fix: register workflow triggers on apply (#160)

Cron, webhook, and email triggers declared in a workflow's YAML were
parsed and validated but never written to workflow_triggers, because the
only function that populated that table (server.SyncTriggers) had no
callers. The cron scheduler, webhook handler, and email poller all read
from that table, so declared triggers never fired.

- workflow.Save now reconciles workflow_triggers in the same transaction
  as the definition insert, so both `mantle apply` and GitOps sync
  register a workflow's triggers atomically with its new version.
- workflow.Disable/Reenable toggle the triggers' enabled flag so a
  pruned/disabled workflow stops firing (schedulers key off
  workflow_triggers.enabled, not the definition's disabled_at) and a
  re-enabled unchanged workflow gets its triggers back.
- The email poller periodically reconciles against the DB so email
  triggers registered by a later apply (a separate process) or a GitOps
  sync activate without a server restart. Cron and webhook triggers are
  already read live per tick/request.
- Removed the dead server.SyncTriggers/TriggerInput; trigger-write logic
  now lives in the workflow package (avoids an import cycle).

Adds integration tests covering registration, version replacement, the
no-trigger case, and the disable/reenable toggle.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WnKgCRfMvFcZAkoVbr8k79

Author: claude

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

@@ -44,14 +44,31 @@ 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.
+	if err := syncTriggersTx(ctx, tx, teamID, name, newVersion, result.Workflow.Triggers); 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 +211,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 +226,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 +252,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 +267,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,53 @@
+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.
+func syncTriggersTx(ctx context.Context, tx *sql.Tx, teamID, name string, version int, triggers []Trigger) 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)
+			 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)`,
+			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),
+		); err != nil {
+			return fmt.Errorf("registering %s trigger for %q: %w", t.Type, name, err)
+		}
+	}
+
+	return nil
+}
+
+// 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
+}