package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

type plan9D2Attempts struct {
	calls  []string
	starts []OperationStart
	again  error
}

func (a *plan9D2Attempts) record(call string, s OperationStart) {
	a.calls = append(a.calls, call)
	a.starts = append(a.starts, s)
}
func (a *plan9D2Attempts) OpenProbe(_ context.Context, s OperationStart, proposed time.Time) (time.Time, error) {
	a.record("open", s)
	return proposed, nil
}
func (a *plan9D2Attempts) CurrentAttempt(s OperationStart) hostclaim.WakeRetryAttempt {
	a.record("current", s)
	return hostclaim.WakeRetryAttempt{ID: "attempt", InvocationID: "invocation", Session: s.Owner.Session, Box: s.Owner.Box}
}
func (a *plan9D2Attempts) Reserve(_ context.Context, s OperationStart, n int) error {
	a.record(fmt.Sprintf("reserve-%d", n), s)
	return nil
}
func (a *plan9D2Attempts) Failed(_ context.Context, s OperationStart, n int) error {
	a.record(fmt.Sprintf("failed-%d", n), s)
	return nil
}
func (a *plan9D2Attempts) Healthy(ctx context.Context, s OperationStart, n int, _ *exec.Cmd, again func(context.Context) error) error {
	a.record(fmt.Sprintf("healthy-%d", n), s)
	a.again = again(ctx)
	return a.again
}

// plan9D2Inputs is the moves Inputs owner's shape for a landing admission.
type plan9D2Inputs struct{ recoveryInputs }

func (in plan9D2Inputs) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
	selected, err := in.recoveryInputs.Select(ctx, rt, o, p)
	h := o.Claim.Holder
	selected.Admission = hostclaim.Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "box-1", Mode: "cold", NewGeneration: fmt.Sprintf("op-gen-%d", in.f.seq), KeepStamp: h.Stamp > h.Session, Region: "ie", PredecessorRegion: "ie"}
	selected.ReadyPath = "/ready"
	return selected, err
}

// plan9D2Landing turns the recovery fixture's created host into the move
// op's landing at phase with box-1 executing, and admits one writer there
// through the real runtime. Moves C1 owners are bound unless bind is false.
func plan9D2Landing(t *testing.T, phase string, p activator.StartPolicy, bind bool) (*recoveryFixture, *recoveryController, *plan9D2Attempts, *int, hostclaim.Observed) {
	t.Helper()
	f := newRecoveryFixture(t)
	ctx := t.Context()
	if _, err := hostclaim.Update(ctx, f.claims, f.read(), func(c *hostclaim.Claim) error {
		c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: "m1", Phase: phase, Exec: "box-1", Participants: []string{"box-0", "box-1"}, Intent: map[string]string{"to": "ie", "source": "box-0", "target": "box-1"}}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	c := f.controller("boot-1", "controller-1", 4101, "start-1")
	c.wake.Inputs = plan9D2Inputs{recoveryInputs{f}}
	ops, probes := &plan9D2Attempts{}, new(int)
	if bind {
		c.wake.Continue = hostclaim.SessionContinues
		c.wake.Operations = ops
		c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error {
			*probes++
			return nil
		}
	}
	live, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	done := make(chan struct{})
	go func() { <-live.Done(); close(done) }()
	o := f.read()
	c.wake.mu.Lock()
	c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
	c.wake.mu.Unlock()
	admitted, err := c.runtime.Admit(ctx, registry.Route{Host: recoveryHost}, o, p)
	if err != nil {
		t.Fatal(err)
	}
	if admitted.Claim.Op.Kind != hostclaim.KindLanding || admitted.Claim.Holder.WriterBox != "box-1" || admitted.Claim.Holder.WakeRetry != nil {
		t.Fatalf("not a landing admission: %+v", admitted.Claim)
	}
	return f, c, ops, probes, admitted
}

func TestPlan9D2OperationAdmissionPath(t *testing.T) {
	landing := func(phase, exec string) hostclaim.Observed {
		return hostclaim.Observed{Host: recoveryHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{}, Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: "m1", Phase: phase, Exec: exec}}}
	}
	serving := hostclaim.Observed{Host: recoveryHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{}, Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box-1"}}}
	moving := hostclaim.Observed{Host: recoveryHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{}, Op: hostclaim.Op{Kind: hostclaim.KindMoving, ID: "m1", Phase: "probing", Exec: "box-1"}}}
	probe, wake := activator.StartPolicy{Probe: true}, activator.StartPolicy{}
	var refusal *ClaimDependencyError
	// Without the moves C1 owners (either flag off), main's refusal stands.
	unbound := &ClaimWakeAssembly{}
	if err := unbound.admitsOperation(serving, wake, "box-1"); err != nil {
		t.Fatalf("ordinary serving wake: %v", err)
	}
	for name, tc := range map[string]struct {
		o hostclaim.Observed
		p activator.StartPolicy
	}{"serving-probe": {serving, probe}, "landing-probe": {landing("probing", "box-1"), probe}, "landing-wake": {landing("committed", "box-1"), wake}} {
		if err := unbound.admitsOperation(tc.o, tc.p, "box-1"); !errors.As(err, &refusal) {
			t.Fatalf("unbound %s: %v", name, err)
		}
	}
	bound := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}, ProbeReady: func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }}
	if err := bound.admitsOperation(landing("probing", "box-1"), probe, "box-1"); err != nil {
		t.Fatalf("landing probe at probing: %v", err)
	}
	if err := bound.admitsOperation(landing("committed", "box-1"), wake, "box-1"); err != nil {
		t.Fatalf("landing wake at committed: %v", err)
	}
	for name, tc := range map[string]struct {
		o    hostclaim.Observed
		p    activator.StartPolicy
		want error
	}{
		"probe-at-committed": {landing("committed", "box-1"), probe, hostbox.ErrChanged},
		"wake-at-probing":    {landing("probing", "box-1"), wake, hostbox.ErrChanged},
		"probe-at-probed":    {landing("probed", "box-1"), probe, hostbox.ErrChanged},
		"other-executor":     {landing("probing", "box-2"), probe, hostbox.ErrChanged},
		"serving-probe":      {serving, probe, hostwrite.ErrUnavailable},
		"moving-probe":       {moving, probe, hostwrite.ErrUnavailable},
	} {
		if err := bound.admitsOperation(tc.o, tc.p, "box-1"); !errors.Is(err, tc.want) {
			t.Fatalf("bound %s: %v, want %v", name, err, tc.want)
		}
	}
	noProbe := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}}
	if err := noProbe.admitsOperation(landing("probing", "box-1"), probe, "box-1"); !errors.As(err, &refusal) {
		t.Fatalf("a probe without a probe readiness check: %v", err)
	}
}

func TestPlan9D2PrepareBindsTheMovesAttemptsToTheAdmittedProbe(t *testing.T) {
	probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
	_, c, ops, probes, admitted := plan9D2Landing(t, hostclaim.PhaseLandingProbing, probe, true)
	ctx := t.Context()
	s, err := c.runtime.Prepare(ctx, registry.Route{Host: recoveryHost}, admitted, probe)
	if err != nil {
		t.Fatal(err)
	}
	if s.OpenProbe == nil || s.CurrentAttempt == nil || s.Reserve == nil || s.Failed == nil || s.Healthy == nil || s.Ready == nil {
		t.Fatal("the probe start is missing its operation bindings")
	}
	opened := time.Date(2026, 10, 8, 9, 0, 0, 0, time.UTC)
	if got, err := s.OpenProbe(ctx, opened); err != nil || !got.Equal(opened) {
		t.Fatalf("open: %v %v", got, err)
	}
	if err := s.Reserve(ctx, 0); err != nil {
		t.Fatal(err)
	}
	if a := s.CurrentAttempt(); a.ID != "attempt" || a.Session != admitted.Claim.Holder.Session {
		t.Fatalf("attempt: %+v", a)
	}
	if err := s.Failed(ctx, 1); err != nil {
		t.Fatal(err)
	}
	// Healthy's second readiness request is the probe check, which needs the
	// started role: before any launch it refuses and the app is never asked.
	if err := s.Healthy(ctx, 2, exec.Command("/bin/true")); !errors.Is(err, hostwrite.ErrUnavailable) || *probes != 0 {
		t.Fatalf("healthy before launch: %v probes=%d", err, *probes)
	}
	want := recoveryTuple(admitted, c.local)
	for i, start := range ops.starts {
		if start.Admitted.ETag != admitted.ETag || start.Owner != want || !start.Policy.Probe || start.Inputs.ReadyPath != "/ready" || start.Inputs.Admission.ObservedOpID != "m1" {
			t.Fatalf("call %s was bound to %+v", ops.calls[i], start)
		}
	}
	if got := strings.Join(ops.calls, ","); got != "open,reserve-0,current,failed-1,healthy-2" {
		t.Fatalf("calls: %s", got)
	}
}

func TestPlan9D2CommittedWakeUsesTheAttemptsAndServingReadiness(t *testing.T) {
	wake := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second}
	_, c, ops, _, admitted := plan9D2Landing(t, hostclaim.PhaseLandingCommitted, wake, true)
	s, err := c.runtime.Prepare(t.Context(), registry.Route{Host: recoveryHost}, admitted, wake)
	if err != nil {
		t.Fatal(err)
	}
	if s.OpenProbe != nil || s.Reserve == nil {
		t.Fatal("a committed wake is not a probe and still reserves through moves")
	}
	if err := s.Reserve(t.Context(), 0); err != nil || len(ops.calls) != 1 || ops.calls[0] != "reserve-0" {
		t.Fatalf("reserve: %v %v", err, ops.calls)
	}
}

// With the attempts unbound (CARLOS_MOVES_PHASE1 off) an operation writer
// that was admitted anyway cannot be prepared.
func TestPlan9D2PrepareRefusesAnOperationWithoutTheMovesOwners(t *testing.T) {
	wake := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second}
	_, c, _, _, admitted := plan9D2Landing(t, hostclaim.PhaseLandingCommitted, wake, false)
	var refusal *ClaimDependencyError
	if _, err := c.runtime.Prepare(t.Context(), registry.Route{Host: recoveryHost}, admitted, wake); !errors.As(err, &refusal) {
		t.Fatalf("unbound operation prepare: %v", err)
	}
}

// plan9D2Cut is a landing target uk-box admitted at probing over the move's
// source ie-box, with the sealed cut, its journal and the source's stop
// evidence. The admitted claim is stored and uk-box's local record is active
// at the kept stamp.
type plan9D2Cut struct {
	handover        claimCrossBoxHandover
	prior, admitted hostclaim.Observed
	evidence        release.Store
	key             string
	report          moves.StoppedReport
}

func plan9D2CrossBox(t *testing.T, mutate func(*hostclaim.Claim)) *plan9D2Cut {
	t.Helper()
	ctx := t.Context()
	at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
	inc, err := hostclaim.Incarnation("bab", "jam", recoveryHost, at)
	if err != nil {
		t.Fatal(err)
	}
	header := ltx.HeaderSize
	seal := moves.Seal{V: moves.SealVersion, N: 42, FromDB: true, SrcCovered: true, SealedAt: at, LeaseUntil: at.Add(time.Minute),
		Replica: []moves.SealedObject{{Key: "moves/" + recoveryHost + "/m1/replica/0009/" + ltx.FormatFilename(1, 42), VersionID: "v1", Size: int64(header)}},
		Files:   []moves.SealedObject{{Key: "moves/" + recoveryHost + "/m1/files/archive", VersionID: "v2", Size: 10}}}
	key := "moves/bab/jam/" + recoveryHost + "/m1/stopped-source.json"
	journal := map[string]string{hostclaim.JournalSeal: seal.Encode(), hostclaim.JournalCutSession: "15", hostclaim.JournalEvidenceSource: key, hostclaim.JournalSourceStamp: "15", hostclaim.JournalSrcGen: "src"}
	priorClaim := &hostclaim.Claim{V: 2, Host: recoveryHost, Gen: 19,
		Holder: &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + recoveryHost + ".json", Account: "bab", App: "jam", CreatedAt: at, Incarnation: inc, Location: "instances/" + recoveryHost, Generation: "src", Stamp: 18, Session: 15, WriterBox: "ie-box",
			Admissions: []hostclaim.AdmissionRecord{{Session: 15, Box: "ie-box", Generation: "src", Predecessor: "older", Region: "ie", PredecessorRegion: "ie", Mode: "cold"}}},
		Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: "m1", Phase: hostclaim.PhaseLandingProbing, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, Intent: map[string]string{"to": "uk", "source": "ie-box", "target": "uk-box"}, Journal: journal}}
	admittedClaim := *priorClaim
	h := *priorClaim.Holder
	h.Session, h.Generation, h.WriterBox = 20, "m1", "uk-box"
	h.Admissions = append(append([]hostclaim.AdmissionRecord(nil), priorClaim.Holder.Admissions...), hostclaim.AdmissionRecord{Session: 20, Box: "uk-box", Generation: "m1", Predecessor: "src", Region: "uk", PredecessorRegion: "ie", Mode: "cold", Coverage: "sealed"})
	admittedClaim.Holder, admittedClaim.Gen = &h, 20
	if mutate != nil {
		mutate(&admittedClaim)
	}
	if err := admittedClaim.Validate(); err != nil {
		t.Fatal(err)
	}
	claims := release.NewDirStore(t.TempDir())
	body, err := json.Marshal(admittedClaim)
	if err != nil {
		t.Fatal(err)
	}
	claimKey, err := hostclaim.Key(recoveryHost)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := claims.Put(ctx, claimKey, body, ""); err != nil {
		t.Fatal(err)
	}
	admitted, err := hostclaim.Read(ctx, claims, recoveryHost)
	if err != nil {
		t.Fatal(err)
	}
	data := t.TempDir()
	guard := hostguard.Guard{DataDir: data, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: claims}}
	if err := os.MkdirAll(filepath.Join(guard.DataDir, "hostguard"), 0770); err != nil {
		t.Fatal(err)
	}
	if err := guard.Activate(ctx, recoveryHost, "uk-box", nil); err != nil {
		t.Fatal(err)
	}
	report := moves.StoppedReport{V: 1, Role: "source", Box: "ie-box", Incarnation: inc, ClaimGen: 17, Host: recoveryHost, OpID: "m1", FinalTXID: 42, CheckedAt: at,
		Steps: map[string]moves.Check{moves.StepTree: {OK: true}, moves.StepStray: {OK: true}, moves.StepFDs: {OK: true}, moves.StepLease: {OK: true}, moves.StepReplica: {OK: true}}}
	evidence := release.NewDirStore(t.TempDir())
	cut := &plan9D2Cut{handover: claimCrossBoxHandover{Guard: guard, Evidence: evidence, Box: "uk-box"}, prior: hostclaim.Observed{Host: recoveryHost, ETag: "prior", Claim: priorClaim}, admitted: admitted, evidence: evidence, key: key, report: report}
	cut.publish(t)
	return cut
}

func (c *plan9D2Cut) publish(t *testing.T) {
	t.Helper()
	body, err := json.Marshal(c.report)
	if err != nil {
		t.Fatal(err)
	}
	etag := ""
	if _, current, err := c.evidence.Get(t.Context(), c.key); err == nil {
		etag = current
	}
	if _, err := c.evidence.Put(t.Context(), c.key, body, etag); err != nil {
		t.Fatal(err)
	}
}

func TestPlan9D2CrossBoxHandoverAuthorizesFromTheSealedCut(t *testing.T) {
	cut := plan9D2CrossBox(t, nil)
	if err := cut.handover.After(t.Context(), cut.prior, cut.admitted); err != nil {
		t.Fatal(err)
	}
}

func TestPlan9D2CrossBoxHandoverRefusesWithoutExactEvidence(t *testing.T) {
	t.Parallel()
	for name, tc := range map[string]struct {
		claim  func(*hostclaim.Claim)
		prior  func(*hostclaim.Claim)
		report func(*moves.StoppedReport)
	}{
		"cut-session":   {prior: func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalCutSession] = "14" }},
		"no-evidence":   {prior: func(c *hostclaim.Claim) { delete(c.Op.Journal, hostclaim.JournalEvidenceSource) }},
		"bad-seal":      {prior: func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalSeal] = "{}" }},
		"same-box":      {prior: func(c *hostclaim.Claim) { c.Holder.WriterBox = "uk-box" }},
		"not-probing":   {prior: func(c *hostclaim.Claim) { c.Op.Phase = "restored" }},
		"report-box":    {report: func(r *moves.StoppedReport) { r.Box = "other-box" }},
		"report-op":     {report: func(r *moves.StoppedReport) { r.OpID = "m0" }},
		"report-n":      {report: func(r *moves.StoppedReport) { r.FinalTXID = 41 }},
		"report-role":   {report: func(r *moves.StoppedReport) { r.Role = "target" }},
		"incomplete":    {report: func(r *moves.StoppedReport) { delete(r.Steps, moves.StepReplica) }},
		"coverage":      {claim: func(c *hostclaim.Claim) { c.Holder.Admissions[1].Coverage = "" }},
		"predecessor":   {claim: func(c *hostclaim.Claim) { c.Holder.Admissions[1].Predecessor = "older" }},
		"stamp-changed": {claim: func(c *hostclaim.Claim) { c.Holder.Stamp = 20 }},
	} {
		t.Run(name, func(t *testing.T) {
			cut := plan9D2CrossBox(t, tc.claim)
			if tc.prior != nil {
				prior := *cut.prior.Claim
				holder := *prior.Holder
				prior.Holder = &holder
				prior.Op.Journal = map[string]string{}
				for k, v := range cut.prior.Claim.Op.Journal {
					prior.Op.Journal[k] = v
				}
				tc.prior(&prior)
				cut.prior.Claim = &prior
			}
			if tc.report != nil {
				tc.report(&cut.report)
				cut.publish(t)
			}
			if err := cut.handover.After(t.Context(), cut.prior, cut.admitted); err == nil {
				t.Fatalf("%s was authorized", name)
			}
		})
	}
	// The kept stamp must be this box's active landing installation.
	cut := plan9D2CrossBox(t, nil)
	cut.handover.Guard.DataDir = t.TempDir()
	if err := cut.handover.After(t.Context(), cut.prior, cut.admitted); err == nil {
		t.Fatal("a box without the landing installation's local record was authorized")
	}
}

func TestPlan9D2AdmissionHandoverDispatch(t *testing.T) {
	var calls []string
	local := func(context.Context, hostclaim.Observed, hostclaim.Observed) error {
		calls = append(calls, "local")
		return nil
	}
	cross := func(context.Context, hostclaim.Observed, hostclaim.Observed) error {
		calls = append(calls, "cross")
		return nil
	}
	prior := func(kind, writer string) hostclaim.Observed {
		return hostclaim.Observed{Host: recoveryHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{WriterBox: writer}, Op: hostclaim.Op{Kind: kind, ID: "m1"}}}
	}
	ctx := t.Context()
	for _, tc := range []struct {
		hook  func(context.Context, hostclaim.Observed, hostclaim.Observed) error
		prior hostclaim.Observed
		want  string
	}{
		{claimAdmissionHandover("uk-box", local, cross), prior(hostclaim.KindLanding, "ie-box"), "cross"},
		{claimAdmissionHandover("uk-box", local, cross), prior(hostclaim.KindLanding, "uk-box"), "local"},
		{claimAdmissionHandover("uk-box", local, cross), prior(hostclaim.KindLanding, ""), "local"},
		{claimAdmissionHandover("uk-box", local, cross), prior(hostclaim.KindServing, "ie-box"), "local"},
		{claimAdmissionHandover("uk-box", local, nil), prior(hostclaim.KindLanding, "ie-box"), "local"},
	} {
		calls = nil
		if err := tc.hook(ctx, tc.prior, hostclaim.Observed{}); err != nil || len(calls) != 1 || calls[0] != tc.want {
			t.Fatalf("%+v went to %v (%v), want %s", tc.prior.Claim, calls, err, tc.want)
		}
	}
}

func TestPlan9D2ProductionPredecessorUsesSourceEvidenceBeforeAdmission(t *testing.T) {
	local, remote := 0, 0
	w := &ClaimWakeAssembly{Stop: func(context.Context, hostclaim.Observed) error { local++; return nil }}
	w.BeforeAdmission = claimPredecessorStop(w, "target", func(context.Context, hostclaim.Observed) error { remote++; return hostbox.ErrChanged })
	o := hostclaim.Observed{Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{WriterBox: "source"}, Op: hostclaim.Op{Kind: hostclaim.KindLanding}}}
	if err := w.stopPredecessor(t.Context(), o); !errors.Is(err, hostbox.ErrChanged) || local != 0 || remote != 1 {
		t.Fatalf("source proof did not gate admission: local=%d remote=%d err=%v", local, remote, err)
	}
	o.Claim.Holder.WriterBox = "target"
	if err := w.stopPredecessor(t.Context(), o); err != nil || local != 1 || remote != 1 {
		t.Fatalf("local predecessor did not use stop: %d %d %v", local, remote, err)
	}
}