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
	// HandlerTimeout bounds a single handler invocation. It must be set slightly
	// below the JobReaper's LeaseTimeout so a healthy long-running handler
	// finishes (or is canceled) before the reaper re-leases the job to another
	// worker and double-runs it. When <= 0 no per-job timeout is applied (the
	// module always sets it from JobReaper.LeaseTimeout when reaping is enabled).
	HandlerTimeout time.Duration
	// 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, w.opts.ID, fmt.Sprintf("unknown job type: %s", job.Type)); w.leaseLost(derr, job, "dead_letter") {
			return
		} else if 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, w.opts.ID); w.leaseLost(cerr, job, "complete") {
			return
		} else if 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, w.opts.ID, msg); w.leaseLost(derr, job, "dead_letter") {
			return
		} else if 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, w.opts.ID, next, msg); w.leaseLost(ferr, job, "fail_retry") {
		return
	} else if 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))
}

// leaseLost reports whether writeErr is ErrLeaseLost. The fencing predicate on
// the terminal/retry updates matched no row, meaning this worker's lease was
// stolen (e.g. its handler outran LeaseTimeout and the reaper re-leased the
// job). The write is correctly dropped: we only log and drop, and must NOT
// treat it as a failure that schedules further work — the new run owns the job.
func (w *Worker) leaseLost(writeErr error, job Job, phase string) bool {
	if errors.Is(writeErr, ErrLeaseLost) {
		w.opts.Log.Warn("job.lease_lost",
			"job_id", job.ID, "job_type", job.Type, "attempt", job.Attempts, "phase", phase)
		return true
	}
	return false
}

// runHandler wraps Handle with panic recovery (converting panics into retryable
// errors) and bounds the handler to HandlerTimeout. Deriving a per-job timeout
// below the reaper's LeaseTimeout ensures a healthy long-running handler is
// canceled before the reaper re-leases the job, so it is never double-run.
func (w *Worker) runHandler(ctx context.Context, h Handler, job Job) (err error) {
	if w.opts.HandlerTimeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, w.opts.HandlerTimeout)
		defer cancel()
	}
	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]
}