package console

// The relay's write side (relay spec §5). What is pinned here, in the order
// it would hurt if lost: an enable delivers a key AND publishes authority, or
// it does neither; the aggregate is rebuilt from records with the etag read
// FIRST, so a disable landing mid-listing fails the write instead of undoing
// itself; the reconcile converges an interrupted enable, a stale config write
// that outlived a disable, and an aggregate naming an app whose record is
// gone; and one console at a time holds the writer lease, with the loser
// stopping rather than writing.

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"testing"
	"time"

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

// relayFixture is storeFixture (an account, the app titogo, a member's token)
// plus everything the relay needs: one armed region, a sealing key whose
// private half only the test holds, and a staged master in this console's own
// environment. It leaves the server HOLDING the writer lease, because every
// operation refuses without it and no test here is about that refusal.
//
// The private key comes back so a test can open the envelope the console
// delivered and prove it is the value the record implies — the console cannot
// do that, by design, so the assertion belongs on this side.
func relayFixture(t *testing.T, code string) (*consoleFixture, string) {
	t.Helper()
	fx, _ := storeFixture(t, code)
	return fx, armRelay(t, fx)
}

// armRelay gives any console fixture a relay: one armed region, a sealing key
// whose private half comes back to the caller, a staged master, and the
// writer lease held.
func armRelay(t *testing.T, fx *consoleFixture) string {
	t.Helper()
	pub, priv, _ := sealedFor(t, "irrelevant")
	fx.srv.cfg.SecretsPub = pub
	fx.srv.cfg.RelayRegions = []string{"eu-west-1"}
	fx.srv.cfg.PlatformDomain = "oncarlos.com"
	t.Setenv(relayMastersEnv("eu-west-1"), "k1:"+relayTestMaster(1))
	if err := fx.srv.relayAcquire(context.Background()); err != nil {
		t.Fatalf("acquire the writer lease: %v", err)
	}
	return priv
}

// relayTestMaster is a deterministic 32-byte master, base64 as the console
// variable carries it.
func relayTestMaster(seed byte) string {
	b := make([]byte, relay.MasterKeyLen)
	for i := range b {
		b[i] = seed*31 + byte(i)
	}
	return base64.StdEncoding.EncodeToString(b)
}

// account is the fixture's account as the operations take it.
func (fx *consoleFixture) account(t *testing.T) *Account {
	t.Helper()
	a, err := fx.st.AccountBySqid(fx.sqid)
	if err != nil {
		t.Fatalf("account %s: %v", fx.sqid, err)
	}
	return a
}

// relayRecord reads the app's record straight out of the bucket.
func relayRecord(t *testing.T, fx *consoleFixture, app string) *relay.Record {
	t.Helper()
	rec, _, err := relay.GetRecord(context.Background(), fx.rel, fx.sqid, app)
	if err != nil {
		t.Fatalf("read relay record: %v", err)
	}
	return rec
}

// relayAuthority reads the published authority record, an empty one when
// nothing has been published.
func relayAuthority(t *testing.T, fx *consoleFixture) *relay.Authority {
	t.Helper()
	a, _, err := relay.ReadAuthority(context.Background(), fx.rel)
	if err != nil {
		t.Fatalf("read authority: %v", err)
	}
	if a == nil {
		return &relay.Authority{}
	}
	return a
}

// TestRelayEnableWritesRecordConfigAndAggregate is the whole happy path: the
// record lands finished, the app's config carries a SEALED CARLOS_RELAY whose
// plaintext is the value the record implies, and the authority every relay
// box reads names the app at that incarnation.
func TestRelayEnableWritesRecordConfigAndAggregate(t *testing.T) {
	fx, priv := relayFixture(t, "code-relay-enable")
	ctx := context.Background()

	rec, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", nil, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if !rec.Enabled || rec.Phase != relayPhaseDone || rec.Op != 1 || len(rec.Incarnation) != 16 {
		t.Fatalf("record = %+v", rec)
	}
	if got := relayRecord(t, fx, "titogo"); got == nil || got.Phase != relayPhaseDone || got.Incarnation != rec.Incarnation {
		t.Fatalf("stored record = %+v, want the returned one finished", got)
	}

	v, ok := appVars(t, fx, "titogo")[relayVar]
	if !ok || v.Kind != "secret" || v.Envelope == "" {
		t.Fatalf("%s = %+v, want a sealed secret", relayVar, v)
	}
	entries, err := relay.ParseConfig(openSecret(t, priv, v))
	if err != nil {
		t.Fatalf("delivered value: %v", err)
	}
	if len(entries) != 1 || entries[0].Region != "eu-west-1" {
		t.Fatalf("entries = %+v, want one for the armed region", entries)
	}
	wantURLs := []string{
		"stun:relay.eu-west-1.oncarlos.com:3478",
		"turn:relay.eu-west-1.oncarlos.com:3478?transport=udp",
		"turn:relay.eu-west-1.oncarlos.com:3478?transport=tcp",
	}
	if strings.Join(entries[0].URLs, ",") != strings.Join(wantURLs, ",") {
		t.Fatalf("urls = %v, want %v", entries[0].URLs, wantURLs)
	}
	wantSuffix := relay.Suffix(fx.sqid, "titogo", rec.Incarnation, "k1")
	if entries[0].UsernameSuffix != wantSuffix || rec.IssuedSuffixes["eu-west-1"] != wantSuffix {
		t.Fatalf("suffix = %q / recorded %q, want %q", entries[0].UsernameSuffix, rec.IssuedSuffixes["eu-west-1"], wantSuffix)
	}

	a := relayAuthority(t, fx)
	x, ok := a.Lookup(fx.sqid, "titogo")
	if !ok || x.Incarnation != rec.Incarnation || strings.Join(x.Regions, ",") != "eu-west-1" {
		t.Fatalf("aggregate = %+v", a)
	}
}

// TestRelayRepeatEnableKeepsIncarnationAndWritesNoConfig: enabling an app
// that is already enabled with the same regions and nothing owed writes
// NOTHING — no op bump, no new envelope, no generation touch. A member who
// re-runs the command must not re-key their own app.
func TestRelayRepeatEnableKeepsIncarnationAndWritesNoConfig(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-repeat")
	ctx := context.Background()

	first, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", nil, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	before := appVars(t, fx, "titogo")[relayVar]
	_, genBefore, err := fx.rel.Get(ctx, appconfig.GenKey(fx.sqid, "titogo"))
	if err != nil {
		t.Fatal(err)
	}

	second, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", []string{"eu-west-1"}, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if second.Incarnation != first.Incarnation || second.Op != first.Op {
		t.Fatalf("second enable = %+v, want the same incarnation and op as %+v", second, first)
	}
	if after := appVars(t, fx, "titogo")[relayVar]; after.Envelope != before.Envelope {
		t.Fatal("a repeat enable re-sealed the value — the app would re-key for nothing")
	}
	_, genAfter, err := fx.rel.Get(ctx, appconfig.GenKey(fx.sqid, "titogo"))
	if err != nil {
		t.Fatal(err)
	}
	if genAfter != genBefore {
		t.Fatalf("config generation moved (%q -> %q) with no config write", genBefore, genAfter)
	}
}

// TestRelayDisableThenEnableIsANewIncarnation: a disable takes the app out of
// authority and takes the variable out of its config; the enable after it
// draws a NEW incarnation, so a credential minted before the disable can
// never be mistaken for a live one.
func TestRelayDisableThenEnableIsANewIncarnation(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-cycle")
	ctx := context.Background()
	acct := fx.account(t)

	first, err := fx.srv.relayEnable(ctx, acct, "titogo", nil, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	off, err := fx.srv.relayDisable(ctx, acct, "titogo", "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if off.Enabled || off.Phase != relayPhaseDone || off.Op != first.Op+1 {
		t.Fatalf("disabled record = %+v", off)
	}
	if _, ok := appVars(t, fx, "titogo")[relayVar]; ok {
		t.Fatalf("%s survived the disable", relayVar)
	}
	if _, ok := relayAuthority(t, fx).Lookup(fx.sqid, "titogo"); ok {
		t.Fatal("the aggregate still names a disabled app")
	}

	again, err := fx.srv.relayEnable(ctx, acct, "titogo", nil, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if again.Incarnation == first.Incarnation {
		t.Fatalf("re-enable kept incarnation %s — a pre-disable key would still verify", again.Incarnation)
	}
	if again.Op != off.Op+1 {
		t.Fatalf("op = %d after a re-enable, want %d (the counter never rewinds)", again.Op, off.Op+1)
	}
	if x, ok := relayAuthority(t, fx).Lookup(fx.sqid, "titogo"); !ok || x.Incarnation != again.Incarnation {
		t.Fatalf("aggregate = %+v, want the new incarnation", x)
	}
}

// relayOpStore records the ORDER of the store calls one publish makes — the
// only way to prove the etag is read before the listing rather than after,
// which is the whole safety of spec §5's publish sequence.
type relayOpStore struct {
	release.Store
	mu  sync.Mutex
	ops []string
}

func (r *relayOpStore) note(op string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.ops = append(r.ops, op)
}

func (r *relayOpStore) seq() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]string(nil), r.ops...)
}

func (r *relayOpStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	r.note("get " + key)
	return r.Store.Get(ctx, key)
}

func (r *relayOpStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	r.note("listdirs " + prefix)
	return r.Store.ListDirs(ctx, prefix)
}

func (r *relayOpStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	r.note("put " + key)
	return r.Store.Put(ctx, key, body, ifMatch)
}

// TestRelayPublishAggregateIsEtagFirst pins the sequence itself: the
// aggregate's etag is read BEFORE any record is listed, and a publish carrying
// an etag that has since moved fails with ErrCASConflict rather than
// overwriting whoever moved it. The retry then publishes the current truth.
func TestRelayPublishAggregateIsEtagFirst(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-etag")
	ctx := context.Background()
	acct := fx.account(t)

	if _, err := fx.srv.relayEnable(ctx, acct, "titogo", nil, "alice@keymail.dev"); err != nil {
		t.Fatal(err)
	}
	if _, err := fx.srv.relayEnable(ctx, acct, "meet", nil, "alice@keymail.dev"); err != nil {
		t.Fatal(err)
	}

	// The order, on a publish with nothing else going on.
	rec := &relayOpStore{Store: fx.rel}
	fx.srv.cfg.Releases = rec
	if err := fx.srv.relayPublishAggregate(ctx); err != nil {
		t.Fatal(err)
	}
	fx.srv.cfg.Releases = rec.Store
	seq := rec.seq()
	getAt, listAt := -1, -1
	for i, op := range seq {
		if getAt < 0 && op == "get "+relay.AuthorityKey {
			getAt = i
		}
		if listAt < 0 && strings.HasPrefix(op, "listdirs apps/") {
			listAt = i
		}
	}
	if getAt < 0 || listAt < 0 || getAt > listAt {
		t.Fatalf("publish sequence = %v, want the aggregate's etag read before the listing", seq)
	}

	// A stale etag must lose. Take one, then move the aggregate underneath it
	// by disabling titogo's record directly and republishing.
	_, stale, err := fx.rel.Get(ctx, relay.AuthorityKey)
	if err != nil {
		t.Fatal(err)
	}
	off, offEtag, err := relay.GetRecord(ctx, fx.rel, fx.sqid, "titogo")
	if err != nil {
		t.Fatal(err)
	}
	off.Enabled = false
	if _, err := relay.PutRecord(ctx, fx.rel, fx.sqid, "titogo", off, offEtag); err != nil {
		t.Fatal(err)
	}
	if err := fx.srv.relayPublishAggregate(ctx); err != nil {
		t.Fatal(err)
	}
	if err := fx.srv.relayPublishOnce(ctx, stale); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("publish with a stale etag = %v, want ErrCASConflict", err)
	}
	if err := fx.srv.relayPublishAggregate(ctx); err != nil {
		t.Fatal(err)
	}
	a := relayAuthority(t, fx)
	if _, ok := a.Lookup(fx.sqid, "titogo"); ok {
		t.Fatalf("aggregate = %+v, want the disabled app gone", a)
	}
	if _, ok := a.Lookup(fx.sqid, "meet"); !ok {
		t.Fatalf("aggregate = %+v, want the still-enabled app kept", a)
	}
}

// TestRelayReconcileFinishesInterruptedEnable: a console that wrote the
// record and died before the config write leaves an enabled record at phase
// `config`, no variable and no authority. The next holder's reconcile
// delivers both and marks it done — without the member running anything.
func TestRelayReconcileFinishesInterruptedEnable(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-resume")
	ctx := context.Background()
	inc := relay.NewIncarnation()
	if _, err := relay.PutRecord(ctx, fx.rel, fx.sqid, "titogo", &relay.Record{
		Enabled: true, Incarnation: inc, Regions: []string{"eu-west-1"},
		Op: 1, Phase: relayPhaseConfig, EnabledAt: time.Now().UTC(),
	}, ""); err != nil {
		t.Fatal(err)
	}

	if err := fx.srv.relayReconcile(ctx); err != nil {
		t.Fatal(err)
	}

	if v, ok := appVars(t, fx, "titogo")[relayVar]; !ok || v.Kind != "secret" {
		t.Fatalf("%s = %+v after the reconcile, want a delivered secret", relayVar, v)
	}
	if x, ok := relayAuthority(t, fx).Lookup(fx.sqid, "titogo"); !ok || x.Incarnation != inc {
		t.Fatal("the reconcile did not publish the interrupted enable")
	}
	if got := relayRecord(t, fx, "titogo"); got.Phase != relayPhaseDone {
		t.Fatalf("phase = %q after the reconcile, want done", got.Phase)
	}
}

// TestRelayReconcileUndoesStaleConfigAfterDisable: a config write that lands
// AFTER a disable — a losing writer's last act — is undone within one
// reconcile. This is the convergence spec §5 promises for the case the phase
// walk alone cannot cover, because the record is already finished.
func TestRelayReconcileUndoesStaleConfigAfterDisable(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-stale")
	ctx := context.Background()
	acct := fx.account(t)

	if _, err := fx.srv.relayEnable(ctx, acct, "titogo", nil, "alice@keymail.dev"); err != nil {
		t.Fatal(err)
	}
	if _, err := fx.srv.relayDisable(ctx, acct, "titogo", "alice@keymail.dev"); err != nil {
		t.Fatal(err)
	}
	// The straggler write.
	if _, err := appconfig.Apply(ctx, fx.rel, fx.sqid, "titogo",
		map[string]appconfig.Var{relayVar: {Kind: "secret", Envelope: "stale-envelope"}}, nil); err != nil {
		t.Fatal(err)
	}

	if err := fx.srv.relayReconcile(ctx); err != nil {
		t.Fatal(err)
	}
	if v, ok := appVars(t, fx, "titogo")[relayVar]; ok {
		t.Fatalf("%s = %+v, want it removed again", relayVar, v)
	}
}

// TestRelayReconcileRepairsDeletionWithoutRecord: an app whose record is gone
// — deleted with the rest of its prefix by a walk that died before publishing
// — disappears from authority on the next unconditional rebuild. That is why
// the reconcile lists records rather than hunting for unfinished ones.
func TestRelayReconcileRepairsDeletionWithoutRecord(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-deleted")
	ctx := context.Background()
	body, err := json.Marshal(relay.Authority{
		Apps: []relay.AuthorityApp{{Account: fx.sqid, App: "gone",
			Incarnation: relay.NewIncarnation(), Regions: []string{"eu-west-1"}}},
		WrittenAt: time.Now().UTC(),
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(ctx, relay.AuthorityKey, body, ""); err != nil {
		t.Fatal(err)
	}

	if err := fx.srv.relayReconcile(ctx); err != nil {
		t.Fatal(err)
	}
	if _, ok := relayAuthority(t, fx).Lookup(fx.sqid, "gone"); ok {
		t.Fatal("the aggregate still names an app with no record")
	}
}

// TestRelayPurgeRepublishesTheAggregate: permanently deleting an app takes it
// out of every relay's authority AT ONCE, not within five minutes. The
// deletion walk sweeps the record; publishing right after is what makes the
// app's credentials stop verifying while the operator is still watching.
func TestRelayPurgeRepublishesTheAggregate(t *testing.T) {
	fx, c, csrf := trashPageFixture(t)
	armRelay(t, fx)
	ctx := context.Background()

	if _, err := fx.srv.relayEnable(ctx, fx.account(t), "kass", nil, "paul@keymail.dev"); err != nil {
		t.Fatal(err)
	}
	if _, ok := relayAuthority(t, fx).Lookup(fx.sqid, "kass"); !ok {
		t.Fatal("the enable did not publish")
	}

	status, _, body := postGated(t, fx, c, fx.u("/trash/delete"),
		url.Values{"csrf": {csrf}, "app": {"kass"}, "confirm": {"kass"}})
	if status != http.StatusSeeOther {
		t.Fatalf("purge = %d:\n%s", status, body)
	}
	if keys := appKeys(t, fx, "kass"); len(keys) != 0 {
		t.Fatalf("the bucket still holds %v", keys)
	}
	// No reconcile anywhere in this test: the purge itself republished.
	if _, ok := relayAuthority(t, fx).Lookup(fx.sqid, "kass"); ok {
		t.Fatal("a purged app still holds relay authority")
	}
}

// TestRelayEnableRefusesRouteWithoutDelivery: a TCP-address route this
// platform does not own, whose bundle has not opted into file delivery, could
// never pick up a rotated key — so it would block every future retirement
// forever. Enable refuses it BY NAME, and writes nothing on the way out.
func TestRelayEnableRefusesRouteWithoutDelivery(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-delivery")
	ctx := context.Background()
	fx.addRoute(t, registry.Route{Host: "x.titogo.oncarlos.com", App: "titogo",
		Kind: "service", Addr: "10.0.0.5:8080", Status: "ready"})

	_, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", nil, "alice@keymail.dev")
	if !errors.Is(err, errRelayUndelivered) {
		t.Fatalf("enable over an undeliverable route = %v, want a refusal", err)
	}
	if !strings.Contains(err.Error(), "x.titogo.oncarlos.com") {
		t.Fatalf("error = %q, want the route named", err)
	}
	if rec := relayRecord(t, fx, "titogo"); rec != nil {
		t.Fatalf("a refused enable left a record: %+v", rec)
	}
	if _, ok := appVars(t, fx, "titogo")[relayVar]; ok {
		t.Fatal("a refused enable delivered a key")
	}

	// Opting the app in makes the same route deliverable.
	if err := appconfig.SetExecDelivery(ctx, fx.rel, fx.sqid, "titogo", appconfig.ExecDeliveryFile); err != nil {
		t.Fatal(err)
	}
	if _, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", nil, "alice@keymail.dev"); err != nil {
		t.Fatalf("enable after opting into file delivery: %v", err)
	}
}

// TestRelayAdoptedRecordsTheIssuedIdentitiesAndRefusesAStranger: a member's
// ack stamps the identities issued AT THAT MOMENT against their own route, so
// an ack given before a later rotation cannot count for it — and a host that
// is not a route of this app is refused rather than appended, which is what
// keeps the record bounded.
//
// The route is a tcp-exec-file one because that is the only shape a member
// may acknowledge at all: anything the config sync restarts itself has its
// cutover observed (TestAPIRelayAdoptedRefusesARouteThePlatformRestarts).
func TestRelayAdoptedRecordsTheIssuedIdentitiesAndRefusesAStranger(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-adopted")
	ctx := context.Background()
	acct := fx.account(t)
	fx.addRoute(t, registry.Route{Host: "titogo.oncarlos.com", App: "titogo",
		Kind: "service", Status: "ready", Addr: "10.0.0.9:8080"})
	if err := appconfig.SetExecDelivery(ctx, fx.rel, fx.sqid, "titogo", appconfig.ExecDeliveryFile); err != nil {
		t.Fatal(err)
	}

	rec, err := fx.srv.relayEnable(ctx, acct, "titogo", nil, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.srv.relayAdopted(ctx, acct, "titogo", "somewhere.else.example", "alice@keymail.dev"); !errors.Is(err, errRelayUnknownHost) {
		t.Fatalf("adopted for a host that is not a route = %v, want a refusal", err)
	}
	after, err := fx.srv.relayAdopted(ctx, acct, "titogo", "titogo.oncarlos.com", "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if len(after.Cutovers) != 1 || after.Cutovers[0].Host != "titogo.oncarlos.com" ||
		after.Cutovers[0].Suffixes["eu-west-1"] != rec.IssuedSuffixes["eu-west-1"] {
		t.Fatalf("cutovers = %+v, want one naming the issued identity", after.Cutovers)
	}
	// A second ack for the same route replaces the first rather than growing
	// the record.
	again, err := fx.srv.relayAdopted(ctx, acct, "titogo", "titogo.oncarlos.com", "alice@keymail.dev")
	if err != nil || len(again.Cutovers) != 1 {
		t.Fatalf("second ack = %+v (%v), want one cutover still", again.Cutovers, err)
	}
}

// TestRelayRotateRefusesUntilHeartbeatsCarryKid: rotation's whole safety is
// that the console will not issue from a master a live relay has not loaded.
// Until the region's heartbeat lists the new kid, rotate refuses and the
// issuer pointer does not move; once it does, every enabled app is re-issued
// under the new kid.
func TestRelayRotateRefusesUntilHeartbeatsCarryKid(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-rotate")
	ctx := context.Background()
	t.Setenv(relayMastersEnv("eu-west-1"), "k1:"+relayTestMaster(1)+",k2:"+relayTestMaster(2))

	before, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", nil, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if !strings.HasSuffix(before.IssuedSuffixes["eu-west-1"], ":k1") {
		t.Fatalf("issued %q, want the first staged master", before.IssuedSuffixes["eu-west-1"])
	}
	hb := relay.Heartbeat{Box: "box1", Region: "eu-west-1", ReadAt: time.Now(),
		Listeners: []string{"udp:3478"}, Kids: []string{"k1"}}
	if err := relay.WriteHeartbeat(ctx, fx.rel, hb); err != nil {
		t.Fatal(err)
	}

	if _, err := fx.srv.relayRotate(ctx, "eu-west-1"); err == nil ||
		!strings.Contains(err.Error(), "k2") || !strings.Contains(err.Error(), "box1") {
		t.Fatalf("rotate before the relay loaded the kid = %v, want a refusal naming the box and kid", err)
	}
	if issuers, _, err := fx.srv.relayIssuers(ctx); err != nil || issuers["eu-west-1"] != "" {
		t.Fatalf("issuer pointer = %v (%v) after a refused rotation, want it untouched", issuers, err)
	}

	hb.Kids = []string{"k1", "k2"}
	if err := relay.WriteHeartbeat(ctx, fx.rel, hb); err != nil {
		t.Fatal(err)
	}
	res, err := fx.srv.relayRotate(ctx, "eu-west-1")
	if err != nil {
		t.Fatal(err)
	}
	if res.Kid != "k2" || res.Apps != 1 {
		t.Fatalf("rotate result = %+v, want kid k2 and one app re-issued", res)
	}
	// A SECOND rotate in the same window must refuse. The staged variable
	// still holds "k1,k2" — the old master stays there for the 28 hours before
	// it may be retired — and rotating "to the entry that is not the issuer"
	// would walk the pointer back onto the master about to be retired.
	if _, err := fx.srv.relayRotate(ctx, "eu-west-1"); err == nil ||
		!strings.Contains(err.Error(), "already the issuer") {
		t.Fatalf("re-rotate with nothing new staged = %v, want a refusal naming the last entry", err)
	}
	if issuers, _, err := fx.srv.relayIssuers(ctx); err != nil || issuers["eu-west-1"] != "k2" {
		t.Fatalf("issuer pointer = %v (%v) after a refused re-rotate, want it still on k2", issuers, err)
	}
	if d := time.Until(res.RetireNoEarlierThan); d < 27*time.Hour || d > 29*time.Hour {
		t.Fatalf("retire-no-earlier-than is %s away, want about 28h", d)
	}
	issuers, _, err := fx.srv.relayIssuers(ctx)
	if err != nil || issuers["eu-west-1"] != "k2" {
		t.Fatalf("issuer pointer = %v (%v), want k2", issuers, err)
	}
	after := relayRecord(t, fx, "titogo")
	if !strings.HasSuffix(after.IssuedSuffixes["eu-west-1"], ":k2") {
		t.Fatalf("issued %q after the rotation, want the new kid", after.IssuedSuffixes["eu-west-1"])
	}
	if after.Incarnation != before.Incarnation {
		t.Fatal("a rotation changed the incarnation — every live allocation would end")
	}
	if after.Op != before.Op+1 || after.Phase != relayPhaseDone {
		t.Fatalf("record = %+v, want one finished operation past %d", after, before.Op)
	}

	// Staging a THIRD master, with the pointer on the middle one, rotates
	// forward to it — the last entry, not merely a different one.
	t.Setenv(relayMastersEnv("eu-west-1"),
		"k1:"+relayTestMaster(1)+",k2:"+relayTestMaster(2)+",k3:"+relayTestMaster(3))
	hb.Kids = []string{"k1", "k2", "k3"}
	if err := relay.WriteHeartbeat(ctx, fx.rel, hb); err != nil {
		t.Fatal(err)
	}
	third, err := fx.srv.relayRotate(ctx, "eu-west-1")
	if err != nil {
		t.Fatal(err)
	}
	if third.Kid != "k3" {
		t.Fatalf("rotate with k1,k2,k3 staged and the pointer on k2 = %q, want k3", third.Kid)
	}
	if !strings.HasSuffix(relayRecord(t, fx, "titogo").IssuedSuffixes["eu-west-1"], ":k3") {
		t.Fatal("the app was not re-issued under the newest kid")
	}
}

// TestRelayReconcileRepairsOverwrittenEnvelope: the bundle holding SOMETHING
// under CARLOS_RELAY is not the same fact as it holding what this console
// wrote. A straggler write from a superseded console, or a member setting the
// variable themselves, leaves suffixes that still agree and a value that does
// not — which is exactly what the record's issued digest catches.
func TestRelayReconcileRepairsOverwrittenEnvelope(t *testing.T) {
	fx, priv := relayFixture(t, "code-relay-envelope")
	ctx := context.Background()

	rec, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", nil, "alice@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if rec.IssuedDigest == "" {
		t.Fatal("enable recorded no issued digest")
	}
	// Someone else's envelope, under the same name, with the record untouched.
	_, _, foreign := sealedFor(t, "not what the console issued")
	if _, err := appconfig.Apply(ctx, fx.rel, fx.sqid, "titogo",
		map[string]appconfig.Var{relayVar: {Kind: "secret", Envelope: foreign}}, nil); err != nil {
		t.Fatal(err)
	}

	if err := fx.srv.relayReconcile(ctx); err != nil {
		t.Fatal(err)
	}

	v := appVars(t, fx, "titogo")[relayVar]
	if v.Envelope == foreign {
		t.Fatal("the reconcile left an envelope this console did not write")
	}
	after := relayRecord(t, fx, "titogo")
	if after.IssuedDigest != relayEnvelopeDigest(v.Envelope) {
		t.Fatalf("record digest %q does not describe the bundle's envelope", after.IssuedDigest)
	}
	entries, err := relay.ParseConfig(openSecret(t, priv, v))
	if err != nil || len(entries) != 1 || entries[0].UsernameSuffix != after.IssuedSuffixes["eu-west-1"] {
		t.Fatalf("re-delivered value = %+v (%v), want the record's own identity", entries, err)
	}
}

// TestRelayWriterLeaseExcludesSecondHolder: two consoles over one bucket, one
// lease. The second is excluded until the first's minute lapses — and the
// first, once it has been taken over, fails its next renewal and drops the
// flag every relay write checks, which is what stops a superseded writer
// finishing a walk the new holder has already started.
func TestRelayWriterLeaseExcludesSecondHolder(t *testing.T) {
	fx, _ := relayFixture(t, "code-relay-lease")
	ctx := context.Background()

	other := newConsoleFixture(t)
	other.srv.cfg.Releases = fx.rel
	other.srv.cfg.RelayRegions = []string{"eu-west-1"}

	base := time.Now()
	first, second := base, base.Add(time.Second)
	fx.srv.relayNow = func() time.Time { return first }
	other.srv.relayNow = func() time.Time { return second }

	if err := fx.srv.relayAcquire(ctx); err != nil || !fx.srv.relayHold() {
		t.Fatalf("first acquire = %v, held = %v", err, fx.srv.relayHold())
	}
	err := other.srv.relayAcquire(ctx)
	if !errors.Is(err, errRelayNotHeld) {
		t.Fatalf("second acquire inside the TTL = %v, want a refusal", err)
	}
	if other.srv.relayHold() {
		t.Fatal("the second console thinks it holds the lease")
	}
	if err := other.srv.relayReconcile(ctx); !errors.Is(err, errRelayNotHeld) {
		t.Fatalf("reconcile without the lease = %v, want a refusal", err)
	}

	// Past the expiry, the lease is takeable.
	second = base.Add(relayLeaseTTL + time.Second)
	if err := other.srv.relayAcquire(ctx); err != nil || !other.srv.relayHold() {
		t.Fatalf("acquire after expiry = %v, held = %v", err, other.srv.relayHold())
	}

	// And the first console's next renewal fails, dropping its own flag.
	first = base.Add(relayLeaseTTL + 2*time.Second)
	if err := fx.srv.relayAcquire(ctx); err == nil {
		t.Fatal("the superseded console renewed a lease it does not hold")
	}
	if fx.srv.relayHold() {
		t.Fatal("the superseded console still believes it holds the lease")
	}
	if _, err := fx.srv.relayEnable(ctx, fx.account(t), "titogo", nil, "alice@keymail.dev"); !errors.Is(err, errRelayNotHeld) {
		t.Fatalf("enable without the lease = %v, want a refusal", err)
	}
}