//go:build unix

package hostbox

import (
	"context"
	"encoding/json"
	"errors"
	"slices"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
)

// completeFromInstalling is the target finishing after the source's
// fence closed: its probe passes on "dst", root stops the probe session,
// and the real serving CAS (hostclaim.CompleteRelocation) queues the
// source's park-local. No reconcile runs on the source meanwhile.
func (s *sourceReconcile) completeFromInstalling(t *testing.T) hostclaim.Observed {
	t.Helper()
	s.update(t, func(c *hostclaim.Claim) error {
		h := c.Holder
		h.Session, h.Generation, h.WriterBox = h.Session+1, "P1", "dst"
		h.Admissions = append(h.Admissions, hostclaim.AdmissionRecord{Session: h.Session, Box: "dst", Generation: "P1", Predecessor: "g1", Mode: "cold", Region: "ie"})
		probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: h.Session, At: time.Unix(90, 0).UTC()})
		stop, _ := json.Marshal(hostclaim.ProbeStop{Box: "dst", Session: h.Session, Receipt: "probe-fence-1", ReceiptSHA256: strings.Repeat("a", 64)})
		c.Op.Phase = hostclaim.PhaseRelocatingProbing
		c.Op.Journal["probe"], c.Op.Journal["probe_stop"] = string(probe), string(stop)
		return nil
	})
	n := 0
	o := s.update(t, hostclaim.CompleteRelocation("rel-1", func() string { n++; return "id-" + strconv.Itoa(n) }))
	if o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Op.Exec != "dst" || !hostclaim.OwesLocal(o.Claim, "box") {
		t.Fatalf("after the completion: op %+v obligations %+v", o.Claim.Op, o.Claim.Obligations)
	}
	return o
}

// L18 joint round, finding 3, through the production ReconcileAdapter,
// the real census (CheckHeld), park.Local and provision's pass: the
// source's fence closes in releasing, its reconcile is delayed past
// installing and the target's completion, and only then does it run. The
// claim now serves on "dst", so Task 10b's relocating scope is gone; the
// catch-up hold keeps the row held for the completed op, never ready,
// with its ownership and every other column as minted, the guard record
// and the claim unchanged. A replay changes nothing.
func TestReconcileCatchUpHoldsTheSourceAfterTheTargetCompleted(t *testing.T) {
	s := newSourceReconcile(t)
	minted := *s.row(t)
	s.releasingThenInstalling(t)
	o := s.completeFromInstalling(t)
	guard := targetGuardFiles(t, s.f.data)
	for i := 0; i < 2; i++ {
		s.pass(t)
		row := s.row(t)
		if row == nil || row.Status != "relocating" || row.HoldOp != "rel-1" {
			t.Fatalf("pass %d after the completion: the source's row is %+v (minted %+v)", i, row, minted)
		}
		held := *row
		held.Status, held.HoldOp = minted.Status, minted.HoldOp
		if held != minted {
			t.Fatalf("pass %d: the hold changed more than status and hold_op: %+v", i, row)
		}
		if got := targetGuardFiles(t, s.f.data); got[s.f.host+".json"] != guard[s.f.host+".json"] {
			t.Fatalf("pass %d: the guard record changed", i)
		}
		if after, err := hostclaim.Read(t.Context(), s.f.store, s.f.host); err != nil || after.ETag != o.ETag {
			t.Fatalf("pass %d: the claim was written: %v", i, err)
		}
	}
}

// The catch-up hold through the pass: a row another op holds is never
// taken, and with region HA off (neither hook set) the pass is main's,
// the row ready and exactly as minted.
func TestReconcileCatchUpHoldGuardsThroughThePass(t *testing.T) {
	t.Run("another op's row", func(t *testing.T) {
		s := newSourceReconcile(t)
		if ok, err := s.reg.SetHoldIf(s.f.host, "moving", "mv-9", ""); err != nil || !ok {
			t.Fatalf("holding for another op: %v %v", ok, err)
		}
		before := *s.row(t)
		s.releasingThenInstalling(t)
		s.completeFromInstalling(t)
		s.pass(t)
		if got := s.row(t); *got != before {
			t.Fatalf("another op's row was taken: %+v (was %+v)", got, before)
		}
	})
	t.Run("region HA off", func(t *testing.T) {
		s := newSourceReconcile(t)
		s.adapter.Reconciler.FenceClosed, s.adapter.Reconciler.HoldRow = nil, nil
		s.adapter.Configure(&s.cfg)
		minted := *s.row(t)
		s.releasingThenInstalling(t)
		s.completeFromInstalling(t)
		s.pass(t)
		if got := s.row(t); *got != minted {
			t.Fatalf("region HA off: the row is %+v (was %+v)", got, minted)
		}
	})
}

// completedServing rewrites the fixture's claim as a completed relocation's
// serving op: exec on exec, the completion recorded on stopBox, and the
// park-local obligations given.
func completedServing(t *testing.T, r Reconciler, host, exec, stopBox string, owed ...hostclaim.Obligation) hostclaim.Observed {
	t.Helper()
	ctx := context.Background()
	obs, err := hostclaim.Read(ctx, r.Claims, host)
	if err != nil {
		t.Fatal(err)
	}
	done, err := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-1", Stop: hostclaim.ProbeStop{Box: stopBox, Session: 7, Receipt: "probe-fence-1", ReceiptSHA256: strings.Repeat("a", 64)}, Phase: hostclaim.PhaseRelocatingProbing})
	if err != nil {
		t.Fatal(err)
	}
	obs, err = hostclaim.Update(ctx, r.Claims, obs, func(c *hostclaim.Claim) error {
		c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: exec, Participants: []string{exec}, Journal: map[string]string{hostclaim.JournalCompleted: string(done)}}
		c.Obligations = append(c.Obligations, owed...)
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	return obs
}

func parkLocal(box, op string) hostclaim.Obligation {
	return hostclaim.Obligation{ID: "ob-" + box + "-" + op, Op: op, Kind: "park-local", Box: box, Retirement: "rt-" + box + "-" + op, Owner: "box:" + box}
}

// The catch-up hold's scope, guard by guard, on the stubbed reconciler
// (relocationReconciler, its HoldRow recording): held for the completed
// op on positive evidence only, hold only, and nowhere else.
func TestReconcileCatchUpHoldScope(t *testing.T) {
	ctx := context.Background()
	install := func(r Reconciler, selected RecordSelection, writes *int) error {
		return r.Install(ctx, selected, func() error { *writes++; return nil })
	}
	t.Run("the source owing the completed op's park-local is held, hold only", func(t *testing.T) {
		r, selected, holds, _ := relocationReconciler(t)
		obs := completedServing(t, r, selected.Host, "dst", "dst", parkLocal("box", "rel-1"))
		writes := 0
		if err := install(r, selected, &writes); err != nil || writes != 0 || !slices.Equal(*holds, []string{"rel-1"}) {
			t.Fatalf("the catch-up hold: %v writes %d holds %v", err, writes, *holds)
		}
		// No installation authority: the activation reader's and the
		// inventory callback's check, which carry no Install decision, and
		// an activation decision, are all refused.
		if err := r.check(ctx, obs, selected); !errors.Is(err, ErrChanged) {
			t.Fatalf("the activation path's check under the catch-up: %v", err)
		}
		d, ok, err := r.completedHold(obs.Claim)
		if err != nil || !ok {
			t.Fatalf("completedHold: %v %v", ok, err)
		}
		if err := r.checkOn(ctx, obs, selected, &decided{row: d}); !errors.Is(err, ErrChanged) {
			t.Fatalf("an activation decision under the catch-up: %v", err)
		}
		if err := r.checkOn(ctx, obs, selected, &decided{row: RowDecision{Case: RowHeld, HoldOp: "rel-1"}, holdOnly: true}); !errors.Is(err, ErrChanged) {
			t.Fatalf("a hold decision the catch-up did not make: %v", err)
		}
	})
	for name, tc := range map[string]struct {
		exec, stop string
		owed       []hostclaim.Obligation
		lost       bool
		wantWrites int // main's serving executor writes through activation
	}{
		"this box serves":                    {exec: "box", stop: "dst", owed: []hostclaim.Obligation{parkLocal("box", "rel-1")}, wantWrites: 1},
		"completed on this box":              {exec: "dst", stop: "box", owed: []hostclaim.Obligation{parkLocal("box", "rel-1")}},
		"no park-local for this box":         {exec: "dst", stop: "dst", owed: []hostclaim.Obligation{parkLocal("other", "rel-1")}},
		"a park-local for another op":        {exec: "dst", stop: "dst", owed: []hostclaim.Obligation{parkLocal("box", "rel-0")}},
		"this box on lost":                   {exec: "dst", stop: "dst", owed: []hostclaim.Obligation{parkLocal("box", "rel-1")}, lost: true},
		"another kind of obligation for box": {exec: "dst", stop: "dst", owed: []hostclaim.Obligation{{ID: "ob-1", Op: "rel-1", Kind: "relocation-release", Box: "box", Region: "ie", Retirement: "rt-1", Owner: "region:ie"}}},
	} {
		t.Run(name, func(t *testing.T) {
			r, selected, holds, _ := relocationReconciler(t)
			completedServing(t, r, selected.Host, tc.exec, tc.stop, tc.owed...)
			if tc.lost {
				obs, err := hostclaim.Read(ctx, r.Claims, selected.Host)
				if err != nil {
					t.Fatal(err)
				}
				if _, err := hostclaim.Update(ctx, r.Claims, obs, func(c *hostclaim.Claim) error { c.Lost = append(c.Lost, "box"); return nil }); err != nil {
					t.Fatal(err)
				}
			}
			writes := 0
			err := install(r, selected, &writes)
			if len(*holds) != 0 || writes != tc.wantWrites {
				t.Fatalf("%v: holds %v writes %d (want %d)", err, *holds, writes, tc.wantWrites)
			}
			if tc.wantWrites == 0 && !errors.Is(err, ErrChanged) {
				t.Fatalf("main's Serving refusal: %v", err)
			}
		})
	}
	t.Run("region HA off", func(t *testing.T) {
		r, selected, holds, _ := relocationReconciler(t)
		r.FenceClosed, r.HoldRow = nil, nil
		completedServing(t, r, selected.Host, "dst", "dst", parkLocal("box", "rel-1"))
		writes := 0
		if err := install(r, selected, &writes); !errors.Is(err, ErrChanged) || writes != 0 || len(*holds) != 0 {
			t.Fatalf("region HA off: %v writes %d holds %v", err, writes, *holds)
		}
	})
	t.Run("the decision is remade under the lock", func(t *testing.T) {
		r, selected, _, _ := relocationReconciler(t)
		completedServing(t, r, selected.Host, "dst", "dst", parkLocal("box", "rel-1"))
		obs, err := hostclaim.Read(ctx, r.Claims, selected.Host)
		if err != nil {
			t.Fatal(err)
		}
		d, ok, err := r.completedHold(obs.Claim)
		if err != nil || !ok {
			t.Fatalf("completedHold: %v %v", ok, err)
		}
		// The park-local is discharged between the decision and the lock.
		if _, err := hostclaim.Update(ctx, r.Claims, obs, func(c *hostclaim.Claim) error { c.Obligations = nil; return nil }); err != nil {
			t.Fatal(err)
		}
		if err := r.holdOnly(ctx, selected, d); !errors.Is(err, ErrChanged) {
			t.Fatalf("a catch-up decision the fresh read no longer makes: %v", err)
		}
	})
}

// RegistryHoldRow's guards hold for the catch-up as they do for Task 10b:
// a held row for the completed op is a replay with nothing written.
func TestReconcileCatchUpHoldReplayWritesNothing(t *testing.T) {
	r, selected, reg, _ := sourceRows(t)
	completedServing(t, r, selected.Host, "dst", "dst", parkLocal("box", "rel-1"))
	if err := r.Install(context.Background(), selected, func() error { return errors.New("activated") }); err != nil {
		t.Fatalf("the catch-up hold: %v", err)
	}
	held := *rowOf(t, reg, selected.Host)
	if held.Status != "relocating" || held.HoldOp != "rel-1" {
		t.Fatalf("held row %+v", held)
	}
	if err := r.Install(context.Background(), selected, func() error { return errors.New("activated") }); err != nil {
		t.Fatalf("the replay: %v", err)
	}
	if got := rowOf(t, reg, selected.Host); *got != held {
		t.Fatalf("the replay changed the row: %+v (was %+v)", got, held)
	}
}

// Review M1: the catch-up hold is a hold only on this box's own existing
// installation (CheckHeld, as for Task 10b). Through the production
// adapter and the pass, a census that refuses (the source's row owned by
// another incarnation, or an inventory of another enrollment) holds
// nothing: the row stays exactly as it was. On the stubbed reconciler a
// refusing CheckHeld is Install's error and HoldRow is never called.
func TestReconcileCatchUpHoldNeedsCheckHeld(t *testing.T) {
	t.Run("a row of another incarnation", func(t *testing.T) {
		s := newSourceReconcile(t)
		s.releasingThenInstalling(t)
		s.completeFromInstalling(t)
		stranger := *s.row(t)
		stranger.ClaimIncarnation = "inc-other"
		// Ownership is never rewritten in place: the row is replaced.
		if err := s.reg.Delete(s.f.host); err != nil {
			t.Fatal(err)
		}
		if err := s.reg.Upsert(stranger); err != nil {
			t.Fatal(err)
		}
		before := *s.row(t)
		s.pass(t)
		if got := s.row(t); *got != before {
			t.Fatalf("a row of another incarnation was held: %+v (was %+v)", got, before)
		}
	})
	t.Run("an inventory of another enrollment", func(t *testing.T) {
		s := newSourceReconcile(t)
		before := *s.row(t)
		s.releasingThenInstalling(t)
		s.completeFromInstalling(t)
		contradicting := *s.census
		contradicting.Enrollment = "en-2"
		s.adapter.Census = &contradicting
		s.adapter.Configure(&s.cfg)
		s.pass(t)
		if got := s.row(t); *got != before {
			t.Fatalf("a contradicting inventory was held: %+v (was %+v)", got, before)
		}
	})
	t.Run("stubbed", func(t *testing.T) {
		r, selected, holds, _ := relocationReconciler(t)
		completedServing(t, r, selected.Host, "dst", "dst", parkLocal("box", "rel-1"))
		r.CheckHeld = func(context.Context, hostclaim.Observed, RecordSelection) error { return ErrTypedInventory }
		writes := 0
		if err := r.Install(context.Background(), selected, func() error { writes++; return nil }); !errors.Is(err, ErrTypedInventory) || writes != 0 || len(*holds) != 0 {
			t.Fatalf("a refusing CheckHeld: %v writes %d holds %v", err, writes, *holds)
		}
	})
}

// Review M2: holdOnly makes the catch-up decision again on the fresh read
// under the host lock and holds nothing when it differs. Install's
// decision is made, then (inside CheckHeld, which runs after the decision
// and before the lock) the claim is rewritten to another completed
// relocation, rel-2, which this box also owes a park-local for: the fresh
// read is still in catch-up scope, but for rel-2, so the locked decision
// no longer holds and nothing is held. In production this state is not
// reachable while the row exists: CompleteRelocation queues no second
// park-local for a box that already owes one, RelocationStartGuard refuses
// a source that owes one, and once rel-1's park-local retires the row,
// RegistryHoldRow finds none. The test pins the re-decision directly.
func TestReconcileCatchUpHoldIsDecidedAgainUnderTheLock(t *testing.T) {
	ctx := context.Background()
	r, selected, holds, _ := relocationReconciler(t)
	completedServing(t, r, selected.Host, "dst", "dst", parkLocal("box", "rel-1"))
	done, err := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-2", Stop: hostclaim.ProbeStop{Box: "dst", Session: 8, Receipt: "probe-fence-2", ReceiptSHA256: strings.Repeat("b", 64)}, Phase: hostclaim.PhaseRelocatingProbing})
	if err != nil {
		t.Fatal(err)
	}
	moved := false
	r.CheckHeld = func(ctx context.Context, _ hostclaim.Observed, _ RecordSelection) error {
		obs, err := hostclaim.Read(ctx, r.Claims, selected.Host)
		if err != nil {
			return err
		}
		_, err = hostclaim.Update(ctx, r.Claims, obs, func(c *hostclaim.Claim) error {
			c.Op.Journal[hostclaim.JournalCompleted] = string(done)
			c.Obligations = []hostclaim.Obligation{parkLocal("box", "rel-2")}
			return nil
		})
		moved = err == nil
		return err
	}
	writes := 0
	if err := r.Install(ctx, selected, func() error { writes++; return nil }); !errors.Is(err, ErrChanged) || !moved || writes != 0 || len(*holds) != 0 {
		t.Fatalf("a catch-up decision the locked read no longer makes: %v moved %v writes %d holds %v", err, moved, writes, *holds)
	}
	obs, err := hostclaim.Read(ctx, r.Claims, selected.Host)
	if err != nil {
		t.Fatal(err)
	}
	if d, ok, err := r.completedHold(obs.Claim); err != nil || !ok || d != (RowDecision{Case: RowRetiring, HoldOp: "rel-2"}) {
		t.Fatalf("the fresh read is not the rel-2 catch-up: %+v %v %v", d, ok, err)
	}
}