// The timers table: scheduled work materialized per (host, name) — spec
// docs/superpowers/specs/2026-08-23-scheduled-work-design.md §5. Rows come
// from two writers with disjoint columns: converge (schedule.Converge) and
// the control socket own spec/path/kind; the runner owns next_at after the
// first fire, last_*, attempts. UpsertTimer encodes that split so neither
// writer can clobber the other. Deliberately NOT in Fingerprint(): a timer
// changes nothing about routing.
package registry

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"github.com/carlosframework/platform/internal/hostwrite"
	"os"
	"sort"
	"strings"
)

type Timer struct {
	ClaimOwnership

	Host, Name, Kind, Spec, Path string
	NextAt, LastAt               int64
	LastStatus                   string
	LastMs                       int64
	Attempts                     int
	CreatedAt                    int64
	// DueAt is the natural occurrence the row is (or was last) delivering —
	// the value X-Carlos-Schedule-At carries, and the key an app dedupes
	// on. It is NOT next_at: after a retry next_at is the retry instant and
	// after `run` it is "now", while the occurrence stays put. Moves only
	// when the row advances to a new occurrence (MarkTimerFired with a new
	// due, FireTimerNow, or an upsert that moves next_at). Zero on rows
	// written before the column existed; readers fall back to NextAt.
	DueAt        int64
	Registration string `json:"-"`
}

var ErrTimerKind = errors.New("registry: timer exists with a different kind")

const timerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at, registration, claim_incarnation, claim_stamp, claim_owner, claim_row_id`

func scanTimer(row interface{ Scan(...any) error }) (Timer, error) {
	var t Timer
	err := row.Scan(&t.Host, &t.Name, &t.Kind, &t.Spec, &t.Path, &t.NextAt, &t.LastAt, &t.LastStatus, &t.LastMs, &t.Attempts, &t.CreatedAt, &t.DueAt, &t.Registration, &t.ClaimIncarnation, &t.ClaimStamp, &t.ClaimOwner, &t.ClaimRowID)
	return t, err
}

// UpsertTimer inserts t, or updates an existing row's spec/path. A changed spec/path or non-zero NextAt starts a new registration and resets attempts; unchanged declarations preserve delivery state.
func (r *Registry) UpsertTimer(t Timer) error {
	if r.remote != nil {
		_, err := r.remote.call(nil, remoteRequest{Method: "UpsertTimer", T: t}, nil)
		return err
	}

	if !hostwrite.Enabled(os.Getenv) {
		return r.upsertLegacyTimer(t)
	}
	tx, err := r.beginClaimTransaction(context.Background(), false)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	rt, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host=?`, t.Host).Scan)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	if err == nil && rt.ClaimOwnership.Valid() {
		if t.ClaimOwner != "" && (t.ClaimIncarnation != rt.ClaimIncarnation || t.ClaimStamp != rt.ClaimStamp || t.ClaimOwner != rt.ClaimOwner) {
			return ErrClaimOwnership
		}
		t.ClaimOwnership = rt.ClaimOwnership
	}
	var id [16]byte
	if _, err = rand.Read(id[:]); err != nil {
		return err
	}
	t.Registration = hex.EncodeToString(id[:])
	if t.ClaimOwnership.Valid() {
		t.ClaimRowID = t.Registration
	}
	if err = upsertTimer(tx, t); err != nil {
		return err
	}
	return tx.Commit()
}
func upsertTimer(tx interface {
	Exec(string, ...any) (sql.Result, error)
}, t Timer) error {
	const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)`
	res, err := tx.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(host, name) DO UPDATE SET
 spec=excluded.spec, path=excluded.path,
 next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END,
 due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END,
 attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END,
 registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END,
 claim_row_id=CASE WHEN `+rearm+` THEN excluded.claim_row_id ELSE timers.claim_row_id END
WHERE timers.kind=excluded.kind AND timers.claim_incarnation=excluded.claim_incarnation
AND timers.claim_stamp=excluded.claim_stamp AND timers.claim_owner=excluded.claim_owner`,
		t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration, t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrTimerKind
	}
	return nil
}

// ReplaceTimers makes host's timer rows exactly rows, in one transaction,
// keeping every column as given: next_at, due_at, attempts and
// registration included. It is how an installation imports the wake state
// it was handed (region HA spec 1 §5.7, moves spec §9.4).
//
// UpsertTimer is the wrong tool for that: it mints a fresh registration
// and resets the runner's bookkeeping on a rearm.
func (r *Registry) ReplaceTimers(host string, rows []Timer) error {
	if r.remote != nil {
		_, err := r.remote.call(nil, remoteRequest{Method: "ReplaceTimers", Host: host, Rows: rows}, nil)
		return err
	}

	if !hostwrite.Enabled(os.Getenv) {
		return r.replaceLegacyTimers(host, rows)
	}
	for _, t := range rows {
		if t.Host != host {
			return fmt.Errorf("registry: timer %q is for host %q, not %q", t.Name, t.Host, host)
		}
	}
	tx, err := r.beginClaimTransaction(context.Background(), false)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	rt, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host=?`, host).Scan)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	if err == nil && rt.ClaimOwnership.Valid() {
		for i := range rows {
			if rows[i].Registration == "" {
				return ErrClaimOwnership
			}
			rows[i].ClaimOwnership = rt.ClaimOwnership
			rows[i].ClaimRowID = rows[i].Registration
		}
	}
	if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil {
		return err
	}
	for _, t := range rows {
		if _, err := tx.Exec(`INSERT INTO timers (`+timerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
			t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration, t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID); err != nil {
			return fmt.Errorf("registry: insert timer %q: %w", t.Name, err)
		}
	}
	return tx.Commit()
}

func (r *Registry) GetTimer(host, name string) (*Timer, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "GetTimer", Host: host, Name: name}, nil)
		return out.Timer, err
	}

	t, err := scanTimer(r.sql.QueryRow(`SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, host, name))
	if errors.Is(err, sql.ErrNoRows) {
		return nil, os.ErrNotExist
	}
	if err != nil {
		return nil, err
	}
	return &t, nil
}

func (r *Registry) DeleteTimer(host, name string) error {
	if r.remote != nil {
		_, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimer", Host: host, Name: name}, nil)
		return err
	}

	_, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=?`, host, name)
	return err
}

func (r *Registry) DeleteTimerRegistration(host, name, registration string) error {
	if r.remote != nil {
		_, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimerRegistration", Host: host, Name: name, Registration: registration}, nil)
		return err
	}

	_, err := r.sql.Exec(`DELETE FROM timers WHERE host=? AND name=? AND registration=?`, host, name, registration)
	return err
}

func (r *Registry) queryTimers(q string, args ...any) ([]Timer, error) {
	rows, err := r.sql.Query(q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Timer
	for rows.Next() {
		t, err := scanTimer(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, t)
	}
	return out, rows.Err()
}

func (r *Registry) TimersFor(host string) ([]Timer, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "TimersFor", Host: host}, nil)
		return out.Timers, err
	}

	return r.queryTimers(`SELECT `+timerCols+` FROM timers WHERE host=? ORDER BY name`, host)
}

// DueTimers is the runner's batch: rows due at or before now, earliest
// first, at most limit of them.
//
// excludeHosts is what keeps the batch FAIR. The runner delivers one timer
// at a time per host and skips a host whose delivery is still running, so a
// host with more due rows than the batch holds — or one parked in a
// 30-minute handler — would otherwise fill every slot of every batch with
// rows the runner is about to skip, and another host's due timer would never
// be seen at all. Passing the busy set here moves the filter into the query,
// where the LIMIT is applied, instead of after it. An empty (or nil) slice
// is no filter, which is the ordinary case: nothing in flight.
func (r *Registry) DueTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "DueTimers", Now: now, Limit: limit, ExcludeHosts: excludeHosts}, nil)
		return out.Timers, err
	}

	return r.dueTimers(now, limit, excludeHosts, false)
}

// DueUnheldTimers is DueTimers without the rows of held routes (moves spec
// §5.2, HoldStatuses). The exclusion is in the query, before the LIMIT, so
// a held host's rows cannot fill the batch with rows the runner would only
// skip and hide another host's due timer behind them (moves Task 6 fix
// round 1). A row whose host has no route is still returned, as DueTimers
// returns it. The runner asks this only with its hold wiring armed;
// DueTimers stays main's query.
func (r *Registry) DueUnheldTimers(now int64, limit int, excludeHosts []string) ([]Timer, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "DueUnheldTimers", Now: now, Limit: limit, ExcludeHosts: excludeHosts}, nil)
		return out.Timers, err
	}

	return r.dueTimers(now, limit, excludeHosts, true)
}

// dueTimers is both queries. Neither returns the rows of a gated route (an
// uncommitted import's, seeded-instances spec §5.4), and the exclusion is
// in the query, before the LIMIT (final review F5): the runner skips such a
// row without advancing it, so gated rows left in the batch would stay due
// for ever and could fill every slot of it, hiding every other host's due
// timer. The runner keeps its own check.
func (r *Registry) dueTimers(now int64, limit int, excludeHosts []string, skipHeld bool) ([]Timer, error) {
	q := `SELECT ` + timerCols + ` FROM timers WHERE next_at > 0 AND next_at <= ?` +
		` AND host NOT IN (SELECT host FROM routes WHERE import_gate != '')`
	args := []any{now}
	if skipHeld {
		statuses := make([]string, 0, len(HoldStatuses))
		for s := range HoldStatuses {
			statuses = append(statuses, s)
		}
		sort.Strings(statuses)
		q += ` AND host NOT IN (SELECT host FROM routes WHERE status IN (?` + strings.Repeat(",?", len(statuses)-1) + `))`
		for _, s := range statuses {
			args = append(args, s)
		}
	}
	if len(excludeHosts) > 0 {
		q += ` AND host NOT IN (?` + strings.Repeat(",?", len(excludeHosts)-1) + `)`
		for _, h := range excludeHosts {
			args = append(args, h)
		}
	}
	q += ` ORDER BY next_at, host, name LIMIT ?`
	return r.queryTimers(q, append(args, limit)...)
}

// FireTimerNow makes the row due at now, and now IS the occurrence being
// delivered (a `carlos schedule run` is its own occurrence, not an early
// delivery of the next natural one).
func (r *Registry) FireTimerNow(host, name string, now int64) error {
	if r.remote != nil {
		_, err := r.remote.call(nil, remoteRequest{Method: "FireTimerNow", Host: host, Name: name, Now: now}, nil)
		return err
	}

	var id [16]byte
	if _, err := rand.Read(id[:]); err != nil {
		return err
	}
	registration := hex.EncodeToString(id[:])
	_, err := r.sql.Exec(`UPDATE timers SET next_at=?, due_at=?, attempts=0, registration=? , claim_row_id=CASE WHEN claim_owner='' THEN claim_row_id ELSE ? END WHERE host=? AND name=?`, now, now, registration, registration, host, name)
	return err
}

// MarkTimerFired records one delivery. next is when the row is due again;
// due is the occurrence THAT next delivery is for: the same value as next
// when the row advances, and the row's current DueAt when next is a retry
// of the occurrence just attempted — the runner decides which.
func (r *Registry) MarkTimerFired(host, name, registration string, at int64, status string, ms int64, attempts int, next, due int64) (bool, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "MarkTimerFired", Host: host, Name: name, Registration: registration, At: at, Status: status, Ms: ms, Attempts: attempts, NextAt: next, Due: due}, nil)
		return out.Changed, err
	}

	res, err := r.sql.Exec(`UPDATE timers SET last_at=?, last_status=?, last_ms=?, attempts=?, next_at=?, due_at=? WHERE host=? AND name=? AND registration=?`,
		at, status, ms, attempts, next, due, host, name, registration)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	return n != 0, err
}

// SetTimerStatus writes last_status alone — converge's one word on a row
// it parked (schedule.StatusNeverFires); last_at stays as it was, so the
// status reads as a condition, not a fire.
func (r *Registry) SetTimerStatus(host, name, status string) error {
	if r.remote != nil {
		_, err := r.remote.call(nil, remoteRequest{Method: "SetTimerStatus", Host: host, Name: name, Status: status}, nil)
		return err
	}

	_, err := r.sql.Exec(`UPDATE timers SET last_status=? WHERE host=? AND name=?`, status, host, name)
	return err
}

func (r *Registry) CountTimers(host, kind string) (int, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "CountTimers", Host: host, Kind: kind}, nil)
		return out.Number, err
	}

	var n int
	err := r.sql.QueryRow(`SELECT COUNT(*) FROM timers WHERE host=? AND kind=?`, host, kind).Scan(&n)
	return n, err
}

// PruneTimers: rows for hosts no longer routed here, and finished once-rows
// older than doneBefore.
//
// An empty alive set deletes EVERY row. That is the right answer for a
// genuinely empty registry — routes and timers live in the same file, so
// every row really is orphaned — but it is a wipe, and there are two callers
// now: schedule.Converger.Pass (every pass, on a box with a deploy store) and
// schedule.Runner.prune (daily, on every box). Both build alive from the
// registry's own route list; the runner declines the sweep outright when that
// list is empty rather than hand the wipe over. A third caller should do the
// same, or say in its own comment why the wipe is what it wants.
func (r *Registry) PruneTimers(alive map[string]bool, doneBefore int64) (int, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "PruneTimers", Alive: alive, DoneBefore: doneBefore}, nil)
		return out.Number, err
	}

	hosts := make([]any, 0, len(alive))
	for h := range alive {
		hosts = append(hosts, h)
	}
	q := `DELETE FROM timers WHERE (kind='once' AND next_at=0 AND last_at < ?)`
	args := []any{doneBefore}
	if len(hosts) == 0 {
		q += ` OR 1=1`
	} else {
		q += ` OR host NOT IN (?` + strings.Repeat(",?", len(hosts)-1) + `)`
		args = append(args, hosts...)
	}
	res, err := r.sql.Exec(q, args...)
	if err != nil {
		return 0, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return 0, err
	}
	return int(n), nil
}

func (r *Registry) DeleteTimersExcept(host, kind string, keep map[string]bool) (int, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "DeleteTimersExcept", Host: host, Kind: kind, Keep: keep}, nil)
		return out.Number, err
	}

	names := make([]any, 0, len(keep))
	for n := range keep {
		names = append(names, n)
	}
	q := `DELETE FROM timers WHERE host=? AND kind=?`
	args := []any{host, kind}
	if len(names) > 0 {
		q += ` AND name NOT IN (?` + strings.Repeat(",?", len(names)-1) + `)`
		args = append(args, names...)
	}
	res, err := r.sql.Exec(q, args...)
	if err != nil {
		return 0, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return 0, err
	}
	return int(n), nil
}

// EarliestTimers is the hibernation guard's batch: for every host with at
// least one armed row, the soonest next_at among them. One query, because
// the activator's sweep reads it BEFORE taking its mutex (activator.go's
// "nothing under a.mu should wait on SQLite") and a per-entry query inside
// the sweep loop would put SQLite exactly where that comment forbids it.
//
// next_at = 0 is "not armed" — a parked cron, or a row converge wrote and
// the runner has not yet dated — and is excluded here rather than by the
// caller, matching DueTimers' own `next_at > 0` gate.
func (r *Registry) EarliestTimers() (map[string]int64, error) {
	if r.remote != nil {
		out, err := r.remote.call(nil, remoteRequest{Method: "EarliestTimers"}, nil)
		return out.Times, err
	}

	rows, err := r.sql.Query(`SELECT host, MIN(next_at) FROM timers WHERE next_at > 0 GROUP BY host`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make(map[string]int64)
	for rows.Next() {
		var host string
		var next int64
		if err := rows.Scan(&host, &next); err != nil {
			return nil, err
		}
		out[host] = next
	}
	return out, rows.Err()
}

const legacyTimerCols = `host, name, kind, spec, path, next_at, last_at, last_status, last_ms, attempts, created_at, due_at, registration`

func (r *Registry) upsertLegacyTimer(t Timer) error {
	const rearm = `(timers.spec != excluded.spec OR timers.path != excluded.path OR excluded.next_at != 0)`
	res, err := r.sql.Exec(`INSERT INTO timers (`+legacyTimerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,lower(hex(randomblob(16))))
ON CONFLICT(host, name) DO UPDATE SET
 spec=excluded.spec, path=excluded.path,
 next_at=CASE WHEN `+rearm+` THEN excluded.next_at ELSE timers.next_at END,
 due_at=CASE WHEN `+rearm+` THEN excluded.due_at ELSE timers.due_at END,
 attempts=CASE WHEN `+rearm+` THEN 0 ELSE timers.attempts END,
 registration=CASE WHEN `+rearm+` THEN excluded.registration ELSE timers.registration END
WHERE timers.kind=excluded.kind`,
		t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrTimerKind
	}
	return nil
}

func (r *Registry) replaceLegacyTimers(host string, rows []Timer) error {
	for _, t := range rows {
		if t.Host != host {
			return fmt.Errorf("registry: timer %q is for host %q, not %q", t.Name, t.Host, host)
		}
	}
	tx, err := r.sql.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil {
		return err
	}
	for _, t := range rows {
		if _, err := tx.Exec(`INSERT INTO timers (`+legacyTimerCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`,
			t.Host, t.Name, t.Kind, t.Spec, t.Path, t.NextAt, t.LastAt, t.LastStatus, t.LastMs, t.Attempts, t.CreatedAt, t.DueAt, t.Registration); err != nil {
			return fmt.Errorf("registry: insert timer %q: %w", t.Name, err)
		}
	}
	return tx.Commit()
}