//go:build unix

package hostbox

import (
	"context"
	"errors"
	"path/filepath"
	"strconv"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// sourceReconcile is the box "box" as a relocation's source, built as the
// claim assembly builds it: the production ReconcileAdapter with the real
// census and park.Local, provision's pass, and Task 10b's production
// hooks. Its row is minted by a serving pass, then one cold writer session
// is admitted at the same stamp.
type sourceReconcile struct {
	f       admissionFixture
	reg     *registry.Registry
	inv     *StopInventoryDB
	census  *EmptyInstallationCensus
	guard   hostguard.Guard
	adapter ReconcileAdapter
	cfg     provision.ReconcileConfig
}

func newSourceReconcile(t *testing.T) *sourceReconcile {
	t.Helper()
	s := &sourceReconcile{f: newAdmissionFixture(t, false)}
	var err error
	if s.reg, err = registry.Open(filepath.Join(t.TempDir(), "registry.db")); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { s.reg.Close() })
	if s.inv, err = OpenStopInventoryDB(filepath.Join(t.TempDir(), "inventory.db")); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { s.inv.Close() })
	launches, err := OpenLaunchJournal(filepath.Join(t.TempDir(), "launches.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { launches.Close() })
	s.census = &EmptyInstallationCensus{Inventory: s.inv, Registry: s.reg, Launches: launches, Box: "box", Enrollment: "en", DirectoryRevision: "1",
		Roots: map[string]string{"data": s.f.data, "opt": t.TempDir(), "units": t.TempDir()}, SharedPaths: map[string][]string{"data": {"hostguard"}}}
	s.guard = hostguard.Guard{DataDir: s.f.data, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: s.f.store}}
	source := WorkSource{Guard: s.guard, Box: "box", Region: "ie", Now: time.Now,
		Policy: func(context.Context, *hostclaim.Claim, Selection, hostwrite.Identity) error { return nil },
		Local:  func(context.Context, *hostclaim.Claim, Selection) error { return nil }}
	s.adapter = ReconcileAdapter{Registry: s.reg, Inventory: s.inv, Census: s.census, Box: "box", Enrollment: "en", DirectoryRevision: "1",
		Reconciler: Reconciler{Claims: s.f.store, Box: "box", Region: "ie", Guard: s.guard, Writes: hostwrite.Config{Enabled: true, Selector: source},
			Local: park.Local{Guard: s.guard, Claims: s.f.store, DataDir: s.f.data, Box: "box"},
			Now:   func() time.Time { return time.Unix(100, 0) }, CheckRecord: func(context.Context, RecordSelection) error { return nil }, CheckLocal: s.census.CheckLocal,
			FenceClosed: AdmissionFenceClosed(s.guard), HoldRow: RegistryHoldRow(s.reg)}}
	s.cfg = provision.ReconcileConfig{DataDir: s.f.data, Region: "ie"}
	s.cfg.Layout = func(context.Context, release.Store) (string, error) { return release.LayoutModeNested, nil }
	s.adapter.Configure(&s.cfg)
	s.pass(t)
	if row := s.row(t); row == nil || row.Status != "ready" || row.HoldOp != "" || !row.ClaimOwnership.Valid() {
		t.Fatalf("the serving pass minted %+v", row)
	}
	s.update(t, func(c *hostclaim.Claim) error {
		return hostclaim.AdmitWriter(hostclaim.Admission{Box: "box", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true})(c)
	})
	return s
}

func (s *sourceReconcile) pass(t *testing.T) {
	t.Helper()
	if err := s.f.pass(t.Context(), s.f.store, s.reg, s.cfg); err != nil {
		t.Fatal(err)
	}
}

func (s *sourceReconcile) row(t *testing.T) *registry.Route {
	t.Helper()
	row, err := s.reg.Get(s.f.host)
	if err != nil {
		t.Fatal(err)
	}
	return row
}

func (s *sourceReconcile) update(t *testing.T, change func(*hostclaim.Claim) error) hostclaim.Observed {
	t.Helper()
	o, err := hostclaim.Read(t.Context(), s.f.store, s.f.host)
	if err != nil {
		t.Fatal(err)
	}
	if o, err = hostclaim.Update(t.Context(), s.f.store, o, change); err != nil {
		t.Fatal(err)
	}
	return o
}

// releasingThenInstalling moves the claim through a drain from "box" to
// "dst": releasing, root's release stop fencing the source's admission
// (BeginFence), then the installing CAS, which advances holder.stamp to
// the target's install stamp. No reconcile runs on the source in between.
func (s *sourceReconcile) releasingThenInstalling(t *testing.T) hostclaim.Observed {
	t.Helper()
	ctx := t.Context()
	o := s.update(t, func(c *hostclaim.Claim) error {
		c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseReleasing, Exec: "box", Participants: []string{"box", "dst"},
			Intent: map[string]string{"region": "ie", "reason": "drain", "source_box": "box", "target_box": "dst"}, Journal: map[string]string{"exec_ack.box": "{}", "exec_ack.dst": "{}"}}
		return nil
	})
	captured, err := hostguard.Capture(s.f.data, s.f.host)
	if err != nil {
		t.Fatal(err)
	}
	target := hostguard.FenceTarget{FenceID: "fence-1", Host: s.f.host, Incarnation: captured.Incarnation, Stamp: captured.Stamp, Session: o.Claim.Holder.Session,
		Op: "rel-1", Phase: hostclaim.PhaseReleasing, Box: "box", Enrollment: "en", BootID: "boot", ControllerID: "ctl", ClaimETag: o.ETag, DirectoryRevision: "1"}
	if _, err := s.guard.BeginFence(ctx, target); err != nil {
		t.Fatal(err)
	}
	return s.update(t, func(c *hostclaim.Claim) error {
		c.Holder.Stamp = c.Gen + 1
		c.Op.Phase, c.Op.Exec = hostclaim.PhaseInstalling, "dst"
		c.Op.Journal["install_stamp"] = strconv.FormatUint(c.Holder.Stamp, 10)
		return nil
	})
}

// Astra Task 10b round 1 (P2; rev 5.4.1): the source's fence closes in
// releasing, but its first reconcile after that runs only once the claim
// is at installing, with holder.stamp at the target's install stamp, which
// this box never had. Through the production adapter, census and
// provision pass, the source's row is held all the same, its ownership
// unchanged: the hold-only path checks the source's own installation (the
// row's stamp and its inventory), never the holder's stamp, and grants
// nothing, so the guard record and the claim are unchanged. A replay
// changes nothing.
func TestReconcileHoldsTheSourceWhenItsFirstPassAfterTheFenceIsAtInstalling(t *testing.T) {
	s := newSourceReconcile(t)
	minted := *s.row(t)
	o := s.releasingThenInstalling(t)
	if o.Claim.Holder.Stamp == minted.ClaimStamp {
		t.Fatal("the installing CAS kept the source's stamp")
	}
	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" || row.ClaimOwnership != minted.ClaimOwnership {
			t.Fatalf("pass %d at installing: 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)
		}
		after, err := hostclaim.Read(t.Context(), s.f.store, s.f.host)
		if err != nil || after.ETag != o.ETag {
			t.Fatalf("pass %d: the claim was written: %v", i, err)
		}
	}
}

// The hold-only check (EmptyInstallationCensus.CheckHeld) accepts the
// source's own existing installation at installing, where main's
// CheckLocal refuses (it asks for the target's stamp). Absence is never
// evidence: an inventory of another box or enrollment, no inventory, a
// row owned by another row identity, another incarnation, no row and an
// unreadable registry all refuse. Through the pass, a contradicting
// inventory leaves the row exactly as it was.
func TestCensusCheckHeldRefusesAContradictingOrMissingSourceInventory(t *testing.T) {
	ctx := t.Context()
	s := newSourceReconcile(t)
	minted := *s.row(t)
	o := s.releasingThenInstalling(t)
	selected := RecordSelection{Host: s.f.host, Incarnation: o.Claim.Holder.Incarnation}
	if err := s.census.CheckHeld(ctx, o, selected); err != nil {
		t.Fatalf("the source's own installation: %v", err)
	}
	if err := s.census.CheckLocal(ctx, o, selected); !errors.Is(err, ErrTypedInventory) {
		t.Fatalf("main's CheckLocal at installing: %v (it asks for the target's stamp)", err)
	}
	empty, err := OpenStopInventoryDB(filepath.Join(t.TempDir(), "inventory.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { empty.Close() })
	otherRows, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { otherRows.Close() })
	stranger := minted
	stranger.ClaimRowID = "another-row"
	if err := otherRows.Upsert(stranger); err != nil {
		t.Fatal(err)
	}
	for name, tc := range map[string]struct {
		change func(*EmptyInstallationCensus, *RecordSelection)
		want   error
	}{
		"another box":         {func(c *EmptyInstallationCensus, _ *RecordSelection) { c.Box = "dst" }, ErrTypedInventory},
		"another enrollment":  {func(c *EmptyInstallationCensus, _ *RecordSelection) { c.Enrollment = "en-2" }, ErrTypedInventory},
		"no inventory":        {func(c *EmptyInstallationCensus, _ *RecordSelection) { c.Inventory = empty }, ErrTypedInventory},
		"another row id":      {func(c *EmptyInstallationCensus, _ *RecordSelection) { c.Registry = otherRows }, ErrTypedInventory},
		"another incarnation": {func(_ *EmptyInstallationCensus, sel *RecordSelection) { sel.Incarnation = "inc-other" }, ErrChanged},
	} {
		c, sel := *s.census, selected
		tc.change(&c, &sel)
		if err := c.CheckHeld(ctx, o, sel); !errors.Is(err, tc.want) {
			t.Fatalf("%s: %v, want %v", name, err, tc.want)
		}
	}
	noRow, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { noRow.Close() })
	c := *s.census
	c.Registry = noRow
	if err := c.CheckHeld(ctx, o, selected); !errors.Is(err, ErrChanged) {
		t.Fatalf("no row: %v", err)
	}
	closed, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	closed.Close()
	c.Registry = closed
	if err := c.CheckHeld(ctx, o, selected); err == nil || errors.Is(err, ErrChanged) || errors.Is(err, ErrTypedInventory) {
		t.Fatalf("an unreadable registry: %v (want the read's error)", err)
	}
	// Through the pass: a contradicting inventory refuses the hold.
	contradicting := *s.census
	contradicting.Enrollment = "en-2"
	s.adapter.Census = &contradicting
	s.adapter.Configure(&s.cfg)
	s.pass(t)
	if row := s.row(t); *row != minted {
		t.Fatalf("a contradicting inventory: the row is %+v (was %+v)", row, minted)
	}
}

// CheckHeld accepts the source's installation only from a complete
// inventory with exactly one materialized route entry for the host, owned
// by a row for the selected incarnation. The inventory as recorded is
// accepted from a fresh inventory database; the same inventory marked
// incomplete, with its route entry removed, with that entry recorded
// twice, or with that entry only intended is ErrTypedInventory. So is a
// row owned by another incarnation while the claim and the selection
// still match, even with a complete inventory of that incarnation that
// passes every later check.
func TestCensusCheckHeldRefusesAnInventoryThatDoesNotProveTheRowsInstallation(t *testing.T) {
	ctx := t.Context()
	s := newSourceReconcile(t)
	minted := *s.row(t)
	o := s.releasingThenInstalling(t)
	selected := RecordSelection{Host: s.f.host, Incarnation: o.Claim.Holder.Incarnation}
	recorded, err := s.inv.ReadInstallation(ctx, s.f.host, minted.ClaimIncarnation, minted.ClaimStamp)
	if err != nil {
		t.Fatal(err)
	}
	isRoute := func(e StopInventoryEntry) bool { return e.Kind == "route" && e.RelativePath == s.f.host }
	routes := 0
	for _, e := range recorded.Entries {
		if isRoute(e) {
			routes++
		}
	}
	if !recorded.Complete || routes != 1 {
		t.Fatalf("the recorded inventory: complete %v, %d route entries", recorded.Complete, routes)
	}
	for _, tc := range []struct {
		name   string
		change func(*StopInventory)
		want   error
		row    func(*registry.Route)
	}{
		{"as recorded", func(*StopInventory) {}, nil, nil},
		{"incomplete", func(in *StopInventory) { in.Complete = false }, ErrTypedInventory, nil},
		{"no route entry", func(in *StopInventory) {
			kept := []StopInventoryEntry{}
			for _, e := range in.Entries {
				if !isRoute(e) {
					kept = append(kept, e)
				}
			}
			in.Entries = kept
		}, ErrTypedInventory, nil},
		{"two route entries", func(in *StopInventory) {
			for _, e := range in.Entries {
				if isRoute(e) {
					e.ArtifactID, e.Inode = e.ArtifactID+"-2", e.Inode+1
					in.Entries = append(in.Entries, e)
					break
				}
			}
		}, ErrTypedInventory, nil},
		{"route entry only intended", func(in *StopInventory) {
			for i, e := range in.Entries {
				if isRoute(e) {
					in.Entries[i].Materialization = "intended"
				}
			}
		}, ErrTypedInventory, nil},
		// The row's own incarnation check is the only guard here: the
		// inventory is complete and recorded at the row's incarnation and
		// stamp, so every later check would accept it.
		{"row owned by another incarnation", func(in *StopInventory) { in.Incarnation = "inc-other" }, ErrTypedInventory,
			func(r *registry.Route) { r.ClaimIncarnation = "inc-other" }},
	} {
		in := recorded
		in.Entries = append([]StopInventoryEntry(nil), recorded.Entries...)
		tc.change(&in)
		db, err := OpenStopInventoryDB(filepath.Join(t.TempDir(), "inventory.db"))
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() { db.Close() })
		if err := db.WriteStopInventory(ctx, in); err != nil {
			t.Fatalf("%s: writing the inventory: %v", tc.name, err)
		}
		c := *s.census
		c.Inventory = db
		if tc.row != nil {
			row := minted
			tc.row(&row)
			if row.ClaimIncarnation == selected.Incarnation {
				t.Fatalf("%s: the row's incarnation still matches the selection", tc.name)
			}
			rows, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
			if err != nil {
				t.Fatal(err)
			}
			t.Cleanup(func() { rows.Close() })
			if err := rows.Upsert(row); err != nil {
				t.Fatal(err)
			}
			c.Registry = rows
		}
		if err := c.CheckHeld(ctx, o, selected); !errors.Is(err, tc.want) || (tc.want == nil && err != nil) {
			t.Fatalf("%s: %v, want %v", tc.name, err, tc.want)
		}
	}
}

func TestPlan14RemoteSourceHoldReadsInventoryOutsideHostLock(t *testing.T) {
	for _, stale := range []bool{false, true} {
		t.Run(strconv.FormatBool(stale), func(t *testing.T) {
			s := newSourceReconcile(t)
			before := *s.row(t)
			s.releasingThenInstalling(t)
			captured := hostguard.Captured{Incarnation: before.ClaimIncarnation, Stamp: before.ClaimStamp}
			calls := 0
			_, remote := RemoteJournals(func(ctx context.Context, q JournalRequest) (JournalResponse, error) {
				if q.Op != "inventory-version" {
					t.Fatalf("hold requested %s", q.Op)
				}
				calls++
				var out JournalResponse
				err := s.guard.WithLocal(ctx, s.f.host, func() error {
					var err error
					out, err = s.inv.inventoryVersion(ctx, q.Target)
					out.ID, out.V = q.ID, 1
					return err
				})
				return out, err
			})
			s.adapter.Inventory = remote
			census := *s.census
			census.Inventory = remote
			s.adapter.Census = &census
			s.adapter.Reconciler.Guard.FenceView = func(ctx context.Context, host string, c hostguard.Captured) (hostguard.FenceView, error) {
				if c != captured {
					return hostguard.FenceView{}, ErrChanged
				}
				_, revision, err := s.inv.Snapshot(ctx, host, c)
				if stale {
					revision++
				}
				return hostguard.FenceView{Schema: 1, Host: host, Captured: c, Installation: hostguard.RootInstallation{InventoryRevision: revision}}, err
			}
			s.adapter.Configure(&s.cfg)
			ctx, cancel := context.WithTimeout(t.Context(), time.Second)
			defer cancel()
			if err := s.f.pass(ctx, s.f.store, s.reg, s.cfg); err != nil {
				t.Fatal(err)
			}
			row := s.row(t)
			if calls != 1 {
				t.Fatalf("inventory RPCs=%d", calls)
			}
			if ctx.Err() != nil {
				t.Fatal("inventory read deadlocked", ctx.Err())
			}
			if stale {
				if *row != before {
					t.Fatal("stale inventory changed row")
				}
				return
			}
			if row.Status != "relocating" || row.HoldOp != "rel-1" || row.ClaimOwnership != before.ClaimOwnership {
				t.Fatalf("hold: %+v", row)
			}
		})
	}
}