package models

import (
	"context"
	"log"
	"time"
)

// DeadWorkerHeartbeatTimeout is the threshold for ReapDeadWorkers — a worker
// whose last_seen is older than this is considered dead and any leased tasks
// it owns are reassigned to the pool. Five minutes mirrors the StaleWorkers()
// check in check_jobs.go so the two reapers cannot disagree about who is
// dead. See docs/todo.md Phase 4 §5.
const DeadWorkerHeartbeatTimeout = 5 * time.Minute

// ReapDeadWorkers marks any non-dead worker whose last_seen is older than
// DeadWorkerHeartbeatTimeout as "dead", then reassigns its leased tasks
// back to the queued pool so other workers (or freshly registered ones)
// can pick them up. It mirrors the structure of ReapExpiredTasks — two
// short UPDATE statements, cheap enough to run from the web process
// every 30s.
//
// The returned tuple is (reaped, reassigned, err): reaped counts the
// workers that flipped to dead during this call; reassigned counts the
// leased tasks that were given back to the pool. A zero count on either
// is normal — the reaper is idempotent and the call is silent when
// nothing is due.
func ReapDeadWorkers() (reaped int, reassigned int, err error) {
	now := time.Now()
	cutoff := now.Add(-DeadWorkerHeartbeatTimeout)

	// First flip the workers to dead so the second UPDATE can match the
	// freshly-stamped ids without having to re-derive them in Go.
	r := DB().Exec(`
		UPDATE worker_nodes
		SET status = ?, updated_at = ?
		WHERE status <> ? AND last_seen IS NOT NULL AND last_seen < ?`,
		"dead", now, "dead", cutoff,
	)
	if r.Error != nil {
		return 0, 0, r.Error
	}
	reaped = int(r.RowsAffected)

	// Nothing flipped → no tasks to return. Cheaper than running an
	// UPDATE that touches 0 rows on every tick when the fleet is healthy.
	if reaped == 0 {
		return 0, 0, nil
	}

	// Second: clear any leased tasks owned by the now-dead workers.
	// The selector stashes the worker's WorkerID string in lease_owner;
	// matching against the (now-stale) worker row's WorkerID is the
	// same identifier the selector uses, so we don't need an extra
	// join. Tasks in other states (queued, succeeded, dead, …) are
	// unaffected — only leased work the dead worker still owned has
	// to go back to the queue.
	r2 := DB().Exec(`
		UPDATE tasks
		SET state = ?, lease_owner = '', lease_expires_at = NULL, updated_at = ?
		WHERE state = ? AND lease_owner IN (
			SELECT worker_id FROM worker_nodes WHERE status = ?
		)`,
		TaskStateQueued, now, TaskStateLeased, "dead",
	)
	if r2.Error != nil {
		return reaped, 0, r2.Error
	}
	reassigned = int(r2.RowsAffected)
	return reaped, reassigned, nil
}

// StartDeadWorkerReaper launches a goroutine that runs ReapDeadWorkers
// on the given interval until ctx is canceled. Mirrors StartTaskReaper
// in this package: same ticker pattern, same logging style, same
// recover() safety net so a malformed row cannot crash the web process.
//
// The default interval is 30s; values <= 0 fall back to the default so
// the helper is safe to call from any call site without a guard. Wire
// from main.init() once per process — the reaper uses short row-level
// locks and is cheap under load (two indexed UPDATEs of <= a few
// hundred rows in steady state).
//
// Passing a nil context falls back to context.Background() so callers
// can write `models.StartDeadWorkerReaper(nil, ...)` in one-liners
// (main, tests) without having to import "context" first.
func StartDeadWorkerReaper(ctx context.Context, interval time.Duration) {
	if interval <= 0 {
		interval = 30 * time.Second
	}
	if ctx == nil {
		ctx = context.Background()
	}
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				func() {
					defer func() {
						if r := recover(); r != nil {
							log.Printf("dead_worker_reaper: panic recovered: %v", r)
						}
					}()
					reaped, reassigned, err := ReapDeadWorkers()
					if err != nil {
						log.Printf("dead_worker_reaper: error: %v", err)
						return
					}
					if reaped > 0 || reassigned > 0 {
						log.Printf("dead_worker_reaper: reaped=%d reassigned=%d", reaped, reassigned)
					}
				}()
			}
		}
	}()
}