package host

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

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/genselect"
	"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/moves/archive"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/superfly/ltx"
)

var poWake = activator.StartPolicy{MaxStarts: 5, Window: 30 * time.Second}

// committedAfterProbe is the owners' fixture after a passing probe and the
// commit: the move id admitted and published, the claim at
// landing/committed, the record committed, the row open.
func committedAfterProbe(t *testing.T) (*poFixture, hostclaim.Observed) {
	t.Helper()
	f := newPOFixture(t)
	admitted, in := f.selectAndAdmit()
	if err := f.owners.Materialize(t.Context(), admitted, in, newFenceWork()); err != nil {
		t.Fatal(err)
	}
	c := *f.read().Claim
	c.Op.Phase = hostclaim.PhaseLandingCommitted
	f.putClaim(&c)
	f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveCommitted, ""
	f.route.Status, f.route.HoldOp = "ready", ""
	return f, f.read()
}

// Task 7 (Task 5's concern): the landing wake after the commit is a cold
// admission into a fresh generation from the certified remote, continuing
// the admitted session's generation; it carries the ordinary launch, no
// probe variable, and is evidenced, materialized, start-checked and
// published as serving's are.
func TestMovesProbeOwnersCommittedWake(t *testing.T) {
	f, o := committedAfterProbe(t)
	ctx := t.Context()
	h := o.Claim.Holder
	in, err := f.owners.Select(ctx, f.route, o, poWake)
	if err != nil {
		t.Fatal(err)
	}
	a := in.Admission
	if a.ObservedOpID != poOp || a.ObservedPhase != hostclaim.PhaseLandingCommitted || a.ObservedSession != h.Session || a.ObservedGeneration != poOp ||
		a.Mode != "cold" || a.NewGeneration == "" || a.NewGeneration == poOp || a.KeepStamp || a.Coverage != "" || a.Region != "uk" || a.PredecessorRegion != "uk" {
		t.Fatalf("the committed admission: %+v", a)
	}
	if in.Prepared || in.Baseline == nil || in.Baseline.Generation != poOp || in.Baseline.MaxTXID != 42 || a.RemoteMaxTXID != 42 {
		t.Fatalf("the committed baseline: %+v", in.Baseline)
	}
	// Computed from the actual launch (fix round 2): its environment's
	// report under the probe's normalisation, and its release; here they
	// are the pins'.
	if want := committedPinsIdentity(f.rec.Move.Inputs); in.Identity != want {
		t.Fatalf("the committed wake's identity %+v, want %+v", in.Identity, want)
	}
	for _, kv := range in.Env {
		if strings.HasPrefix(kv, "CARLOS_MOVE_PROBE=") {
			t.Fatalf("the committed wake carries %s", kv)
		}
	}
	if err := f.owners.AdmissionEvidence(ctx, o, a, in.Baseline); err != nil {
		t.Fatal(err)
	}
	stale := *in.Baseline
	stale.MaxTXID = 41
	if err := f.owners.AdmissionEvidence(ctx, o, a, &stale); !errors.Is(err, hostbox.ErrChanged) {
		t.Fatalf("a stale baseline: %v", err)
	}
	admitted := f.admit(o, in)
	f.writeLocal(poOp, admitted.Claim.Holder.Generation, f.inc)
	if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
		t.Fatalf("materialize: %v", err)
	}
	root := "instances/" + poHost + "/g-" + a.NewGeneration + "/0009/" + ltx.FormatFilename(1, 43)
	f.bucket.Put("data", root, f.ltxHeader(1, 43))
	pub := ClaimPublication{Sync: func(context.Context) error { return nil }, Check: func(context.Context) error { return nil }}
	if _, err := f.owners.PublishFenced(ctx, admitted, in, pub); err != nil {
		t.Fatal(err)
	}
	if err := f.owners.PublicationEvidence(ctx, admitted); err != nil {
		t.Fatal(err)
	}
	// A probe is never admitted at committed, and a committed wake never at
	// probing; a record that is not committed refuses.
	if _, err := f.owners.Select(ctx, f.route, o, poProbe); err == nil {
		t.Fatal("a probe at committed")
	}
	f.rec.Move.State = provision.MoveLanding
	if _, err := f.owners.Select(ctx, f.route, o, poWake); err == nil {
		t.Fatal("a committed wake over a landing record")
	}
}

// filesGen reads the local .files-gen.
func (f *poFixture) filesGen() moves.FilesGen {
	f.t.Helper()
	raw, err := os.ReadFile(filepath.Join(f.files, archive.MarkerName))
	if err != nil {
		f.t.Fatal(err)
	}
	var fg moves.FilesGen
	if err := json.Unmarshal(raw, &fg); err != nil {
		f.t.Fatal(err)
	}
	return fg
}

// restore is ColdRestore's local effect on a fresh admission: <dbPath>.gen
// names the admitted generation.
func (f *poFixture) restore(gen string) {
	f.t.Helper()
	if err := hostbox.WriteGenerationMarker(f.route.DBPath+".gen", gen); err != nil {
		f.t.Fatal(err)
	}
}

// publishFresh puts the fresh generation's first root and publishes it
// fenced, as the claim's sequence does once the session's root is durable.
func (f *poFixture) publishFresh(admitted hostclaim.Observed, in ClaimLaunchInputs) error {
	f.t.Helper()
	gen := admitted.Claim.Holder.Generation
	f.bucket.Put("data", "instances/"+poHost+"/g-"+gen+"/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 }}
	_, err := f.owners.PublishFenced(f.t.Context(), admitted, in, pub)
	return err
}

// restartedProbePublished runs a dead first attempt and a restarted attempt
// that passes and publishes its fresh generation.
func restartedProbePublished(t *testing.T) (*poFixture, hostclaim.Observed, ClaimLaunchInputs) {
	t.Helper()
	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)
	f.restore(restarted.Claim.Holder.Generation)
	if err := f.owners.Materialize(ctx, restarted, again, newFenceWork()); err != nil {
		t.Fatalf("the restarted probe's start: %v", err)
	}
	return f, restarted, again
}

// commit moves the fixture to the commit: the claim at landing/committed,
// the record committed, the row open.
func (f *poFixture) commit() hostclaim.Observed {
	f.t.Helper()
	c := *f.read().Claim
	c.Op.Phase = hostclaim.PhaseLandingCommitted
	f.putClaim(&c)
	f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveCommitted, ""
	f.route.Status, f.route.HoldOp = "ready", ""
	return f.read()
}

// Task 7 fix round 1, finding 1: a restarted probe's successful publication
// refreshes .files-gen to the published generation, under the writer's
// fence, after the files on disk are proved the landing's. After the commit,
// a process death and a wake, and then several more, each start passes
// Task 4's unweakened start check.
func TestMovesProbeOwnersRestartedProbeThenCommittedWakes(t *testing.T) {
	f, restarted, again := restartedProbePublished(t)
	ctx := t.Context()
	if err := f.publishFresh(restarted, again); err != nil {
		t.Fatal(err)
	}
	fresh := restarted.Claim.Holder.Generation
	if fg := f.filesGen(); fg.Op != fresh || fg.Incarnation != f.inc {
		t.Fatalf(".files-gen after the restarted probe's publication: %+v, want %s", fg, fresh)
	}
	o := f.commit()
	for i := 0; i < 3; i++ {
		in, err := f.owners.Select(ctx, f.route, o, poWake)
		if err != nil {
			t.Fatalf("wake %d: %v", i, err)
		}
		admitted := f.admit(o, in)
		f.restore(admitted.Claim.Holder.Generation)
		if err := f.owners.Materialize(ctx, admitted, in, newFenceWork()); err != nil {
			t.Fatalf("wake %d's start after the probe's %s: %v", i, fresh, err)
		}
		if err := f.publishFresh(admitted, in); err != nil {
			t.Fatalf("wake %d's publication: %v", i, err)
		}
		if fg := f.filesGen(); fg.Op != admitted.Claim.Holder.Generation {
			t.Fatalf("wake %d's .files-gen: %+v", i, fg)
		}
		o = f.read()
	}
}

// A files directory that is not the one the landing placed fails the
// publication: no marker is written and .files-gen is not refreshed.
func TestMovesProbeOwnersPublicationRefusesMismatchedFiles(t *testing.T) {
	f, restarted, again := restartedProbePublished(t)
	before := f.bucket.LatestID("data", f.markerKey())
	// Another directory in its place: made while the placed one still
	// exists, so it cannot reuse its inode.
	old := f.files + ".old"
	if err := os.Rename(f.files, old); err != nil {
		t.Fatal(err)
	}
	if err := os.Mkdir(f.files, 0o700); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(old) })
	f.writeLocal(poOp, restarted.Claim.Holder.Generation, f.inc)
	if err := f.publishFresh(restarted, again); err == nil {
		t.Fatal("published over files the landing did not place")
	}
	if f.bucket.LatestID("data", f.markerKey()) != before {
		t.Fatal("a marker was written")
	}
	if fg := f.filesGen(); fg.Op != poOp {
		t.Fatalf(".files-gen was refreshed: %+v", fg)
	}
}

// platform-1e (claim Plan 10 rev 5): the moves owner's input policy binds
// the committed wake's spawn to Move.Inputs, as MovesClaimPolicy binds the
// landing's rows: a committed cold admission with the pins' launch identity
// is accepted, and refused once the pins differ.
func TestMovesProbeOwnersInputPolicyBindsTheCommittedWake(t *testing.T) {
	f, o := committedAfterProbe(t)
	ctx := t.Context()
	in, err := f.owners.Select(ctx, f.route, o, poWake)
	if err != nil {
		t.Fatal(err)
	}
	admitted := f.admit(o, in)
	c := admitted.Claim
	sel := hostbox.Selection{Host: poHost, Action: "spawn", 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}
	policy := MovesClaimPolicyWithInputs(f.owners.Records, f.owners.InputPolicy)
	if err := f.owners.InputPolicy(ctx, c, sel, in.Identity); err != nil {
		t.Fatalf("matching pins: %v", err)
	}
	if err := policy(ctx, c, sel, in.Identity); err != nil {
		t.Fatalf("matching pins through the policy: %v", err)
	}
	for name, mutate := range map[string]func(){
		"another release":    func() { f.rec.Move.Inputs.Release = "v9" },
		"another env report": func() { f.rec.Move.Inputs.EnvReport = "sha256:other" },
	} {
		pins := *f.rec.Move.Inputs
		mutate()
		if err := f.owners.InputPolicy(ctx, c, sel, in.Identity); err == nil {
			t.Fatalf("%s: accepted", name)
		}
		if err := policy(ctx, c, sel, in.Identity); err == nil {
			t.Fatalf("%s: accepted through the policy", name)
		}
		f.rec.Move.Inputs = &pins
	}
	if err := f.owners.InputPolicy(ctx, c, sel, f.launch.Identity); err == nil {
		t.Fatal("the ordinary identity was accepted")
	}
	_ = genselect.Marker{}
	_ = hostwrite.Identity{}
}

// writeFilesGenOnly writes .files-gen alone.
func (f *poFixture) writeFilesGenOnly(op string) {
	f.t.Helper()
	body, _ := json.Marshal(moves.FilesGen{Op: op, Incarnation: f.inc})
	if err := os.WriteFile(filepath.Join(f.files, archive.MarkerName), body, 0o600); err != nil {
		f.t.Fatal(err)
	}
}

// committedWake admits, restores and materializes one committed wake.
func (f *poFixture) committedWake(o hostclaim.Observed) (hostclaim.Observed, ClaimLaunchInputs, error) {
	f.t.Helper()
	in, err := f.owners.Select(f.t.Context(), f.route, o, poWake)
	if err != nil {
		f.t.Fatal(err)
	}
	admitted := f.admit(o, in)
	f.restore(admitted.Claim.Holder.Generation)
	return admitted, in, f.owners.Materialize(f.t.Context(), admitted, in, newFenceWork())
}

// Task 7 fix round 2, item 1: a publication interrupted between its marker
// write and the .files-gen refresh (local g1, published g2) is repaired
// before the next admission's (g3) start check: the published marker names
// g3's cold predecessor, and the files on disk are proved the landing's, so
// .files-gen moves to g2 and the start check, unweakened, passes.
func TestMovesProbeOwnersRepairAnInterruptedFilesGenRefresh(t *testing.T) {
	f, restarted, again := restartedProbePublished(t)
	if err := f.publishFresh(restarted, again); err != nil {
		t.Fatal(err)
	}
	g1 := restarted.Claim.Holder.Generation
	o := f.commit()
	w1, in1, err := f.committedWake(o)
	if err != nil {
		t.Fatal(err)
	}
	if err := f.publishFresh(w1, in1); err != nil {
		t.Fatal(err)
	}
	g2 := w1.Claim.Holder.Generation
	// The crash: the marker names g2, .files-gen still g1.
	f.writeFilesGenOnly(g1)
	w2, in2, err := f.committedWake(f.read())
	if err != nil {
		t.Fatalf("the next admission after an interrupted refresh: %v", err)
	}
	if fg := f.filesGen(); fg.Op != g2 {
		t.Fatalf(".files-gen was not repaired to %s: %+v", g2, fg)
	}
	if err := f.publishFresh(w2, in2); err != nil {
		t.Fatal(err)
	}
	if fg := f.filesGen(); fg.Op != w2.Claim.Holder.Generation {
		t.Fatalf("after the next publication: %+v", fg)
	}
}

// Item 1: files that are not the landing's are never repaired over; the
// start check refuses as before.
func TestMovesProbeOwnersRepairRefusesMismatchedFiles(t *testing.T) {
	f, restarted, again := restartedProbePublished(t)
	if err := f.publishFresh(restarted, again); err != nil {
		t.Fatal(err)
	}
	g1 := restarted.Claim.Holder.Generation
	o := f.commit()
	w1, in1, err := f.committedWake(o)
	if err != nil {
		t.Fatal(err)
	}
	if err := f.publishFresh(w1, in1); err != nil {
		t.Fatal(err)
	}
	old := f.files + ".old"
	if err := os.Rename(f.files, old); err != nil {
		t.Fatal(err)
	}
	if err := os.Mkdir(f.files, 0o700); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(old) })
	f.writeFilesGenOnly(g1)
	if _, _, err := f.committedWake(f.read()); !errors.Is(err, moves.ErrStartFiles) {
		t.Fatalf("a repair over files the landing did not place: %v", err)
	}
	if fg := f.filesGen(); fg.Op != g1 {
		t.Fatalf(".files-gen was rewritten: %+v", fg)
	}
}

// Item 2: the committed wake's input check judges the real launch. A changed
// launch environment with the record's pins unchanged is refused, by Select
// and by the input policy on the identity that launch would carry; matching
// inputs are accepted.
func TestMovesProbeOwnersCommittedWakeJudgesTheRealLaunch(t *testing.T) {
	f, o := committedAfterProbe(t)
	ctx := t.Context()
	in, err := f.owners.Select(ctx, f.route, o, poWake)
	if err != nil {
		t.Fatalf("matching inputs: %v", err)
	}
	c := o.Claim
	sel := hostbox.Selection{Host: poHost, Action: "spawn", 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 := f.owners.InputPolicy(ctx, c, sel, in.Identity); err != nil {
		t.Fatalf("matching inputs through the input policy: %v", err)
	}
	f.launch.Env = []string{"A=2"}
	if _, err := f.owners.Select(ctx, f.route, o, poWake); err == nil {
		t.Fatal("a changed launch environment was selected")
	}
	changed := committedLaunchIdentity(envDigest(poEnvKey, f.launch.Env), f.launch.Version)
	if err := f.owners.InputPolicy(ctx, c, sel, changed); err == nil {
		t.Fatal("the changed launch's identity was accepted")
	}
}

// committedPinsIdentity is the identity a committed launch matching the pins
// carries.
func committedPinsIdentity(p *provision.MoveInputs) hostwrite.Identity {
	return hostwrite.Identity{Config: committedEnvPrefix + p.EnvReport, Release: p.Release}
}

// Task 7 fix round 3: the repair verifies content, never the directory's
// identity alone. With g2's publication interrupted before its refresh, a
// file modified, deleted or added inside the original directory (the same
// inode) refuses the repair: .files-gen stays g1 and the start check
// refuses. Matching contents still recover (TestMovesProbeOwnersRepairAnInterruptedFilesGenRefresh).
func TestMovesProbeOwnersRepairVerifiesTheFilesContent(t *testing.T) {
	for name, change := range map[string]func(dir string) error{
		"a modified file": func(dir string) error { return os.WriteFile(filepath.Join(dir, "a.txt"), []byte("stale"), 0o600) },
		"a deleted file":  func(dir string) error { return os.Remove(filepath.Join(dir, "a.txt")) },
		"an added file":   func(dir string) error { return os.WriteFile(filepath.Join(dir, "b.txt"), []byte("new"), 0o600) },
	} {
		t.Run(name, func(t *testing.T) {
			f, restarted, again := restartedProbePublished(t)
			if err := f.publishFresh(restarted, again); err != nil {
				t.Fatal(err)
			}
			g1 := restarted.Claim.Holder.Generation
			w1, in1, err := f.committedWake(f.commit())
			if err != nil {
				t.Fatal(err)
			}
			if err := f.publishFresh(w1, in1); err != nil {
				t.Fatal(err)
			}
			f.writeFilesGenOnly(g1)
			if err := change(f.files); err != nil {
				t.Fatal(err)
			}
			if _, _, err := f.committedWake(f.read()); !errors.Is(err, moves.ErrStartFiles) {
				t.Fatalf("a repair over changed contents: %v", err)
			}
			if fg := f.filesGen(); fg.Op != g1 {
				t.Fatalf(".files-gen was repaired over changed contents: %+v", fg)
			}
		})
	}
}

// publishedP is the fixture at a published committed generation P: a
// restarted probe published (P), the commit, .files-gen naming P and P's
// files digest journalled.
func publishedP(t *testing.T) (*poFixture, string) {
	t.Helper()
	f, restarted, again := restartedProbePublished(t)
	if err := f.publishFresh(restarted, again); err != nil {
		t.Fatal(err)
	}
	f.commit()
	return f, restarted.Claim.Holder.Generation
}

// Final review, ruling 1: a committed wake G1 that failed before its marker
// leaves .files-gen at the published P while the next admission G2's cold
// predecessor is G1. The repair walks G1 back to P through unpublished cold
// admissions on this box and region and, the files matching P's recorded
// digest, sets .files-gen to G1: G2 starts (StartCheck unchanged) and
// publishes.
func TestMovesProbeOwnersRecoverACommittedWakeThatFailedBeforePublishing(t *testing.T) {
	f, p := publishedP(t)
	if _, _, err := f.committedWake(f.read()); err != nil { // G1, then it dies before PutMarker
		t.Fatal(err)
	}
	if fg := f.filesGen(); fg.Op != p {
		t.Fatalf("the fixture: %+v", fg)
	}
	g2, in2, err := f.committedWake(f.read())
	if err != nil {
		t.Fatalf("G2 after an unpublished G1: %v", err)
	}
	if err := f.publishFresh(g2, in2); err != nil {
		t.Fatal(err)
	}
	if fg := f.filesGen(); fg.Op != g2.Claim.Holder.Generation {
		t.Fatalf(".files-gen after G2's publication: %+v", fg)
	}
}

// Ruling 1: two failed wakes, G1 and G2, then G3 starts.
func TestMovesProbeOwnersRecoverTwoFailedCommittedWakes(t *testing.T) {
	f, _ := publishedP(t)
	f.committedWake(f.read()) // G1, dies before PutMarker
	f.committedWake(f.read()) // G2, dies before PutMarker
	if _, _, err := f.committedWake(f.read()); err != nil {
		t.Fatalf("G3 after two unpublished wakes: %v", err)
	}
}

// Ruling 1: a file changed after P's publication refuses the walk's repair;
// the start check refuses G2 and .files-gen stays P.
func TestMovesProbeOwnersCommittedWalkVerifiesContent(t *testing.T) {
	f, p := publishedP(t)
	if _, _, err := f.committedWake(f.read()); err != nil { // G1
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(f.files, "a.txt"), []byte("changed after P"), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, _, err := f.committedWake(f.read()); !errors.Is(err, moves.ErrStartFiles) {
		t.Fatalf("G2 over changed files: %v", err)
	}
	if fg := f.filesGen(); fg.Op != p {
		t.Fatalf(".files-gen was repaired over changed files: %+v", fg)
	}
}