fix: register workflow triggers on apply (#162)

Author: michaelmcnees

.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.

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
 }

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
-}

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,14 +51,32 @@ 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,
 	)
 	if err != nil {
 		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

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
+}