feat: add schema and migrations for push queue (#36)

flanksource · Feb 22, 2023 · 24b4218 · 24b4218
1 parent e6d316d
commit 24b4218
Show file tree

Hide file tree

Showing 4 changed files with 116 additions and 14 deletions.
diff --git a/models/canary.go b/models/canary.go
@@ -24,3 +24,7 @@ type Canary struct {
 func (c Canary) GetCheckID(checkName string) string {
 	return c.Checks[checkName]
 }
+
+func (c Canary) TableName() string {
+	return "canaries"
+}
diff --git a/models/checks.go b/models/checks.go
@@ -80,3 +80,21 @@ func (u Uptime) String() string {
 	percentage := 100.0 * (1 - (float64(u.Failed) / float64(u.Passed+u.Failed)))
 	return fmt.Sprintf("%d/%d (%0.1f%%)", u.Passed, u.Passed+u.Failed, percentage)
 }
+
+type CheckStatus struct {
+	Status   bool   `json:"status"`
+	Invalid  bool   `json:"invalid,omitempty"`
+	Time     string `json:"time"`
+	Duration int    `json:"duration"`
+	Message  string `json:"message,omitempty"`
+	Error    string `json:"error,omitempty"`
+	Detail   any    `json:"-"`
+}
+
+func (s CheckStatus) GetTime() (time.Time, error) {
+	return time.Parse("2006-01-02 15:04:05", s.Time)
+}
+
+func (s CheckStatus) TableName() string {
+	return "check_statuses"
+}
diff --git a/models/components.go b/models/components.go
@@ -78,20 +78,6 @@ type Link struct {
 	Text `json:",inline"`
 }
 
-type CheckStatus struct {
-	Status   bool   `json:"status"`
-	Invalid  bool   `json:"invalid,omitempty"`
-	Time     string `json:"time"`
-	Duration int    `json:"duration"`
-	Message  string `json:"message,omitempty"`
-	Error    string `json:"error,omitempty"`
-	Detail   any    `json:"-"`
-}
-
-func (s CheckStatus) GetTime() (time.Time, error) {
-	return time.Parse("2006-01-02 15:04:05", s.Time)
-}
-
 type Latency struct {
 	Percentile99 float64 `json:"p99,omitempty" db:"p99"`
 	Percentile97 float64 `json:"p97,omitempty" db:"p97"`

diff --git a/views/012_changelog.sql b/views/012_changelog.sql
@@ -0,0 +1,94 @@
+-- Trigger def
+CREATE OR REPLACE FUNCTION change_trigger() 
+RETURNS trigger AS $$
+BEGIN
+  IF TG_TABLE_NAME = 'component_relationships' THEN
+    IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'component_id', NEW.component_id, 'relationship_id', NEW.relationship_id, 'selector_id', NEW.selector_id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN NEW;
+    ELSIF TG_OP = 'DELETE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'component_id', OLD.component_id, 'relationship_id', OLD.relationship_id, 'selector_id', OLD.selector_id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN OLD;
+    END IF;
+
+  ELSIF TG_TABLE_NAME = 'config_component_relationships' THEN
+    IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'component_id', NEW.component_id, 'config_id', NEW.config_id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN NEW;
+    ELSIF TG_OP = 'DELETE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'component_id', OLD.component_id, 'config_id', OLD.config_id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN OLD;
+    END IF;
+
+  ELSIF TG_TABLE_NAME = 'config_relationships' THEN
+    IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'related_id', NEW.related_id, 'config_id', NEW.config_id, 'selector_id', NEW.selector_id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN NEW;
+    ELSIF TG_OP = 'DELETE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'related_id', OLD.related_id, 'config_id', OLD.config_id, 'selector_id', OLD.selector_id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN OLD;
+    END IF;
+
+  ELSIF TG_TABLE_NAME = 'check_statuses' THEN
+    IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'check_id', NEW.check_id, 'time', NEW.time));
+      NOTIFY event_queue_updates, 'update';
+      RETURN NEW;
+    ELSIF TG_OP = 'DELETE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'check_id', OLD.check_id, 'time', OLD.time));
+      NOTIFY event_queue_updates, 'update';
+      RETURN OLD;
+    END IF;
+
+  ELSE
+    IF TG_OP = 'INSERT' OR TG_OP = 'UPDATE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'id', NEW.id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN NEW;
+    ELSIF TG_OP = 'DELETE' THEN
+      INSERT INTO event_queue (name, properties) VALUES ('push_queue.create', jsonb_build_object('table', TG_TABLE_NAME, 'id', OLD.id));
+      NOTIFY event_queue_updates, 'update';
+      RETURN OLD;
+    END IF;
+  END IF;
+END;
+$$ LANGUAGE 'plpgsql' SECURITY DEFINER;
+
+-- Apply trigger
+DO $$ 
+DECLARE 
+  table_name TEXT;
+BEGIN 
+  FOR table_name IN 
+    SELECT t.table_name 
+    FROM information_schema.tables  t
+    WHERE t.table_schema = 'public' AND t.table_type = 'BASE TABLE'
+      AND t.table_name IN (
+        'canaries',
+        'checks',
+        'components',
+        'config_analysis',
+        'config_changes',
+        'config_items',
+        'check_statuses',
+        'config_component_relationships',
+        'component_relationships',
+        'config_relationships'
+      )
+  LOOP 
+    EXECUTE format(
+      'DROP TRIGGER IF EXISTS %1$I_change_trigger ON %1$I;
+      CREATE TRIGGER %1$I_change_trigger
+      BEFORE INSERT OR UPDATE OR DELETE ON %1$I
+      FOR EACH ROW
+      EXECUTE PROCEDURE change_trigger()',
+      table_name
+    );
+  END LOOP; 
+END $$;