package main

// Unit + in-process integration tests for the activator mesh (mesh.go): HRW
// determinism and minimal redistribution, membership staleness, the
// owner-vs-proxy routing decision, proxy header/Host semantics, drain
// ordering (heartbeat first), the due-sweep ownership gate, and the mesh-off
// tripwire. Two "boxes" here are two activator+router pairs sharing an
// in-memory heartbeat store — the store IS the bucket, which is the whole
// membership design. The real-S3, real-process version (proxy-to-owner and
// the kill-a-box failover, drain and hard-kill variants) is in
// mesh_live_test.go.

import (
	"context"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"net/http/httputil"
	"net/url"
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"lchat/internal/activator"
)

// --- fakes ---------------------------------------------------------------------

// fakeMeshStore is an in-memory stand-in for the S3 heartbeat prefix. Shared
// between boxes it plays the bucket; rec (optional) feeds a fakeBackend's
// order log so drain tests can pin event ordering.
type fakeMeshStore struct {
	mu          sync.Mutex
	beats       map[string]time.Time
	regions     map[string]string // peer -> advertised region (like the real body's "region")
	regionsCall int               // how many times Regions was invoked — proves the beat gate
	now         func() time.Time
	rec         func(ev string)
}

func newFakeMeshStore(now func() time.Time) *fakeMeshStore {
	return &fakeMeshStore{beats: make(map[string]time.Time), regions: make(map[string]string), now: now}
}

func (s *fakeMeshStore) Put(_ context.Context, peer, region string) error {
	s.mu.Lock()
	s.beats[peer] = s.now()
	if region != "" {
		s.regions[peer] = region // mirror the real store: region omitted when empty
	}
	rec := s.rec
	s.mu.Unlock()
	if rec != nil {
		rec("hb-put")
	}
	return nil
}

func (s *fakeMeshStore) List(context.Context) ([]activator.Heartbeat, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]activator.Heartbeat, 0, len(s.beats))
	for peer, seen := range s.beats {
		out = append(out, activator.Heartbeat{Peer: peer, LastSeen: seen})
	}
	return out, nil
}

func (s *fakeMeshStore) Regions(context.Context) (map[string]string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.regionsCall++
	out := make(map[string]string, len(s.regions))
	for peer, region := range s.regions {
		out[peer] = region
	}
	return out, nil
}

func (s *fakeMeshStore) regionsCalls() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.regionsCall
}

func (s *fakeMeshStore) Delete(_ context.Context, peer string) error {
	s.mu.Lock()
	delete(s.beats, peer)
	delete(s.regions, peer)
	rec := s.rec
	s.mu.Unlock()
	if rec != nil {
		rec("hb-delete")
	}
	return nil
}

func (s *fakeMeshStore) setBeat(peer string, at time.Time) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.beats[peer] = at
}

// setRegion advertises a peer's placement region, as if its heartbeat body
// carried it — the fake counterpart of the real box PUTting with a region.
func (s *fakeMeshStore) setRegion(peer, region string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.regions[peer] = region
}

func (s *fakeMeshStore) hasBeat(peer string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	_, ok := s.beats[peer]
	return ok
}

// hostOwnedBy generates deterministic host names until one's HRW owner is the
// wanted box — tests can't hardcode hosts because box identities (httptest
// ports) vary per run.
func hostOwnedBy(boxes []string, owner string) string {
	for i := 0; ; i++ {
		h := fmt.Sprintf("space-%d.test", i)
		if hrwOwner(h, boxes) == owner {
			return h
		}
	}
}

// meshTestBox is one in-process "box": activator (fake backend) + router +
// front door + a live unix-socket listener standing in for any woken instance
// (it echoes the box's name and records what the backend saw).
type meshTestBox struct {
	name  string
	act   *activatorSvc
	b     *fakeBackend
	reg   *Registry
	rt    *router
	front *httptest.Server
	mesh  *meshSvc
	sock  string

	seenHost  atomic.Value // last Host the backend instance saw
	fwdLeaked atomic.Bool  // the forwarded marker reached the backend
}

func (box *meshTestBox) self() string { return box.front.URL }

func newMeshTestBox(t *testing.T, name string, store meshStore, clk *fakeClock, domains []string) *meshTestBox {
	t.Helper()
	box := &meshTestBox{name: name, b: &fakeBackend{}}

	reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}
	box.reg = reg
	box.act = newActivator(hibernateConfig{Bucket: "test-bucket", Idle: time.Minute}, box.b, reg)
	box.act.now = clk.now

	box.sock = tmpSock(t)
	ln, err := net.Listen("unix", box.sock)
	if err != nil {
		t.Fatalf("listen unix: %v", err)
	}
	srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		box.seenHost.Store(r.Host)
		if r.Header.Get(meshForwardedHeader) != "" {
			box.fwdLeaked.Store(true)
		}
		io.WriteString(w, name)
	})}
	go srv.Serve(ln)
	t.Cleanup(func() { srv.Close() })

	box.rt, err = newRouter(reg, box.act)
	if err != nil {
		t.Fatalf("new router: %v", err)
	}
	box.front = httptest.NewServer(box.rt)
	t.Cleanup(box.front.Close)

	box.mesh = newMesh(box.front.URL, store, domains)
	box.mesh.now = clk.now
	box.mesh.reg = reg // region-scoped placement reads home_region from the registry
	box.act.mesh = box.mesh
	return box
}

// addHost registers a hibernating route for host on this box (pointing at its
// stand-in instance socket) and reloads the router.
func (box *meshTestBox) addHost(t *testing.T, host string) {
	t.Helper()
	if err := box.reg.upsert(Route{Host: host, Kind: "instance", Socket: box.sock, Hibernate: true}); err != nil {
		t.Fatal(err)
	}
	if err := box.rt.reload(); err != nil {
		t.Fatalf("reload: %v", err)
	}
}

// getVia is proxyGet plus arbitrary headers.
func getVia(t *testing.T, base, host string, hdr map[string]string) (string, int) {
	t.Helper()
	req, err := http.NewRequest(http.MethodGet, base+"/", nil)
	if err != nil {
		t.Fatalf("new request: %v", err)
	}
	req.Host = host
	for k, v := range hdr {
		req.Header.Set(k, v)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("request: %v", err)
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return string(b), resp.StatusCode
}

// --- rendezvous hashing ----------------------------------------------------------

// hrwOwner is a pure function: same answer for any ordering of the member
// set, on every box, forever — that's what lets placement need no directory.
func TestHRWOwnerDeterministic(t *testing.T) {
	boxes := []string{"http://a", "http://b", "http://c"}
	perms := [][]string{
		{"http://a", "http://b", "http://c"},
		{"http://c", "http://a", "http://b"},
		{"http://b", "http://c", "http://a"},
	}
	owned := map[string]int{}
	for i := 0; i < 300; i++ {
		h := fmt.Sprintf("host-%d.test", i)
		want := hrwOwner(h, boxes)
		for _, p := range perms {
			if got := hrwOwner(h, p); got != want {
				t.Fatalf("owner of %s depends on member order: %q vs %q", h, got, want)
			}
		}
		owned[want]++
	}
	// The hash must actually spread load — every box owns a healthy share.
	for _, b := range boxes {
		if owned[b] < 50 {
			t.Fatalf("box %s owns only %d/300 hosts — distribution is broken: %v", b, owned[b], owned)
		}
	}
	if hrwOwner("x.test", []string{"http://solo"}) != "http://solo" {
		t.Fatal("single box does not own everything")
	}
	if hrwOwner("x.test", nil) != "" {
		t.Fatal("empty member set should own nothing")
	}
}

// The property that makes membership churn cheap: adding a box moves only the
// ~1/N of hosts it steals (all TO the joiner), removing one moves only the
// leaver's hosts (scattered over survivors) — everyone else stays put, so a
// scale event never stampedes the fleet.
func TestHRWRedistributionMinimal(t *testing.T) {
	four := []string{"http://a", "http://b", "http://c", "http://d"}
	five := append(append([]string{}, four...), "http://e")
	const n = 1000

	before := make(map[string]string, n)
	for i := 0; i < n; i++ {
		h := fmt.Sprintf("host-%d.test", i)
		before[h] = hrwOwner(h, four)
	}

	// Join: every moved host must have moved onto the new box.
	moved := 0
	for h, was := range before {
		now := hrwOwner(h, five)
		if now != was {
			moved++
			if now != "http://e" {
				t.Fatalf("join moved %s from %s to %s — only the joiner may steal", h, was, now)
			}
		}
	}
	if moved < n/10 || moved > n/2 {
		t.Fatalf("join moved %d/%d hosts, want ~1/5 (a fifth box should steal about its share)", moved, n)
	}

	// Leave: only the leaver's hosts move.
	three := []string{"http://a", "http://c", "http://d"}
	for h, was := range before {
		now := hrwOwner(h, three)
		if was == "http://b" {
			if now == "http://b" {
				t.Fatalf("%s still owned by the removed box", h)
			}
		} else if now != was {
			t.Fatalf("leave moved %s from %s to %s — hosts not owned by the leaver must not move", h, was, now)
		}
	}
}

// --- membership --------------------------------------------------------------

// Staleness: a heartbeat older than staleAfter means a dead box — judged at
// read time, so a box that dies between polls drops out on the clock, not on
// the next LIST. Self is always a member (a box counts itself even before its
// first PUT lands), and garbage records never enter the hash.
func TestMeshMembershipStaleness(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	store := newFakeMeshStore(clk.now)
	m := newMesh("http://a", store, nil)
	m.now = clk.now

	// Before any beat: membership of one — the degenerate single box.
	if got := m.liveBoxes(); len(got) != 1 || got[0] != "http://a" {
		t.Fatalf("liveBoxes before any beat = %v, want just self", got)
	}

	store.setBeat("http://b", clk.now())
	store.setBeat("this is not a url", clk.now()) // a foreign/garbage record
	m.beat()
	if !store.hasBeat("http://a") {
		t.Fatal("beat did not PUT our own heartbeat")
	}
	live := m.liveBoxes()
	if len(live) != 2 {
		t.Fatalf("liveBoxes = %v, want self+b (garbage excluded)", live)
	}

	// b goes quiet: past staleAfter it is dead — with NO new list needed.
	clk.advance(m.staleAfter + time.Second)
	if got := m.liveBoxes(); len(got) != 1 || got[0] != "http://a" {
		t.Fatalf("liveBoxes with a stale peer = %v, want just self", got)
	}
}

// The routing decision: proxy only when the HRW owner is another LIVE box;
// own it (wake locally) when the owner is us — or dead, because a stale owner
// is out of the member set and the hash has already reassigned. The lease
// arbitrates if this view is ever wrong; the decision only places.
func TestMeshRoutingDecision(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	store := newFakeMeshStore(clk.now)
	m := newMesh("http://a", store, nil)
	m.now = clk.now
	store.setBeat("http://b", clk.now())
	m.beat()

	boxes := []string{"http://a", "http://b"}
	theirs := hostOwnedBy(boxes, "http://b")
	ours := hostOwnedBy(boxes, "http://a")

	if peer, remote := m.remoteOwner(theirs); !remote || peer != "http://b" {
		t.Fatalf("remoteOwner(%s) = %q, %v; want b, true", theirs, peer, remote)
	}
	if _, remote := m.remoteOwner(ours); remote {
		t.Fatalf("remoteOwner(%s) says proxy a host we own", ours)
	}
	if m.owns(theirs) || !m.owns(ours) {
		t.Fatal("owns() disagrees with remoteOwner()")
	}

	// The owner dies: its hosts fall through to a local wake.
	clk.advance(m.staleAfter + time.Second)
	if _, remote := m.remoteOwner(theirs); remote {
		t.Fatal("remoteOwner still proxies to a stale (dead) box — its hosts must wake locally")
	}
	if !m.owns(theirs) {
		t.Fatal("survivor does not own the dead box's host")
	}
}

// --- domain scoping ------------------------------------------------------------

// inScope's matching rule: exact domain or a proper subdomain of one, case-
// insensitive (hosts and configured domains alike); nothing else.
func TestMeshInScope(t *testing.T) {
	m := newMesh("http://a", newFakeMeshStore(time.Now), []string{"11ch.at", "11relay.com"})
	cases := []struct {
		host string
		want bool
	}{
		{"11ch.at", true},     // the bare domain itself
		{"abc.11ch.at", true}, // a subdomain
		{"ABC.11CH.AT", true}, // case-insensitive
		{"11relay.com", true},
		{"abc.11relay.com", true},
		{"family.luan.chat", false}, // a different domain entirely
		{"evil11ch.at", false},      // must not match on a bare suffix
		{"11ch.at.evil.com", false}, // must not match as a prefix
	}
	for _, c := range cases {
		if got := m.inScope(c.host); got != c.want {
			t.Errorf("inScope(%q) = %v, want %v", c.host, got, c.want)
		}
	}
}

// An out-of-scope host always resolves locally, regardless of what HRW would
// otherwise say — it's never proxied away, and this box always reports owning
// it (the sweep's due-wake gate must never skip it). Domain scoping exists
// because a registry row for an out-of-scope host is only ever replicated to
// the box(es) it already lives on (control.go's registerInstance): a host in
// scope legitimately has rows on every mesh member and needs real HRW
// placement; a host out of scope does not, and hashing it elsewhere would
// just 404 there.
func TestMeshDomainScopingBypassesHRW(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	store := newFakeMeshStore(clk.now)
	m := newMesh("http://a", store, []string{"11ch.at"})
	m.now = clk.now
	store.setBeat("http://b", clk.now())
	m.beat()

	boxes := []string{"http://a", "http://b"}
	// A host under the in-scope domain that genuinely hashes to the peer.
	// hostOwnedBy's "space-N.test" hosts fall under no configured domain, so
	// build one under 11ch.at directly.
	var inScopeTheirs string
	for i := 0; ; i++ {
		h := fmt.Sprintf("space-%d.11ch.at", i)
		if hrwOwner(h, boxes) == "http://b" {
			inScopeTheirs = h
			break
		}
	}
	if peer, remote := m.remoteOwner(inScopeTheirs); !remote || peer != "http://b" {
		t.Fatalf("in-scope host: remoteOwner = %q, %v; want b, true", peer, remote)
	}

	// The same host name, but under a domain the mesh isn't scoped to. Find
	// one that HRW would (incorrectly, if honored) assign to the peer.
	var outOfScope string
	for i := 0; ; i++ {
		h := fmt.Sprintf("family-%d.luan.chat", i)
		if hrwOwner(h, boxes) == "http://b" {
			outOfScope = h
			break
		}
	}
	if _, remote := m.remoteOwner(outOfScope); remote {
		t.Fatalf("out-of-scope host %q was proxied away — domain scoping did not apply", outOfScope)
	}
	if !m.owns(outOfScope) {
		t.Fatalf("out-of-scope host %q: owns() = false — the due-wake gate would wrongly skip it", outOfScope)
	}
}

// --- region placement (docs/multi-edge.md, Phase 1) -----------------------------

// candidatesFor scopes HRW to the live boxes whose region matches a space's
// home_region, with an all-boxes fallback that keeps pre-region behaviour
// byte-identical. One setup, four rules the task pins:
//   (a) a space pinned to a region routes to that region's box;
//   (b) a space with empty home_region falls back to all-boxes HRW (== today);
//   (c) a home_region no live box advertises falls back to all-boxes;
//   (d) a peer that advertises no region is treated as agnostic (only ever a
//       fallback candidate, never a region match).
func TestMeshRegionPlacement(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	store := newFakeMeshStore(clk.now)
	reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatalf("open registry: %v", err)
	}

	// Three live boxes: self in "ie", a peer in "fi", and an agnostic peer that
	// advertises no region at all (rule d).
	m := newMesh("http://a", store, []string{"11ch.at"})
	m.now = clk.now
	m.reg = reg
	m.region = "ie"
	store.setBeat("http://b", clk.now())
	store.setRegion("http://b", "fi")
	store.setBeat("http://c", clk.now()) // agnostic peer — no setRegion
	m.beat()                             // our PUT records self's region; List+Regions load the view

	all := []string{"http://a", "http://b", "http://c"}
	if got := m.liveBoxes(); !sameSet(got, all) {
		t.Fatalf("liveBoxes = %v, want %v", got, all)
	}

	// A helper to register an in-scope space with a given home_region.
	pin := func(host, region string) {
		if err := reg.upsert(Route{Host: host, Kind: "instance", Socket: "/x.sock", Hibernate: true, HomeRegion: region}); err != nil {
			t.Fatalf("upsert %s: %v", host, err)
		}
	}

	// (a) Pinned to "fi" → only the fi box is a candidate, so it owns and every
	// other box proxies to it, regardless of what bare HRW over all three says.
	pin("fi-space.11ch.at", "fi")
	if got := m.candidatesFor("fi-space.11ch.at"); !sameSet(got, []string{"http://b"}) {
		t.Fatalf("home_region fi: candidates = %v, want [http://b]", got)
	}
	if m.owns("fi-space.11ch.at") {
		t.Fatal("box a wrongly owns an fi-pinned space")
	}
	if peer, remote := m.remoteOwner("fi-space.11ch.at"); !remote || peer != "http://b" {
		t.Fatalf("remoteOwner(fi-space) = %q,%v; want http://b,true", peer, remote)
	}

	// Self-region match: a space pinned to our own "ie" is owned locally.
	pin("ie-space.11ch.at", "ie")
	if got := m.candidatesFor("ie-space.11ch.at"); !sameSet(got, []string{"http://a"}) {
		t.Fatalf("home_region ie: candidates = %v, want [http://a]", got)
	}
	if !m.owns("ie-space.11ch.at") {
		t.Fatal("box a does not own an ie-pinned space it is the region for")
	}

	// (b) Empty home_region (a host with no row at all is empty too) → all boxes,
	// and the chosen owner is identical to bare HRW over the whole live set.
	noRegion := "plain.11ch.at" // no reg row → homeRegionOf ""
	if got := m.candidatesFor(noRegion); !sameSet(got, all) {
		t.Fatalf("empty home_region: candidates = %v, want all %v", got, all)
	}
	if hrwOwner(noRegion, m.candidatesFor(noRegion)) != hrwOwner(noRegion, m.liveBoxes()) {
		t.Fatal("empty home_region must place identically to pre-region HRW")
	}

	// (c) A home_region no live box advertises → fall back to all boxes.
	pin("de-space.11ch.at", "de")
	if got := m.candidatesFor("de-space.11ch.at"); !sameSet(got, all) {
		t.Fatalf("unmatched home_region de: candidates = %v, want all %v", got, all)
	}

	// (d) The agnostic peer "http://c" is never a region match (it advertised
	// none), yet is present in every all-boxes fallback above — already asserted
	// by the sameSet(all) checks that include it.
}

// TestMeshRegionAgnosticUnchanged pins the core back-compat promise directly: a
// mesh with no region set and no registry (exactly how the existing unit tests
// and today's fleet run) has candidatesFor == liveBoxes for every host, so
// owns/remoteOwner cannot diverge from pre-region behaviour.
func TestMeshRegionAgnosticUnchanged(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	store := newFakeMeshStore(clk.now)
	m := newMesh("http://a", store, []string{"11ch.at"})
	m.now = clk.now // no region, no reg — the pre-region shape
	store.setBeat("http://b", clk.now())
	m.beat()

	for i := 0; i < 50; i++ {
		host := fmt.Sprintf("space-%d.11ch.at", i)
		if !sameSet(m.candidatesFor(host), m.liveBoxes()) {
			t.Fatalf("region-agnostic mesh scoped %s away from the full live set", host)
		}
	}
}

// TestMeshBeatRegionFetchGate proves the zero-cost-when-inert promise: a box
// with no region declared (prod today) fetches NO peer regions on a beat, so
// region placement adds zero S3 requests over pre-region behaviour; a box that
// declares a region does fetch them.
func TestMeshBeatRegionFetchGate(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}

	// Agnostic box: beats never touch Regions.
	store := newFakeMeshStore(clk.now)
	agnostic := newMesh("http://a", store, []string{"11ch.at"})
	agnostic.now = clk.now // no region
	agnostic.beat()
	agnostic.beat()
	if got := store.regionsCalls(); got != 0 {
		t.Fatalf("agnostic box made %d Regions call(s) across two beats, want 0 — an inert feature must cost nothing", got)
	}

	// Region-scoped box: each beat fetches peer regions.
	store2 := newFakeMeshStore(clk.now)
	scoped := newMesh("http://b", store2, []string{"11ch.at"})
	scoped.now = clk.now
	scoped.region = "ie"
	scoped.beat()
	scoped.beat()
	if got := store2.regionsCalls(); got != 2 {
		t.Fatalf("region-scoped box made %d Regions call(s) across two beats, want 2", got)
	}
}

// sameSet compares two peer slices ignoring order (liveBoxes/candidatesFor draw
// from a map, so order is not defined).
func sameSet(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	seen := make(map[string]int, len(a))
	for _, x := range a {
		seen[x]++
	}
	for _, y := range b {
		seen[y]--
	}
	for _, n := range seen {
		if n != 0 {
			return false
		}
	}
	return true
}

// meshFromEnv reads LCHAT_MESH_REGION (optional): set = this box's placement
// region, unset = region-agnostic (the empty string, today's behaviour).
func TestMeshFromEnvRegion(t *testing.T) {
	t.Setenv("LCHAT_MESH_SELF", "https://box-a.internal")
	t.Setenv("LCHAT_MESH_DOMAINS", "11ch.at")
	t.Setenv("LCHAT_MESH_TOKEN", "shared-secret")

	t.Setenv("LCHAT_MESH_REGION", "")
	m, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"})
	if err != nil || m == nil {
		t.Fatalf("unset region: %v, %v", m, err)
	}
	if m.region != "" {
		t.Fatalf("region = %q, want \"\" (agnostic) when LCHAT_MESH_REGION is unset", m.region)
	}

	t.Setenv("LCHAT_MESH_REGION", " ie ")
	m, err = meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"})
	if err != nil || m == nil {
		t.Fatalf("set region: %v, %v", m, err)
	}
	if m.region != "ie" {
		t.Fatalf("region = %q, want the trimmed \"ie\"", m.region)
	}
}

// meshFromEnv refuses to start the mesh without an explicit domain allowlist —
// failing loudly beats silently letting HRW assign a host to a box that has
// no registry row for it (see the inScope doc comment).
func TestMeshFromEnvRequiresDomains(t *testing.T) {
	t.Setenv("LCHAT_MESH_SELF", "https://box-a.internal")
	t.Setenv("LCHAT_MESH_TOKEN", "shared-secret")
	t.Setenv("LCHAT_MESH_DOMAINS", "")
	if _, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"}); err == nil {
		t.Fatal("mesh without LCHAT_MESH_DOMAINS should be a config error")
	}

	t.Setenv("LCHAT_MESH_DOMAINS", "11ch.at, 11relay.com")
	m, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"})
	if err != nil || m == nil {
		t.Fatalf("valid domains: %v, %v", m, err)
	}
	if len(m.domains) != 2 || m.domains[0] != "11ch.at" || m.domains[1] != "11relay.com" {
		t.Fatalf("domains = %v, want the parsed, trimmed, lowercased list", m.domains)
	}
}

// meshFromEnv refuses to start the mesh without a shared route-replication
// secret — an unauthenticated /api/mesh/route would let any caller hand this
// box arbitrary routes (see meshreplicate.go).
func TestMeshFromEnvRequiresToken(t *testing.T) {
	t.Setenv("LCHAT_MESH_SELF", "https://box-a.internal")
	t.Setenv("LCHAT_MESH_DOMAINS", "11ch.at")
	t.Setenv("LCHAT_MESH_TOKEN", "")
	if _, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"}); err == nil {
		t.Fatal("mesh without LCHAT_MESH_TOKEN should be a config error")
	}

	t.Setenv("LCHAT_MESH_TOKEN", "shared-secret")
	m, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"})
	if err != nil || m == nil {
		t.Fatalf("valid token: %v, %v", m, err)
	}
	if m.token != "shared-secret" {
		t.Fatalf("token = %q, want it read through", m.token)
	}
}

// --- the proxy hop -------------------------------------------------------------

// Scheme/Host/SNI semantics of the peer hop: the instance hostname survives
// (the peer routes by Host); for https peers the outgoing URL keeps the
// instance host so SNI and cert verification run against a name the peer can
// actually prove (its wildcard/per-host cert), while the dialer pins TCP to
// the peer's address; for http peers the URL targets the peer directly. The
// forwarded marker rides every hop. InsecureSkipVerify appears nowhere.
func TestPeerProxyRewrite(t *testing.T) {
	rewrite := func(p *httputil.ReverseProxy, host string) *http.Request {
		in := httptest.NewRequest("GET", "http://placeholder/api/x", nil)
		in.Host = host
		in.RemoteAddr = "192.0.2.7:4711"
		out := in.Clone(context.Background())
		p.Rewrite(&httputil.ProxyRequest{In: in, Out: out})
		return out
	}

	httpsPeer, _ := url.Parse("https://box-b.internal")
	out := rewrite(newPeerProxy("http://self", httpsPeer), "space.11ch.at")
	if out.URL.Scheme != "https" || out.URL.Host != "space.11ch.at" || out.Host != "space.11ch.at" {
		t.Fatalf("https hop: scheme=%s urlHost=%s host=%s — SNI/verification must target the instance host", out.URL.Scheme, out.URL.Host, out.Host)
	}
	if got := peerDialAddr(httpsPeer); got != "box-b.internal:443" {
		t.Fatalf("https dial addr = %q, want the peer's address", got)
	}
	if out.Header.Get(meshForwardedHeader) != "http://self" {
		t.Fatal("forwarded marker missing — the one-hop loop guard would not hold")
	}

	httpPeer, _ := url.Parse("http://10.0.0.5:9000")
	out = rewrite(newPeerProxy("http://self", httpPeer), "space.11ch.at")
	if out.URL.Scheme != "http" || out.URL.Host != "10.0.0.5:9000" || out.Host != "space.11ch.at" {
		t.Fatalf("http hop: scheme=%s urlHost=%s host=%s — dial the peer, Host stays the instance", out.URL.Scheme, out.URL.Host, out.Host)
	}
	if got := peerDialAddr(httpPeer); got != "10.0.0.5:9000" {
		t.Fatalf("http dial addr = %q", got)
	}
	bare, _ := url.Parse("http://box-c.internal")
	if got := peerDialAddr(bare); got != "box-c.internal:80" {
		t.Fatalf("default http port: %q", got)
	}
}

// Two boxes, one shared "bucket": a request landing on the non-owner is
// reverse-proxied to the owner (which wakes the instance), the instance
// hostname survives the hop, the forwarded marker never reaches a backend,
// a request already carrying the marker is served where it stands (one hop
// max), and a host with a live local entry stays local even when the hash
// points elsewhere.
func TestMeshProxyToOwner(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	store := newFakeMeshStore(clk.now)
	box1 := newMeshTestBox(t, "box1", store, clk, nil)
	box2 := newMeshTestBox(t, "box2", store, clk, nil)
	// Beat both (twice for box1, whose first beat predates box2's PUT) so
	// each box sees the full membership.
	box1.mesh.beat()
	box2.mesh.beat()
	box1.mesh.beat()

	boxes := []string{box1.self(), box2.self()}
	host := hostOwnedBy(boxes, box2.self())
	box1.addHost(t, host)
	box2.addHost(t, host)

	// (a) Request to the NON-owner: proxied to the owner and served there.
	if body, code := proxyGet(t, box1.front.URL, host); code != 200 || body != "box2" {
		t.Fatalf("non-owner request: got %d %q, want 200 box2 (proxied to owner)", code, body)
	}
	if acq, _, _ := box2.b.counts(); acq != 1 {
		t.Fatalf("owner acquires = %d, want 1 (the wake happened on the owner)", acq)
	}
	if acq, _, starts := box1.b.counts(); acq != 0 || starts != 0 {
		t.Fatalf("non-owner touched its own activator (acquires=%d starts=%d)", acq, starts)
	}
	if got, _ := box2.seenHost.Load().(string); got != host {
		t.Fatalf("owner backend saw Host %q, want %q — the hostname must survive the hop", got, host)
	}
	if box2.fwdLeaked.Load() {
		t.Fatal("the forwarded marker leaked through to the backend")
	}

	// (b) Request straight to the owner: local, no extra wake.
	if body, code := proxyGet(t, box2.front.URL, host); code != 200 || body != "box2" {
		t.Fatalf("owner request: got %d %q", code, body)
	}
	if acq, _, _ := box2.b.counts(); acq != 1 {
		t.Fatalf("owner re-acquired for a host it already has awake (acquires=%d)", acq)
	}

	// (c) One hop max: a request already carrying the forwarded marker is
	// served where it stands — here the non-owner, which wakes it locally
	// (the lease would arbitrate a real conflict; the fake just observes).
	var host2 string // a second box2-owned host, distinct from the first
	for i := 0; ; i++ {
		if h := fmt.Sprintf("space-b-%d.test", i); hrwOwner(h, boxes) == box2.self() {
			host2 = h
			break
		}
	}
	box1.addHost(t, host2)
	box2.addHost(t, host2)
	if body, code := getVia(t, box1.front.URL, host2, map[string]string{meshForwardedHeader: "http://elsewhere"}); code != 200 || body != "box1" {
		t.Fatalf("forwarded request: got %d %q, want 200 box1 (served in place, never re-proxied)", code, body)
	}
	if acq, _, _ := box1.b.counts(); acq != 1 {
		t.Fatalf("non-owner acquires = %d after forwarded request, want 1", acq)
	}

	// (d) Live-local-entry rule: host2 is now awake on box1, so an ordinary
	// request to box1 stays local despite the hash pointing at box2 — an
	// awake instance finishes its life where it is.
	if body, code := proxyGet(t, box1.front.URL, host2); code != 200 || body != "box1" {
		t.Fatalf("live-entry request: got %d %q, want 200 box1 (local entry wins over the hint)", code, body)
	}
	if acq, _, _ := box2.b.counts(); acq != 1 {
		t.Fatalf("owner was woken for a host living on its peer (acquires=%d)", acq)
	}
}

// --- mesh off: the tripwire ------------------------------------------------------

// With the mesh off (LCHAT_MESH_SELF unset → act.mesh nil) the activator is
// byte-identical to the single-box shape: a host whose HRW owner WOULD be a
// peer is woken locally, nothing is proxied, and no heartbeat is ever
// written. The same box with only the mesh flipped on proxies the same-shaped
// request — proving the nil check is the single divergence point.
func TestMeshOffTripwire(t *testing.T) {
	clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
	store := newFakeMeshStore(clk.now)
	box1 := newMeshTestBox(t, "box1", store, clk, nil)
	box2 := newMeshTestBox(t, "box2", store, clk, nil)
	box1.act.mesh = nil // the tripwire: box1 is mesh-off
	box2.mesh.beat()    // box2 participates; its heartbeat is in the store

	boxes := []string{box1.self(), box2.self()}
	host := hostOwnedBy(boxes, box2.self())
	box1.addHost(t, host)
	box2.addHost(t, host)

	if body, code := proxyGet(t, box1.front.URL, host); code != 200 || body != "box1" {
		t.Fatalf("mesh-off request: got %d %q, want 200 box1 (woken locally, never proxied)", code, body)
	}
	if acq, _, _ := box1.b.counts(); acq != 1 {
		t.Fatalf("mesh-off box acquires = %d, want 1", acq)
	}
	if acq, _, _ := box2.b.counts(); acq != 0 {
		t.Fatalf("peer was touched by a mesh-off box (acquires=%d)", acq)
	}
	if store.hasBeat(box1.self()) {
		t.Fatal("mesh-off box wrote a heartbeat")
	}

	// Contrast: flip the mesh on — the ONLY change — and the same-shaped
	// request (a fresh host, since the first now has a live local entry)
	// proxies to the owner.
	box1.act.mesh = box1.mesh
	box1.mesh.beat()
	var host2 string
	for i := 0; ; i++ {
		h := fmt.Sprintf("contrast-%d.test", i)
		if hrwOwner(h, boxes) == box2.self() {
			host2 = h
			break
		}
	}
	box1.addHost(t, host2)
	box2.addHost(t, host2)
	if body, code := proxyGet(t, box1.front.URL, host2); code != 200 || body != "box2" {
		t.Fatalf("mesh-on contrast: got %d %q, want 200 box2 (proxied)", code, body)
	}
}

// meshFromEnv: unset = off (nil, nil — the zero-change default); set without
// the hibernation bucket = a loud config error; malformed URLs rejected.
func TestMeshFromEnv(t *testing.T) {
	// A valid domain allowlist and token throughout —
	// TestMeshFromEnvRequiresDomains/TestMeshFromEnvRequiresToken cover their
	// missing cases on their own.
	t.Setenv("LCHAT_MESH_DOMAINS", "11ch.at")
	t.Setenv("LCHAT_MESH_TOKEN", "shared-secret")

	t.Setenv("LCHAT_MESH_SELF", "")
	if m, err := meshFromEnv(&hibernateConfig{Bucket: "b"}); err != nil || m != nil {
		t.Fatalf("unset: got %v, %v; want nil, nil (mesh off)", m, err)
	}

	t.Setenv("LCHAT_MESH_SELF", "https://box-a.internal")
	if _, err := meshFromEnv(nil); err == nil {
		t.Fatal("mesh without a hibernation bucket should be a config error")
	}

	for _, bad := range []string{"box-a.internal", "ftp://box-a", "https://box-a/path"} {
		t.Setenv("LCHAT_MESH_SELF", bad)
		if _, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"}); err == nil {
			t.Fatalf("LCHAT_MESH_SELF %q should be rejected", bad)
		}
	}

	t.Setenv("LCHAT_MESH_SELF", "https://box-a.internal/")
	m, err := meshFromEnv(&hibernateConfig{Bucket: "b", Region: "eu-west-1"})
	if err != nil || m == nil {
		t.Fatalf("valid self: %v, %v", m, err)
	}
	if m.self != "https://box-a.internal" {
		t.Fatalf("self = %q, want the trailing slash trimmed", m.self)
	}
}

// --- drain ordering ---------------------------------------------------------------

// The drain contract, order-pinned: the heartbeat is deleted FIRST — before
// the next-due probe, before the child stops, before the final checkpoint and
// lease release — so peers take over the moment the leases free up. Same code
// path for deploys, box failure drills, and scale-down.
func TestMeshDrainDeletesHeartbeatFirst(t *testing.T) {
	a, b, clk, _ := testActivator(t, time.Hour)
	store := newFakeMeshStore(clk.now)
	store.rec = b.record // interleave heartbeat ops with the backend's order log
	m := newMesh("http://box-a", store, nil)
	m.now = clk.now
	m.interval = time.Hour // the loop must never beat on its own mid-test
	a.mesh = m
	m.start()

	release, err := a.beginRequest(context.Background(), testRoute("drain.example.com"))
	if err != nil {
		t.Fatalf("beginRequest: %v", err)
	}
	release()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	a.shutdown(ctx)

	if store.hasBeat("http://box-a") {
		t.Fatal("drain left our heartbeat behind")
	}
	ev := b.events()
	del := indexOf(ev, "hb-delete")
	if del == -1 {
		t.Fatalf("no heartbeat delete in %v", ev)
	}
	if put := indexOf(ev, "hb-put"); put == -1 || put > del {
		t.Fatalf("heartbeat lifecycle out of order: %v", ev)
	}
	for _, later := range []string{"nextDue", "stop:instance"} {
		if i := indexOf(ev, later); i == -1 || i < del {
			t.Fatalf("drain order wrong: %q at %d, hb-delete at %d — the heartbeat must go first (%v)", later, i, del, ev)
		}
	}
}

// --- due-time sweeps under the mesh -----------------------------------------------

// Only the HRW owner due-wakes a host; a non-owner skips it every tick. When
// the owner dies, staleness reassigns ownership and the next sweep fires the
// wake here — failover for schedules is the same eventual path as requests.
func TestMeshDueSweepOnlyOwner(t *testing.T) {
	a, b, clk, reg := testActivator(t, 10*time.Second)
	store := newFakeMeshStore(clk.now)
	m := newMesh("http://box-a", store, nil)
	m.now = clk.now
	a.mesh = m
	store.setBeat("http://box-b", clk.now())
	m.beat()

	host := hostOwnedBy([]string{"http://box-a", "http://box-b"}, "http://box-b")
	route := testRoute(host)
	if err := reg.upsert(route); err != nil {
		t.Fatal(err)
	}
	if err := reg.setNextDue(host, clk.now().Add(-time.Second).Unix()); err != nil {
		t.Fatal(err)
	}

	// The peer owns it: our sweeps must not wake it, however overdue.
	for i := 0; i < 3; i++ {
		a.sweepOnce()
	}
	if _, _, starts := b.counts(); starts != 0 {
		t.Fatalf("starts=%d — a non-owner due-woke a host", starts)
	}

	// The owner dies (heartbeat goes stale): ownership reassigns to us on the
	// next read, and the sweep fires the wake.
	clk.advance(m.staleAfter + time.Second)
	a.sweepOnce()
	waitFor(t, "post-failover due wake", func() bool { return a.entryAwake(host) })
	if _, _, starts := b.counts(); starts != 1 {
		t.Fatalf("starts=%d after owner death, want 1", starts)
	}
}