// The relay's write side (relay spec §5): the writer lease that makes one
// console the only process writing relay state, the operations that turn it
// on and off for an app, the etag-first publish of the authority record every
// relay box reads, and the five-minute reconcile that converges both derived
// artefacts from the records whatever happened in between.
//
// THREE OBJECTS, ONE SOURCE OF TRUTH. `apps/<acct>/<app>/relay.json` (the
// relay.Record) is the truth for an app. `control/relay.json` (the
// relay.Authority) is DERIVED from every record and never patched:
// publishing it is read-the-etag-first, list, build, CAS, and start over on
// conflict, so a disable that lands during the listing makes the write fail
// rather than quietly restore the app. `control/relay-issuer.json` is the
// unsealed pointer saying WHICH staged master a region issues from; adding a
// master to the console's secret issues nothing until that pointer moves.
//
// ONE WRITER, ENFORCED, AND THE WALK IS RESUMABLE. Every operation here
// writes its intent to the record first (`op` bumped, `phase` set), then
// writes config, then publishes, advancing `phase` after each step. That is
// what makes an interrupted operation finishable: the reconcile does not look
// for unfinished records, it re-derives what every record implies and fixes
// whatever disagrees. The lease is the fence around all of it — a process
// that loses a renewal stops before its next bucket write and abandons the
// operation, and the next holder's reconcile finishes it.
//
// THE CONSOLE HOLDS MASTERS BUT OPENS NOTHING. relayMasters reads the
// region's masters out of this console's OWN environment
// (CARLOS_RELAY_MASTERS_<REGION>, delivered like every other console
// variable) and derives an app key from them; the value it delivers is
// SEALED with CARLOS_SECRETS_PUBKEY, which this console cannot open — the
// same one-way street apiconfig.go's package comment describes, and the
// reason a console with no sealing key refuses to enable rather than mint a
// key it could never deliver.
//
// Spec: docs/superpowers/specs/2026-09-07-relay-stun-turn-design.md §4, §5.
package console

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"maps"
	"os"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/relay"
	"github.com/carlosframework/platform/internal/release"
)

const (
	// relayWriterKey is the writer lease: {holder, expires_at}, CAS-written.
	relayWriterKey = "control/relay-writer.json"
	// relayIssuerKey is the issuer pointer: {"<region>": "<kid>"}. Unsealed
	// on purpose — it names a key, it is not one.
	relayIssuerKey = "control/relay-issuer.json"
	// relayAckPrefix is where each armed relay writes its heartbeat.
	relayAckPrefix = "control/relay-ack/"
	// configAckPrefix is where each box's config sync writes a per-route
	// config-adoption acknowledgement (spec §5, internal/appconfig/
	// configack.go): configAckPrefix + "<box>/<route host>.json".
	configAckPrefix = "control/config-ack/"

	// The lease's two numbers, from spec §5: sixty seconds of validity,
	// renewed every twenty. Three renewals inside one TTL means a single
	// slow bucket round trip costs nothing, while a console that dies is
	// replaced within a minute.
	relayLeaseTTL   = 60 * time.Second
	relayLeaseRenew = 20 * time.Second
	// relayLeaseWait bounds how long StartRelay waits for the lease before
	// its first reconcile. A console that does not get it simply reconciles
	// nothing — the holder is doing it.
	relayLeaseWait = 5 * time.Second

	// relayReconcileInterval is the convergence floor from spec §5: a stale
	// config write is undone, an interrupted enable delivered, and a
	// half-done deletion repaired, within this.
	relayReconcileInterval = 5 * time.Minute

	// relayHeartbeatFresh is how recent a relay's last SUCCESSFUL authority
	// read must be for its heartbeat to count as a live answer — the same
	// five minutes `carlos relay status` calls "stale" past.
	relayHeartbeatFresh = 5 * time.Minute
	// relayRetireAfter is the earliest a rotated-away master may be retired
	// (spec §5, step 4): 28 hours, comfortably past the four-hour ceiling on
	// any credential minted just before the pointer moved.
	relayRetireAfter = 28 * time.Hour

	// relayRouteMax caps the instance records one delivery check reads. An
	// app with more routes than this has bigger questions than the relay.
	relayRouteMax = 500

	// relayVar is the one variable this whole file exists to deliver.
	relayVar = "CARLOS_RELAY"
)

// The three phases of an operation's walk (spec §5). `done` is the only one
// that means "nothing is owed".
const (
	relayPhaseConfig    = "config"
	relayPhaseAggregate = "aggregate"
	relayPhaseDone      = "done"
)

// The refusals callers (and, in Task 11, doors) match on. Each is a
// configuration fact rather than a failure of the request, which is why they
// are sentinels rather than sentences built at the call site: the door turns
// them into a 503 naming the variable, and the CLI prints them verbatim.
var (
	errRelayNoBucket  = errors.New("no deployment bucket on this host — this console cannot serve the relay")
	errRelayNoRegions = errors.New(
		"this deployment runs no relay (CARLOS_RELAY_REGIONS is empty) — ask the operator to arm a region before enabling it for an app")
	errRelayNoSealKey = errors.New(
		"this console has no secrets key (CARLOS_SECRETS_PUBKEY) — it cannot deliver a relay key, so it will not issue one")
	// The commonest cause is not another console at all: it is THIS console
	// having just restarted, leaving its own previous process's lease to
	// expire. That takes up to relayLeaseTTL, so the message says to retry
	// and names the wait rather than leaving an operator to guess whether
	// the deployment is wedged (discussion #120).
	errRelayNotHeld = errors.New(
		"this console does not hold the relay writer lease. another console is driving the relay, " +
			"or this one restarted and its previous lease has not expired yet. the operation was not started, " +
			"so retry in up to a minute")
	errRelayNotEnabled  = errors.New("the relay is not enabled for this app")
	errRelayUndelivered = errors.New("a route has no config delivery")
	// errRelayUnknownHost is `carlos relay adopted --host` naming something
	// that is not a route of this app — the door's 404.
	errRelayUnknownHost = errors.New("no such route on this app")
	// errRelayNotMemberRoute is `carlos relay adopted --host` naming a route
	// the config sync restarts itself. Spec §5, step 5 gives a route exactly
	// one way to establish its cutover, decided by shape: a unit or
	// socket-exec route's cutover is the ack the config sync writes after it
	// restarted the process, and a member's word would override an observed
	// fact with a claim.
	errRelayNotMemberRoute = errors.New("this route's cutover is observed, not declared")
)

// relayLease is control/relay-writer.json.
type relayLease struct {
	Holder    string    `json:"holder"`
	ExpiresAt time.Time `json:"expires_at"`
}

// relayHolderID draws this process's lease identity: 16 random hex
// characters, like an incarnation and for the same reason — two consoles
// must never collide, and a counter or a hostname can.
func relayHolderID() string {
	var b [8]byte
	if _, err := rand.Read(b[:]); err != nil {
		panic("console: crypto/rand: " + err.Error())
	}
	return hex.EncodeToString(b[:])
}

// relayTime is every relay path's clock, seamed for the lease tests.
func (s *Server) relayTime() time.Time {
	if s.relayNow != nil {
		return s.relayNow()
	}
	return time.Now()
}

// relayHold reports whether this process currently holds the writer lease.
func (s *Server) relayHold() bool { return s.relayHeld.Load() }

// requireRelayLease is the check every relay bucket write makes FIRST. A
// process that lost its renewal stops here rather than completing a walk
// whose next step another console has already started — spec §5's "stops
// every relay write in flight before its next bucket write".
func (s *Server) requireRelayLease() error {
	if !s.relayHold() {
		return errRelayNotHeld
	}
	return nil
}

// relayAcquire is one acquire-or-renew attempt, by CAS on the lease object.
// Absent, expired, or already ours means we take it; anyone else's live lease
// means we do not, and — this is the load-bearing half — we drop `held` so
// any operation in flight in this process fails its next check.
func (s *Server) relayAcquire(ctx context.Context) error {
	if s.cfg.Releases == nil {
		return errRelayNoBucket
	}
	now := s.relayTime()
	body, etag, err := s.cfg.Releases.Get(ctx, relayWriterKey)
	switch {
	case errors.Is(err, os.ErrNotExist):
		body, etag = nil, ""
	case err != nil:
		s.relayHeld.Store(false)
		return err
	}
	var cur relayLease
	if len(body) > 0 {
		if err := json.Unmarshal(body, &cur); err != nil {
			// An unparseable lease is not a live one: it names nobody this
			// process can defer to. Taking it over with the etag we read is
			// still a CAS, so a concurrent repair wins the race, not us.
			log.Printf("console: relay writer lease is unreadable, taking it over: %v", err)
			cur = relayLease{}
		}
	}
	if cur.Holder != "" && cur.Holder != s.relayHolder && now.Before(cur.ExpiresAt) {
		s.relayHeld.Store(false)
		return fmt.Errorf("relay writer lease is held by %s until %s: %w",
			cur.Holder, cur.ExpiresAt.UTC().Format(time.RFC3339), errRelayNotHeld)
	}
	next, err := json.Marshal(relayLease{Holder: s.relayHolder, ExpiresAt: now.Add(relayLeaseTTL)})
	if err != nil {
		s.relayHeld.Store(false)
		return err
	}
	if _, err := s.cfg.Releases.Put(ctx, relayWriterKey, next, etag); err != nil {
		// A lost CAS is another console taking the same lapsed lease. Held
		// goes false either way: a write we did not land is a lease we do
		// not hold.
		s.relayHeld.Store(false)
		return err
	}
	s.relayHeld.Store(true)
	return nil
}

// relayWriterLoop acquires the lease now and renews it every twenty seconds
// until ctx ends. It is the ONLY writer of s.relayHeld outside a test.
//
// A failed attempt is logged at most as a line, never fatal: the ordinary
// steady state of a two-console deployment is one process failing this call
// forever while the other holds the lease, and a deployment overlap must not
// fill a log with alarm.
func (s *Server) relayWriterLoop(ctx context.Context) {
	t := time.NewTicker(relayLeaseRenew)
	defer t.Stop()
	held := false
	for {
		err := s.relayAcquire(ctx)
		switch {
		case err == nil && !held:
			log.Printf("console: relay writer lease acquired (holder %s)", s.relayHolder)
			held = true
		case err != nil && held:
			log.Printf("console: relay writer lease lost: %v", err)
			held = false
		case err != nil && !errors.Is(err, errRelayNotHeld) && !errors.Is(err, release.ErrCASConflict):
			log.Printf("console: relay writer lease: %v", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// StartRelay runs the relay's two background loops: the writer lease, and the
// converging reconcile every five minutes and once at start. Started from
// cmd/console beside the other passes — never from NewServer, which stays
// goroutine-free so constructing a Server in a test starts nothing.
//
// Gated on a configured region list rather than on holding the lease: a
// deployment with no relay has nothing to converge and should not be reading
// a lease object every twenty seconds forever.
func (s *Server) StartRelay(ctx context.Context) {
	if s.cfg.Releases == nil || len(s.relayRegions()) == 0 {
		return
	}
	go s.relayWriterLoop(ctx)
	// The first reconcile is the one that finishes whatever the previous
	// holder abandoned, so it is worth a short wait for the lease rather
	// than five minutes of drift.
	s.relayAwaitLease(ctx, relayLeaseWait)
	t := time.NewTicker(relayReconcileInterval)
	defer t.Stop()
	for {
		if s.relayHold() {
			if err := s.relayReconcile(ctx); err != nil {
				log.Printf("console: relay reconcile: %v", err)
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// relayAwaitLease waits up to d for the writer loop's first acquire.
func (s *Server) relayAwaitLease(ctx context.Context, d time.Duration) {
	deadline := time.Now().Add(d)
	for !s.relayHold() && time.Now().Before(deadline) {
		select {
		case <-ctx.Done():
			return
		case <-time.After(25 * time.Millisecond):
		}
	}
}

// ---- masters and the issuer pointer -----------------------------------------

// relayMastersEnv is the console variable holding one region's masters:
// CARLOS_RELAY_MASTERS_EU_WEST_1 for eu-west-1 (uppercased, hyphens to
// underscores, since variable names refuse hyphens — spec §4).
func relayMastersEnv(region string) string {
	return "CARLOS_RELAY_MASTERS_" + strings.ToUpper(strings.ReplaceAll(region, "-", "_"))
}

// relayStagedMaster is one `kid:key` pair out of that variable, in the order
// the operator wrote them.
type relayStagedMaster struct {
	Kid string
	Key []byte
}

// relayStagedMasters parses CARLOS_RELAY_MASTERS_<REGION> — read from the
// process environment AT USE TIME rather than cached at boot, so an operator
// who stages a master and restarts nothing still has it available to the next
// rotate.
//
// Every defect is a refusal naming the variable: this is key material, and a
// half-parsed list would silently issue from the wrong master.
func relayStagedMasters(region string) ([]relayStagedMaster, error) {
	name := relayMastersEnv(region)
	raw := strings.TrimSpace(os.Getenv(name))
	if raw == "" {
		return nil, fmt.Errorf("this console holds no master for region %s (%s is unset) — ask the operator to stage one", region, name)
	}
	var out []relayStagedMaster
	for _, pair := range strings.Split(raw, ",") {
		pair = strings.TrimSpace(pair)
		if pair == "" {
			continue
		}
		kid, keyB64, ok := strings.Cut(pair, ":")
		if !ok {
			return nil, fmt.Errorf("%s holds an entry that is not <kid>:<base64 key>", name)
		}
		kid = strings.TrimSpace(kid)
		if !relay.ValidKid(kid) {
			return nil, fmt.Errorf("%s holds an invalid kid %q (1-8 lowercase letters or digits)", name, kid)
		}
		key, err := base64.StdEncoding.DecodeString(strings.TrimSpace(keyB64))
		if err != nil || len(key) != relay.MasterKeyLen {
			return nil, fmt.Errorf("%s: master %s must be base64 of %d bytes", name, kid, relay.MasterKeyLen)
		}
		if slices.ContainsFunc(out, func(m relayStagedMaster) bool { return m.Kid == kid }) {
			return nil, fmt.Errorf("%s names kid %s twice", name, kid)
		}
		out = append(out, relayStagedMaster{Kid: kid, Key: key})
	}
	if len(out) == 0 {
		return nil, fmt.Errorf("%s is set but holds no <kid>:<base64 key> entry", name)
	}
	return out, nil
}

// relayIssuers reads the issuer pointer with its etag. Absent is an empty
// map, not an error: a deployment that has never rotated has never written
// one, and the fallback below (the first staged master) is what it means.
func (s *Server) relayIssuers(ctx context.Context) (map[string]string, string, error) {
	body, etag, err := s.cfg.Releases.Get(ctx, relayIssuerKey)
	if errors.Is(err, os.ErrNotExist) {
		return map[string]string{}, "", nil
	}
	if err != nil {
		return nil, "", err
	}
	out := map[string]string{}
	if err := json.Unmarshal(body, &out); err != nil {
		return nil, "", fmt.Errorf("console: %s: %w", relayIssuerKey, err)
	}
	return out, etag, nil
}

// relayMasters answers which master a region ISSUES from right now: the kid
// the issuer pointer names, or — when the pointer says nothing about this
// region — the FIRST entry in the console's variable, which is what a
// deployment that has never rotated has.
//
// A pointer naming a kid this console does not hold is an error rather than a
// fallback. It means the operator moved the pointer without staging the key
// here (or removed the key too early), and issuing from a different master
// than the pointer names would mint credentials every relay refuses.
func (s *Server) relayMasters(ctx context.Context, region string) (string, []byte, error) {
	staged, err := relayStagedMasters(region)
	if err != nil {
		return "", nil, err
	}
	issuers, _, err := s.relayIssuers(ctx)
	if err != nil {
		return "", nil, err
	}
	kid, ok := issuers[region]
	if !ok || kid == "" {
		return staged[0].Kid, staged[0].Key, nil
	}
	for _, m := range staged {
		if m.Kid == kid {
			return m.Kid, m.Key, nil
		}
	}
	return "", nil, fmt.Errorf("%s names kid %s as region %s's issuer, but %s does not hold it",
		relayIssuerKey, kid, region, relayMastersEnv(region))
}

// relayRegionMasters gathers the issuing master for each of an app's regions
// — everything relay.BuildConfig needs, and nothing it does not.
func (s *Server) relayRegionMasters(ctx context.Context, regions []string) ([]relay.RegionMaster, error) {
	out := make([]relay.RegionMaster, 0, len(regions))
	for _, r := range regions {
		kid, master, err := s.relayMasters(ctx, r)
		if err != nil {
			return nil, err
		}
		out = append(out, relay.RegionMaster{Region: r, Kid: kid, Master: master})
	}
	return out, nil
}

// ---- regions, hostnames, and the config value -------------------------------

// relayRegions is the deployment's region list, normalized and deduped, in
// the operator's order.
func (s *Server) relayRegions() []string {
	var out []string
	for _, r := range s.cfg.RelayRegions {
		r = strings.ToLower(strings.TrimSpace(r))
		if r != "" && !slices.Contains(out, r) {
			out = append(out, r)
		}
	}
	return out
}

// relayDomain is the domain relay hostnames hang off. CARLOS_PLATFORM_DOMAIN
// when the operator set it; otherwise the apps domain with a leading "apps."
// label stripped, which is the one shape this console can derive safely
// (apps.oncarlos.com -> oncarlos.com; oncarlos.com -> oncarlos.com).
func (s *Server) relayDomain() string {
	if d := strings.ToLower(strings.TrimSpace(s.cfg.PlatformDomain)); d != "" {
		return d
	}
	d := strings.ToLower(strings.TrimSpace(s.cfg.AppsDomain))
	if label, rest, ok := strings.Cut(d, "."); ok && label == "apps" {
		return rest
	}
	return d
}

// relayURLs is the ICE server list for one region (spec §2): the STUN
// listener and its two TURN transports, all on one port.
func (s *Server) relayURLs(region string) []string {
	host := "relay." + region + "." + s.relayDomain() + ":3478"
	return []string{
		"stun:" + host,
		"turn:" + host + "?transport=udp",
		"turn:" + host + "?transport=tcp",
	}
}

// relayReady is the three configuration facts every operation needs before it
// touches anything: a bucket to write, a region to issue for, and a key to
// seal with. Checked together, and BEFORE the first write, for storeSealKey's
// reason: a key minted without delivery is worse than no key at all.
func (s *Server) relayReady() error {
	if s.cfg.Releases == nil {
		return errRelayNoBucket
	}
	if len(s.relayRegions()) == 0 {
		return errRelayNoRegions
	}
	if s.cfg.SecretsPub == "" {
		return errRelayNoSealKey
	}
	return nil
}

// relayResolveRegions turns an app's request into the region list its record
// carries: every armed region in the CONSOLE's order when the caller named
// none, or the named subset — still in the console's order, since that order
// is the one every browser tries them in.
func (s *Server) relayResolveRegions(want []string) ([]string, error) {
	armed := s.relayRegions()
	if len(armed) == 0 {
		return nil, errRelayNoRegions
	}
	var asked []string
	for _, r := range want {
		if r = strings.ToLower(strings.TrimSpace(r)); r != "" {
			asked = append(asked, r)
		}
	}
	if len(asked) == 0 {
		return armed, nil
	}
	for _, r := range asked {
		if !slices.Contains(armed, r) {
			return nil, fmt.Errorf("this deployment runs no relay in %s — armed regions are %s",
				r, strings.Join(armed, ", "))
		}
	}
	out := make([]string, 0, len(asked))
	for _, r := range armed {
		if slices.Contains(asked, r) {
			out = append(out, r)
		}
	}
	return out, nil
}

// relayConfigValue builds the CARLOS_RELAY value for a record and the
// username suffixes it issues, from the masters each region issues from now.
func (s *Server) relayConfigValue(ctx context.Context, acct, app string, rec *relay.Record) (string, map[string]string, error) {
	masters, err := s.relayRegionMasters(ctx, rec.Regions)
	if err != nil {
		return "", nil, err
	}
	return relay.BuildConfig(acct, app, rec, masters, s.relayURLs)
}

// relayExpectedSuffixes is what an enabled record SHOULD have issued, given
// the masters in force right now — the reconcile's comparison value.
//
// Suffixes, not the sealed value: the console cannot open an envelope it
// wrote, and two seals of identical plaintext differ anyway (each carries a
// fresh ephemeral key). The suffix carries account, app, incarnation and kid
// — every identity the value derives from — so a suffix set that matches is
// a value that matches, and a rotation, a re-enable or a region change all
// move it.
func (s *Server) relayExpectedSuffixes(ctx context.Context, acct, app string, rec *relay.Record) (map[string]string, error) {
	masters, err := s.relayRegionMasters(ctx, rec.Regions)
	if err != nil {
		return nil, err
	}
	out := make(map[string]string, len(masters))
	for _, m := range masters {
		out[m.Region] = relay.Suffix(acct, app, rec.Incarnation, m.Kid)
	}
	return out, nil
}

// relayEnvelopeDigest identifies a sealed envelope without opening it:
// hex(sha256) of the envelope string as it sits in the bundle. It is what
// makes "the bundle still holds exactly what we wrote" checkable by a console
// that holds no decryption key — see relay.Record.IssuedDigest.
func relayEnvelopeDigest(envelope string) string {
	sum := sha256.Sum256([]byte(envelope))
	return hex.EncodeToString(sum[:])
}

// relayConfigVar reads the app's delivered CARLOS_RELAY entry, if any.
func (s *Server) relayConfigVar(ctx context.Context, acct, app string) (appconfig.Var, bool, error) {
	m, err := appconfig.Get(ctx, s.cfg.Releases, acct, app)
	if err != nil {
		return appconfig.Var{}, false, err
	}
	v, ok := m.Vars[relayVar]
	return v, ok, nil
}

// relayApplyConfig derives the app's key, seals it and writes it into the
// DEFAULT config bundle, recording on the record which identities it just
// issued. It mutates rec.IssuedSuffixes; the caller writes the record.
func (s *Server) relayApplyConfig(ctx context.Context, acct, app string, rec *relay.Record) error {
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	if s.cfg.SecretsPub == "" {
		return errRelayNoSealKey
	}
	value, suffixes, err := s.relayConfigValue(ctx, acct, app, rec)
	if err != nil {
		return err
	}
	pub, err := appconfig.ParsePublicKey(s.cfg.SecretsPub, "CARLOS_SECRETS_PUBKEY")
	if err != nil {
		return err
	}
	env, err := appconfig.Seal(pub, appconfig.Binding{Account: acct, App: app}, []byte(value))
	if err != nil {
		return err
	}
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	if _, err := appconfig.Apply(ctx, s.cfg.Releases, acct, app,
		map[string]appconfig.Var{relayVar: {Kind: "secret", Envelope: env}}, nil); err != nil {
		return err
	}
	rec.IssuedSuffixes = suffixes
	rec.IssuedDigest = relayEnvelopeDigest(env)
	return nil
}

// relayUnsetConfig removes CARLOS_RELAY from the app's default bundle, and
// reports whether there was anything to remove. It READS first because an
// unset-only Apply against an app with no bundle at all is an error
// (appconfig.applyAt), and "this app has no config" is not a failed disable.
func (s *Server) relayUnsetConfig(ctx context.Context, acct, app string) (bool, error) {
	if err := s.requireRelayLease(); err != nil {
		return false, err
	}
	_, present, err := s.relayConfigVar(ctx, acct, app)
	if err != nil || !present {
		return false, err
	}
	if err := s.requireRelayLease(); err != nil {
		return false, err
	}
	n, err := appconfig.Apply(ctx, s.cfg.Releases, acct, app, nil, []string{relayVar})
	return n > 0, err
}

// ---- the record walk --------------------------------------------------------

// relayAppLock is the per-app mutex every operation and the reconcile's
// per-record work takes. The lease keeps two CONSOLES apart; this keeps two
// handlers inside one console apart, which the lease says nothing about.
func (s *Server) relayAppLock(acct, app string) *sync.Mutex {
	v, _ := s.relayAppLocks.LoadOrStore(acct+"/"+app, &sync.Mutex{})
	return v.(*sync.Mutex)
}

// relayPutRecord is relay.PutRecord behind the lease check. EVERY record
// write in this file goes through it rather than calling PutRecord directly:
// a walk is several writes long, the lease can lapse between any two of them,
// and spec §5's rule is a check before each bucket write, not one per
// operation.
func (s *Server) relayPutRecord(ctx context.Context, acct, app string, rec *relay.Record, etag string) (string, error) {
	if err := s.requireRelayLease(); err != nil {
		return "", err
	}
	return relay.PutRecord(ctx, s.cfg.Releases, acct, app, rec, etag)
}

// relayAppRecord is one app's record with the coordinates that name it and
// the etag it was read at — the unit both the publish listing and the
// reconcile walk over.
type relayAppRecord struct {
	Account string
	App     string
	Record  *relay.Record
	Etag    string
}

// relayListRecords walks every app in the bucket and returns the ones that
// have a relay record. Two delimiter-aware listings (accounts, then their
// apps) rather than one flat List of the whole `apps/` prefix: a deployment's
// object count is dominated by releases, and this must not read them.
//
// A record that will not parse is logged and skipped, not fatal. The
// alternative is one corrupt object stopping the aggregate from being
// published for every other app in the deployment.
func (s *Server) relayListRecords(ctx context.Context) ([]relayAppRecord, error) {
	accts, err := s.cfg.Releases.ListDirs(ctx, "apps/")
	if err != nil {
		return nil, err
	}
	var out []relayAppRecord
	for _, acctPrefix := range accts {
		acct := strings.Trim(strings.TrimPrefix(acctPrefix, "apps/"), "/")
		if !release.ValidAccount(acct) {
			continue
		}
		apps, err := s.cfg.Releases.ListDirs(ctx, acctPrefix)
		if err != nil {
			return nil, err
		}
		for _, appPrefix := range apps {
			app := strings.Trim(strings.TrimPrefix(appPrefix, acctPrefix), "/")
			if !release.ValidApp(app) {
				continue
			}
			rec, etag, err := relay.GetRecord(ctx, s.cfg.Releases, acct, app)
			if err != nil {
				log.Printf("console: relay record %s/%s: %v", acct, app, err)
				continue
			}
			if rec == nil {
				continue
			}
			out = append(out, relayAppRecord{Account: acct, App: app, Record: rec, Etag: etag})
		}
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Account != out[j].Account {
			return out[i].Account < out[j].Account
		}
		return out[i].App < out[j].App
	})
	return out, nil
}

// relayPublishOnce is ONE attempt at spec §5's publish sequence, from an etag
// the CALLER read before it did anything else: list every record, build the
// authority from the enabled ones whatever their phase, and write with that
// etag. A conflict comes straight back — this is the function the retry loop
// wraps, and the function a test can hand a deliberately stale etag.
//
// The etag being an argument is the whole design. Read after the listing, it
// would let a disable that landed mid-listing be overwritten by an aggregate
// built before it; read before, that same disable makes this write fail.
func (s *Server) relayPublishOnce(ctx context.Context, etag string) error {
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	recs, err := s.relayListRecords(ctx)
	if err != nil {
		return err
	}
	a := relay.Authority{WrittenAt: s.relayTime().UTC()}
	for _, r := range recs {
		if !r.Record.Enabled {
			continue
		}
		a.Apps = append(a.Apps, relay.AuthorityApp{
			Account:     r.Account,
			App:         r.App,
			Incarnation: r.Record.Incarnation,
			Regions:     r.Record.Regions,
		})
	}
	body, err := json.MarshalIndent(a, "", "  ")
	if err != nil {
		return err
	}
	if s.relayGated() {
		return s.relayPublishGated(ctx, etag, append(body, '\n'))
	}
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	_, err = s.cfg.Releases.Put(ctx, relay.AuthorityKey, append(body, '\n'), etag)
	return err
}

// relayPublishAggregate publishes control/relay.json: read the etag FIRST,
// then the whole sequence, restarting it on conflict, four attempts. Then a
// serial bump so relays notice in seconds.
//
// The bump is a LATENCY HINT, not the lease (spec §5): a relay that never
// sees it still re-reads on its own timer, so a failed bump is a log line and
// a successful publish, never a failed operation.
func (s *Server) relayPublishAggregate(ctx context.Context) error {
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	var lastErr error
	for attempt := 0; attempt < 4; attempt++ {
		_, etag, err := s.cfg.Releases.Get(ctx, relay.AuthorityKey)
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = ""
		case err != nil:
			return err
		}
		err = s.relayPublishOnce(ctx, etag)
		if err == nil {
			if err := release.BumpSerial(ctx, s.cfg.Releases, s.relayTime()); err != nil {
				log.Printf("console: relay serial bump (relays converge on their own timer anyway): %v", err)
			}
			return nil
		}
		if !errors.Is(err, release.ErrCASConflict) {
			return err
		}
		lastErr = err
	}
	return fmt.Errorf("publishing %s: %w", relay.AuthorityKey, lastErr)
}

// ---- route delivery ---------------------------------------------------------

// relayRoute is one of an app's routes as the delivery check sees it: enough
// to say which of spec §5's three shapes it is, and nothing else.
type relayRoute struct {
	Host        string
	Kind        string // instance | service | static ("" reads as instance)
	Addr        string // a TCP upstream this platform does not own
	Socket      string
	Unit        string
	Environment string
}

// relayAppRoutes enumerates an app's routes from both places the console can
// see them: the instance records it writes itself, and — for routes an
// operator added directly on a box — the platform registry.
//
// Records first, registry second, deduped on host: a route that exists in
// both is one route, and the record is the authority on its shape.
func (s *Server) relayAppRoutes(ctx context.Context, acct, app string) ([]relayRoute, error) {
	recs, truncated, err := provision.ListRecords(ctx, s.cfg.Releases, acct, app, relayRouteMax)
	if err != nil {
		return nil, err
	}
	// A truncated listing is a REFUSAL, not a partial answer. The one caller
	// that matters is the delivery check, and a check that only saw the first
	// 500 routes would enable the relay over an undeliverable route it never
	// read — the exact state spec §5 refuses because it blocks retirement
	// forever, arrived at silently.
	if truncated {
		return nil, fmt.Errorf("%s/%s has more than %d instance records — the relay cannot check them all, so it will not act on a partial list",
			acct, app, relayRouteMax)
	}
	var out []relayRoute
	seen := map[string]bool{}
	for _, rec := range recs {
		seen[rec.Host] = true
		// Fold Addrs the way the registry branch does (routeAddr, below). A
		// record picks ONE upstream shape — the legacy Addr or the Addrs set,
		// never both (provision.ValidateShape) — so a multi-upstream external
		// route carries Addr == "", and "no address" is exactly what the
		// delivery check reads as deliverable. Left unfolded, such a route
		// slips past the refusal enable exists to make, shows as shape
		// "unknown" in status, and can be acknowledged with `carlos relay
		// adopted`. What every caller here asks is "is there a process at the
		// far end this platform never spawned", and the first upstream
		// answers that as well as five.
		addr := rec.Addr
		if all := rec.AddrsOrDefault(); len(all) > 0 {
			addr = all[0]
		}
		out = append(out, relayRoute{Host: rec.Host, Kind: rec.Kind, Addr: addr,
			Unit: rec.Unit, Environment: rec.Environment})
	}
	if s.cfg.Registry == "" {
		return out, nil
	}
	reg, err := openRegistry(s.cfg.Registry)
	if err != nil {
		// A registry this console cannot open is not a reason to refuse an
		// enable: the records above are the platform's own writing, and the
		// registry copy is the belt to their braces.
		log.Printf("console: relay routes %s/%s: registry: %v", acct, app, err)
		return out, nil
	}
	routes, err := reg.List()
	if err != nil {
		log.Printf("console: relay routes %s/%s: registry list: %v", acct, app, err)
		return out, nil
	}
	for _, rt := range routes {
		if rt.App != app || release.RouteAccount(rt) != acct || seen[rt.Host] {
			continue
		}
		seen[rt.Host] = true
		out = append(out, relayRoute{Host: rt.Host, Kind: rt.Kind, Addr: routeAddr(rt),
			Socket: rt.Socket, Unit: rt.Unit, Environment: rt.Environment})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Host < out[j].Host })
	return out, nil
}

// routeAddr folds a route's two upstream shapes into "does it have one".
func routeAddr(rt registry.Route) string {
	if rt.Addr != "" {
		return rt.Addr
	}
	return rt.Addrs
}

// relayExecDelivery answers whether a route's BOUND bundle opts into file
// delivery, applying the same precedence appconfig's own resolver does
// (#195): the route's named bundle when it states something, the app default
// otherwise. Absence is inheritance, never "off".
func (s *Server) relayExecDelivery(ctx context.Context, acct, app, envName string) (bool, error) {
	if envName != "" && appconfig.ValidEnvironmentName(envName) {
		m, err := appconfig.LoadEnvironment(ctx, s.cfg.Releases, acct, app, envName)
		if err != nil {
			return false, err
		}
		if m.ExecDelivery != "" {
			return m.ExecDelivery == appconfig.ExecDeliveryFile, nil
		}
	}
	m, err := appconfig.Get(ctx, s.cfg.Releases, acct, app)
	if err != nil {
		return false, err
	}
	return m.ExecDelivery == appconfig.ExecDeliveryFile, nil
}

// relayNoDelivery answers the ONE question three callers need about a route:
// can the config sync put a relay key in front of its process at all?
//
// The three shapes: an `instance` route is the platform's own — it has a
// socket or a unit and the config sync restarts it, so it is deliverable. A
// `static` route runs no process at all and is ignored. Everything else is
// deliverable only if it has a socket or a unit to bounce, or its bundle opts
// into file delivery — a bare TCP address whose process this platform never
// spawned is exactly the route the spec refuses.
//
// It lives in one function because the three callers must agree: enable
// REFUSES such a route (below), `carlos relay status` NAMES it (relayReply,
// apirelay.go), and `carlos relay adopted` refuses to record a cutover for
// one (relayAdopted). Three copies of this switch would eventually let a
// member acknowledge a key that had never been written for them.
func (s *Server) relayNoDelivery(ctx context.Context, acct, app string, rt relayRoute) (bool, error) {
	switch {
	case rt.Kind == "static",
		rt.Kind == "" || rt.Kind == "instance",
		rt.Socket != "" || rt.Unit != "",
		rt.Addr == "":
		return false, nil
	}
	on, err := s.relayExecDelivery(ctx, acct, app, rt.Environment)
	return !on, err
}

// relayDeliverable is enable's second refusal (spec §5, step 2): a route that
// could never receive a rotated key must not be enabled, because it would
// block every future retirement forever and nobody would find out until the
// rotation stalled.
func (s *Server) relayDeliverable(ctx context.Context, acct, app string) error {
	routes, err := s.relayAppRoutes(ctx, acct, app)
	if err != nil {
		return err
	}
	for _, rt := range routes {
		none, err := s.relayNoDelivery(ctx, acct, app, rt)
		if err != nil {
			return err
		}
		if none {
			return fmt.Errorf("route %s is a TCP address this platform does not own and its config bundle does not opt into file delivery, "+
				"so it could never pick up a rotated relay key — set it with `carlos env exec-delivery file` first: %w", rt.Host, errRelayUndelivered)
		}
	}
	return nil
}

// ---- the operations ---------------------------------------------------------

// relayEnable turns the relay on for one app and delivers its key (spec §5,
// steps 3-7). The transition table decides what happens to the record; the
// walk after it is always the same three steps, each recorded in `phase` so
// the reconcile can finish it.
//
// A repeat enable of a FINISHED, identically-configured record is the one
// path that writes nothing at all: no op bump, no config write, no new
// envelope. Anything else — an unfinished operation, a region change, a
// re-enable after a disable — walks.
func (s *Server) relayEnable(ctx context.Context, acct *Account, app string, regions []string, by string) (*relay.Record, error) {
	if err := s.relayReady(); err != nil {
		return nil, err
	}
	want, err := s.relayResolveRegions(regions)
	if err != nil {
		return nil, err
	}
	mu := s.relayAppLock(acct.Sqid, app)
	mu.Lock()
	defer mu.Unlock()
	var out *relay.Record
	err = s.relayAppMutation(ctx, acct.Sqid, app, "relay-enable", func(ctx context.Context) error {
		var err error
		out, err = s.relayEnableWalk(ctx, acct, app, want, by)
		return err
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// relayEnableWalk is relayEnable's body under the app's lock and permit.
func (s *Server) relayEnableWalk(ctx context.Context, acct *Account, app string, want []string, by string) (*relay.Record, error) {
	if err := s.requireRelayLease(); err != nil {
		return nil, err
	}
	if err := s.relayDeliverable(ctx, acct.Sqid, app); err != nil {
		return nil, err
	}
	rec, etag, err := relay.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app)
	if err != nil {
		return nil, err
	}
	now := s.relayTime().UTC()
	switch {
	case rec != nil && rec.Enabled && rec.Phase == relayPhaseDone && slices.Equal(rec.Regions, want):
		return rec, nil
	case rec != nil && rec.Enabled && slices.Equal(rec.Regions, want):
		// An unfinished operation RESUMES rather than starting a new one:
		// same incarnation, same op, and the walk below repeats the steps it
		// may already have done, all of which are idempotent.
	case rec != nil && rec.Enabled:
		// A region change keeps the incarnation — the app's identity did not
		// move, only where it may connect (spec §5's transition table).
		rec.Regions = want
		rec.Op++
		rec.Phase = relayPhaseConfig
	default:
		op := int64(0)
		if rec != nil {
			op = rec.Op
		}
		// A NEW incarnation for a first enable and for every re-enable after
		// a disable, so a key from before the disable can never be mistaken
		// for a live one. Cutovers go with it: they acknowledged identities
		// that no longer exist.
		rec = &relay.Record{
			Enabled: true, Incarnation: relay.NewIncarnation(), Regions: want,
			Op: op + 1, Phase: relayPhaseConfig, EnabledAt: now,
		}
	}
	if etag, err = s.relayPutRecord(ctx, acct.Sqid, app, rec, etag); err != nil {
		return nil, err
	}
	if err := s.relayApplyConfig(ctx, acct.Sqid, app, rec); err != nil {
		return nil, err
	}
	rec.Phase = relayPhaseAggregate
	if etag, err = s.relayPutRecord(ctx, acct.Sqid, app, rec, etag); err != nil {
		return nil, err
	}
	if err := s.relayPublishAggregate(ctx); err != nil {
		return nil, err
	}
	rec.Phase = relayPhaseDone
	if _, err = s.relayPutRecord(ctx, acct.Sqid, app, rec, etag); err != nil {
		return nil, err
	}
	log.Printf("console: relay enabled %s/%s in %s by %s (incarnation %s, op %d)",
		acct.Sqid, app, strings.Join(rec.Regions, ","), by, rec.Incarnation, rec.Op)
	return rec, nil
}

// relayDisable is the same walk in reverse (spec §5): record the intent,
// remove the variable, publish an aggregate without the app, mark done. The
// incarnation stays in the record — it is what `status` reports the pending
// disable against — and a later enable draws a new one.
func (s *Server) relayDisable(ctx context.Context, acct *Account, app string, by string) (*relay.Record, error) {
	if s.cfg.Releases == nil {
		return nil, errRelayNoBucket
	}
	mu := s.relayAppLock(acct.Sqid, app)
	mu.Lock()
	defer mu.Unlock()
	var out *relay.Record
	err := s.relayAppMutation(ctx, acct.Sqid, app, "relay-disable", func(ctx context.Context) error {
		var err error
		out, err = s.relayDisableWalk(ctx, acct, app, by)
		return err
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// relayDisableWalk is relayDisable's body under the app's lock and permit.
func (s *Server) relayDisableWalk(ctx context.Context, acct *Account, app string, by string) (*relay.Record, error) {
	if err := s.requireRelayLease(); err != nil {
		return nil, err
	}
	rec, etag, err := relay.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app)
	if err != nil {
		return nil, err
	}
	if rec == nil {
		return nil, errRelayNotEnabled
	}
	if !rec.Enabled && rec.Phase == relayPhaseDone {
		return rec, nil // already off and finished: nothing owed
	}
	if rec.Enabled {
		rec.Enabled = false
		rec.Op++
		rec.Phase = relayPhaseConfig
		rec.DisabledAt = s.relayTime().UTC()
		// Nothing is issued any more. Cutovers stay: they are a historical
		// record of what a member acknowledged, and a re-enable clears them.
		rec.IssuedSuffixes = nil
		rec.IssuedDigest = ""
		if etag, err = s.relayPutRecord(ctx, acct.Sqid, app, rec, etag); err != nil {
			return nil, err
		}
	}
	if _, err := s.relayUnsetConfig(ctx, acct.Sqid, app); err != nil {
		return nil, err
	}
	rec.Phase = relayPhaseAggregate
	if etag, err = s.relayPutRecord(ctx, acct.Sqid, app, rec, etag); err != nil {
		return nil, err
	}
	if err := s.relayPublishAggregate(ctx); err != nil {
		return nil, err
	}
	rec.Phase = relayPhaseDone
	if _, err = s.relayPutRecord(ctx, acct.Sqid, app, rec, etag); err != nil {
		return nil, err
	}
	log.Printf("console: relay disabled %s/%s by %s (op %d)", acct.Sqid, app, by, rec.Op)
	return rec, nil
}

// relayAdopted records a member's acknowledgement that a TCP-address route
// they restarted by hand now runs the identities the console last issued
// (spec §5, step 5).
//
// The identities are stamped from the record AT THIS MOMENT, not named by the
// caller, and they are what makes the ack meaningful: an acknowledgement
// given before a later rotation names the older suffixes and so does not
// count for the newer ones. One cutover per host — a second ack for the same
// route replaces the first.
func (s *Server) relayAdopted(ctx context.Context, acct *Account, app, host, by string) (*relay.Record, error) {
	if s.cfg.Releases == nil {
		return nil, errRelayNoBucket
	}
	mu := s.relayAppLock(acct.Sqid, app)
	mu.Lock()
	defer mu.Unlock()
	var out *relay.Record
	err := s.relayAppMutation(ctx, acct.Sqid, app, "relay-adopted", func(ctx context.Context) error {
		var err error
		out, err = s.relayAdoptedWalk(ctx, acct, app, host, by)
		return err
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// relayAdoptedWalk is relayAdopted's body under the app's lock and permit.
func (s *Server) relayAdoptedWalk(ctx context.Context, acct *Account, app, host, by string) (*relay.Record, error) {
	if err := s.requireRelayLease(); err != nil {
		return nil, err
	}
	rec, etag, err := relay.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app)
	if err != nil {
		return nil, err
	}
	if rec == nil || !rec.Enabled {
		return nil, errRelayNotEnabled
	}
	// The host must be one of the app's own routes. Cheap, and it is what
	// keeps the record bounded: without it, an ack naming anything at all
	// appends a cutover forever, and `status` would report identities against
	// routes that do not exist.
	routes, err := s.relayAppRoutes(ctx, acct.Sqid, app)
	if err != nil {
		return nil, err
	}
	i := slices.IndexFunc(routes, func(rt relayRoute) bool { return rt.Host == host })
	if i < 0 {
		return nil, fmt.Errorf("%s is not a route of this app: %w", host, errRelayUnknownHost)
	}
	// A route the config sync cannot reach has been sent NOTHING to adopt, so
	// an acknowledgement of it would be false — and worse than merely false:
	// a cutover is what the retirement of an old master is eventually checked
	// against, and one recorded here would say a process has a key it has
	// never seen. Such a route can only appear after an enable (enable itself
	// refuses it), which is exactly when a member might reach for this verb
	// to make the warning go away.
	if none, err := s.relayNoDelivery(ctx, acct.Sqid, app, routes[i]); err != nil {
		return nil, err
	} else if none {
		return nil, fmt.Errorf("%s has no config delivery, so no relay key has been written for it to adopt — "+
			"opt it into file delivery with `carlos env exec-delivery file` or move it to a unit: %w", host, errRelayUndelivered)
	}
	// And the route must be the ONE shape whose cutover a member establishes.
	// Spec §5, step 5: a unit or socket-exec route is restarted by the config
	// sync, so its cutover is the ack that sync writes — an observed fact,
	// with the identities the host actually rendered. Accepting a member's
	// word for such a route would stamp the identities the console ISSUED
	// over that observation, and a route whose restart silently failed would
	// read as cut over. The member verb exists only for the process this
	// platform cannot restart.
	switch shape := relayRouteShape(routes[i]); shape {
	case relayShapeFile: // the one shape a member acknowledges
	case relayShapeUnit, relayShapeSocket:
		return nil, fmt.Errorf("%s is a %s route: the platform restarts this route itself; its adoption is observed, not declared — "+
			"`carlos relay status` reports it as adopted once the config sync acknowledges the write: %w",
			host, shape, errRelayNotMemberRoute)
	default:
		return nil, fmt.Errorf("%s has no unit, no socket and no address, so this console cannot say what would run a relay key, "+
			"let alone that it has: %w", host, errRelayNotMemberRoute)
	}
	cut := relay.RouteCutover{
		Host:     host,
		Suffixes: maps.Clone(rec.IssuedSuffixes),
		At:       s.relayTime().UTC(),
		By:       by,
	}
	replaced := false
	for i := range rec.Cutovers {
		if rec.Cutovers[i].Host == host {
			rec.Cutovers[i] = cut
			replaced = true
			break
		}
	}
	if !replaced {
		rec.Cutovers = append(rec.Cutovers, cut)
	}
	if _, err := s.relayPutRecord(ctx, acct.Sqid, app, rec, etag); err != nil {
		return nil, err
	}
	return rec, nil
}

// ---- rotation ---------------------------------------------------------------

// relayRotateResult is what `carlos relay rotate` prints: which kid the
// region now issues from, how many apps were re-issued, and the earliest the
// old master may be retired.
type relayRotateResult struct {
	Region              string
	Kid                 string
	Apps                int
	RetireNoEarlierThan time.Time
}

// relayHeartbeats reads every armed relay's heartbeat.
func (s *Server) relayHeartbeats(ctx context.Context) ([]relay.Heartbeat, error) {
	keys, err := s.cfg.Releases.List(ctx, relayAckPrefix)
	if err != nil {
		return nil, err
	}
	var out []relay.Heartbeat
	for _, key := range keys {
		if !strings.HasSuffix(key, ".json") {
			continue
		}
		body, _, err := s.cfg.Releases.Get(ctx, key)
		if err != nil {
			if errors.Is(err, os.ErrNotExist) {
				continue
			}
			return nil, err
		}
		var hb relay.Heartbeat
		if err := json.Unmarshal(body, &hb); err != nil {
			log.Printf("console: relay heartbeat %s: %v", key, err)
			continue
		}
		out = append(out, hb)
	}
	return out, nil
}

// relayConfigAcks reads every per-route config-adoption acknowledgement
// under configAckPrefix whose HOST — a key's basename, ConfigAckKey's own
// "<box>/<host>.json" shape — is in hosts. There is no per-host box list
// to narrow the LISTING by, the same limit relayHeartbeats accepts for
// control/relay-ack/, so this still lists every key under the prefix; but
// hosts filters that listing down to this app's own routes before a
// single Get is issued, so a status call costs one List plus a Get per
// THIS APP's route, never a Get per ack in the whole fleet (review
// finding 2026-09-07 — relayReply used to pass every ack for every app on
// the deployment through this filter instead of narrowing the read
// itself).
func (s *Server) relayConfigAcks(ctx context.Context, hosts map[string]bool) ([]appconfig.ConfigAck, error) {
	keys, err := s.cfg.Releases.List(ctx, configAckPrefix)
	if err != nil {
		return nil, err
	}
	var out []appconfig.ConfigAck
	for _, key := range keys {
		if !strings.HasSuffix(key, ".json") {
			continue
		}
		host := strings.TrimSuffix(key, ".json")
		if idx := strings.LastIndexByte(host, '/'); idx >= 0 {
			host = host[idx+1:]
		}
		if !hosts[host] {
			continue
		}
		body, _, err := s.cfg.Releases.Get(ctx, key)
		if err != nil {
			if errors.Is(err, os.ErrNotExist) {
				continue
			}
			return nil, err
		}
		var a appconfig.ConfigAck
		if err := json.Unmarshal(body, &a); err != nil {
			log.Printf("console: relay config ack %s: %v", key, err)
			continue
		}
		out = append(out, a)
	}
	return out, nil
}

// relayRotate moves a region's issuer pointer to the master the operator
// staged, then re-issues every app enabled in that region (spec §5, step 4).
//
// IT REFUSES UNTIL EVERY LIVE RELAY IN THE REGION HAS THE NEW KID. That gate
// is the whole safety of a rotation: issuing from a master a box has not
// loaded mints credentials that box refuses, and every call in the region
// fails until it catches up. A heartbeat older than five minutes is not an
// answer — a dead relay must not be able to consent by silence — so a region
// with no fresh heartbeat at all refuses too.
//
// The new kid is the LAST staged entry that is not the current issuer, which
// is where an operator following the runbook appends it.
func (s *Server) relayRotate(ctx context.Context, region string) (relayRotateResult, error) {
	var res relayRotateResult
	if err := s.relayReady(); err != nil {
		return res, err
	}
	region = strings.ToLower(strings.TrimSpace(region))
	if !slices.Contains(s.relayRegions(), region) {
		return res, fmt.Errorf("this deployment runs no relay in %s — armed regions are %s",
			region, strings.Join(s.relayRegions(), ", "))
	}
	if err := s.requireRelayLease(); err != nil {
		return res, err
	}
	staged, err := relayStagedMasters(region)
	if err != nil {
		return res, err
	}
	current, issuerEtag, err := s.relayIssuers(ctx)
	if err != nil {
		return res, err
	}
	curKid := current[region]
	if curKid == "" {
		curKid = staged[0].Kid
	}
	// The new kid is the LAST staged entry, full stop — never "the last one
	// that is not the issuer". After a rotation the variable holds
	// "<old>,<new>" with the pointer on <new> for the 28 hours before the old
	// one may be retired, and a second rotate in that window must refuse, not
	// walk the pointer BACK onto the master about to be retired and re-issue
	// every app under it.
	next := staged[len(staged)-1].Kid
	if next == curKid {
		return res, fmt.Errorf("no new master is staged for %s: the last entry in %s is already the issuer",
			region, relayMastersEnv(region))
	}
	hbs, err := s.relayHeartbeats(ctx)
	if err != nil {
		return res, err
	}
	now := s.relayTime()
	live := 0
	for _, hb := range hbs {
		if hb.Region != region || now.Sub(hb.ReadAt) > relayHeartbeatFresh {
			continue
		}
		live++
		if !slices.Contains(hb.Kids, next) {
			return res, fmt.Errorf("relay %s in %s has not loaded kid %s yet (it has %s) — add the line to /etc/carlos/relay-masters there and try again",
				hb.Box, region, next, strings.Join(hb.Kids, ","))
		}
	}
	if live == 0 {
		return res, fmt.Errorf("no relay in %s has heartbeated in the last %s — rotating now would issue keys nothing can verify",
			region, relayHeartbeatFresh)
	}
	// The pointer moves FIRST, under CAS, and every key derived after this
	// line uses the new master. An interrupted rotation therefore leaves a
	// moved pointer and some apps still on the old kid, which is exactly what
	// the reconcile converges — and why the old kid stays accepted for 28
	// hours.
	current[region] = next
	body, err := json.MarshalIndent(current, "", "  ")
	if err != nil {
		return res, err
	}
	if err := s.requireRelayLease(); err != nil {
		return res, err
	}
	if _, err := s.cfg.Releases.Put(ctx, relayIssuerKey, append(body, '\n'), issuerEtag); err != nil {
		return res, err
	}
	recs, err := s.relayListRecords(ctx)
	if err != nil {
		return res, err
	}
	reissued := 0
	for _, r := range recs {
		if !r.Record.Enabled || !slices.Contains(r.Record.Regions, region) {
			continue
		}
		if err := s.relayReissue(ctx, r); err != nil {
			// One app that cannot be re-issued must not stop the rest: the
			// pointer has already moved, so every remaining app is one the
			// reconcile would otherwise have to catch five minutes later.
			log.Printf("console: relay rotate %s: %s/%s: %v", region, r.Account, r.App, err)
			continue
		}
		reissued++
	}
	log.Printf("console: relay rotated %s to kid %s, re-issued %d app(s)", region, next, reissued)
	return relayRotateResult{Region: region, Kid: next, Apps: reissued,
		RetireNoEarlierThan: now.Add(relayRetireAfter).UTC()}, nil
}

// relayReissue is a rotation's per-app operation: op bumped, config
// re-applied from the master in force now, aggregate published, done. The
// aggregate does not CHANGE on a rotation (the incarnation is untouched), but
// it is published anyway because the phase walk is what makes the operation
// resumable, and a publish is cheap next to a half-recorded rotation.
func (s *Server) relayReissue(ctx context.Context, r relayAppRecord) error {
	mu := s.relayAppLock(r.Account, r.App)
	mu.Lock()
	defer mu.Unlock()
	return s.relayAppMutation(ctx, r.Account, r.App, "relay-reissue", func(ctx context.Context) error {
		return s.relayReissueWalk(ctx, r)
	})
}

// relayReissueWalk is relayReissue's body under the app's lock and permit.
func (s *Server) relayReissueWalk(ctx context.Context, r relayAppRecord) error {
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	rec, etag, err := relay.GetRecord(ctx, s.cfg.Releases, r.Account, r.App)
	if err != nil {
		return err
	}
	if rec == nil || !rec.Enabled {
		return nil // disabled under us: the reconcile owns it now
	}
	rec.Op++
	rec.Phase = relayPhaseConfig
	if etag, err = s.relayPutRecord(ctx, r.Account, r.App, rec, etag); err != nil {
		return err
	}
	if err := s.relayApplyConfig(ctx, r.Account, r.App, rec); err != nil {
		return err
	}
	rec.Phase = relayPhaseAggregate
	if etag, err = s.relayPutRecord(ctx, r.Account, r.App, rec, etag); err != nil {
		return err
	}
	if err := s.relayPublishAggregate(ctx); err != nil {
		return err
	}
	rec.Phase = relayPhaseDone
	_, err = s.relayPutRecord(ctx, r.Account, r.App, rec, etag)
	return err
}

// ---- the reconcile ----------------------------------------------------------

// relayReconcile converges BOTH derived artefacts from every record, every
// five minutes and once at console start (spec §5).
//
// It deliberately does not look for unfinished records. For each enabled
// record it derives what the config should hold from the record and the
// masters in force, and re-applies when the config holds anything else; for
// each disabled record it removes the variable if present; and it ALWAYS
// rebuilds and publishes the aggregate, which is what repairs a deletion that
// removed a record but died before publishing.
//
// One app's failure is logged and skipped, never fatal. The aggregate publish
// is the last step and the only one whose failure is returned, because it is
// the one that speaks for every app at once.
func (s *Server) relayReconcile(ctx context.Context) error {
	if s.cfg.Releases == nil {
		return errRelayNoBucket
	}
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	recs, err := s.relayListRecords(ctx)
	if err != nil {
		return err
	}
	// Records whose walk reached `aggregate` in this pass, with the etag to
	// mark them done once the publish below has landed.
	type pending struct {
		rec  relayAppRecord
		etag string
	}
	var advanced []pending
	for _, r := range recs {
		etag, changed, err := s.relayConverge(ctx, r)
		if err != nil {
			log.Printf("console: relay reconcile %s/%s: %v", r.Account, r.App, err)
			continue
		}
		if changed {
			advanced = append(advanced, pending{rec: r, etag: etag})
		}
	}
	if err := s.relayPublishAggregate(ctx); err != nil {
		return err
	}
	for _, p := range advanced {
		if err := s.relayMarkDone(ctx, p.rec, p.etag); err != nil {
			log.Printf("console: relay reconcile %s/%s: marking done: %v", p.rec.Account, p.rec.App, err)
		}
	}
	return nil
}

// relayConverge brings ONE record's config into agreement with it, returning
// the etag it left the record at and whether the record still needs marking
// done after the publish.
func (s *Server) relayConverge(ctx context.Context, r relayAppRecord) (string, bool, error) {
	mu := s.relayAppLock(r.Account, r.App)
	mu.Lock()
	defer mu.Unlock()
	var (
		etag    string
		changed bool
	)
	err := s.relayAppMutation(ctx, r.Account, r.App, "relay-converge", func(ctx context.Context) error {
		var err error
		etag, changed, err = s.relayConvergeWalk(ctx, r)
		return err
	})
	if err != nil {
		return "", false, err
	}
	return etag, changed, nil
}

// relayConvergeWalk is relayConverge's body under the app's lock and permit.
func (s *Server) relayConvergeWalk(ctx context.Context, r relayAppRecord) (string, bool, error) {
	if err := s.requireRelayLease(); err != nil {
		return "", false, err
	}
	// Re-read under the lock: the listing is a snapshot, and an operation may
	// have moved this record between it and here.
	rec, etag, err := relay.GetRecord(ctx, s.cfg.Releases, r.Account, r.App)
	if err != nil || rec == nil {
		return "", false, err
	}
	if rec.Enabled {
		want, err := s.relayExpectedSuffixes(ctx, r.Account, r.App, rec)
		if err != nil {
			return "", false, err
		}
		v, present, err := s.relayConfigVar(ctx, r.Account, r.App)
		if err != nil {
			return "", false, err
		}
		// Current means BOTH halves agree: the identities the record says it
		// issued are the ones the masters in force imply (a rotation,
		// re-enable or region change moves those), AND the bundle still holds
		// the exact envelope this console last wrote (a straggler write from
		// a superseded console, or a member setting CARLOS_RELAY themselves,
		// moves that). A record from before IssuedDigest existed has none, so
		// it re-applies once and then carries one.
		current := present && v.Kind == "secret" && v.Envelope != "" &&
			maps.Equal(rec.IssuedSuffixes, want) &&
			rec.IssuedDigest != "" && relayEnvelopeDigest(v.Envelope) == rec.IssuedDigest
		if current && rec.Phase == relayPhaseDone {
			return etag, false, nil
		}
		if !current {
			if err := s.relayApplyConfig(ctx, r.Account, r.App, rec); err != nil {
				return "", false, err
			}
		}
	} else {
		if _, err := s.relayUnsetConfig(ctx, r.Account, r.App); err != nil {
			return "", false, err
		}
		if rec.Phase == relayPhaseDone {
			return etag, false, nil
		}
	}
	rec.Phase = relayPhaseAggregate
	etag, err = s.relayPutRecord(ctx, r.Account, r.App, rec, etag)
	if err != nil {
		return "", false, err
	}
	return etag, true, nil
}

// relayMarkDone is the last step of a reconciled record's walk, after the
// aggregate that carries it has been published.
func (s *Server) relayMarkDone(ctx context.Context, r relayAppRecord, etag string) error {
	mu := s.relayAppLock(r.Account, r.App)
	mu.Lock()
	defer mu.Unlock()
	return s.relayAppMutation(ctx, r.Account, r.App, "relay-mark-done", func(ctx context.Context) error {
		return s.relayMarkDoneWalk(ctx, r, etag)
	})
}

// relayMarkDoneWalk is relayMarkDone's body under the app's lock and permit.
func (s *Server) relayMarkDoneWalk(ctx context.Context, r relayAppRecord, etag string) error {
	if err := s.requireRelayLease(); err != nil {
		return err
	}
	rec, cur, err := relay.GetRecord(ctx, s.cfg.Releases, r.Account, r.App)
	if err != nil || rec == nil {
		return err
	}
	if cur != etag {
		return nil // moved on under us; its own walk owns the phase now
	}
	rec.Phase = relayPhaseDone
	_, err = s.relayPutRecord(ctx, r.Account, r.App, rec, etag)
	return err
}