Feat v0.3.3 public entry background jobs (#17)

Co-authored-by: Jeffrey Smith <jasafpro@gmail.com>
Co-committed-by: Jeffrey Smith <jasafpro@gmail.com>

server/workflow/scanner.go

@@ -0,0 +1,246 @@
+package workflow
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"sync"
+	"time"
+
+	"switchboard-core/events"
+	"switchboard-core/models"
+	"switchboard-core/store"
+)
+
+// Scanner runs periodic SLA and staleness checks on active workflow instances.
+type Scanner struct {
+	stores   store.Stores
+	bus      *events.Bus
+	stopCh   chan struct{}
+	wg       sync.WaitGroup
+	interval time.Duration
+}
+
+// NewScanner creates a workflow scanner with a default 5-minute interval.
+func NewScanner(stores store.Stores, bus *events.Bus) *Scanner {
+	return &Scanner{
+		stores:   stores,
+		bus:      bus,
+		stopCh:   make(chan struct{}),
+		interval: 5 * time.Minute,
+	}
+}
+
+// Start begins the background scan loop.
+func (sc *Scanner) Start() {
+	sc.wg.Add(1)
+	go func() {
+		defer sc.wg.Done()
+		ticker := time.NewTicker(sc.interval)
+		defer ticker.Stop()
+
+		// Delay first run to avoid startup contention
+		time.Sleep(2 * time.Minute)
+		sc.runScan()
+
+		for {
+			select {
+			case <-ticker.C:
+				sc.runScan()
+			case <-sc.stopCh:
+				return
+			}
+		}
+	}()
+	log.Printf("[workflow-scanner] started (interval=%s)", sc.interval)
+}
+
+// Stop halts the background scan loop and waits for completion.
+func (sc *Scanner) Stop() {
+	close(sc.stopCh)
+	sc.wg.Wait()
+	log.Printf("[workflow-scanner] stopped")
+}
+
+// runScan performs one SLA + staleness check cycle.
+func (sc *Scanner) runScan() {
+	ctx := context.Background()
+	instances, err := sc.stores.Workflows.ListActiveInstances(ctx)
+	if err != nil {
+		log.Printf("[workflow-scanner] list active instances: %v", err)
+		return
+	}
+	if len(instances) == 0 {
+		return
+	}
+
+	// Cache version snapshots and workflow definitions to avoid N+1 queries
+	type versionKey struct {
+		workflowID string
+		version    int
+	}
+	versionCache := map[versionKey][]models.WorkflowStage{}
+	workflowCache := map[string]*models.Workflow{}
+
+	getStages := func(wfID string, ver int) []models.WorkflowStage {
+		key := versionKey{wfID, ver}
+		if stages, ok := versionCache[key]; ok {
+			return stages
+		}
+		v, err := sc.stores.Workflows.GetVersion(ctx, wfID, ver)
+		if err != nil {
+			return nil
+		}
+		var stages []models.WorkflowStage
+		if json.Unmarshal(v.Snapshot, &stages) != nil {
+			return nil
+		}
+		versionCache[key] = stages
+		return stages
+	}
+
+	getWorkflow := func(wfID string) *models.Workflow {
+		if wf, ok := workflowCache[wfID]; ok {
+			return wf
+		}
+		wf, err := sc.stores.Workflows.GetByID(ctx, wfID)
+		if err != nil {
+			return nil
+		}
+		workflowCache[wfID] = wf
+		return wf
+	}
+
+	var slaBreaches, staleMarked int
+	for i := range instances {
+		inst := &instances[i]
+
+		// ── SLA check ──
+		stages := getStages(inst.WorkflowID, inst.WorkflowVersion)
+		if stages != nil {
+			sc.checkSLA(ctx, inst, stages)
+		}
+
+		// ── Staleness check ──
+		wf := getWorkflow(inst.WorkflowID)
+		if wf != nil && wf.StalenessTimeoutHours != nil && *wf.StalenessTimeoutHours > 0 {
+			threshold := time.Duration(*wf.StalenessTimeoutHours) * time.Hour
+			if time.Since(inst.UpdatedAt) > threshold {
+				sc.markStale(ctx, inst)
+				staleMarked++
+			}
+		}
+	}
+
+	if slaBreaches > 0 || staleMarked > 0 {
+		log.Printf("[workflow-scanner] cycle: %d SLA breaches, %d stale", slaBreaches, staleMarked)
+	}
+	_ = slaBreaches // used in future if we count from checkSLA
+}
+
+// checkSLA checks whether the current stage has an SLA and whether it's been breached.
+func (sc *Scanner) checkSLA(ctx context.Context, inst *models.WorkflowInstance, stages []models.WorkflowStage) {
+	// Find current stage
+	var currentStage *models.WorkflowStage
+	for i := range stages {
+		if stages[i].Name == inst.CurrentStage {
+			currentStage = &stages[i]
+			break
+		}
+	}
+	if currentStage == nil || currentStage.SLASeconds == nil {
+		return
+	}
+
+	slaDuration := time.Duration(*currentStage.SLASeconds) * time.Second
+	if time.Since(inst.StageEnteredAt) <= slaDuration {
+		return
+	}
+
+	// Check if already breached (avoid re-firing)
+	if metadataHasKey(inst.Metadata, "sla_breached") {
+		return
+	}
+
+	// Mark breached in metadata
+	inst.Metadata = setMetadataKey(inst.Metadata, "sla_breached", true)
+	if err := sc.stores.Workflows.UpdateInstance(ctx, inst); err != nil {
+		log.Printf("[workflow-scanner] update SLA metadata for %s: %v", inst.ID, err)
+		return
+	}
+
+	sc.emit("workflow.sla_breach", inst.ID, map[string]any{
+		"instance_id":   inst.ID,
+		"workflow_id":   inst.WorkflowID,
+		"stage":         inst.CurrentStage,
+		"sla_seconds":   *currentStage.SLASeconds,
+		"entered_at":    inst.StageEnteredAt.Format(time.RFC3339),
+	})
+
+	log.Printf("[workflow-scanner] SLA breach: instance=%s stage=%s (limit=%ds)",
+		inst.ID, inst.CurrentStage, *currentStage.SLASeconds)
+}
+
+// markStale transitions an instance to stale and cancels open assignments.
+func (sc *Scanner) markStale(ctx context.Context, inst *models.WorkflowInstance) {
+	if err := sc.stores.Workflows.MarkInstanceStale(ctx, inst.ID); err != nil {
+		log.Printf("[workflow-scanner] mark stale %s: %v", inst.ID, err)
+		return
+	}
+
+	// Cancel open assignments
+	assignments, _ := sc.stores.Workflows.ListAssignmentsByInstance(ctx, inst.ID)
+	for _, a := range assignments {
+		if a.Status == models.AssignmentStatusUnassigned || a.Status == models.AssignmentStatusClaimed {
+			sc.stores.Workflows.CancelAssignment(ctx, a.ID)
+		}
+	}
+
+	sc.emit("workflow.stale", inst.ID, map[string]any{
+		"instance_id": inst.ID,
+		"workflow_id": inst.WorkflowID,
+		"stage":       inst.CurrentStage,
+	})
+
+	log.Printf("[workflow-scanner] stale: instance=%s", inst.ID)
+}
+
+// emit publishes an event on the bus.
+func (sc *Scanner) emit(label, room string, payload map[string]any) {
+	if sc.bus == nil {
+		return
+	}
+	sc.bus.Publish(events.Event{
+		Label:   label,
+		Room:    room,
+		Payload: events.MustJSON(payload),
+		Ts:      time.Now().UnixMilli(),
+	})
+}
+
+// ── Metadata helpers ────────────────────────
+
+func metadataHasKey(meta json.RawMessage, key string) bool {
+	if len(meta) == 0 {
+		return false
+	}
+	var m map[string]json.RawMessage
+	if json.Unmarshal(meta, &m) != nil {
+		return false
+	}
+	_, ok := m[key]
+	return ok
+}
+
+func setMetadataKey(meta json.RawMessage, key string, val any) json.RawMessage {
+	var m map[string]any
+	if len(meta) > 0 {
+		json.Unmarshal(meta, &m)
+	}
+	if m == nil {
+		m = map[string]any{}
+	}
+	m[key] = val
+	out, _ := json.Marshal(m)
+	return out
+}