//go:build unix

package host

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

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostrelocate"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// Ruling 66 (Astra round 3 finding 2) and ruling 76 (Astra round 6
// findings 1 and 3): after the serving CAS, the target discharges main's
// participation, releases its row's relocating hold, lifts the fence its
// probe stop installed under the persisted owner, and marks the completed
// relocation discharged, in that order. A refused discharge leaves all of
// it, so the edge keeps the row held, the activator wakes nothing and no
// successor starts; a holder moved off the probe session while the row is
// held is refused; another operation's fence is never lifted; a replay
// after the hold went discharges again (positive evidence, never the
// released hold's inference), then lifts and marks; a row held by another
// op, or a record whose stop was on another box, is refused with nothing
// done; a claim that does not serve on this box is left alone and is
// ErrChanged.
func TestRelocationReleaseHoldWaitsForDischarge(t *testing.T) {
	t.Parallel()
	const host, box = "shop.example", "dst"
	ctx := t.Context()
	created := time.Date(2026, 10, 7, 0, 0, 0, 0, time.UTC)
	inc, err := hostclaim.Incarnation("acct", "app", host, created)
	if err != nil {
		t.Fatal(err)
	}
	done, _ := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-1", Stop: hostclaim.ProbeStop{Box: box, Session: 12, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("a", 64)}, Stamp: 9, Phase: hostclaim.PhaseRelocatingProbing, Generation: "P1"})
	claim := func(change func(*hostclaim.Claim)) *hostclaim.Claim {
		c := &hostclaim.Claim{V: 2, Gen: 31, Host: host,
			Holder: &hostclaim.Holder{Family: "instance", Key: "apps/acct/app/instances/" + host + ".json", Account: "acct", App: "app", Incarnation: inc, CreatedAt: created,
				Location: "instances/" + host, Stamp: 9, Session: 12, Generation: "P1", WriterBox: box, Admissions: []hostclaim.AdmissionRecord{{Session: 12, Box: box, Generation: "P1", Mode: "cold", Region: "ie"}}},
			Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: box, Participants: []string{box}, Journal: map[string]string{hostclaim.JournalCompleted: string(done)}}}
		if change != nil {
			change(c)
		}
		return c
	}
	type world struct {
		order     []string
		store     release.Store
		reg       *registry.Registry
		act       *stopProbeActivator
		owner     string
		marker    string
		discharge error
		// heldAt is, per discharge, whether the row was still held for
		// rel-1 when it ran.
		heldAt []bool
		hold   func(context.Context, string, string) error
	}
	setup := func(t *testing.T, c *hostclaim.Claim) *world {
		t.Helper()
		w := &world{store: release.NewDirStore(t.TempDir())}
		raw, _ := json.Marshal(c)
		key, _ := hostclaim.Key(host)
		if _, err := w.store.Put(ctx, key, raw, ""); err != nil {
			t.Fatal(err)
		}
		if w.reg, err = registry.Open(filepath.Join(t.TempDir(), "registry.db")); err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() { w.reg.Close() })
		if err := w.reg.Upsert(registry.Route{Host: host, Status: "relocating", HoldOp: "rel-1", Socket: "/run/shop.sock"}); err != nil {
			t.Fatal(err)
		}
		// The probe stop's fence, under the owner it persisted.
		dataDir := t.TempDir()
		marker, err := relocationProbeFencePath(dataDir, "rel-1")
		if err != nil {
			t.Fatal(err)
		}
		w.marker = marker
		if w.owner, err = relocationProbeFenceOwner(marker, host, 12); err != nil {
			t.Fatal(err)
		}
		w.act = newStopProbeActivator(&w.order)
		fctx, cancel := context.WithTimeout(ctx, time.Second)
		defer cancel()
		if _, err := w.act.svc.FenceAndStopAs(fctx, host, 12, w.owner); err != nil && !errors.Is(err, activator.ErrStopUnproven) {
			t.Fatal(err)
		}
		// The claim writes go through markStore, so the mark has its place
		// in the order.
		a := ClaimAssembly{Claims: markStore{Store: w.store, order: &w.order}, stopLocal: func(context.Context) (claimStopLocal, error) {
			return claimStopLocal{Actor: hostbox.Actor{Box: box}}, nil
		}}
		cfg := RelocationAssemblyConfig{Registry: w.reg, Activator: w.act, DataDir: dataDir, Discharge: func(_ context.Context, o hostclaim.Observed) error {
			row, _ := w.reg.Get(host)
			w.heldAt = append(w.heldAt, row != nil && row.HoldOp == "rel-1")
			w.order = append(w.order, "discharge")
			return w.discharge
		}}
		w.hold = a.relocationReleaseHold(cfg)
		return w
	}
	held := func(w *world) bool {
		row, err := w.reg.Get(host)
		return err == nil && row != nil && row.Status == "relocating" && row.HoldOp == "rel-1"
	}
	marked := func(w *world) bool {
		obs, err := hostclaim.Read(ctx, w.store, host)
		if err != nil {
			t.Fatal(err)
		}
		r, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim)
		if err != nil {
			t.Fatal(err)
		}
		return ok && r.Discharged
	}
	t.Run("the discharge first, then the hold, the fence and the mark", func(t *testing.T) {
		w := setup(t, claim(nil))
		if err := w.hold(ctx, host, "rel-1"); err != nil {
			t.Fatal(err)
		}
		if want := []string{"discharge", "lift", "mark"}; !slices.Equal(w.order, want) || !slices.Equal(w.heldAt, []bool{true}) || held(w) || w.act.svc.Fenced(host) || !marked(w) {
			t.Fatalf("order %v held at discharge %v held=%v fenced=%v marked=%v", w.order, w.heldAt, held(w), w.act.svc.Fenced(host), marked(w))
		}
		// A replay of a marked relocation does nothing at all.
		w.order = nil
		if err := w.hold(ctx, host, "rel-1"); err != nil || len(w.order) != 0 {
			t.Fatalf("replay: %v %v", err, w.order)
		}
	})
	t.Run("a refused discharge keeps everything", func(t *testing.T) {
		w := setup(t, claim(nil))
		w.discharge = errors.New("the participation slot is still open")
		if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, w.discharge) {
			t.Fatalf("got %v", err)
		}
		if want := []string{"discharge"}; !slices.Equal(w.order, want) || !held(w) || w.act.svc.FenceGeneration(host) != 12 || marked(w) {
			t.Fatalf("order %v held=%v fence=%d marked=%v", w.order, held(w), w.act.svc.FenceGeneration(host), marked(w))
		}
	})
	t.Run("a row held by another op is refused with nothing done", func(t *testing.T) {
		w := setup(t, claim(nil))
		if ok, err := w.reg.HandHold(host, "relocating", "rel-9", "rel-1"); err != nil || !ok {
			t.Fatalf("hand the hold to rel-9: %v %v", ok, err)
		}
		err := w.hold(ctx, host, "rel-1")
		row, _ := w.reg.Get(host)
		if !errors.Is(err, hostrelocate.ErrRefused) || len(w.order) != 0 || row == nil || row.HoldOp != "rel-9" || w.act.svc.FenceGeneration(host) != 12 || marked(w) {
			t.Fatalf("%v order %v row %+v fence=%d marked=%v", err, w.order, row, w.act.svc.FenceGeneration(host), marked(w))
		}
	})
	t.Run("a record whose stop was on another box is refused with nothing done", func(t *testing.T) {
		other, _ := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-1", Stop: hostclaim.ProbeStop{Box: "src", Session: 12, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("a", 64)}, Stamp: 9, Phase: hostclaim.PhaseRelocatingProbing, Generation: "P1"})
		w := setup(t, claim(func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalCompleted] = string(other) }))
		if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, hostrelocate.ErrRefused) || len(w.order) != 0 || !held(w) || w.act.svc.FenceGeneration(host) != 12 || marked(w) {
			t.Fatalf("%v order %v held=%v fence=%d marked=%v", err, w.order, held(w), w.act.svc.FenceGeneration(host), marked(w))
		}
	})
	t.Run("another operation's fence after a restart stays", func(t *testing.T) {
		w := setup(t, claim(nil))
		w.act.svc = &activator.Service{} // the restart: this box's fence is gone
		w.act.foreignFence(t, host, 12)
		if err := w.hold(ctx, host, "rel-1"); err != nil {
			t.Fatal(err)
		}
		if held(w) || w.act.svc.FenceGeneration(host) != 12 || !marked(w) {
			t.Fatalf("held=%v fence=%d marked=%v", held(w), w.act.svc.FenceGeneration(host), marked(w))
		}
	})
	t.Run("a replay after the hold went discharges, lifts and marks", func(t *testing.T) {
		// The hold was released, then the agent stopped before the lift;
		// a serving wake has since moved the holder on. The discharge runs
		// again, without the holder check, and its answer is the evidence.
		w := setup(t, claim(func(c *hostclaim.Claim) { c.Holder.Session = 13 }))
		if _, err := w.reg.ReleaseHold(host, "rel-1"); err != nil {
			t.Fatal(err)
		}
		if err := w.hold(ctx, host, "rel-1"); err != nil {
			t.Fatal(err)
		}
		if want := []string{"discharge", "lift", "mark"}; !slices.Equal(w.order, want) || !slices.Equal(w.heldAt, []bool{false}) || w.act.svc.Fenced(host) || !marked(w) {
			t.Fatalf("order %v held at discharge %v fenced=%v marked=%v", w.order, w.heldAt, w.act.svc.Fenced(host), marked(w))
		}
	})
	t.Run("a replay after the hold went with a refused discharge keeps the fence and the record", func(t *testing.T) {
		w := setup(t, claim(nil))
		if _, err := w.reg.ReleaseHold(host, "rel-1"); err != nil {
			t.Fatal(err)
		}
		w.discharge = errors.New("the participation slot is still open")
		if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, w.discharge) {
			t.Fatalf("got %v", err)
		}
		if want := []string{"discharge"}; !slices.Equal(w.order, want) || w.act.svc.FenceGeneration(host) != 12 || marked(w) {
			t.Fatalf("order %v fence=%d marked=%v", w.order, w.act.svc.FenceGeneration(host), marked(w))
		}
	})
	t.Run("a holder moved off the probe session while held", func(t *testing.T) {
		w := setup(t, claim(func(c *hostclaim.Claim) { c.Holder.Session, c.Holder.Stamp = 40, 40 }))
		if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, hostrelocate.ErrRefused) || len(w.order) != 0 || !held(w) || marked(w) {
			t.Fatalf("%v order %v held=%v", err, w.order, held(w))
		}
	})
	t.Run("no row is refused", func(t *testing.T) {
		w := setup(t, claim(nil))
		if err := w.reg.Delete(host); err != nil {
			t.Fatal(err)
		}
		if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, hostrelocate.ErrRefused) || len(w.order) != 0 || marked(w) {
			t.Fatalf("%v order %v", err, w.order)
		}
	})
	for name, change := range map[string]func(*hostclaim.Claim){
		"still relocating": func(c *hostclaim.Claim) {
			c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseRelocatingProbing, Exec: box, Participants: []string{"src", box},
				Intent: map[string]string{"region": "ie", "source_box": "src", "target_box": box}}
		},
		"serving on another box": func(c *hostclaim.Claim) {
			c.Op.Exec, c.Op.Participants, c.Holder.WriterBox = "src", []string{"src"}, "src"
		},
	} {
		t.Run(name, func(t *testing.T) {
			w := setup(t, claim(change))
			if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, hostbox.ErrChanged) || len(w.order) != 0 || !held(w) || w.act.svc.FenceGeneration(host) != 12 {
				t.Fatalf("%v order %v held=%v", err, w.order, held(w))
			}
		})
	}
	t.Run("an absent claim", func(t *testing.T) {
		w := setup(t, claim(nil))
		key, _ := hostclaim.Key(host)
		if err := w.store.Delete(ctx, key); err != nil {
			t.Fatal(err)
		}
		if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, hostbox.ErrChanged) || len(w.order) != 0 || !held(w) || w.act.svc.FenceGeneration(host) != 12 {
			t.Fatalf("%v order %v held=%v", err, w.order, held(w))
		}
	})
	// Absence is never evidence (the Global Constraint): the hold release
	// reads its evidence before it releases anything, and without it
	// refuses with nothing done: the row stays held, the probe's fence
	// stands and the record is not marked.
	t.Run("an unreadable completion record", func(t *testing.T) {
		w := setup(t, claim(func(c *hostclaim.Claim) { c.Op.Journal[hostclaim.JournalCompleted] = "{not json" }))
		if err := w.hold(ctx, host, "rel-1"); !errors.Is(err, hostclaim.ErrRelocation) || len(w.order) != 0 || !held(w) || w.act.svc.FenceGeneration(host) != 12 {
			t.Fatalf("%v order %v held=%v fence=%d", err, w.order, held(w), w.act.svc.FenceGeneration(host))
		}
	})
	for name, change := range map[string]func(*testing.T, string){
		"no probe-fence owner": func(t *testing.T, path string) {
			if err := os.Remove(path); err != nil {
				t.Fatal(err)
			}
		},
		"an unreadable probe-fence owner": func(t *testing.T, path string) {
			if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil {
				t.Fatal(err)
			}
		},
		"a probe-fence owner for another session": func(t *testing.T, path string) {
			if err := os.WriteFile(path, []byte(`{"host":"shop.example","session":13,"owner":"`+strings.Repeat("b", 32)+`"}`), 0o600); err != nil {
				t.Fatal(err)
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			w := setup(t, claim(nil))
			change(t, w.marker)
			err := w.hold(ctx, host, "rel-1")
			if !errors.Is(err, hostrelocate.ErrRefused) || len(w.order) != 0 || !held(w) || w.act.svc.FenceGeneration(host) != 12 || marked(w) {
				t.Fatalf("%v order %v held=%v fence=%d marked=%v", err, w.order, held(w), w.act.svc.FenceGeneration(host), marked(w))
			}
			// An unreadable owner is reported as such, never as a missing
			// one: the operator who reads the refusal must find the file.
			if unreadable := strings.Contains(name, "unreadable"); unreadable != strings.Contains(err.Error(), "cannot be read") {
				t.Fatalf("the refusal names the wrong reason: %v", err)
			}
		})
	}
}

// Astra round 6, finding 3, through the target's real hold release and
// the real start CAS: a successor relocation from the box a relocation
// completed on starts only once that box has discharged its probe's
// participation and marked it.
func TestRelocationSuccessorStartsOnlyAfterTheTargetsDischarge(t *testing.T) {
	const host, box = "shop.example", "dst"
	ctx := t.Context()
	created := time.Date(2026, 10, 7, 0, 0, 0, 0, time.UTC)
	inc, err := hostclaim.Incarnation("acct", "app", host, created)
	if err != nil {
		t.Fatal(err)
	}
	done, _ := json.Marshal(hostclaim.CompletedRelocation{Op: "rel-1", Stop: hostclaim.ProbeStop{Box: box, Session: 12, Receipt: "fence-1", ReceiptSHA256: strings.Repeat("a", 64)}, Stamp: 9, Phase: hostclaim.PhaseRelocatingProbing, Generation: "P1"})
	c := &hostclaim.Claim{V: 2, Gen: 31, Host: host,
		Holder: &hostclaim.Holder{Family: "instance", Key: "apps/acct/app/instances/" + host + ".json", Account: "acct", App: "app", Incarnation: inc, CreatedAt: created,
			Location: "instances/" + host, Stamp: 9, Session: 12, Generation: "P1", WriterBox: box, Admissions: []hostclaim.AdmissionRecord{{Session: 12, Box: box, Generation: "P1", Mode: "cold", Region: "ie"}}},
		Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: box, Participants: []string{box}, Journal: map[string]string{hostclaim.JournalCompleted: string(done)}}}
	store := release.NewDirStore(t.TempDir())
	raw, _ := json.Marshal(c)
	key, _ := hostclaim.Key(host)
	if _, err := store.Put(ctx, key, raw, ""); err != nil {
		t.Fatal(err)
	}
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	if err := reg.Upsert(registry.Route{Host: host, Status: "relocating", HoldOp: "rel-1", Socket: "/run/shop.sock"}); err != nil {
		t.Fatal(err)
	}
	var order []string
	refused := errors.New("the participation slot is still open")
	discharge := refused
	a := ClaimAssembly{Claims: store, stopLocal: func(context.Context) (claimStopLocal, error) {
		return claimStopLocal{Actor: hostbox.Actor{Box: box}}, nil
	}}
	// The probe stop's persisted owner: the hold release reads it before
	// it releases anything.
	dataDir := t.TempDir()
	marker, err := relocationProbeFencePath(dataDir, "rel-1")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := relocationProbeFenceOwner(marker, host, 12); err != nil {
		t.Fatal(err)
	}
	hold := a.relocationReleaseHold(RelocationAssemblyConfig{Registry: reg, Activator: newStopProbeActivator(&order), DataDir: dataDir,
		Discharge: func(context.Context, hostclaim.Observed) error { return discharge }})
	start := func() error {
		obs, err := hostclaim.Read(ctx, store, host)
		if err != nil {
			t.Fatal(err)
		}
		_, err = hostclaim.StartRelocation(ctx, store, obs, hostclaim.RelocationStart{OpID: "rel-2", By: "box:wrk", Now: created.Add(3 * time.Hour), Region: "ie", Reason: hostclaim.ReasonDrain,
			SourceBox: box, SourceEnrollment: strings.Repeat("c", 32), TargetBox: "spa", TargetEnrollment: strings.Repeat("d", 32), DirectoryRevision: 2, AdmissionID: "adm-2"}, nil)
		return err
	}
	if err := hold(ctx, host, "rel-1"); !errors.Is(err, refused) {
		t.Fatalf("hold: %v", err)
	}
	if err := start(); !errors.Is(err, hostclaim.ErrPendingChore) {
		t.Fatalf("the successor started before the target's discharge: %v", err)
	}
	discharge = nil
	if err := hold(ctx, host, "rel-1"); err != nil {
		t.Fatal(err)
	}
	if err := start(); err != nil {
		t.Fatalf("the successor after the discharge: %v", err)
	}
}

// markStore records every claim write in order as "mark": the hold
// release writes the claim once, to mark the completion discharged.
type markStore struct {
	release.Store
	order *[]string
}

func (s markStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	*s.order = append(*s.order, "mark")
	return s.Store.Put(ctx, key, body, ifMatch)
}