engine.go

package runnerq

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"math"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/google/uuid"

	"github.com/alob-mtc/runnerq-go/storage"
)

// workerPoolHeartbeatInterval is how often the engine refreshes its
// runnerq_worker_pools row when the backend supports registration. Should be
// roughly 1/6 of the backend's liveness window so a single missed beat
// doesn't drop the pool out of cluster-wide capacity reporting.
const workerPoolHeartbeatInterval = 10 * time.Second

// backoff implements simple exponential backoff for idle polls.
type backoff struct {
	current time.Duration
	base    time.Duration
	max     time.Duration
}

func newBackoff(base, max time.Duration) *backoff {
	return &backoff{current: base, base: base, max: max}
}

func (b *backoff) reset() {
	b.current = b.base
}

func (b *backoff) next() time.Duration {
	n := b.current
	b.current = min(b.current*2, b.max)
	return n
}

// WorkerEngine is the main activity processing engine.
type WorkerEngine struct {
	queue      activityQueue
	backend    storage.Storage
	handlers   map[string]ActivityHandler
	config     WorkerConfig
	running    atomic.Bool
	cancelFunc context.CancelFunc
	shutdownCh chan struct{}
	metrics    MetricsSink
	resultWg   sync.WaitGroup // tracks in-flight result storage goroutines

	poolID uuid.UUID // identity used for worker_pools registration; zero if backend doesn't support it

	// SuspendOnAwait dispatcher state. These are nil and unused when the
	// flag is off; the engine falls back to the fixed-goroutine pool.
	workerSem  chan struct{}  // primary slot pool; capacity = MaxConcurrentActivities - SuspendLeavesReserved
	leafSem    chan struct{}  // reservation pool, only populated when SuspendLeavesReserved>0 && len(SuspendLeafActivityTypes)>0
	leafQueue  activityQueue  // adapter filtered to leaf types (paired with leafSem)
	activityWg sync.WaitGroup // tracks in-flight activity goroutines (suspend mode only)
}

// NewWorkerEngineWithBackend creates a WorkerEngine from a custom backend.
func NewWorkerEngineWithBackend(backend storage.Storage, config WorkerConfig) *WorkerEngine {
	// Propagate lease config to backends that support it.
	if lc, ok := backend.(storage.LeaseConfigurer); ok && config.LeaseMS != nil {
		leaseMS := min(*config.LeaseMS, math.MaxInt64)
		lc.SetLeaseMS(int64(leaseMS))
	}

	adapter := newBackendQueueAdapter(backend, config.ActivityTypes)
	shutdownCh := make(chan struct{})
	return &WorkerEngine{
		queue:      adapter,
		backend:    backend,
		handlers:   make(map[string]ActivityHandler),
		config:     config,
		shutdownCh: shutdownCh,
		metrics:    NoopMetrics{},
	}
}

// SetMetrics sets the metrics sink.
func (e *WorkerEngine) SetMetrics(sink MetricsSink) {
	e.metrics = sink
}

// Inspector returns a QueueInspector for observability operations.
// Import the observability package and use NewQueueInspector(backend) instead
// for decoupled usage.
func (e *WorkerEngine) Backend() storage.Storage {
	return e.backend
}

// MaxConcurrentActivities returns the max workers config for inspector use.
func (e *WorkerEngine) MaxConcurrentActivities() int {
	return e.config.MaxConcurrentActivities
}

// RegisterActivity registers an activity handler for a given activity type.
func (e *WorkerEngine) RegisterActivity(activityType string, handler ActivityHandler) {
	e.handlers[activityType] = handler
}

// GetActivityExecutor returns an ActivityExecutor for orchestrating activities.
// Spawns made through the returned executor are roots (no parent lineage).
func (e *WorkerEngine) GetActivityExecutor() ActivityExecutor {
	return newWorkerEngineWrapperWithDepth(e.queue, e.config.MaxActivityDepth)
}

// Start starts the worker engine and blocks until shutdown or error.
func (e *WorkerEngine) Start(ctx context.Context) error {
	if len(e.config.ActivityTypes) > 0 {
		var missing []string
		for _, t := range e.config.ActivityTypes {
			if _, ok := e.handlers[t]; !ok {
				missing = append(missing, t)
			}
		}
		if len(missing) > 0 {
			panic(fmt.Sprintf("activity_types filter contains types with no registered handler: %v", missing))
		}
	}

	// SuspendOnAwait config sanity. Fail fast on the misconfigurations that
	// would otherwise silently degrade capacity, waste reserved slots, or
	// (worst case) cause this engine's leaf dispatcher to dequeue activities
	// it can't run — stealing work that another engine in the cluster would
	// have handled.
	if e.config.SuspendOnAwait {
		if e.config.SuspendLeavesReserved < 0 {
			return &WorkerError{
				Kind:    ErrConfiguration,
				Message: "SuspendLeavesReserved must be >= 0",
			}
		}
		if e.config.SuspendLeavesReserved >= e.config.MaxConcurrentActivities {
			return &WorkerError{
				Kind: ErrConfiguration,
				Message: fmt.Sprintf(
					"SuspendLeavesReserved (%d) must be < MaxConcurrentActivities (%d); otherwise the primary pool collapses to 1 slot",
					e.config.SuspendLeavesReserved, e.config.MaxConcurrentActivities,
				),
			}
		}
		if e.config.SuspendLeavesReserved > 0 && len(e.config.SuspendLeafActivityTypes) == 0 {
			return &WorkerError{
				Kind:    ErrConfiguration,
				Message: "SuspendLeavesReserved > 0 requires SuspendLeafActivityTypes to be non-empty; otherwise reserved slots become dead capacity",
			}
		}
		var missingLeaf []string
		for _, t := range e.config.SuspendLeafActivityTypes {
			if _, ok := e.handlers[t]; !ok {
				missingLeaf = append(missingLeaf, t)
			}
		}
		if len(missingLeaf) > 0 {
			return &WorkerError{
				Kind: ErrConfiguration,
				Message: fmt.Sprintf(
					"SuspendLeafActivityTypes contains types with no registered handler: %v — leaf dispatcher would dequeue them and mark them handler_not_found, stealing work from other engines",
					missingLeaf,
				),
			}
		}
	}

	if !e.running.CompareAndSwap(false, true) {
		return &WorkerError{Kind: ErrAlreadyRunning}
	}

	slog.Info("Starting worker engine", "max_concurrent_activities", e.config.MaxConcurrentActivities)

	engineCtx, cancel := context.WithCancel(ctx)
	e.cancelFunc = cancel
	e.shutdownCh = make(chan struct{})

	// Register this pool so cluster-wide capacity reporting stays accurate.
	// Registration failure is non-fatal — the engine still runs, the KPI just
	// under-reports until a later heartbeat re-establishes the row.
	e.poolID = uuid.New()
	info := storage.WorkerPoolInfo{
		PoolID:        e.poolID,
		QueueName:     e.config.QueueName,
		MaxWorkers:    e.config.MaxConcurrentActivities,
		ActivityTypes: e.config.ActivityTypes,
	}
	if err := e.backend.RegisterWorkerPool(engineCtx, info); err != nil {
		slog.Warn("Failed to register worker pool", "error", err, "pool_id", e.poolID)
		e.poolID = uuid.Nil
	} else {
		slog.Info("Registered worker pool", "pool_id", e.poolID, "max_workers", info.MaxWorkers)
	}

	var wg sync.WaitGroup

	// Heartbeat the worker_pools row so we keep counting toward cluster capacity.
	if e.poolID != uuid.Nil {
		wg.Go(func() {
			e.runWorkerPoolHeartbeat(engineCtx)
		})
	}

	// Scheduled activities processor (skipped if backend handles it natively)
	if !e.queue.SchedulesNatively() {
		wg.Go(func() {
			e.runScheduledProcessor(engineCtx)
		})
	}

	// Reaper processor
	wg.Go(func() {
		e.runReaperProcessor(engineCtx)
	})

	if e.config.SuspendOnAwait {
		// Semaphore-based dispatcher model. A parent's slot is released
		// while it waits on a child future (see suspend.go) so leaves can
		// keep dequeueing — eliminates the parent-blocking starvation
		// pattern on recursive fan-out workloads.
		primary := max(e.config.MaxConcurrentActivities-e.config.SuspendLeavesReserved, 1)
		e.workerSem = make(chan struct{}, primary)
		wg.Go(func() {
			e.runSuspendDispatcher(engineCtx, e.queue, e.workerSem, "dispatcher")
		})

		// Optional reservation: a separate dispatcher with its own slot
		// pool that dequeues only leaf types, so leaves can always run
		// even when every primary slot holds a suspended parent that's
		// about to wake.
		if e.config.SuspendLeavesReserved > 0 && len(e.config.SuspendLeafActivityTypes) > 0 {
			e.leafSem = make(chan struct{}, e.config.SuspendLeavesReserved)
			e.leafQueue = newBackendQueueAdapter(e.backend, e.config.SuspendLeafActivityTypes)
			wg.Go(func() {
				e.runSuspendDispatcher(engineCtx, e.leafQueue, e.leafSem, "dispatcher-leaf")
			})
		}
	} else {
		// Existing fixed-goroutine pool. One worker per slot.
		for i := 0; i < e.config.MaxConcurrentActivities; i++ {
			workerID := i
			wg.Go(func() {
				e.runWorkerLoop(engineCtx, workerID)
			})
		}
	}

	// Wait for shutdown signal or context cancellation
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

	select {
	case sig := <-sigCh:
		slog.Info("Received shutdown signal", "signal", sig)
	case <-ctx.Done():
		slog.Info("Context cancelled")
	case <-e.shutdownCh:
		slog.Info("Shutdown requested")
	}

	e.stop()
	cancel()

	// Single shutdown grace covering everything in parallel — worker loops,
	// dispatchers, in-flight activity goroutines, result-storage goroutines,
	// and pool deregistration. Previous design ran them sequentially with
	// per-stage timeouts (wg.Wait unbounded + 30s + 10s + 5s) which on a
	// busy engine could push shutdown past a minute and time out the
	// orchestrator's SIGTERM grace. Worst case is now ShutdownGraceSeconds
	// regardless of how many goroutines are still in flight.
	graceSec := uint64(30)
	if e.config.ShutdownGraceSeconds != nil {
		graceSec = max(*e.config.ShutdownGraceSeconds, 1)
	}
	graceCtx, graceCancel := context.WithTimeout(context.Background(), time.Duration(graceSec)*time.Second)
	defer graceCancel()

	type drainTask struct {
		name string
		fn   func()
	}
	tasks := []drainTask{
		// Worker loops, dispatchers, reaper, scheduled processor, heartbeat —
		// everything that wg.Add'd into the supervisor wg above.
		{"workers", func() { wg.Wait() }},
		// In-flight activity goroutines (suspend mode only — non-suspend
		// activities are inside the worker goroutines covered by wg).
		{"activities", func() {
			if e.config.SuspendOnAwait {
				e.activityWg.Wait()
			}
		}},
		// Async result-storage goroutines spawned by handleSuccess /
		// handleNonRetryableFailure.
		{"results", func() { e.resultWg.Wait() }},
		// Best-effort pool deregister — uses graceCtx so it can't outlive
		// the budget on its own.
		{"deregister", func() {
			if e.poolID == uuid.Nil {
				return
			}
			if err := e.backend.DeregisterWorkerPool(graceCtx, e.poolID); err != nil && graceCtx.Err() == nil {
				slog.Warn("Failed to deregister worker pool", "error", err, "pool_id", e.poolID)
			}
		}},
	}

	done := make(chan string, len(tasks))
	for _, t := range tasks {
		t := t
		go func() {
			t.fn()
			done <- t.name
		}()
	}

	finished := 0
	for finished < len(tasks) {
		select {
		case name := <-done:
			slog.Debug("Shutdown drain complete", "stage", name)
			finished++
		case <-graceCtx.Done():
			pending := len(tasks) - finished
			slog.Warn("Shutdown grace exceeded; returning with drains in flight",
				"grace_seconds", graceSec, "pending_drains", pending)
			signal.Stop(sigCh)
			return nil
		}
	}

	signal.Stop(sigCh)
	slog.Info("Worker engine stopped")
	return nil
}

// Stop initiates a graceful shutdown of the worker engine.
func (e *WorkerEngine) Stop() {
	e.stop()
}

func (e *WorkerEngine) stop() {
	slog.Info("Stopping worker engine")
	e.running.Store(false)
	if e.cancelFunc != nil {
		e.cancelFunc()
	}
	select {
	case <-e.shutdownCh:
	default:
		close(e.shutdownCh)
	}
}

func (e *WorkerEngine) runWorkerLoop(ctx context.Context, workerID int) {
	slog.Debug("Starting worker loop", "worker_id", workerID)
	workerLabel := fmt.Sprintf("worker-%d", workerID)
	bo := newBackoff(100*time.Millisecond, 5*time.Second)

	for e.running.Load() {
		select {
		case <-ctx.Done():
			slog.Debug("Worker loop stopped (context)", "worker_id", workerID)
			return
		default:
		}

		act, err := e.queue.Dequeue(ctx, time.Second, workerLabel)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			slog.Error("Failed to dequeue activity", "worker_id", workerID, "error", err)
			select {
			case <-time.After(time.Second):
			case <-ctx.Done():
				return
			}
			continue
		}

		if act == nil {
			sleepFor := bo.next()
			select {
			case <-time.After(sleepFor):
			case <-ctx.Done():
				return
			}
			continue
		}

		bo.reset()
		e.processActivity(ctx, act, workerLabel, workerID)
	}

	slog.Debug("Worker loop stopped", "worker_id", workerID)
}

// runSuspendDispatcher is the SuspendOnAwait equivalent of runWorkerLoop.
// One dispatcher goroutine acquires a slot from sem, dequeues, and hands the
// activity off to a per-activity goroutine that owns the slot via a
// slotHolder. The slotHolder is installed in the activity's context so that
// ActivityFuture.GetResult inside the handler can release the slot for the
// duration of the wait — that's the whole point of SuspendOnAwait.
//
// Two dispatchers can run concurrently against different (queue, sem) pairs
// to implement leaf-slot reservation; they share the engine's activityWg so
// shutdown waits for both.
func (e *WorkerEngine) runSuspendDispatcher(ctx context.Context, q activityQueue, sem chan struct{}, label string) {
	slog.Debug("Starting suspend dispatcher", "label", label, "slots", cap(sem))
	bo := newBackoff(100*time.Millisecond, 5*time.Second)

	for e.running.Load() {
		// Acquire a slot before dequeueing — keeps in-flight activities
		// bounded by the semaphore capacity even under bursty load.
		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
			slog.Debug("Suspend dispatcher stopped (context)", "label", label)
			return
		}

		act, err := q.Dequeue(ctx, time.Second, label)
		if err != nil {
			<-sem
			if ctx.Err() != nil {
				return
			}
			slog.Error("Suspend dispatcher dequeue failed", "label", label, "error", err)
			select {
			case <-time.After(time.Second):
			case <-ctx.Done():
				return
			}
			continue
		}

		if act == nil {
			<-sem
			sleepFor := bo.next()
			select {
			case <-time.After(sleepFor):
			case <-ctx.Done():
				return
			}
			continue
		}

		bo.reset()
		holder := newSlotHolder(sem)
		e.activityWg.Add(1)
		go func(act *activity) {
			defer e.activityWg.Done()
			defer holder.release() // safety net — release() is idempotent

			// MUST pass the same label that was sent to Dequeue above. The
			// AckSuccess/AckFailure SQL guards on `current_worker_id = $2`,
			// which was set to `label` at dequeue time. Using a per-activity
			// label here (e.g. "<label>-<short-uuid>") would always fail
			// the WHERE clause: handler completes, MarkCompleted returns
			// "not in a claimable state", handleSuccess logs
			// activity_lost_completion, the reaper requeues the row, and
			// the activity re-runs forever. Activity identity for slog is
			// already covered by the activity_id field on every log line
			// processActivity emits.
			ctxWithSlot := withSuspendSlot(ctx, holder)
			e.processActivity(ctxWithSlot, act, label, 0)
		}(act)
	}
	slog.Debug("Suspend dispatcher stopped", "label", label)
}

func (e *WorkerEngine) processActivity(ctx context.Context, act *activity, workerLabel string, workerID int) {
	activityID := act.ID
	activityType := act.ActivityType

	slog.Debug("Worker processing activity", "worker_id", workerID, "activity_id", activityID, "activity_type", activityType)

	handler, ok := e.handlers[activityType]
	if !ok {
		slog.Error("No handler found for activity type", "worker_id", workerID, "activity_id", activityID, "activity_type", activityType)
		if _, err := e.queue.MarkFailed(ctx, act, "handler_not_found", false, workerLabel); err != nil {
			slog.Error("Failed to mark activity as failed", "worker_id", workerID, "activity_id", activityID, "error", err)
		}
		return
	}

	activityTimeout := time.Duration(act.TimeoutSeconds) * time.Second
	timeoutCtx, timeoutCancel := context.WithTimeout(ctx, activityTimeout)
	defer timeoutCancel()

	scopedExecutor := newWorkerEngineWrapperWithDepth(e.queue, e.config.MaxActivityDepth).scopedForChild(act)

	actCtx := ActivityContext{
		ActivityID:       activityID,
		ActivityType:     activityType,
		RetryCount:       act.RetryCount,
		Metadata:         act.Metadata,
		Ctx:              timeoutCtx,
		ActivityExecutor: scopedExecutor,
		ParentActivityID: act.ParentActivityID,
		RootActivityID:   act.RootActivityID,
		Depth:            act.Depth,
	}

	payloadForDL := make(json.RawMessage, len(act.Payload))
	copy(payloadForDL, act.Payload)

	result, handlerErr := e.safeHandle(handler, actCtx, act.Payload)

	// Check if we timed out
	if timeoutCtx.Err() == context.DeadlineExceeded {
		e.handleTimeout(ctx, act, handler, actCtx, payloadForDL, workerLabel, workerID, activityID, activityType, activityTimeout)
		return
	}

	if handlerErr == nil {
		e.handleSuccess(ctx, act, result, workerLabel, workerID, activityID, activityType)
		return
	}

	retryable := true // default: unknown errors are retryable
	if re, ok := handlerErr.(RetryableError); ok {
		retryable = re.IsRetryable()
	}

	if retryable {
		e.handleRetryableFailure(ctx, act, handler, actCtx, payloadForDL, handlerErr.Error(), workerLabel, workerID, activityID, activityType)
	} else {
		e.handleNonRetryableFailure(ctx, act, handlerErr.Error(), workerLabel, workerID, activityID, activityType)
	}
}

// safeHandle calls the handler with panic recovery.
func (e *WorkerEngine) safeHandle(handler ActivityHandler, ctx ActivityContext, payload json.RawMessage) (result json.RawMessage, err error) {
	defer func() {
		if r := recover(); r != nil {
			var errMsg string
			switch v := r.(type) {
			case string:
				errMsg = fmt.Sprintf("panic: %s", v)
			case error:
				errMsg = fmt.Sprintf("panic: %s", v.Error())
			default:
				errMsg = "panic (unknown)"
			}
			err = NewRetryError(errMsg)
		}
	}()
	return handler.Handle(ctx, payload)
}

func (e *WorkerEngine) handleSuccess(ctx context.Context, act *activity, result json.RawMessage, workerLabel string, workerID int, activityID any, activityType string) {
	if err := e.queue.MarkCompleted(ctx, act, workerLabel); err != nil {
		// MarkCompleted failed — the row was no longer in 'processing' with
		// this worker (lease likely expired and the reaper requeued it).
		// Don't claim success: another worker will rerun the activity.
		// Don't store the result either, since the rerun will produce its own.
		slog.Warn("Activity completed in handler but row was no longer claimable; another worker will rerun it",
			"worker_id", workerID, "activity_id", activityID, "activity_type", activityType, "error", err)
		e.metrics.IncCounter("activity_lost_completion", 1)
		return
	}
	e.metrics.IncCounter("activity_completed", 1)
	slog.Info("Activity completed successfully", "worker_id", workerID, "activity_id", activityID, "activity_type", activityType)

	// Async result storage — tracked for graceful shutdown
	e.resultWg.Go(func() {
		storeCtx, storeCancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer storeCancel()
		res := activityResult{Data: result, State: ResultOk}
		if err := e.queue.StoreResult(storeCtx, act.ID, res); err != nil {
			slog.Error("Failed to store activity result", "activity_id", activityID, "error", err)
		}
	})
}

func (e *WorkerEngine) handleRetryableFailure(ctx context.Context, act *activity, handler ActivityHandler, actCtx ActivityContext, payloadForDL json.RawMessage, reason string, workerLabel string, workerID int, activityID any, activityType string) {
	e.metrics.IncCounter("activity_retry", 1)
	slog.Warn("Activity requesting retry", "worker_id", workerID, "activity_id", activityID, "activity_type", activityType, "reason", reason)

	deadLettered, err := e.queue.MarkFailed(ctx, act, reason, true, workerLabel)
	if err != nil {
		slog.Error("Failed to mark activity for retry", "worker_id", workerID, "activity_id", activityID, "error", err)
		return
	}
	if deadLettered {
		dlCtx := ActivityContext{
			ActivityID:       act.ID,
			ActivityType:     activityType,
			RetryCount:       0,
			Metadata:         make(map[string]string),
			Ctx:              ctx,
			ActivityExecutor: newWorkerEngineWrapperWithDepth(e.queue, e.config.MaxActivityDepth).scopedForChild(act),
			ParentActivityID: act.ParentActivityID,
			RootActivityID:   act.RootActivityID,
			Depth:            act.Depth,
		}
		handler.OnDeadLetter(dlCtx, payloadForDL, reason)
	}
}

func (e *WorkerEngine) handleNonRetryableFailure(ctx context.Context, act *activity, reason string, workerLabel string, workerID int, activityID any, activityType string) {
	e.metrics.IncCounter("activity_failed_non_retry", 1)
	slog.Error("Activity failed", "worker_id", workerID, "activity_id", activityID, "activity_type", activityType, "reason", reason)

	if _, err := e.queue.MarkFailed(ctx, act, reason, false, workerLabel); err != nil {
		slog.Error("Failed to mark activity as failed", "worker_id", workerID, "activity_id", activityID, "error", err)
	}

	// Async result storage — tracked for graceful shutdown
	e.resultWg.Go(func() {
		storeCtx, storeCancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer storeCancel()
		errorResult, _ := json.Marshal(map[string]any{
			"error":     reason,
			"type":      "non_retryable",
			"failed_at": time.Now().UTC().Format(time.RFC3339),
		})
		res := activityResult{Data: errorResult, State: ResultErr}
		if err := e.queue.StoreResult(storeCtx, act.ID, res); err != nil {
			slog.Error("Failed to store activity result", "activity_id", activityID, "error", err)
		}
	})
}

func (e *WorkerEngine) handleTimeout(ctx context.Context, act *activity, handler ActivityHandler, actCtx ActivityContext, payloadForDL json.RawMessage, workerLabel string, workerID int, activityID any, activityType string, timeout time.Duration) {
	e.metrics.IncCounter("activity_timeout", 1)
	errorMsg := "Activity execution timed out"
	slog.Error("Activity timed out", "worker_id", workerID, "activity_id", activityID, "activity_type", activityType, "timeout", timeout)

	deadLettered, err := e.queue.MarkFailed(ctx, act, errorMsg, true, workerLabel)
	if err != nil {
		slog.Error("Failed to mark activity as failed", "worker_id", workerID, "activity_id", activityID, "error", err)
		return
	}
	if deadLettered {
		dlCtx := ActivityContext{
			ActivityID:       act.ID,
			ActivityType:     activityType,
			RetryCount:       0,
			Metadata:         make(map[string]string),
			Ctx:              ctx,
			ActivityExecutor: newWorkerEngineWrapperWithDepth(e.queue, e.config.MaxActivityDepth).scopedForChild(act),
			ParentActivityID: act.ParentActivityID,
			RootActivityID:   act.RootActivityID,
			Depth:            act.Depth,
		}
		handler.OnDeadLetter(dlCtx, payloadForDL, errorMsg)
	}
}

func (e *WorkerEngine) runScheduledProcessor(ctx context.Context) {
	pollInterval := uint64(5)
	if e.config.SchedulePollIntervalSeconds != nil {
		pollInterval = max(*e.config.SchedulePollIntervalSeconds, 1)
	}
	ticker := time.NewTicker(time.Duration(pollInterval) * time.Second)
	defer ticker.Stop()

	slog.Debug("Starting scheduled activities processor")
	for e.running.Load() {
		select {
		case <-ctx.Done():
			slog.Debug("Scheduled activities processor stopped")
			return
		case <-ticker.C:
			if _, err := e.queue.ProcessScheduledActivities(ctx); err != nil {
				slog.Error("Failed to process scheduled activities", "error", err)
			}
		}
	}
	slog.Debug("Scheduled activities processor stopped")
}

// runWorkerPoolHeartbeat keeps this engine's runnerq_worker_pools row marked
// as alive so the cluster-wide MaxWorkers reported by Stats() stays accurate.
// A failure to heartbeat is logged but doesn't shut down the engine — the row
// will simply age out of the liveness window until the next successful beat.
func (e *WorkerEngine) runWorkerPoolHeartbeat(ctx context.Context) {
	ticker := time.NewTicker(workerPoolHeartbeatInterval)
	defer ticker.Stop()
	for e.running.Load() {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := e.backend.HeartbeatWorkerPool(ctx, e.poolID); err != nil {
				slog.Warn("Worker pool heartbeat failed", "error", err, "pool_id", e.poolID)
			}
		}
	}
}

func (e *WorkerEngine) runReaperProcessor(ctx context.Context) {
	intervalSec := uint64(5)
	if e.config.ReaperIntervalSeconds != nil {
		intervalSec = max(*e.config.ReaperIntervalSeconds, 1)
	}
	batchSize := 100
	if e.config.ReaperBatchSize != nil {
		batchSize = *e.config.ReaperBatchSize
	}

	ticker := time.NewTicker(time.Duration(intervalSec) * time.Second)
	defer ticker.Stop()

	slog.Debug("Starting reaper processor")
	for e.running.Load() {
		select {
		case <-ctx.Done():
			slog.Debug("Reaper processor stopped")
			return
		case <-ticker.C:
			if _, err := e.queue.RequeueExpired(ctx, batchSize); err != nil {
				slog.Error("Reaper failed to requeue expired items", "error", err)
			}
		}
	}
	slog.Debug("Reaper processor stopped")
}

// WorkerEngineBuilder provides fluent configuration for WorkerEngine.
type WorkerEngineBuilder struct {
	queueName        *string
	maxWorkers       *int
	pollInterval     *time.Duration
	metrics          MetricsSink
	backend          storage.Storage
	activityTypes    []string
	suspendOnAwait   bool
	suspendLeafTypes []string
	suspendLeavesRes int
	shutdownGrace    *time.Duration
}

// Builder creates a new WorkerEngineBuilder.
func Builder() *WorkerEngineBuilder {
	return &WorkerEngineBuilder{}
}

// QueueName sets the queue name.
func (b *WorkerEngineBuilder) QueueName(name string) *WorkerEngineBuilder {
	b.queueName = &name
	return b
}

// MaxWorkers sets the maximum concurrent workers.
func (b *WorkerEngineBuilder) MaxWorkers(max int) *WorkerEngineBuilder {
	b.maxWorkers = &max
	return b
}

// SchedulePollInterval sets the interval for polling scheduled activities.
func (b *WorkerEngineBuilder) SchedulePollInterval(interval time.Duration) *WorkerEngineBuilder {
	b.pollInterval = &interval
	return b
}

// ActivityTypes restricts this engine to only dequeue the specified types.
func (b *WorkerEngineBuilder) ActivityTypes(types []string) *WorkerEngineBuilder {
	b.activityTypes = types
	return b
}

// Metrics sets the metrics sink.
func (b *WorkerEngineBuilder) Metrics(sink MetricsSink) *WorkerEngineBuilder {
	b.metrics = sink
	return b
}

// Backend sets the storage backend. Required.
func (b *WorkerEngineBuilder) Backend(backend storage.Storage) *WorkerEngineBuilder {
	b.backend = backend
	return b
}

// SuspendOnAwait toggles the semaphore-based dispatcher that releases a
// parent activity's worker slot while it waits for child futures. Default
// false (existing fixed-goroutine pool). See WorkerConfig.SuspendOnAwait for
// the operational notes.
func (b *WorkerEngineBuilder) SuspendOnAwait(enabled bool) *WorkerEngineBuilder {
	b.suspendOnAwait = enabled
	return b
}

// ReserveSlotsForLeaves reserves `n` slots out of MaxWorkers for the listed
// "leaf" activity types when SuspendOnAwait is on. Prevents the wake-up
// deadlock where every freed slot is immediately retaken by a waking parent
// and no leaf can run. No effect when SuspendOnAwait is off.
func (b *WorkerEngineBuilder) ReserveSlotsForLeaves(n int, leafTypes []string) *WorkerEngineBuilder {
	b.suspendLeavesRes = n
	b.suspendLeafTypes = leafTypes
	return b
}

// ShutdownGrace bounds the entire shutdown drain — workers, dispatchers,
// in-flight activity goroutines, result-storage goroutines, and pool
// deregistration all run in parallel under this single budget. When the
// budget expires Start() returns even if some goroutines remain in flight.
// Defaults to 30s when not set; clamped to >= 1s.
func (b *WorkerEngineBuilder) ShutdownGrace(d time.Duration) *WorkerEngineBuilder {
	b.shutdownGrace = &d
	return b
}

// Build creates the WorkerEngine with configured settings.
func (b *WorkerEngineBuilder) Build() (*WorkerEngine, error) {
	maxConcurrent := 10
	if b.maxWorkers != nil {
		maxConcurrent = *b.maxWorkers
	}
	pollIntervalSec := uint64(5)
	if b.pollInterval != nil {
		pollIntervalSec = uint64(b.pollInterval.Seconds())
	}

	if b.backend == nil {
		return nil, &WorkerError{
			Kind:    ErrConfiguration,
			Message: "No backend configured. Call .Backend(yourBackend) before .Build(). Use PostgresBackend for PostgreSQL.",
		}
	}

	queueName := "default"
	if b.queueName != nil {
		queueName = *b.queueName
	}

	leaseMS := uint64(60_000)
	reaperInterval := uint64(5)
	reaperBatch := 100

	config := WorkerConfig{
		QueueName:                   queueName,
		MaxConcurrentActivities:     maxConcurrent,
		SchedulePollIntervalSeconds: &pollIntervalSec,
		LeaseMS:                     &leaseMS,
		ReaperIntervalSeconds:       &reaperInterval,
		ReaperBatchSize:             &reaperBatch,
		ActivityTypes:               b.activityTypes,
		SuspendOnAwait:              b.suspendOnAwait,
		SuspendLeafActivityTypes:    b.suspendLeafTypes,
		SuspendLeavesReserved:       b.suspendLeavesRes,
	}
	if b.shutdownGrace != nil {
		secs := uint64((*b.shutdownGrace).Seconds())
		if secs == 0 {
			secs = 1
		}
		config.ShutdownGraceSeconds = &secs
	}

	engine := NewWorkerEngineWithBackend(b.backend, config)
	if b.metrics != nil {
		engine.SetMetrics(b.metrics)
	}

	return engine, nil
}