feat: notification delay & event_queue delay

flanksource · Sep 17, 2024 · d04d339 · d04d339
1 parent ca2b9a8
commit d04d339
Show file tree

Hide file tree

Showing 8 changed files with 19 additions and 6 deletions.
diff --git a/connection/zz_generated.deepcopy.go b/connection/zz_generated.deepcopy.go
diff --git a/models/event_queue.go b/models/event_queue.go
@@ -16,6 +16,7 @@ type Event struct {
 	Name        string              `json:"name"`
 	CreatedAt   time.Time           `json:"created_at"`
 	Properties  types.JSONStringMap `json:"properties"`
+	Delay       *time.Duration      `json:"delay,omitempty"`
 	Error       *string             `json:"error,omitempty"`
 	Attempts    int                 `json:"attempts"`
 	LastAttempt *time.Time          `json:"last_attempt"`

diff --git a/models/notifications.go b/models/notifications.go
@@ -16,6 +16,7 @@ type Notification struct {
 	Namespace      string              `json:"namespace,omitempty"`
 	Events         pq.StringArray      `json:"events" gorm:"type:[]text"`
 	Title          string              `json:"title,omitempty"`
+	WaitFor        *time.Duration      `json:"wait_for,omitempty"`
 	Template       string              `json:"template,omitempty"`
 	Filter         string              `json:"filter,omitempty"`
 	PersonID       *uuid.UUID          `json:"person_id,omitempty"`

diff --git a/postq/event.go b/postq/event.go
@@ -35,6 +35,7 @@ func fetchEvents(ctx context.Context, tx *gorm.DB, watchEvents []string, batchSi
 		WHERE id IN (
 			SELECT id FROM event_queue
 			WHERE
+				(delay IS NULL OR created_at + (delay * INTERVAL '1 second' / 1000000000)  <= NOW()) AND
 				attempts <= @MaxAttempts AND
 				name = ANY(@Events) AND
 				(last_attempt IS NULL OR last_attempt <= NOW() - INTERVAL '1 SECOND' * @BaseDelay * POWER(attempts, @Exponent))

diff --git a/schema/notifications.hcl b/schema/notifications.hcl
@@ -6,8 +6,8 @@ table "notifications" {
     default = sql("generate_ulid()")
   }
   column "name" {
-    null = false
-    type = text
+    null    = false
+    type    = text
     default = sql("generate_ulid()") # temporary default value to make the migration possible. we can remove this later.
   }
   column "namespace" {
@@ -54,6 +54,11 @@ table "notifications" {
     null = true
     type = text
   }
+  column "wait_for" {
+    null    = true
+    type    = bigint
+    comment = "duration in nanoseconds"
+  }
   column "group_by" {
     null    = true
     type    = sql("text[]")

diff --git a/schema/system.hcl b/schema/system.hcl
@@ -59,6 +59,11 @@ table "event_queue" {
     null = true
     type = text
   }
+  column "delay" {
+    null    = true
+    type    = bigint
+    comment = "wait for this duration (nanoseconds) before consuming"
+  }
   column "created_at" {
     null    = false
     type    = timestamptz

diff --git a/shell/zz_generated.deepcopy.go b/shell/zz_generated.deepcopy.go
diff --git a/types/zz_generated.deepcopy.go b/types/zz_generated.deepcopy.go