//go:build linux

package host

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"net"
	"path/filepath"
	"strconv"
	"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/provision"
	"github.com/carlosframework/platform/internal/registry"
)

// rootPeer is the fixture's live requester: the agent's process tuple.
// ServeHTTP authenticates it with SO_PEERCRED and the controller registry in
// production, which this test host does not have.
var rootPeer = StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: "boot-1", ControllerID: "controller-1"}

// serve puts root's current handler on a Unix socket, as instancectl does,
// with peer authentication fixed to rootPeer. A restart needs a new socket.
func (w *assemblyWorld) serve() string {
	w.t.Helper()
	root := w.root
	root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return rootPeer, nil }
	return startRetirementRecoveryTransport(w.t, root)
}

// recoveryAgent is the agent's cross-box handover as claimconstruction.go
// builds it in agent mode: root's recovery rebind, inventory and fence view
// all through root's socket, the production identity, the box's registry.
func (w *assemblyWorld) recoveryAgent(socket string, sent *recoveryRebindRequest) claimCrossBoxHandover {
	w.t.Helper()
	_, inventory := hostbox.RemoteJournals(rootJournalClient(socket))
	guard := hostguard.Guard{DataDir: w.r.DataDir, LockDir: w.locks, Claims: hostguard.StoreReader{Store: w.store}, FenceView: rootFenceViewClient(rootJournalClient(socket))}
	client := recoveryRebindClient(socket)
	return claimCrossBoxHandover{RecoveryRebind: func(ctx context.Context, q recoveryRebindRequest) error {
		if sent != nil {
			*sent = q
		}
		return client(ctx, q)
	}, Guard: guard, Evidence: w.store, Box: "box-1", Inventory: inventory, Identity: w.owners.base.Identity, Registry: w.reg}
}

// recoveryAdmission moves host's claim into a recovery this box executes
// (a restoring op at probing whose source stamp is the installed one) by a
// TEST-ONLY claim transition, then admits a cold writer through the real
// hostclaim.AdmitWriter. It returns the claim before and after admission.
func (w *assemblyWorld) recoveryAdmission(host string) (prior, admitted hostclaim.Observed) {
	w.t.Helper()
	ctx := w.t.Context()
	o := w.claim(host)
	stamp := o.Claim.Holder.Stamp
	prior, err := hostclaim.Update(ctx, w.store, o, func(c *hostclaim.Claim) error {
		c.Op = plan10J2Op(hostclaim.KindRestoring, hostclaim.PhaseProbing)
		c.Op.Journal = map[string]string{hostclaim.JournalSourceStamp: strconv.FormatUint(stamp, 10)}
		return nil
	})
	if err != nil {
		w.t.Fatal(err)
	}
	h := prior.Claim.Holder
	admitted, err = hostclaim.Update(ctx, w.store, prior, hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: prior.Claim.Op.ID, ObservedPhase: prior.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "box-1", Mode: "cold", NewGeneration: "recovered", Region: "ie", PredecessorRegion: "ie"}))
	if err != nil {
		w.t.Fatal(err)
	}
	if admitted.Claim.Holder.Stamp <= stamp {
		w.t.Fatal("admission did not move the stamp")
	}
	return prior, admitted
}

// restampInventory rewrites ONLY the stamp of host's installed inventory in
// root's journal database, leaving its revision and every other field: the
// one fact the predecessor check reads.
func (w *assemblyWorld) restampInventory(host string, captured hostguard.Captured, stamp string) {
	w.t.Helper()
	db, err := sql.Open("sqlite", w.journal)
	if err != nil {
		w.t.Fatal(err)
	}
	defer db.Close()
	var body []byte
	if err = db.QueryRow("SELECT body FROM stop_inventory WHERE host=? AND incarnation=? AND stamp=?", host, captured.Incarnation, captured.Stamp).Scan(&body); err != nil {
		w.t.Fatal(err)
	}
	var in hostbox.StopInventory
	if err = json.Unmarshal(body, &in); err != nil {
		w.t.Fatal(err)
	}
	in.DirectoryRevision = stamp
	if body, err = json.Marshal(in); err != nil {
		w.t.Fatal(err)
	}
	if _, err = db.Exec("UPDATE stop_inventory SET body=? WHERE host=? AND incarnation=? AND stamp=?", body, host, captured.Incarnation, captured.Stamp); err != nil {
		w.t.Fatal(err)
	}
}

func (w *assemblyWorld) view(host string, c hostguard.Captured) hostguard.FenceView {
	w.t.Helper()
	v, err := w.assembly.Guard.Root.View(w.t.Context(), host, c)
	if err != nil {
		w.t.Fatal(err)
	}
	return v
}

// successorInventory is root's inventory of admitted's installation, or
// revision 0 when there is none.
func (w *assemblyWorld) successorInventory(admitted hostclaim.Observed) (hostbox.StopInventory, uint64) {
	w.t.Helper()
	h := admitted.Claim.Holder
	in, revision, err := w.assembly.Journals.Inventory.Snapshot(w.t.Context(), admitted.Host, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp})
	if err != nil {
		w.t.Fatal(err)
	}
	return in, revision
}

// N4 (roster binding task 6, R3): a recovery rebind through the real
// producer, installRecoveryRemote, and root's serveRecoveryRebind over the
// Unix socket, with real hold churn between the predecessor's install and
// the recovery and a hold held through it. The rebound inventory carries the
// binding. Then the same, interrupted after root's intent is durable and
// finished by a restarted root's RecoverPending (the startup replay of the
// pending rebind intent), then the agent's retry. Then only the predecessor
// inventory's stamp is changed: the rebind is refused at the predecessor
// check, after the successor check has passed.
func TestRemoteRecoveryRebindsSameBinding(t *testing.T) {
	ctx := t.Context()
	w := newAssemblyWorld(t)
	w.mustStart()
	b0 := w.fresh()
	agent := w.agent(b0)
	recA, routeA := w.newHost("a")
	recB, routeB := w.newHost("b")
	recC, routeC := w.newHost("c")
	for _, h := range []struct {
		rec   provision.Record
		route registry.Route
	}{{recA, routeA}, {recB, routeB}, {recC, routeC}} {
		if err := agent.install(h.rec, h.route); err != nil {
			t.Fatal(err)
		}
	}
	w.roster.holdChurn()
	w.roster.holdChurn()
	held := w.roster.reserveHold()
	if actor := w.fresh(); actor.Revision == b0.Revision || actor.Binding != b0.Binding {
		t.Fatalf("hold churn: revision %d -> %d, binding moved %v", b0.Revision, actor.Revision, actor.Binding != b0.Binding)
	}

	// Success through the producer and root's handler.
	socket := w.serve()
	prior, admitted := w.recoveryAdmission(recA.Host)
	var sent recoveryRebindRequest
	if err := w.recoveryAgent(socket, &sent).installRecoveryRemote(ctx, prior, admitted); err != nil {
		t.Fatal("recovery rebind after hold churn:", err)
	}
	if sent.Next.DirectoryRevision != b0.Binding {
		t.Fatalf("producer sent successor stamp %q, want the binding", sent.Next.DirectoryRevision)
	}
	rebound, revision := w.successorInventory(admitted)
	if revision != 1 || rebound.DirectoryRevision != b0.Binding || rebound.InstallationStamp != admitted.Claim.Holder.Stamp {
		t.Fatalf("rebound inventory: revision %d stamp %q installation %d", revision, rebound.DirectoryRevision, rebound.InstallationStamp)
	}
	sh := admitted.Claim.Holder
	if v := w.view(recA.Host, hostguard.Captured{Incarnation: sh.Incarnation, Stamp: sh.Stamp}); v.Installation.Stamp != sh.Stamp {
		t.Fatalf("root installation not rebound: %+v", v.Installation)
	}
	if row, err := w.reg.Get(recA.Host); err != nil || row == nil || row.ClaimStamp != sh.Stamp {
		t.Fatalf("rows not rebound: %+v %v", row, err)
	}
	w.roster.releaseHold(held)

	// Startup replay: root dies after its rebind intent is durable (the
	// intent write succeeds, the commit write fails), restarts, and its
	// RecoverPending finishes the intent under the unchanged binding.
	prior, admitted = w.recoveryAdmission(recB.Host)
	*w.interrupt = 1
	err := w.recoveryAgent(socket, nil).installRecoveryRemote(ctx, prior, admitted)
	*w.interrupt = -1
	if err == nil || w.pending() != 1 {
		t.Fatalf("rebind was not interrupted after its intent: %v, pending %d", err, w.pending())
	}
	w.roster.holdChurn()
	if err := w.restart(); err != nil {
		t.Fatal(err)
	}
	if err := w.root.RecoverPending(ctx); err != nil {
		t.Fatal("restart with a pending same-binding rebind:", err)
	}
	if w.pending() != 0 {
		t.Fatal("startup did not finish the pending rebind")
	}
	if in, revision := w.successorInventory(admitted); revision != 1 || in.DirectoryRevision != b0.Binding {
		t.Fatalf("replayed rebind: revision %d stamp %q", revision, in.DirectoryRevision)
	}
	socket = w.serve()
	if err := w.recoveryAgent(socket, nil).installRecoveryRemote(ctx, prior, admitted); err != nil {
		t.Fatal("agent retry after the startup replay:", err)
	}
	if row, err := w.reg.Get(recB.Host); err != nil || row == nil || row.ClaimStamp != admitted.Claim.Holder.Stamp {
		t.Fatalf("rows not rebound after the replay: %+v %v", row, err)
	}

	// Only the predecessor's stamp changes: refused at the predecessor check.
	ch := w.claim(recC.Host).Claim.Holder
	source := hostguard.Captured{Incarnation: ch.Incarnation, Stamp: ch.Stamp}
	w.restampInventory(recC.Host, source, strings.Repeat("d", 64))
	before := w.view(recC.Host, source)
	prior, admitted = w.recoveryAdmission(recC.Host)
	if err := w.recoveryAgent(socket, &sent).installRecoveryRemote(ctx, prior, admitted); err == nil {
		t.Fatal("a predecessor stamped under another binding was rebound")
	}
	if sent.Next.DirectoryRevision != w.fresh().Binding {
		t.Fatal("the successor stamp was not today's binding")
	}
	if _, revision := w.successorInventory(admitted); revision != 0 || w.pending() != 0 {
		t.Fatalf("refusal left successor state: revision %d pending %d", revision, w.pending())
	}
	if after := w.view(recC.Host, source); after.Installation != before.Installation {
		t.Fatalf("refusal changed root: %+v", after.Installation)
	}
	// The refusal is the predecessor check's (ErrTypedInventory), not the
	// successor check's (ErrChanged).
	if err := w.root.rebindRecoveryInventory(ctx, sent); !errors.Is(err, hostbox.ErrTypedInventory) {
		t.Fatalf("refused at %v, want the predecessor check", err)
	}
}

// L6 (roster binding R3): recovery never adopts a predecessor stamped under
// another binding. Rebind (remote) and handover recovery (local) refuse a
// predecessor installed at B0 once readmission makes the binding B1, and a
// numeral-stamped predecessor; the local path still adopts a predecessor at
// today's binding after hold churn. A numeral predecessor's receipt (no
// binding field) fails verifyRetirementReceipt, so the operation cold
// handover cannot park it. (The cold handover that parks a B0 predecessor
// with a matching receipt and starts a successor at B1 is N5.)
func TestClaimBindingRecoveryRefusesPredecessor(t *testing.T) {
	t.Run("rebind", func(t *testing.T) {
		ctx := t.Context()
		w := newAssemblyWorld(t)
		w.mustStart()
		b0 := w.fresh()
		agent := w.agent(b0)
		recA, routeA := w.newHost("a")
		recN, routeN := w.newHost("n")
		for _, h := range []struct {
			rec   provision.Record
			route registry.Route
		}{{recA, routeA}, {recN, routeN}} {
			if err := agent.install(h.rec, h.route); err != nil {
				t.Fatal(err)
			}
		}
		nh := w.claim(recN.Host).Claim.Holder
		w.restampInventory(recN.Host, hostguard.Captured{Incarnation: nh.Incarnation, Stamp: nh.Stamp}, strconv.FormatUint(b0.Revision, 10))
		w.roster.quarantine("loss_1")
		w.roster.readmit()
		b1 := w.fresh()
		socket := w.serve()
		for _, host := range []string{recA.Host, recN.Host} {
			var sent recoveryRebindRequest
			prior, admitted := w.recoveryAdmission(host)
			if err := w.recoveryAgent(socket, &sent).installRecoveryRemote(ctx, prior, admitted); err == nil {
				t.Fatalf("%s: a stale predecessor was rebound", host)
			}
			if sent.Next.DirectoryRevision != b1.Binding {
				t.Fatalf("%s: successor stamp %q, want B1", host, sent.Next.DirectoryRevision)
			}
			if err := w.root.rebindRecoveryInventory(ctx, sent); !errors.Is(err, hostbox.ErrTypedInventory) {
				t.Fatalf("%s: refused at %v, want the predecessor check", host, err)
			}
			if _, revision := w.successorInventory(admitted); revision != 0 {
				t.Fatalf("%s: refusal wrote a successor inventory", host)
			}
		}
	})

	t.Run("handover recovery", func(t *testing.T) {
		for _, stale := range []string{"none", "readmission", "numeral"} {
			t.Run(stale, func(t *testing.T) {
				ctx := t.Context()
				roster := newBindingRosterFixture(t, "box-1", "ie")
				identity := roster.identity()
				b0, err := identity.Verify(ctx)
				if err != nil {
					t.Fatal(err)
				}
				data := t.TempDir()
				inventory, err := openLegacyInventory(t, filepath.Join(t.TempDir(), "inventory.db"))
				if err != nil {
					t.Fatal(err)
				}
				t.Cleanup(func() { inventory.Close() })
				reg := restartTestRegistry(t)
				host := "shop.bab.oncarlos.com"
				prior := hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc-1", Stamp: 4}, Op: hostclaim.Op{Journal: map[string]string{hostclaim.JournalSourceStamp: "4"}}}}
				admitted := hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc-1", Stamp: 9}}}
				route := registry.Route{Host: host, Kind: "instance", DBPath: provision.NestedDBPath(data, host), ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: "inc-1", ClaimStamp: 4, ClaimOwner: "old-owner", ClaimRowID: "old-row"}}
				if err := reg.Upsert(route); err != nil {
					t.Fatal(err)
				}
				stamp := b0.Binding
				if stale == "numeral" {
					stamp = strconv.FormatUint(b0.Revision, 10)
				}
				if err := inventory.WriteStopInventory(ctx, hostbox.StopInventory{Complete: true, Host: host, Incarnation: "inc-1", InstallationStamp: 4, Box: "box-1", Enrollment: b0.Enrollment, DirectoryRevision: stamp, Entries: []hostbox.StopInventoryEntry{{Kind: "route", ArtifactID: "route", ManagedRoot: "registry", RelativePath: host, RowIdentity: "old-row", Ownership: "old-owner", Materialization: "materialized"}}}); err != nil {
					t.Fatal(err)
				}
				switch stale {
				case "none":
					roster.holdChurn()
				case "readmission":
					roster.quarantine("loss_1")
					roster.readmit()
				}
				h := claimCrossBoxHandover{Guard: hostguard.Guard{DataDir: data}, Box: "box-1", Inventory: inventory, Identity: identity, Registry: reg}
				err = h.installRecoveryInventory(ctx, prior, admitted)
				_, readErr := inventory.ReadInstallation(ctx, host, "inc-1", 9)
				if stale == "none" {
					if err != nil {
						t.Fatal("handover recovery at today's binding after hold churn:", err)
					}
					successor, e := inventory.ReadInstallation(ctx, host, "inc-1", 9)
					if e != nil || successor.DirectoryRevision != b0.Binding {
						t.Fatalf("successor inventory %q %v", successor.DirectoryRevision, e)
					}
					return
				}
				if !errors.Is(err, hostbox.ErrTypedInventory) || !errors.Is(readErr, sql.ErrNoRows) {
					t.Fatalf("stale predecessor: %v, successor read %v", err, readErr)
				}
				if row, e := reg.Get(host); e != nil || row.ClaimOwnership != route.ClaimOwnership {
					t.Fatalf("refusal rebound rows: %+v %v", row, e)
				}
			})
		}
	})

	t.Run("numeral receipt", func(t *testing.T) {
		target := hostguard.FenceTarget{FenceID: "fence-1", Region: "ie", Host: "shop.bab.oncarlos.com", Incarnation: "inc-1", Stamp: 4, Session: 5, Op: "purge-1", Phase: "stopping", Box: "box-1", Enrollment: strings.Repeat("a", 32), BootID: "boot-1", ControllerID: "controller-1", ClaimETag: "etag", DirectoryRevision: "7", RosterRevision: 7}
		pre := StopReceipt{Kind: "stopped", ID: "receipt-1", FenceID: target.FenceID, Region: target.Region, Host: target.Host, Incarnation: target.Incarnation, InstallationStamp: target.Stamp, WriterSession: target.Session, OperationID: target.Op, Phase: target.Phase, Box: target.Box, Enrollment: target.Enrollment, BootID: target.BootID, ControllerID: target.ControllerID, ClaimETag: target.ClaimETag, DirectoryRevision: "7"}
		if err := verifyRetirementReceipt(t.Context(), hostguard.FenceReceipt{Target: target, Bytes: mustMarshalStopReceipt(pre), EvidenceAt: pre.EvidenceAt}); !errors.Is(err, ErrClaimStopReceipt) {
			t.Fatalf("a receipt without a binding proved a numeral predecessor: %v", err)
		}
		// Control: the binding field is the only difference.
		carried := pre
		carried.Binding = target.DirectoryRevision
		if err := verifyRetirementReceipt(t.Context(), hostguard.FenceReceipt{Target: target, Bytes: mustMarshalStopReceipt(carried), EvidenceAt: carried.EvidenceAt}); err != nil {
			t.Fatalf("control receipt refused: %v", err)
		}
	})
}