package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"testing"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The cut guard (Paul's ruling, 2026-10-11): the source refuses to start a
// cut unless the target's latest report states that its box can land the
// move and roll it back (T13), and unless the source itself can recover.
// The refusal comes before anything stops, the writer's lease is taken or
// the claim is sealed, and the move stays ready with the reason recorded.

// errStandIn is what a harness stand-in answers. The stand-ins arm a box
// so that the arming predicates hold, as they do once the wake-source gate
// opens (movesWakeSourceRegistered, Plan 5 Task 9b);
// a test that reaches one has run a step its harness never wired.
var errStandIn = errors.New("harness stand-in: this hook is not wired in this harness")

func standIn(name string) error { return fmt.Errorf("%s: %w", name, errStandIn) }

// armStandIns fills every hook landingArmed, rollbackTargetArmed and
// recoverySourceArmed read that c lacks, with the park fakes (as
// armParkFakes arms them, for role) and a stand-in for each other hook. A hook c already has is
// kept, so a harness's own fakes win.
func armStandIns(c *moveCfg, s release.Store, role string) {
	if c.Guard.DataDir == "" {
		c.Guard = hostguard.Guard{DataDir: c.DataDir}
	}
	if c.Local.DataDir == "" {
		c.Local = park.Local{DataDir: c.DataDir, Box: c.Box}
	}
	if c.ParkLeases == nil {
		c.ParkLeases = claimParkLeases(&fakeParkLeases{})
	}
	if c.ParkView == nil {
		c.ParkView = recordView(s, c.Region, role)
	}
	if c.ParkAuthorize == nil {
		c.ParkAuthorize = allowParkAuthorize
	}
	if c.ParkRunHeld == nil {
		c.ParkRunHeld = movesRunHeld
	}
	if c.ParkLocalAuthorize == nil {
		c.ParkLocalAuthorize = allowLocalAuthorize
	}
	if c.OwnLease == nil {
		c.OwnLease = func(context.Context, string, string) (park.HeldLease, error) { return nil, standIn("OwnLease") }
	}
	if c.RenderHeld == nil {
		c.RenderHeld = func(context.Context, release.Store, registry.Route, string, *provision.MoveInputs, func(string) error) (appconfig.HeldRender, error) {
			return appconfig.HeldRender{}, standIn("RenderHeld")
		}
	}
	if c.ReplaceTimers == nil {
		c.ReplaceTimers = func(context.Context, string, []registry.Timer) error { return standIn("ReplaceTimers") }
	}
	if c.SetReplication == nil {
		c.SetReplication = func(string, string) error { return standIn("SetReplication") }
	}
	if c.SetNextDue == nil {
		c.SetNextDue = func(string, int64) error { return standIn("SetNextDue") }
	}
	if c.RemoveVerifyEntry == nil {
		c.RemoveVerifyEntry = func(context.Context, release.Store, string) error { return standIn("RemoveVerifyEntry") }
	}
	if c.WriteGen == nil {
		c.WriteGen = func(string, string) error { return standIn("WriteGen") }
	}
	if c.ReleaseReady == nil {
		c.ReleaseReady = func(context.Context, release.Store, provision.Record, string) moves.Check {
			return moves.Check{Reason: standIn("ReleaseReady").Error()}
		}
	}
	if c.HandHold == nil {
		c.HandHold = func(string, string, string, string, func() error) (bool, error) { return false, standIn("HandHold") }
	}
	if c.ReleaseHeld == nil {
		c.ReleaseHeld = func(string, string) (bool, error) { return false, standIn("ReleaseHeld") }
	}
	if c.Timers == nil {
		c.Timers = func(string) ([]registry.Timer, error) { return nil, standIn("Timers") }
	}
	if role != roleTarget {
		return
	}
	if c.DeleteHeld == nil {
		c.DeleteHeld = func(string, string) (bool, error) { return false, standIn("DeleteHeld") }
	}
	if c.Fence == nil {
		c.Fence = func(context.Context, string, uint64) error { return standIn("Fence") }
	}
	if c.TimersDrained == nil {
		c.TimersDrained = func(context.Context, string) error { return standIn("TimersDrained") }
	}
	if c.Evidence == nil {
		c.Evidence = func(context.Context, string, string, []string) (moves.StoppedReport, error) {
			return moves.StoppedReport{}, standIn("Evidence")
		}
	}
	if c.FinalCheckpoint == nil {
		c.FinalCheckpoint = func(context.Context, string, string, hostclaim.Holder, func(context.Context) error) error {
			return standIn("FinalCheckpoint")
		}
	}
	if c.LocalPosition == nil {
		c.LocalPosition = func(string) (uint64, error) { return 0, standIn("LocalPosition") }
	}
}

// unarmedTarget is the cut harness's target with the landing or the
// rollback left unwired, as every box is in production until the
// wake-source gate opens (Plan 5 Task 9b).
func (h *cutHarness) unarmedTarget(t *testing.T, unarm func(*moveCfg)) moveCfg {
	t.Helper()
	c := h.target()
	if !landingArmed(c) || !rollbackTargetArmed(c) {
		t.Fatal("fixture: the harness's target is not armed")
	}
	unarm(&c)
	if landingArmed(c) && rollbackTargetArmed(c) {
		t.Fatal("fixture: the target is still armed")
	}
	return c
}

// nothingStopped fails unless the move is ready again with reason recorded
// and the pass did nothing irreversible: no hold, fence, drain, stop
// evidence (which reads the writer's lease) or final checkpoint, no staging
// lease taken and no status set since the counts before the pass, and the
// claim still the moving op under the source with nothing sealed.
func (h *cutHarness) nothingStopped(t *testing.T, reason string, acquired, statuses int) {
	t.Helper()
	m := h.move(t)
	if m.State != provision.MoveReady || m.CutPhase != "" || m.Inputs != nil || m.CutAt != nil || m.Reason != reason {
		t.Fatalf("not refused back to ready with %q: %+v", reason, m)
	}
	if h.ev.calls != 0 || len(h.holds) != 0 || h.fences != 0 || h.drains != 0 || h.checkpoints != 0 || len(h.locations) != 0 {
		t.Fatalf("something stopped: evidence %d, holds %v, fences %d, drains %d, checkpoints %d, leases read %v",
			h.ev.calls, h.holds, h.fences, h.drains, h.checkpoints, h.locations)
	}
	if got := h.lease.acquired; got != acquired {
		t.Fatalf("the refusal took the staging lease: %d acquisitions, want %d", got, acquired)
	}
	if len(h.statuses) != statuses {
		t.Fatalf("the source's row changed status: %v", h.statuses[statuses:])
	}
	c := h.claim(t)
	if c.Op.Exec != "ie-box" || c.Op.Kind != hostclaim.KindMoving || c.Op.Phase != hostclaim.PhaseMoving {
		t.Fatalf("the claim moved: %+v", c.Op)
	}
	if _, err := journalledSeal(c); err == nil {
		t.Fatal("the claim carries a seal")
	}
}

// The harnesses arm both sides, so the cut runs exactly as before: the
// target's report states T13, the source's recovery is armed.
func TestCutGuardBothArmedCutsAsBefore(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	if r := h.targetReport(t); !r.Armed() || r.Checks[moves.CheckArmed] != (moves.Check{OK: true}) {
		t.Fatalf("the armed target's report: %+v", r.Checks[moves.CheckArmed])
	}
	if !recoverySourceArmed(h.cfg) {
		t.Fatal("fixture: the source's recovery is not armed")
	}
	h.cut(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped || m.Reason != "" {
		t.Fatalf("the armed cut did not flip: %+v", m)
	}
}

// A target whose report says it cannot land or roll back is refused at
// requested, before the source's round takes the staging lease: the move
// is ready again, unpinned, with the reason, and nothing stopped. So is a
// report from an older box, which has no T13 at all.
func TestCutGuardRefusesAnUnarmedTarget(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	for name, spoil := range map[string]func(t *testing.T, h *cutHarness){
		"its landing is not armed": func(t *testing.T, h *cutHarness) {
			h.targetPass(t, h.unarmedTarget(t, func(c *moveCfg) { c.OwnLease = nil }))
		},
		"its rollback is not armed": func(t *testing.T, h *cutHarness) {
			h.targetPass(t, h.unarmedTarget(t, func(c *moveCfg) { c.DeleteHeld = nil }))
		},
		"an older box's report has no T13": func(t *testing.T, h *cutHarness) {
			rec := h.read(t, p1Host)
			key := moves.TargetReportKey(moves.ReportPrefix(rec.Account, rec.App, rec.Host, rec.Move.ID))
			body, _, err := h.s.Get(ctx, key)
			if err != nil {
				t.Fatal(err)
			}
			var raw map[string]json.RawMessage
			if err := json.Unmarshal(body, &raw); err != nil {
				t.Fatal(err)
			}
			var checks map[string]json.RawMessage
			if err := json.Unmarshal(raw["checks"], &checks); err != nil {
				t.Fatal(err)
			}
			delete(checks, moves.CheckArmed)
			raw["checks"], _ = json.Marshal(checks)
			body, _ = json.Marshal(raw)
			if err := moves.PutReplace(ctx, h.s, key, body); err != nil {
				t.Fatal(err)
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			spoil(t, h)
			if h.targetReport(t).Armed() {
				t.Fatal("fixture: the report still states T13")
			}
			h.cut(t)
			acquired, statuses := h.lease.acquired, len(h.statuses)
			h.pass(t)
			h.nothingStopped(t, provision.RefusedTargetUnarmed, acquired, statuses)
		})
	}
}

// A source that cannot recover a failed cut refuses it the same way.
func TestCutGuardRefusesAnUnarmedSource(t *testing.T) {
	t.Parallel()
	for name, unarm := range map[string]func(*moveCfg){
		"no recovery open":      func(c *moveCfg) { c.ReleaseHeld = nil },
		"no hand-over":          func(c *moveCfg) { c.HandHold = nil },
		"no landing for return": func(c *moveCfg) { c.ParkRunHeld = nil },
	} {
		t.Run(name, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			unarm(&h.cfg)
			if recoverySourceArmed(h.cfg) {
				t.Fatal("fixture: the source is still armed")
			}
			h.cut(t)
			acquired, statuses := h.lease.acquired, len(h.statuses)
			h.pass(t)
			h.nothingStopped(t, provision.RefusedSourceUnarmed, acquired, statuses)
		})
	}
}

// At admitted (a record an older source admitted without the guard), the
// guard refuses before the step to stopping. The target writes its report
// while the move is ready, so the unarmed one is published before the cut.
func TestCutGuardRefusesAtAdmitted(t *testing.T) {
	ctx := context.Background()
	h := newCutHarness(t)
	h.ready(t)
	h.targetPass(t, h.unarmedTarget(t, func(c *moveCfg) { c.SetNextDue = nil }))
	h.cut(t)
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	now := h.clock.Now()
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
		t.Fatal(err)
	}
	acquired, statuses := h.lease.acquired, len(h.statuses)
	h.pass(t)
	h.nothingStopped(t, provision.RefusedTargetUnarmed, acquired, statuses)
}

// The refusal stands without a loop: an automatic cut refused because the
// source cannot recover is not pinned again or cut again while that holds,
// so later passes write nothing. Once the source is armed, the next passing
// round pins again and the automatic cut proceeds.
func TestCutGuardRefusalStandsUntilArmed(t *testing.T) {
	h := newCutHarness(t)
	h.start(t, "auto")
	rootHolder(t, h.phase1Harness)
	armed := h.cfg
	h.cfg.ReleaseHeld = nil
	h.ev.stopped = false // stop at stopping once armed, to look at it
	h.pass(t)
	h.targetPass(t, h.target())
	for i := 0; i < 3; i++ {
		h.pass(t)
	}
	refused := h.move(t)
	if refused.State != provision.MoveReady || refused.Inputs != nil || refused.Reason != provision.RefusedSourceUnarmed {
		t.Fatalf("the automatic cut was not refused: %+v", refused)
	}
	h.targetPass(t, h.target())
	h.pass(t)
	h.pass(t)
	if m := h.move(t); m.UpdatedAt != refused.UpdatedAt || m.Inputs != nil || m.Reason != provision.RefusedSourceUnarmed {
		t.Fatalf("passes after the refusal wrote the record: %+v", m)
	}
	if h.ev.calls != 0 || len(h.holds) != 0 {
		t.Fatalf("something stopped: evidence %d, holds %v", h.ev.calls, h.holds)
	}
	h.cfg = armed
	h.pass(t) // pins again and requests the cut
	h.pass(t) // admits and starts stopping
	if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping || m.Reason != "" {
		t.Fatalf("the armed source did not cut: %+v", m)
	}
}

// With the switch off, a requested cut on an unarmed source is left as main
// leaves it: nothing refuses, nothing is written.
func TestCutGuardSwitchOffIsMain(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	h.cut(t)
	before := h.move(t)
	h.cfg.Enabled = false
	h.cfg.ReleaseHeld = nil
	h.pass(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutRequested || m.Reason != "" || m.UpdatedAt != before.UpdatedAt {
		t.Fatalf("the switch-off pass wrote the record: %+v", m)
	}
	if h.ev.calls != 0 || len(h.holds) != 0 {
		t.Fatalf("something stopped: evidence %d, holds %v", h.ev.calls, h.holds)
	}
}

// T13 gates readiness (§7.3): an unarmed target's report fails T13 with its
// sentence, and the move stays preparing, so `carlos move status` shows why
// before anyone asks to cut. Armed, it passes and the move is ready.
func TestTargetReportStatesItsArming(t *testing.T) {
	h := newCutHarness(t)
	h.start(t, "manual")
	rootHolder(t, h.phase1Harness)
	h.pass(t)
	h.targetPass(t, h.unarmedTarget(t, func(c *moveCfg) { c.ParkView = nil }))
	r := h.targetReport(t)
	if f := failing(r); len(f) != 1 || f[moves.CheckArmed] != t13NotArmed {
		t.Fatalf("an unarmed target's failing checks: %v", f)
	}
	h.pass(t)
	if m := h.move(t); m.State != provision.MovePreparing {
		t.Fatalf("an unarmed target's move: %+v", m)
	}
	h.targetPass(t, h.target())
	if f := failing(h.targetReport(t)); len(f) != 0 {
		t.Fatalf("an armed target's failing checks: %v", f)
	}
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveReady || m.Inputs == nil {
		t.Fatalf("an armed target's move: %+v", m)
	}
}

// Fix round 1: a refusal whose record write loses to another writer still
// ends the step. Here another writer takes the move back to ready just
// before the refusal's write, so the refusal's CAS conflicts; the pass must
// not carry on with the cut (the source's round and its staging lease at
// requested, the step to stopping at admitted). The next pass re-reads.
func TestCutGuardRefusalEndsTheStepOnAConflict(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	for _, phase := range []string{provision.CutRequested, provision.CutAdmitted} {
		t.Run(phase, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			h.targetPass(t, h.unarmedTarget(t, func(c *moveCfg) { c.OwnLease = nil }))
			h.cut(t)
			g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
			if phase == provision.CutAdmitted {
				now := h.clock.Now()
				if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
					t.Fatal(err)
				}
			}
			inner, conflicted := h.cfg.cutFault, false
			h.cfg.cutFault = func(point string) error {
				if point == "refusal" && !conflicted {
					conflicted = true
					if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveBackToReady(g)); err != nil {
						t.Errorf("the conflicting write: %v", err)
					}
				}
				return inner(point)
			}
			acquired, statuses := h.lease.acquired, len(h.statuses)
			h.pass(t)
			if !conflicted {
				t.Fatal("the pass never reached the refusal")
			}
			// The other writer's decision stands: ready, unpinned, and
			// no reason, since its write carried none.
			h.nothingStopped(t, "", acquired, statuses)
		})
	}
}

// Fix round 1 (ruling on concern 2): the scoped-holder refusal (ruling
// T5-lease) ends the step whatever its write returns, as the cut guard's
// does. Another writer takes the move back to ready just before the
// refusal's write, so its CAS conflicts; the pass ends there. At admitted
// the step must not go on to the cut guard or to stopping (the target is
// left unarmed so that a step that carried on would reach the guard's
// "refusal" point); at requested it must not go on to the admission. The
// staging lease the requested round takes before the holder check is
// main's order (ruling T5-lease) and is not this refusal's.
func TestCutHolderRefusalEndsTheStepOnAConflict(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	for _, phase := range []string{provision.CutRequested, provision.CutAdmitted} {
		t.Run(phase, func(t *testing.T) {
			h := newCutHarness(t)
			h.scoped = true
			h.ready(t)
			if phase == provision.CutAdmitted {
				h.targetPass(t, h.unarmedTarget(t, func(c *moveCfg) { c.OwnLease = nil }))
			}
			h.cut(t)
			g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
			if phase == provision.CutAdmitted {
				now := h.clock.Now()
				if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
					t.Fatal(err)
				}
			}
			inner, conflicted := h.cfg.cutFault, false
			var after []string
			h.cfg.cutFault = func(point string) error {
				if conflicted {
					after = append(after, point)
				}
				if point == "holder-refusal" && !conflicted {
					conflicted = true
					if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveBackToReady(g)); err != nil {
						t.Errorf("the conflicting write: %v", err)
					}
				}
				return inner(point)
			}
			acquired, statuses := h.lease.acquired, len(h.statuses)
			h.pass(t)
			if !conflicted {
				t.Fatal("the pass never reached the holder refusal")
			}
			if len(after) != 0 {
				t.Fatalf("the step carried on after the conflicting refusal: %v", after)
			}
			if phase == provision.CutRequested {
				acquired = h.lease.acquired // the round's own, before the holder check
			}
			h.nothingStopped(t, "", acquired, statuses)
			if m := h.move(t); m.CutAt != nil || m.CutBudgetS != 0 {
				t.Fatalf("the cut was admitted: %+v", m)
			}
		})
	}
}