package edge

// Issuer-side certificate publishing (spec:
// docs/superpowers/specs/2026-08-20-multi-region-ingress-design.md, Part 2).
//
// With more than one edge in a deployment, exactly ONE of them talks to the
// CA — the housekeeper (internal/host/housekeeper.go) — and every artifact
// that issuance produces is published to the control bucket under
// control/certs/ so every OTHER edge can serve it. Two halves live here:
//
//   - publishingCache wraps the autocert.Cache the whole issuance path
//     already funnels through (certcache.go's seam, reserved for exactly
//     this). autocert writes its ACME account key, every per-host cert+key
//     PEM, and every HTTP-01 token response through its Cache — cluster
//     support is that interface's documented purpose — and wildcard.go's
//     DNS-01 manager shares the same handle. Wrapping it therefore captures
//     per-host certs, wildcards, tokens and account keys with no autocert
//     fork and no second abstraction.
//
//   - the issuance gate: HostPolicy (per-host HTTP-01) and the wildcard
//     manager's order path (DNS-01) both refuse to START work on a box that
//     is not the housekeeper. Serving is never gated: an edge serves from
//     its local cache and from the mirror of this prefix regardless.
//
// Cert writes never bump control/serial — the serial is the console's
// (writer map, Part 1); the mirror side polls this prefix on its own
// cadence. Renewals run ~30 days before expiry, so a poll cadence measured
// in seconds is never on a deadline.

import (
	"bytes"
	"context"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"log"
	"math/rand/v2"
	"net"
	"net/http"
	"os"
	"path"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"time"

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

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

// certPrefix is where every issuance artifact lands in the control bucket.
// One prefix for the lot, named by the cache key autocert (or wildcard.go)
// chose, so the mirror can rebuild a whole cache from a single list — and so
// the fleet door's ACL has exactly one prefix to keep denying.
const certPrefix = "control/certs/"

// certKey is the bucket key for cache entry name. The cache name goes
// through verbatim: "example.com", "example.com+rsa", "acme_account+key",
// "<token>+http-01", "wildcard-<parent>.pem", "wildcard-account.key".
func certKey(name string) string { return certPrefix + name }

// IssuerGate answers "may this process start certificate issuance, and under
// what authority does that work run?" — the housekeeper's gate
// (internal/host), handed DOWN at edge construction because internal/host
// imports internal/edge and never the reverse. The context is cancelled the
// instant this box stops being the housekeeper, so an order in flight
// unwinds with the authority it started under.
//
// A nil gate means "ungated": `carlos edge` run without the agent, and every
// test, has no election in the process at all, and refusing issuance there
// would leave the box with no certificates rather than with today's
// behaviour.
type IssuerGate func() (context.Context, bool)

// issuerGate is the Edge's read of its gate, with the ungated default
// applied. See the type's doc for why nil means "yes".
func (e *Edge) issuerGate() (context.Context, bool) {
	if e.issuer == nil {
		return context.Background(), true
	}
	return e.issuer()
}

// errNotIssuer is what HostPolicy answers on a box that is not the
// deployment's certificate issuer. Sentinel (errors.Is-able) so the gate is
// distinguishable from "that host isn't mine", which is a different fault
// with a different fix.
//
// The message text is deliberately generic, not diagnostic: autocert's own
// HTTPHandler writes this exact string, verbatim, as the body of an
// unauthenticated :80 response (http.Error(w, err.Error(), 403) at
// autocert.go's HTTPHandler) — OUR code never gets a chance to intercept or
// reword it before it reaches the wire. "acme: this box is not the
// deployment's certificate issuer" told any anonymous prober both that
// multi-edge issuance exists and which box is not the issuer; the identity
// that matters everywhere else in this file is errors.Is(err, errNotIssuer),
// never the string, so softening it costs nothing internal.
var errNotIssuer = errors.New("acme: not available here")

// armCertIssuance wires cert DISTRIBUTION and the issuance GATE together,
// because they are one decision and not two: a box may only be told "you are
// not the issuer" if some other box's certificates can reach it.
//
// WITH a deployment store: every issuance artifact publishes to
// control/certs/ and the gate applies — one elected issuer mints, every
// other edge serves the published copy.
//
// WITHOUT one there is nothing to publish to and, just as importantly,
// nothing to mirror FROM. The gate is cleared and the box issues for itself
// exactly as it always has. Three real box classes land here, and gating any
// of them would leave it permanently certificate-less — no new orders AND no
// serving out of its own on-disk cache, since autocert consults HostPolicy
// before its cache lookup:
//
//   - fleet boxes (CARLOS_FLEET) carrying no deployment bucket of their own:
//     the housekeeper never contends there, and there is nothing to mirror
//     from. The fleet store DOOR is not what decides this and never was: the
//     edge does not use the door at all — enableStaticFromEnv reads
//     CARLOS_DEPLOYMENT_BUCKET / CARLOS_DEPLOYMENT_DIR and opens the store
//     directly, and the door is the agent's client. A grandfathered fleet box
//     that still carries the bucket env therefore reads control/certs/ over
//     its own IAM, and is gated AND mirrored like any other non-issuer — a
//     transition state, resolved by the rollout dropping that env. It also
//     never BACKFILLS (#318): mayBackfill requires contending in the
//     election, and a fleet box's housekeeper mode is "never" — its disk
//     holds a customer's certs, not this deployment's;
//   - boxes with neither CARLOS_DEPLOYMENT_BUCKET nor CARLOS_DEPLOYMENT_DIR;
//   - boxes with a bucket but no CARLOS_RELEASE_PUBKEY, where
//     enableStaticFromEnv leaves staticStore nil.
//
// Clearing the field (rather than only skipping the HostPolicy wrap) is what
// keeps the wildcard manager symmetric too: newWildcardFromEnv reads the
// same gate a few lines later.
func (e *Edge) armCertIssuance(inner autocert.Cache) autocert.Cache {
	if e.staticStore == nil {
		e.issuer = nil
		log.Printf("edge: cert publishing off (no deployment store configured) — this box issues and keeps its own certs, as before")
		return inner
	}
	log.Printf("edge: publishing issued certs to %s* in the deployment bucket", certPrefix)
	return &publishingCache{inner: inner, s: e.staticStore, issuer: e.issuerGate}
}

// issuerHostPolicy is the autocert HostPolicy: the registry check this edge
// has always applied, plus the issuance gate.
//
// The registry check is deploymentHost (edge.go) rather than Allowed: a
// deployment's issuer must be able to mint for any EXACTLY-named route in the
// deployment, not only the ones this box happens to home, or a home-boxed
// route whose hellos reach the issuer through the backhaul can never get a
// certificate at all (#277). Exact-only, so PR #180's rule is untouched: a
// label under a catch-all still drives no order. The errNotIssuer gate below
// is unaffected — non-issuers still order nothing, for any name.
//
// Two autocert call sites reach this, and both are deliberate:
// GetCertificate consults the policy BEFORE its cache lookup
// (autocert.go:292 in x/crypto v0.52.0), and HTTPHandler consults it before
// serving an HTTP-01 token (autocert.go:405). On a non-issuer both therefore
// decline — which is correct only because the mirror layer sits behind them
// and serves both certs and challenges from control/certs/. Nothing here may
// be relaxed into "allow if we have it cached": autocert treats an EXPIRED
// cached cert as a miss and would go straight to an order, so a policy that
// says yes for a cached name is a policy that lets a non-issuer order.
//
// It does NOT stop renewals, and nothing here should be read as claiming it
// does: autocert's renewal timers (renewal.go's domainRenewal.do) never
// consult HostPolicy. A timer started while this box held the lease keeps
// firing after a stand-down; each firing re-checks the cache first, and only
// orders when the cache has nothing fresh. Two consequences, both landing on
// Task 5: the mirror MUST write fresh certs into this box's LOCAL cache, so
// that cache-first check short-circuits the timer before it orders; and
// until it does, a stood-down box's timer fails HTTP-01 against our own
// gated handler every ~30 minutes, burning the shared account's
// failed-validation budget. (A tls-alpn-01 validation would even succeed —
// the token-cert branch at autocert.go:288 precedes the policy check — and
// then the gated publisher declines to publish the result.)
//
// The flag is not a fence. Under a hung lease renewal two boxes can
// transiently both believe they hold — the housekeeper's own stand-down is
// bounded by its 30s poll, and a lost renewal is only noticed on that poll.
// Brief double-issuance against the shared ACME account is inside the
// spec's accepted envelope (one account, one rate-limit bucket, renewals ~30
// days early), so there is deliberately NO fencing token here and no
// readiness predicate on the election: a fence would trade a rare duplicate
// order for a deployment that can stop issuing altogether.
func (e *Edge) issuerHostPolicy() autocert.HostPolicy {
	return func(_ context.Context, host string) error {
		if !e.deploymentHost(host) {
			return fmt.Errorf("acme: host %q not in registry", host)
		}
		if _, ok := e.issuerGate(); !ok {
			return errNotIssuer
		}
		return nil
	}
}

// publishingCache is the autocert.Cache every issuance artifact goes
// through: the local cache first (durability, and autocert's own read path
// keeps working byte-for-byte), then a best-effort copy into the control
// bucket.
//
// Bucket failures are LOUD and never returned: a returned error aborts an
// issuance that has already succeeded locally, trading a cert this box could
// have served for a bucket blip. The mirror's next poll (and, failing that,
// the next renewal) re-publishes.
type publishingCache struct {
	inner autocert.Cache
	s     release.Store
	// issuer gates the BUCKET half only — the local cache is written
	// unconditionally, because a non-issuer must still be able to serve what
	// it was handed. nil = ungated (see IssuerGate).
	//
	// It is the gate's boolean that is used and not its context: publishing
	// is not an order, and cancelling a publish because the holding ended
	// mid-write would strand a freshly minted cert on one box's disk with
	// nothing scheduled to re-publish it. The write runs on its own budget
	// (publishCtx) instead.
	issuer IssuerGate
}

// certPublishTimeout budgets ONE bucket publish or delete. It exists because
// autocert calls this cache from under its own locks: deleteHTTPToken and
// deleteCertToken call Cache.Delete while HOLDING m.challengeMu for write,
// and both httpToken (the :80 challenge handler) and GetCertificate's
// tls-alpn-01 branch read under that same mutex. An unbudgeted bucket call
// there wedges challenge serving and ALPN handshakes on the issuer for as
// long as the network takes. The Put path is no better off inheriting
// autocert's own 5-10 minute order deadline.
//
// A var, not a const, so tests can shorten it.
var certPublishTimeout = 10 * time.Second

// publishCtx budgets one bucket call, detached from the caller's
// cancellation. WithoutCancel is deliberate: an order that has already been
// abandoned still has to clean up its challenge token, and a cancelled
// context would skip exactly the delete that prevents litter every edge
// would go on to mirror and serve.
func publishCtx(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.WithoutCancel(ctx), certPublishTimeout)
}

func (p *publishingCache) Get(ctx context.Context, key string) ([]byte, error) {
	return p.inner.Get(ctx, key)
}

// Put writes locally, then publishes. Local first: if the disk write fails
// there is no cert, and publishing one the box itself cannot serve would be
// a lie in the bucket.
func (p *publishingCache) Put(ctx context.Context, key string, data []byte) error {
	if err := p.inner.Put(ctx, key, data); err != nil {
		return err
	}
	if !p.publishes(key) {
		return nil
	}
	pctx, cancel := publishCtx(ctx)
	defer cancel()
	if err := putCert(pctx, p.s, key, data); err != nil {
		log.Printf("certsync: bucket publish failed for %q: %v — the cert is live on this box; the issuer's next renewal republishes", key, err)
	}
	return nil
}

// Delete removes both copies. The bucket copy goes even when the local
// delete fails: the entry autocert deletes is a spent HTTP-01 token, and a
// token left in a prefix every edge mirrors is litter every edge would keep
// serving.
//
// It is gated like Put, and that is a deliberate trade: an issuer that
// stands down mid-order leaves its token behind in the bucket, because
// ISSUANCE artifacts stay the housekeeper's to write and un-write (#318
// widened only the backfill of settled cert entries, never tokens), and a
// box that is no longer the housekeeper does not get to delete there. The
// spec's answer to that litter is a TTL stamp swept from the mirror side,
// not a second writer.
func (p *publishingCache) Delete(ctx context.Context, key string) error {
	err := p.inner.Delete(ctx, key)
	if !p.publishes(key) {
		return err
	}
	pctx, cancel := publishCtx(ctx)
	defer cancel()
	if derr := p.s.Delete(pctx, certKey(key)); derr != nil {
		log.Printf("certsync: bucket delete failed for %q: %v — a spent challenge token may linger until it is overwritten", key, derr)
	}
	return err
}

// publishes reports whether this box should write key to the bucket: only
// the issuer publishes, and only names that survive the round trip through a
// bucket key do.
func (p *publishingCache) publishes(key string) bool {
	if p.s == nil {
		return false
	}
	if p.issuer != nil {
		if _, ok := p.issuer(); !ok {
			// Not the issuer. Silent, like every other not-held path in this
			// deployment: it is the permanent normal state of N-1 boxes.
			return false
		}
	}
	if !publishableCertName(key) {
		log.Printf("certsync: refusing to publish cache entry %q — the name does not round-trip as a bucket key", key)
		return false
	}
	return true
}

// publishableCertName rejects cache names that would not survive as
// "control/certs/<name>": a "/" would put the entry outside the flat prefix
// the mirror maps back onto cache names, and a traversal segment has no
// business in a key derived from a hostname. Nothing autocert or wildcard.go
// generates fails this (autocert's HTTP-01 key is path.Base + "+http-01",
// and domains are punycode-validated before they reach a cache key); it is
// here so that a future key shape fails visibly rather than scattering
// objects through the control prefix.
func publishableCertName(name string) bool {
	if name == "" || strings.Contains(name, "/") {
		return false
	}
	for _, seg := range []string{"..", "\\"} {
		if strings.Contains(name, seg) {
			return false
		}
	}
	return true
}

// putCert writes key's body regardless of prior state: create-only when
// absent, etag-replace when present, re-reading on a lost race — the
// putReplacing idiom from internal/host's snapshot loop, needed here for the
// same reason (release.Store.Put with an empty ifMatch is create-only, so a
// renewal would otherwise lose to its own previous cert).
//
// Last-writer-wins is the DESIGN here, not a compromise. On the issuance
// path, the transient double-holder the housekeeper's poll window allows
// writes an equally valid cert for the same name, and whichever lands last
// is servable. On the backfill path (#318: any contending box), a writer
// racing a stale view can land an OLDER leaf — repaired, not prevented:
// fileStamp's bucket-identity half un-settles the box holding the newer
// one, whose next pass republishes it. A freshness check inside this
// function would still race (check-then-put), so the repair loop is the
// guarantee, and it converges because the NotAfter compare is monotone.
func putCert(ctx context.Context, s release.Store, name string, body []byte) error {
	return putCAS(ctx, s, certKey(name), body)
}

// putCAS is putCert's write, by key rather than by cert name — the same
// read-etag-then-If-Match loop, shared so the cert warm's outcome records
// (#299 §3e, certwarm.go) get the identical concurrency posture instead of a
// second copy of it that can drift.
func putCAS(ctx context.Context, s release.Store, key string, body []byte) error {
	for attempt := 0; attempt < 3; attempt++ {
		_, etag, err := s.Get(ctx, key)
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = ""
		case err != nil:
			return fmt.Errorf("get %s: %w", key, err)
		}
		if _, err := s.Put(ctx, key, body, etag); err == nil {
			return nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return fmt.Errorf("put %s: %w", key, err)
		}
	}
	return fmt.Errorf("%s: lost CAS race three times", key)
}

// ---- the mirror: every edge serves every cert --------------------------------
//
// certMirror is the serving half of the pair at the top of this file, and the
// reason the issuance gate above is safe to shut. It POLLS control/certs/
// (its serving side never writes there; the one writer it carries is the
// backfill below, bounded to cert entries this box owns — #318) and keeps
// the whole prefix in memory, so any edge can answer a
// handshake for a name some OTHER edge issued, and can answer an HTTP-01
// challenge for an order some OTHER edge is running.
//
// Three consumers, in the order a request meets them:
//
//   - GetCertificate (wrapGetCertificate, wildcard.go): the mirror is the
//     PRIMARY serving path on a non-issuer, ahead of autocert, because
//     autocert's HostPolicy runs BEFORE its cache lookup — a gated manager
//     serves nothing at all, cached or not.
//   - the :80 challenge handler (challengeHandler): in FRONT of autocert's
//     HTTPHandler for the same reason — that handler 403s a non-issuer before
//     it ever looks a token up.
//   - the LOCAL autocert cache: every refresh copies fresh cert material into
//     it. autocert's renewal timers never consult HostPolicy (see
//     issuerHostPolicy), and renewal.go's do() checks the cache FIRST, so a
//     mirror-refreshed local cache is what stops a stood-down box's stale
//     timers from ordering — and what carries a box through the boot and
//     crash windows where the election has not yet been decided.
//
// Account keys are the one thing that is mirrored but never served: they are
// issuer material and have no business in a map a handshake can reach.

// certMirrorInterval is the poll cadence. Renewals run ~30 days before
// expiry, so nothing on this path is on a deadline — except live HTTP-01
// validations, which the poll deliberately does NOT try to catch (see
// challengeResponse's read-through).
var certMirrorInterval = 30 * time.Second

// certMirrorTimeout budgets one whole poll: list plus the gets. Generous,
// because it runs on its own goroutine and blocks nothing; bounded, because a
// stalled bucket must not wedge the loop forever.
var certMirrorTimeout = 2 * time.Minute

// certMirrorBootTimeout budgets the ONE refresh that runs before the
// listeners bind. Much shorter than a poll's: a stalled bucket must delay the
// front door by seconds, not minutes, and the box still has its own cache to
// serve from while the loop catches up.
var certMirrorBootTimeout = 20 * time.Second

// certRefreshConcurrency bounds the in-flight gets in Refresh. It is a
// latency fix, not a throughput one: the pass costs
// ceil(count/concurrency) x latency instead of count x latency, which at
// 101 certs and a 250ms round trip is about 2s instead of 25s.
//
// Ten, not sixteen, because the AWS SDK's default transport keeps ten idle
// connections per host (DefaultHTTPTransportMaxIdleConnsPerHost) and nothing
// here overrides it. Sixteen in flight means six connections closed at the
// end of every pass and re-handshaked on the next one, forever, on a 30s
// poll — about a dozen extra cross-region TLS handshakes a minute for one
// extra wave's worth of latency.
const certRefreshConcurrency = 10

// certSnapshot is one poll's result, swapped in whole. Immutable once
// published, so readers hold no lock beyond the pointer read.
type certSnapshot struct {
	// entries is every mirrored name → body, account keys included (the local
	// sync needs them; Cert refuses to serve them).
	entries map[string][]byte
	// certs is the parsed form of every cert-shaped entry, keyed by the same
	// name. Parsing happens ONCE per poll per changed entry, never per
	// handshake. A name whose body does not parse is absent, not nil.
	certs map[string]*tls.Certificate
}

var emptyCertSnapshot = &certSnapshot{entries: map[string][]byte{}, certs: map[string]*tls.Certificate{}}

type certMirror struct {
	s release.Store
	// local is this box's OWN autocert cache — the raw one, NOT the
	// publishing wrapper: a mirror write is a copy inwards, and pushing it
	// back out to the bucket would have the issuer republishing its own
	// material on every poll.
	local autocert.Cache

	// mayBackfill reports whether this box may publish its disk's certs at
	// all — the backfill's switch (see backfill): true on a box run under an
	// agent (whose election makes it an accountable member of the
	// deployment), false on a hand-run `carlos edge` and nil — never — for a
	// mirror built without an Edge.
	mayBackfill func() bool
	// issuing reports whether this box holds the issuer gate RIGHT NOW —
	// what remains single-writer inside backfill: seeding the ACME account
	// key (#318 widened cert publishing to every agent-run box). nil = never
	// the issuer, the safe default for a mirror built without an Edge.
	issuing func() bool
	// wildcardParents is the wildcard manager's current parent set, the
	// bound on which wildcard PEMs backfill may publish. nil = no manager =
	// none.
	wildcardParents func() []string
	// settled memoes the stamp each on-disk entry was last found settled at
	// (backfill); backfillFailures counts failed passes for the log
	// cadence; backfillOnce guards the one-time "cannot walk" line.
	settled          map[string]fileStamp
	backfillFailures int
	backfillOnce     sync.Once

	// wake is called after a poll that changed wildcard material — the
	// wildcard manager's refresh (R4), so a takeover propagates on the poll
	// instead of on the manager's own 12h tick. nil when wildcards are off.
	wake func()

	// allowed reports whether a hostname is one this box serves, or one
	// another box in this deployment exactly homes (Edge.deploymentHost) —
	// the bound on BOTH read-throughs, whose keys are attacker-chosen SNI and
	// an attacker-chosen Host header. Exact names only, either side: a label
	// under a catch-all spends neither budget (#180, and KnownHost's doc).
	// nil = unbounded, which is tests only.
	allowed func(string) bool

	// gated reports whether issuance is currently BARRED on this box, which
	// is what decides where the mirror sits in the handshake chain (see
	// wrapGetCertificate). Read per hello, because the election moves under a
	// running process. nil = treat as gated, the safe default: it serves the
	// same certificates either way, it only costs an issuer its lazily-armed
	// renewal timers, and a mirror built without a gate (tests) has no
	// issuance to arm.
	gated func() bool

	// refreshMu serializes Refresh's whole body. Production has exactly one
	// caller today (the 30s loop, run()), so this is insurance rather than a
	// live fix: two overlapping polls (a second caller added later, or a
	// boot-time refresh racing the loop's first tick) could otherwise both
	// read m.snapshot() as prev, build independently, and publish out of
	// order — the SECOND Store to finish under m.mu.Lock() wins regardless
	// of which list was actually more recent, so a slow poll could clobber a
	// fresher one. Locking the body makes Refresh calls queue and apply in
	// the order they were made.
	refreshMu sync.Mutex

	mu   sync.RWMutex
	snap *certSnapshot
	// partialPolls counts polls that could not read every object, for the
	// first-and-every-tenth log cadence.
	partialPolls int

	// health is the load balancer health state the first landed refresh
	// latches (region HA spec 2 §6); nil without region HA.
	health *lbhealth.State

	// watch resolves the process's change-index watch (#336), handed down by
	// the agent through WithChangeWatch and read on every poll: the agent
	// arms its watch on its own schedule, possibly after the edge starts.
	// nil, or a nil watch, is today's mirror: every poll a full Refresh.
	// now is the poll's clock (nil is time.Now; tests inject one).
	watch func() *release.ChangeWatch
	now   func() time.Time
	// watchTimeout bounds the poll of the watch (0 is
	// release.ChangeIndexPollTimeout; tests shorten it).
	watchTimeout time.Duration

	// Guarded by refreshMu: what the published snapshot was read at.
	// etags is each entry's ETag as its Get returned it ("" = unknown, read
	// it again); lastFull is the last full rebuild that landed and fullAge
	// the jittered backstop after which the next poll rebuilds in full;
	// actedGen is the control/certs generation, captured BEFORE the read,
	// that the snapshot is known to reflect (actedOK false = no such
	// generation: the watch could not vouch, or some object went unread).
	etags    map[string]string
	lastFull time.Time
	fullAge  time.Duration
	actedGen uint64
	actedOK  bool
	// localPending: the last copy into the local cache left entries it
	// could not write; an unchanged poll retries it from the snapshot.
	localPending bool

	// synced remembers, per name, the mirrored bytes last reconciled into the
	// local cache, so a steady state costs no local I/O at all.
	syncMu sync.Mutex
	synced map[string][]byte

	// rt is the read-through memo (hits and misses), its in-flight set, and
	// the global token bucket that bounds live bucket reads.
	rtMu       sync.Mutex
	rt         map[string]readThroughEntry
	rtInflight map[string]chan struct{}
	rtHTTP     rtBudget // :80, the HTTP-01 challenge path
	rtALPN     rtBudget // 443, the tls-alpn-01 token-cert path
	lastRTLog  time.Time

	// localCerts memoes the last-resort reads of this box's own cache;
	// lastLocalLog throttles the line that says the mirror had nothing.
	localMu      sync.Mutex
	localCerts   map[string]localCertEntry
	lastLocalLog time.Time
}

// newCertMirror is the Edge's mirror, or nil when there is nothing to mirror
// FROM. The nil case is exactly armCertIssuance's no-store case — the two
// decisions are the same decision, and a box that lands there is ungated and
// serves its own cache through autocert as it always has.
func (e *Edge) newCertMirror(local autocert.Cache) *certMirror {
	if e.staticStore == nil {
		return nil
	}
	m := newCertMirror(e.staticStore, local)
	m.health = e.lbHealth
	m.watch = e.changeWatch
	m.gated = func() bool {
		_, held := e.issuerGate()
		return !held
	}
	// Both read-throughs are keyed by remote input (SNI on 443, the Host
	// header on :80), so they are bounded to the names this DEPLOYMENT serves:
	// this box's own routes, plus the exact names another box's edges file
	// homes. deploymentHost is the same exact-match, registry-backed check
	// autocert's HostPolicy uses (issuerHostPolicy, above), which is the right
	// bound for the same reason it always was — a per-host order only ever
	// exists for a name that passed it — now that the policy admits the
	// deployment's remote-homed routes too (#277).
	m.allowed = e.deploymentHost
	// Backfill runs on any box that CONTENDS in the deployment's issuer
	// election (#318): contending is what makes the box an accountable
	// member of the deployment, whether or not it currently holds the
	// lease. Two box classes are excluded, each for its own reason: a
	// hand-run `carlos edge` with a bucket but no agent is "held" for
	// serving and must NOT dump its disk into the deployment's prefix
	// (e.issuer nil, and no contender predicate was ever handed down); and
	// a grandfathered FLEET box still carrying the bucket env is agent-run
	// with a gate func, but its housekeeper mode is "never" — its disk
	// holds a customer's certs, which are not this deployment's to
	// replicate onto every platform edge (WithIssuerContender, and the
	// armCertIssuance doc's grandfathered-fleet-box paragraph).
	m.mayBackfill = func() bool { return e.issuer != nil && e.contends != nil && e.contends() }
	// Strict: only the box holding the lease RIGHT NOW may seed the
	// deployment's ACME account identity (see backfill's account-key rule).
	m.issuing = func() bool {
		if e.issuer == nil {
			return false
		}
		_, held := e.issuer()
		return held
	}
	return m
}

// attachWildcardToMirror is the mirror↔wildcard-manager wiring, in one place
// so a test can see it: a wildcard the ISSUER minted reaches this box's
// handshakes on the next poll rather than on the manager's 12h tick (wake),
// and the manager's current parent set bounds which wildcard PEMs the
// issuer's backfill may publish. Either side nil is a no-op.
func attachWildcardToMirror(m *certMirror, w *wildcardManager) {
	if m == nil || w == nil {
		return
	}
	m.wake = w.refresh
	m.wildcardParents = w.currentParents
}

func newCertMirror(s release.Store, local autocert.Cache) *certMirror {
	return &certMirror{
		s:          s,
		local:      local,
		snap:       emptyCertSnapshot,
		synced:     map[string][]byte{},
		rt:         map[string]readThroughEntry{},
		rtInflight: map[string]chan struct{}{},
		localCerts: map[string]localCertEntry{},
	}
}

// Refresh rebuilds the whole map from one list plus a get per key. It is the
// boot read before the listeners bind, every poll while the change index is
// disarmed, and the armed loop's backstop and "unknown" read (poll, #336).
// A rebuild from the listing is what makes a withdrawn entry (a spent
// challenge token) actually leave; poll's changed-only read builds from a
// listing too, for the same reason.
//
// An error leaves the last-good snapshot in place and is returned for the
// caller to log. That is the edge's standing degrade-not-die posture, and it
// matters more here than usual: an empty map on a bucket blip would be a TLS
// outage on every non-issuer in the deployment.
func (m *certMirror) Refresh(ctx context.Context) error {
	m.refreshMu.Lock()
	defer m.refreshMu.Unlock()
	return m.refreshFull(ctx)
}

// refreshFull is Refresh's body, under refreshMu.
func (m *certMirror) refreshFull(ctx context.Context) error {
	// The generation this read will reflect, captured BEFORE the list: a
	// write that lands during the read leaves control/certs moved past it,
	// so the next poll looks again (#336). Without a usable watch there is
	// none, and the next poll cannot skip.
	gen, genOK := m.certsGen()
	keys, err := m.s.List(ctx, certPrefix)
	if err != nil {
		return fmt.Errorf("list %s: %w", certPrefix, err)
	}
	prev := m.snapshot()
	next := make(map[string][]byte, len(keys))
	etags := make(map[string]string, len(keys))

	// The eligible names first, then their bodies CONCURRENTLY. The fetch is
	// one round trip per object and the objects are independent, so doing
	// them one after another makes the whole pass cost
	// count x latency — which is fine beside the bucket and is not fine
	// across the planet.
	//
	// That is not hypothetical (amadan #118 thread, diagnosed 2026-09-19).
	// The Sydney box reads this deployment's bucket in eu-west-1 at ~250ms a
	// round trip. At 101 published certs the serial pass took ~25s, blowing
	// certMirrorBootTimeout, and since this refresh runs BEFORE either
	// listener binds, the edge was not serving until it finished. The
	// self-update gate waits 20s for the edge to answer on loopback, so that
	// box failed its gate, rolled back, failed the rollback's gate too, and
	// took no update at all from 2026-09-07 to 2026-09-19.
	//
	// The count is what makes it a fleet problem rather than an AU one: every
	// box pays count x latency, so each new published cert moves every box
	// closer to the same cliff. us-east-1 at ~75ms is about a third of the
	// way there at today's count.
	var items []certFetch
	for _, key := range keys {
		name, ok := strings.CutPrefix(key, certPrefix)
		// publishableCertName is the same filter the publish side applies:
		// anything nested under the prefix was not written by this deployment's
		// issuer and has no cache name to map back onto.
		if !ok || !publishableCertName(name) {
			continue
		}
		items = append(items, certFetch{key: key, name: name})
	}
	tried := len(items)
	failed, lastErr := applyCertFetches(items, m.fetchCerts(ctx, items), prev, next, etags)
	if tried > 0 && failed == tried {
		// Nothing at all could be read: that is a bucket fault, not an object
		// fault, and it is reported like a failed List — last-good stands.
		return fmt.Errorf("get %s*: every one of %d object(s) failed, last: %w", certPrefix, tried, lastErr)
	}
	if failed > 0 {
		m.logPartialPoll(failed, tried, lastErr)
	}
	m.publish(ctx, prev, next, etags)
	// A partial rebuild still counts as the full rebuild: the objects it
	// could not read carry an unknown ETag, so the next poll's changed-only
	// read fetches exactly those again. It reflects no generation, though,
	// so that next poll cannot skip.
	m.lastFull, m.fullAge = m.clock(), jitteredBackstop()
	m.actedGen, m.actedOK = gen, genOK && failed == 0
	return nil
}

// poll is the 30s loop's refresh (#336). Disarmed, it is Refresh. Armed, it
// polls the change index (a 304 when nothing moved, one GET either way) and
// then does the least that keeps the snapshot current:
//
//   - nothing, when control/certs has not moved since the generation the
//     snapshot reflects and the last full rebuild is younger than the
//     jittered backstop;
//   - a changed-only read (refreshChanged) when it has moved;
//   - today's full Refresh when the watch cannot vouch for the scope (the
//     index unreadable: "unknown", never "unchanged"), when no full rebuild
//     has landed yet, or at the backstop.
//
// Poll's moved list is not consulted: the watch is shared with the agent's
// tick and record view, so whichever polls first consumes it. Generations
// are compared with the one this mirror last acted on instead.
func (m *certMirror) poll(ctx context.Context) error {
	w := m.changeWatch()
	if w == nil {
		return m.Refresh(ctx)
	}
	// Bounded on its own: the watch is shared, and a hung index GET from
	// another caller holds its poll lock. Any error, giving up on that lock
	// included, is "unknown": a full read, as before the index.
	_, perr := pollWatch(ctx, w, m.watchTimeout)
	m.refreshMu.Lock()
	defer m.refreshMu.Unlock()
	gen, ok := w.Gen(certScope)
	switch {
	case perr != nil, !ok, m.lastFull.IsZero(), !m.clock().Before(m.lastFull.Add(m.fullAge)):
		return m.refreshFull(ctx)
	case m.actedOK && gen == m.actedGen:
		// Nothing moved in the bucket, but a copy into the local cache
		// that failed is retried from the snapshot already held: no
		// bucket read (final review R1, finding 7).
		if m.localPending {
			m.localPending = m.syncLocal(ctx, m.snapshot())
		}
		return nil
	}
	return m.refreshChanged(ctx, gen)
}

// refreshChanged rebuilds the snapshot from one ETag listing plus a get per
// new or rewritten object, under refreshMu. gen is the control/certs
// generation captured before the listing. It keeps every guarantee of the
// full rebuild:
//
//   - a key absent from the listing (a spent challenge token) is absent from
//     the new snapshot, because the snapshot is built from the listing;
//   - a failed get of a changed object carries the previous copy forward
//     with an unknown ETag, so the next poll fetches it again, and is never
//     a failed poll, even when it is the only object that changed;
//   - the new snapshot is built whole and swapped in by publish, which also
//     copies changed material into the local cache and wakes the wildcard
//     manager, exactly as Refresh does.
//
// A store without the ETag listing, or a listing that fails, is "unknown":
// today's full Refresh runs instead.
func (m *certMirror) refreshChanged(ctx context.Context, gen uint64) error {
	lister, ok := m.s.(release.ETagLister)
	if !ok {
		return m.refreshFull(ctx)
	}
	listed, err := lister.ListETags(ctx, certPrefix)
	if err != nil {
		return m.refreshFull(ctx)
	}
	prev := m.snapshot()
	next := make(map[string][]byte, len(listed))
	etags := make(map[string]string, len(listed))
	var items []certFetch
	for _, le := range listed {
		name, ok := strings.CutPrefix(le.Key, certPrefix)
		if !ok || !publishableCertName(name) {
			continue
		}
		if was := m.etags[name]; was != "" && was == le.ETag {
			if body, ok := prev.entries[name]; ok {
				next[name], etags[name] = body, was
				continue
			}
		}
		items = append(items, certFetch{key: le.Key, name: name})
	}
	tried := len(items)
	failed, lastErr := applyCertFetches(items, m.fetchCerts(ctx, items), prev, next, etags)
	// No "every object failed" rule here, unlike Refresh: the listing just
	// answered, so the bucket is up, and a changed-only read is often ONE
	// object. Treating that one failure as a bucket fault would turn a single
	// unreadable key into a failed poll until the backstop, holding back
	// every deletion in the listing and the loop's backfill with it. The
	// failed objects carry their previous copies with unknown ETags and the
	// generation stays unacted, so the next poll reads them again.
	if failed > 0 {
		m.logPartialPoll(failed, tried, lastErr)
	}
	m.publish(ctx, prev, next, etags)
	m.actedGen, m.actedOK = gen, failed == 0
	return nil
}

// certFetch is one object a refresh reads; certFetchResult its outcome.
type certFetch struct {
	key, name string
}

type certFetchResult struct {
	body []byte
	etag string
	err  error
}

// fetchCerts gets every item's body, at most certRefreshConcurrency at once,
// and returns the results in item order.
func (m *certMirror) fetchCerts(ctx context.Context, items []certFetch) []certFetchResult {
	results := make([]certFetchResult, len(items))
	// A fixed pool pulling indices off a channel, NOT a goroutine per object
	// parked on a semaphore. The distinction only shows up at scale, and it is
	// the whole point of having a bound: goroutine-per-object costs one stack
	// per cert before a single one is read, so a deployment with a million
	// published certs would spend gigabytes on goroutines that are all waiting
	// for one of sixteen slots.
	//
	// Workers also select on ctx.Done, so a cancelled boot refresh stops
	// handing out work instead of draining the whole queue through Get. (It
	// still waits for the gets already in flight: Refresh has never been a
	// hard bound on its own wall clock, and certMirrorBootTimeout bounds the
	// requests, not this function's return.)
	workers := min(certRefreshConcurrency, len(items))
	idx := make(chan int)
	var wg sync.WaitGroup
	for range workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := range idx {
				body, etag, err := m.s.Get(ctx, items[i].key)
				results[i] = certFetchResult{body: body, etag: etag, err: err}
			}
		}()
	}
	for i := range items {
		select {
		case idx <- i:
		case <-ctx.Done():
			// Leaves results[i:] zero — a nil body with a nil error, which
			// the apply loop reads as a successful empty get. Set the
			// context's error on the rest so they take the failed-get branch
			// and carry the previous copy forward instead.
			for j := i; j < len(items); j++ {
				results[j] = certFetchResult{err: ctx.Err()}
			}
			close(idx)
			wg.Wait()
			return results
		}
	}
	close(idx)
	wg.Wait()
	return results
}

// applyCertFetches folds fetched bodies into next (and their ETags into
// etags), carrying prev's copy of any object that failed, and returns the
// failure count and the last failure.
func applyCertFetches(items []certFetch, results []certFetchResult, prev *certSnapshot, next map[string][]byte, etags map[string]string) (failed int, lastErr error) {
	// Applied in key order, not completion order, so a partial poll reports
	// the same lastErr whichever way the fetches happened to interleave.
	for i, it := range items {
		r := results[i]
		switch {
		case errors.Is(r.err, os.ErrNotExist):
			// Raced a delete between the list and the get — a spent token. Not
			// a fault; the entry is simply gone.
			continue
		case r.err != nil:
			// ONE bad object is not a bad poll. Aborting here would discard the
			// whole rebuild — every unrelated cert that landed this tick with
			// it — and skip syncLocal and the wildcard wake; a single
			// permanently unreadable key would freeze the mirror forever. Carry
			// the previous copy of this name forward and keep going, with an
			// unknown ETag so the next changed-only read fetches it again.
			failed++
			lastErr = r.err
			if body, ok := prev.entries[it.name]; ok {
				next[it.name], etags[it.name] = body, ""
			}
			continue
		}
		next[it.name], etags[it.name] = r.body, r.etag
	}
	return failed, lastErr
}

// publish swaps in the snapshot built from next, whole, and does what every
// landed refresh does after it: the local cache copy-in, the health latch and
// the wildcard wake. Under refreshMu.
func (m *certMirror) publish(ctx context.Context, prev *certSnapshot, next map[string][]byte, etags map[string]string) {
	snap := buildCertSnapshot(prev, next)
	m.mu.Lock()
	m.snap = snap
	m.mu.Unlock()
	m.etags = etags
	m.localPending = m.syncLocal(ctx, snap)
	// The local cache the handshake reads now holds what the bucket does:
	// the certificate condition is met for good (§6). A later failed
	// refresh keeps serving this copy and leaves the latch set.
	m.health.CertsLoaded()
	if m.wake != nil && wildcardMaterialChanged(prev, snap) {
		// A takeover (or a first wildcard for a new account) reaches the
		// serving path in one poll instead of up to twelve hours.
		m.wake()
	}
}

// pollWatch polls w under a bound of its own (timeout, or
// release.ChangeIndexPollTimeout when 0) inside ctx. The watch is shared with
// the agent's tick, whose hung index GET would otherwise hold this loop, and
// the health heartbeat it stamps, behind the watch's poll lock.
func pollWatch(ctx context.Context, w *release.ChangeWatch, timeout time.Duration) ([]string, error) {
	if timeout <= 0 {
		timeout = release.ChangeIndexPollTimeout
	}
	pctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	return w.Poll(pctx)
}

// certScope is the change-index scope every write under certPrefix bumps.
const certScope = "control/certs"

func (m *certMirror) changeWatch() *release.ChangeWatch {
	if m.watch == nil {
		return nil
	}
	return m.watch()
}

// certsGen is control/certs' generation as the watch last saw it, ok false
// when there is no usable watch.
func (m *certMirror) certsGen() (uint64, bool) {
	if w := m.changeWatch(); w != nil {
		return w.Gen(certScope)
	}
	return 0, false
}

func (m *certMirror) clock() time.Time {
	if m.now != nil {
		return m.now()
	}
	return time.Now()
}

// jitteredBackstop is release.ChangeIndexBackstop jittered by up to 10% either
// way (R3), drawn per full read so a fleet's rebuilds do not line up.
func jitteredBackstop() time.Duration {
	b := release.ChangeIndexBackstop
	return b - b/10 + time.Duration(rand.Int64N(int64(b/5)+1))
}

// logPartialPoll reports unreadable objects on the loop's own first-and-every-
// tenth cadence: a permanently bad key would otherwise say the same thing
// every 30 seconds forever.
func (m *certMirror) logPartialPoll(failed, tried int, err error) {
	m.mu.Lock()
	m.partialPolls++
	n := m.partialPolls
	m.mu.Unlock()
	if n == 1 || n%10 == 0 {
		log.Printf("certsync: %d of %d object(s) under %s unreadable this poll (#%d, last: %v) — the previous copy of each is carried forward", failed, tried, certPrefix, n, err)
	}
}

// buildCertSnapshot parses what changed and carries the rest of the previous
// parse forward — a handshake never pays for x509 parsing, and a quiet
// deployment pays for it once.
func buildCertSnapshot(prev *certSnapshot, entries map[string][]byte) *certSnapshot {
	snap := &certSnapshot{entries: entries, certs: make(map[string]*tls.Certificate, len(entries))}
	for name, body := range entries {
		if !certShapedName(name) {
			continue
		}
		if strings.HasSuffix(name, alpnTokenSuffix) {
			// tls-alpn-01 token certs are never served from a snapshot — see
			// alpnTokenCert — so parsing them here would be work done every
			// poll for an answer nothing may use. Leaving them unparsed also
			// makes serving a stale one structurally impossible.
			continue
		}
		if _, seen := prev.entries[name]; seen && bytes.Equal(prev.entries[name], body) {
			// Unchanged. Carry the previous RESULT forward, failure included:
			// re-parsing identical bytes cannot answer differently, and
			// re-logging a corrupt entry every 30s forever is not "loud", it is
			// noise.
			if old, ok := prev.certs[name]; ok {
				snap.certs[name] = old
			}
			continue
		}
		cert, err := parseCertEntry(body)
		if err != nil {
			// Loud, and skipped rather than served: a body that does not parse
			// is a body no handshake can use, and the next poll re-reads it.
			log.Printf("certsync: mirrored entry %q does not parse as a certificate (%v) — skipping it", name, err)
			continue
		}
		snap.certs[name] = cert
	}
	return snap
}

func (m *certMirror) snapshot() *certSnapshot {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.snap
}

// Cert returns the mirrored body for a cache name. Account keys are refused
// here, not filtered at poll time, because the local sync legitimately needs
// them: the refusal belongs at the SERVING door.
func (m *certMirror) Cert(name string) ([]byte, bool) {
	if accountKeyName(name) {
		return nil, false
	}
	b, ok := m.snapshot().entries[name]
	return b, ok
}

// Challenge returns the HTTP-01 response body for token, if the mirror holds
// it. The cache name is autocert's own (autocert.go:896,
// httpTokenCacheKey): path.Base(<challenge path>) + "+http-01" — flat, with
// no /.well-known/ prefix.
func (m *certMirror) Challenge(token string) ([]byte, bool) {
	b, ok := m.snapshot().entries[token+httpTokenSuffix]
	return b, ok
}

const (
	httpTokenSuffix = "+http-01" // autocert's HTTP-01 response entries
	alpnTokenSuffix = "+token"   // autocert's tls-alpn-01 challenge certs
	// autocertAccountKeyName is autocert's own ACME account key entry
	// (autocert.go's accountKey cache name). wildcardAccountKeyName
	// (wildcard.go) is the DNS-01 manager's.
	autocertAccountKeyName = "acme_account+key"
)

// accountKeyName reports whether name is ACME ACCOUNT material — the issuer's
// identity at the CA, never a serving artifact.
func accountKeyName(name string) bool {
	return name == autocertAccountKeyName || name == wildcardAccountKeyName
}

// certShapedName reports whether an entry is expected to parse as a
// certificate: everything except account keys and HTTP-01 response bodies.
// That leaves per-host certs (<host>, <host>+rsa), wildcard PEMs
// (wildcard-<parent>.pem) and tls-alpn-01 token certs (<host>+token) — the
// last of which only the live read-through parses (buildCertSnapshot skips
// them).
func certShapedName(name string) bool {
	return !accountKeyName(name) && !strings.HasSuffix(name, httpTokenSuffix)
}

// parseCertEntry parses an autocert cache entry — the private key PEM
// followed by the chain, which is also wildcard.go's encodeWildcardPEM shape —
// into a usable tls.Certificate. Both halves come out of the one blob, and
// X509KeyPair is what checks that the key actually matches the leaf.
func parseCertEntry(raw []byte) (*tls.Certificate, error) {
	cert, err := tls.X509KeyPair(raw, raw)
	if err != nil {
		return nil, err
	}
	if cert.Leaf == nil {
		// X509KeyPair has populated Leaf since Go 1.23; belt and braces, since
		// every serving decision below reads it.
		leaf, err := x509.ParseCertificate(cert.Certificate[0])
		if err != nil {
			return nil, err
		}
		cert.Leaf = leaf
	}
	return &cert, nil
}

// run polls until stop closes — the edge's standing background-loop shape
// (Watch): fixed ticker, errors counted and logged at the first and then
// every tenth, recovery logged once.
func (m *certMirror) run(stop <-chan struct{}) {
	log.Printf("edge: cert mirror armed (%s poll of %s*) — this box serves every cert and challenge the issuer publishes", certMirrorInterval, certPrefix)
	tick := time.NewTicker(certMirrorInterval)
	defer tick.Stop()
	errs := 0
	for {
		select {
		case <-stop:
			return
		case <-tick.C:
		}
		ctx, cancel := context.WithTimeout(context.Background(), certMirrorTimeout)
		err := m.poll(ctx)
		cancel()
		if err != nil {
			errs++
			if errs == 1 || errs%10 == 0 {
				log.Printf("edge: cert mirror refresh failed (#%d): %v — serving the last good copy", errs, err)
			}
			continue
		}
		if errs > 0 {
			log.Printf("edge: cert mirror refresh recovered after %d failure(s)", errs)
			errs = 0
		}
		ctx, cancel = context.WithTimeout(context.Background(), certMirrorTimeout)
		m.backfill(ctx)
		cancel()
	}
}

// backfill publishes what this box's cache holds and the bucket does not —
// every box's half of "every edge serves every cert" for material minted
// BEFORE publishing existed (#291), on whichever box's disk it sits (#318).
//
// publishingCache publishes on Put, which is to say on issuance and renewal.
// A deployment that arms a second edge after a year of single-box service
// therefore starts with a bucket holding nothing but what was minted since
// the gate went in: the flagship roll found 4 objects against 50 live
// certs, and every handshake on a new region edge failed until renewal —
// 30 to 60 days out. Renewal is the wrong repair for that; this is a
// one-pass copy, run after every successful poll, that converges in one
// tick. Steady state on the issuer is one directory listing per poll plus
// a stat per entry: an entry whose size and mtime have not moved since it
// was last settled — in sync with the bucket, or found not to be this
// deployment's — is not re-read (m.settled).
//
// Rules, each doing work:
//
//   - Only on a box that contends in the election, NOT only on the issuer
//     (m.mayBackfill: agent-run AND contending in the issuer election).
//     The 2026-08-27 rung-1 rehearsal (#318) found the issuer-only rule
//     left every cert minted on a box that never held the lease since
//     #291 sitting on that box's disk alone — exactly the bytes a
//     failover loses with the box. Cert entries in control/certs/** are
//     therefore multi-writer now, kept convergent by the freshness rules
//     below (bytes-equal settles; an older leaf never displaces a newer
//     one), and the writer map's housekeeper-only line for this prefix is
//     amended to "any contending box, certs only". A concurrent writer
//     racing a stale view CAN transiently land an older leaf over a newer
//     one; the identity half of fileStamp is what makes that transient —
//     the box holding the newer leaf un-settles on the changed bucket sum
//     and republishes next pass. The serving-side "held
//     or ungated" read is still the wrong predicate: a hand-run `carlos
//     edge` with a bucket but no agent is ungated for serving and must
//     NOT dump its disk into the deployment's prefix — m.mayBackfill is
//     false exactly there. One consequence to hold in mind when a cert
//     PURGE ever lands: any box's mirrored on-disk copy now resurrects a
//     deleted bucket entry on its next pass, so a purge must tombstone or
//     it will not stick.
//   - Only a DirCache is walked. The mirror holds the cache as an
//     autocert.Cache, and that interface cannot enumerate; the on-disk
//     layout is the only backend that exists (certcache.go) and the one
//     whose entries ARE plain files named by cache key.
//   - Only names this deployment owns publish. The disk of a long-lived box
//     holds a cert for every route ever detached or renamed (autocert never
//     deletes), and DirCache.Put leaves "<name><random>" temp files behind
//     a kill mid-write — neither is the deployment's to replicate onto every
//     edge. A per-host entry (<host>, <host>+rsa) must pass m.allowed, the
//     same registry-backed exact check both read-throughs use; a wildcard
//     PEM must be for a parent in the wildcard manager's current set and
//     its leaf must cover it. Account keys are their own shape.
//   - Challenge tokens never publish: they are seconds long, and a copy of
//     a spent one is litter every edge would go on serving.
//   - Account keys publish only when the bucket has none, and only from
//     the box holding the STRICT issuer gate (m.issuing) — the one
//     backfill decision that stays single-writer. The deployment's ACME
//     identity is whichever key was published first (the mirror seeds it
//     onto every box, and a takeover issuer orders under it); a box's own
//     older key must never displace it, or the deployment would swap
//     accounts on every failover — and N agent boxes racing an EMPTY
//     bucket with their own keys would leave certs ordered under one
//     account stored beside another's key, which is why the widened cert
//     rule above does not extend here. A non-issuer's unseeded key is
//     deliberately NOT settled: the box may hold the lease on a later
//     pass, and a settled stamp would sit on exactly the pass that could
//     finally publish it.
//   - A cert-shaped entry publishes when the bucket lacks it, or holds an
//     older leaf (NotAfter earlier than this box's) — the renewed-before-
//     the-gate case. Anything that does not parse, or has already expired,
//     stays on disk: a stale rehearsal name is nothing to distribute, and
//     the expiry check is what keeps this from re-publishing it forever.
//   - Failures are per-entry, retried next poll, and logged on the file's
//     first-and-every-tenth cadence: a write-denied bucket must not say so
//     every 30 seconds forever.
func (m *certMirror) backfill(ctx context.Context) {
	if m.mayBackfill == nil || !m.mayBackfill() || m.s == nil {
		return
	}
	dir, ok := m.local.(autocert.DirCache)
	if !ok {
		m.backfillOnce.Do(func() {
			log.Printf("certsync: backfill off — the local cert cache is not an on-disk cache and cannot be walked; only certs minted from now on reach %s*", certPrefix)
		})
		return
	}
	entries, err := os.ReadDir(string(dir))
	if err != nil {
		if !errors.Is(err, os.ErrNotExist) {
			m.logBackfillFailure(0, err)
		}
		return
	}
	snap := m.snapshot()
	now := time.Now()
	var published, failed int
	var lastErr error
	for _, ent := range entries {
		name := ent.Name()
		if !ent.Type().IsRegular() || !publishableCertName(name) {
			continue
		}
		if strings.HasSuffix(name, httpTokenSuffix) || strings.HasSuffix(name, alpnTokenSuffix) {
			continue
		}
		remote, inBucket := snap.entries[name]
		info, err := ent.Info()
		if err != nil {
			continue
		}
		stamp := fileStamp{size: info.Size(), mtime: info.ModTime().UnixNano(), bucketSum: bucketSum(remote, inBucket)}
		// Settled at this stamp on an earlier pass — in sync with the
		// bucket, or not this deployment's to publish — and nothing has
		// moved since. The bucket copy's IDENTITY is part of the stamp: an
		// entry that was in sync but has since left the bucket, or been
		// overwritten by another box, re-reads — that re-read is the #318
		// repair path (see fileStamp). This is what keeps steady state at
		// a listing plus a stat per entry.
		if m.settledAt(name) == stamp {
			continue
		}
		body, err := os.ReadFile(filepath.Join(string(dir), name))
		if err != nil {
			failed++
			lastErr = err
			continue
		}
		if inBucket && bytes.Equal(remote, body) {
			// Also the backstop for names the snapshot never parses: the
			// freshness compare below cannot see those, and without this
			// they would republish every poll.
			m.markSettled(name, stamp)
			continue
		}
		if accountKeyName(name) {
			if inBucket {
				m.markSettled(name, stamp)
				continue
			}
			if m.issuing == nil || !m.issuing() {
				// Only the elected issuer seeds the deployment's ACME
				// identity (the rule above). Not settled, deliberately —
				// see the same rule for why.
				continue
			}
		} else {
			cert, err := parseCertEntry(body)
			if err != nil || !now.Before(cert.Leaf.NotAfter) || !m.ownsCertName(name, cert) {
				// Expiry is a function of time, but a settled entry can only
				// move from live to expired, never back; and ownership can
				// return (a route re-attached), which is a registry change —
				// so an entry settled as not-owned is re-examined only when
				// its stamp moves. A re-attached route mints afresh anyway.
				m.markSettled(name, stamp)
				continue
			}
			if inBucket {
				if have := snap.certs[name]; have != nil && !have.Leaf.NotAfter.Before(cert.Leaf.NotAfter) {
					m.markSettled(name, stamp)
					continue
				}
			}
		}
		if err := putCert(ctx, m.s, name, body); err != nil {
			failed++
			lastErr = err
			continue
		}
		// Not settled: the bucket side of the stamp changes on the next
		// poll, and the bytes.Equal path settles it then.
		published++
	}
	if published > 0 {
		log.Printf("certsync: backfilled %d cache entr%s from this box into %s* — every edge serves them on its next poll", published, plural(published, "y", "ies"), certPrefix)
	}
	if failed > 0 {
		m.logBackfillFailure(failed, lastErr)
	}
}

// ownsCertName reports whether a cert-shaped cache entry is one this
// deployment should replicate: a per-host entry for a name the registry
// still serves (m.allowed — nil admits nothing, the safe default for a
// mirror built without an Edge), or a wildcard PEM for a parent in the
// wildcard manager's CURRENT parent set (m.wildcardParents — nil, no
// manager, admits none) whose leaf covers that parent.
func (m *certMirror) ownsCertName(name string, cert *tls.Certificate) bool {
	if parent, ok := strings.CutPrefix(name, "wildcard-"); ok && strings.HasSuffix(parent, ".pem") {
		parent = strings.TrimSuffix(parent, ".pem")
		if parent == "" || m.wildcardParents == nil || !slices.Contains(m.wildcardParents(), parent) {
			return false
		}
		return slices.Contains(cert.Leaf.DNSNames, "*."+parent)
	}
	host := strings.TrimSuffix(name, "+rsa")
	if m.allowed == nil || !m.allowed(host) {
		return false
	}
	return cert.Leaf.VerifyHostname(host) == nil
}

// fileStamp is the state a backfill pass last settled an entry at: the
// file's size+mtime, and the IDENTITY of the bucket's copy (a sha256 of
// its bytes; zero = absent), not merely its presence. The local half is
// not a hash: DirCache writes are atomic renames, so a changed body is a
// changed stamp (nanosecond mtime, so a same-size rewrite within one
// second on a coarse filesystem is the only alias, and autocert never
// rewrites a name that fast). The bucket half MUST be identity: with
// cert entries multi-writer (#318), another box racing a stale view can
// land an older leaf over a newer one this box already settled against —
// a presence bool would leave the settled box blind to the clobber
// forever (its file unmoved, the name still "in bucket"), while the sum
// changes with the clobber, un-settles exactly the box holding the newer
// leaf, and its next pass republishes it. Convergence within a poll or
// two, newest leaf wins.
type fileStamp struct {
	size      int64
	mtime     int64
	bucketSum [sha256.Size]byte
}

// bucketSum is fileStamp's bucket-identity half. The zero value means
// "absent", which cannot collide with a real entry: nothing under
// control/certs/ is ever zero bytes, and sha256 of any body is nonzero.
func bucketSum(body []byte, inBucket bool) [sha256.Size]byte {
	if !inBucket {
		return [sha256.Size]byte{}
	}
	return sha256.Sum256(body)
}

func (m *certMirror) settledAt(name string) fileStamp {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.settled[name]
}

func (m *certMirror) markSettled(name string, st fileStamp) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.settled == nil {
		m.settled = map[string]fileStamp{}
	}
	m.settled[name] = st
}

// logBackfillFailure is logPartialPoll's cadence for the publish side.
func (m *certMirror) logBackfillFailure(failed int, err error) {
	m.mu.Lock()
	m.backfillFailures++
	n := m.backfillFailures
	m.mu.Unlock()
	if n == 1 || n%10 == 0 {
		log.Printf("certsync: backfill: %d entr%s not published (#%d, last: %v) — retried on the next poll", failed, plural(failed, "y", "ies"), n, err)
	}
}

func plural(n int, one, many string) string {
	if n == 1 {
		return one
	}
	return many
}

// syncLocal copies mirrored material into this box's OWN autocert cache. It
// is the quiet half of the mirror, and the one that matters when nothing is
// looking:
//
//   - autocert's renewal timers bypass HostPolicy and check the cache FIRST
//     (renewal.go's do()), so a stood-down box whose timers are still armed
//     finds a fresh cert here and never reaches the CA;
//   - a box that boots, or crashes and restarts, has no election result for
//     the first ~20 minutes; a warm local cache serves through that window
//     even before the mirror map is consulted;
//   - the wildcard manager loads parents from this same cache (loadCached),
//     which is how a wildcard the issuer minted reaches a non-issuer's
//     handshakes at all.
//
// Never written locally: challenge tokens of either kind. They are seconds
// long, the issuer owns their deletion, and a copy on every box's disk is
// litter nothing sweeps.
//
// Writes are best-effort and loud: a full disk must not stop the mirror from
// SERVING what it already holds in memory. It reports whether any entry could
// not be reconciled, so the armed poll can retry an unchanged snapshot.
func (m *certMirror) syncLocal(ctx context.Context, snap *certSnapshot) (failedAny bool) {
	if m.local == nil {
		return false
	}
	m.syncMu.Lock()
	defer m.syncMu.Unlock()
	var failed int
	var lastErr error
	for name, body := range snap.entries {
		if strings.HasSuffix(name, httpTokenSuffix) || strings.HasSuffix(name, alpnTokenSuffix) {
			continue
		}
		// Unchanged since the last reconcile: the local copy was already
		// settled then, and nothing but autocert itself (which only ever writes
		// fresher material) can have moved it since. Steady state costs no
		// local I/O at all.
		if prev, ok := m.synced[name]; ok && bytes.Equal(prev, body) {
			continue
		}
		switch err := m.reconcileLocal(ctx, name, body, snap.certs[name]); {
		case err == nil:
			m.synced[name] = body
		case errors.Is(err, errUnparseableEntry):
			// Already reported by the snapshot build; nothing to say twice.
		default:
			failed++
			lastErr = err
		}
	}
	if failed > 0 {
		// One line per poll, not one per name: a local cache that has stopped
		// accepting writes fails for EVERY name at once, and the fault is the
		// cache, not the certificate.
		log.Printf("certsync: %d mirrored entry/entries could not be reconciled into this box's local cert cache (last: %v) — they are still served from memory", failed, lastErr)
	}
	// Forget names the bucket no longer holds, so the map tracks the prefix
	// rather than growing with every token this box has ever seen.
	for name := range m.synced {
		if _, ok := snap.entries[name]; !ok {
			delete(m.synced, name)
		}
	}
	return failed > 0
}

// errUnparseableEntry marks the one reconcile failure that is the ENTRY's
// fault rather than the local cache's — already reported once by the snapshot
// build, and not worth a second line.
var errUnparseableEntry = errors.New("certsync: mirrored entry does not parse")

// reconcileLocal writes one entry into the local cache when the local copy is
// worse. A nil error means the local copy can be considered settled.
func (m *certMirror) reconcileLocal(ctx context.Context, name string, body []byte, cert *tls.Certificate) error {
	local, err := m.local.Get(ctx, name)
	if err != nil && !errors.Is(err, autocert.ErrCacheMiss) {
		// An unreadable local cache is a real fault (permissions, disk).
		return fmt.Errorf("read %q: %w", name, err)
	}
	have := err == nil
	if accountKeyName(name) {
		// ACME ACCOUNT material. Seeded when absent, so that a box winning the
		// election later takes over the DEPLOYMENT's account — one account, one
		// rate-limit bucket, which is the envelope the issuance gate's
		// no-fencing decision is priced against. Never overwritten: a box that
		// already registered its own account has orders and authorizations tied
		// to that key, and replacing it under them buys nothing.
		if have {
			return nil
		}
		if err := m.local.Put(ctx, name, body); err != nil {
			return fmt.Errorf("seed the deployment's ACME account key %q: %w", name, err)
		}
		log.Printf("certsync: seeded the deployment's ACME account key (%s) into the local cache", name)
		return nil
	}
	if cert == nil {
		// Did not parse (already reported by buildCertSnapshot). Writing an
		// unparseable body over a working local cert would be a downgrade.
		return errUnparseableEntry
	}
	if have {
		if cur, err := parseCertEntry(local); err == nil && !cur.Leaf.NotAfter.Before(cert.Leaf.NotAfter) {
			// The local copy is at least as fresh — this box may BE the issuer,
			// one write ahead of what it has published. Leave it alone.
			return nil
		}
	}
	if err := m.local.Put(ctx, name, body); err != nil {
		return fmt.Errorf("write %q: %w", name, err)
	}
	return nil
}

// wildcardPEMName reports whether an entry is a wildcard cert PEM
// (wildcard.go's wildcardPEMKey shape).
func wildcardPEMName(name string) bool {
	return strings.HasPrefix(name, "wildcard-") && strings.HasSuffix(name, ".pem")
}

// wildcardMaterialChanged reports whether this poll brought a wildcard PEM
// that is new or different. Only then is the wildcard manager nudged: its own
// loop sleeps 12h between passes, so a takeover — or a first wildcard for a
// new account, minted on the issuer — would otherwise wait that long to reach
// this box's handshakes. A withdrawal is not a wake: there is nothing new to
// load, and the manager keeps serving what it has until it expires.
func wildcardMaterialChanged(prev, next *certSnapshot) bool {
	for name, body := range next.entries {
		if !wildcardPEMName(name) {
			continue
		}
		if !bytes.Equal(prev.entries[name], body) {
			return true
		}
	}
	return false
}

// ---- serving: the four-layer chain ------------------------------------------
//
// wrapGetCertificate is the mirror's half of the handshake path. The FULL
// chain a hello meets, outermost first (RunWithActivatorReady composes it):
//
//	wildcard (wildcard.go) → mirror → local autocert → local cache, direct
//
// The order is forced by autocert's own shape, not by preference. autocert
// runs HostPolicy — which carries the issuance gate — BEFORE its cache
// lookup (autocert.go:292), so on a non-issuer autocert serves NOTHING: not
// what the mirror handed it, not what it holds on disk itself. Putting the
// mirror behind it would therefore serve nothing at all on every box but one.
//
// The last layer is the local cache read DIRECTLY, bypassing autocert, and it
// is not redundant: it is what a box serves when the bucket is unreachable at
// boot (empty mirror) but its own disk still holds yesterday's certs, and
// what covers the ~20 minute window after a boot or a crash while the
// housekeeper election is undecided.
//
// The short-circuit ahead of autocert applies while this box is GATED, and
// only then. autocert arms its renewal timers LAZILY, out of GetCertificate
// (m.cert -> startRenew, autocert.go:464) — a mirror that answered every
// hello would leave an ISSUER with no renewal timers at all, and its certs
// would expire ninety days later with nothing in the journal. An ungated box
// loses nothing by going through autocert first: syncLocal has already put
// every mirrored cert in the cache autocert reads, and the mirror stays
// behind it as a fallback for anything autocert cannot serve.
func (m *certMirror) wrapGetCertificate(inner func(*tls.ClientHelloInfo) (*tls.Certificate, error)) func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
	return func(hello *tls.ClientHelloInfo) (*tls.Certificate, error) {
		// Converted ONCE, here, and used by every layer below: the mirror
		// lookup, the tls-alpn-01 lookup and the local cache read all have to
		// agree with the name autocert published under.
		name := serverName(hello.ServerName)
		if len(hello.SupportedProtos) == 1 && hello.SupportedProtos[0] == acme.ALPNProto {
			// A tls-alpn-01 validation. autocert answers its OWN order from
			// memory; anything else is another edge's order, and the CA reached
			// this box because every edge answers the same names.
			cert, err := inner(hello)
			if err == nil {
				return cert, nil
			}
			if c := m.alpnTokenCert(helloContext(hello), name); c != nil {
				return c, nil
			}
			return nil, err
		}
		wantRSA := !helloSupportsECDSA(hello)
		if m.isGated() {
			if c := m.certificate(name, wantRSA); c != nil {
				return c, nil
			}
		}
		cert, err := inner(hello)
		if err == nil {
			return cert, nil
		}
		// Ungated and autocert could not answer (an order that failed, a name
		// it will not order for): the mirror is still the better answer than a
		// dropped handshake. Gated, this repeat is a cheap map miss.
		if c := m.certificate(name, wantRSA); c != nil {
			return c, nil
		}
		if c := m.localCertificate(helloContext(hello), name, wantRSA); c != nil {
			return c, nil
		}
		return nil, err
	}
}

// serverName normalizes a client-supplied name to the cache name autocert
// would have used: lowercased, no trailing dot, and punycode via
// idna.Lookup.ToASCII — the same conversion autocert applies before its own
// cache lookup (autocert.go:307), because some clients (curl among them) send
// unicode SNI verbatim and a cert is always published under the A-label.
//
// A name IDNA refuses is kept in its lowercased form rather than dropped:
// every layer below treats it as an ordinary miss, which is what it is, and
// the refusal belongs to autocert (which rejects it outright) rather than to
// the mirror.
func serverName(sni string) string {
	name := strings.ToLower(strings.TrimSuffix(sni, "."))
	if ascii, err := idna.Lookup.ToASCII(name); err == nil {
		return ascii
	}
	return name
}

// isGated reads the issuance gate, defaulting to "gated" (see the field).
func (m *certMirror) isGated() bool {
	return m.gated == nil || m.gated()
}

// helloContext is the handshake's context, or a background one for the
// hand-built hellos tests use.
func helloContext(hello *tls.ClientHelloInfo) context.Context {
	if ctx := hello.Context(); ctx != nil {
		return ctx
	}
	return context.Background()
}

// certificate returns a mirrored cert for name, or nil. The two cache names
// are autocert's own ("<host>" for ECDSA, "<host>+rsa" for legacy clients);
// the preferred one is tried first and the other accepted rather than
// dropping the handshake, since on a non-issuer no order can produce the
// missing type.
func (m *certMirror) certificate(name string, wantRSA bool) *tls.Certificate {
	if name == "" {
		return nil
	}
	snap := m.snapshot()
	for _, key := range certEntryNames(name, wantRSA) {
		if c := snap.certs[key]; usableFor(c, name) {
			return c
		}
	}
	return nil
}

// certEntryNames lists the cache names that can hold a cert for host, most
// preferred first.
func certEntryNames(host string, wantRSA bool) [2]string {
	if wantRSA {
		return [2]string{host + "+rsa", host}
	}
	return [2]string{host, host + "+rsa"}
}

// usableFor reports whether cert can be served for name right now. An expired
// or wrong-named entry is treated as absent so the chain keeps falling
// through — a dead cert fails the handshake just as surely as no cert, and
// the next layer may have a live one.
func usableFor(cert *tls.Certificate, name string) bool {
	return cert != nil && cert.Leaf != nil &&
		time.Now().Before(cert.Leaf.NotAfter) &&
		cert.Leaf.VerifyHostname(name) == nil
}

// helloSupportsECDSA is autocert's own supportsECDSA (autocert.go), ported
// because it is unexported there and the mirror must look under the SAME
// cache name autocert would have chosen for this client.
func helloSupportsECDSA(hello *tls.ClientHelloInfo) bool {
	if hello.SignatureSchemes != nil {
		ecdsaOK := false
	schemeLoop:
		for _, scheme := range hello.SignatureSchemes {
			const tlsECDSAWithSHA1 tls.SignatureScheme = 0x0203
			switch scheme {
			case tlsECDSAWithSHA1, tls.ECDSAWithP256AndSHA256,
				tls.ECDSAWithP384AndSHA384, tls.ECDSAWithP521AndSHA512:
				ecdsaOK = true
				break schemeLoop
			}
		}
		if !ecdsaOK {
			return false
		}
	}
	if hello.SupportedCurves != nil {
		ecdsaOK := false
		for _, curve := range hello.SupportedCurves {
			if curve == tls.CurveP256 {
				ecdsaOK = true
				break
			}
		}
		if !ecdsaOK {
			return false
		}
	}
	for _, suite := range hello.CipherSuites {
		switch suite {
		case tls.TLS_ECDHE_ECDSA_WITH_RC4_128_SHA,
			tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
			tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
			tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256,
			tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
			tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
			tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305:
			return true
		}
	}
	return false
}

// localCertTTL bounds how long a last-resort read is remembered. Short
// enough that a renewal reaching the local cache is picked up promptly, long
// enough that a flood of handshakes for one name is not a file read each.
const localCertTTL = 30 * time.Second

type localCertEntry struct {
	cert *tls.Certificate
	at   time.Time
}

// localCertificate is the last layer: this box's own autocert cache, read
// DIRECTLY, because a gated autocert.Manager refuses the name before it ever
// looks there. Both outcomes are remembered briefly so a busy name costs one
// file read per TTL rather than one per handshake.
func (m *certMirror) localCertificate(ctx context.Context, name string, wantRSA bool) *tls.Certificate {
	if m.local == nil || name == "" {
		return nil
	}
	for _, key := range certEntryNames(name, wantRSA) {
		m.localMu.Lock()
		e, ok := m.localCerts[key]
		fresh := ok && time.Since(e.at) < localCertTTL
		m.localMu.Unlock()
		if fresh {
			if usableFor(e.cert, name) {
				return e.cert
			}
			continue
		}
		var cert *tls.Certificate
		if raw, err := m.local.Get(ctx, key); err == nil {
			if c, err := parseCertEntry(raw); err == nil {
				cert = c
			}
		}
		m.localMu.Lock()
		if len(m.localCerts) > readThroughCacheMax {
			// Crude but bounded: the map is a cache, and every entry is
			// re-readable from disk.
			clear(m.localCerts)
		}
		m.localCerts[key] = localCertEntry{cert: cert, at: time.Now()}
		m.localMu.Unlock()
		if usableFor(cert, name) {
			// Worth saying: the mirror had nothing for a name this box is
			// serving anyway. Once a minute for the whole box, not once per
			// name — every host on a box whose bucket is unreachable lands here
			// at once, and one line is enough to know it.
			m.localMu.Lock()
			quiet := time.Since(m.lastLocalLog) < time.Minute
			if !quiet {
				m.lastLocalLog = time.Now()
			}
			m.localMu.Unlock()
			if !quiet {
				log.Printf("edge: serving %s from this box's local cert cache — the mirror has no copy and issuance is gated here", name)
			}
			return cert
		}
	}
	return nil
}

// ---- the :80 challenge path -------------------------------------------------

const acmeChallengePrefix = "/.well-known/acme-challenge/"

// challengeHandler answers HTTP-01 challenges from the mirror BEFORE
// autocert's own handler sees them, and delegates everything else. The order
// is forced, again, by autocert: HTTPHandler runs HostPolicy first
// (autocert.go:405), so on a non-issuer it 403s a challenge whose answer is
// sitting in the mirror. Behind this, autocert still answers its OWN live
// orders from memory, which is what the issuer needs.
func (m *certMirror) challengeHandler(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !strings.HasPrefix(r.URL.Path, acmeChallengePrefix) {
			next.ServeHTTP(w, r)
			return
		}
		// path.Base is the token autocert itself would derive
		// (httpTokenCacheKey), so both handlers agree on the name.
		if body, ok := m.challengeResponse(r.Context(), requestHost(r.Host), path.Base(r.URL.Path)); ok {
			w.Header().Set("Content-Type", "text/plain; charset=utf-8")
			_, _ = w.Write(body)
			return
		}
		next.ServeHTTP(w, r)
	})
}

// challengeResponse answers from the polled map, then — and this is the part
// the 30s poll cannot do — reads through to the bucket. A validation's whole
// life is a few seconds: the issuer writes the token, sends Accept, and the
// CA arrives at whichever edge DNS handed it. A poll that runs every 30s
// misses that window every time.
func (m *certMirror) challengeResponse(ctx context.Context, host, token string) ([]byte, bool) {
	if body, ok := m.Challenge(token); ok {
		// Already polled: a map hit costs nothing and the token is the CA's own
		// 128+ bits of entropy, so it needs no host bound of its own.
		return body, true
	}
	if !validChallengeToken(token) {
		return nil, false
	}
	// The LIVE lookup is bounded by the host under validation, exactly as
	// autocert bounds its own token lookup (HTTPHandler runs hostPolicy on
	// r.Host first, autocert.go:405). A real CA always sends the domain it is
	// validating, so the bound costs a genuine validation nothing — and
	// without it an unauthenticated :80 flood sustains a bucket GET per
	// distinct path, forever, from anywhere. The bound is the DEPLOYMENT's
	// exact names, not this box's (m.allowed, Edge.deploymentHost): DNS for a
	// home-boxed route can point at any edge, and the one it points at is the
	// one the CA arrives at (#277).
	if m.allowed != nil && !m.allowed(host) {
		return nil, false
	}
	body, _, ok := m.readThrough(ctx, token+httpTokenSuffix, &m.rtHTTP)
	return body, ok
}

// requestHost is the Host header without its port, normalized the same way
// SNI is (serverName) so it meets Edge.Allowed's exact registry map in the
// shape routes are stored in.
func requestHost(hostport string) string {
	host := hostport
	if h, _, err := net.SplitHostPort(host); err == nil {
		host = h
	}
	return serverName(host)
}

// validChallengeToken bounds what an unauthenticated URL path may turn into a
// bucket read: ACME tokens are base64url of at least 128 bits of entropy, so
// nothing shorter or otherwise shaped can be one. It also keeps the key
// inside the flat prefix (publishableCertName's concern, from the other
// side).
func validChallengeToken(token string) bool {
	if len(token) < 16 || len(token) > 128 {
		return false
	}
	for _, r := range token {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_':
		default:
			return false
		}
	}
	return true
}

// alpnTokenCert answers a tls-alpn-01 validation for an order some OTHER edge
// is running. autocert prefers tls-alpn-01 over http-01 and only falls back
// to it after a whole failed order (autocert.go's AuthorizeOrderLoop), so
// without this every issuance in an N-edge deployment burns a failed
// validation with probability (N-1)/N before it ever tries the HTTP path.
//
// Bounded harder than the HTTP-01 path, because the name here is
// attacker-chosen SNI rather than a high-entropy token: only names this
// deployment serves — this box's own routes, or another box's exactly-named
// ones (m.allowed, Edge.deploymentHost) — are ever looked up. A stranger's
// invented label under a mirrored catch-all is not one of them.
func (m *certMirror) alpnTokenCert(ctx context.Context, host string) *tls.Certificate {
	if host == "" {
		return nil
	}
	if m.allowed != nil && !m.allowed(host) {
		return nil
	}
	// Deliberately NOT from the poll's snapshot. Unlike an HTTP-01 entry —
	// whose name carries the order's unique token — a token CERT is keyed by
	// hostname alone, so the previous order's entry for the same host is a
	// live collision, not a stale copy of the same answer. Serving it hands
	// the CA the wrong key authorization and burns the order. The snapshot is
	// a poll behind at best, and far worse if a gated Delete left litter
	// behind, so this material is only ever read live.
	_, cert, _ := m.readThrough(ctx, host+alpnTokenSuffix, &m.rtALPN)
	if usableFor(cert, host) {
		return cert
	}
	return nil
}

// ---- read-through: the bounded live lookup ----------------------------------

const (
	// readThroughTimeout budgets one live bucket read. It runs inside a
	// handshake or a :80 request, so it is short by construction.
	readThroughTimeout = 2 * time.Second
	// readThroughHitTTL is about one poll: after that the poll itself holds
	// the entry (or it has been withdrawn, which is also the answer). It
	// applies to names that are unique per order — HTTP-01 tokens.
	readThroughHitTTL = 30 * time.Second
	// readThroughMissTTL keeps a scanner walking the challenge path from
	// turning into a bucket GET per request.
	readThroughMissTTL = 5 * time.Second
	// readThroughCacheMax bounds the memo; over it, the map is dropped whole.
	readThroughCacheMax = 512
	// readThroughBurst / readThroughRefill are one budget's rate: distinct
	// names defeat the negative cache, so the last bound is a rate. Under a
	// flood the read-through degrades to "off" — the 30s poll still covers
	// everything that is not a live validation.
	//
	// There is one budget PER LISTENER (rtHTTP, rtALPN), never one shared:
	// :80's challenge path is reachable by anyone, and a flood there must not
	// spend the budget 443's tls-alpn-01 rescue needs. They fail
	// independently or not at all.
	readThroughBurst  = 20
	readThroughRefill = 5 // per second
)

// rtBudget is one token bucket. Guarded by certMirror.rtMu.
type rtBudget struct {
	tokens float64
	last   time.Time
}

// spend takes one token, reporting whether there was one. Caller holds rtMu.
func (b *rtBudget) spend(now time.Time) bool {
	if b.last.IsZero() {
		b.tokens = readThroughBurst
	} else {
		b.tokens = min(readThroughBurst, b.tokens+now.Sub(b.last).Seconds()*readThroughRefill)
	}
	b.last = now
	if b.tokens < 1 {
		return false
	}
	b.tokens--
	return true
}

type readThroughEntry struct {
	body []byte
	cert *tls.Certificate
	ok   bool
	at   time.Time
}

// readThroughTokenTTL is how long a tls-alpn-01 token CERT may be remembered.
// Seconds, not a poll: the name is the bare hostname, so the next order for
// that host reuses it, and two orders can be seconds apart. A var so tests
// can zero it.
var readThroughTokenTTL = 3 * time.Second

// readThroughTTL is how long a HIT for name may be reused.
func readThroughTTL(name string) time.Duration {
	if strings.HasSuffix(name, alpnTokenSuffix) {
		return readThroughTokenTTL
	}
	return readThroughHitTTL
}

func (e readThroughEntry) fresh(name string) bool {
	ttl := readThroughMissTTL
	if e.ok {
		ttl = readThroughTTL(name)
	}
	return time.Since(e.at) < ttl
}

// readThrough fetches one control/certs/ entry outside the poll cycle: memoed
// (hits and misses), single-flighted per name, rate-limited globally, and
// budgeted per call. Errors are answers here, not faults — a bucket that is
// down simply means this box cannot help with someone else's live
// validation, and the CA retries.
func (m *certMirror) readThrough(ctx context.Context, name string, budget *rtBudget) ([]byte, *tls.Certificate, bool) {
	m.rtMu.Lock()
	if e, ok := m.rt[name]; ok && e.fresh(name) {
		m.rtMu.Unlock()
		return e.body, e.cert, e.ok
	}
	if done, ok := m.rtInflight[name]; ok {
		// Somebody is already asking. The CA opens several connections for one
		// validation; they share the one bucket read.
		m.rtMu.Unlock()
		select {
		case <-done:
		case <-ctx.Done():
			return nil, nil, false
		}
		m.rtMu.Lock()
		e := m.rt[name]
		m.rtMu.Unlock()
		return e.body, e.cert, e.ok
	}
	if !budget.spend(time.Now()) {
		m.rtMu.Unlock()
		return nil, nil, false
	}
	done := make(chan struct{})
	m.rtInflight[name] = done
	m.rtMu.Unlock()

	// Detached from the caller: a client that hangs up mid-validation must not
	// cancel the read every other in-flight waiter is sharing.
	fetchCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), readThroughTimeout)
	body, _, err := m.s.Get(fetchCtx, certKey(name))
	cancel()
	e := readThroughEntry{at: time.Now()}
	switch {
	case err == nil:
		e.body, e.ok = body, true
		if certShapedName(name) {
			if cert, perr := parseCertEntry(body); perr == nil {
				e.cert = cert
			} else {
				log.Printf("certsync: live read of %q does not parse as a certificate: %v", name, perr)
			}
		}
	case errors.Is(err, os.ErrNotExist):
		// The ordinary case: no such token. Negative-cached, not logged.
	default:
		m.logReadThroughError(name, err)
	}
	m.rtMu.Lock()
	if len(m.rt) > readThroughCacheMax {
		clear(m.rt)
	}
	m.rt[name] = e
	delete(m.rtInflight, name)
	m.rtMu.Unlock()
	close(done)
	return e.body, e.cert, e.ok
}

// logReadThroughError says once a minute that live lookups are failing. Not
// per call: the rate limiter still allows several a second, and a bucket
// outage is one fault however many validations hit it.
func (m *certMirror) logReadThroughError(name string, err error) {
	m.rtMu.Lock()
	quiet := time.Since(m.lastRTLog) < time.Minute
	if !quiet {
		m.lastRTLog = time.Now()
	}
	m.rtMu.Unlock()
	if !quiet {
		log.Printf("certsync: live read of %q failed: %v — the %s poll still covers everything but live validations", name, err, certMirrorInterval)
	}
}