//go:build linux

package host

import (
	"bytes"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/registry"
)

// operationStop moves host's claim into an operation session this box
// executes (a moving op, with a writer session numbered as an admission at
// this write would number it) by a TEST-ONLY claim transition, then stops
// it through root's real stop coordinator: the prior's completed fence
// receipt, stamped with stamp.
func (w *assemblyWorld) operationStop(host, stamp string) hostclaim.Observed {
	w.t.Helper()
	o := w.operationSession(host)
	if _, err := w.root.Stop.Coordinator.FenceAndStop(w.t.Context(), w.priorStop(o, stamp)); err != nil {
		w.t.Fatal("operation stop of the prior:", err)
	}
	return o
}

// operationSession is operationStop's TEST-ONLY claim transition alone.
func (w *assemblyWorld) operationSession(host string) hostclaim.Observed {
	w.t.Helper()
	o, err := hostclaim.Update(w.t.Context(), w.store, w.claim(host), func(c *hostclaim.Claim) error {
		h := c.Holder
		h.Session, h.WriterBox, h.Generation = c.Gen+1, "box-1", "g5"
		h.Admissions = []hostclaim.AdmissionRecord{{Region: "ie", Session: h.Session, Box: "box-1", Generation: "g5", Mode: "fresh"}}
		c.Op = hostclaim.Op{Kind: hostclaim.KindMoving, ID: "purge-1", Phase: "stopping", Exec: "box-1"}
		return nil
	})
	if err != nil {
		w.t.Fatal(err)
	}
	return o
}

// priorStop is the agent's stop request for o's writer session.
func (w *assemblyWorld) priorStop(o hostclaim.Observed, stamp string) StopRequest {
	h := o.Claim.Holder
	return StopRequest{Region: "ie", Host: o.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session, Box: "box-1", Enrollment: w.roster.row().Enrollment,
		BootID: rootPeer.BootID, ControllerID: rootPeer.ControllerID, ExpectedClaimETag: o.ETag, ExpectedDirectoryRevision: stamp, RosterRevision: w.roster.read().Roster.Revision,
		OperationID: "purge-1", Phase: "stopping", ObligationID: "pre-teardown-stop"}
}

// coldAdmit is the executor's real cold writer admission on host's claim.
func (w *assemblyWorld) coldAdmit(host, generation string) hostclaim.Observed {
	w.t.Helper()
	o := w.claim(host)
	h := o.Claim.Holder
	next, err := hostclaim.Update(w.t.Context(), w.store, o, hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "box-1", Mode: "cold", NewGeneration: generation, Region: "ie", PredecessorRegion: "ie"}))
	if err != nil {
		w.t.Fatal(err)
	}
	return next
}

// agentGuard is the agent's guard: root's fence view over the socket.
func (w *assemblyWorld) agentGuard(socket string) hostguard.Guard {
	return hostguard.Guard{DataDir: w.r.DataDir, LockDir: w.locks, Claims: hostguard.StoreReader{Store: w.store}, FenceView: rootFenceViewClient(rootJournalClient(socket))}
}

// predecessorReceipt is the prior's completed receipt as the agent's
// production cold admission hook finds it.
func (w *assemblyWorld) predecessorReceipt(socket string, prior hostclaim.Observed) hostguard.FenceReceipt {
	w.t.Helper()
	h := prior.Claim.Holder
	receipt, err := w.agentGuard(socket).CompletedColdPredecessor(w.t.Context(), prior.Host, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}, h.Session, prior.Claim.Op.ID, prior.Claim.Op.Phase)
	if err != nil {
		w.t.Fatal("prior receipt:", err)
	}
	return receipt
}

func coldRequestFor(prior hostguard.FenceTarget, admitted hostclaim.Observed, enrollment, stamp string) coldHandoverRequest {
	h := admitted.Claim.Holder
	return coldHandoverRequest{Prior: prior, Next: hostbox.ColdSuccessor{Host: admitted.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Generation: h.Generation, Box: "box-1", Enrollment: enrollment, DirectoryRevision: stamp, ETag: admitted.ETag, Op: admitted.Claim.Op.ID, Phase: admitted.Claim.Op.Phase}}
}

// coldJournals is root's cold handover journals by file name and phase.
func (w *assemblyWorld) coldJournals() map[string]string {
	w.t.Helper()
	out := map[string]string{}
	entries, err := os.ReadDir(filepath.Join(w.retire, "cold"))
	if errors.Is(err, os.ErrNotExist) {
		return out
	}
	if err != nil {
		w.t.Fatal(err)
	}
	for _, e := range entries {
		if filepath.Ext(e.Name()) != ".json" {
			continue
		}
		raw, err := os.ReadFile(filepath.Join(w.retire, "cold", e.Name()))
		if err != nil {
			w.t.Fatal(err)
		}
		var j struct{ Phase string }
		if err := json.Unmarshal(raw, &j); err != nil {
			w.t.Fatal(err)
		}
		out[e.Name()] = j.Phase
	}
	return out
}

// priorIntact asserts nothing of the prior moved: its route row is still
// owned by its installation and root's active stamp is still its own.
func (w *assemblyWorld) priorIntact(what string, prior hostguard.FenceTarget, route registry.Route) {
	w.t.Helper()
	row, err := w.reg.Get(prior.Host)
	if err != nil || row == nil || row.ClaimStamp != prior.Stamp || row.ClaimOwnership != route.ClaimOwnership {
		w.t.Fatalf("%s: the prior's row moved: %+v %v", what, row, err)
	}
	if v := w.view(prior.Host, hostguard.Captured{Incarnation: prior.Incarnation, Stamp: prior.Stamp}); v.Installation.Stamp != prior.Stamp {
		w.t.Fatalf("%s: root's active stamp moved: %+v", what, v.Installation)
	}
}

// installedRoute is host's registry row as the agent's install left it.
func (w *assemblyWorld) installedRoute(host string) registry.Route {
	w.t.Helper()
	row, err := w.reg.Get(host)
	if err != nil || row == nil {
		w.t.Fatalf("no route for %s: %v", host, err)
	}
	return *row
}

// N5 (roster binding task 6b): root's operation cold handover admits only a
// successor stamped with today's binding, checked before any successor
// state. Through root's handler over the socket: a B0 prior (installed and
// stopped before a quarantine) hands over to a B1 successor after
// readmission (parked and started); while quarantined the handover is
// refused; a successor at the old binding is refused with no journal file,
// no successor inventory and the prior not parked.
func TestColdHandoverVerifiesSuccessorAtRoot(t *testing.T) {
	ctx := t.Context()
	w := newAssemblyWorld(t)
	w.mustStart()
	b0 := w.fresh()
	rec, route := w.newHost("a")
	if err := w.agent(b0).install(rec, route); err != nil {
		t.Fatal(err)
	}
	route = w.installedRoute(rec.Host)
	w.roster.holdChurn()
	prior := w.operationStop(rec.Host, b0.Binding)
	socket := w.serve()
	receipt := w.predecessorReceipt(socket, prior)
	if receipt.Target.DirectoryRevision != b0.Binding {
		t.Fatalf("prior receipt target stamp %q", receipt.Target.DirectoryRevision)
	}
	admitted := w.coldAdmit(rec.Host, "g6")
	enrollment := w.roster.row().Enrollment

	// Quarantined: refused before any successor state. Through the socket
	// ServeHTTP's RequireNewWork answers 403; the handler's own admission
	// answers 409.
	w.roster.quarantine("loss_1")
	quarantined := coldRequestFor(receipt.Target, admitted, enrollment, b0.Binding)
	if status := postRecoveryHandover(t, socket, quarantined); status != http.StatusForbidden {
		t.Fatalf("quarantined handover through the socket: HTTP %d", status)
	}
	body, _ := json.Marshal(quarantined)
	direct := httptest.NewRecorder()
	w.root.serveColdHandover(direct, httptest.NewRequest(http.MethodPost, "http://instance-control/claim/handover", bytes.NewReader(body)).WithContext(ctx), rootPeer)
	if direct.Code != http.StatusConflict {
		t.Fatalf("quarantined handover at the handler: HTTP %d", direct.Code)
	}

	// Readmitted: a successor at the old binding is refused before any state.
	w.roster.readmit()
	b1 := w.fresh()
	if b1.Binding == b0.Binding {
		t.Fatal("readmission did not move the binding")
	}
	for _, stamp := range []string{b0.Binding, strings.Repeat("e", 64)} {
		if status := postRecoveryHandover(t, socket, coldRequestFor(receipt.Target, admitted, enrollment, stamp)); status != http.StatusConflict {
			t.Fatalf("successor stamped %q: HTTP %d", stamp, status)
		}
	}
	if j := w.coldJournals(); len(j) != 0 {
		t.Fatalf("refused handover wrote a journal: %v", j)
	}
	if _, revision := w.successorInventory(admitted); revision != 0 {
		t.Fatal("refused handover wrote a successor inventory")
	}
	w.priorIntact("refused successor", receipt.Target, route)

	// B0 prior with its receipt, B1 successor through the agent's production
	// cold admission hook: parked and started.
	if err := coldAdmissionHook(w.agentGuard(socket), w.owners.base.Identity, socket)(ctx, prior, admitted); err != nil {
		t.Fatal("B0 prior to B1 successor:", err)
	}
	successor, revision := w.successorInventory(admitted)
	if revision == 0 || successor.DirectoryRevision != b1.Binding {
		t.Fatalf("successor inventory: revision %d stamp %q", revision, successor.DirectoryRevision)
	}
	if row, err := w.reg.Get(rec.Host); err != nil || row == nil || row.ClaimStamp != admitted.Claim.Holder.Stamp {
		t.Fatalf("successor not started: %+v %v", row, err)
	}
	if v := w.view(rec.Host, hostguard.Captured{Incarnation: successor.Incarnation, Stamp: successor.InstallationStamp}); v.Installation.Stamp != admitted.Claim.Holder.Stamp || v.Barrier != nil {
		t.Fatalf("root after the handover: %+v barrier %v", v.Installation, v.Barrier)
	}
	for name, phase := range w.coldJournals() {
		if phase != "complete" {
			t.Fatalf("journal %s: %s", name, phase)
		}
	}
}

// N5 extended (B1) and B3: a cold handover interrupted after
// BeginColdHandover (its barrier durable, nothing of the prior moved); the
// box is quarantined and readmitted. The agent's replay of the journaled
// handover is refused (not resumed). Root restarts: startup comes up, its
// RecoverPending does not resume the handover and settles it instead (the
// barrier cleared, the successor stamp retired, the journal marked settled,
// logged). The prior keeps its historical authority: its cleanup stop
// returns its receipt, and a clean successor admitted under B1 takes over
// from it through root's handler, parking the prior. Another restart leaves
// the settled journal alone.
func TestColdHandoverObsoleteBarrierSettledAtStartup(t *testing.T) {
	ctx := t.Context()
	w := newAssemblyWorld(t)
	w.mustStart()
	b0 := w.fresh()
	rec, route := w.newHost("a")
	if err := w.agent(b0).install(rec, route); err != nil {
		t.Fatal(err)
	}
	route = w.installedRoute(rec.Host)
	prior := w.operationStop(rec.Host, b0.Binding)
	socket := w.serve()
	receipt := w.predecessorReceipt(socket, prior)
	admitted := w.coldAdmit(rec.Host, "g6")
	enrollment := w.roster.row().Enrollment
	request := coldRequestFor(receipt.Target, admitted, enrollment, b0.Binding)
	successorStamp := hostguard.Captured{Incarnation: admitted.Claim.Holder.Incarnation, Stamp: admitted.Claim.Holder.Stamp}
	captured := hostguard.Captured{Incarnation: receipt.Target.Incarnation, Stamp: receipt.Target.Stamp}

	// Root dies right after BeginColdHandover's write: the journal's prepare
	// write and the barrier write succeed, the prior's retirement write fails.
	*w.interrupt = 2
	status := postRecoveryHandover(t, socket, request)
	*w.interrupt = -1
	if status != http.StatusConflict {
		t.Fatalf("interrupted handover: HTTP %d", status)
	}
	if v := w.view(rec.Host, captured); v.Barrier == nil || v.Barrier.Stamp != successorStamp.Stamp {
		t.Fatalf("no durable barrier after the interruption: %+v", v.Barrier)
	}
	journals := w.coldJournals()
	if len(journals) != 1 {
		t.Fatalf("journals %v", journals)
	}
	for _, phase := range journals {
		if phase != "prepared" {
			t.Fatalf("interrupted journal phase %q", phase)
		}
	}
	w.priorIntact("interrupted", receipt.Target, route)

	// Binding change, then the agent's replay: refused, not resumed.
	w.roster.quarantine("loss_1")
	if status := postRecoveryHandover(t, socket, request); status == http.StatusNoContent {
		t.Fatal("quarantined replay resumed")
	}
	w.roster.readmit()
	b1 := w.fresh()
	if status := postRecoveryHandover(t, socket, request); status != http.StatusConflict {
		t.Fatalf("replay with an obsolete successor: HTTP %d", status)
	}
	if v := w.view(rec.Host, captured); v.Barrier == nil {
		t.Fatal("replay refusal cleared the barrier")
	}
	w.priorIntact("refused replay", receipt.Target, route)

	// Startup: up, not resumed, settled.
	logs := captureLog(t)
	if err := w.restart(); err != nil {
		t.Fatal("startup with an obsolete handover barrier:", err)
	}
	if err := w.root.RecoverPending(ctx); err != nil {
		t.Fatal("startup recovery with an obsolete handover barrier:", err)
	}
	if !strings.Contains(logs(), "obsolete cold handover: stamped "+b0.Binding+" current "+b1.Binding) {
		t.Fatalf("no settlement log: %s", logs())
	}
	v := w.view(rec.Host, captured)
	if v.Barrier != nil || v.Pending == nil || *v.Pending != receipt.Target || v.Installation.Stamp != receipt.Target.Stamp {
		t.Fatalf("after settlement: barrier %v pending %v installation %+v", v.Barrier, v.Pending, v.Installation)
	}
	retired := false
	for _, old := range v.Retired {
		retired = retired || old == successorStamp
	}
	if !retired {
		t.Fatalf("obsolete successor stamp not retired: %v", v.Retired)
	}
	for name, phase := range w.coldJournals() {
		if phase != "settled" {
			t.Fatalf("journal %s after settlement: %s", name, phase)
		}
	}
	if _, revision := w.successorInventory(admitted); revision != 0 {
		t.Fatal("the obsolete successor got an inventory")
	}
	w.priorIntact("settled", receipt.Target, route)

	// The prior's historical authority: its cleanup stop completes with its
	// original receipt.
	socket = w.serve()
	cleanup := w.priorStop(prior, b0.Binding)
	cleanup.Cleanup = true
	stopped, err := w.root.Stop.Coordinator.FenceAndStop(ctx, cleanup)
	if err != nil || !bytes.Equal(stopped.Bytes, receipt.Bytes) {
		t.Fatalf("prior's cleanup stop after settlement: %v (same receipt %v)", err, bytes.Equal(stopped.Bytes, receipt.Bytes))
	}
	// The obsolete successor is never reused, at any binding.
	if status := postRecoveryHandover(t, socket, coldRequestFor(receipt.Target, admitted, enrollment, b1.Binding)); status != http.StatusConflict {
		t.Fatalf("settled successor reused: HTTP %d", status)
	}

	// Clean reinstall at B1: a new successor admitted after the settlement,
	// handed the prior through root's handler.
	clean := w.coldAdmit(rec.Host, "g7")
	if status := postRecoveryHandover(t, socket, coldRequestFor(receipt.Target, clean, enrollment, b1.Binding)); status != http.StatusNoContent {
		t.Fatalf("clean reinstall at B1: HTTP %d", status)
	}
	successor, revision := w.successorInventory(clean)
	if revision == 0 || successor.DirectoryRevision != b1.Binding {
		t.Fatalf("clean successor inventory: revision %d stamp %q", revision, successor.DirectoryRevision)
	}
	if row, err := w.reg.Get(rec.Host); err != nil || row == nil || row.ClaimStamp != clean.Claim.Holder.Stamp {
		t.Fatalf("clean successor not started: %+v %v", row, err)
	}

	// Another restart: the settled journal is left alone.
	if err := w.restart(); err != nil {
		t.Fatal(err)
	}
	if err := w.root.RecoverPending(ctx); err != nil {
		t.Fatal("restart after the clean reinstall:", err)
	}
	settled := 0
	for _, phase := range w.coldJournals() {
		if phase == "settled" {
			settled++
		}
	}
	if settled != 1 {
		t.Fatalf("journals after the reinstall: %v", w.coldJournals())
	}
}

// B1, independently (fix round 1, finding 4): a journaled cold handover
// whose successor matches root's startup binding is not obsolete, so only
// startup's fresh admission stands between it and a resume. Root restarts
// with the roster unchanged (startup binding B0), then the fresh authority
// changes before RecoverPending: a quarantine, then a readmission (B1).
// Each time startup refuses the resume, and nothing of the successor
// exists: the barrier stays, the journal stays prepared, the successor has
// no inventory, the prior's row and active stamp are its own. Removing only
// startup's admission (the Admit of RecoverPending's cold handover) resumes
// the handover here.
func TestColdHandoverStartupAdmitsAgainstFreshAuthority(t *testing.T) {
	ctx := t.Context()
	w := newAssemblyWorld(t)
	w.mustStart()
	b0 := w.fresh()
	rec, route := w.newHost("a")
	if err := w.agent(b0).install(rec, route); err != nil {
		t.Fatal(err)
	}
	route = w.installedRoute(rec.Host)
	prior := w.operationStop(rec.Host, b0.Binding)
	socket := w.serve()
	receipt := w.predecessorReceipt(socket, prior)
	admitted := w.coldAdmit(rec.Host, "g6")
	request := coldRequestFor(receipt.Target, admitted, w.roster.row().Enrollment, b0.Binding)
	successorStamp := hostguard.Captured{Incarnation: admitted.Claim.Holder.Incarnation, Stamp: admitted.Claim.Holder.Stamp}
	captured := hostguard.Captured{Incarnation: receipt.Target.Incarnation, Stamp: receipt.Target.Stamp}

	// Root dies right after BeginColdHandover's write, under B0.
	*w.interrupt = 2
	status := postRecoveryHandover(t, socket, request)
	*w.interrupt = -1
	if status != http.StatusConflict {
		t.Fatalf("interrupted handover: HTTP %d", status)
	}
	// Restart with the roster unchanged: the journal matches the startup
	// binding, so it is not obsolete.
	if err := w.restart(); err != nil {
		t.Fatal(err)
	}
	if w.root.startupBinding != b0.Binding {
		t.Fatalf("startup binding %q, want B0 %q", w.root.startupBinding, b0.Binding)
	}
	untouched := func(what string) {
		t.Helper()
		if v := w.view(rec.Host, captured); v.Barrier == nil || v.Barrier.Stamp != successorStamp.Stamp {
			t.Fatalf("%s: barrier %+v", what, v.Barrier)
		}
		journals := w.coldJournals()
		if len(journals) != 1 {
			t.Fatalf("%s: journals %v", what, journals)
		}
		for _, phase := range journals {
			if phase != "prepared" {
				t.Fatalf("%s: journal phase %q", what, phase)
			}
		}
		if _, revision := w.successorInventory(admitted); revision != 0 {
			t.Fatalf("%s: the successor got an inventory", what)
		}
		w.priorIntact(what, receipt.Target, route)
	}
	for _, change := range []struct {
		what  string
		apply func()
	}{{"quarantined", func() { w.roster.quarantine("loss_1") }}, {"readmitted", func() { w.roster.readmit() }}} {
		change.apply()
		logs := captureLog(t)
		err := w.root.RecoverPending(ctx)
		if strings.Contains(logs(), "obsolete cold handover") {
			t.Fatalf("%s: settled as obsolete, so admission was not what refused: %s", change.what, logs())
		}
		untouched(change.what)
		if err == nil {
			t.Fatalf("%s: startup recovery reported no refusal", change.what)
		}
	}
	if b1 := w.fresh(); b1.Binding == b0.Binding {
		t.Fatal("readmission did not move the binding")
	}
}