agentic-coding-workflow/internal/jobs/worker.go

package jobs

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"github.com/yan1h/agent-coding-workflow/internal/infra/clock"
)

// WorkerOptions configures a single Worker. ID identifies the worker in
// jobs.leased_by. Backoff is the per-attempt retry delay sequence; if attempts
// exceeds len(Backoff), the last entry is used.
type WorkerOptions struct {
	ID           string
	PollInterval time.Duration
	Backoff      []time.Duration
	Clock        clock.Clock
	Log          *slog.Logger
	// OnDeadLetter, if non-nil, is invoked once after a job is moved to dead-letter
	// (either because IsPermanent(err) or attempts exhausted). The hook runs in
	// the worker goroutine; panics are recovered and logged but do not stop the
	// worker. Use this for fallback admin notifications when external delivery
	// fails permanently.
	OnDeadLetter func(ctx context.Context, job Job, err error)
}

// Worker pulls leased jobs from the repository, dispatches them to handlers,
// and writes back the appropriate completion / retry / dead-letter state.
type Worker struct {
	repo     Repository
	handlers map[JobType]Handler
	opts     WorkerOptions
	wg       sync.WaitGroup
}

// NewWorker builds a Worker. The caller registers handlers by JobType.
func NewWorker(repo Repository, handlers map[JobType]Handler, opts WorkerOptions) *Worker {
	if opts.PollInterval <= 0 {
		opts.PollInterval = 100 * time.Millisecond
	}
	if opts.Clock == nil {
		opts.Clock = clock.Real()
	}
	if opts.Log == nil {
		opts.Log = slog.Default()
	}
	if len(opts.Backoff) == 0 {
		opts.Backoff = []time.Duration{30 * time.Second}
	}
	return &Worker{repo: repo, handlers: handlers, opts: opts}
}

// Run loops until ctx is canceled, polling for and processing jobs.
func (w *Worker) Run(ctx context.Context) {
	w.wg.Add(1)
	defer w.wg.Done()
	t := time.NewTicker(w.opts.PollInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			w.pollOnce(ctx)
		}
	}
}

// Wait blocks until the goroutine started by Run returns.
func (w *Worker) Wait() { w.wg.Wait() }

func (w *Worker) pollOnce(ctx context.Context) {
	job, err := w.repo.Lease(ctx, w.opts.ID)
	if err != nil {
		w.opts.Log.Error("job.lease_failed", "err", err.Error())
		return
	}
	if job == nil {
		return
	}
	w.execute(ctx, *job)
}

func (w *Worker) execute(ctx context.Context, job Job) {
	w.opts.Log.Info("job.start", "job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts)
	start := w.opts.Clock.Now()

	h, ok := w.handlers[job.Type]
	if !ok {
		if derr := w.repo.DeadLetter(ctx, job.ID, fmt.Sprintf("unknown job type: %s", job.Type)); derr != nil {
			w.opts.Log.Error("job.dead_letter_write_failed", "job_id", job.ID, "err", derr.Error())
		}
		w.opts.Log.Error("job.dead_letter", "job_id", job.ID, "job_type", job.Type, "reason", "unknown_type")
		w.fireOnDeadLetter(ctx, job, errors.New("unknown job type: "+string(job.Type)))
		return
	}

	err := w.runHandler(ctx, h, job)
	if err == nil {
		if cerr := w.repo.Complete(ctx, job.ID); cerr != nil {
			w.opts.Log.Error("job.complete_write_failed", "job_id", job.ID, "err", cerr.Error())
		}
		w.opts.Log.Info("job.complete",
			"job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts,
			"duration_ms", w.opts.Clock.Now().Sub(start).Milliseconds())
		return
	}

	// Permanent or attempts exhausted -> dead-letter
	if IsPermanent(err) || job.Attempts >= job.MaxAttempts {
		msg := truncate(err.Error(), 2000)
		if derr := w.repo.DeadLetter(ctx, job.ID, msg); derr != nil {
			w.opts.Log.Error("job.dead_letter_write_failed", "job_id", job.ID, "err", derr.Error())
		}
		w.opts.Log.Error("job.dead_letter",
			"job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts, "err", msg)
		w.fireOnDeadLetter(ctx, job, err)
		return
	}

	// Retryable
	delay := backoffFor(w.opts.Backoff, int(job.Attempts))
	next := w.opts.Clock.Now().Add(delay)
	msg := truncate(err.Error(), 2000)
	if ferr := w.repo.FailWithRetry(ctx, job.ID, next, msg); ferr != nil {
		w.opts.Log.Error("job.fail_retry_write_failed", "job_id", job.ID, "err", ferr.Error())
	}
	w.opts.Log.Warn("job.fail_retry",
		"job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts,
		"err", msg, "next_at", next.Format(time.RFC3339))
}

// runHandler wraps Handle with panic recovery, converting panics into retryable errors.
func (w *Worker) runHandler(ctx context.Context, h Handler, job Job) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("handler panic: %v", r)
		}
	}()
	return h.Handle(ctx, job)
}

// backoffFor returns the backoff for attempt index (1-based "after first try").
// If index exceeds the slice, the last element is used.
func backoffFor(seq []time.Duration, attempt int) time.Duration {
	if attempt <= 0 {
		return seq[0]
	}
	idx := attempt - 1
	if idx >= len(seq) {
		idx = len(seq) - 1
	}
	return seq[idx]
}

// fireOnDeadLetter invokes opts.OnDeadLetter, if set, recovering any panics.
func (w *Worker) fireOnDeadLetter(ctx context.Context, job Job, err error) {
	if w.opts.OnDeadLetter == nil {
		return
	}
	defer func() {
		if r := recover(); r != nil {
			w.opts.Log.Error("job.dead_letter_hook_panic", "job_id", job.ID, "panic", r)
		}
	}()
	w.opts.OnDeadLetter(ctx, job, err)
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n]
}