package edge

// Task 6 of the multi-region backhaul plan: the acceptance test the whole of
// Part 3 exists for — ONE control bucket, TWO edges, and a request that lands
// in the wrong region and is served anyway.
//
// Edge A ("annie") homes app.backhaul.test and runs the tenant's app behind it.
// Edge B ("betty") has never heard of that host: no registry route, no
// certificate of its own, no way to reach the app except through A. Everything
// between the two is the bucket — the per-box edges files (Task 1's schema,
// written here directly because the publishing loop lives in internal/host),
// the mesh key (Task 2), the certs Part 2 distributes — and one TLS hop.
//
// What makes this an e2e rather than a bigger unit test is that BOTH doors are
// real and neither is stubbed:
//
//   - the client completes a real TLS handshake with the SENDING edge, with a
//     real x509 verification against the test CA, for a host that edge does not
//     route. That is the invariant the whole feature rides on and the one the
//     unit tests can only assert one branch of at a time: certMirror must
//     answer a hello for a host this box has no route for, or the client never
//     gets far enough to be backhauled at all (P1 below);
//   - the hop itself is a second real TLS handshake, from B's own transport to
//     A's front door, verified the same way.
//
// The two edges are wired exactly as RunWithActivatorReady wires one (raw local
// cache kept aside, publishing wrapper over it, mirror built from the RAW
// handle, autocert behind the mirror, the :80 chain composed in the same
// order), and every poll is DRIVEN — Refresh is called, never waited for. There
// are no sleeps in this file.
//
// Neither edge is the deployment's certificate issuer. That is deliberate and
// it is the sharp case, not a convenience: issuance is Part 2's subject, while
// what Part 3 needs is that a box which may never talk to a CA can still answer
// handshakes for hosts it does not serve. Both managers point at a dead ACME
// directory as well, so nothing here can reach a CA even by accident.
//
// Conventions are the neighbouring harnesses' (certsync_e2e_test.go for the
// two-edge shape and the e2e* naming, homemap_test.go for the hop vocabulary
// helpers): real loopback listeners, helpers reused rather than re-cut, bounds
// asserted rather than narrated. No t.Parallel anywhere — this package's tests
// swap package-level seams (backhaulPort, backhaulRootCAs) and the log writer.

import (
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"fmt"
	"io"
	"math/big"
	"net"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"golang.org/x/crypto/acme/autocert"

	"github.com/carlosframework/platform/internal/geoip"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

const (
	// e2eHomedHost is the host the whole scenario turns on: routed on edge A,
	// unknown to edge B, homed on A in every box's map.
	e2eHomedHost = "app.backhaul.test"
	// e2eMovedHost is homed on A in the MAP but no longer routed there — the
	// stale-map case, which is what a placement move looks like from the
	// sending edge for one publish cycle.
	e2eMovedHost = "moved.backhaul.test"
	// e2eDeadHost is homed on a box that answers nothing: the poisoned map.
	e2eDeadHost = "dead.backhaul.test"
	// e2eUnhomedHost is in nobody's edges file and nobody's registry — the
	// answer for it must stay the plain unknown-host 404 on every box.
	e2eUnhomedHost = "nothere.backhaul.test"
	// e2eSenderHost is edge B's OWN route, so B is a real box with traffic of
	// its own rather than a bare forwarder.
	e2eSenderHost = "local.betty.test"

	e2eHomeMarker   = "served by the home edge's own app"
	e2eSenderMarker = "served by the sending edge's own app"
)

// ---- the tenant app ----------------------------------------------------------

// e2eApp is a tenant's app behind an edge: a marker body that says WHICH box's
// upstream produced it, plus the trust-bearing headers it was handed. One
// response therefore proves both halves at once — that the bytes crossed the
// hop, and what the app was told about who called.
//
// X-Forwarded-For is rendered from Values(), joined visibly, because the
// property under test is the LENGTH of the chain as much as its content: a
// backhaul that appended the sending edge would show up here as a second
// element (or a second header line), and Get() would hide it.
func e2eApp(t *testing.T, marker string) *httptest.Server {
	t.Helper()
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintf(w, "%s|xff=%s|xfh=%s|hop=%s|auth=%s|geo=%s|hopgeo=%s", marker,
			strings.Join(r.Header.Values("X-Forwarded-For"), " + "),
			r.Header.Get("X-Forwarded-Host"),
			r.Header.Get(hopHeader), r.Header.Get(hopAuthHeader),
			r.Header.Get(geoip.HeaderCountry), r.Header.Get(geoip.HeaderHop))
	}))
	t.Cleanup(ts.Close)
	return ts
}

// e2eSeen is one parsed e2eApp response.
type e2eSeen struct {
	marker, xff, xfh, hop, auth string
	geoCountry, hopGeo          string
}

func e2eParseApp(t *testing.T, body string) e2eSeen {
	t.Helper()
	parts := strings.Split(body, "|")
	if len(parts) != 7 {
		t.Fatalf("body %q did not come from the tenant app at all", body)
	}
	seen := e2eSeen{marker: parts[0]}
	for _, p := range parts[1:] {
		k, v, _ := strings.Cut(p, "=")
		switch k {
		case "xff":
			seen.xff = v
		case "xfh":
			seen.xfh = v
		case "hop":
			seen.hop = v
		case "auth":
			seen.auth = v
		case "geo":
			seen.geoCountry = v
		case "hopgeo":
			seen.hopGeo = v
		}
	}
	return seen
}

// ---- the deployment's certificate authority ----------------------------------

// e2eCA stands in for the public CA Part 2's issuer orders from. A real CA
// rather than a self-signed leaf per host, because every handshake in this file
// is verified the way a browser verifies one — RootCAs, hostname, expiry — and
// a leaf trusted as its own root would quietly excuse a chain that production
// would reject.
type e2eCA struct {
	root *x509.Certificate
	key  *ecdsa.PrivateKey
	pool *x509.CertPool
}

func newE2ECA(t *testing.T) *e2eCA {
	t.Helper()
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	tmpl := &x509.Certificate{
		SerialNumber:          big.NewInt(1),
		Subject:               pkix.Name{CommonName: "CARLOS backhaul test CA"},
		NotBefore:             time.Now().Add(-time.Hour),
		NotAfter:              time.Now().Add(365 * 24 * time.Hour),
		KeyUsage:              x509.KeyUsageCertSign,
		BasicConstraintsValid: true,
		IsCA:                  true,
	}
	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
	if err != nil {
		t.Fatal(err)
	}
	root, err := x509.ParseCertificate(der)
	if err != nil {
		t.Fatal(err)
	}
	pool := x509.NewCertPool()
	pool.AddCert(root)
	return &e2eCA{root: root, key: key, pool: pool}
}

// publish mints a certificate for host and writes it to control/certs/<host>,
// in autocert's own cache-entry shape and through putCert — byte-for-byte what
// the issuer's publishingCache leaves in the bucket. Every edge in the
// deployment then serves it; none of them issued it.
func (ca *e2eCA) publish(t *testing.T, s release.Store, host string) *x509.Certificate {
	t.Helper()
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 96))
	if err != nil {
		t.Fatal(err)
	}
	tmpl := &x509.Certificate{
		SerialNumber: serial,
		Subject:      pkix.Name{CommonName: host},
		DNSNames:     []string{host},
		NotBefore:    time.Now().Add(-time.Hour),
		NotAfter:     time.Now().Add(90 * 24 * time.Hour),
		KeyUsage:     x509.KeyUsageDigitalSignature,
		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
	}
	der, err := x509.CreateCertificate(rand.Reader, tmpl, ca.root, &key.PublicKey, ca.key)
	if err != nil {
		t.Fatal(err)
	}
	leaf, err := x509.ParseCertificate(der)
	if err != nil {
		t.Fatal(err)
	}
	raw, err := encodeWildcardPEM(&tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key, Leaf: leaf}, key)
	if err != nil {
		t.Fatal(err)
	}
	putBucket(t, s, host, raw)
	return leaf
}

// ---- one edge ----------------------------------------------------------------

// backhaulEdge is one box: a real Edge over its OWN handle on the shared
// bucket, with the full handshake chain and the full :80 chain composed the way
// RunWithActivatorReady composes them, on real loopback listeners.
type backhaulEdge struct {
	edge   *Edge
	store  release.Store
	homes  *homeMap
	mirror *certMirror
	mgr    *autocert.Manager
	// local is the RAW autocert cache — this box's own disk, which starts
	// empty and is only ever written by the mirror.
	local autocert.Cache

	tlsAddr, plainAddr string
	// calls counts requests that reached this box's 443 door, refusals
	// included: "exactly one dial" and "never dialled at all" are both this
	// number moving, or not, on the OTHER box.
	calls atomic.Int64
	// plainCalls is the same for :80.
	plainCalls atomic.Int64
	// peerAs, when set to an address, is what this box's 443 door reports as
	// the TCP peer of every request (the port is kept). A sending box dials
	// from its own public address in production; on loopback every box is
	// 127.0.0.1, so a test that needs the peer to be that address says so
	// here. Unset, the peer is the real one.
	peerAs atomic.Value // string

	client *http.Client
}

// newBackhaulEdge builds one box. routes are host → upstream address; anything
// not listed is a host this box genuinely does not serve.
func newBackhaulEdge(t *testing.T, bucketDir, box string, ca *e2eCA, from *net.TCPAddr, routes map[string]string) *backhaulEdge {
	t.Helper()
	be := &backhaulEdge{store: release.NewDirStore(bucketDir)}
	reg := openReg(t)
	for host, addr := range routes {
		if err := reg.Upsert(registry.Route{Host: host, Kind: "service", Addr: addr, Status: "ready"}); err != nil {
			t.Fatalf("upsert %s: %v", host, err)
		}
	}
	e, err := New(reg, nil)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	t.Cleanup(e.probers.Stop)
	be.edge = e
	e.staticStore = be.store
	WithBoxLabel(box)(e)
	// Not the issuer, and never able to become one: the gate is shut and the
	// manager below points at a dead directory. Everything this box serves on
	// 443 therefore came out of the bucket.
	WithIssuerGate(func() (context.Context, bool) { return nil, false })(e)

	// The RAW cache, kept before the publishing wrapper goes on (edge.go).
	be.local = autocert.DirCache(t.TempDir())
	cache := e.armCertIssuance(be.local)
	be.mirror = e.newCertMirror(be.local)
	if be.mirror == nil {
		t.Fatal("a store-backed edge must have a mirror to serve from")
	}
	be.mgr = offlineManager(t, cache, e.issuerHostPolicy())
	tlsCfg := be.mgr.TLSConfig()
	tlsCfg.GetCertificate = be.mirror.wrapGetCertificate(tlsCfg.GetCertificate)

	be.homes = e.newHomeMap()
	if be.homes == nil {
		t.Fatal("a store-backed edge must have a home map")
	}
	e.homes.Store(be.homes)

	// 443: the edge itself, counted.
	https := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		be.calls.Add(1)
		if p, _ := be.peerAs.Load().(string); p != "" {
			_, port, _ := net.SplitHostPort(r.RemoteAddr)
			r.RemoteAddr = net.JoinHostPort(p, port)
		}
		e.ServeHTTP(w, r)
	}))
	https.TLS = tlsCfg
	https.StartTLS()
	t.Cleanup(https.Close)
	be.tlsAddr = https.Listener.Addr().String()

	// :80, outermost first, exactly as RunWithActivatorReady composes it: the
	// mirror's challenge handler, then autocert's, then the redirect fallback.
	// Composed ONCE, before anything listens, for the reason edge.go composes
	// it there — HTTPHandler is what arms autocert's HTTP-01 path.
	acmeAndFallback := be.mirror.challengeHandler(be.mgr.HTTPHandler(e.httpFallbackHandler()))
	plain := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		be.plainCalls.Add(1)
		acmeAndFallback.ServeHTTP(w, r)
	}))
	t.Cleanup(plain.Close)
	be.plainAddr = plain.Listener.Addr().String()

	be.client = e2eClientFor(be, ca, from)
	return be
}

// e2eClientFor is a client that resolves every hostname to ONE box's two
// listeners — the deployment's DNS, in other words, pointed wherever the test
// wants this request to land. https connections are handshaken here rather than
// by the transport so that the client verifies the edge's certificate against
// the test CA, by name, exactly as a browser would: a mirrored certificate that
// did not actually cover the host would fail here and not quietly downstream.
func e2eClientFor(be *backhaulEdge, ca *e2eCA, from *net.TCPAddr) *http.Client {
	dialer := &net.Dialer{Timeout: 10 * time.Second, LocalAddr: from}
	return &http.Client{
		Timeout: 30 * time.Second,
		Transport: &http.Transport{
			DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
				return dialer.DialContext(ctx, network, be.plainAddr)
			},
			DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
				name, _, err := net.SplitHostPort(addr)
				if err != nil {
					return nil, err
				}
				raw, err := dialer.DialContext(ctx, network, be.tlsAddr)
				if err != nil {
					return nil, err
				}
				conn := tls.Client(raw, &tls.Config{ServerName: name, RootCAs: ca.pool, MinVersion: tls.VersionTLS12})
				if err := conn.HandshakeContext(ctx); err != nil {
					raw.Close()
					return nil, err
				}
				return conn, nil
			},
		},
	}
}

// ---- the scene ---------------------------------------------------------------

type backhaulScene struct {
	ca   *e2eCA
	seed release.Store // the housekeeper's handle: what writes the control plane
	a, b *backhaulEdge
	key  []byte // the deployment's mesh key, as both edges mirror it
	// leaf is the certificate the bucket carries per host.
	leaf map[string]*x509.Certificate
	// clientIP is the source address the test client dials from, or "" if the
	// platform would not give us a second loopback address.
	clientIP string
	deadEIP  string
}

func newBackhaulScene(t *testing.T) *backhaulScene {
	t.Helper()
	e2eQuietLogs(t)
	bucket := t.TempDir()
	s := &backhaulScene{
		seed:    release.NewDirStore(bucket),
		ca:      newE2ECA(t),
		leaf:    map[string]*x509.Certificate{},
		deadEIP: e2eDeadEIP(t),
	}

	// The housekeeper's one-time mint (Task 2), and the certs Part 2's issuer
	// publishes. Both edges read them; neither wrote them.
	if err := MintMeshKey(context.Background(), s.seed); err != nil {
		t.Fatalf("mint mesh key: %v", err)
	}
	for _, host := range []string{e2eHomedHost, e2eMovedHost, e2eDeadHost, e2eUnhomedHost, e2eSenderHost} {
		s.leaf[host] = s.ca.publish(t, s.seed, host)
	}

	from := e2eClientAddr(t)
	if from != nil {
		s.clientIP = from.IP.String()
	}
	s.a = newBackhaulEdge(t, bucket, "annie", s.ca, from, map[string]string{
		e2eHomedHost: e2eApp(t, e2eHomeMarker).Listener.Addr().String(),
	})
	s.b = newBackhaulEdge(t, bucket, "betty", s.ca, from, map[string]string{
		e2eSenderHost: e2eApp(t, e2eSenderMarker).Listener.Addr().String(),
	})

	// The dial seams, armed for the test's lifetime: 443 is an ephemeral port
	// here, and the home edge's certificate is signed by the test CA rather
	// than a public one. Nothing about verification is weakened — the hop still
	// checks the name and the chain (see backhaulRootCAs' own doc).
	//
	// One consequence of backhaulPort being a single global: every box in this
	// scene is dialled on the SAME port, so a box's EIP is the whole of its
	// identity on the wire. That is why the dead box below gets an address of
	// its own rather than a port of its own.
	eip, port, err := net.SplitHostPort(s.a.tlsAddr)
	if err != nil {
		t.Fatalf("split home edge address: %v", err)
	}
	oldPort, oldCAs := backhaulPort, backhaulRootCAs
	backhaulPort, backhaulRootCAs = port, s.ca.pool
	t.Cleanup(func() { backhaulPort, backhaulRootCAs = oldPort, oldCAs })

	// The per-box edges files, Task 1's schema, written straight into the store
	// (the publishing loop is internal/host's and is not in this process).
	bEIP, _, err := net.SplitHostPort(s.b.tlsAddr)
	if err != nil {
		t.Fatalf("split sending edge address: %v", err)
	}
	s.publishEdgesFile(t, "annie", eip, e2eHomedHost, e2eMovedHost)
	s.publishEdgesFile(t, "betty", bEIP, e2eSenderHost)
	s.publishEdgesFile(t, "carol", s.deadEIP, e2eDeadHost)
	return s
}

// publishEdgesFile writes one box's control/edges/<box>.json — homedOn's shape,
// written through the seed handle so neither edge is the publisher of its own
// map (which is the production writer arrangement: the box's own host loop
// writes it, and every edge only ever READS the prefix).
func (s *backhaulScene) publishEdgesFile(t *testing.T, 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, s.seed, "control/edges/"+box+".json", fmt.Sprintf(
		`{"v":1,"box":%q,"eip":%q,"hosts":{%s}}`, box, eip, strings.Join(entries, ",")))
}

// converge drives every mirror and every map ONE poll forward. Explicit, never
// waited for: the loops (certMirror.run, homeMap.run) are 30s tickers, and a
// test that slept for them would be testing the clock.
func (s *backhaulScene) converge(t *testing.T) {
	t.Helper()
	for _, be := range []*backhaulEdge{s.a, s.b} {
		if err := be.mirror.Refresh(context.Background()); err != nil {
			t.Fatalf("cert mirror refresh: %v", err)
		}
		refreshHomeMap(t, be.homes)
	}
	key, next := s.a.homes.meshKeys()
	if len(key) == 0 || len(next) != 0 {
		t.Fatalf("mesh keys mirrored as current=%d next=%d bytes, want one current key", len(key), len(next))
	}
	s.key = key
}

// get drives one client request at one box and reads the whole body. follow
// says whether redirects are followed — the :80 leg wants both answers.
func (s *backhaulScene) get(t *testing.T, be *backhaulEdge, url string, headers map[string]string, follow bool) (*http.Response, string) {
	t.Helper()
	resp, err := s.try(be, url, headers, follow)
	if err != nil {
		t.Fatalf("GET %s: %v", url, err)
	}
	body, err := io.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
		t.Fatalf("read body of %s: %v", url, err)
	}
	return resp, string(body)
}

func (s *backhaulScene) try(be *backhaulEdge, url string, headers map[string]string, follow bool) (*http.Response, error) {
	req, err := http.NewRequest(http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}
	for k, v := range headers {
		req.Header.Set(k, v)
	}
	client := be.client
	if !follow {
		c := *be.client
		c.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
		client = &c
	}
	return client.Do(req)
}

// e2eClientAddr is the source address the test client dials FROM: a second
// loopback address, so the client is distinguishable ON THE WIRE from the
// sending edge, which dials the home edge from the default one. That is what
// turns "X-Forwarded-For names the client, not the edge that forwarded it" into
// an assertion rather than a coincidence of both being 127.0.0.1.
//
// nil on a platform that will not hand out a second loopback address; the
// chain-LENGTH assertions still hold there, and the identity ones are skipped
// explicitly rather than silently passing.
func e2eClientAddr(t *testing.T) *net.TCPAddr {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.2:0")
	if err != nil {
		return nil
	}
	ln.Close()
	return &net.TCPAddr{IP: net.ParseIP("127.0.0.2")}
}

// e2eDeadEIP is an address nothing listens on, for the poisoned-map case: a
// loopback one when the platform allows it, so the dial is REFUSED immediately
// rather than hanging. The documentation address is the fallback, and costs the
// backhaul dial timeout instead — the same 502, more slowly.
func e2eDeadEIP(t *testing.T) string {
	t.Helper()
	if ln, err := net.Listen("tcp", "127.0.0.9:0"); err == nil {
		ln.Close()
		return "127.0.0.9"
	}
	return "192.0.2.1" // TEST-NET-1, RFC 5737
}

// ---- the scenario ------------------------------------------------------------

func TestBackhaulE2EAcrossTwoEdges(t *testing.T) {
	s := newBackhaulScene(t)

	// Before either mirror has polled, edge B holds nothing for the homed host:
	// empty mirror, empty local cache, autocert gated shut. The handshake must
	// FAIL — otherwise every success below could be autocert's own doing, or a
	// certificate that was on B's disk all along.
	if _, err := s.try(s.b, "https://"+e2eHomedHost+"/", nil, true); err == nil {
		t.Fatal("the sending edge completed a handshake for a certificate no box has published to it yet")
	}
	s.converge(t)

	t.Run("P1 the sending edge answers a hello for a host it does not route", func(t *testing.T) {
		// The premise, asserted rather than assumed: B has no route for this
		// host, will not issue for it, and its own manager refuses it outright.
		if _, ok := s.b.edge.lookupUpstream(e2eHomedHost); ok {
			t.Fatal("test bug: the sending edge must not route the host it is supposed to backhaul")
		}
		if s.b.edge.Allowed(e2eHomedHost) {
			t.Fatal("test bug: the sending edge must not claim the host for issuance either")
		}
		if _, err := s.b.mgr.GetCertificate(ecdsaHello(e2eHomedHost)); err == nil {
			t.Fatal("the sending edge's own autocert manager must refuse a host it does not serve")
		}

		// And yet a real client, verifying against the CA by name, completes the
		// handshake — with the certificate the BUCKET carries. This is the
		// invariant the feature rides on: without it a remote-homed host dies at
		// the TLS layer on every edge but its home, and nothing downstream ever
		// runs.
		resp, _ := s.get(t, s.b, "https://"+e2eHomedHost+"/", nil, true)
		if resp.TLS == nil || len(resp.TLS.PeerCertificates) == 0 {
			t.Fatal("no peer certificate — the handshake did not happen")
		}
		if got := resp.TLS.PeerCertificates[0]; !got.Equal(s.leaf[e2eHomedHost]) {
			t.Fatalf("the sending edge served a certificate that is not the mirrored one (subject %q)", got.Subject.CommonName)
		}
		if c := s.b.mirror.certificate(e2eHomedHost, false); c == nil {
			t.Fatal("the mirror map holds no certificate for the host — it cannot have been the source")
		}
	})

	t.Run("the marker body backhauls, the chain names the client, one dial", func(t *testing.T) {
		home, sender := s.a.calls.Load(), s.b.calls.Load()
		resp, body := s.get(t, s.b, "https://"+e2eHomedHost+"/deep?q=1", nil, true)
		if resp.StatusCode != http.StatusOK {
			t.Fatalf("status = %d, want 200 — a remote-homed host is served through its home edge", resp.StatusCode)
		}
		seen := e2eParseApp(t, body)
		if seen.marker != e2eHomeMarker {
			t.Fatalf("marker = %q, want the HOME edge's app (%q)", seen.marker, e2eHomeMarker)
		}
		if got := s.a.calls.Load() - home; got != 1 {
			t.Fatalf("the home edge saw %d request(s), want exactly one — one hop, ever", got)
		}
		if got := s.b.calls.Load() - sender; got != 1 {
			t.Fatalf("the sending edge saw %d request(s) for one client request", got)
		}
		// The chain the app sees is a single element and it is the CLIENT's
		// address — not the sending edge's, which is a different loopback
		// address here precisely so the two can be told apart. Every consumer in
		// this deployment takes the last element as the caller
		// (internal/console/authcli.go's clientIP), so an appended edge would
		// bill, rate-limit and log a real visitor as infrastructure.
		if strings.Contains(seen.xff, "+") || strings.Contains(seen.xff, ",") {
			t.Fatalf("X-Forwarded-For = %q, want a single-element chain", seen.xff)
		}
		if s.clientIP != "" && seen.xff != s.clientIP {
			t.Fatalf("X-Forwarded-For = %q, want the client's own address %q", seen.xff, s.clientIP)
		}
		if seen.xfh != e2eHomedHost {
			t.Fatalf("X-Forwarded-Host = %q, want the host the client asked for", seen.xfh)
		}
		// The hop vocabulary is the edges' own and stops at the home edge.
		if seen.hop != "" || seen.auth != "" {
			t.Fatalf("hop=%q auth=%q reached the tenant's app", seen.hop, seen.auth)
		}
	})

	t.Run("the sending edge still serves its own routes without dialling", func(t *testing.T) {
		home := s.a.calls.Load()
		resp, body := s.get(t, s.b, "https://"+e2eSenderHost+"/", nil, true)
		if resp.StatusCode != http.StatusOK {
			t.Fatalf("status = %d, want 200", resp.StatusCode)
		}
		if seen := e2eParseApp(t, body); seen.marker != e2eSenderMarker {
			t.Fatalf("marker = %q, want the sending edge's OWN app", seen.marker)
		}
		if got := s.a.calls.Load() - home; got != 0 {
			t.Fatalf("the home edge saw %d request(s) for a host the sending edge serves itself", got)
		}
	})

	t.Run("the :80 leg redirects, and the redirect it hands out backhauls", func(t *testing.T) {
		// A first-time visitor typing a bare hostname lands on whichever edge
		// DNS picked, on :80. Task 5's fix: a remote-homed host is routable, so
		// it gets the ordinary redirect rather than a 404 for an app that is up.
		plain := s.b.plainCalls.Load()
		resp, _ := s.get(t, s.b, "http://"+e2eHomedHost+"/deep?q=1", nil, false)
		if resp.StatusCode != http.StatusFound {
			t.Fatalf("status = %d, want 302 — :80 must send a remote-homed host to 443", resp.StatusCode)
		}
		if got := s.b.plainCalls.Load() - plain; got != 1 {
			t.Fatalf("the sending edge's :80 door saw %d request(s), want one — this leg must be plain HTTP", got)
		}
		if got, want := resp.Header.Get("Location"), "https://"+e2eHomedHost+"/deep?q=1"; got != want {
			t.Fatalf("Location = %q, want %q", got, want)
		}
		// A host nothing in the deployment homes keeps the unchanged 404 on the
		// same handler, so the redirect is a routing answer and not a blanket one.
		if got, _ := s.get(t, s.b, "http://"+e2eUnhomedHost+"/", nil, false); got.StatusCode != http.StatusNotFound {
			t.Fatalf("status = %d, want the unchanged 404 for a host nothing homes", got.StatusCode)
		}

		// Both legs, one client call: the visitor follows the redirect and ends
		// on the home edge's app, having crossed :80 here and 443 there.
		home := s.a.calls.Load()
		final, body := s.get(t, s.b, "http://"+e2eHomedHost+"/deep?q=1", nil, true)
		if final.StatusCode != http.StatusOK {
			t.Fatalf("status = %d after following the redirect, want 200", final.StatusCode)
		}
		if final.Request.URL.Scheme != "https" {
			t.Fatalf("the followed request ended on %q, want https", final.Request.URL.Scheme)
		}
		if final.TLS == nil {
			t.Fatal("the second leg was not TLS")
		}
		if seen := e2eParseApp(t, body); seen.marker != e2eHomeMarker {
			t.Fatalf("marker = %q, want the home edge's app at the end of both legs", seen.marker)
		}
		if got := s.a.calls.Load() - home; got != 1 {
			t.Fatalf("the home edge saw %d request(s) across the two legs, want one", got)
		}
	})

	t.Run("a poisoned map is a bad gateway, and never a loop", func(t *testing.T) {
		if _, box, ok := s.b.homes.Home(e2eDeadHost); !ok || box != "carol" {
			t.Fatalf("test bug: the map must name the dead box for %s (got %q, %v)", e2eDeadHost, box, ok)
		}
		home, sender := s.a.calls.Load(), s.b.calls.Load()
		resp, body := s.get(t, s.b, "https://"+e2eDeadHost+"/", nil, true)
		if resp.StatusCode != http.StatusBadGateway {
			t.Fatalf("status = %d, want 502 — this edge believed a specific box was there", resp.StatusCode)
		}
		if got := resp.Header.Get("Retry-After"); got != "" {
			t.Fatalf("Retry-After = %q — that answer is reserved for KNOWING nobody is home", got)
		}
		if strings.Contains(body, e2eHomeMarker) {
			t.Fatal("a dead home somehow served the app")
		}
		if got := s.b.calls.Load() - sender; got != 1 {
			t.Fatalf("the sending edge handled %d request(s) for one client request — a failed dial must not be retried around the fleet", got)
		}
		if got := s.a.calls.Load() - home; got != 0 {
			t.Fatalf("the live home edge saw %d request(s) for a host it does not home — a failed dial must not scan for a second target", got)
		}
	})

	t.Run("a stale map gets the home edge's 503, and the sender does not re-dial", func(t *testing.T) {
		// The map still homes this host on A; A no longer routes it. That is
		// what one publish cycle of a placement move looks like from here.
		if _, box, ok := s.b.homes.Home(e2eMovedHost); !ok || box != "annie" {
			t.Fatalf("test bug: the map must still name the home edge for %s (got %q, %v)", e2eMovedHost, box, ok)
		}
		if _, ok := s.a.edge.lookupUpstream(e2eMovedHost); ok {
			t.Fatal("test bug: the home edge must NOT serve the moved host")
		}
		home, sender := s.a.calls.Load(), s.b.calls.Load()
		resp, body := s.get(t, s.b, "https://"+e2eMovedHost+"/", nil, true)
		if resp.StatusCode != http.StatusServiceUnavailable {
			t.Fatalf("status = %d, want the home edge's 503", resp.StatusCode)
		}
		if got := resp.Header.Get("Retry-After"); got != "2" {
			t.Fatalf("Retry-After = %q, want 2 — converge heals this within a publish cycle", got)
		}
		// The refusal reaches the client as the home edge wrote it, and says
		// nothing about where the host went.
		if strings.Contains(body, e2eMovedHost) || strings.Contains(body, "annie") || strings.Contains(body, "betty") {
			t.Fatalf("body = %q — the refusal must not name the host or any box", body)
		}
		if got := s.a.calls.Load() - home; got != 1 {
			t.Fatalf("the home edge saw %d request(s), want exactly one — a 503 is an answer, not a reason to try again", got)
		}
		if got := s.b.calls.Load() - sender; got != 1 {
			t.Fatalf("the sending edge handled %d request(s) for one client request", got)
		}
	})

	t.Run("a tampered hop reaches the home edge as an ordinary client", func(t *testing.T) {
		// Tampering happens on the wire INTO the home edge: a genuine signature
		// this deployment's own key produced, with one hex nibble flipped, plus
		// the forwarding claims a tamperer would want believed.
		tampered := []byte(signHop(s.key, e2eHomedHost, time.Now()))
		tampered[len(tampered)-1] ^= 'f' ^ '0'
		headers := map[string]string{
			hopHeader:           "1",
			hopAuthHeader:       string(tampered),
			"X-Forwarded-For":   "203.0.113.9",
			"X-Forwarded-Host":  "evil.test",
			"X-Forwarded-Proto": "https",
		}

		resp, body := s.get(t, s.a, "https://"+e2eHomedHost+"/", headers, true)
		if resp.StatusCode != http.StatusOK {
			t.Fatalf("status = %d, want 200 — a failed verification is 'you are an ordinary client', not a rejection", resp.StatusCode)
		}
		seen := e2eParseApp(t, body)
		if seen.marker != e2eHomeMarker {
			t.Fatalf("marker = %q, want the home edge's app", seen.marker)
		}
		if strings.Contains(seen.xff, "203.0.113.9") {
			t.Fatalf("X-Forwarded-For = %q — a chain nothing vouched for must be discarded", seen.xff)
		}
		if s.clientIP != "" && seen.xff != s.clientIP {
			t.Fatalf("X-Forwarded-For = %q, want the tamperer's own address %q", seen.xff, s.clientIP)
		}
		if seen.xfh == "evil.test" {
			t.Fatalf("X-Forwarded-Host = %q — an unverified sender does not get to state it", seen.xfh)
		}
		if seen.hop != "" || seen.auth != "" {
			t.Fatalf("hop=%q auth=%q reached the app — hop vocabulary is stripped either way", seen.hop, seen.auth)
		}

		// And no hop SEMANTICS at all: for a host the home edge does not serve,
		// a tampered hop gets the ordinary unknown-host answer. The same request
		// with an intact signature gets the stale-map refusal instead — the
		// signature is the whole of the difference, which is what makes the 503
		// unreachable to a stranger who cannot mint one.
		bad := map[string]string{hopHeader: "1", hopAuthHeader: string(tampered)}
		if got, _ := s.get(t, s.a, "https://"+e2eUnhomedHost+"/", bad, true); got.StatusCode != http.StatusNotFound {
			t.Fatalf("status = %d, want the unchanged 404 — a tampered hop earns no hop semantics", got.StatusCode)
		} else if ra := got.Header.Get("Retry-After"); ra != "" {
			t.Fatalf("Retry-After = %q — a stranger must not be told to come back", ra)
		}
		good := map[string]string{hopHeader: "1", hopAuthHeader: signHop(s.key, e2eUnhomedHost, time.Now())}
		if got, _ := s.get(t, s.a, "https://"+e2eUnhomedHost+"/", good, true); got.StatusCode != http.StatusServiceUnavailable {
			t.Fatalf("status = %d, want 503 — an INTACT signature is what buys the stale-map refusal", got.StatusCode)
		}
	})
}

// ---- attribution: the hop annotation, end to end -----------------------------

// The attribution half of the egress ruling, on the real two-edge scene: a
// backhauled request is metered ONCE, on the box that holds the route's
// authoritative row, and that one row also carries the hop annotation. The
// sending edge records nothing at all for the host — it has no route row and
// therefore no account or app to bill, which is exactly why the annotation
// lives here and not there.
func TestHopServedRequestMetersOnceWithHopCounters(t *testing.T) {
	s := newBackhaulScene(t)
	s.converge(t)

	home, sender := NewUsageMeter(), NewUsageMeter()
	s.a.edge.EnableUsage(home)
	s.b.edge.EnableUsage(sender)

	resp, body := s.get(t, s.b, "https://"+e2eHomedHost+"/deep?q=1", nil, true)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	if seen := e2eParseApp(t, body); seen.marker != e2eHomeMarker {
		t.Fatalf("marker = %q, want the home edge's app", seen.marker)
	}

	rows := home.Drain()
	row, ok := rows[e2eHomedHost]
	if !ok {
		t.Fatalf("the home edge metered nothing for %s: %+v", e2eHomedHost, rows)
	}
	if row.Requests != 1 {
		t.Fatalf("row = %+v, want exactly one request — a hop is ONE metered exchange, not two", row)
	}
	if row.BytesOut == 0 {
		t.Fatalf("row = %+v, want the wire bytes the home edge sent", row)
	}
	if row.HopRequests != 1 || row.HopBytesOut != row.BytesOut {
		t.Fatalf("row = %+v, want the whole exchange annotated as hopped (hop_bytes_out == bytes_out)", row)
	}
	if row.HopBytesOut > row.BytesOut || row.HopRequests > row.Requests {
		t.Fatalf("row = %+v — the hop counters are a subset of the totals", row)
	}

	// The sending edge is meterless for this host BY DESIGN: no route row, no
	// identity, so anything it wrote would be unattributed — and a second
	// Record for one exchange is the double count this arrangement refuses.
	if got, ok := sender.Drain()[e2eHomedHost]; ok {
		t.Fatalf("the sending edge metered %+v for a host it has no route for", got)
	}
}

// ---- the single-box regression pin -------------------------------------------

// Everything above is dormant on a deployment with one platform box, and this
// is the pin that says so: an edge whose control/edges/ prefix holds only its
// OWN file serves its own routes exactly as it always has, answers the
// unchanged 404 for everything else, and never opens a connection to anything.
//
// The witness is a zero-call dialer — a listener that counts every connection
// made to the backhaul port — with the single box's own edges file naming the
// listener's ADDRESS as its EIP. So the one failure this pin exists to catch,
// an edge that offers itself as a backhaul target, lands ON the counter rather
// than merely failing somewhere else.
//
// The file lists a host the box does not route, which is what makes the counter
// a real witness rather than a formality: a route removed between two publishes
// leaves exactly that state for one cycle, it is the ONLY shape in which a
// single box could dial at all (a host it serves is served locally, before the
// map is ever consulted), and dropping own-box exclusion turns it into a dial to
// this box's own front door.
func TestBackhaulE2ESingleBoxNeverDials(t *testing.T) {
	e2eQuietLogs(t)
	bucket := t.TempDir()
	seed := release.NewDirStore(bucket)
	if err := MintMeshKey(context.Background(), seed); err != nil {
		t.Fatalf("mint mesh key: %v", err)
	}

	const soloHost = "solo.backhaul.test"
	// A host this box's own file still claims but its registry no longer
	// serves — one publish cycle after a route was removed.
	const soloStaleHost = "stale.backhaul.test"
	app := e2eApp(t, e2eHomeMarker)
	reg := openReg(t)
	if err := reg.Upsert(registry.Route{Host: soloHost, Kind: "service", Addr: app.Listener.Addr().String(), Status: "ready"}); err != nil {
		t.Fatalf("upsert: %v", err)
	}
	e, err := New(reg, nil)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	t.Cleanup(e.probers.Stop)
	e.staticStore = release.NewDirStore(bucket)
	WithBoxLabel("solo")(e)

	eip, dials := e2eDialCounter(t)
	putJSON(t, seed, "control/edges/solo.json", fmt.Sprintf(
		`{"v":1,"box":"solo","eip":%q,"hosts":{%q:{"role":"home"},%q:{"role":"home"}}}`,
		eip, soloHost, soloStaleHost))

	m := e.newHomeMap()
	refreshHomeMap(t, m)
	e.homes.Store(m)
	for _, host := range []string{soloHost, soloStaleHost} {
		if _, _, ok := m.Home(host); ok {
			t.Fatalf("%s: the only box in a deployment must never be offered as a backhaul target for its own host", host)
		}
	}

	if resp := hopRequest(t, e, soloHost, nil); resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200 — a single box serves its own routes as before", resp.StatusCode)
	} else if seen := e2eParseApp(t, readBody(t, resp)); seen.marker != e2eHomeMarker {
		t.Fatalf("marker = %q, want the box's own app", seen.marker)
	}
	for _, host := range []string{soloStaleHost, "nothing.backhaul.test"} {
		if resp := hopRequest(t, e, host, nil); resp.StatusCode != http.StatusNotFound {
			t.Fatalf("%s: status = %d, want the unchanged 404", host, resp.StatusCode)
		}
	}
	if n := dials.Load(); n != 0 {
		t.Fatalf("%d backhaul dial(s) from a single-box deployment, want none", n)
	}
	n := 0
	e.backhaulTransports.Range(func(_, _ any) bool { n++; return true })
	if n != 0 {
		t.Fatalf("%d backhaul transport(s) built, want none — nothing was ever a target", n)
	}
}

// e2eDialCounter is the zero-call witness: a listener on the backhaul port that
// counts connections and answers nothing. It returns the address a dial would
// have to name to reach it, and arms backhaulPort for the test's lifetime.
func e2eDialCounter(t *testing.T) (string, *atomic.Int64) {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	t.Cleanup(func() { ln.Close() })
	var n atomic.Int64
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			n.Add(1)
			conn.Close()
		}
	}()
	host, port, err := net.SplitHostPort(ln.Addr().String())
	if err != nil {
		t.Fatalf("split: %v", err)
	}
	old := backhaulPort
	backhaulPort = port
	t.Cleanup(func() { backhaulPort = old })
	return host, &n
}