package schedule

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// Converger turns apps/<acct>/<app>/schedules.json into timers rows for every
// exec-backed instance this box serves, acts on schedule-run.json, and prunes
// what no longer belongs (spec §6.2). One instance per agent; Pass is called
// on every serial move, hourly, and once at start.
//
// Etag memory is what keeps an idle box cheap and a run-signal
// exactly-once: an unchanged declaration is one GET and no writes; a run
// signal fires when its etag MOVES, and the first sighting only seeds
// (a signal older than this process is history — restart-gen's rule).
//
// One instance, but THREE callers, and they are separate goroutines: the
// agent's fast converge loop runs Pass at arming and on every serial move,
// and its adopt loop runs Pass hourly. Pass is therefore serialized on mu —
// see the field.
type Converger struct {
	Reg        *registry.Registry
	Store      release.Store
	Now        func() time.Time
	Box        string                             // reported in Status.Box
	Report     func(rt registry.Route, st Status) // called for every host whose rows changed or that is unsupported; nil = no reports
	Idle       time.Duration                      // the activator's idle window; zero when no hibernation is armed on this box
	HostWrites hostwrite.Config
	// Frozen answers whether a move holds host's claim (moves spec §9.4,
	// §19 K9): its timers are neither upserted nor pruned. Nil: none is.
	Frozen func(host string) bool
	// Gate is the per-host gate shared with the runner and a move's drain;
	// convergeHost holds it across its Frozen check and its writes. Nil:
	// no gate, main's converger.
	Gate *HostGate

	// afterFrozenCheck is the tests' pause inside the gate, after the
	// Frozen check (TestConvergerCannotWriteAfterTheBarrier). Nil in the
	// agent.
	afterFrozenCheck func(host string)

	declEtag map[[2]string]string // (acct, app) → last etag seen
	declDoc  map[[2]string]Declaration
	runEtag  map[[2]string]string

	// reportedUnsupported tracks which unit-backed hosts have already had
	// their one report — Report fires for them once, not every pass.
	reportedUnsupported map[string]bool

	// mu serializes Pass, and is held for the WHOLE of it. Two reasons, and
	// the first one alone is fatal:
	//
	//  1. The four maps above are plain maps written on every pass. The agent
	//     calls Pass from two goroutines (fastConvergeLoop's seed and serial
	//     leg; adoptLoop's hourly floor), and they genuinely overlap — an
	//     hourly pass mid-flight when a console bump fires the fast one. A
	//     concurrent map read and write is a runtime THROW, not a data race
	//     you find later: it would kill the agent, and the activator and every
	//     exec child with it.
	//  2. The etag memory IS the exactly-once mechanism for a manual run. Two
	//     interleaved passes can both read runEtag[k] before either writes it,
	//     both see movement, and both fire the same `carlos schedule run` —
	//     the one promise this struct's comment makes, broken.
	//
	// The lock lives HERE rather than at the agent's call sites (the shape
	// configSyncMu/reconcileMu use in host/agent.go) because a Converger is a
	// stateful object with an exported method: a future caller cannot forget a
	// lock it never has to take.
	mu sync.Mutex
}

const onceRetention = 7 * 24 * time.Hour

func NewConverger(reg *registry.Registry, store release.Store, box string, report func(registry.Route, Status)) *Converger {
	return &Converger{
		Reg: reg, Store: store, Now: time.Now, Box: box, Report: report,
		declEtag: map[[2]string]string{}, declDoc: map[[2]string]Declaration{},
		runEtag: map[[2]string]string{}, reportedUnsupported: map[string]bool{},
	}
}

// Pass is die-free per app: one app's unreadable declaration is logged and
// its rows left exactly as they were (a baseline with holes manufactures
// deletions), and every other app still converges.
//
// Serialized end to end (see mu): concurrent callers queue rather than
// interleave. A pass is store-bound, so a queued caller can wait on a slow
// bucket — that is the intended cost, and cheaper than the alternative for
// both reasons mu documents.
func (c *Converger) Pass(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	routes, err := c.Reg.List()
	if err != nil {
		return err
	}
	now := c.Now()
	// alive is every row, held ones included, so PruneTimers never treats a
	// frozen host's rows as orphans. frozenAny holds back the table-wide
	// prune itself while a move has frozen any host here: its retention leg
	// would otherwise delete a frozen host's finished once-rows. The
	// runner's daily sweep still runs (it cannot be frozen; the timers
	// digest leaves that history out, joint N14), so nothing accumulates.
	alive := map[string]bool{}
	frozenAny := false
	byApp := map[[2]string][]registry.Route{}
	for _, rt := range routes {
		alive[rt.Host] = true
		if c.Frozen != nil && !frozenAny && c.Frozen(rt.Host) {
			frozenAny = true
		}
		if rt.Kind != "instance" {
			continue
		}
		k := [2]string{release.RouteAccount(rt), rt.App}
		if k[1] == "" { // no k[0] check: RouteAccount's floor is release.OpsAccount, never ""
			continue
		}
		byApp[k] = append(byApp[k], rt)
	}
	var errs []error
	for k, rts := range byApp {
		if err := c.convergeApp(ctx, k[0], k[1], rts, now); err != nil {
			errs = append(errs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
		}
	}
	if !c.HostWrites.Enabled && !frozenAny {
		if n, err := c.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
			errs = append(errs, err)
		} else if n > 0 {
			log.Printf("schedule: pruned %d timer row(s)", n)
		}
	}
	return errors.Join(errs...)
}

func (c *Converger) convergeApp(ctx context.Context, acct, app string, rts []registry.Route, now time.Time) error {
	// Capture once before either grouped remote input read. Reuse on every effect.
	selected := map[string]hostwrite.Work{}
	var errs []error
	for _, rt := range rts {
		work, err := c.HostWrites.Select(ctx, rt.Host, "timers")
		if err != nil {
			errs = append(errs, fmt.Errorf("%s: select lifecycle authority: %w", rt.Host, err))
			continue
		}
		selected[rt.Host] = work
	}
	k := [2]string{acct, app}
	// Reads the object directly rather than calling ReadDeclaration: the
	// etag has to be compared against declEtag[k] BEFORE deciding whether
	// to unmarshal, so an unchanged declaration costs one GET and zero JSON
	// parses (the struct's own doc comment's promise). ReadDeclaration
	// always unmarshals what it reads — right for a one-off caller, wrong
	// for a loop that runs every 2s per app.
	raw, etag, err := c.Store.Get(ctx, DeclarationKey(acct, app))
	var decl Declaration
	switch {
	case errors.Is(err, os.ErrNotExist):
		etag = ""
	case err != nil:
		return fmt.Errorf("read schedules.json: %w", err)
	case etag == c.declEtag[k]:
		decl = c.declDoc[k]
	default:
		if err := json.Unmarshal(raw, &decl); err != nil {
			return fmt.Errorf("parse schedules.json: %w", err)
		}
	}
	c.declEtag[k], c.declDoc[k] = etag, decl

	// Run signal: seed silently on first sight, fire on movement.
	runRaw, runEtag, err := c.Store.Get(ctx, RunKey(acct, app))
	fireName := ""
	switch {
	case errors.Is(err, os.ErrNotExist):
		runEtag = ""
	case err != nil:
		return fmt.Errorf("read schedule-run.json: %w", err)
	}
	if prev, seen := c.runEtag[k]; seen && runEtag != prev && runEtag != "" {
		var sig RunSignal
		if err := json.Unmarshal(runRaw, &sig); err == nil {
			if _, ok := decl.Find(sig.Name); ok {
				fireName = sig.Name
			} else {
				log.Printf("schedule: %s/%s: run requested for undeclared schedule %q — ignored", acct, app, sig.Name)
			}
		}
	}
	c.runEtag[k] = runEtag

	for _, rt := range rts {
		work, ok := selected[rt.Host]
		if !ok {
			continue
		}
		hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work})
		changed, err := c.convergeHost(hctx, rt, decl, fireName, now)
		if err != nil {
			errs = append(errs, fmt.Errorf("%s: %w", rt.Host, err))
			continue
		}
		if changed && c.Report != nil {
			var st Status
			err := withTimerWork(hctx, func() error {
				var statusErr error
				st, statusErr = StatusFor(c.Reg, rt, c.Box, now, c.Idle)
				return statusErr
			})
			if err != nil {
				errs = append(errs, err)
				continue
			}
			c.Report(rt, st)
		}
	}
	return errors.Join(errs...)
}

// convergeHost returns whether anything about the host's rows changed (or
// the host is unsupported and should be reported so).
//
// A host a move has frozen (Frozen, moves spec §9.4 and §19 K9) is left
// exactly as it is: nothing upserted, nothing pruned, a run signal not
// fired. The check and every write below happen inside the host's gate, so
// the cut's drain (HostGate.Drain after the hold) waits for a convergence
// that checked before the freeze, and one that starts after the drain sees
// it. A host the gate refuses is being drained: skipped the same way.
func (c *Converger) convergeHost(ctx context.Context, rt registry.Route, decl Declaration, fireName string, now time.Time) (bool, error) {
	if !c.Gate.tryEnter(rt.Host) {
		return false, nil
	}
	defer c.Gate.leave(rt.Host)
	if c.Frozen != nil && c.Frozen(rt.Host) {
		return false, nil
	}
	if c.afterFrozenCheck != nil {
		c.afterFrozenCheck(rt.Host)
	}
	var before []registry.Timer
	if c.HostWrites.Enabled {
		err := withTimerWork(ctx, func() error {
			var readErr error
			before, readErr = c.Reg.TimersFor(rt.Host)
			return readErr
		})
		if err != nil {
			return false, err
		}
		cutoff := now.Add(-onceRetention).Unix()
		kept := before[:0]
		for _, row := range before {
			if row.Kind == KindOnce && row.NextAt == 0 && row.LastAt < cutoff {
				if err := deleteSelectedTimer(ctx, c.Reg, row); err != nil {
					return false, err
				}
				continue
			}
			kept = append(kept, row)
		}
		before = kept
	}
	if rt.Backing == registry.BackingUnit {
		// A host can flip exec -> unit mid-lifetime (the app-capabilities
		// migration door, registry.SetBacking): drop any schedule rows it
		// still carries from before the flip, so "no rows" stays true for
		// every unit-backed host, not just ones that were always unit-backed.
		var n int
		n, err := c.deleteSchedules(ctx, rt.Host, before, map[string]bool{})
		if err != nil {
			return false, err
		}
		changed := n > 0
		// Reported once per host, and only when there is something to warn
		// about — an undeclared app on a unit-backed host is silent.
		if len(decl.Schedules) > 0 && c.unsupportedUnreported(rt.Host) {
			changed = true
		}
		return changed, nil
	}
	if !c.HostWrites.Enabled {
		var err error
		before, err = c.Reg.TimersFor(rt.Host)
		if err != nil {
			return false, err
		}
	}
	registrations := map[string]string{}
	keep := map[string]bool{}
	changed := false
	for _, sc := range decl.Schedules {
		sp, err := sc.Spec()
		if err != nil {
			log.Printf("schedule: %s: schedule %q has an invalid spec (%v) — skipped", rt.Host, sc.Name, err)
			continue
		}
		keep[sc.Name] = true
		// prev is the ONLY-schedule-kind baseline: a same-named once-row
		// (the instance's own, off its control socket) is never "previous
		// state" for this comparison — a kind collision decides itself
		// below, at the write, not here.
		prev := findTimer(before, sc.Name, KindSchedule)
		wouldChange := prev == nil || prev.Spec != sp.String() || prev.Path != sc.Path
		var next int64
		neverFires := false
		if wouldChange {
			// Schedule.Validate refuses a spec whose first Next is zero at
			// `set` time; one written some other way must not become a row
			// due at the zero time's unix value (-62135596800 — in the past,
			// so due on the very next tick, forever). It is parked at
			// next_at=0 and told why, below.
			if n := sp.Next(now); !n.IsZero() {
				next = n.Unix()
			} else {
				neverFires = true
			}
		}
		// A new occurrence is due whenever next moves (DueAt rides NextAt —
		// UpsertTimer leaves both alone when NextAt is 0 and nothing changed).
		var installed *registry.Timer
		t := registry.Timer{Host: rt.Host, Name: sc.Name, Kind: KindSchedule, Spec: sp.String(), Path: sc.Path, NextAt: next, DueAt: next, CreatedAt: now.Unix()}
		handled := false
		if c.HostWrites.Enabled {
			expected := ""
			if prev != nil {
				expected = prev.Registration
			}
			if prev != nil && !wouldChange && hasTimerMutator(ctx) {
				installed, handled = prev, true
			} else {
				installed, handled, err = mutateSelectedTimer(ctx, t, expected, false)
			}
		}
		if !handled {
			err = withTimerWork(ctx, func() error {
				if !c.HostWrites.Enabled {
					return c.Reg.UpsertTimer(t)
				}
				expected := ""
				if prev != nil {
					expected = prev.Registration
				}
				var err error
				installed, err = c.Reg.UpsertTimerRegistration(t, expected)
				return err
			})
		}

		if errors.Is(err, registry.ErrTimerKind) {
			// Nothing was written — a collision must not pin `changed`,
			// or a permanently-colliding name reports every single pass.
			log.Printf("schedule: %s: %q is a once-timer the instance registered; the declared schedule of that name is skipped", rt.Host, sc.Name)
			continue
		}
		if err != nil {
			return changed, err
		}
		if installed != nil {
			registrations[sc.Name] = installed.Registration
		}
		if neverFires {
			log.Printf("schedule: %s: schedule %q never fires (no occurrence within the search window) — parked", rt.Host, sc.Name)
			if err := withTimerWork(ctx, func() error {
				if c.HostWrites.Enabled {
					return c.Reg.SetTimerStatusRegistration(rt.Host, sc.Name, installed.Registration, StatusNeverFires)
				}
				return c.Reg.SetTimerStatus(rt.Host, sc.Name, StatusNeverFires)
			}); err != nil {
				return changed, err
			}
		}
		if wouldChange {
			changed = true
		}
	}
	n, err := c.deleteSchedules(ctx, rt.Host, before, keep)
	if err != nil {
		return changed, err
	}
	if n > 0 {
		changed = true
	}
	if fireName != "" && keep[fireName] {
		fire := func() error {
			if c.HostWrites.Enabled {
				row, err := c.Reg.GetTimer(rt.Host, fireName)
				if err != nil {
					return err
				}
				if row.Registration != registrations[fireName] {
					return registry.ErrTimerChanged
				}
				row.NextAt, row.DueAt, row.Attempts = now.Unix(), now.Unix(), 0
				expected := row.Registration
				row.Registration = ""
				if _, handled, err := mutateSelectedTimer(ctx, *row, expected, false); handled {
					return err
				}
			}
			return withTimerWork(ctx, func() error {
				if c.HostWrites.Enabled {
					return c.Reg.FireTimerNowRegistration(rt.Host, fireName, registrations[fireName], now.Unix())
				}
				return c.Reg.FireTimerNow(rt.Host, fireName, now.Unix())
			})
		}
		if err := fire(); err != nil {
			return changed, err
		}
		changed = true
	}
	return changed, nil
}

// unsupportedUnreported reports whether host has NOT yet had its
// unsupported-host report sent, and marks it sent — so the report is
// written once per host's lifetime in this Converger, not every pass.
func (c *Converger) unsupportedUnreported(host string) bool {
	if c.reportedUnsupported == nil {
		c.reportedUnsupported = map[string]bool{}
	}
	if c.reportedUnsupported[host] {
		return false
	}
	c.reportedUnsupported[host] = true
	return true
}

// findTimer returns the row named name with the given kind, or nil. Kind
// matters: a once-row and a declared schedule can share a name (the
// once-row is the instance's own, registered off its control socket), and
// only a same-kind row is ever "previous state" for the change comparison
// below — a kind collision is a skip (ErrTimerKind, at the UpsertTimer
// call), never a manufactured "changed".
func findTimer(rows []registry.Timer, name, kind string) *registry.Timer {
	for i := range rows {
		if rows[i].Name == name && rows[i].Kind == kind {
			return &rows[i]
		}
	}
	return nil
}

const UnsupportedUnit = "unit-backed instance — scheduled work delivers to exec-backed instances in this release"

// StatusNeverFires is the last_status of a parked row: a declared cron
// expression with no occurrence inside CronExpr.Next's window (Validate
// refuses these at `set`; this is the defence for one that slipped past it).
const StatusNeverFires = "never fires"

// StatusFor builds the host's report from its rows. idle is the activator's
// idle window on this box, threaded through to holdsAwake — zero when no
// hibernation is armed on the box at all.
//
// Hibernation is also armed PER ROUTE (rt.Hibernate — activator.go will not
// sweep a host unless Hibernate && Status == "ready"), so a box-wide window
// is not sufficient on its own: a route with Hibernate false was never
// going to sleep, schedule or no schedule, and holds_awake must say so by
// reporting false for every one of its rows, not by naming a tight timer
// that changes nothing for it.
func StatusFor(reg *registry.Registry, rt registry.Route, box string, now time.Time, idle time.Duration) (Status, error) {
	st := Status{Schema: 1, Host: rt.Host, Box: box, ReportedAt: now.UTC().Format(time.RFC3339), Timers: []TimerStatus{}}
	if rt.Backing == registry.BackingUnit {
		st.Unsupported = UnsupportedUnit
		return st, nil
	}
	if !rt.Hibernate {
		idle = 0
	}
	rows, err := reg.TimersFor(rt.Host)
	if err != nil {
		return st, err
	}
	for _, r := range rows {
		st.Timers = append(st.Timers, toStatus(r, idle, now))
	}
	return st, nil
}

// toStatus maps one timers row onto its wire shape. Task 5 (the `carlos
// schedule ls` reader / the runner's own report path) reuses this so the
// row-to-status mapping is written once.
func toStatus(t registry.Timer, idle time.Duration, now time.Time) TimerStatus {
	return TimerStatus{
		Name: t.Name, Kind: t.Kind, Spec: t.Spec, Path: t.Path,
		NextAt: t.NextAt, DueAt: t.DueAt, LastAt: t.LastAt,
		LastStatus: t.LastStatus, LastMs: t.LastMs, Attempts: t.Attempts,
		HoldsAwake: holdsAwake(t, idle, now),
	}
}

// holdsAwake reports whether t's interval is tight enough that its instance
// never gets to sleep: the gap between consecutive occurrences is no wider
// than idle, so the activator's guard (and, before it, the tick's own wake)
// keeps the process resident indefinitely.
//
// The gap comes from two Next calls rather than Spec.Every, so a cron gets a
// real answer instead of a special case — but for an irregular cron (e.g.
// "0,5 * * * *": five minutes then fifty-five) that first gap is a SAMPLE,
// not a bound: the same row can answer true at :59 and false at :01,
// depending on when the report happens to be built. That is the direction
// to err in — over-warning about a bill beats missing one — and is chosen
// deliberately, not an oversight.
//
// idle <= 0 means hibernation is not armed — either the box has no
// activator, or (StatusFor) the route itself has Hibernate off — so
// nothing holds anything awake, because nothing sleeps. A once-timer fires
// once and cannot hold anything. An unparseable spec is not evidence of a
// problem, so it makes no claim.
func holdsAwake(t registry.Timer, idle time.Duration, now time.Time) bool {
	if idle <= 0 || t.Kind != KindSchedule {
		return false
	}
	sp, err := ParseSpec(t.Spec)
	if err != nil {
		return false
	}
	first := sp.Next(now)
	if first.IsZero() {
		return false
	}
	second := sp.Next(first)
	if second.IsZero() {
		return false
	}
	return second.Sub(first) <= idle
}

// deleteSchedules only removes rows observed by this pass. A same-name
// replacement keeps its registration and survives the stale deletion.
func (c *Converger) deleteSchedules(ctx context.Context, host string, before []registry.Timer, keep map[string]bool) (int, error) {
	if !c.HostWrites.Enabled {
		return c.Reg.DeleteTimersExcept(host, KindSchedule, keep)
	}
	if err := withTimerWork(ctx, func() error { return nil }); err != nil {
		return 0, err
	}
	n := 0
	for _, row := range before {
		if row.Kind != KindSchedule || keep[row.Name] {
			continue
		}
		if err := deleteSelectedTimer(ctx, c.Reg, row); err != nil {
			return n, err
		}
		n++
	}
	return n, nil
}