feat: playbook action consumer & playbook_action_agent_data table

models/playbooks.go

@@ -11,13 +11,14 @@ import (
 type PlaybookRunStatus string
 
 const (
-	PlaybookRunStatusPending   PlaybookRunStatus = "pending"
-	PlaybookRunStatusScheduled PlaybookRunStatus = "scheduled"
-	PlaybookRunStatusRunning   PlaybookRunStatus = "running"
 	PlaybookRunStatusCancelled PlaybookRunStatus = "cancelled"
-	PlaybookRunStatusFailed    PlaybookRunStatus = "failed"
 	PlaybookRunStatusCompleted PlaybookRunStatus = "completed"
+	PlaybookRunStatusFailed    PlaybookRunStatus = "failed"
+	PlaybookRunStatusPending   PlaybookRunStatus = "pending" // pending approval
+	PlaybookRunStatusRunning   PlaybookRunStatus = "running"
+	PlaybookRunStatusScheduled PlaybookRunStatus = "scheduled"
 	PlaybookRunStatusSleeping  PlaybookRunStatus = "sleeping"
+	PlaybookRunStatusWaiting   PlaybookRunStatus = "waiting" // waiting for a consumer
 )
 
 // PlaybookRunStatus are statuses for a playbook run and its actions.
@@ -27,6 +28,7 @@ const (
 	PlaybookActionStatusCompleted PlaybookActionStatus = "completed"
 	PlaybookActionStatusFailed    PlaybookActionStatus = "failed"
 	PlaybookActionStatusRunning   PlaybookActionStatus = "running"
+	PlaybookActionStatusScheduled PlaybookActionStatus = "scheduled"
 	PlaybookActionStatusSkipped   PlaybookActionStatus = "skipped"
 	PlaybookActionStatusSleeping  PlaybookActionStatus = "sleeping"
 )
@@ -67,7 +69,7 @@ type PlaybookRun struct {
 	PlaybookID    uuid.UUID           `json:"playbook_id"`
 	Status        PlaybookRunStatus   `json:"status,omitempty"`
 	CreatedAt     time.Time           `json:"created_at,omitempty" time_format:"postgres_timestamp" gorm:"<-:false"`
-	StartTime     time.Time           `json:"start_time,omitempty" time_format:"postgres_timestamp"`
+	StartTime     *time.Time          `json:"start_time,omitempty" time_format:"postgres_timestamp"`
 	ScheduledTime time.Time           `json:"scheduled_time,omitempty" time_format:"postgres_timestamp" gorm:"default:NOW(), NOT NULL"`
 	EndTime       *time.Time          `json:"end_time,omitempty" time_format:"postgres_timestamp"`
 	CreatedBy     *uuid.UUID          `json:"created_by,omitempty"`
@@ -111,3 +113,15 @@ type PlaybookApproval struct {
 func (p PlaybookApproval) AsMap(removeFields ...string) map[string]any {
 	return asMap(p, removeFields...)
 }
+
+type PlaybookActionAgentData struct {
+	ActionID   uuid.UUID  `json:"action_id"`
+	RunID      uuid.UUID  `json:"run_id"`
+	PlaybookID uuid.UUID  `json:"playbook_id"`
+	Spec       types.JSON `json:"spec"`
+	Env        types.JSON `json:"env,omitempty"`
+}
+
+func (t *PlaybookActionAgentData) TableName() string {
+	return "playbook_action_agent_data"
+}

schema/playbooks.hcl

@@ -221,6 +221,41 @@ table "playbook_runs" {
   }
 }
 
+table "playbook_action_agent_data" {
+  schema  = schema.public
+  comment = "saves the necessary details for the agent to run a playbook action (eg: template env vars). Only applicable to agent runners."
+  column "action_id" {
+    null = false
+    type = uuid
+  }
+  column "playbook_id" {
+    comment = "saves the linked upstream playbook id"
+    null = false
+    type = uuid
+  }
+  column "run_id" {
+    comment = "saves the linked upstream playbook run id"
+    null = false
+    type = uuid
+  }
+  column "spec" {
+    comment = "Action spec provided by upstream"
+    null    = false
+    type    = jsonb
+  }
+  column "env" {
+    comment = "templateEnv for the action provided by the upstream"
+    null    = true
+    type    = jsonb
+  }
+  foreign_key "playbook_action_template_env_agent_action_id_fkey" {
+    columns     = [column.action_id]
+    ref_columns = [table.playbook_run_actions.column.id]
+    on_update   = NO_ACTION
+    on_delete   = CASCADE
+  }
+}
+
 table "playbook_run_actions" {
   schema = schema.public
   column "id" {
@@ -238,14 +273,13 @@ table "playbook_run_actions" {
     default = "running"
   }
   column "playbook_run_id" {
-    null = true
-    type = uuid
+    null    = true
+    type    = uuid
     comment = "a run id is mandatory except for an agent"
   }
   column "start_time" {
-    null    = true
-    type    = timestamptz
-    default = sql("now()")
+    null = true
+    type = timestamptz
   }
   column "scheduled_time" {
     null    = false

upstream/commands.go

@@ -195,7 +195,11 @@ func InsertUpstreamMsg(ctx context.Context, req *PushData) error {
 			"error":    req.PlaybookActions[i].Error,
 		}
 		if err := db.Model(&models.PlaybookRunAction{}).Where("id = ?", req.PlaybookActions[i].ID).Updates(updates).Error; err != nil {
-			return fmt.Errorf("error updating playbook action [%s]: %w", req.PlaybookActions[i].ID.String(), err)
+			return fmt.Errorf("error updating playbook action [%s]: %w", req.PlaybookActions[i].ID, err)
+		}
+
+		if err := db.Exec("UPDATE playbook_runs SET status = ? WHERE id = (SELECT playbook_run_id FROM playbook_run_actions WHERE id = ?)", models.PlaybookRunStatusScheduled, req.PlaybookActions[i].ID).Error; err != nil {
+			return fmt.Errorf("error updating playbook run [%s]  status to %s : %w", req.PlaybookActions[i].PlaybookRunID, models.PlaybookRunStatusScheduled, err)
 		}
 	}
 

views/018_playbooks.sql

@@ -1,3 +1,23 @@
+-- Notify playbook action created or status updated
+CREATE OR REPLACE FUNCTION notify_playbook_action_update() RETURNS TRIGGER AS $$
+BEGIN
+  IF TG_OP = 'INSERT' THEN
+    NOTIFY playbook_action_updates;
+  ELSEIF TG_OP = 'UPDATE' THEN
+    IF OLD.status != NEW.status AND NEW.status = 'scheduled' THEN
+      NOTIFY playbook_action_updates;
+    END IF;
+  END IF;
+
+  RETURN NULL;
+END
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE TRIGGER playbook_action_updates
+AFTER INSERT OR UPDATE ON playbook_run_actions
+FOR EACH ROW
+EXECUTE PROCEDURE notify_playbook_action_update();
+
 -- Notify playbook run created or status updated
 CREATE OR REPLACE FUNCTION notify_playbook_run_update() RETURNS TRIGGER AS $$
 BEGIN