package host

import (
	"context"
	"errors"
	"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/hostwrite"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

// plan10J2Rollback is the rollback id a restoring op carries in its intent.
const plan10J2Rollback = "rb_abcdefghijklmnopqrstuvwxyz"

// plan10J2Op is the source's recovery op as box-1 (the source) executes it.
func plan10J2Op(kind, phase string) hostclaim.Op {
	op := hostclaim.Op{Kind: kind, ID: "m1", Phase: phase, Exec: "box-1", Participants: []string{"box-1", "box-2"}, Intent: map[string]string{"to": "uk", "source": "box-1", "target": "box-2"}}
	switch kind {
	case hostclaim.KindRestoring:
		op.Intent["rollback"] = plan10J2Rollback
	case hostclaim.KindReverting:
		op.Journal = map[string]string{hostclaim.JournalRevPhase: phase}
	}
	return op
}

func plan10J2Observed(kind, phase, exec string) hostclaim.Observed {
	op := plan10J2Op(kind, phase)
	op.Exec = exec
	return hostclaim.Observed{Host: recoveryHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{WriterBox: "box-1"}, Op: op}}
}

// plan10J2Recovery turns the recovery fixture's created host into the
// source's recovery op at phase, with box-1 executing, and admits one writer
// there through the real runtime. The moves C1 owners are bound.
func plan10J2Recovery(t *testing.T, kind, phase string, p activator.StartPolicy) (*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 = plan10J2Op(kind, phase)
		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)
	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 != kind || admitted.Claim.Op.Phase != phase || admitted.Claim.Holder.WriterBox != "box-1" || admitted.Claim.Holder.WakeRetry != nil {
		t.Fatalf("not a %s admission at %s: %+v", kind, phase, admitted.Claim)
	}
	return f, c, ops, probes, admitted
}

// The recovery probe is admitted at probing and nowhere else; an ordinary
// wake of a recovery op is not admitted here at any phase, although
// hostbox.WakePhase's own table grants probed and opened to a wake, because
// the assembly admits exactly the probe shape for these kinds.
func TestPlan10J2RecoveryAdmissionPath(t *testing.T) {
	probe, wake := activator.StartPolicy{Probe: true}, activator.StartPolicy{}
	bound := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}, ProbeReady: func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }}
	unbound := &ClaimWakeAssembly{}
	noProbe := &ClaimWakeAssembly{Operations: &plan9D2Attempts{}}
	var refusal *ClaimDependencyError
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		t.Run(kind, func(t *testing.T) {
			if err := bound.admitsOperation(plan10J2Observed(kind, "probing", "box-1"), probe, "box-1"); err != nil {
				t.Fatalf("recovery probe at probing: %v", err)
			}
			for name, a := range map[string]*ClaimWakeAssembly{"unbound": unbound, "no probe readiness": noProbe} {
				if err := a.admitsOperation(plan10J2Observed(kind, "probing", "box-1"), probe, "box-1"); !errors.As(err, &refusal) {
					t.Fatalf("%s: %v", name, err)
				}
			}
			for _, phase := range []string{"probed", "opened", "claimed", "released", "recovered", ""} {
				if err := bound.admitsOperation(plan10J2Observed(kind, phase, "box-1"), probe, "box-1"); !errors.Is(err, hostbox.ErrChanged) {
					t.Fatalf("probe at %q: %v", phase, err)
				}
			}
			for _, phase := range []string{"probing", "probed", "opened"} {
				if err := bound.admitsOperation(plan10J2Observed(kind, phase, "box-1"), wake, "box-1"); !errors.As(err, &refusal) {
					t.Fatalf("ordinary wake at %q was not refused: %v", phase, err)
				}
			}
			if err := bound.admitsOperation(plan10J2Observed(kind, "probing", "box-2"), probe, "box-1"); !errors.Is(err, hostbox.ErrChanged) {
				t.Fatalf("another executor: %v", err)
			}
			lost := plan10J2Observed(kind, "probing", "box-1")
			lost.Claim.Lost = []string{"box-1"}
			if err := bound.admitsOperation(lost, probe, "box-1"); !errors.Is(err, hostbox.ErrChanged) {
				t.Fatalf("a box declared lost: %v", err)
			}
		})
	}
	for _, kind := range []string{hostclaim.KindMoving, hostclaim.KindPurging, hostclaim.KindRelocating} {
		if err := bound.admitsOperation(plan10J2Observed(kind, "probing", "box-1"), probe, "box-1"); !errors.As(err, &refusal) && !errors.Is(err, hostwrite.ErrUnavailable) {
			t.Fatalf("%s probe: %v", kind, err)
		}
	}
	if err := bound.admitsOperation(plan10J2Observed(hostclaim.KindLanding, "committed", "box-1"), wake, "box-1"); err != nil {
		t.Fatalf("landing wake at committed: %v", err)
	}
}

func TestPlan10J2PrepareBindsTheAttemptsAndProbeReadinessToARecoveryProbe(t *testing.T) {
	probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		t.Run(kind, func(t *testing.T) {
			_, c, ops, probes, admitted := plan10J2Recovery(t, kind, hostclaim.PhaseProbing, probe)
			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 recovery 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.Session != admitted.Claim.Holder.Session {
				t.Fatalf("attempt: %+v", a)
			}
			// The second readiness request is the probe check, which needs
			// the started role's cgroup: before any launch it refuses and
			// the app is never asked.
			if err := s.Healthy(ctx, 1, 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.Admitted.Claim.Op.Kind != kind || start.Admitted.Claim.Op.Phase != hostclaim.PhaseProbing {
					t.Fatalf("call %s was bound to %+v", ops.calls[i], start)
				}
			}
			if got := strings.Join(ops.calls, ","); got != "open,reserve-0,current,healthy-1" {
				t.Fatalf("calls: %s", got)
			}
			// A probe of this kind without the probe readiness is refused.
			c.wake.ProbeReady = nil
			if _, err := c.wake.readiness(admitted, probe, registry.Route{Host: recoveryHost}, ClaimLaunchInputs{}, c.wake.get(recoveryHost), nil); err == nil {
				t.Fatal("a recovery probe got readiness without its launch")
			}
		})
	}
}

// A recovery writer admitted at probing keeps working one adjacent forward
// step at a time (claimSameWriter under Continue): probing to probed to
// opened. A skipped or backward phase, or any change of op, kind, session,
// stamp, generation or executor ends the attempt's session.
func TestPlan10J2AttemptAdvancesOnlyAlongTheRecoveryRow(t *testing.T) {
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		other := hostclaim.KindReverting
		if kind == hostclaim.KindReverting {
			other = hostclaim.KindRestoring
		}
		at := func(phase string, mutate func(*hostclaim.Claim)) hostclaim.Observed {
			o := plan10J2Observed(kind, phase, "box-1")
			o.Claim.Holder = &hostclaim.Holder{Incarnation: "inc", Stamp: 10, Session: 11, Generation: "g1", WriterBox: "box-1"}
			if mutate != nil {
				mutate(o.Claim)
			}
			return o
		}
		start := func(phase string) *claimWakeAttempt {
			return &claimWakeAttempt{observed: at(phase, nil)}
		}
		t.Run(kind+"/forward", func(t *testing.T) {
			a := start("probing")
			for _, phase := range []string{"probing", "probed", "opened"} {
				if !a.advance(at(phase, nil), hostclaim.SessionContinues) {
					t.Fatalf("did not advance to %s", phase)
				}
			}
		})
		for name, tc := range map[string]struct {
			from, to string
			mutate   func(*hostclaim.Claim)
		}{
			"skipped phase":   {"probing", "opened", nil},
			"backward phase":  {"probed", "probing", nil},
			"before the row":  {"probing", "claimed", nil},
			"changed session": {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Session++ }},
			"changed stamp":   {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Stamp++ }},
			"changed gen":     {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Generation = "fresh" }},
			"changed inc":     {"probing", "probed", func(c *hostclaim.Claim) { c.Holder.Incarnation = "other" }},
			"changed op":      {"probing", "probed", func(c *hostclaim.Claim) { c.Op.ID = "m2" }},
			"changed exec":    {"probing", "probed", func(c *hostclaim.Claim) { c.Op.Exec = "box-2" }},
			"changed kind":    {"probing", "probed", func(c *hostclaim.Claim) { c.Op.Kind = other }},
			"landing kind":    {"probing", "probed", func(c *hostclaim.Claim) { c.Op.Kind = hostclaim.KindLanding }},
		} {
			t.Run(kind+"/"+name, func(t *testing.T) {
				a := start(tc.from)
				before := a.observed
				if a.advance(at(tc.to, tc.mutate), hostclaim.SessionContinues) {
					t.Fatalf("%s advanced", name)
				}
				if a.observed.Claim.Op.Phase != before.Claim.Op.Phase {
					t.Fatalf("a refused advance moved the checkpoint to %s", a.observed.Claim.Op.Phase)
				}
			})
		}
	}
}

// Teardown of an admitted recovery writer follows the same row: a stop
// names the session at any phase it went on to reach, never an earlier one.
func TestPlan10J2TeardownReachesForwardOnlyAlongTheRecoveryRow(t *testing.T) {
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		at := func(phase string) hostclaim.Observed {
			o := plan10J2Observed(kind, phase, "box-1")
			o.Claim.Holder = &hostclaim.Holder{Incarnation: "inc", Stamp: 10, Session: 11, Generation: "g1", WriterBox: "box-1", Location: "loc"}
			return o
		}
		reaches := hostclaim.SessionReaches
		if !sameStopWriter(at("probing"), at("opened"), reaches) || !sameStopWriter(at("probed"), at("opened"), reaches) {
			t.Fatalf("%s: a stop did not reach the later phase", kind)
		}
		if sameStopWriter(at("opened"), at("probing"), reaches) || sameStopWriter(at("probed"), at("probing"), reaches) {
			t.Fatalf("%s: a stop named an earlier phase", kind)
		}
		if sameStopWriter(at("claimed"), at("probing"), reaches) {
			t.Fatalf("%s: a stop crossed from before the row", kind)
		}
	}
}

// An admission error for a recovery probe releases its pending lease,
// exactly as a landing's does (before any prelaunch record), and a stop of an
// attempt with no prelaunch record never runs the local root's stop.
func TestPlan10J2AdmissionFailureReleasesTheRecoveryLease(t *testing.T) {
	t.Parallel()
	probe := activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second, Probe: true}
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		t.Run(kind, func(t *testing.T) {
			f := newRecoveryFixture(t)
			ctx := t.Context()
			if _, err := hostclaim.Update(ctx, f.claims, f.read(), func(c *hostclaim.Claim) error {
				c.Op = plan10J2Op(kind, hostclaim.PhaseProbing)
				return nil
			}); err != nil {
				t.Fatal(err)
			}
			c := f.controller("boot-1", "controller-1", 4101, "start-1")
			c.wake.Inputs = plan9D2Inputs{recoveryInputs{f}}
			c.wake.Continue = hostclaim.SessionContinues
			c.wake.Operations = &plan9D2Attempts{}
			c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
			stops := 0
			c.wake.Stop = func(context.Context, hostclaim.Observed) error { stops++; return nil }
			dead, kill := context.WithCancel(context.Background())
			kill()
			done := make(chan struct{})
			close(done)
			o := f.read()
			c.wake.mu.Lock()
			c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: dead, cancel: kill, done: done}
			c.wake.mu.Unlock()
			if _, err := c.runtime.Admit(ctx, registry.Route{Host: recoveryHost}, o, probe); err == nil {
				t.Fatal("admission under a dead lease succeeded")
			}
			if c.wake.get(recoveryHost) != nil {
				t.Fatal("the failed recovery admission left its pending lease")
			}
			if stops != 0 {
				t.Fatalf("the local stop ran %d times for an attempt with no launch", stops)
			}
		})
	}
}

// The recovery's local artifact retirement stays fail-closed: the landing's
// retire-artifacts is bound to the landing writer (hostguard.LandingWriter),
// and a recovery op is not one. Recover has nothing to re-drive for it.
func TestPlan10J2ArtifactRetirementRefusesUnadmittedRecovery(t *testing.T) {
	f := newRecoveryFixture(t)
	dataDir := f.guard.DataDir
	r := claimArtifactRetirement{Local: park.Local{Claims: f.claims, Box: "box-1", DataDir: dataDir}, Continue: hostclaim.SessionContinues}
	in := ClaimLaunchInputs{DBPath: filepath.Join(dataDir, "state", "data.db")}
	for _, kind := range []string{hostclaim.KindRestoring, hostclaim.KindReverting} {
		o := plan10J2Observed(kind, "probing", "box-1")
		o.Claim.Holder = &hostclaim.Holder{Incarnation: "inc", Stamp: 10, Session: 11, Generation: "g1", WriterBox: "box-1"}
		if err := r.Retire(t.Context(), o, in, "m1"); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("%s retire: %v", kind, err)
		}
		if err := r.Recover(t.Context(), o, in); !errors.Is(err, hostbox.ErrChanged) {
			t.Fatalf("%s recover: %v", kind, err)
		}
	}
}

// A clean return records its current inputs at probing before the held
// render and launch. No configured phase is needed. The adapter must call
// recoveryProbeRequest.Record first; movesrecoveryprobe.go supplies the adapter.
// This test exercises the existing recording transition and the policy checks
// for the recorded configuration, release and launch.
func TestPlan10J2CleanReturnRecordsRecoveryInputsBeforeRenderAndLaunch(t *testing.T) {
	g := provision.MoveGuard{ID: mpOp, Incarnation: mpInc}
	rec := rpRestoringRecord(provision.RbSourceProbing)
	rec.Move.Rollback.Dirty, rec.Move.Rollback.RecoveryInputs = false, nil
	c := rpRestoringClaim(hostclaim.PhaseProbing)
	read := (&mpReader{rec: rec}).read
	check := func(action string, id hostwrite.Identity) error {
		p := hostbox.Policy{Moves: MovesClaimPolicy(read), Held: func(context.Context, string) (string, error) { return rpRB, nil }}
		return p.Check(t.Context(), c, rpSelection(c, action), id)
	}
	recovered := rpRecovery()
	rpRefused(t, check("spawn", rpLaunch(recovered)), "no recorded recovery inputs")
	if err := provision.MoveRecordRecoveryInputs(g, rpRB, *recovered)(rec); err != nil {
		t.Fatalf("recording at probing: %v", err)
	}
	if rec.Move.Rollback.RecoveryInputs == nil || rec.Move.Rollback.Dirty {
		t.Fatalf("the record is %+v", rec.Move.Rollback)
	}
	if err := check("held-config", rpIdentity("held-config")); err != nil {
		t.Fatalf("render of the recorded inputs: %v", err)
	}
	if err := check("held-release", rpIdentity("held-release")); err != nil {
		t.Fatalf("release of the recorded inputs: %v", err)
	}
	if err := check("spawn", rpLaunch(recovered)); err != nil {
		t.Fatalf("the launch of the recorded inputs: %v", err)
	}
	other := *recovered
	other.Release = "v9.9.9"
	rpRefused(t, check("spawn", rpLaunch(&other)), "the launch is not the recovery's")
	// Recording is not legal at any other source phase.
	for _, phase := range []string{provision.RbSourceRestored, provision.RbSourceProbed, provision.RbSourceOpened} {
		later := rpRestoringRecord(phase)
		later.Move.Rollback.Dirty, later.Move.Rollback.RecoveryInputs = false, nil
		if err := provision.MoveRecordRecoveryInputs(g, rpRB, *recovered)(later); err == nil {
			t.Fatalf("recording was legal at %s", phase)
		}
	}
}