package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/superfly/ltx"
)

// fenceWork is an admitted spawn selection whose fence can be closed: once
// shut, every check refuses. It counts the checks it answered.
type fenceWork struct {
	shut   *bool
	checks *int
}

func newFenceWork() fenceWork { return fenceWork{shut: new(bool), checks: new(int)} }

func (w fenceWork) With(_ context.Context, _ hostwrite.Identity, fn func() error) error {
	*w.checks++
	if *w.shut {
		return hostbox.ErrChanged
	}
	return fn()
}

func (f *poFixture) markerKey() string { return "instances/" + poHost + "/generation.json" }

func (f *poFixture) currentMarker() (genselect.Marker, bool) {
	f.t.Helper()
	body, ok := f.bucket.Current("data", f.markerKey())
	var m genselect.Marker
	if ok && json.Unmarshal(body, &m) != nil {
		f.t.Fatalf("marker %s", body)
	}
	return m, ok
}

// selectAndAdmit runs Select and the admission CAS for the claim as it is.
func (f *poFixture) selectAndAdmit() (hostclaim.Observed, ClaimLaunchInputs) {
	f.t.Helper()
	o := f.read()
	in, err := f.owners.Select(f.t.Context(), f.route, o, poProbe)
	if err != nil {
		f.t.Fatal(err)
	}
	return f.admit(o, in), in
}

// Finding 1: AdmissionEvidence, immediately before the admission's CAS,
// refuses a landing past its deadline, a record attempt outside what T11
// admitted, an install the record does not vouch for, and a baseline whose
// pin is not the build's.
func TestMovesProbeOwnersAdmissionEvidenceChecksTheLanding(t *testing.T) {
	past := time.Date(2026, 10, 9, 11, 59, 0, 0, time.UTC)
	for name, tc := range map[string]struct {
		rec  func(*provision.Record)
		plan func(*genselect.Plan)
	}{
		"past the cut deadline":     {rec: func(r *provision.Record) { r.Move.CutDeadline = &past }},
		"past the landing deadline": {rec: func(r *provision.Record) { r.Move.LandingDeadline = &past }},
		"no deadline":               {rec: func(r *provision.Record) { r.Move.CutDeadline = nil }},
		"no attempt counted":        {rec: func(r *provision.Record) { r.Move.Landing.Attempts = 0 }},
		"beyond the admitted":       {rec: func(r *provision.Record) { r.Move.Landing.Attempts = 4 }},
		"another installed db": {rec: func(r *provision.Record) {
			r.Move.Landing.InstalledDBSHA256 = "sha256:" + hex.EncodeToString(make([]byte, 32))
		}},
		"no installed db": {rec: func(r *provision.Record) { r.Move.Landing.InstalledDBSHA256 = "" }},
		"another pin version": {plan: func(p *genselect.Plan) {
			pin := *p.Pin
			pin.SnapshotVersion = "other"
			p.Pin = &pin
		}},
		"no pin": {plan: func(p *genselect.Plan) { p.Pin = nil }},
	} {
		t.Run(name, func(t *testing.T) {
			f := newPOFixture(t)
			ctx := t.Context()
			o := f.read()
			in, err := f.owners.Select(ctx, f.route, o, poProbe)
			if err != nil {
				t.Fatal(err)
			}
			if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); err != nil {
				t.Fatalf("the unchanged landing: %v", err)
			}
			plan := *in.Baseline
			if tc.rec != nil {
				tc.rec(f.rec)
			}
			if tc.plan != nil {
				tc.plan(&plan)
			}
			if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, &plan); err == nil {
				t.Fatal("admitted")
			}
		})
	}
}

// Finding 2: the prepared marker is written only under the admitted
// writer's live fence, checked by the store before every request attempt,
// and carries the install manifest's digest, verified against install.json.
func TestMovesProbeOwnersPreparedMarkerIsFenced(t *testing.T) {
	f := newPOFixture(t)
	ctx := t.Context()
	admitted, in := f.selectAndAdmit()
	// Group D's prepared Publish with no fence held for the session.
	if err := f.owners.Publish(ctx, admitted, in); err == nil {
		t.Fatal("an unfenced prepared marker was published")
	}
	if _, ok := f.currentMarker(); ok {
		t.Fatal("a marker was written without the fence")
	}
	fence := newFenceWork()
	if err := f.owners.Materialize(ctx, admitted, in, fence); err != nil {
		t.Fatal(err)
	}
	m, ok := f.currentMarker()
	install, _ := f.bucket.Current("data", moves.InstallManifestKey(moves.GenerationPrefix("instances/"+poHost, poOp)))
	sum := sha256.Sum256(install)
	if !ok || m.Op != poOp || m.ManifestSHA256 != hex.EncodeToString(sum[:]) || m.SnapshotTXID != 42 {
		t.Fatalf("marker %+v", m)
	}
	// Group D's call now replays the marker this session wrote.
	written := f.bucket.LatestID("data", f.markerKey())
	if err := f.owners.Publish(ctx, admitted, in); err != nil || f.bucket.LatestID("data", f.markerKey()) != written {
		t.Fatalf("replay: %v", err)
	}
	if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
		t.Fatal(err)
	}
	// Another writer replaced it and the session's fence has closed: nothing
	// is written over it.
	f.bucket.Put("data", f.markerKey(), mustJSON(t, genselect.Marker{Op: "other", Incarnation: f.inc, Host: poHost, Box: "uk-box", InstalledAt: time.Unix(3, 0).UTC()}))
	other := f.bucket.LatestID("data", f.markerKey())
	*fence.shut = true
	if err := f.owners.Publish(ctx, admitted, in); err == nil || f.bucket.LatestID("data", f.markerKey()) != other {
		t.Fatalf("a marker written after the fence closed: %v", err)
	}
}

// Finding 2: an install.json that is not the journalled manifest is not
// published, and the probe does not start.
func TestMovesProbeOwnersPreparedMarkerNeedsTheInstallManifest(t *testing.T) {
	f := newPOFixture(t)
	other := f.manifest
	other.BuiltAt = time.Unix(9, 0).UTC()
	f.bucket.Put("data", moves.InstallManifestKey(moves.GenerationPrefix("instances/"+poHost, poOp)), mustJSON(t, other))
	admitted, in := f.selectAndAdmit()
	if err := f.owners.Materialize(t.Context(), admitted, in, newFenceWork()); err == nil {
		t.Fatal("published over an install.json the journal does not name")
	}
	if _, ok := f.currentMarker(); ok {
		t.Fatal("a marker was written")
	}
}

// Finding 3: the env pin is the report of the launch Select selected: a
// launch environment that changed is refused though the pin and every other
// input are unchanged; the probe-only CARLOS_MOVE_PROBE is not part of it.
func TestMovesProbeOwnersEnvPinIsTheSelectedLaunch(t *testing.T) {
	f := newPOFixture(t)
	if _, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe); err != nil {
		t.Fatalf("the pinned launch: %v", err)
	}
	f.launch.Env = []string{"A=2", "CARLOS_MOVE_PROBE=0"}
	_, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe)
	var changed *ProbeInputsChanged
	if !errors.As(err, &changed) || changed.Reason != provision.RollbackConfigChanged {
		t.Fatalf("a changed launch environment: %v", err)
	}
	f.launch.Env = []string{"A=1"}
	if _, err := f.owners.Select(t.Context(), f.route, f.read(), poProbe); err != nil {
		t.Fatalf("without the probe-only variable: %v", err)
	}
}

// policyWork is a held selection that asks the real MovesClaimPolicy, as
// hostbox.WorkSource does, before it runs anything.
type policyWork struct {
	f      *poFixture
	action string
}

func (w policyWork) With(ctx context.Context, id hostwrite.Identity, fn func() error) error {
	o := w.f.read()
	c := o.Claim
	s := hostbox.Selection{Host: poHost, Action: w.action, Box: "uk-box", Region: "uk", Captured: hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp},
		Kind: c.Op.Kind, Op: c.Op.ID, Phase: c.Op.Phase, Generation: c.Holder.Generation, Session: c.Holder.Session}
	if err := MovesClaimPolicy(func(context.Context, string, string, string) (*provision.Record, error) { return w.f.rec, nil })(ctx, c, s, id); err != nil {
		return err
	}
	return fn()
}

// Finding 4: a render whose etags are not the pins is ProbeInputsChanged
// (the landing's RollbackConfigChanged), decided before any held work is
// selected or written, through the real policy, which would otherwise
// refuse the write with its own error.
func TestMovesProbeOwnersChangedConfigThroughThePolicy(t *testing.T) {
	f := newPOFixture(t)
	ctx := t.Context()
	f.owners.Works = func(_ context.Context, _, action string, _ hostguard.Captured) (hostwrite.Work, error) {
		f.order = append(f.order, "select:"+action)
		return policyWork{f: f, action: action}, nil
	}
	admitted, in := f.selectAndAdmit()
	if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
		t.Fatalf("the pinned render through the policy: %v", err)
	}
	f.order = nil
	f.renderOverlay = "e-changed"
	err := f.owners.Materialize(ctx, admitted, in, newFenceWork())
	var changed *ProbeInputsChanged
	var refusal *MovesPolicyRefusal
	if !errors.As(err, &changed) || changed.Reason != provision.RollbackConfigChanged || errors.As(err, &refusal) {
		t.Fatalf("a changed render: %v", err)
	}
	for _, step := range f.order {
		if step == "select:held-config" || step == "select:held-release" {
			t.Fatalf("a held work was selected for a changed render: %v", f.order)
		}
	}
}

// Finding 8, ruled order: admit, publish the verified prepared marker under
// the writer's fence (superseding the earlier stay's marker by If-Match),
// then the start check, then the spawn. Fix round 2: the earlier stay is
// really in the admission history, a cold admission on this box and region
// that does not descend from the move's admission; it is still an earlier
// stay, superseded, never taken for a restart of this landing.
func TestMovesProbeOwnersSupersedeTheEarlierStaysMarker(t *testing.T) {
	f := newPOFixture(t)
	ctx := t.Context()
	c := f.poFlipClaim()
	c.Holder.Admissions = []hostclaim.AdmissionRecord{
		{Session: 5, Box: "uk-box", Generation: "earlier", Predecessor: "", Region: "uk", Mode: "cold"},
		{Session: 15, Box: "ie-box", Generation: "src", Predecessor: "earlier", Region: "ie", PredecessorRegion: "uk", Mode: "cold"},
	}
	if err := c.Validate(); err != nil {
		t.Fatal(err)
	}
	f.putClaim(c)
	f.bucket.Put("data", f.markerKey(), mustJSON(t, genselect.Marker{Op: "earlier", Incarnation: f.inc, Host: poHost, Box: "uk-box", InstalledAt: time.Unix(1, 0).UTC(), Pin: genselect.Pin{SnapshotTXID: 3}}))
	admitted, in := f.selectAndAdmit()
	if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
		t.Fatalf("the first probe over an earlier stay's marker: %v", err)
	}
	if m, _ := f.currentMarker(); m.Op != poOp {
		t.Fatalf("the earlier stay's marker was not superseded: %+v", m)
	}
	if err := f.owners.Publish(ctx, admitted, in); err != nil {
		t.Fatalf("Group D's prepared publication: %v", err)
	}
	if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
		t.Fatalf("publication evidence: %v", err)
	}
}

// Fix round 2: a genuine restart after a crash is still this landing's. The
// first attempt published the move id; the restarted attempt fresh1
// published its own marker after its root and crashed. The next attempt's
// baseline is the certified remote from fresh1, whose chain reaches the move
// id, and fresh1's marker is kept, not superseded by the prepared one. Since
// Task 7's fix round 1 fresh1's publication refreshed .files-gen to fresh1,
// so that attempt passes Task 4's files rule too.
func TestMovesProbeOwnersRecogniseARestartAfterACrash(t *testing.T) {
	f := newPOFixture(t)
	ctx := t.Context()
	first, in := f.selectAndAdmit()
	if err := f.owners.Materialize(ctx, first, in, newFenceWork()); err != nil {
		t.Fatal(err)
	}
	again, err := f.owners.Select(ctx, f.route, first, poProbe)
	if err != nil {
		t.Fatal(err)
	}
	restarted := f.admit(first, again)
	fresh := again.Admission.NewGeneration
	f.restore(fresh)
	if err := f.owners.Materialize(ctx, restarted, again, newFenceWork()); err != nil {
		t.Fatal(err)
	}
	f.bucket.Put("data", "instances/"+poHost+"/g-"+fresh+"/0009/"+ltx.FormatFilename(1, 44), f.ltxHeader(1, 44))
	pub := ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { return nil }}
	if _, err := f.owners.PublishFenced(ctx, restarted, again, pub); err != nil {
		t.Fatal(err)
	}
	// The crash: the next record attempt admits again over fresh1.
	next, err := f.owners.Select(ctx, f.route, restarted, poProbe)
	if err != nil {
		t.Fatal(err)
	}
	if next.Baseline.Generation != fresh || next.Baseline.Marker == nil || next.Baseline.Marker.Op != fresh || next.Baseline.MaxTXID != 44 {
		t.Fatalf("the restart's baseline is not the certified remote of %s: %+v", fresh, next.Baseline)
	}
	admitted := f.admit(restarted, next)
	f.restore(admitted.Claim.Holder.Generation)
	written := f.bucket.LatestID("data", f.markerKey())
	err = f.owners.Materialize(ctx, admitted, next, newFenceWork())
	if m, _ := f.currentMarker(); m.Op != fresh || f.bucket.LatestID("data", f.markerKey()) != written {
		t.Fatalf("the restart's marker was superseded: %+v", m)
	}
	if err != nil {
		t.Fatalf("the second restart: %v", err)
	}
}

// Fix round 2, the predicate itself: a generation is this landing's only
// with positive proof, the chain of cold admissions on this box and region
// back from it reaching the move id.
func TestMovesProbeOwnersLandingMarkerNeedsDescent(t *testing.T) {
	f := newPOFixture(t)
	v := probeLanding{op: poOp, manifest: f.manifest, holder: &hostclaim.Holder{Incarnation: f.inc, Admissions: []hostclaim.AdmissionRecord{
		{Session: 5, Box: "uk-box", Generation: "earlier", Region: "uk", Mode: "cold"},
		{Session: 15, Box: "ie-box", Generation: "src", Predecessor: "earlier", Region: "ie", Mode: "cold"},
		{Session: 20, Box: "uk-box", Generation: poOp, Predecessor: "src", Region: "uk", Mode: "cold", Coverage: "sealed"},
		{Session: 21, Box: "uk-box", Generation: "f1", Predecessor: poOp, Region: "uk", Mode: "cold"},
		{Session: 22, Box: "uk-box", Generation: "f2", Predecessor: "f1", Region: "uk", Mode: "cold"},
		{Session: 23, Box: "uk-box", Generation: "orphan", Predecessor: "gone", Region: "uk", Mode: "cold"},
		{Session: 24, Box: "other-box", Generation: "f3", Predecessor: "f2", Region: "uk", Mode: "cold"},
	}}}
	marker := func(op string) *genselect.Marker {
		return &genselect.Marker{Op: op, Incarnation: f.inc, Host: poHost, Box: "uk-box"}
	}
	for op, want := range map[string]bool{"f1": true, "f2": true, "earlier": false, "src": false, "orphan": false, "f3": false, "unknown": false} {
		if got := f.owners.landingMarker(marker(op), poHost, v, ""); got != want {
			t.Errorf("%s: %v, want %v", op, got, want)
		}
	}
}

// Finding 8: a history trimmed so that the admitted generation's ancestry
// back to the published marker cannot be proved is refused by the start
// check, after the publication and before any spawn. StartCheck is not
// weakened.
func TestMovesProbeOwnersTrimmedHistoryIsRefused(t *testing.T) {
	f := newPOFixture(t)
	ctx := t.Context()
	first, in := f.selectAndAdmit()
	if err := f.owners.Materialize(ctx, first, in, newFenceWork()); err != nil {
		t.Fatal(err)
	}
	// Two restarted attempts: fresh1 from the move id, fresh2 from fresh1.
	again, err := f.owners.Select(ctx, f.route, first, poProbe)
	if err != nil {
		t.Fatal(err)
	}
	second := f.admit(first, again)
	third, err := f.owners.Select(ctx, f.route, second, poProbe)
	if err != nil {
		t.Fatal(err)
	}
	admitted := f.admit(second, third)
	// Trim the history to the last admission only (fresh2 from fresh1).
	c := *admitted.Claim
	h := *c.Holder
	h.Admissions = h.Admissions[len(h.Admissions)-1:]
	c.Holder = &h
	f.putClaim(&c)
	f.writeLocal(poOp, h.Generation, f.inc)
	trimmed := f.read()
	before := f.bucket.LatestID("data", f.markerKey())
	// C8 refuses the unprovable history first (item 1: the establishment
	// runs before the start check, and a chain it cannot prove is
	// ErrNoEvidence); behind it the start check stands unweakened.
	err = f.owners.Materialize(ctx, trimmed, third, newFenceWork())
	if !errors.Is(err, hostfiles.ErrNoEvidence) {
		t.Fatalf("a trimmed history through C8: %v", err)
	}
	f.owners.Files = noopFiles{}
	err = f.owners.Materialize(ctx, trimmed, third, newFenceWork())
	if !errors.Is(err, moves.ErrStartMarker) {
		t.Fatalf("a trimmed history: %v", err)
	}
	if f.bucket.LatestID("data", f.markerKey()) != before {
		t.Fatal("the verified marker was replaced")
	}
}