core/server/workflow/scanner.go

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
		}
		stages, parseErr := parseSnapshotStages(v.Snapshot)
		if parseErr != 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
}