diff --git a/.changeset/fix-trigger-registration.md b/.changeset/fix-trigger-registration.md new file mode 100644 index 00000000..e4d3f644 --- /dev/null +++ b/.changeset/fix-trigger-registration.md @@ -0,0 +1,5 @@ +--- +"@mantle/engine": patch +--- + +Register workflow triggers on apply. Cron, webhook, and email triggers declared in a workflow's YAML were parsed and validated but never written to `workflow_triggers`, so they never fired. `workflow.Save` now reconciles the table in the same transaction as the definition insert (covering both `mantle apply` and GitOps sync). Re-applying an unchanged workflow backfills its triggers if the table has drifted (so upgrading and re-applying activates triggers for workflows created before this fix, without churning rows in steady state). Pruned/disabled workflows have their triggers disabled so they stop firing, and the email poller periodically reloads so new email triggers activate without a server restart. diff --git a/packages/engine/internal/server/email_trigger.go b/packages/engine/internal/server/email_trigger.go index fdfe2e22..31b196c7 100644 --- a/packages/engine/internal/server/email_trigger.go +++ b/packages/engine/internal/server/email_trigger.go @@ -22,6 +22,12 @@ const ( emailReconnectBaseDelay = 5 * time.Second emailReconnectMaxDelay = 5 * time.Minute defaultMaxEmailConns = 5 + // emailReloadInterval is how often the poller reconciles its running + // pollers against workflow_triggers, so email triggers registered by a + // later `mantle apply` (a separate process) or a GitOps sync start + // without requiring a server restart. Cron and webhook triggers are read + // from the DB on every tick/request, so they need no equivalent. + emailReloadInterval = 30 * time.Second ) // EmailTriggerPoller starts one goroutine per email trigger and maintains a @@ -58,6 +64,25 @@ func (e *EmailTriggerPoller) Start(ctx context.Context) error { e.startPoller(ctx, t) } + // Periodically reconcile against the DB so triggers added or removed + // after startup take effect without a restart. + e.wg.Add(1) + go func() { + defer e.wg.Done() + ticker := time.NewTicker(emailReloadInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := e.Reload(ctx); err != nil { + e.server.Logger.Error("email poller: periodic reload failed", "error", err) + } + } + } + }() + e.server.Logger.Info("email trigger poller started", "triggers", len(triggers)) return nil } diff --git a/packages/engine/internal/server/trigger.go b/packages/engine/internal/server/trigger.go index 796ac12d..f7308d8a 100644 --- a/packages/engine/internal/server/trigger.go +++ b/packages/engine/internal/server/trigger.go @@ -3,7 +3,6 @@ package server import ( "context" "database/sql" - "fmt" "github.com/dvflw/mantle/internal/auth" ) @@ -27,53 +26,10 @@ type TriggerRecord struct { PollInterval string } -// SyncTriggers replaces all triggers for a workflow with the given set. -// Called by `mantle apply` when a workflow definition changes. -func SyncTriggers(ctx context.Context, db *sql.DB, workflowName string, version int, triggers []TriggerInput) error { - teamID := auth.TeamIDFromContext(ctx) - - tx, err := db.BeginTx(ctx, nil) - if err != nil { - return fmt.Errorf("starting transaction: %w", err) - } - defer tx.Rollback() - - // Remove existing triggers for this workflow, scoped to team. - _, err = tx.ExecContext(ctx, - `DELETE FROM workflow_triggers WHERE workflow_name = $1 AND team_id = $2`, workflowName, teamID) - if err != nil { - return fmt.Errorf("deleting old triggers: %w", err) - } - - // Insert new triggers. - for _, t := range triggers { - _, err = tx.ExecContext(ctx, - `INSERT INTO workflow_triggers - (workflow_name, workflow_version, type, schedule, path, secret, team_id, mailbox, folder, filter, poll_interval) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`, - workflowName, version, t.Type, t.Schedule, t.Path, t.Secret, teamID, - nullableString(t.Mailbox), nullableString(t.Folder), nullableString(t.Filter), nullableString(t.PollInterval)) - if err != nil { - return fmt.Errorf("inserting trigger: %w", err) - } - } - - return tx.Commit() -} - -// TriggerInput is the data needed to register a trigger. -type TriggerInput struct { - Type string - Schedule string - Path string - Secret string - - // Email trigger fields (used only when Type == "email"). - Mailbox string - Folder string - Filter string - PollInterval string -} +// Trigger registration lives in the workflow package (workflow.Save), which +// reconciles workflow_triggers whenever a definition is applied. The functions +// below are the read side consumed by the cron scheduler, webhook handler, and +// email poller. // ListCronTriggers returns all enabled cron triggers, including team_id for proper scoping. func ListCronTriggers(ctx context.Context, db *sql.DB) ([]TriggerRecord, error) { @@ -148,12 +104,3 @@ func scanEmailTriggers(rows *sql.Rows) ([]TriggerRecord, error) { } return triggers, rows.Err() } - -// nullableString converts an empty string to nil so that nullable TEXT columns -// receive NULL rather than an empty string. -func nullableString(s string) interface{} { - if s == "" { - return nil - } - return s -} diff --git a/packages/engine/internal/workflow/store.go b/packages/engine/internal/workflow/store.go index 6499121f..a00d45ae 100644 --- a/packages/engine/internal/workflow/store.go +++ b/packages/engine/internal/workflow/store.go @@ -30,6 +30,13 @@ func Save(ctx context.Context, database *sql.DB, result *ParseResult, rawContent return 0, fmt.Errorf("checking latest hash: %w", err) } if latestHash == hash { + // Content unchanged, so no new version — but still backfill triggers + // if the table drifted from the definition (e.g. a workflow applied + // before trigger registration existed). Otherwise upgrading and + // re-applying would leave triggers inert until a byte-level change. + if err := reconcileTriggersIfDrifted(ctx, database, teamID, name, result.Workflow.Triggers); err != nil { + return 0, err + } return 0, nil } @@ -44,7 +51,13 @@ func Save(ctx context.Context, database *sql.DB, result *ParseResult, rawContent return 0, fmt.Errorf("marshaling workflow: %w", err) } - _, err = database.ExecContext(ctx, + tx, err := database.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("starting transaction: %w", err) + } + defer tx.Rollback() //nolint:errcheck + + _, err = tx.ExecContext(ctx, `INSERT INTO workflow_definitions (name, version, content, content_hash, team_id) VALUES ($1, $2, $3, $4, $5)`, name, newVersion, content, hash, teamID, ) @@ -52,6 +65,18 @@ func Save(ctx context.Context, database *sql.DB, result *ParseResult, rawContent return 0, fmt.Errorf("inserting workflow definition: %w", err) } + // Register the workflow's declared triggers so the cron/webhook/email + // schedulers actually fire it. Done in the same transaction as the + // definition insert so version and triggers commit atomically. A new + // version is always active, so its triggers are enabled. + if err := syncTriggersTx(ctx, tx, teamID, name, newVersion, result.Workflow.Triggers, true); err != nil { + return 0, err + } + + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("committing workflow save: %w", err) + } + return newVersion, nil } @@ -194,7 +219,13 @@ func GetVersion(ctx context.Context, database *sql.DB, name string, version int) // speculatively during a GitOps prune pass). func Disable(ctx context.Context, database *sql.DB, name string) error { teamID := auth.TeamIDFromContext(ctx) - _, err := database.ExecContext(ctx, + tx, err := database.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("disabling workflow %q: %w", name, err) + } + defer tx.Rollback() //nolint:errcheck + + if _, err := tx.ExecContext(ctx, `UPDATE workflow_definitions SET disabled_at = NOW() WHERE id = ( @@ -203,8 +234,23 @@ func Disable(ctx context.Context, database *sql.DB, name string) error { ORDER BY version DESC LIMIT 1 )`, name, teamID, - ) - if err != nil { + ); err != nil { + return fmt.Errorf("disabling workflow %q: %w", name, err) + } + + // Disable the workflow's triggers too — the schedulers key off + // workflow_triggers.enabled, not the definition's disabled_at, so a + // pruned workflow would otherwise keep firing. Guarded on enabled = true + // so repeated GitOps passes don't churn updated_at. + if _, err := tx.ExecContext(ctx, + `UPDATE workflow_triggers SET enabled = false, updated_at = NOW() + WHERE workflow_name = $1 AND team_id = $2 AND enabled = true`, + name, teamID, + ); err != nil { + return fmt.Errorf("disabling triggers for %q: %w", name, err) + } + + if err := tx.Commit(); err != nil { return fmt.Errorf("disabling workflow %q: %w", name, err) } return nil @@ -214,7 +260,13 @@ func Disable(ctx context.Context, database *sql.DB, name string) error { // workflow. Safe to call when already enabled — becomes a no-op. func Reenable(ctx context.Context, database *sql.DB, name string) error { teamID := auth.TeamIDFromContext(ctx) - _, err := database.ExecContext(ctx, + tx, err := database.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("reenabling workflow %q: %w", name, err) + } + defer tx.Rollback() //nolint:errcheck + + if _, err := tx.ExecContext(ctx, `UPDATE workflow_definitions SET disabled_at = NULL WHERE id = ( @@ -223,8 +275,24 @@ func Reenable(ctx context.Context, database *sql.DB, name string) error { ORDER BY version DESC LIMIT 1 )`, name, teamID, - ) - if err != nil { + ); err != nil { + return fmt.Errorf("reenabling workflow %q: %w", name, err) + } + + // Restore the workflow's triggers. This covers the GitOps case where a + // workflow was pruned (triggers disabled) and later reappears with + // byte-identical content: Save is a no-op so it won't re-register them, + // but Reenable must. Guarded on enabled = false to stay a true no-op on + // the common already-enabled path. + if _, err := tx.ExecContext(ctx, + `UPDATE workflow_triggers SET enabled = true, updated_at = NOW() + WHERE workflow_name = $1 AND team_id = $2 AND enabled = false`, + name, teamID, + ); err != nil { + return fmt.Errorf("reenabling triggers for %q: %w", name, err) + } + + if err := tx.Commit(); err != nil { return fmt.Errorf("reenabling workflow %q: %w", name, err) } return nil diff --git a/packages/engine/internal/workflow/triggers.go b/packages/engine/internal/workflow/triggers.go new file mode 100644 index 00000000..cb7795af --- /dev/null +++ b/packages/engine/internal/workflow/triggers.go @@ -0,0 +1,106 @@ +package workflow + +import ( + "context" + "database/sql" + "fmt" +) + +// syncTriggersTx reconciles the workflow_triggers rows for a workflow within an +// existing transaction. It removes the workflow's current triggers (scoped to +// the team) and inserts the declared set, all enabled. This is what makes +// cron/webhook/email triggers declared in a workflow's YAML actually fire: +// the cron scheduler, webhook handler, and email poller all read from +// workflow_triggers, and without this reconciliation that table stays empty. +// +// It runs inside Save's transaction so a new definition version and its +// triggers are committed atomically — a workflow is never left registered at +// one version with triggers pointing at another. enabled sets the rows' +// enabled flag: true for an active workflow, false when backfilling a +// currently-disabled one so it does not start firing. +func syncTriggersTx(ctx context.Context, tx *sql.Tx, teamID, name string, version int, triggers []Trigger, enabled bool) error { + // A new version supersedes the previous one, so replace whatever was + // registered before rather than trying to diff. Scoped to team so we + // never touch another tenant's rows. + if _, err := tx.ExecContext(ctx, + `DELETE FROM workflow_triggers WHERE workflow_name = $1 AND team_id = $2`, + name, teamID); err != nil { + return fmt.Errorf("clearing triggers for %q: %w", name, err) + } + + for _, t := range triggers { + if _, err := tx.ExecContext(ctx, + `INSERT INTO workflow_triggers + (workflow_name, workflow_version, type, schedule, path, secret, team_id, mailbox, folder, filter, poll_interval, enabled) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`, + name, version, t.Type, + nullableString(t.Schedule), nullableString(t.Path), nullableString(t.Secret), teamID, + nullableString(t.Mailbox), nullableString(t.Folder), nullableString(t.Filter), nullableString(t.PollInterval), + enabled, + ); err != nil { + return fmt.Errorf("registering %s trigger for %q: %w", t.Type, name, err) + } + } + + return nil +} + +// reconcileTriggersIfDrifted backfills a workflow's triggers on the +// unchanged-content apply path. Save early-returns when the content hash is +// unchanged, so without this a workflow applied before trigger registration +// existed (definition present, workflow_triggers empty) would never register +// its triggers unless the user made a byte-level YAML change. It reconciles +// only when the stored rows are actually out of sync with the latest version's +// declared triggers, so steady-state GitOps re-applies stay churn-free (which +// matters for the email poller, whose IMAP connections key off trigger IDs). +// The backfilled rows inherit the workflow's current enabled/disabled state. +func reconcileTriggersIfDrifted(ctx context.Context, database *sql.DB, teamID, name string, desired []Trigger) error { + // Latest version and whether it is currently active (not disabled). + var version int + var active bool + err := database.QueryRowContext(ctx, + `SELECT version, disabled_at IS NULL FROM workflow_definitions + WHERE name = $1 AND team_id = $2 ORDER BY version DESC LIMIT 1`, + name, teamID).Scan(&version, &active) + if err == sql.ErrNoRows { + return nil // no definition to reconcile against + } + if err != nil { + return fmt.Errorf("loading latest version for %q: %w", name, err) + } + + // Already in sync when the row count matches and every row points at the + // latest version. Covers both the empty-table (pre-fix) case and rows + // left at a stale version. + var total, atVersion int + if err := database.QueryRowContext(ctx, + `SELECT COUNT(*), COUNT(*) FILTER (WHERE workflow_version = $3) + FROM workflow_triggers WHERE workflow_name = $1 AND team_id = $2`, + name, teamID, version).Scan(&total, &atVersion); err != nil { + return fmt.Errorf("checking trigger drift for %q: %w", name, err) + } + if total == len(desired) && atVersion == len(desired) { + return nil + } + + tx, err := database.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("starting transaction: %w", err) + } + defer tx.Rollback() //nolint:errcheck + + if err := syncTriggersTx(ctx, tx, teamID, name, version, desired, active); err != nil { + return err + } + return tx.Commit() +} + +// nullableString maps an empty string to NULL so that nullable TEXT columns +// receive NULL rather than an empty string. Reads COALESCE these back to +// their defaults. +func nullableString(s string) any { + if s == "" { + return nil + } + return s +} diff --git a/packages/engine/internal/workflow/triggers_test.go b/packages/engine/internal/workflow/triggers_test.go new file mode 100644 index 00000000..3641c0d5 --- /dev/null +++ b/packages/engine/internal/workflow/triggers_test.go @@ -0,0 +1,328 @@ +package workflow + +import ( + "context" + "database/sql" + "testing" +) + +// countTriggers returns the number of workflow_triggers rows for a workflow, +// optionally filtering to enabled rows only. +func countTriggers(t *testing.T, database *sql.DB, name string, enabledOnly bool) int { + t.Helper() + q := `SELECT COUNT(*) FROM workflow_triggers WHERE workflow_name = $1` + if enabledOnly { + q += ` AND enabled = true` + } + var n int + if err := database.QueryRowContext(context.Background(), q, name).Scan(&n); err != nil { + t.Fatalf("counting triggers: %v", err) + } + return n +} + +var triggeredWorkflowYAML = []byte(`name: triggered-wf +description: workflow with all trigger types + +triggers: + - type: cron + schedule: "*/5 * * * *" + - type: webhook + path: /incoming + secret: shh + - type: email + mailbox: inbox-cred + folder: INBOX + filter: unseen + poll_interval: 30s + +steps: + - name: s + action: http/request + params: + method: GET + url: "https://example.com" +`) + +// TestSave_RegistersTriggers is the regression test for the bug where triggers +// declared in YAML were parsed but never written to workflow_triggers, so they +// never fired. Applying a workflow must populate the table the schedulers read. +func TestSave_RegistersTriggers(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + result, err := ParseBytes(triggeredWorkflowYAML) + if err != nil { + t.Fatalf("ParseBytes: %v", err) + } + version, err := Save(ctx, database, result, triggeredWorkflowYAML) + if err != nil { + t.Fatalf("Save: %v", err) + } + if version != 1 { + t.Fatalf("version = %d, want 1", version) + } + + if got := countTriggers(t, database, "triggered-wf", false); got != 3 { + t.Fatalf("registered triggers = %d, want 3", got) + } + + // The cron trigger row must carry the schedule and point at the version. + var ( + schedule string + ver int + ) + if err := database.QueryRowContext(ctx, + `SELECT COALESCE(schedule, ''), workflow_version FROM workflow_triggers + WHERE workflow_name = 'triggered-wf' AND type = 'cron'`, + ).Scan(&schedule, &ver); err != nil { + t.Fatalf("querying cron trigger: %v", err) + } + if schedule != "*/5 * * * *" { + t.Errorf("cron schedule = %q, want %q", schedule, "*/5 * * * *") + } + if ver != 1 { + t.Errorf("cron trigger version = %d, want 1", ver) + } + + // The email trigger row must carry its email-specific fields. + var mailbox, folder, filter, poll string + if err := database.QueryRowContext(ctx, + `SELECT COALESCE(mailbox,''), COALESCE(folder,''), COALESCE(filter,''), COALESCE(poll_interval,'') + FROM workflow_triggers WHERE workflow_name = 'triggered-wf' AND type = 'email'`, + ).Scan(&mailbox, &folder, &filter, &poll); err != nil { + t.Fatalf("querying email trigger: %v", err) + } + if mailbox != "inbox-cred" || folder != "INBOX" || filter != "unseen" || poll != "30s" { + t.Errorf("email fields = (%q,%q,%q,%q), want (inbox-cred,INBOX,unseen,30s)", + mailbox, folder, filter, poll) + } + + // The webhook trigger row must carry its path and secret. + var path, secret string + if err := database.QueryRowContext(ctx, + `SELECT COALESCE(path,''), COALESCE(secret,'') FROM workflow_triggers + WHERE workflow_name = 'triggered-wf' AND type = 'webhook'`, + ).Scan(&path, &secret); err != nil { + t.Fatalf("querying webhook trigger: %v", err) + } + if path != "/incoming" || secret != "shh" { + t.Errorf("webhook fields = (%q,%q), want (/incoming,shh)", path, secret) + } +} + +// triggerIDs returns the sorted IDs of a workflow's trigger rows. +func triggerIDs(t *testing.T, database *sql.DB, name string) []string { + t.Helper() + rows, err := database.QueryContext(context.Background(), + `SELECT id::text FROM workflow_triggers WHERE workflow_name = $1 ORDER BY id`, name) + if err != nil { + t.Fatalf("querying trigger ids: %v", err) + } + defer rows.Close() + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + t.Fatalf("scanning id: %v", err) + } + ids = append(ids, id) + } + return ids +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// TestSave_BackfillsTriggersOnUnchangedApply is the regression test for the +// upgrade path: a workflow whose definition already exists but has no +// workflow_triggers rows (applied before trigger registration existed) must +// get its triggers backfilled when identical content is re-applied — without +// churning rows on subsequent no-op applies. +func TestSave_BackfillsTriggersOnUnchangedApply(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + result, _ := ParseBytes(triggeredWorkflowYAML) + if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil { + t.Fatalf("initial Save: %v", err) + } + + // Simulate a pre-fix workflow: definition present, trigger rows absent. + if _, err := database.ExecContext(ctx, + `DELETE FROM workflow_triggers WHERE workflow_name = 'triggered-wf'`); err != nil { + t.Fatalf("clearing triggers: %v", err) + } + if got := countTriggers(t, database, "triggered-wf", false); got != 0 { + t.Fatalf("precondition: triggers = %d, want 0", got) + } + + // Re-applying identical content must backfill the triggers (version stays 0). + version, err := Save(ctx, database, result, triggeredWorkflowYAML) + if err != nil { + t.Fatalf("backfill Save: %v", err) + } + if version != 0 { + t.Errorf("version = %d, want 0 (unchanged content)", version) + } + if got := countTriggers(t, database, "triggered-wf", true); got != 3 { + t.Fatalf("backfilled enabled triggers = %d, want 3", got) + } + + // A further identical apply must be a true no-op: rows already in sync, + // so their IDs must not churn (the email poller keys off trigger IDs). + ids1 := triggerIDs(t, database, "triggered-wf") + if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil { + t.Fatalf("no-op Save: %v", err) + } + ids2 := triggerIDs(t, database, "triggered-wf") + if !equalStrings(ids1, ids2) { + t.Errorf("triggers churned on no-op apply: %v -> %v", ids1, ids2) + } +} + +// TestSave_BackfillPreservesDisabledState verifies that backfilling triggers +// for a currently-disabled workflow inserts them disabled, so a pruned +// workflow does not start firing on a re-apply. +func TestSave_BackfillPreservesDisabledState(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + result, _ := ParseBytes(triggeredWorkflowYAML) + if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil { + t.Fatalf("initial Save: %v", err) + } + if err := Disable(ctx, database, "triggered-wf"); err != nil { + t.Fatalf("Disable: %v", err) + } + // Drop the rows to force a backfill on the next apply. + if _, err := database.ExecContext(ctx, + `DELETE FROM workflow_triggers WHERE workflow_name = 'triggered-wf'`); err != nil { + t.Fatalf("clearing triggers: %v", err) + } + + if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil { + t.Fatalf("backfill Save: %v", err) + } + if got := countTriggers(t, database, "triggered-wf", false); got != 3 { + t.Errorf("backfilled rows = %d, want 3", got) + } + if got := countTriggers(t, database, "triggered-wf", true); got != 0 { + t.Errorf("enabled backfilled rows = %d, want 0 (workflow is disabled)", got) + } +} + +// TestSave_ReplacesTriggersOnNewVersion verifies a new version supersedes the +// prior triggers rather than accumulating duplicates. +func TestSave_ReplacesTriggersOnNewVersion(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + result, _ := ParseBytes(triggeredWorkflowYAML) + if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil { + t.Fatalf("first Save: %v", err) + } + + // New version with a single cron trigger on a different schedule. + updated := []byte(`name: triggered-wf +description: now with one trigger + +triggers: + - type: cron + schedule: "0 * * * *" + +steps: + - name: s + action: http/request + params: + method: GET + url: "https://example.com" +`) + result2, err := ParseBytes(updated) + if err != nil { + t.Fatalf("ParseBytes updated: %v", err) + } + version, err := Save(ctx, database, result2, updated) + if err != nil { + t.Fatalf("second Save: %v", err) + } + if version != 2 { + t.Fatalf("version = %d, want 2", version) + } + + if got := countTriggers(t, database, "triggered-wf", false); got != 1 { + t.Fatalf("triggers after new version = %d, want 1", got) + } + var schedule string + var ver int + if err := database.QueryRowContext(ctx, + `SELECT COALESCE(schedule,''), workflow_version FROM workflow_triggers WHERE workflow_name = 'triggered-wf'`, + ).Scan(&schedule, &ver); err != nil { + t.Fatalf("querying trigger: %v", err) + } + if schedule != "0 * * * *" { + t.Errorf("schedule = %q, want %q", schedule, "0 * * * *") + } + if ver != 2 { + t.Errorf("trigger version = %d, want 2", ver) + } +} + +// TestSave_NoTriggersLeavesTableEmpty verifies a workflow without triggers +// registers none (and doesn't error on the empty set). +func TestSave_NoTriggersLeavesTableEmpty(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + result, _ := ParseBytes(testWorkflowYAML) // defined in store_test.go, has no triggers + if _, err := Save(ctx, database, result, testWorkflowYAML); err != nil { + t.Fatalf("Save: %v", err) + } + if got := countTriggers(t, database, "test-workflow", false); got != 0 { + t.Errorf("triggers = %d, want 0", got) + } +} + +// TestDisableReenable_TogglesTriggerEnabled verifies that disabling a workflow +// (e.g. a GitOps prune) stops its triggers from firing, and re-enabling +// restores them — including the case where the definition is unchanged so Save +// would not re-register them. +func TestDisableReenable_TogglesTriggerEnabled(t *testing.T) { + database := setupTestDB(t) + ctx := context.Background() + + result, _ := ParseBytes(triggeredWorkflowYAML) + if _, err := Save(ctx, database, result, triggeredWorkflowYAML); err != nil { + t.Fatalf("Save: %v", err) + } + if got := countTriggers(t, database, "triggered-wf", true); got != 3 { + t.Fatalf("enabled triggers = %d, want 3", got) + } + + if err := Disable(ctx, database, "triggered-wf"); err != nil { + t.Fatalf("Disable: %v", err) + } + if got := countTriggers(t, database, "triggered-wf", true); got != 0 { + t.Errorf("enabled triggers after Disable = %d, want 0", got) + } + // Rows are retained (disabled), not deleted. + if got := countTriggers(t, database, "triggered-wf", false); got != 3 { + t.Errorf("total triggers after Disable = %d, want 3", got) + } + + if err := Reenable(ctx, database, "triggered-wf"); err != nil { + t.Fatalf("Reenable: %v", err) + } + if got := countTriggers(t, database, "triggered-wf", true); got != 3 { + t.Errorf("enabled triggers after Reenable = %d, want 3", got) + } +}