oikos/internal/execworker/worker.go

// Package execworker processes pending executions as a background daemon.
// This provides a Postgres-backed queue: executions survive restarts, and
// per-execution advisory locks prevent duplicate processing across instances.
//
// Since Phase 4 of the hexagonal refactor, the dispatch logic lives in
// ExecutionService.DispatchQueued (over CommandExecutor + TargetResolver
// ports); this package is a thin poller adapter that claims queued rows and
// delegates to the service.
package execworker

import (
	"context"
	"log/slog"
	"time"

	"github.com/dtoro/oikos/internal/adapters/postgres"
	"github.com/dtoro/oikos/internal/adapters/remote"
	"github.com/dtoro/oikos/internal/adapters/ssh"
	"github.com/dtoro/oikos/internal/config"
	"github.com/dtoro/oikos/internal/core/app"
	"github.com/dtoro/oikos/internal/core/domain"
	"github.com/dtoro/oikos/internal/health"
	"github.com/google/uuid"
)

// Run starts the execution worker loop. Blocks until ctx is cancelled.
func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
	slog.Info("execworker: starting")

	// Build the ExecutionService with the same wiring as cmd/oikos/main.go.
	executor := ssh.NewExecutor(ssh.FileSignerSource(), 5*time.Minute)
	resolver := remote.NewResolver(pool)
	policySvc := app.NewPolicyService(db.NewGovernanceRepo(pool))
	execSvc := app.NewExecutionService(policySvc, executor, resolver, db.NewExecRunRepo(pool))

	probe := health.New(2 * time.Minute)
	probe.Serve(ctx, cfg.HealthListen)

	recoverOrphaned(ctx, pool)
	probe.Bump()

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

	for {
		select {
		case <-ctx.Done():
			slog.Info("execworker: shutting down")
			return
		case <-ticker.C:
			processPending(ctx, pool, execSvc)
			probe.Bump()
		}
	}
}

// recoverOrphaned marks executions stuck in 'running' as failed.
func recoverOrphaned(ctx context.Context, pool *db.Pool) {
	tag, err := pool.Exec(ctx, `UPDATE executions SET status = 'failed', result = '{"error":"worker restarted while execution was running"}'::jsonb, completed_at = now() WHERE status = 'running'`)
	if err != nil {
		slog.Error("execworker: recover orphaned", "error", err)
		return
	}
	if tag.RowsAffected() > 0 {
		slog.Warn("execworker: recovered orphaned executions", "count", tag.RowsAffected())
	}
}

// processPending polls for pending executions and dispatches them via
// ExecutionService.DispatchQueued.
func processPending(ctx context.Context, pool *db.Pool, execSvc *app.ExecutionService) {
	rows, err := pool.Query(ctx, `
		SELECT e.entity_id, COALESCE(t.slug, '') AS target_slug, e.action
		FROM executions e
		LEFT JOIN entities t ON t.id = e.target_entity_id
		WHERE e.status = 'proposed'
		ORDER BY e.created_at ASC
		LIMIT 10`)
	if err != nil {
		slog.Error("execworker: query pending", "error", err)
		return
	}
	defer rows.Close()

	for rows.Next() {
		var execID uuid.UUID
		var targetSlug, action string
		if err := rows.Scan(&execID, &targetSlug, &action); err != nil {
			slog.Error("execworker: scan row", "error", err)
			continue
		}

		// At-most-once: try advisory lock on execution entity_id.
		lockKey := hashUUID(execID)
		lockConn, err := pool.Acquire(ctx)
		if err != nil {
			slog.Error("execworker: acquire lock conn", "error", err)
			continue
		}
		var locked bool
		if err := lockConn.QueryRow(ctx, "SELECT pg_try_advisory_lock($1)", lockKey).Scan(&locked); err != nil || !locked {
			lockConn.Release()
			continue
		}

		// Release the lock on the same connection even if the dispatch
		// panics — an un-released advisory lock would permanently orphan the
		// execution (every future worker skips it at pg_try_advisory_lock).
		func() {
			defer func() {
				if r := recover(); r != nil {
					slog.Error("execworker: panic dispatching execution", "execution", execID, "panic", r)
				}
				lockConn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", lockKey)
				lockConn.Release()
			}()

			_, dispatchErr := execSvc.DispatchQueued(ctx, domain.UUID(execID.String()), targetSlug, action)
			if dispatchErr != nil {
				slog.Error("execworker: dispatch failed", "execution", execID, "error", dispatchErr)
			}
		}()
	}
}

func hashUUID(id uuid.UUID) int {
	h := 0
	for _, b := range id {
		h = (h*31 + int(b)) & 0x7fffffff
	}
	return h
}