//go:build unix

package host

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostbox"
	"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/registry"
	"github.com/carlosframework/platform/internal/release"
)

const sourceHoldHost = "shop.example"

// sourceHoldBox is the relocation's source "box" with its installation
// active in the host guard, one cold writer session admitted, a ready row
// owned by that installation, and the reconciler the claim assembly
// builds (the WorkSource selector, park.Local), its hooks set by
// relocationSourceRows under getenv and counted.
type sourceHoldBox struct {
	t        *testing.T
	store    release.Store
	reg      *registry.Registry
	guard    hostguard.Guard
	r        hostbox.Reconciler
	selected hostbox.RecordSelection
	owner    registry.ClaimOwnership
	asked    int // FenceClosed calls
	holds    int // HoldRow calls
	writes   int // the activation path's row writes
}

func newSourceHoldBox(t *testing.T, getenv func(string) string) *sourceHoldBox {
	t.Helper()
	ctx := t.Context()
	b := &sourceHoldBox{t: t, store: release.NewDirStore(t.TempDir())}
	dir := t.TempDir()
	locks := filepath.Join(dir, "locks")
	for _, p := range []string{locks, filepath.Join(dir, "hostguard")} {
		if err := os.Mkdir(p, 0770); err != nil {
			t.Fatal(err)
		}
	}
	obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost)
	if err != nil {
		t.Fatal(err)
	}
	obs, err = hostclaim.Update(ctx, b.store, obs, hostclaim.Adopt(hostclaim.NewHolder{Family: "instance", Key: "apps/acct/app/instances/" + sourceHoldHost + ".json", Account: "acct", App: "app", Host: sourceHoldHost, Region: "ie"}, "box", nil, nil, "adopt", func() string { return "id" }))
	if err != nil {
		t.Fatal(err)
	}
	b.guard = hostguard.Guard{DataDir: dir, LockDir: locks, Claims: hostguard.StoreReader{Store: b.store}}
	if err := b.guard.Activate(ctx, sourceHoldHost, "box", nil); err != nil {
		t.Fatal(err)
	}
	obs, err = hostclaim.Update(ctx, b.store, obs, hostclaim.AdmitWriter(hostclaim.Admission{Box: "box", Mode: "cold", NewGeneration: "g1", Region: "ie", KeepStamp: true}))
	if err != nil {
		t.Fatal(err)
	}
	captured, err := hostguard.Capture(dir, sourceHoldHost)
	if err != nil {
		t.Fatal(err)
	}
	if b.reg, err = registry.Open(filepath.Join(t.TempDir(), "registry.db")); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { b.reg.Close() })
	b.owner = registry.ClaimOwnership{ClaimIncarnation: captured.Incarnation, ClaimStamp: captured.Stamp, ClaimOwner: "owner-1", ClaimRowID: "row-1"}
	if err := b.reg.Upsert(registry.Route{Host: sourceHoldHost, Status: "ready", Socket: "/run/shop.sock", ClaimOwnership: b.owner}); err != nil {
		t.Fatal(err)
	}
	now := func() time.Time { return time.Unix(100, 0) }
	source := hostbox.WorkSource{Guard: b.guard, Box: "box", Region: "ie", Now: now,
		Policy: hostbox.Policy{Held: func(context.Context, string) (string, error) { return "", nil }, Relocation: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil }}.Check,
		Local:  func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }}
	h := obs.Claim.Holder
	b.selected = hostbox.RecordSelection{Host: sourceHoldHost, Key: h.Key, ETag: "record-etag", Incarnation: h.Incarnation, Family: h.Family, Layout: "flat", Paths: []string{"shop.example.db"}}
	b.r = hostbox.Reconciler{Claims: b.store, Guard: b.guard, Local: park.Local{Guard: b.guard, Claims: b.store, DataDir: dir, Box: "box"}, Writes: hostwrite.Config{Enabled: true, Selector: source}, Box: "box", Region: "ie", Now: now,
		CheckRecord: func(context.Context, hostbox.RecordSelection) error { return nil },
		CheckLocal:  func(context.Context, hostclaim.Observed, hostbox.RecordSelection) error { return nil },
		CheckHeld:   func(context.Context, hostclaim.Observed, hostbox.RecordSelection) error { return nil }}
	relocationSourceRows(getenv, &b.r, b.guard, b.reg)
	if fence := b.r.FenceClosed; fence != nil {
		b.r.FenceClosed = func(ctx context.Context, o hostclaim.Observed) (bool, error) { b.asked++; return fence(ctx, o) }
	}
	if hold := b.r.HoldRow; hold != nil {
		b.r.HoldRow = func(ctx context.Context, host, op string) error { b.holds++; return hold(ctx, host, op) }
	}
	return b
}

func (b *sourceHoldBox) update(change func(*hostclaim.Claim)) hostclaim.Observed {
	b.t.Helper()
	ctx := b.t.Context()
	obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost)
	if err != nil {
		b.t.Fatal(err)
	}
	obs, err = hostclaim.Update(ctx, b.store, obs, func(c *hostclaim.Claim) error { change(c); return nil })
	if err != nil {
		b.t.Fatal(err)
	}
	return obs
}

// releasing is a drain from "box" to "dst" in releasing, both
// acknowledged.
func (b *sourceHoldBox) releasing() hostclaim.Observed {
	return b.update(func(c *hostclaim.Claim) {
		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": "{}"}}
	})
}

// closeFence is root beginning the source's release stop: the durable
// admission fence in the guard record (hostguard.Guard.BeginFence).
func (b *sourceHoldBox) closeFence() {
	b.t.Helper()
	ctx := b.t.Context()
	captured, err := hostguard.Capture(b.guard.DataDir, sourceHoldHost)
	if err != nil {
		b.t.Fatal(err)
	}
	obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost)
	if err != nil {
		b.t.Fatal(err)
	}
	target := hostguard.FenceTarget{FenceID: "fence-1", Host: sourceHoldHost, Incarnation: captured.Incarnation, Stamp: captured.Stamp, Session: obs.Claim.Holder.Session,
		Op: "rel-1", Phase: hostclaim.PhaseReleasing, Box: "box", Enrollment: "enrollment-1", BootID: "boot", ControllerID: "ctl", ClaimETag: obs.ETag, DirectoryRevision: "4"}
	if _, err := b.guard.BeginFence(ctx, target); err != nil {
		b.t.Fatal(err)
	}
}

func (b *sourceHoldBox) install(dry bool) error {
	sel := b.selected
	sel.DryRun = dry
	return b.r.Install(b.t.Context(), sel, func() error { b.writes++; return nil })
}

func (b *sourceHoldBox) row() *registry.Route {
	b.t.Helper()
	row, err := b.reg.Get(sourceHoldHost)
	if err != nil {
		b.t.Fatal(err)
	}
	return row
}

func (b *sourceHoldBox) guardRecord() []byte {
	b.t.Helper()
	raw, err := os.ReadFile(filepath.Join(b.guard.DataDir, "hostguard", sourceHoldHost+".json"))
	if err != nil {
		b.t.Fatal(err)
	}
	return raw
}

// retireRow is the source's park-local retiring its row: the typed
// retirement's registry step (hostbox.RegistryRows, the one park-local
// row write), by the row's claim ownership, held or not.
func (b *sourceHoldBox) retireRow() {
	b.t.Helper()
	rows := hostbox.RegistryRows{Registry: b.reg, Inventory: hostbox.RetirementInventory{Host: sourceHoldHost, Captured: hostguard.Captured{Incarnation: b.owner.ClaimIncarnation, Stamp: b.owner.ClaimStamp}, Ownership: b.owner.ClaimOwner}}
	route := hostbox.Artifact{Kind: hostbox.ArtifactRoute, Root: "registry", Name: sourceHoldHost, Owner: b.owner.ClaimOwner, RowID: b.owner.ClaimRowID}
	if err := rows.RetireRowsJournal(b.t.Context(), "retire-1", b.owner.ClaimOwner, []hostbox.Artifact{route}, func(json.RawMessage) error { return nil }); err != nil {
		b.t.Fatalf("the park-local's row retirement: %v", err)
	}
}

func regionHA(on bool) func(string) string {
	return func(k string) string {
		if k == "CARLOS_REGION_HA" && on {
			return "1"
		}
		return ""
	}
}

// complete is the target's probe passing and root's stop of the probe
// session, then the real serving CAS (hostclaim.CompleteRelocation),
// which queues the source's park-local unless a fence discharged it.
func (b *sourceHoldBox) complete() hostclaim.Observed {
	b.t.Helper()
	obs := b.completeOwing()
	if !hostclaim.OwesLocal(obs.Claim, "box") {
		b.t.Fatalf("after the completion: obligations %+v", obs.Claim.Obligations)
	}
	return obs
}

// completeOwing is complete without asking which boxes owe a park-local.
func (b *sourceHoldBox) completeOwing() hostclaim.Observed {
	b.t.Helper()
	b.update(func(c *hostclaim.Claim) {
		h := c.Holder
		h.Stamp, h.Session, h.Generation, h.WriterBox = h.Stamp+5, 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, c.Op.Exec = hostclaim.PhaseRelocatingProbing, "dst"
		c.Op.Journal["probe"], c.Op.Journal["probe_stop"] = string(probe), string(stop)
	})
	ctx := b.t.Context()
	obs, err := hostclaim.Read(ctx, b.store, sourceHoldHost)
	if err != nil {
		b.t.Fatal(err)
	}
	n := 0
	obs, err = hostclaim.Update(ctx, b.store, obs, hostclaim.CompleteRelocation("rel-1", func() string { n++; return "id-" + strconv.Itoa(n) }))
	if err != nil {
		b.t.Fatalf("the serving CAS: %v", err)
	}
	if obs.Claim.Op.Kind != hostclaim.KindServing || obs.Claim.Op.Exec != "dst" {
		b.t.Fatalf("after the completion: op %+v obligations %+v", obs.Claim.Op, obs.Claim.Obligations)
	}
	return obs
}

// leaves asserts reconcile, dry run or not, refuses with nothing done:
// the row exactly as before, the fence not asked, nothing held or
// written, and the guard record as it was.
func (b *sourceHoldBox) leaves(what string, before registry.Route, record []byte) {
	b.t.Helper()
	asked, holds, writes := b.asked, b.holds, b.writes
	for _, dry := range []bool{true, false} {
		if err := b.install(dry); !errors.Is(err, hostbox.ErrChanged) {
			b.t.Fatalf("%s (dry run %v): %v", what, dry, err)
		}
	}
	if got := b.row(); got == nil || *got != before || b.asked != asked || b.holds != holds || b.writes != writes || !bytes.Equal(b.guardRecord(), record) {
		b.t.Fatalf("%s: row %+v (was %+v) fence asked %d holds %d writes %d", what, got, before, b.asked-asked, b.holds-holds, b.writes-writes)
	}
}

// replays asserts reconcile, dry run or not, answers the catch-up hold's
// replay (L18 finding 3): nil, with the row exactly as before, the fence
// not asked, nothing activated or written, and the guard record as it was.
func (b *sourceHoldBox) replays(what string, before registry.Route, record []byte) {
	b.t.Helper()
	asked, writes := b.asked, b.writes
	for _, dry := range []bool{true, false} {
		if err := b.install(dry); err != nil {
			b.t.Fatalf("%s (dry run %v): %v", what, dry, err)
		}
	}
	if got := b.row(); got == nil || *got != before || b.asked != asked || b.writes != writes || !bytes.Equal(b.guardRecord(), record) {
		b.t.Fatalf("%s: row %+v (was %+v) fence asked %d writes %d", what, got, before, b.asked-asked, b.writes-writes)
	}
}

// Task 10b test 3. The completion: the source's row, held at its fence's
// close, stays held through the real serving CAS. Reconcile never
// releases it: under the serving claim the source owes the completed op's
// park-local, so reconcile's catch-up hold (L18 finding 3) replays the
// hold and writes nothing. The park-local's retirement deletes the held
// row by its ownership: that ends the hold, and nothing is minted after. A failover
// from a fenced source: the source is on lost, so reconcile neither asks
// its fence nor holds its row, whether the row was ready or already held
// when the box was fenced, both while the failover runs and after its
// completion queues the park-local; the row is left exactly as it was,
// and the park-local's retirement removes it.
func TestRelocationSourceHoldEndsWithTheCompletionOrTheParkLocal(t *testing.T) {
	t.Parallel()
	t.Run("completion", func(t *testing.T) {
		b := newSourceHoldBox(t, regionHA(true))
		b.releasing()
		b.closeFence()
		if err := b.install(false); err != nil {
			t.Fatalf("the fenced source's hold: %v", err)
		}
		held := *b.row()
		if held.Status != "relocating" || held.HoldOp != "rel-1" || b.holds != 1 {
			t.Fatalf("held row %+v holds %d", held, b.holds)
		}
		b.complete()
		b.replays("the source after the completion", held, b.guardRecord())
		b.retireRow()
		if err := b.install(false); !errors.Is(err, hostbox.ErrChanged) || b.row() != nil || b.writes != 0 {
			t.Fatalf("after the park-local: %v row %+v writes %d", err, b.row(), b.writes)
		}
	})
	for _, fencedHeld := range []bool{false, true} {
		name := "failover, row ready when fenced"
		if fencedHeld {
			name = "failover, row already held when fenced"
		}
		t.Run(name, func(t *testing.T) {
			b := newSourceHoldBox(t, regionHA(true))
			b.releasing()
			if fencedHeld {
				b.closeFence()
				if err := b.install(false); err != nil {
					t.Fatalf("the fenced source's hold: %v", err)
				}
			}
			before, record := *b.row(), b.guardRecord()
			// hostclaim.FenceSource on a source terminated before its
			// release is journaled: a failover in pinning, the source off
			// the participants and on lost, its fence journaled.
			b.update(func(c *hostclaim.Claim) {
				c.Op.Phase, c.Op.Exec, c.Op.Participants = hostclaim.PhasePinning, "none", []string{"dst"}
				c.Op.Journal["reason"], c.Op.Journal["fence.box"] = hostclaim.ReasonFailover, `{"key":"fences/box.json"}`
				c.Lost = append(c.Lost, "box")
			})
			b.leaves("the fenced source during the failover", before, record)
			// No qualified discharge: the completion queues its park-local.
			b.complete()
			b.leaves("the fenced source after the failover's completion", before, record)
			b.retireRow()
			if got := b.row(); got != nil {
				t.Fatalf("the park-local left the row: %+v", got)
			}
		})
	}
}

// Task 10b test 4. With CARLOS_REGION_HA off, or any value that is not
// true, relocationSourceRows sets neither hook and reconcile is main's:
// the serving executor writes its row through the activation path, and a
// relocating claim is refused by Serving with the row, the claim and the
// guard record unchanged, even once the source's fence has closed. With
// it on, the same box holds the row at the fence's close.
func TestRelocationSourceHoldFollowsTheFlag(t *testing.T) {
	t.Parallel()
	for _, v := range []string{"", "0", "false", "off", "yes please", "1", "true"} {
		on := v == "1" || v == "true"
		t.Run("CARLOS_REGION_HA="+v, func(t *testing.T) {
			b := newSourceHoldBox(t, func(k string) string {
				if k == "CARLOS_REGION_HA" {
					return v
				}
				return ""
			})
			if (b.r.FenceClosed != nil) != on || (b.r.HoldRow != nil) != on {
				t.Fatalf("hooks set %v %v with the flag %q", b.r.FenceClosed != nil, b.r.HoldRow != nil, v)
			}
			if err := b.install(false); err != nil || b.writes != 1 {
				t.Fatalf("the serving executor: %v writes %d", err, b.writes)
			}
			b.releasing()
			b.closeFence()
			before, record := *b.row(), b.guardRecord()
			obs, err := hostclaim.Read(t.Context(), b.store, sourceHoldHost)
			if err != nil {
				t.Fatal(err)
			}
			if on {
				if err := b.install(false); err != nil {
					t.Fatalf("region HA on: %v", err)
				}
				if got := b.row(); got.Status != "relocating" || got.HoldOp != "rel-1" {
					t.Fatalf("region HA on: row %+v", got)
				}
				return
			}
			for _, dry := range []bool{true, false} {
				if err := b.install(dry); !errors.Is(err, hostbox.ErrChanged) {
					t.Fatalf("region HA off (dry run %v): %v", dry, err)
				}
			}
			now, err := hostclaim.Read(t.Context(), b.store, sourceHoldHost)
			if err != nil {
				t.Fatal(err)
			}
			if got := b.row(); *got != before || b.writes != 1 || now.ETag != obs.ETag || !bytes.Equal(b.guardRecord(), record) {
				t.Fatalf("region HA off changed something: row %+v (was %+v) writes %d claim %v", got, before, b.writes, now.ETag != obs.ETag)
			}
		})
	}
}

// L18 joint round, finding 3: the source releases and its fence closes,
// its reconcile is delayed, the target completes, and only then does the
// source reconcile. The claim serves on another box, so Task 10b's
// relocating scope no longer reaches it; the catch-up hold does. Through
// the production hooks (relocationSourceRows) and the real reconcile: the
// row ends held for the completed op, never ready, its ownership and
// every other column as they were, the fence not asked, nothing activated
// and the claim and the guard record unchanged; a replay changes nothing;
// the park-local's retirement then ends it and nothing is minted.
//
// The guards: a row another op holds is never taken; with region HA off
// reconcile is main's (refused, the row ready and unchanged); a box that
// was not the relocation's source, or owes a park-local for another op
// only, does nothing; a box on lost is left to its park-local (Task 10b
// decision 3, the failover test above); a completion record that cannot
// be read is its error, never "no completion", with the row unchanged.
func TestRelocationSourceCatchUpHoldAfterTheCompletion(t *testing.T) {
	t.Parallel()
	t.Run("delayed source reconcile", func(t *testing.T) {
		b := newSourceHoldBox(t, regionHA(true))
		minted := *b.row()
		b.releasing()
		b.closeFence()
		done := b.complete()
		record := b.guardRecord()
		if err := b.install(false); err != nil {
			t.Fatalf("the source's first reconcile after the completion: %v", err)
		}
		got := *b.row()
		if got.Status != "relocating" || got.HoldOp != "rel-1" || b.holds != 1 || b.asked != 0 || b.writes != 0 {
			t.Fatalf("the source's row %+v holds %d fence asked %d writes %d", got, b.holds, b.asked, b.writes)
		}
		held := got
		held.Status, held.HoldOp = minted.Status, minted.HoldOp
		if held != minted {
			t.Fatalf("the hold changed more than status and hold_op: %+v (minted %+v)", got, minted)
		}
		if now, err := hostclaim.Read(t.Context(), b.store, sourceHoldHost); err != nil || now.ETag != done.ETag || !bytes.Equal(b.guardRecord(), record) {
			t.Fatalf("the catch-up hold wrote the claim or the guard record: %v", err)
		}
		b.replays("the replay", got, record)
		b.retireRow()
		if err := b.install(false); !errors.Is(err, hostbox.ErrChanged) || b.row() != nil || b.writes != 0 {
			t.Fatalf("after the park-local: %v row %+v writes %d", err, b.row(), b.writes)
		}
	})
	t.Run("another op's row is never taken", func(t *testing.T) {
		b := newSourceHoldBox(t, regionHA(true))
		b.releasing()
		b.closeFence()
		if ok, err := b.reg.SetHoldIf(sourceHoldHost, "moving", "mv-9", ""); err != nil || !ok {
			t.Fatalf("holding the row for another op: %v %v", ok, err)
		}
		before := *b.row()
		b.complete()
		for _, dry := range []bool{true, false} {
			if err := b.install(dry); !dry && !errors.Is(err, hostbox.ErrChanged) {
				t.Fatalf("another op's row: %v", err)
			}
		}
		if got := b.row(); *got != before || b.writes != 0 {
			t.Fatalf("another op's row was taken: %+v (was %+v) writes %d", got, before, b.writes)
		}
	})
	t.Run("region HA off", func(t *testing.T) {
		b := newSourceHoldBox(t, regionHA(false))
		b.releasing()
		b.closeFence()
		b.complete()
		b.leaves("region HA off after the completion", *b.row(), b.guardRecord())
		if got := b.row(); got.Status != "ready" || got.HoldOp != "" {
			t.Fatalf("region HA off: row %+v", got)
		}
	})
	t.Run("not the source", func(t *testing.T) {
		b := newSourceHoldBox(t, regionHA(true))
		before, record := *b.row(), b.guardRecord()
		// A drain of the host from "other" to "dst": this box holds a
		// stale local record and row, but was never a participant.
		b.update(func(c *hostclaim.Claim) {
			c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseReleasing, Exec: "other", Participants: []string{"other", "dst"},
				Intent: map[string]string{"region": "ie", "reason": "drain", "source_box": "other", "target_box": "dst"}, Journal: map[string]string{"exec_ack.other": "{}", "exec_ack.dst": "{}"}}
		})
		done := b.completeOwing()
		if hostclaim.OwesLocal(done.Claim, "box") || !hostclaim.OwesLocal(done.Claim, "other") {
			t.Fatalf("the completion's park-locals: %+v", done.Claim.Obligations)
		}
		b.leaves("a box that was not the source", before, record)
	})
	t.Run("a park-local for another op only", func(t *testing.T) {
		b := newSourceHoldBox(t, regionHA(true))
		b.releasing()
		b.closeFence()
		b.complete()
		b.update(func(c *hostclaim.Claim) {
			for i := range c.Obligations {
				if c.Obligations[i].Kind == "park-local" && c.Obligations[i].Box == "box" {
					c.Obligations[i].Op = "rel-0"
				}
			}
		})
		b.leaves("a park-local for another op", *b.row(), b.guardRecord())
	})
	t.Run("an unreadable completion record", func(t *testing.T) {
		b := newSourceHoldBox(t, regionHA(true))
		b.releasing()
		b.closeFence()
		b.complete()
		b.update(func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalCompleted] = "{" })
		before, record := *b.row(), b.guardRecord()
		for _, dry := range []bool{true, false} {
			if err := b.install(dry); err == nil || errors.Is(err, hostbox.ErrChanged) || !errors.Is(err, hostclaim.ErrRelocation) {
				t.Fatalf("an unreadable completion record (dry run %v): %v (want the read's error)", dry, err)
			}
		}
		if got := b.row(); *got != before || b.holds != 0 || b.writes != 0 || !bytes.Equal(b.guardRecord(), record) {
			t.Fatalf("an unreadable completion record changed something: row %+v holds %d writes %d", got, b.holds, b.writes)
		}
	})
}