package host

import (
	"context"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
)

// Ruling R-PARK (settings-4 Task 8b): <holder.Location>/build-state.json is
// live state beside the lease, the generation marker's twin. Every path
// that lists the holder location passes it over where it passes over the
// marker; anywhere else it is residue like any other object.

// A landing onto a location that already holds a build state (an earlier
// stay's) parks the earlier stay's generations, passes the build state over
// as it passes the marker over, and completes: the install then places the
// moved state over it.
func TestALandingOntoALocationHoldingABuildStateCompletes(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
	h.flip(t)
	keys := h.residue(t)
	dst := cutRoot + "/build-state.json"
	if _, err := h.vs.PutIfAbsent(context.Background(), "data-uk", dst, []byte(p1StateBody("v3"))); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 4 && h.read(t, p1Host).Move.LandingPhase != provision.LandRestored; i++ {
		h.land(t)
	}
	h.installed(t)
	if got := h.f.buckets["data-uk"][dst].body; got != p1StateBody("v6") {
		t.Fatalf("the location holds %q, want the moved v6", got)
	}
	for _, k := range keys {
		if _, ok := h.f.buckets["data-uk"][k]; ok && k != cutRoot+"/generation.json" {
			t.Fatalf("the earlier stay's %s was not parked", k)
		}
	}
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, "parked/") && strings.Contains(k, "/build-state.json@") {
			t.Fatalf("the location's build state was parked: %s", k)
		}
	}
}

// A dirty return whose source location holds the pre-move build state
// classifies the location, parks the pre-move generations and installs;
// the returned state is placed over the pre-move one.
func TestADirtyReturnOverAPreMoveBuildStateClassifies(t *testing.T) {
	t.Parallel()
	h := newReturnHarness(t, true)
	ctx := context.Background()
	if _, err := h.vs.PutIfAbsent(ctx, "data-ie", cutRoot+"/build-state.json", []byte(p1StateBody("v5"))); err != nil {
		t.Fatal(err)
	}
	key := moves.BuildStateKey(p1Host, p1RB)
	ver, err := h.vs.PutIfAbsent(ctx, "data-ie", key, []byte(p1StateBody("v8")))
	if err != nil {
		t.Fatal(err)
	}
	h.seal.State = []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(p1StateBody("v8")))}}
	if err := h.seal.Validate(); err != nil {
		t.Fatal(err)
	}
	pinned := h.read(t, p1Host).Move.Inputs
	if refused, err := h.install(t); err != nil || refused != "" {
		t.Fatalf("install: %q %v", refused, err)
	}
	h.installedReturn(t, pinned)
	if got := h.f.buckets["data-ie"][cutRoot+"/build-state.json"].body; got != p1StateBody("v8") {
		t.Fatalf("the source's location holds %q, want the returned v8", got)
	}
}

// TestMovesHolderBuildStateMatchesTheEngine keeps the host's twin key in
// step with the park engine's, in the spirit of
// TestMovesParkLockMatchesTheEngine: park.Decide is the oracle. In the
// authoritative region, with a classifier that places every key (so a
// skip is the engine's alone), Decide makes no candidate of a one-key
// census exactly when the host says the key is the holder location's
// build state. The same file name anywhere else is a candidate.
func TestMovesHolderBuildStateMatchesTheEngine(t *testing.T) {
	for name, bare := range map[string]bool{"scoped": false, "bare": true} {
		t.Run(name, func(t *testing.T) {
			f := newEvFixture(t)
			if bare {
				f.bare(t)
			}
			obs := f.observe(t)
			view := park.View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: evMoveID,
				Marker: park.Ref{Key: f.location + "/generation.json"},
				GenerationOf: func(r park.Ref) (string, bool) {
					if g, ok := landingGenerationOf(f.location, r.Key); ok {
						return g, true
					}
					return "stale", true
				}}
			other := strings.Repeat("ab", 32)
			keys := []string{
				f.location + "/build-state.json",
				f.location + "/g-old/build-state.json",
				f.location + "/build-state.json.old",
				f.location + "/state/build-state.json",
				"instances/" + evHost + "/" + other + "/build-state.json",
				"moves/" + evHost + "/build-state.json",
				"moves/" + evHost + "/" + evMoveID + "/state/build-state.json",
			}
			if bare {
				keys = append(keys, "instances/"+evHost+"/"+f.inc+"/build-state.json")
			} else {
				keys = append(keys, "instances/"+evHost+"/build-state.json")
			}
			skipped := 0
			for _, key := range keys {
				d, err := park.Decide(obs, view, []lease.ObjectVersion{{Key: key, VersionID: "v1"}, {Key: key, VersionID: "v0"}}, f.inc)
				if err != nil {
					t.Fatalf("%s: %v", key, err)
				}
				engine, mirror := len(d.Candidates) == 0, movesHolderBuildState(f.location, key)
				if engine != mirror {
					t.Errorf("%s: the engine skips it %v, the host's twin key says %v", key, engine, mirror)
				}
				if engine {
					skipped++
				}
			}
			if skipped != 1 {
				t.Errorf("the engine skipped %d keys, want the holder location's build state alone", skipped)
			}
		})
	}
}

// The close need not vouch for the live location's build state: it is the
// location's own, as the marker is. A lane's copy written after the close
// still has an unknown writer.
func TestTheLiveLocationsBuildStateNeedsNoVouching(t *testing.T) {
	for name, bare := range map[string]bool{"scoped": false, "bare": true} {
		t.Run(name, func(t *testing.T) {
			f := newEvFixture(t)
			if bare {
				f.bare(t)
			}
			f.lane(evMoveID, "move.json")
			f.bucket.Put("data", f.genKey(evMoveID), []byte("live"))
			f.bucket.Put("data", f.location+"/generation.json", []byte("marker"))
			f.bucket.Put("data", f.location+"/build-state.json", []byte("older"))
			f.bucket.Put("data", f.location+"/build-state.json", []byte("current"))
			if _, _, err := f.provider().VerifyRetirement(t.Context(), f.observe(t), f.census(t)); err != nil {
				t.Fatalf("VerifyRetirement refused the live build state: %v", err)
			}
		})
	}
	t.Run("a lane's state written after the close", func(t *testing.T) {
		f := newEvFixture(t)
		f.rec.LastMove.At = time.Date(2026, 10, 4, 0, 0, 0, 0, time.UTC) // before every write
		f.lane("mv_unknown", moves.BuildStateSuffix)
		bothRefuse(t, f, true)
	})
}

// returnBeforeInstall is a rollback harness whose landing stopped at
// building, before the install placed anything, with the forward seal's
// state edited by edit.
func returnBeforeInstall(t *testing.T, edit func(h *rollbackHarness, s *moves.SealedObject)) *rollbackHarness {
	t.Helper()
	h := &rollbackHarness{landingHarness: newLandingHarness(t)}
	h.f.keepHistory = true
	storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
	h.flip(t)
	h.landAt = "after-record-" + provision.LandBuilding
	h.crashLand(t)
	h.uncover(t)
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		s, err := journalledSeal(c)
		if err != nil {
			t.Fatal(err)
		}
		if len(s.State) != 1 {
			t.Fatalf("seal state %+v", s.State)
		}
		edit(h, &s.State[0])
		c.Op.Journal[hostclaim.JournalSeal] = s.Encode()
	})
	h.rollBack(t)
	h.rollbackPass(t)
	return h
}

// Task 8 re-review minor 1: a return before the install reads the forward
// seal's state only at the move's own lane key, and only at the size the
// seal names, as copyAsGiven reads every other sealed object.
func TestAReturnBeforeTheInstallReadsOnlyItsOwnSealedState(t *testing.T) {
	t.Parallel()
	refused := func(t *testing.T, h *rollbackHarness) {
		t.Helper()
		rb := h.rb(t)
		if rb.TargetPhase == provision.RbTargetReturned {
			t.Fatalf("the return sealed: %+v", rb)
		}
		if _, ok := h.f.buckets["data-ie"][moves.BuildStateKey(p1Host, rb.ID)]; ok {
			t.Fatal("a build state went back")
		}
	}
	t.Run("the control returns", func(t *testing.T) {
		h := returnBeforeInstall(t, func(*rollbackHarness, *moves.SealedObject) {})
		if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturned {
			t.Fatalf("rollback %+v", rb)
		}
	})
	t.Run("another lane's key", func(t *testing.T) {
		h := returnBeforeInstall(t, func(h *rollbackHarness, s *moves.SealedObject) {
			key := moves.BuildStateKey(p1Host, "mv_other")
			body := p1StateBody("v4")
			ver, err := h.vs.PutIfAbsent(context.Background(), "data-uk", key, []byte(body))
			if err != nil {
				t.Fatal(err)
			}
			s.Key, s.VersionID, s.Size = key, ver, int64(len(body))
		})
		refused(t, h)
	})
	t.Run("a size the version is not", func(t *testing.T) {
		h := returnBeforeInstall(t, func(_ *rollbackHarness, s *moves.SealedObject) { s.Size++ })
		refused(t, h)
	})
}

// Task 8 re-review minor 2: once the landing reached probing a writer may
// have recorded over the placed state, so the target's location is the
// source of the returned state even if the landing's journal lost its
// install step (installManifest rebuilds it then; the build state is not
// rebuilt).
func TestAReturnPastProbingTakesTheTargetsStateWithoutTheInstallStep(t *testing.T) {
	t.Parallel()
	h := &rollbackHarness{landingHarness: newLandingHarness(t)}
	storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
	h.flip(t)
	h.land(t)
	h.installed(t)
	h.step(t, provision.LandRestored, provision.LandProbing)
	h.tpos = h.n
	storeState(h.phase1Harness, "data-uk", cutRoot+"/build-state.json", "v8")
	j := h.landJournal(t)
	j.Unmark(landStepInstall)
	if err := j.Commit(h.tdir); err != nil {
		t.Fatal(err)
	}
	h.uncover(t)
	h.rollBack(t)
	h.rollbackPass(t)
	rb := h.rb(t)
	if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("rollback %+v", rb)
	}
	if _, body := returnSealState(t, h, rb.ID); body != p1StateBody("v8") {
		t.Fatalf("the return lane holds %q, want the target's v8", body)
	}
}