package host

import (
	"context"
	"log"
	"slices"
	"sync/atomic"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/schedule"
)

// movesHoldView is what this box's other passes ask about running moves
// (moves spec §5.2, §9.4, §9.6, §19 K9): reconcile mints and keeps a
// participant's row through hold and keep, and the schedule converger
// leaves a frozen host's timers alone through frozen. Strictness is not
// here: the moves worker's setStrict is the one strict source (G9).
//
// It is rebuilt from every moves pass's record list, plus one claim read
// per host with a Phase 1 move in flight, and read from the reconcile and
// schedule loops in between. It fails closed: until the first build, keep
// and frozen answer true for every host (a removal or a schedule converge
// waits a pass rather than act on a host it cannot vouch for), and a host
// whose claim would not read is kept and frozen until it does. hold judges
// the record the caller passes, not the view's copy, so a flip the view has
// not seen yet still mints the landing row held.
//
// Only built with the Phase 1 switch on: off, armedHoldView is nil and no
// pass is given a hook.
//
// Ending a hold that hold would re-assert (the landing's open, the
// finalisation: Task 10, movesland.go) must happen under
// provision.WithHostHoldLock, the lock reconcile's hold decision holds from
// its record re-read through its conditional write. See that function.
type movesHoldView struct {
	box, region string
	read        func(ctx context.Context, s release.Store, host string) (hostclaim.Observed, error)
	// written is reconcile's HoldWritten: the stage-requested marker after
	// a hold write (G10). Nil writes nothing.
	written func(host string)
	// gate is the schedule wiring's HostGate, set by armScheduleHolds; nil
	// until then (no scheduled work armed: nothing to drain).
	gate atomic.Pointer[schedule.HostGate]
	snap atomic.Pointer[holdSnapshot]
}

// holdSnapshot is one build: what each in-flight move's claim said.
type holdSnapshot struct {
	claims  map[string]holdClaim // host -> the claim's op, for hosts whose claim read
	unknown map[string]bool      // hosts whose claim would not read
}

type holdClaim struct {
	kind         string
	participants []string
}

func newMovesHoldView(box, region string, read func(context.Context, release.Store, string) (hostclaim.Observed, error), written func(host string)) *movesHoldView {
	return &movesHoldView{box: box, region: region, read: read, written: written}
}

// moveKinds are the claim op kinds a move holds (§19.1). Region HA's
// relocating is its own op and freezes through its own wiring.
var moveKinds = map[string]bool{hostclaim.KindMoving: true, hostclaim.KindLanding: true,
	hostclaim.KindReverting: true, hostclaim.KindRestoring: true}

// refresh rebuilds the view from recs: every host with a Phase 1 move in
// flight has its claim read. Called only with a list that was read; a
// failed listing leaves the last build in place.
func (v *movesHoldView) refresh(ctx context.Context, s release.Store, recs []provision.Record) {
	next := &holdSnapshot{claims: map[string]holdClaim{}, unknown: map[string]bool{}}
	for _, rec := range recs {
		if rec.Move == nil || rec.Move.Legacy() {
			continue
		}
		if _, seen := next.claims[rec.Host]; seen || next.unknown[rec.Host] {
			continue
		}
		obs, err := v.read(ctx, s, rec.Host)
		if err != nil {
			log.Printf("moves: %s: claim unreadable (%v); its row is kept and its timers frozen until it reads", rec.Host, err)
			next.unknown[rec.Host] = true
			continue
		}
		if obs.Absent || obs.Claim == nil {
			continue
		}
		next.claims[rec.Host] = holdClaim{kind: obs.Claim.Op.Kind, participants: slices.Clone(obs.Claim.Op.Participants)}
	}
	v.snap.Store(next)
}

// keep is reconcile's KeepParticipant: this box is a participant of a
// move-kind op on host.
func (v *movesHoldView) keep(host string) bool {
	snap := v.snap.Load()
	if snap == nil || snap.unknown[host] {
		return true
	}
	c, ok := snap.claims[host]
	return ok && moveKinds[c.kind] && slices.Contains(c.participants, v.box)
}

// frozen is the converger's Frozen: a move-kind op holds host's claim.
func (v *movesHoldView) frozen(host string) bool {
	snap := v.snap.Load()
	if snap == nil || snap.unknown[host] {
		return true
	}
	c, ok := snap.claims[host]
	return ok && moveKinds[c.kind]
}

// The roles a box plays in one move.
const (
	roleNone   = ""
	roleSource = "source"
	roleTarget = "target"
)

// role is this box's part in rec's move: by the claim's participants
// ({source, target}, §19.1) when the view read the claim, else by the
// record's regions. A box the claim does not name plays none.
func (v *movesHoldView) role(rec provision.Record) (role, claimKind string) {
	if snap := v.snap.Load(); snap != nil {
		if c, ok := snap.claims[rec.Host]; ok && moveKinds[c.kind] && len(c.participants) == 2 {
			switch v.box {
			case c.participants[0]:
				return roleSource, c.kind
			case c.participants[1]:
				return roleTarget, c.kind
			}
			return roleNone, c.kind
		}
	}
	m := rec.Move
	switch {
	case m.From == v.region && m.To != v.region:
		return roleSource, ""
	case m.To == v.region && m.From != v.region:
		return roleTarget, ""
	case m.To == v.region:
		// Both ends in this region and no claim to say which box is which:
		// the state decides, and either answer holds the row.
		switch m.State {
		case provision.MoveLanding, provision.MoveCommitted:
			return roleTarget, ""
		}
		return roleSource, ""
	}
	return roleNone, ""
}

// hold is reconcile's Hold: the status and op a participant's row is held
// under, from the record's state and this box's role (§5.2, §7.4, §7.6,
// §7.8, §19.1): the source moving from stopping until its row is deleted at
// finalization, restoring through a revert or a rollback; the target landing
// from the flip until the commit opens its row, releasing through a
// rollback. A rollback's holds are under the rollback id; a rollback that
// has none (malformed) holds under the move id rather than not at all.
func (v *movesHoldView) hold(rec provision.Record) (status, op string, ok bool) {
	m := rec.Move
	if m == nil || m.Legacy() {
		return "", "", false
	}
	role, kind := v.role(rec)
	rollbackOp := m.ID
	if m.Rollback != nil && m.Rollback.ID != "" {
		rollbackOp = m.Rollback.ID
	}
	switch role {
	case roleSource:
		switch m.State {
		case provision.MoveCutting:
			switch m.CutPhase {
			case provision.CutStopping, provision.CutStopped, provision.CutStaged, provision.CutArchived, provision.CutFlipped:
				return "moving", m.ID, true
			}
		case provision.MoveLanding, provision.MoveCommitted:
			return "moving", m.ID, true
		case provision.MoveReverting:
			return "restoring", m.ID, true
		case provision.MoveAborting:
			// Aborting follows a revert (claim reverting, the row held
			// restoring) or comes before any cut (claim moving, the row
			// serving): only the first is held, and only the claim can say.
			if kind == hostclaim.KindReverting {
				return "restoring", m.ID, true
			}
		case provision.MoveRollingBack, provision.MoveStuck:
			return "restoring", rollbackOp, true
		}
	case roleTarget:
		switch m.State {
		case provision.MoveLanding:
			return "landing", m.ID, true
		case provision.MoveRollingBack, provision.MoveStuck:
			return "releasing", rollbackOp, true
		}
	}
	return "", "", false
}

// drainTimers is the cut's timer drain (moves spec §9.4 "drained", Task 5's
// moveCfg.TimersDrained): after SetHold, it waits until no delivery or
// convergence that could have missed the hold is still running for host.
// No gate (no scheduled work armed on this box) has nothing to drain.
func (v *movesHoldView) drainTimers(ctx context.Context, host string) error {
	return v.gate.Load().Drain(ctx, host)
}

// armedHoldView is the armed worker's hold view, or nil when no worker is
// armed or the Phase 1 switch is off.
func armedHoldView() *movesHoldView {
	w := armedMovesWorker.Load()
	if w == nil || !w.cfg.Enabled {
		return nil
	}
	return w.holds
}

// withMovesHolds gives a reconcile pass the hold view's hooks, unless the
// caller set its own. With the switch off it returns cfg unchanged: main.
func withMovesHolds(cfg provision.ReconcileConfig) provision.ReconcileConfig {
	v := armedHoldView()
	if v == nil || cfg.Hold != nil || cfg.KeepParticipant != nil {
		return cfg
	}
	cfg.Hold, cfg.KeepParticipant = v.hold, v.keep
	if v.written != nil && cfg.HoldWritten == nil {
		cfg.HoldWritten = v.written
	}
	return cfg
}

// armScheduleHolds wires the schedule halves to the hold view: one HostGate
// shared by the runner, the converger (across both of the agent's converge
// loops: there is one Converger) and the cut's drain, and the converger's
// freeze. v nil (switch off) or sd nil (no scheduled work) leaves both as
// main has them.
func armScheduleHolds(sd *scheduleDeps, v *movesHoldView) {
	if sd == nil || v == nil {
		return
	}
	gate := &schedule.HostGate{}
	sd.runner.Gate = gate
	if sd.conv != nil {
		sd.conv.Gate = gate
		sd.conv.Frozen = v.frozen
	}
	v.gate.Store(gate)
}

// refreshMovesHolds builds the hold view once, outside a moves pass: the
// agent's startup converge runs before the first moves pass, and an unbuilt
// view freezes every host. A failed listing leaves it unbuilt; the first
// moves pass that lists builds it.
func refreshMovesHolds(ctx context.Context, s release.Store) {
	v := armedHoldView()
	if v == nil || s == nil {
		return
	}
	recs, err := provision.ListAllRecords(ctx, s)
	if err != nil {
		log.Printf("moves: hold view: list records: %v; rows are kept and timers frozen until a moves pass lists them", err)
		return
	}
	v.refresh(ctx, s, recs)
}

// holdStageRequester is the hold view's HoldWritten: the stage-requested
// marker, under its own throttle, so root re-judges what it staged under
// the new hold (registry.SetHold's contract, G10).
func holdStageRequester(dd string) func(host string) {
	req := markerRequester(dd, func(host, _ string, err error) {
		if err != nil {
			log.Printf("agent: %s: held for a move, but cannot request a root stage pass: %v", host, err)
			return
		}
		log.Printf("agent: %s: held for a move; asked root to re-judge what it staged", host)
	})
	return func(host string) { req(host, "", nil) }
}

// handHold is the box's hand-over of a move's hold to its rollback (ruling
// J3): registry.HandHold under WithHostHoldLock, then written (in production
// holdStageRequester) only when the hand-over landed, so root re-judges what
// it staged under the new hold. A refusal, an absent row or a replay touches
// nothing. Every caller hands a hold over through this wrapper, never through
// registry.HandHold directly, which cannot reach the marker.
//
// validate, when not nil, is the caller's own fresh check (Task 4 fix round
// 1, ruling 1; J3's record check): it runs under the hold lock, after the
// row is re-read and before anything is decided, and an error from it moves
// nothing and is returned. It runs inside WithHostHoldLock, so it must not
// take that lock again; a provision record read is what it is for.
//
// true means the row is held status under toOp: the hand-over landed, or a
// replay found it already done. false with no error means nothing matched:
// the row is held by some other op, unheld, or absent.
func handHold(reg *registry.Registry, written func(host string)) func(host, status, toOp, fromOp string, validate func() error) (bool, error) {
	return func(host, status, toOp, fromOp string, validate func() error) (bool, error) {
		var handed bool
		var err error
		provision.WithHostHoldLock(host, func() {
			var rt *registry.Route
			if rt, err = reg.Get(host); err != nil || rt == nil {
				return
			}
			if validate != nil {
				if err = validate(); err != nil {
					return
				}
			}
			if rt.HoldOp == toOp && rt.Status == status && toOp != "" {
				handed = true
				return
			}
			if handed, err = reg.HandHold(host, status, toOp, fromOp); err != nil || !handed {
				return
			}
			if written != nil {
				written(host)
			}
		})
		return handed, err
	}
}