package edge

import (
	"bufio"
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/edgeproof"
	"github.com/carlosframework/platform/internal/lbhealth"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// putJSON writes a control-plane object body verbatim, replacing whatever is
// there. release.Store.Put with an empty ifMatch is create-only, so a test
// that rewrites a key (the last-good and stale-file cases below) has to carry
// the etag — the same putReplacing shape certsync.go uses in production.
func putJSON(t *testing.T, s release.Store, key, body string) {
	t.Helper()
	_, etag, err := s.Get(context.Background(), key)
	switch {
	case errors.Is(err, os.ErrNotExist):
		etag = ""
	case err != nil:
		t.Fatalf("get %s: %v", key, err)
	}
	if _, err := s.Put(context.Background(), key, []byte(body), etag); err != nil {
		t.Fatalf("put %s: %v", key, err)
	}
}

func refreshHomeMap(t *testing.T, m *homeMap) {
	t.Helper()
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatalf("refresh: %v", err)
	}
}

func TestHomeMapFoldsAndExcludesOwnBox(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"a.test":{"role":"home"}}}`)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	if eip, box, ok := m.Home("b.test"); !ok || eip != "2.2.2.2" || box != "betty" {
		t.Fatalf("b.test home = %s/%s/%v", eip, box, ok)
	}
	if _, _, ok := m.Home("a.test"); ok {
		t.Fatal("own-box hosts must not offer a backhaul target")
	}
}

// Task 1's writer publishes registry row hosts VERBATIM, catch-all rows
// included, so the reader owes lookupUpstream's one-label wildcard rule: an
// exact match wins, then "*.<parent>" — and only ONE label, so a two-label
// host misses exactly as it does locally.
func TestHomeMapAppliesOneLabelWildcardRule(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json",
		`{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"*.apps.test":{"role":"home"},"exact.apps.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)

	// Exact form.
	if eip, _, ok := m.Home("exact.apps.test"); !ok || eip != "2.2.2.2" {
		t.Fatalf("exact host = %s/%v, want the home edge", eip, ok)
	}
	// Wildcard form: one label under the catch-all parent.
	if eip, box, ok := m.Home("paul.apps.test"); !ok || eip != "2.2.2.2" || box != "betty" {
		t.Fatalf("one-label wildcard host = %s/%s/%v", eip, box, ok)
	}
	// Two labels: a miss here exactly as in lookupUpstream, which composes
	// "*.b.apps.test" and finds nothing.
	if _, _, ok := m.Home("a.b.apps.test"); ok {
		t.Fatal("a two-label host must not match a one-label wildcard")
	}
	// The wildcard key itself is not a request host and must not resolve as
	// one via the fallback ("*.apps.test" composes "*.apps.test" again).
	if _, _, ok := m.Home("apps.test"); ok {
		t.Fatal("the catch-all parent must not match its own wildcard")
	}
}

// The role object is read-local anticipation (Task 1). Only "home" routes;
// anything else is a role this build does not implement, and an unknown role
// must neither route nor fail the poll.
func TestHomeMapIgnoresUnknownRoles(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json",
		`{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"read.test":{"role":"read"},"home.test":{"role":"home"},"blank.test":{}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m) // an unknown role is not an error
	if _, _, ok := m.Home("read.test"); ok {
		t.Fatal("role:\"read\" is not a home — it must not offer a backhaul target")
	}
	if _, _, ok := m.Home("blank.test"); ok {
		t.Fatal("an absent role is not a home")
	}
	if _, _, ok := m.Home("home.test"); !ok {
		t.Fatal("role:\"home\" must still route alongside roles this build ignores")
	}
}

// A box with no discoverable public IP publishes its hosts list with eip:""
// (Task 1's DR-runbook case). There is nothing to dial, so it must never be
// offered as a target — a "" EIP would otherwise become a dial to port 443 of
// nothing.
func TestHomeMapSkipsBoxesWithNoEIP(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	if _, _, ok := m.Home("b.test"); ok {
		t.Fatal("a box with no EIP has no dial address and must not be a backhaul target")
	}
}

// Two boxes claiming the same host is a placement bug, but every edge must
// still pick the SAME one — a coin flip per box would split traffic across a
// move instead of converging on it.
func TestHomeMapPicksTheSameHomeOnADuplicateClaim(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	putJSON(t, s, "control/edges/carol.json", `{"v":1,"box":"carol","eip":"3.3.3.3","hosts":{"b.test":{"role":"home"}}}`)
	for i := 0; i < 5; i++ {
		m := newHomeMap(s, "annie")
		refreshHomeMap(t, m)
		if _, box, _ := m.Home("b.test"); box != "betty" {
			t.Fatalf("duplicate claim resolved to %q, want the deterministic first (betty)", box)
		}
	}
}

// A box that is retired (its file deleted) must LEAVE the map — the fold is a
// full rebuild from each list, never a merge into the previous one.
func TestHomeMapForgetsADeletedEdgesFile(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	if _, _, ok := m.Home("b.test"); !ok {
		t.Fatal("precondition: b.test should be homed on betty")
	}
	if err := s.Delete(context.Background(), "control/edges/betty.json"); err != nil {
		t.Fatalf("delete: %v", err)
	}
	refreshHomeMap(t, m)
	if _, _, ok := m.Home("b.test"); ok {
		t.Fatal("a retired box's hosts must leave the map on the next poll")
	}
}

// One unreadable object is not a bad poll: the previous copy of that box's
// file carries forward and every other box's file still lands, exactly as the
// cert mirror does it.
func TestHomeMapCarriesForwardOneUnreadableFile(t *testing.T) {
	inner := meshTestStore(t)
	s := &oneKeyFailingStore{Store: inner}
	putJSON(t, inner, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)

	// betty's file goes unreadable; carol appears in the same poll.
	putJSON(t, inner, "control/edges/carol.json", `{"v":1,"box":"carol","eip":"3.3.3.3","hosts":{"c.test":{"role":"home"}}}`)
	s.failKey("control/edges/betty.json")
	refreshHomeMap(t, m)

	if eip, _, ok := m.Home("b.test"); !ok || eip != "2.2.2.2" {
		t.Fatalf("b.test = %s/%v — the previous copy of an unreadable file must carry forward", eip, ok)
	}
	if _, _, ok := m.Home("c.test"); !ok {
		t.Fatal("one bad object must not discard the rest of the rebuild")
	}
}

// A corrupt body is the same class of fault as an unreadable one: carry the
// last good parse forward rather than dropping the box off the map.
func TestHomeMapCarriesForwardACorruptFile(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty",`)
	refreshHomeMap(t, m)
	if eip, _, ok := m.Home("b.test"); !ok || eip != "2.2.2.2" {
		t.Fatalf("b.test = %s/%v — a corrupt file must leave the last good parse standing", eip, ok)
	}
}

// A failed List is a bucket fault: last-good stands and the error is returned
// for the loop to log. An empty map here would silently stop every backhaul
// in the deployment on one blip.
func TestHomeMapKeepsLastGoodWhenTheBucketIsDown(t *testing.T) {
	inner := meshTestStore(t)
	s := &listFailingStore{Store: inner}
	putJSON(t, inner, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)

	s.setFail(true)
	if err := m.Refresh(context.Background()); err == nil {
		t.Fatal("a failed List must be reported, not swallowed")
	}
	if _, _, ok := m.Home("b.test"); !ok {
		t.Fatal("last-good must stand through a bucket blip")
	}
}

// An edge with no box identity of its own cannot exclude itself, so it must
// never offer a target at all — fail closed rather than backhaul to itself.
func TestHomeMapWithoutASelfLabelNeverOffersATarget(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "")
	refreshHomeMap(t, m)
	if _, _, ok := m.Home("b.test"); ok {
		t.Fatal("an edge that does not know its own box label must not backhaul")
	}
}

// A nil map (no deployment store, so no mesh at all) answers like an empty
// one rather than panicking on the request path.
func TestNilHomeMapIsInert(t *testing.T) {
	var m *homeMap
	if _, _, ok := m.Home("b.test"); ok {
		t.Fatal("a nil home map must offer nothing")
	}
	if m.KnownHost("b.test") {
		t.Fatal("a nil home map knows no host")
	}
	if cur, next := m.meshKeys(); cur != nil || next != nil {
		t.Fatal("a nil home map holds no mesh keys")
	}
}

// KnownHost is the membership question the three cert gates ask through
// Edge.deploymentHost (#277), and it is EXACT-only where Home is not. Home's
// one-label fallback is what makes an invented subdomain under a remote
// catch-all DIALABLE; letting the same fallback into a cert gate would let
// that invented label drive an ACME order and spend a read-through budget,
// which is precisely what PR #180's Allowed rule refuses. The two answers
// diverging on the same map, in the same test, is the pin.
func TestKnownHostExactOnly(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/annie.json",
		`{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"own.test":{"role":"home"}}}`)
	putJSON(t, s, "control/edges/betty.json",
		`{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"*.apps.test":{"role":"home"},"exact.apps.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)

	if !m.KnownHost("exact.apps.test") {
		t.Fatal("an exactly-named remote-homed host is the whole point of the predicate")
	}
	// Precondition: the fallback Home applies is live on this very map — so
	// the refusal below is KnownHost's own, not an empty map's.
	if _, _, ok := m.Home("invented.apps.test"); !ok {
		t.Fatal("test bug: the one-label wildcard rule did not apply")
	}
	if m.KnownHost("invented.apps.test") {
		t.Fatal("a label under a mirrored catch-all must never be admitted — that is #180's rule")
	}
	if m.KnownHost("*.apps.test") {
		t.Fatal("the catch-all key itself is not a certifiable name")
	}
	if m.KnownHost("stranger.example") {
		t.Fatal("a host no box in the deployment homes is a stranger")
	}
	// Own-box rows are excluded by the fold and stay excluded here. Harmless:
	// deploymentHost's other half (Allowed) is what admits this box's routes.
	if m.KnownHost("own.test") {
		t.Fatal("the fold excludes this box's own file, so KnownHost inherits that exclusion")
	}
	// Nothing polled yet: no membership, and no panic on the gate's path.
	if newHomeMap(s, "annie").KnownHost("exact.apps.test") {
		t.Fatal("a map that has never polled knows no host")
	}
}

// A box publishing eip:"" is in the fold's no-dial exclusion, and KnownHost
// inherits it: there is no DNS answer pointing at such a box, so a validation
// for its hosts has nowhere legitimate to land.
func TestKnownHostSkipsBoxesWithNoEIP(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	if m.KnownHost("b.test") {
		t.Fatal("a box with no public IP offers no validation target")
	}
}

// ---- hop acceptance ---------------------------------------------------------

// forwardEcho is an upstream that reports back exactly the trust-bearing
// headers it was handed, so a test can assert both what survived and what was
// stripped on the way in.
func forwardEcho(t *testing.T) *httptest.Server {
	t.Helper()
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintf(w, "xff=%s|xfh=%s|hop=%s|auth=%s",
			r.Header.Get("X-Forwarded-For"), r.Header.Get("X-Forwarded-Host"),
			r.Header.Get(hopHeader), r.Header.Get(hopAuthHeader))
	}))
	t.Cleanup(ts.Close)
	return ts
}

// hopEdge builds an edge serving hosts from one echo backend, with a mesh key
// minted and a home map armed over the same store — the receiving box of a
// backhaul hop.
func hopEdge(t *testing.T, self string, hosts ...string) (*Edge, *homeMap, []byte) {
	t.Helper()
	s := meshTestStore(t)
	if err := mintMeshKey(context.Background(), s); err != nil {
		t.Fatalf("mint: %v", err)
	}
	backend := forwardEcho(t)
	reg := openReg(t)
	for _, h := range hosts {
		if err := reg.Upsert(registry.Route{Host: h, Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready"}); err != nil {
			t.Fatalf("upsert %s: %v", h, err)
		}
	}
	e, err := New(reg, nil)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	t.Cleanup(e.probers.Stop)
	m := newHomeMap(s, self)
	refreshHomeMap(t, m)
	e.homes.Store(m)
	key, _ := m.meshKeys()
	if len(key) == 0 {
		t.Fatal("precondition: the mesh key must be mirrored")
	}
	return e, m, key
}

// hopRequest drives one request at the edge with the given headers.
func hopRequest(t *testing.T, e *Edge, host string, headers map[string]string) *http.Response {
	t.Helper()
	srv := httptest.NewServer(e)
	t.Cleanup(srv.Close)
	req, err := http.NewRequest("GET", srv.URL+"/", nil)
	if err != nil {
		t.Fatalf("new request: %v", err)
	}
	req.Host = host
	for k, v := range headers {
		req.Header.Set(k, v)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("do: %v", err)
	}
	t.Cleanup(func() { resp.Body.Close() })
	return resp
}

func readBody(t *testing.T, resp *http.Response) string {
	t.Helper()
	b, err := io.ReadAll(resp.Body)
	if err != nil {
		t.Fatalf("read body: %v", err)
	}
	return string(b)
}

// A verified hop is served locally, and the chain the sending edge appended
// is the one thing an ordinary client is never trusted for.
func TestVerifiedHopIsServedLocallyAndHonoursTheInboundXFF(t *testing.T) {
	e, _, key := hopEdge(t, "annie", "app.test")
	resp := hopRequest(t, e, "app.test", map[string]string{
		hopHeader:         "1",
		hopAuthHeader:     signHop(key, "app.test", time.Now()),
		"X-Forwarded-For": "203.0.113.9",
	})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	body := readBody(t, resp)
	// Verbatim, with NO element appended: the last element must still be the
	// client. internal/console/authcli.go's clientIP takes the last element as
	// the caller and every tenant framework's rightmost-public heuristic does
	// the same, so appending this edge's peer (the sibling edge) would make a
	// backhauled request look as though infrastructure called it.
	if !strings.HasPrefix(body, "xff=203.0.113.9|") {
		t.Fatalf("body = %q — a verified hop's inbound XFF must pass through verbatim, last element still the client", body)
	}
	// The hop's own trust headers are the edge's vocabulary, never the app's.
	if !strings.Contains(body, "|hop=|auth=") {
		t.Fatalf("body = %q — the hop headers must not reach the app", body)
	}
}

// The MAC binds the CONCRETE request host, and a catch-all route serves it
// through lookupUpstream's wildcard fallback — the two halves of carry C1 on
// the receiving side.
func TestVerifiedHopServesAWildcardRoutedHost(t *testing.T) {
	e, _, key := hopEdge(t, "annie", "*.apps.test")
	resp := hopRequest(t, e, "paul.apps.test", map[string]string{
		hopHeader:     "1",
		hopAuthHeader: signHop(key, "paul.apps.test", time.Now()),
	})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 — a catch-all route serves a hop like any other client", resp.StatusCode)
	}
}

// The stale-map refusal: the sender's map named us for a host we do not
// serve. 503 + Retry-After: 2, and structurally never a re-forward — the home
// map here is POISONED to name another box for exactly that host, so anything
// that consulted it would leave through a dial instead.
func TestVerifiedHopForAnUnservedHostRefusesWithoutForwarding(t *testing.T) {
	e, m, key := hopEdge(t, "annie", "app.test")
	putJSON(t, m.s, "control/edges/betty.json",
		`{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"moved.test":{"role":"home"}}}`)
	refreshHomeMap(t, m)
	if _, _, ok := m.Home("moved.test"); !ok {
		t.Fatal("precondition: the map must name a remote home for moved.test")
	}

	resp := hopRequest(t, e, "moved.test", map[string]string{
		hopHeader:     "1",
		hopAuthHeader: signHop(key, "moved.test", time.Now()),
	})
	if resp.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 for a verified hop naming a host we do not serve", resp.StatusCode)
	}
	if got := resp.Header.Get("Retry-After"); got != "2" {
		t.Fatalf("Retry-After = %q, want 2", got)
	}
	if body := readBody(t, resp); strings.Contains(body, "moved.test") || strings.Contains(body, "betty") {
		t.Fatalf("body = %q — the refusal must not name the host or where it went", body)
	}
}

// Task 1: the refusal used to be silent. A home move stuck mid-way can leave
// a sender retrying into this refusal for minutes (the loop cadences,
// edgesfile.go/homemap.go's own docs), so it now logs — but first-and-
// every-tenth, logPartialPoll's own cadence, or a stuck move would fill the
// journal with one line per hop.
func TestStaleHopRefusalLogsFirstAndEveryTenth(t *testing.T) {
	e, m, key := hopEdge(t, "annie", "app.test")
	putJSON(t, m.s, "control/edges/betty.json",
		`{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"moved.test":{"role":"home"}}}`)
	refreshHomeMap(t, m)

	buf := captureLog(t)
	for i := 0; i < 10; i++ {
		resp := hopRequest(t, e, "moved.test", map[string]string{
			hopHeader:     "1",
			hopAuthHeader: signHop(key, "moved.test", time.Now()),
		})
		if resp.StatusCode != http.StatusServiceUnavailable {
			t.Fatalf("refusal #%d: status = %d, want 503", i+1, resp.StatusCode)
		}
	}
	got := buf.String()
	if n := strings.Count(got, "moved.test"); n != 2 {
		t.Fatalf("log mentions the stale host %d time(s), want exactly 2 (occurrence #1 and #10, not one line per refusal): %s", n, got)
	}
	if !strings.Contains(got, "(#1)") {
		t.Fatalf("log output = %q, want the first occurrence logged immediately", got)
	}
	if !strings.Contains(got, "(#10)") {
		t.Fatalf("log output = %q, want the tenth occurrence logged", got)
	}
	if !strings.Contains(got, "box annie") {
		t.Fatalf("log output = %q, want this box's own label named in the line", got)
	}
}

// Review catch on this task: host is client-controlled (a wildcard-homed
// route lets a sending box mint a signed hop for any invented subdomain,
// staleHopN's own doc), so staleHopN must not grow without bound. Drives
// logStaleHop directly rather than through HTTP, so cap-plus-a-few distinct
// hosts stays a cheap, focused test of the invariant: the map's key count
// stops at cap+1 (every tracked key plus the one shared overflow bucket),
// and hosts past the cap accumulate under that bucket instead of growing
// the map further.
func TestStaleHopNIsCappedWithAnOverflowBucket(t *testing.T) {
	_, m, _ := hopEdge(t, "annie", "local.test")

	for i := 0; i < staleHopKeyCap; i++ {
		m.logStaleHop(fmt.Sprintf("host-%d.example.com", i))
	}
	if got := len(m.staleHopN); got != staleHopKeyCap {
		t.Fatalf("staleHopN has %d key(s) at the cap, want exactly %d", got, staleHopKeyCap)
	}

	// Three MORE distinct hosts, past the cap.
	for i := 0; i < 3; i++ {
		m.logStaleHop(fmt.Sprintf("overflow-%d.example.com", i))
	}
	if got, want := len(m.staleHopN), staleHopKeyCap+1; got != want {
		t.Fatalf("staleHopN has %d key(s) past the cap, want %d (cap distinct keys + one overflow bucket, not %d+3)", got, want, staleHopKeyCap)
	}
	const overflowKey = "annie (other)"
	if n := m.staleHopN[overflowKey]; n != 3 {
		t.Fatalf("overflow key %q count = %d, want 3 — each new host past the cap must fold into the shared bucket, not mint its own", overflowKey, n)
	}

	// A key that was already tracked BEFORE the cap keeps its own identity
	// and cadence — the cap only refuses NEW keys, never demotes an
	// existing one.
	m.logStaleHop("host-0.example.com")
	if n := m.staleHopN["annie host-0.example.com"]; n != 2 {
		t.Fatalf("a pre-cap key's count = %d, want 2 — it must keep incrementing under its own key, not fold into overflow", n)
	}
}

// Verification failure is not an error path: it is "you are an ordinary
// client", with every forwarding claim you made discarded.
func TestUnverifiedHopIsServedAsAnOrdinaryClientWithForwardedHeadersStripped(t *testing.T) {
	e, _, key := hopEdge(t, "annie", "app.test")
	tampered := []byte(signHop(key, "app.test", time.Now()))
	tampered[len(tampered)-1] ^= 'f' ^ '0' // flip the last hex nibble

	resp := hopRequest(t, e, "app.test", map[string]string{
		hopHeader:           "1",
		hopAuthHeader:       string(tampered),
		"X-Forwarded-For":   "203.0.113.9",
		"X-Forwarded-Host":  "evil.test",
		"X-Forwarded-Proto": "https",
	})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 — an unverified hop is an ordinary client, not a rejection", resp.StatusCode)
	}
	body := readBody(t, resp)
	if strings.Contains(body, "203.0.113.9") {
		t.Fatalf("body = %q — an unverified sender's X-Forwarded-For must not be honoured", body)
	}
	if strings.Contains(body, "evil.test") {
		t.Fatalf("body = %q — an unverified sender's X-Forwarded-Host must not be honoured", body)
	}
	if !strings.Contains(body, "|hop=|auth=") {
		t.Fatalf("body = %q — the hop headers must be stripped either way", body)
	}
}

// The strip is at the header level, not just a proxy side effect: every route
// class behind the edge (static trees, the status host, the activator) reads
// the same request.
func TestAcceptHopStripsEveryForwardedHeaderOnFailedVerification(t *testing.T) {
	e, _, _ := hopEdge(t, "annie", "app.test")
	r := httptest.NewRequest("GET", "http://app.test/", nil)
	r.Header.Set(hopHeader, "1")
	r.Header.Set(hopAuthHeader, "0:deadbeef")
	r.Header.Set("Forwarded", "for=203.0.113.9")
	r.Header.Set("X-Forwarded-For", "203.0.113.9")
	r.Header.Set("X-Forwarded-Host", "evil.test")
	r.Header.Set("X-Forwarded-Proto", "https")

	r, verified := e.acceptHop(r, "app.test")
	if verified {
		t.Fatal("a garbage auth header must not verify")
	}
	for _, h := range []string{"Forwarded", "X-Forwarded-For", "X-Forwarded-Host", "X-Forwarded-Proto", hopHeader, hopAuthHeader} {
		if got := r.Header.Get(h); got != "" {
			t.Fatalf("%s survived an unverified hop as %q", h, got)
		}
	}
	if hopVerified(r.Context()) {
		t.Fatal("an unverified request must not carry the hop mark")
	}

	// The half-formed case: a signature with no X-Carlos-Hop beside it. It is
	// not a hop and must not be treated as one — but the header is still this
	// edge's vocabulary, and leaving it on the request would let an app that
	// reflects its request headers launder it back onto the wire (the
	// scrubInboundPurgeHeader precedent, purge_header.go).
	stray := httptest.NewRequest("GET", "http://app.test/", nil)
	stray.Header.Set(hopAuthHeader, "0:deadbeef")
	stray, verified = e.acceptHop(stray, "app.test")
	if verified {
		t.Fatal("an auth header with no hop marker is not a hop")
	}
	if got := stray.Header.Get(hopAuthHeader); got != "" {
		t.Fatalf("%s survived without a hop marker as %q — hop vocabulary must never reach the app", hopAuthHeader, got)
	}
}

// The MAC is over the host, so a signature minted for one hop cannot be
// replayed against another — and the timestamp bounds it in time.
func TestHopVerificationIsBoundToTheHostAndTheClock(t *testing.T) {
	e, _, key := hopEdge(t, "annie", "app.test", "other.test")

	wrongHost := hopRequest(t, e, "app.test", map[string]string{
		hopHeader:         "1",
		hopAuthHeader:     signHop(key, "other.test", time.Now()),
		"X-Forwarded-For": "203.0.113.9",
	})
	if body := readBody(t, wrongHost); strings.Contains(body, "203.0.113.9") {
		t.Fatalf("body = %q — a signature minted for another host must not verify", body)
	}

	stale := hopRequest(t, e, "app.test", map[string]string{
		hopHeader:         "1",
		hopAuthHeader:     signHop(key, "app.test", time.Now().Add(-2*hopSkew)),
		"X-Forwarded-For": "203.0.113.9",
	})
	if body := readBody(t, stale); strings.Contains(body, "203.0.113.9") {
		t.Fatalf("body = %q — a signature outside the skew bound must not verify", body)
	}
}

// Fail closed: with no mesh key mirrored (a brand new deployment, or a box
// whose bucket is unreachable at boot) nothing verifies.
func TestHopWithoutAMirroredMeshKeyIsAnOrdinaryClient(t *testing.T) {
	e, _, key := hopEdge(t, "annie", "app.test")
	e.homes.Store(nil) // no mesh at all — the no-deployment-store case
	resp := hopRequest(t, e, "app.test", map[string]string{
		hopHeader:         "1",
		hopAuthHeader:     signHop(key, "app.test", time.Now()),
		"X-Forwarded-For": "203.0.113.9",
	})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	if body := readBody(t, resp); strings.Contains(body, "203.0.113.9") {
		t.Fatalf("body = %q — with no key to verify against, the sender is an ordinary client", body)
	}
}

// An unserved host that NO box homes keeps today's answer exactly: 404, with
// no Retry-After, and nothing said about the rest of the deployment.
//
// The home map here is deliberately non-empty and deliberately about a
// DIFFERENT host: the map is consulted (Task 4's backhaul) and misses, which
// is the only route to this answer now that a homed host is dialled instead.
// The stale-map 503 stays reachable only by a party holding the mesh key, so
// the two answers still cannot be told apart by a stranger.
func TestUnknownHostWithNoHomeIsStillThe404(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "app.test")
	putJSON(t, m.s, "control/edges/betty.json",
		`{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"moved.test":{"role":"home"}}}`)
	refreshHomeMap(t, m)
	resp := hopRequest(t, e, "nowhere.test", nil)
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("status = %d, want the unchanged 404 for a host nothing in the deployment homes", resp.StatusCode)
	}
	if got := resp.Header.Get("Retry-After"); got != "" {
		t.Fatalf("Retry-After = %q — a stranger must not be told to come back", got)
	}
}

// Regression pin: an ordinary client's forged X-Forwarded-For is no more
// trusted after this change than before it.
func TestOrdinaryClientForgedXFFIsStillDiscarded(t *testing.T) {
	e, _, _ := hopEdge(t, "annie", "app.test")
	resp := hopRequest(t, e, "app.test", map[string]string{"X-Forwarded-For": "203.0.113.9"})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	if body := readBody(t, resp); strings.Contains(body, "203.0.113.9") {
		t.Fatalf("body = %q — a client-supplied XFF is never honoured", body)
	}
}

// ---- backhaul dialing (the sending half) ------------------------------------

// homeEdgeSeen is what a fake home edge was handed on the wire.
type homeEdgeSeen struct {
	n                    int
	sni, host, hop, auth string
	proof                string
	xff                  []string
}

type homeEdgeRecord struct {
	mu   sync.Mutex
	seen homeEdgeSeen
}

func (h *homeEdgeRecord) get() homeEdgeSeen {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.seen
}

const homeEdgeBody = "served by the home edge"

// fakeHomeEdge stands in for the box that homes a host: a TLS server that
// records the ClientHello's SNI and the trust-bearing headers of whatever
// arrives, then runs h (a plain body writer unless the test needs a protocol
// switch).
//
// It also arms the two dial seams for the test's lifetime — backhaulPort, so
// the dial reaches an ephemeral httptest port instead of 443, and
// backhaulRootCAs, so this server's self-signed certificate is trusted the way
// a real home edge's publicly-trusted one is. Nothing about verification is
// weakened: httptest's certificate names example.com, so every test below asks
// for exactly that host and a name mismatch would fail the handshake here just
// as it would in production.
func fakeHomeEdge(t *testing.T, h http.Handler) (*homeEdgeRecord, string) {
	t.Helper()
	rec := &homeEdgeRecord{}
	if h == nil {
		h = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, homeEdgeBody) })
	}
	ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		rec.mu.Lock()
		rec.seen.n++
		rec.seen.host = r.Host
		rec.seen.hop = r.Header.Get(hopHeader)
		rec.seen.auth = r.Header.Get(hopAuthHeader)
		rec.seen.xff = r.Header.Values("X-Forwarded-For")
		rec.seen.proof = r.Header.Get(edgeproof.Header)
		rec.mu.Unlock()
		h.ServeHTTP(w, r)
	}))
	ts.TLS = &tls.Config{GetConfigForClient: func(chi *tls.ClientHelloInfo) (*tls.Config, error) {
		rec.mu.Lock()
		rec.seen.sni = chi.ServerName
		rec.mu.Unlock()
		return nil, nil // the server's own config, unchanged — this hook only watches
	}}
	ts.StartTLS()
	t.Cleanup(ts.Close)

	eip, port, err := net.SplitHostPort(ts.Listener.Addr().String())
	if err != nil {
		t.Fatalf("split home edge addr: %v", err)
	}
	pool := x509.NewCertPool()
	pool.AddCert(ts.Certificate())
	oldPort, oldCAs := backhaulPort, backhaulRootCAs
	backhaulPort, backhaulRootCAs = port, pool
	t.Cleanup(func() { backhaulPort, backhaulRootCAs = oldPort, oldCAs })
	return rec, eip
}

// homedOn publishes one box's edges file naming hosts as its own, and lands it
// in the map — the second box a single-box deployment never has.
func homedOn(t *testing.T, m *homeMap, box, eip string, hosts ...string) {
	t.Helper()
	entries := make([]string, 0, len(hosts))
	for _, h := range hosts {
		entries = append(entries, fmt.Sprintf("%q:{\"role\":\"home\"}", h))
	}
	putJSON(t, m.s, "control/edges/"+box+".json", fmt.Sprintf(
		`{"v":1,"box":%q,"eip":%q,"hosts":{%s}}`, box, eip, strings.Join(entries, ",")))
	refreshHomeMap(t, m)
}

// The whole of the sending half in one request: a host this box does not serve
// but another box homes reaches that box over real TLS, with the SNI it will
// need to present a certificate for, the real Host header, a hop signature the
// home edge can verify, and an X-Forwarded-For that names only the client.
func TestBackhaulDialsTheHomeEdgeOverVerifiedTLS(t *testing.T) {
	e, m, key := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "example.com")

	resp := hopRequest(t, e, "example.com", nil)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 — a remote-homed host must be served through its home edge", resp.StatusCode)
	}
	if body := readBody(t, resp); body != homeEdgeBody {
		t.Fatalf("body = %q, want %q", body, homeEdgeBody)
	}
	seen := rec.get()
	if seen.n != 1 {
		t.Fatalf("home edge saw %d request(s), want exactly one — one hop, ever", seen.n)
	}
	if seen.sni != "example.com" {
		t.Fatalf("SNI = %q, want the request host — the home edge answers with the cert for the host being asked for", seen.sni)
	}
	if seen.host != "example.com" {
		t.Fatalf("Host = %q, want the real hostname — the home edge routes on it", seen.host)
	}
	if seen.hop != "1" {
		t.Fatalf("%s = %q, want 1", hopHeader, seen.hop)
	}
	if !verifyHopEitherKey(key, nil, "example.com", seen.auth, time.Now()) {
		t.Fatalf("%s = %q did not verify under the deployment's mesh key", hopAuthHeader, seen.auth)
	}
	// One element, the client. Neither this edge's own address nor the home EIP
	// may be appended: the home edge restores this chain verbatim on a verified
	// hop, and every consumer takes the LAST element as the caller
	// (setForwarded, homemap.go).
	if len(seen.xff) != 1 || strings.Contains(seen.xff[0], ",") {
		t.Fatalf("X-Forwarded-For = %q, want a single-element chain naming only the client", seen.xff)
	}
}

// Precedence, first rule: a host served HERE is served here, whatever the home
// map says about it. The map is poisoned to name a reachable home for exactly
// that host, so a lookup in the wrong order would show up as a dial.
func TestLocalHitIgnoresTheHomeMap(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "example.com")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "example.com")

	resp := hopRequest(t, e, "example.com", nil)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	if body := readBody(t, resp); !strings.HasPrefix(body, "xff=") {
		t.Fatalf("body = %q, want the LOCAL backend's answer — a local hit must never consult the home map", body)
	}
	if n := rec.get().n; n != 0 {
		t.Fatalf("home edge saw %d request(s), want none", n)
	}
}

// One hop, ever: a request that ARRIVED as a verified hop is refused, never
// forwarded onward, even when the map offers a live target for it. The fake
// home edge is the zero-call witness — a re-forward would answer 200 here.
func TestArrivedHopNeverDialsTheHomeEdge(t *testing.T) {
	e, m, key := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "example.com")

	resp := hopRequest(t, e, "example.com", map[string]string{
		hopHeader:     "1",
		hopAuthHeader: signHop(key, "example.com", time.Now()),
	})
	if resp.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 — an arrived hop is refused, not re-forwarded", resp.StatusCode)
	}
	if got := resp.Header.Get("Retry-After"); got != "2" {
		t.Fatalf("Retry-After = %q, want 2", got)
	}
	if n := rec.get().n; n != 0 {
		t.Fatalf("home edge saw %d request(s), want none — a hop must never become a second hop", n)
	}
}

// The backhaul's own guard, exercised directly rather than through ServeHTTP's
// branch above it: if this call ever moves, an arrived hop must still fail
// closed here instead of being forwarded.
func TestBackhaulHelperRefusesAnArrivedHopOnItsOwn(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "example.com")

	r := httptest.NewRequest("GET", "http://example.com/", nil)
	r = r.WithContext(context.WithValue(r.Context(), hopVerifiedKey{}, true))
	w := httptest.NewRecorder()
	if !e.backhaul(w, r, "example.com") {
		t.Fatal("backhaul must ANSWER an arrived hop, not fall through to the unknown-host path")
	}
	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want the stale-map refusal", w.Code)
	}
	if n := rec.get().n; n != 0 {
		t.Fatalf("home edge saw %d request(s), want none", n)
	}
}

// Fail closed on the SENDING side too: with no mesh key mirrored there is
// nothing to prove the hop with, and signing under an empty key would cross a
// region only to be served as an anonymous client. The request gets today's
// unknown-host answer instead.
func TestBackhaulRefusesToDialWithoutAMeshKey(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "example.com")
	if err := m.s.Delete(context.Background(), meshKeyKey); err != nil {
		t.Fatalf("delete mesh key: %v", err)
	}
	refreshHomeMap(t, m)
	if cur, next := m.meshKeys(); len(cur) != 0 || len(next) != 0 {
		t.Fatal("precondition: the mirror must hold no key")
	}

	resp := hopRequest(t, e, "example.com", nil)
	if resp.StatusCode != http.StatusNotFound {
		t.Fatalf("status = %d, want the unchanged 404 — an unprovable hop is not dialled", resp.StatusCode)
	}
	if n := rec.get().n; n != 0 {
		t.Fatalf("home edge saw %d request(s), want none — never sign with an empty key", n)
	}
}

// A stranger forging the hop vocabulary for a remote-homed host is backhauled
// like any other client, and gains exactly nothing by having tried.
//
// Two halves have to hold at once and they meet only here. acceptHop deletes
// both hop headers and — verification having failed — strips every forwarding
// claim the forger made, so the request reaching this point is an ordinary
// one; backhaul then mints a FRESH signature under the deployment's own key,
// so the home edge sees a hop it can verify and an X-Forwarded-For naming the
// forger. What the forger must NOT get is the stale-map 503 (which would tell
// them the host lives elsewhere), a chain of their own choosing, or their
// garbage signature carried onward under this edge's name.
func TestForgedHopForARemoteHostIsBackhauledAsAnOrdinaryClient(t *testing.T) {
	e, m, key := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "example.com")

	const garbage = "0:deadbeef"
	resp := hopRequest(t, e, "example.com", map[string]string{
		hopHeader:         "1",
		hopAuthHeader:     garbage,
		"X-Forwarded-For": "203.0.113.9",
	})
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 — a forged hop is an ordinary client, not a refusal", resp.StatusCode)
	}
	if got := resp.Header.Get("Retry-After"); got != "" {
		t.Fatalf("Retry-After = %q — the stale-map answer must stay unreachable without the mesh key", got)
	}
	if body := readBody(t, resp); body != homeEdgeBody {
		t.Fatalf("body = %q, want the home edge's answer %q", body, homeEdgeBody)
	}
	seen := rec.get()
	if seen.n != 1 {
		t.Fatalf("home edge saw %d request(s), want exactly one", seen.n)
	}
	if seen.auth == garbage {
		t.Fatalf("%s = %q — the forger's signature must never be forwarded under this edge's name", hopAuthHeader, seen.auth)
	}
	if seen.hop != "1" || !verifyHopEitherKey(key, nil, "example.com", seen.auth, time.Now()) {
		t.Fatalf("hop=%q auth=%q — the hop must carry a FRESH signature this edge minted", seen.hop, seen.auth)
	}
	// The forger's claimed chain is gone and the one that arrives names the
	// forger's own address, once. Anything else would let a stranger choose
	// what the app (and the console's rate limiter) believes about who called.
	if len(seen.xff) != 1 || strings.Contains(seen.xff[0], ",") {
		t.Fatalf("X-Forwarded-For = %q, want a single-element chain", seen.xff)
	}
	if strings.Contains(seen.xff[0], "203.0.113.9") {
		t.Fatalf("X-Forwarded-For = %q — a forged chain must be discarded, not carried", seen.xff)
	}
	if ip := net.ParseIP(seen.xff[0]); ip == nil || !ip.IsLoopback() {
		t.Fatalf("X-Forwarded-For = %q, want the forger's own (loopback) address", seen.xff)
	}
}

// A catch-all row in another box's edges file resolves through the same
// one-label wildcard rule lookupUpstream uses — but the signature must be over
// the CONCRETE request host, because that is what the home edge verifies
// against. Signing the map KEY would produce a MAC for "*.com", which is not a
// request host under any canonicalization.
func TestBackhaulSignsTheConcreteHostNotTheWildcardKey(t *testing.T) {
	e, m, key := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "*.com")
	if _, _, ok := m.Home("example.com"); !ok {
		t.Fatal("precondition: the wildcard row must offer a home for example.com")
	}

	if resp := hopRequest(t, e, "example.com", nil); resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	seen := rec.get()
	if !verifyHop(key, "example.com", seen.auth, time.Now()) {
		t.Fatalf("%s = %q must verify against the concrete request host", hopAuthHeader, seen.auth)
	}
	if verifyHop(key, "*.com", seen.auth, time.Now()) {
		t.Fatalf("%s = %q verifies against the wildcard map key — the MAC must bind the concrete host", hopAuthHeader, seen.auth)
	}
}

// A WebSocket must compose through the hop with no code of its own: the dial
// branch sits above ServeHTTP's cache/compression wrapping, so the client's
// hijackable writer reaches httputil.ReverseProxy's protocol-switch path, and
// the transport speaks HTTP/1.1 (no ForceAttemptHTTP2) so that path works.
// Bytes must then flow both ways with no whole-body deadline cutting them off.
func TestBackhaulCarriesAWebSocketUpgrade(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		hj, ok := w.(http.Hijacker)
		if !ok {
			http.Error(w, "not hijackable", http.StatusInternalServerError)
			return
		}
		conn, buf, err := hj.Hijack()
		if err != nil {
			return
		}
		defer conn.Close()
		buf.WriteString("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n")
		if err := buf.Flush(); err != nil {
			return
		}
		line, err := buf.ReadString('\n')
		if err != nil {
			return
		}
		buf.WriteString("home:" + line)
		buf.Flush()
	}))
	homedOn(t, m, "betty", eip, "example.com")

	srv := httptest.NewServer(e)
	t.Cleanup(srv.Close)
	c, err := net.Dial("tcp", strings.TrimPrefix(srv.URL, "http://"))
	if err != nil {
		t.Fatalf("dial edge: %v", err)
	}
	defer c.Close()
	if err := c.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {
		t.Fatalf("set deadline: %v", err)
	}
	if _, err := fmt.Fprint(c, "GET /socket HTTP/1.1\r\nHost: example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"); err != nil {
		t.Fatalf("write handshake: %v", err)
	}
	br := bufio.NewReader(c)
	resp, err := http.ReadResponse(br, nil)
	if err != nil {
		t.Fatalf("read handshake response: %v", err)
	}
	if resp.StatusCode != http.StatusSwitchingProtocols {
		t.Fatalf("handshake status = %d, want 101 — an upgrade must switch protocols through the hop", resp.StatusCode)
	}
	if _, err := fmt.Fprint(c, "ping\n"); err != nil {
		t.Fatalf("write post-upgrade frame: %v", err)
	}
	line, err := br.ReadString('\n')
	if err != nil {
		t.Fatalf("read post-upgrade frame: %v", err)
	}
	if line != "home:ping\n" {
		t.Fatalf("post-upgrade echo = %q, want %q — bytes must flow both ways across the backhaul", line, "home:ping\n")
	}
	if seen := rec.get(); seen.hop != "1" {
		t.Fatalf("%s = %q — an upgrade is a hop like any other", hopHeader, seen.hop)
	}
}

// A home whose edge cannot be reached is a bad gateway, not a 503: this box
// believed a specific box was there. The unreachable target here is the
// documented cert-less bootstrap window's shape too — a dial or handshake that
// does not complete, healed by the next mirror poll.
func TestBackhaulToAnUnreachableHomeIsA502(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	// A port nothing listens on, reached through the same seam the fake home
	// edge uses. 127.0.0.1 refuses immediately rather than hanging for the dial
	// budget.
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	eip, port, err := net.SplitHostPort(ln.Addr().String())
	if err != nil {
		t.Fatalf("split: %v", err)
	}
	ln.Close()
	old := backhaulPort
	backhaulPort = port
	t.Cleanup(func() { backhaulPort = old })
	homedOn(t, m, "betty", eip, "example.com")

	resp := hopRequest(t, e, "example.com", nil)
	if resp.StatusCode != http.StatusBadGateway {
		t.Fatalf("status = %d, want 502 for a home edge that could not be reached", resp.StatusCode)
	}
	if got := resp.Header.Get("Retry-After"); got != "" {
		t.Fatalf("Retry-After = %q — 503 + Retry-After is reserved for knowing nobody is home", got)
	}
}

// Task 1: an unreachable home used to log unthrottled, once per failed dial —
// a home box down for any real stretch would fill the journal with one line
// per request. It now goes through e.logThrottled, keyed "backhaul-error
// "+box, so a run of failures against the SAME home box produces one line
// (occurrence #1), not one per failure.
func TestBackhaulErrorHandlerRoutesThroughLogThrottled(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	eip, port, err := net.SplitHostPort(ln.Addr().String())
	if err != nil {
		t.Fatalf("split: %v", err)
	}
	ln.Close()
	old := backhaulPort
	backhaulPort = port
	t.Cleanup(func() { backhaulPort = old })
	homedOn(t, m, "betty", eip, "example.com")

	buf := captureLog(t)
	for i := 0; i < 3; i++ {
		resp := hopRequest(t, e, "example.com", nil)
		if resp.StatusCode != http.StatusBadGateway {
			t.Fatalf("attempt #%d: status = %d, want 502", i+1, resp.StatusCode)
		}
	}
	got := buf.String()
	// logThrottled's own format is "edge: <msg> (occurrence #N)" — it prints
	// no key of its own, so the key's effect is observed as throttling: three
	// failed dials against the SAME "backhaul-error "+box key must produce
	// exactly one logged line (#1), not three.
	if n := strings.Count(got, "backhaul"); n != 1 {
		t.Fatalf("log mentions the backhaul failure %d time(s), want exactly 1 (logThrottled's first-and-every-300th, not one line per failed dial): %s", n, got)
	}
	if !strings.Contains(got, "occurrence #1") {
		t.Fatalf("log output = %q, want the first occurrence logged (logThrottled's own format)", got)
	}
	if strings.Contains(got, "occurrence #2") || strings.Contains(got, "occurrence #3") {
		t.Fatalf("log output = %q, want the second and third failures throttled, not logged again", got)
	}
	if !strings.Contains(got, "example.com") || !strings.Contains(got, "betty") {
		t.Fatalf("log output = %q, want the host and the home box named", got)
	}
}

// One transport per home BOX. The pool is what makes a backhaul cost one
// handshake rather than one per request; keying it per box rather than per
// host is what keeps a catch-all home — where every invented subdomain is a
// backhaul target — from growing the map without limit. Per-host identity
// still exists, one layer down: the outbound authority is the hostname, so SNI
// and pooled connections are per host inside the box's transport (the SNI
// assertion in TestBackhaulDialsTheHomeEdgeOverVerifiedTLS is that half).
func TestBackhaulTransportsAreCachedPerHomeBox(t *testing.T) {
	e, _, _ := hopEdge(t, "annie", "local.test")
	a := e.backhaulTransport("1.1.1.1")
	if b := e.backhaulTransport("1.1.1.1"); b != a {
		t.Fatal("the same home box must reuse one transport, or every request pays a fresh TLS handshake")
	}
	if b := e.backhaulTransport("2.2.2.2"); b == a {
		t.Fatal("a different home box must get its own transport — the dial address is baked into it")
	}
	if sn := a.TLSClientConfig.ServerName; sn != "" {
		t.Fatalf("ServerName = %q, want it unset — pinning one host per transport is what a wildcard home would multiply", sn)
	}
	if a.TLSClientConfig.InsecureSkipVerify {
		t.Fatal("the hop verifies the home edge's certificate — always")
	}
}

// Two hosts homed on one box share that box's transport, and each still
// reaches the home edge under its OWN name: the shared transport is the memory
// property, the per-host SNI is the correctness one.
func TestBackhaulSharesOneTransportAcrossAWildcardHome(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "*.com")

	if resp := hopRequest(t, e, "example.com", nil); resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	if sni := rec.get().sni; sni != "example.com" {
		t.Fatalf("SNI = %q, want the concrete request host", sni)
	}
	n := 0
	e.backhaulTransports.Range(func(_, _ any) bool { n++; return true })
	if n != 1 {
		t.Fatalf("%d transports cached for one home box, want 1 — a catch-all home must not mint one per hostname", n)
	}
	if _, ok := e.backhaulTransports.Load(eip); !ok {
		t.Fatalf("the cached transport must be keyed by the home box's EIP (%s)", eip)
	}
}

// The :80 half of the backhaul, carried from Task 4's review. A host homed on
// another box is answered on 443 (the hop), so plain http:// must redirect it
// there like any other routable host — otherwise a first-time visitor typing a
// bare hostname in the wrong region gets a 404 for an app that is up.
//
// The redirect grants nothing that the 443 answer does not: it is a 302 to the
// SAME host, never an ACME order (httpFallbackHandler's own doc records why
// this gate is wider than Allowed's). And a verified hop can never arrive
// here — the sending edge dials 443 — so there is no hop logic on this path.
func TestHTTPFallbackRedirectsARemoteHomedHost(t *testing.T) {
	e, m, _ := hopEdge(t, "annie") // serves nothing itself
	homedOn(t, m, "betty", "2.2.2.2", "remote.test", "*.wild.test")
	h := e.httpFallbackHandler()

	for _, host := range []string{"remote.test", "paul.wild.test"} {
		req := httptest.NewRequest(http.MethodGet, "http://"+host+"/deep?q=1", nil)
		req.Host = host
		rec := httptest.NewRecorder()
		h.ServeHTTP(rec, req)
		if rec.Code != http.StatusFound {
			t.Fatalf("%s: status = %d, want 302 — the host is homed on another box and 443 backhauls it", host, rec.Code)
		}
		if loc := rec.Header().Get("Location"); loc != "https://"+host+"/deep?q=1" {
			t.Fatalf("%s: Location = %q", host, loc)
		}
	}

	// A host nothing in the deployment homes keeps today's answer exactly.
	req := httptest.NewRequest(http.MethodGet, "http://nowhere.test/", nil)
	req.Host = "nowhere.test"
	rec := httptest.NewRecorder()
	h.ServeHTTP(rec, req)
	if rec.Code != http.StatusNotFound {
		t.Fatalf("nowhere.test: status = %d, want the unchanged 404", rec.Code)
	}
	if loc := rec.Header().Get("Location"); loc != "" {
		t.Fatalf("nowhere.test: Location = %q, want none", loc)
	}
}

// ---- containment: the per-home-box in-flight cap --------------------------

// backhaulCap lowers the in-flight cap for one test and restores it after —
// the package-seam pattern backhaulPort and homeMapInterval already follow.
func backhaulCap(t *testing.T, n int64) {
	t.Helper()
	old := backhaulMaxInflight.Load()
	backhaulMaxInflight.Store(n)
	t.Cleanup(func() { backhaulMaxInflight.Store(old) })
}

// backhaulGetter drives requests for host at one edge over a single reused
// server, so several can be in flight at once (hopRequest stands a server per
// call, which is fine for one-shot tests and useless for a concurrency one).
func backhaulGetter(t *testing.T, e *Edge) func(host string) (*http.Response, error) {
	t.Helper()
	srv := httptest.NewServer(e)
	t.Cleanup(srv.Close)
	return func(host string) (*http.Response, error) {
		req, err := http.NewRequest(http.MethodGet, srv.URL+"/", nil)
		if err != nil {
			return nil, err
		}
		req.Host = host
		return http.DefaultClient.Do(req)
	}
}

// The cap itself: with two hops held open toward one home box, a third is
// answered 503 + Retry-After: 2 HERE — no dial, no cross-region bytes — and
// the slot a finished hop gives back is immediately usable again.
//
// This is what bounds a scanner's forged-Host flood: egress is (concurrent
// streams × per-stream bandwidth), and this caps the first factor per box.
func TestBackhaulRefusesAtInflightCap(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	backhaulCap(t, 2)

	started, hold := make(chan struct{}, 4), make(chan struct{})
	rec, eip := fakeHomeEdge(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		started <- struct{}{}
		<-hold
		fmt.Fprint(w, homeEdgeBody)
	}))
	homedOn(t, m, "betty", eip, "example.com")
	get := backhaulGetter(t, e)

	done := make(chan error, 2)
	for i := 0; i < 2; i++ {
		go func() {
			resp, err := get("example.com")
			if err != nil {
				done <- err
				return
			}
			defer resp.Body.Close()
			if resp.StatusCode != http.StatusOK {
				done <- fmt.Errorf("held hop: status = %d, want 200", resp.StatusCode)
				return
			}
			body, err := io.ReadAll(resp.Body)
			if err != nil {
				done <- err
				return
			}
			if string(body) != homeEdgeBody {
				done <- fmt.Errorf("held hop: body = %q", body)
				return
			}
			done <- nil
		}()
	}
	for i := 0; i < 2; i++ {
		select {
		case <-started:
		case err := <-done:
			t.Fatalf("a hop finished before the cap was full: %v", err)
		case <-time.After(15 * time.Second):
			t.Fatal("the two hops that fill the cap never reached the home edge")
		}
	}

	// Both slots are held. The third request must be refused by THIS box.
	dialled := rec.get().n
	resp, err := get("example.com")
	if err != nil {
		t.Fatalf("third request: %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 at the in-flight cap", resp.StatusCode)
	}
	if got := resp.Header.Get("Retry-After"); got != "2" {
		t.Fatalf("Retry-After = %q, want 2 — the cap clears as hops finish", got)
	}
	if got := rec.get().n; got != dialled {
		t.Fatalf("the home edge saw %d request(s), want %d — a refusal must cost no dial and no cross-region bytes", got, dialled)
	}

	// Released slots are usable again: the cap is a bound on CONCURRENCY, not
	// a quota that a box burns through and never gets back.
	close(hold)
	for i := 0; i < 2; i++ {
		select {
		case err := <-done:
			if err != nil {
				t.Fatal(err)
			}
		case <-time.After(15 * time.Second):
			t.Fatal("a held hop never completed after its home edge was released")
		}
	}
	// Retried for a bounded window rather than asserted on the first attempt:
	// a held hop's client-side goroutine reports completion when the BODY is
	// read, which can be a hair before the edge's own deferred release runs.
	// The property under test is that the slots come back, not that they come
	// back within one scheduling quantum.
	var (
		after   *http.Response
		lastErr error
	)
	deadline := time.Now().Add(10 * time.Second)
	for {
		after, lastErr = get("example.com")
		if lastErr != nil {
			t.Fatalf("fourth request: %v", lastErr)
		}
		if after.StatusCode == http.StatusOK {
			break
		}
		after.Body.Close()
		if after.StatusCode != http.StatusServiceUnavailable {
			t.Fatalf("status = %d after the held hops finished, want 200", after.StatusCode)
		}
		if time.Now().After(deadline) {
			t.Fatal("the cap never cleared after both held hops finished — a released slot was not given back")
		}
		time.Sleep(10 * time.Millisecond)
	}
	defer after.Body.Close()
	if got := rec.get().n; got < dialled+1 {
		t.Fatalf("the home edge saw %d request(s), want at least %d — the fourth request must dial normally", got, dialled+1)
	}
}

// The cap is per HOME BOX: one saturated box must not stop this edge serving
// hosts homed on any other. A deployment-wide counter would let one busy (or
// one attacked) box take the whole mesh down.
func TestBackhaulInflightCapIsPerHomeBox(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	backhaulCap(t, 1)
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "carol", eip, "example.com")
	// Betty's EIP is never dialled in this test — her slot is filled directly
	// below, so the refusal happens before any dial. (TEST-NET-1: a dial
	// would hang for the whole 5s budget and be visible as a failure.)
	homedOn(t, m, "betty", "192.0.2.5", "saturated.test")

	release, ok := e.backhaulAcquire("betty")
	if !ok {
		t.Fatal("the first slot toward a box must always be available")
	}
	t.Cleanup(release)

	if resp := hopRequest(t, e, "saturated.test", nil); resp.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status = %d, want 503 — betty's only slot is held", resp.StatusCode)
	}
	resp := hopRequest(t, e, "example.com", nil)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 — carol's cap is her own", resp.StatusCode)
	}
	if body := readBody(t, resp); body != homeEdgeBody {
		t.Fatalf("body = %q, want the home edge's own answer", body)
	}
	if n := rec.get().n; n != 1 {
		t.Fatalf("carol saw %d request(s), want exactly one", n)
	}
}

// A slot is held for the LIFE of the stream, not until the response headers
// arrive: a long-lived stream (a WebSocket, an SSE feed, a large download) is
// precisely the sustained cross-region egress the cap exists to bound, so it
// must count against the cap for as long as it flows.
func TestBackhaulSlotHeldForStreamLife(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	backhaulCap(t, 1)

	streaming, finish := make(chan struct{}), make(chan struct{})
	_, eip := fakeHomeEdge(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprint(w, "first ")
		w.(http.Flusher).Flush() // headers and a first chunk are on the wire
		close(streaming)
		<-finish
		fmt.Fprint(w, "last")
	}))
	homedOn(t, m, "betty", eip, "example.com")
	get := backhaulGetter(t, e)

	stream, err := get("example.com")
	if err != nil {
		t.Fatalf("stream request: %v", err)
	}
	defer stream.Body.Close()
	if stream.StatusCode != http.StatusOK {
		t.Fatalf("stream status = %d, want 200", stream.StatusCode)
	}
	select {
	case <-streaming:
	case <-time.After(15 * time.Second):
		t.Fatal("the streaming hop never reached the home edge")
	}

	// Headers have arrived and the body is still flowing: the slot is held.
	busy, err := get("example.com")
	if err != nil {
		t.Fatalf("second request: %v", err)
	}
	defer busy.Body.Close()
	if busy.StatusCode != http.StatusServiceUnavailable {
		t.Fatalf("status = %d while a hop is mid-stream, want 503 — the slot is held for the stream's life", busy.StatusCode)
	}

	close(finish)
	body, err := io.ReadAll(stream.Body)
	if err != nil {
		t.Fatalf("read stream: %v", err)
	}
	if string(body) != "first last" {
		t.Fatalf("stream body = %q, want both chunks", body)
	}
	stream.Body.Close()

	// And the moment it ends, the slot is back.
	after, err := get("example.com")
	if err != nil {
		t.Fatalf("third request: %v", err)
	}
	defer after.Body.Close()
	if after.StatusCode != http.StatusOK {
		t.Fatalf("status = %d after the stream ended, want 200 — the slot must have been released", after.StatusCode)
	}
}

// Saturation is one throttled journal line per home box, never silence and
// never one line per refused request: a flood is a STANDING condition, and
// the key is the box (bounded by the deployment) and never the host (which a
// stranger picks, and which a catch-all home would multiply without limit).
func TestBackhaulSaturationLogsThrottledPerBox(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "local.test")
	backhaulCap(t, 1)
	// Neither EIP is ever dialled: both boxes' only slot is held below, so
	// every request in this test is refused before the dial.
	homedOn(t, m, "betty", "192.0.2.5", "betty.test")
	homedOn(t, m, "carol", "192.0.2.6", "carol.test")
	for _, box := range []string{"betty", "carol"} {
		release, ok := e.backhaulAcquire(box)
		if !ok {
			t.Fatalf("%s: the first slot toward a box must always be available", box)
		}
		t.Cleanup(release)
	}

	buf := captureLog(t)
	for _, host := range []string{"betty.test", "betty.test", "carol.test"} {
		if resp := hopRequest(t, e, host, nil); resp.StatusCode != http.StatusServiceUnavailable {
			t.Fatalf("%s: status = %d, want 503", host, resp.StatusCode)
		}
	}

	got := buf.String()
	// logThrottled prints no key of its own, so key isolation is observed as
	// TWO first occurrences (one bucket per box) with betty's second refusal
	// throttled inside her own bucket.
	if n := strings.Count(got, "occurrence #1"); n != 2 {
		t.Fatalf("%d first-occurrence line(s), want one per home box (two buckets): %s", n, got)
	}
	if strings.Contains(got, "occurrence #2") {
		t.Fatalf("log output = %q, want the second refusal against the same box throttled", got)
	}
	if n := strings.Count(got, "in-flight cap"); n != 2 {
		t.Fatalf("%d saturation line(s), want exactly two: %s", n, got)
	}
	if !strings.Contains(got, "betty") || !strings.Contains(got, "carol") {
		t.Fatalf("log output = %q, want each saturated home box named", got)
	}
}

// The knob: unset, unparseable and non-positive all mean the default. There
// is deliberately no "off" — an unbounded cross-region path is what the cap
// closes — and a typo must not silently zero-cap a deployment either.
func TestBackhaulMaxInflightFromEnv(t *testing.T) {
	// The ambient value is captured and restored by hand: t.Setenv restores
	// what IT set, but the unset case below has to clear the variable, and an
	// operator's own CARLOS_BACKHAUL_MAX_INFLIGHT must survive this test.
	ambient, had := os.LookupEnv("CARLOS_BACKHAUL_MAX_INFLIGHT")
	t.Cleanup(func() {
		if had {
			os.Setenv("CARLOS_BACKHAUL_MAX_INFLIGHT", ambient)
			return
		}
		os.Unsetenv("CARLOS_BACKHAUL_MAX_INFLIGHT")
	})

	for _, tc := range []struct {
		set  bool
		raw  string
		want int
	}{
		{set: false, want: defaultBackhaulMaxInflight},
		{set: true, raw: "", want: defaultBackhaulMaxInflight},
		{set: true, raw: "  ", want: defaultBackhaulMaxInflight},
		{set: true, raw: "sixty-four", want: defaultBackhaulMaxInflight},
		{set: true, raw: "0", want: defaultBackhaulMaxInflight},
		{set: true, raw: "-5", want: defaultBackhaulMaxInflight},
		{set: true, raw: "128", want: 128},
		{set: true, raw: " 8 ", want: 8},
	} {
		if tc.set {
			t.Setenv("CARLOS_BACKHAUL_MAX_INFLIGHT", tc.raw)
		} else {
			os.Unsetenv("CARLOS_BACKHAUL_MAX_INFLIGHT")
		}
		if got := backhaulInflightFromEnv(); got != tc.want {
			t.Fatalf("CARLOS_BACKHAUL_MAX_INFLIGHT=%q (set=%v) → %d, want %d", tc.raw, tc.set, got, tc.want)
		}
	}
}

// oneKeyFailingStore fails Get for one named key — a single unreadable object
// mid-poll, the case the carry-forward exists for.
type oneKeyFailingStore struct {
	release.Store
	mu  sync.Mutex
	bad string
}

func (o *oneKeyFailingStore) failKey(key string) {
	o.mu.Lock()
	o.bad = key
	o.mu.Unlock()
}

func (o *oneKeyFailingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	o.mu.Lock()
	bad := o.bad
	o.mu.Unlock()
	if key == bad {
		return nil, "", errors.New("bucket down for this object")
	}
	return o.Store.Get(ctx, key)
}

// ---- owners (#175 P7, the edge-local door) ---------------------------------

// Owner answers through the same lookup as Home, exact and one-label
// wildcard, and its dial target is Home's for the same host.
func TestOwnerFoldsThroughExactAndWildcardLookup(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{`+
		`"acme.go.tito.io":{"role":"home","account":"bae","app":"titogo"},`+
		`"*.classic.tito.io":{"role":"home","account":"bae","app":"classic"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	for _, c := range []struct{ host, app string }{
		{"acme.go.tito.io", "titogo"},
		{"ti.classic.tito.io", "classic"},
	} {
		acct, app, eip, box, ok := m.Owner(c.host)
		if !ok || acct != "bae" || app != c.app {
			t.Fatalf("Owner(%s) = %s/%s %v, want bae/%s", c.host, acct, app, ok, c.app)
		}
		heip, hbox, hok := m.Home(c.host)
		if !hok || eip != heip || box != hbox {
			t.Fatalf("Owner(%s) dials %s/%s, Home dials %s/%s", c.host, eip, box, heip, hbox)
		}
	}
	if _, _, _, _, ok := m.Owner("a.b.classic.tito.io"); ok {
		t.Fatal("a two-label host must miss the one-label wildcard, as Home does")
	}
}

// No published owner (a mixed hostname, or a publisher that predates the
// field), a cert-only entry, an own-box entry, or a nil map: ok is false and
// nothing else is answered.
func TestOwnerIsFalseWithoutAnOwnerOrADialTarget(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"own.test":{"role":"home","account":"bae","app":"titogo"}}}`)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{`+
		`"mixed.test":{"role":"home"},`+
		`"halfowned.test":{"role":"home","account":"bae"},`+
		`"*.certonly.test":{"role":"wildcard","wildcard":true,"account":"bae","app":"titogo"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	for _, host := range []string{"mixed.test", "halfowned.test", "x.certonly.test", "own.test", "nowhere.test"} {
		if acct, app, eip, box, ok := m.Owner(host); ok || acct != "" || app != "" || eip != "" || box != "" {
			t.Errorf("Owner(%s) = %q/%q %q/%q %v, want zeros and false", host, acct, app, eip, box, ok)
		}
	}
	// An exact entry with no owner is the winner: the lookup does not fall
	// through to a wildcard that would have named one.
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{`+
		`"exact.wild.test":{"role":"home"},"*.wild.test":{"role":"home","account":"bae","app":"titogo"}}}`)
	refreshHomeMap(t, m)
	if _, _, _, _, ok := m.Owner("exact.wild.test"); ok {
		t.Error("an ownerless exact entry borrowed the wildcard's owner")
	}
	var nilMap *homeMap
	if acct, app, eip, box, ok := nilMap.Owner("acme.go.tito.io"); ok || acct != "" || app != "" || eip != "" || box != "" {
		t.Error("a nil map answered an owner")
	}
}

// A route replacing an alias on a name collision carries the ROUTE's owner,
// whatever box sorted first. The owner is never merged across entries.
func TestOwnerOfARouteReplacingAnAliasIsTheRoutes(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"shared.test":{"role":"home","alias":true,"account":"bae","app":"squatter"},`+
		`"ownerless.test":{"role":"home","alias":true,"account":"bae","app":"titogo"}}}`)
	putJSON(t, s, "control/edges/carol.json", `{"v":1,"box":"carol","eip":"3.3.3.3","hosts":{"shared.test":{"role":"home","account":"bae","app":"titogo"},`+
		`"ownerless.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	refreshHomeMap(t, m)
	acct, app, eip, box, ok := m.Owner("shared.test")
	if !ok || acct != "bae" || app != "titogo" || eip != "3.3.3.3" || box != "carol" {
		t.Fatalf("Owner(shared.test) = %s/%s %s/%s %v, want the route's bae/titogo on carol", acct, app, eip, box, ok)
	}
	// The ownerless route on carol replaces betty's owned alias: its lack of
	// an owner is not filled in from the alias it beat.
	if _, _, _, box, ok := m.Owner("ownerless.test"); ok {
		t.Fatalf("Owner(ownerless.test) = %s, true; the losing alias's owner leaked into the winner", box)
	}
	if _, box, _ := m.Home("ownerless.test"); box != "carol" {
		t.Fatalf("Home(ownerless.test) = %s, want carol", box)
	}
}

func TestHomeMapHigherStampWinsWhateverTheSortOrder(t *testing.T) {
	s := meshTestStore(t)
	// "annie" sorts first and is the dead box; "betty" holds the newer installation.
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home","stamp":41}}}`)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"shop.test":{"role":"home","stamp":57}}}`)
	m := newHomeMap(s, "carol")
	refreshHomeMap(t, m)
	if eip, box, ok := m.Home("shop.test"); !ok || box != "betty" || eip != "2.2.2.2" {
		t.Fatalf("Home = %q %q %v, want betty", eip, box, ok)
	}
	// And the other way round, so it is the stamp and not the name.
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home","stamp":90}}}`)
	refreshHomeMap(t, m)
	if _, box, _ := m.Home("shop.test"); box != "annie" {
		t.Fatalf("Home box = %q, want annie", box)
	}
}

func TestHomeMapStampedBeatsUnstamped(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home"}}}`)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"shop.test":{"role":"home","stamp":3}}}`)
	m := newHomeMap(s, "carol")
	refreshHomeMap(t, m)
	if _, box, _ := m.Home("shop.test"); box != "betty" {
		t.Fatalf("Home box = %q, want betty", box)
	}
}

func TestHomeMapUnstampedKeepsTodaysRule(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home","alias":true}}}`)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"shop.test":{"role":"home"}}}`)
	m := newHomeMap(s, "carol")
	refreshHomeMap(t, m)
	if _, box, _ := m.Home("shop.test"); box != "betty" {
		t.Fatalf("a route still beats an alias: got %q", box)
	}
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home"}}}`)
	refreshHomeMap(t, m)
	if _, box, _ := m.Home("shop.test"); box != "annie" {
		t.Fatalf("two plain routes: first box in sorted order; got %q", box)
	}
}

func TestHomeMapAStampedAliasDoesNotBeatAnUnstampedRoute(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home"}}}`)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"shop.test":{"role":"home","alias":true,"stamp":9}}}`)
	m := newHomeMap(s, "carol")
	refreshHomeMap(t, m)
	if _, box, _ := m.Home("shop.test"); box != "annie" {
		t.Fatalf("got %q: #287's rule (a route beats an alias) comes before stamps", box)
	}
}

func TestHomeMapAStampedEntryWithNoDialAddressDoesNotWin(t *testing.T) {
	s := meshTestStore(t)
	putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home","stamp":2}}}`)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"","hosts":{"shop.test":{"role":"home","stamp":99}}}`)
	m := newHomeMap(s, "carol")
	refreshHomeMap(t, m)
	if eip, box, ok := m.Home("shop.test"); !ok || box != "annie" || eip != "1.1.1.1" {
		t.Fatalf("Home = %q %q %v, want annie: a box nothing can dial is skipped today and stays skipped", eip, box, ok)
	}
}

func file(box, eip string, hosts map[string]edgesHostView) edgesFileView {
	return edgesFileView{V: 1, Box: box, EIP: eip, Hosts: hosts}
}

func TestHomeMapReportsTwoBoxesAtTheSameStamp(t *testing.T) {
	files := map[string]edgesFileView{
		"control/edges/annie.json": file("annie", "1.1.1.1", map[string]edgesHostView{
			"tied.test": {Role: "home", Stamp: 7}, "fine.test": {Role: "home", Stamp: 7}, "plain.test": {Role: "home"}}),
		"control/edges/betty.json": file("betty", "2.2.2.2", map[string]edgesHostView{
			"tied.test": {Role: "home", Stamp: 7}, "fine.test": {Role: "home", Stamp: 9}, "plain.test": {Role: "home"}}),
	}
	hosts, tied := foldHomes(files, "carol", false)
	if len(tied) != 1 || tied[0] != "tied.test" {
		t.Fatalf("tied = %v, want [tied.test]: unstamped duplicates are normal (static sites run on every box) and are not reported", tied)
	}
	if hosts["tied.test"].box != "annie" {
		t.Fatalf("a tie keeps the first box in sorted order; got %q", hosts["tied.test"].box)
	}
	if hosts["fine.test"].box != "betty" {
		t.Fatalf("fine.test box = %q", hosts["fine.test"].box)
	}
}

// A tie between two boxes is reported even when a third box holds a higher
// stamp and wins the dial: the report does not depend on the winner.
func TestHomeMapReportsATieBelowTheWinner(t *testing.T) {
	files := map[string]edgesFileView{
		"control/edges/annie.json": file("annie", "1.1.1.1", map[string]edgesHostView{"h.test": {Role: "home", Stamp: 9}}),
		"control/edges/betty.json": file("betty", "2.2.2.2", map[string]edgesHostView{"h.test": {Role: "home", Stamp: 7}}),
		"control/edges/carol.json": file("carol", "3.3.3.3", map[string]edgesHostView{"h.test": {Role: "home", Stamp: 7}}),
	}
	hosts, tied := foldHomes(files, "dave", false)
	if len(tied) != 1 || tied[0] != "h.test" {
		t.Fatalf("tied = %v, want [h.test]", tied)
	}
	if hosts["h.test"].box != "annie" {
		t.Fatalf("the highest stamp still wins the dial; got %q", hosts["h.test"].box)
	}
}

// In a two-box region each box excludes itself from routing. It must
// still see that it and its neighbour publish one host at one stamp.
func TestHomeMapReportsATieWithThisBoxItself(t *testing.T) {
	files := map[string]edgesFileView{
		"control/edges/annie.json": file("annie", "1.1.1.1", map[string]edgesHostView{"h.test": {Role: "home", Stamp: 7}}),
		"control/edges/betty.json": file("betty", "2.2.2.2", map[string]edgesHostView{"h.test": {Role: "home", Stamp: 7}}),
	}
	hosts, tied := foldHomes(files, "annie", false)
	if len(tied) != 1 || tied[0] != "h.test" {
		t.Fatalf("tied = %v, want [h.test]", tied)
	}
	if hosts["h.test"].box != "betty" {
		t.Fatalf("this box is still excluded from its own routing; got %q", hosts["h.test"].box)
	}
}

// A route catch-all here does not beat another box's EXACT host: the
// precedence one box keeps (exact, then "*.<parent>") holds across boxes.
// A label nobody serves exactly still falls to the local catch-all.
func TestLocalWildcardYieldsToARemoteExactHost(t *testing.T) {
	e, m, _ := hopEdge(t, "annie", "*.example.com")
	rec, eip := fakeHomeEdge(t, nil)
	homedOn(t, m, "betty", eip, "acme.example.com")

	resp := hopRequest(t, e, "acme.example.com", nil)
	if body := readBody(t, resp); resp.StatusCode != http.StatusOK || body != homeEdgeBody {
		t.Fatalf("acme.example.com = %d %q, want the home edge's answer — another box serves it exactly", resp.StatusCode, body)
	}
	if n := rec.get().n; n != 1 {
		t.Fatalf("home edge saw %d request(s), want 1", n)
	}

	resp = hopRequest(t, e, "other.example.com", nil)
	if body := readBody(t, resp); resp.StatusCode != http.StatusOK || !strings.HasPrefix(body, "xff=") {
		t.Fatalf("other.example.com = %d %q, want the LOCAL catch-all — nobody serves it exactly", resp.StatusCode, body)
	}
	if n := rec.get().n; n != 1 {
		t.Fatalf("home edge saw %d request(s), want still 1", n)
	}
}

// ---- #336: only the hosts half is gated on the change index ------------------

// armedHomeMap is a home map over s with the change index armed (one watch over
// the same store, the agent's in production) and a clock that starts now, so
// the health state's real-time reads agree with it.
func armedHomeMap(t *testing.T, s *changeIndexStore, self string) (*homeMap, *testClock) {
	t.Helper()
	w := release.NewChangeWatch(s)
	clk := &testClock{t: time.Now()}
	m := newHomeMap(s, self)
	m.watch = func() *release.ChangeWatch { return w }
	m.now = clk.now
	return m, clk
}

func homePoll(t *testing.T, m *homeMap) {
	t.Helper()
	if err := m.poll(context.Background()); err != nil {
		t.Fatalf("poll: %v", err)
	}
}

func edgesReads(s *changeIndexStore) int {
	return s.count("List", edgesPrefix) + s.count("ListETags", edgesPrefix) + s.count("Get", edgesPrefix)
}

// codex 8: a mesh-key rotation writes control/edge-mesh-key, which bumps no
// scope, so the keys must not wait on control/edges.
func TestHomeMapKeysPolledIndependently(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	putJSON(t, s, meshKeyKey, strings.Repeat("aa", 32))
	m, clk := armedHomeMap(t, s, "annie")
	homePoll(t, m)
	if _, _, ok := m.Home("b.test"); !ok {
		t.Fatal("precondition: the first poll folds the edges files")
	}

	putJSON(t, s, meshKeyKey, strings.Repeat("bb", 32))
	s.reset()
	clk.advance(homeMapInterval)
	homePoll(t, m)
	if cur, _ := m.meshKeys(); string(cur) != strings.Repeat("\xbb", 32) {
		t.Fatalf("the rotated mesh key must load within one poll, got %x", cur)
	}
	if n := edgesReads(s); n != 0 {
		t.Fatalf("control/edges did not move, yet the poll made %d read(s) under it", n)
	}
	if n := s.count("Get", meshKeyKey); n < 1 {
		t.Fatal("the mesh key must be read on every poll")
	}
	if _, _, ok := m.Home("b.test"); !ok {
		t.Fatal("a skipped hosts read must keep the fold")
	}
}

// codex 8: the five-minute refresh heartbeat holds across polls whose hosts
// read is skipped, and the forwarding latch stays set.
func TestHomeMapHeartbeatOnUnchangedPoll(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	putJSON(t, s, meshKeyKey, strings.Repeat("aa", 32))
	m, clk := armedHomeMap(t, s, "annie")
	health := &lbhealth.State{}
	m.health = health
	homePoll(t, m)
	for i := 0; i < 4; i++ {
		s.reset()
		clk.advance(lbhealth.RefreshWindow - time.Minute)
		homePoll(t, m)
		if n := edgesReads(s); n != 0 {
			t.Fatalf("unchanged poll %d made %d read(s) under control/edges", i, n)
		}
		for _, c := range health.Failing(clk.now()) {
			if c == lbhealth.CondRefresh || c == lbhealth.CondForwarding {
				t.Fatalf("unchanged poll %d: health fails %q — a skipped hosts read must still beat", i, c)
			}
		}
	}
}

// The hosts half re-reads when control/edges moves, and at the backstop with
// the index unmoved.
func TestHomeMapHostsReadOnMoveAndAtBackstop(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m, clk := armedHomeMap(t, s, "annie")
	homePoll(t, m)

	putJSON(t, s, "control/edges/carol.json", `{"v":1,"box":"carol","eip":"3.3.3.3","hosts":{"c.test":{"role":"home"}}}`)
	s.reset()
	clk.advance(homeMapInterval)
	homePoll(t, m)
	if _, _, ok := m.Home("c.test"); !ok {
		t.Fatal("a moved control/edges must be read on the next poll")
	}
	if n := s.count("List", edgesPrefix); n != 1 {
		t.Fatalf("moved: %d list(s) of control/edges, want 1", n)
	}

	s.reset()
	clk.advance(release.ChangeIndexBackstop*9/10 - time.Minute)
	homePoll(t, m)
	if n := edgesReads(s); n != 0 {
		t.Fatalf("young and unmoved: %d read(s), want none", n)
	}
	s.reset()
	clk.advance(release.ChangeIndexBackstop*2/10 + 2*time.Minute)
	homePoll(t, m)
	if n := s.count("List", edgesPrefix); n != 1 {
		t.Fatalf("at the backstop: %d list(s) of control/edges, want 1", n)
	}
}

// Disarmed, every poll reads the hosts as today.
func TestHomeMapDisarmedPollReadsHosts(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	m := newHomeMap(s, "annie")
	m.watch = func() *release.ChangeWatch { return nil }
	for i := 0; i < 2; i++ {
		s.reset()
		homePoll(t, m)
		if n := s.count("List", edgesPrefix); n != 1 {
			t.Fatalf("disarmed poll %d: %d list(s), want 1", i, n)
		}
		if n := s.count("GetIfChanged", ""); n != 0 {
			t.Fatalf("disarmed poll %d read the index", i)
		}
	}
}

// codex 8 again: a hung index GET elsewhere in the process holds the shared
// watch's poll lock, and the home map must still finish its poll and stamp
// RefreshAttempted, or the load balancer pulls a working box after five
// minutes.
func TestHomeMapStampsRefreshWhileTheWatchIsStuck(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
	putJSON(t, s, meshKeyKey, strings.Repeat("aa", 32))
	m, clk := armedHomeMap(t, s, "annie")
	m.watchTimeout = 50 * time.Millisecond
	health := &lbhealth.State{}
	m.health = health
	homePoll(t, m)

	entered, release := s.blockIndex(t)
	go func() { _, _ = m.watch().Poll(context.Background()) }()
	<-entered
	for i := 0; i < 2; i++ {
		clk.advance(lbhealth.RefreshWindow - time.Minute)
		done := make(chan error, 1)
		go func() { done <- m.poll(context.Background()) }()
		select {
		case err := <-done:
			if err != nil {
				t.Fatalf("poll %d: %v", i, err)
			}
		case <-time.After(5 * time.Second):
			release()
			t.Fatalf("poll %d waited on the stuck watch", i)
		}
		for _, c := range health.Failing(clk.now()) {
			if c == lbhealth.CondRefresh {
				t.Fatalf("poll %d: the refresh heartbeat lapsed behind a stuck watch", i)
			}
		}
	}
}