package console

import (
	"bytes"
	"encoding/json"
	"log"
	"os"
	"strings"
	"testing"
	"time"

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

// The bucket is the control plane, and the console is not the only minter:
// `carlos accounts create` writes accounts/<sqid>/account.json and a registry
// row without ever opening the console database. Before this, the console
// simply did not know such an account existed — so `console member -add` would
// refuse a sqid that demonstrably exists, on the very box that created it.
func TestMaterializeAccountsFromTheBucket(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	created := time.Unix(1700000000, 0).UTC()

	const kassID, blogID = int64(793481920157), int64(42)
	for _, a := range []struct {
		id   int64
		name string
	}{{kassID, "Kass"}, {blogID, "Blog"}} {
		if err := WriteAccountManifest(ctx, rel, a.id, domains.Sqid(a.id), a.name, created); err != nil {
			t.Fatal(err)
		}
	}

	n, err := MaterializeAccounts(ctx, rel, st)
	if err != nil {
		t.Fatal(err)
	}
	if n != 2 {
		t.Errorf("materialized %d accounts, want 2", n)
	}
	acct, err := st.AccountBySqid(domains.Sqid(kassID))
	if err != nil || acct == nil {
		t.Fatalf("AccountBySqid(Kass) = %v %v, want the materialized account", acct, err)
	}
	// The id in the manifest is taken as GIVEN. Minting a new one here would
	// re-home every platform domain the account owns.
	if acct.ID != kassID {
		t.Errorf("materialized id = %d, want the manifest's %d", acct.ID, kassID)
	}
	if acct.Name != "Kass" || !acct.CreatedAt.Equal(created) {
		t.Errorf("materialized account = %+v, want Kass created %v", acct, created)
	}

	// Idempotent: a second pass finds nothing new, which is what makes it safe
	// to run on every console start and before every membership operation.
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 0 {
		t.Errorf("second pass materialized %d (%v), want 0", n, err)
	}
}

// A manifest whose sqid disagrees with its id is corruption, not a new
// account: materializing it would put an account at a domain that stops
// resolving the moment anything recomputes the sqid. Skipped loudly, never
// guessed at — the same rule internal/domains applies to the same objects.
func TestMaterializeSkipsManifestsThatDisagreeWithThemselves(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()

	// sqid in the path and body agree with each other but not with Sqid(id).
	if _, err := rel.Put(ctx, "accounts/wrong/account.json",
		[]byte(`{"id":42,"sqid":"wrong","name":"Impostor"}`), ""); err != nil {
		t.Fatal(err)
	}
	// Body sqid disagrees with the path.
	if _, err := rel.Put(ctx, "accounts/"+domains.Sqid(43)+"/account.json",
		[]byte(`{"id":43,"sqid":"elsewhere","name":"Mislabelled"}`), ""); err != nil {
		t.Fatal(err)
	}
	if _, err := rel.Put(ctx, "accounts/broken/account.json", []byte("{not json"), ""); err != nil {
		t.Fatal(err)
	}
	if _, err := rel.Put(ctx, "accounts/empty/account.json", []byte("{}"), ""); err != nil {
		t.Fatal(err)
	}
	// One good manifest alongside: a bad neighbour must not stop the pass.
	if err := WriteAccountManifest(ctx, rel, 42, domains.Sqid(42), "Blog", time.Now()); err != nil {
		t.Fatal(err)
	}

	n, err := MaterializeAccounts(ctx, rel, st)
	if err != nil {
		t.Fatal(err)
	}
	if n != 1 {
		t.Errorf("materialized %d, want only the well-formed manifest", n)
	}
	for _, sqid := range []string{"wrong", "elsewhere", "broken", "empty", domains.Sqid(43)} {
		if a, _ := st.AccountBySqid(sqid); a != nil {
			t.Errorf("materialized %q from a manifest that disagrees with itself: %+v", sqid, a)
		}
	}
	if a, _ := st.AccountBySqid(domains.Sqid(42)); a == nil {
		t.Error("the well-formed manifest was not materialized")
	}
}

// Materialization is INSERT-only. An account the console already knows keeps
// its row: the console is where names are edited, the manifest is written
// once, and letting a stale manifest name win would rename accounts behind
// the operator's back. Above all it must never reassign an id.
func TestMaterializeNeverRewritesAKnownAccount(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0).UTC()

	if err := st.CreateAccount(42, domains.Sqid(42), "Current name", "paul@keymail.dev", now); err != nil {
		t.Fatal(err)
	}
	if err := WriteAccountManifest(ctx, rel, 42, domains.Sqid(42), "Stale name", now); err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 0 {
		t.Fatalf("materialized %d (%v) for an account the console already had", n, err)
	}
	acct, _ := st.AccountBySqid(domains.Sqid(42))
	if acct == nil || acct.Name != "Current name" || acct.ID != 42 {
		t.Errorf("known account was rewritten: %+v", acct)
	}
	// And its membership survives — materialization touches memberships never.
	if role, _ := st.Role(42, "paul@keymail.dev"); role != RoleOwner {
		t.Errorf("owner role after materialize = %q, want owner", role)
	}
}

// FOUND IN PRODUCTION: `carlos accounts create` recorded an owner NOWHERE, so
// the accounts kass and keymail were migrated into had no members — and
// membership IS authorization, so the operator could not reach either of them
// in the console at all. The account existed, the apps were in it, and the only
// door was a `console member -add … -role owner` that nothing told anybody to
// run.
//
// The fix keeps the boundary the carlos binary is held to (it must never open
// this database): `create` names the owner in the BUCKET manifest, and this
// pass — the console's own reader of that manifest — turns it into the owner
// membership. Same object, same reader, on every box.
func TestMaterializeGivesANewAccountTheManifestOwner(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()

	// Exactly what `carlos accounts create -name Kass -owner paul@keymail.dev`
	// leaves behind, minted through the same allocator.
	id, sqid, err := domains.MintAccount(ctx, rel, "Kass", "paul@keymail.dev", time.Unix(1700000000, 0))
	if err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want 1", n, err)
	}
	if role, _ := st.Role(id, "paul@keymail.dev"); role != RoleOwner {
		t.Errorf("owner role after materialize = %q, want owner — the account is unreachable without it", role)
	}
	// And it lands where the console looks it up from: the switcher.
	accts, err := st.AccountsFor("paul@keymail.dev")
	if err != nil {
		t.Fatal(err)
	}
	if len(accts) != 1 || accts[0].Sqid != sqid || accts[0].Role != RoleOwner {
		t.Errorf("AccountsFor(owner) = %+v, want the new account as owner", accts)
	}
	// Idempotent: the account is known and owned, so a second pass does nothing.
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 0 {
		t.Errorf("second pass materialized %d (%v), want 0", n, err)
	}
}

// THE MANIFEST IS NOT WRITE-ONCE, and the owner is only adopted at BIRTH.
//
// `MintAccount` claims the object with a create-only Put, but that is a
// convention of one code path, not a property of the object: release.Store.Put
// takes its ifMatch from the caller, the deployment bucket has no Object Lock,
// and the host IAM role holds unconditional s3:PutObject over the whole bucket
// with nothing excluding accounts/*/account.json. So anyone who can write to
// the bucket can REWRITE a live account's manifest.
//
// If materialization adopted an owner on any pass, that rewrite would be an
// authority grant: name yourself owner of a vacant account, wait for the next
// console start or `console member` invocation — both run this pass — and you
// hold owner over a live account with its apps in it. That is a genuine
// widening of the trust model, not something bucket write access already
// implied: forging a channel pointer needs CARLOS_RELEASE_KEY, which most
// bucket-writers do not have, and this path needs no key at all.
//
// The transactional vacancy check in AdoptManifestOwner does not help here. It
// is TOCTOU-safe, but the account was LEGITIMATELY vacant; the manifest is what
// changed afterwards.
//
// So adoption happens once, when the console first creates the account row, and
// never again. Safe now that -owner is required at create time: manifests are
// born with their owner.
func TestMaterializeNeverAdoptsAnOwnerAddedToAManifestAfterTheFact(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0)

	// A legitimately vacant account: an old manifest, from before `owner`.
	if err := WriteAccountManifest(ctx, rel, 42, domains.Sqid(42), "Kass", now); err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want 1", n, err)
	}
	if members, _ := st.Members(42); len(members) != 0 {
		t.Fatalf("Members = %+v, want the account vacant to begin with", members)
	}

	// Somebody with plain bucket write access — no signing key — rewrites the
	// manifest naming themselves.
	rewriteManifest(t, rel, 42, "attacker@keymail.dev")
	// This pass runs on every console start and before every `console member`.
	if _, err := MaterializeAccounts(ctx, rel, st); err != nil {
		t.Fatal(err)
	}
	if role, _ := st.Role(42, "attacker@keymail.dev"); role != "" {
		t.Errorf("a rewritten manifest granted %q over a live account — bucket write became "+
			"console authority, with no signing key involved", role)
	}
	if members, _ := st.Members(42); len(members) != 0 {
		t.Errorf("Members = %+v, want the account still vacant", members)
	}
}

// The same rule stated the other way, and the case an operator will actually
// hit: an owner added to a manifest AFTER the account first materialized is
// deliberately not adopted, however well-meant. The operator path for that is
// `console member -add <address> <sqid> -role owner`, which is box-side
// authority rather than bucket-side.
func TestMaterializeLeavesALateManifestOwnerToTheOperator(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0)

	if err := WriteAccountManifest(ctx, rel, 42, domains.Sqid(42), "Kass", now); err != nil {
		t.Fatal(err)
	}
	if _, err := MaterializeAccounts(ctx, rel, st); err != nil {
		t.Fatal(err)
	}
	rewriteManifest(t, rel, 42, "paul@keymail.dev")
	if _, err := MaterializeAccounts(ctx, rel, st); err != nil {
		t.Fatal(err)
	}
	if role, _ := st.Role(42, "paul@keymail.dev"); role != RoleOwner {
		// The box-side verb is what fills it, and it still works.
		if err := st.AddMember(42, "paul@keymail.dev", RoleOwner, now); err != nil {
			t.Fatal(err)
		}
	} else {
		t.Error("a late manifest owner was adopted — the manifest is not write-once, " +
			"so that is a standing grant anyone with bucket write can issue")
	}
}

// THE COUPLING TEST. Once an owner exists, the only thing standing between "the
// operator removed them" and "the manifest puts them back" is that the account
// never becomes vacant again — and that is enforced by lastOwnerGuard, which
// lives in member removal, not here. This diff leans on that guard implicitly;
// exercising it end to end means a future change to removal (or to account
// deletion) cannot silently reopen the adoption path.
func TestARemovedOwnerIsNeverRegrantedByTheManifest(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0)

	id, _, err := domains.MintAccount(ctx, rel, "Kass", "paul@keymail.dev", now)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := MaterializeAccounts(ctx, rel, st); err != nil {
		t.Fatal(err)
	}
	if role, _ := st.Role(id, "paul@keymail.dev"); role != RoleOwner {
		t.Fatalf("role = %q, want the manifest owner adopted at birth", role)
	}

	// Link one: the account cannot be emptied. Removing the last owner is
	// refused, so there is no vacancy for a re-read to fill.
	if err := st.RemoveMember(id, "paul@keymail.dev"); err == nil {
		t.Fatal("removed the last owner — the account is now vacant and the manifest can refill it")
	}

	// Link two: with ownership transferred, the original CAN be removed — and
	// must stay removed across every subsequent pass, forever.
	if err := st.AddMember(id, "successor@keymail.dev", RoleOwner, now); err != nil {
		t.Fatal(err)
	}
	if err := st.RemoveMember(id, "paul@keymail.dev"); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 3; i++ {
		if _, err := MaterializeAccounts(ctx, rel, st); err != nil {
			t.Fatal(err)
		}
	}
	if role, _ := st.Role(id, "paul@keymail.dev"); role != "" {
		t.Errorf("the removed owner came back as %q — the manifest is behaving as a standing grant", role)
	}
	if role, _ := st.Role(id, "successor@keymail.dev"); role != RoleOwner {
		t.Errorf("successor role = %q, want owner", role)
	}
}

// The manifest is written ONCE and read on every start, so it must never be a
// standing grant. An account that has an owner is administered by the console;
// re-reading the object must not demote them, replace them, or add the
// manifest's address alongside them.
func TestMaterializeNeverReplacesAnExistingOwner(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0)

	id, sqid, err := domains.MintAccount(ctx, rel, "Kass", "original@keymail.dev", now)
	if err != nil {
		t.Fatal(err)
	}
	// Ownership was transferred in the console after creation.
	if err := st.CreateAccount(id, sqid, "Kass", "successor@keymail.dev", now); err != nil {
		t.Fatal(err)
	}
	if _, err := MaterializeAccounts(ctx, rel, st); err != nil {
		t.Fatal(err)
	}
	if role, _ := st.Role(id, "successor@keymail.dev"); role != RoleOwner {
		t.Errorf("the console's owner became %q after a materialize pass", role)
	}
	if role, _ := st.Role(id, "original@keymail.dev"); role != "" {
		t.Errorf("the manifest re-granted %q to the address it named at creation — "+
			"a write-once object must not be a standing grant", role)
	}
}

// An owner address the sign-in form would refuse is a membership nobody can
// ever claim. Skipped loudly, and the ACCOUNT still materializes: the identity
// is sound even when the owner field is not, and refusing the whole manifest
// would make one bad address hide an account from the console entirely.
func TestMaterializeSkipsAnUnusableManifestOwner(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()

	sqid := domains.Sqid(42)
	if _, err := rel.Put(ctx, domains.AccountManifestKey(sqid),
		[]byte(`{"id":42,"sqid":"`+sqid+`","name":"Kass","owner":"not-an-address"}`), ""); err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want the account itself", n, err)
	}
	if members, _ := st.Members(42); len(members) != 0 {
		t.Errorf("Members = %+v, want nobody — an address keymail cannot authenticate is not an owner", members)
	}
}

// An old manifest — written before the field existed — has no owner and stays
// ownerless. Additive means the pre-owner world is unchanged.
func TestMaterializeLeavesAnOwnerlessManifestOwnerless(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	if err := WriteAccountManifest(ctx, rel, 42, domains.Sqid(42), "Legacy", time.Unix(1700000000, 0)); err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want 1", n, err)
	}
	if members, _ := st.Members(42); len(members) != 0 {
		t.Errorf("Members = %+v, want nobody — inventing an owner is minting authority from nothing", members)
	}
}

// THE KASS GAP, closed: a deletion performed anywhere becomes real here.
//
// Before this, MaterializeAccounts merely SKIPPED a tombstoned manifest, so an
// account deleted on another box (or by `carlos accounts delete`) kept its
// console row, its memberships and its apps forever — the console said the
// account was live, the control plane said it was gone, and only hand-run SQL
// could settle it. The bucket is the control plane: it has already ruled.
func TestMaterializeReapsTombstonedAccounts(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0).UTC()

	id, sqid, err := domains.MintAccount(ctx, rel, "Testco", "paul@keymail.dev", now)
	if err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want the new account", n, err)
	}
	if _, err := st.CreateAppIn(id, "blog", "paul@keymail.dev", now); err != nil {
		t.Fatal(err)
	}
	if err := st.SetLastAccount("paul@keymail.dev", id); err != nil {
		t.Fatal(err)
	}

	// Deleted somewhere else entirely, while this console knew nothing about it.
	if err := domains.TombstoneAccount(ctx, rel, sqid, now); err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 0 {
		t.Fatalf("materialized %d (%v) on the reaping pass, want 0 — a reap is not a birth", n, err)
	}

	if a, err := st.AccountBySqid(sqid); err != nil || a != nil {
		t.Errorf("AccountBySqid after the tombstone = %+v %v, want nothing: the control plane has ruled", a, err)
	}
	// Invisible, but REMEMBERED — see the resurrection test for what forgetting costs.
	if a, deleted, err := st.AccountAnyBySqid(sqid); err != nil || a == nil || !deleted {
		t.Fatalf("AccountAnyBySqid = %+v %v %v, want a marked row", a, deleted, err)
	}
	if members, _ := st.Members(id); len(members) != 0 {
		t.Errorf("Members = %+v, want none — membership against a dead account is access to nothing", members)
	}
	if apps, _ := st.AppsIn(id); len(apps) != 0 {
		t.Errorf("AppsIn = %v, want none", apps)
	}
	if trashed, _ := st.TrashedAppsIn(id); len(trashed) != 0 {
		t.Errorf("TrashedAppsIn = %+v, want none — the app rows are reaped with the account", trashed)
	}
	if last, err := st.LastAccount("paul@keymail.dev"); err != nil || last != 0 {
		t.Errorf("LastAccount = %d %v, want 0 — the MRU must not land anyone in a reaped account", last, err)
	}
	// Stable: a second pass finds a marked row and a tombstoned manifest agreeing.
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 0 {
		t.Errorf("second pass materialized %d (%v), want 0", n, err)
	}
}

// THE SECURITY TEST. Owner adoption is birth-only, and "birth" must mean
// FIRST-EVER — not "first since the row went away".
//
// Deleting the row would forget the account was ever seen, and forgetting
// re-arms the whole attack materializeOwner exists to prevent: tombstone an
// account (or wait for someone to delete one), let the console reap it, then
// rewrite the manifest un-tombstoned with yourself as owner. A console that
// treats that as a first materialization re-births the account — same id, same
// sqid, same platform hostnames, same apps prefix — and adopts the attacker.
// Bucket write access, no signing key, a whole account.
//
// So a marked row is tombstone MEMORY, and memory does not argue with a
// rewritten manifest. An account id is spent once.
func TestMaterializeRefusesResurrection(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0).UTC()

	id, sqid, err := domains.MintAccount(ctx, rel, "Testco", "paul@keymail.dev", now)
	if err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want the new account", n, err)
	}
	if err := domains.TombstoneAccount(ctx, rel, sqid, now); err != nil {
		t.Fatal(err)
	}
	if _, err := MaterializeAccounts(ctx, rel, st); err != nil {
		t.Fatal(err)
	}
	if _, deleted, _ := st.AccountAnyBySqid(sqid); !deleted {
		t.Fatal("the account was not reaped, so this test is not testing what it says")
	}

	// Somebody with plain s3:PutObject — no CARLOS_RELEASE_KEY, no console
	// session — writes the manifest back, live, naming themselves.
	rewriteManifest(t, rel, id, "eve@evil.example")

	// The refusal must be EXPLICIT, not an accident of some later check. Delete
	// the corruption arm and every assertion below still passes — refusal falls
	// through to "known account, stop here" — while the operator loses the only
	// signal that somebody rewrote a dead account's manifest. So the log line is
	// part of the contract and is pinned here.
	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	// Every console start runs this pass, and so does every `console member`.
	for i := 0; i < 2; i++ {
		if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 0 {
			t.Fatalf("pass %d materialized %d (%v) — a spent account id was re-birthed from a rewritten manifest", i, n, err)
		}
	}
	if a, _ := st.AccountBySqid(sqid); a != nil {
		t.Errorf("the account came back as %+v — reap, rewrite, re-birth is an account takeover", a)
	}
	if _, deleted, _ := st.AccountAnyBySqid(sqid); !deleted {
		t.Error("the marked row lost its mark — the console has forgotten the tombstone")
	}
	if role, _ := st.Role(id, "eve@evil.example"); role != "" {
		t.Errorf("the rewritten manifest granted %q — bucket write became console authority", role)
	}
	if accts, _ := st.AccountsFor("eve@evil.example"); len(accts) != 0 {
		t.Errorf("AccountsFor(attacker) = %+v, want nothing adopted", accts)
	}
	if members, _ := st.Members(id); len(members) != 0 {
		t.Errorf("Members = %+v, want none", members)
	}
	if !strings.Contains(logged.String(), "refusing to resurrect") {
		t.Errorf("nothing said a word about the rewritten manifest; log was:\n%s\n"+
			"a manifest that came back to life over a tombstoned account is corruption, "+
			"and an operator has to hear about it", logged.String())
	}
}

// App-row convergence: the console's rows follow the bucket, because trash and
// purge can both happen while this console is down (headless `carlos`, another
// box, the auto-purge pass). Four cases, and case (d) is the one that keeps
// convergence honest.
func TestMaterializeAppsConverges(t *testing.T) {
	st := openTestStore(t)
	rel := release.NewDirStore(t.TempDir())
	ctx := t.Context()
	now := time.Unix(1700000000, 0).UTC()
	trashedAt := now.Add(-48 * time.Hour)

	id, sqid, err := domains.MintAccount(ctx, rel, "Testco", "paul@keymail.dev", now)
	if err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, rel, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want the account", n, err)
	}
	for _, app := range []string{"alpha", "beta", "gamma", "delta"} {
		if _, err := st.CreateAppIn(id, app, "paul@keymail.dev", now); err != nil {
			t.Fatal(err)
		}
	}
	shipped := func(app string) {
		t.Helper()
		if _, err := rel.Put(ctx, "apps/"+sqid+"/"+app+"/channels/stable.json", []byte(`{}`), ""); err != nil {
			t.Fatal(err)
		}
	}
	// (a) live row, trashed headlessly: the marker exists in the bucket.
	shipped("alpha")
	if err := release.TrashApp(ctx, rel, sqid, "alpha", "paul@keymail.dev", trashedAt); err != nil {
		t.Fatal(err)
	}
	// (b) marked row, restored headlessly: marker gone, objects still there.
	shipped("beta")
	if err := st.TrashAppRow(id, "beta", trashedAt); err != nil {
		t.Fatal(err)
	}
	// (c) marked row, purged: no marker and no objects at all.
	if err := st.TrashAppRow(id, "gamma", trashedAt); err != nil {
		t.Fatal(err)
	}
	// (d) live row, never shipped: no marker, no objects — a legitimate claim.
	// Nothing about it is distinguishable from (c) except that its row is LIVE.

	// Another account's keys, for a sqid this console has never heard of: they
	// must contribute nothing to anybody's decision.
	if _, err := rel.Put(ctx, "apps/"+domains.Sqid(999999)+"/alpha/channels/stable.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}

	if err := MaterializeApps(ctx, rel, st); err != nil {
		t.Fatal(err)
	}

	if live, _ := st.HasAppIn(id, "alpha"); live {
		t.Error("(a) alpha has a trash marker in the bucket but its row is still live")
	}
	if trashed, _ := st.HasTrashedAppIn(id, "alpha"); !trashed {
		t.Error("(a) alpha was not marked trashed from its bucket marker")
	}
	if live, _ := st.HasAppIn(id, "beta"); !live {
		t.Error("(b) beta's marker is gone and its objects remain — the row should be live again")
	}
	if trashed, _ := st.HasTrashedAppIn(id, "gamma"); trashed {
		t.Error("(c) gamma is purged in the bucket but its row still holds the name")
	}
	if live, _ := st.HasAppIn(id, "gamma"); live {
		t.Error("(c) gamma came back to life")
	}
	if accts, _ := st.AppAccounts("gamma"); len(accts) != 0 {
		t.Errorf("(c) gamma still has rows %v — the row IS the name reservation, so purge must free it", accts)
	}
	if live, _ := st.HasAppIn(id, "delta"); !live {
		t.Error("(d) a claimed-but-never-shipped app was eaten by convergence — " +
			"an app with no objects is indistinguishable from a purged one except by its own row")
	}

	// The cooloff clock comes from the MARKER, not from this pass: the bucket
	// records when the app was trashed, and a console catching up days later
	// must not restart the 30 days.
	trashedRows, err := st.TrashedAppsIn(id)
	if err != nil {
		t.Fatal(err)
	}
	if len(trashedRows) != 1 || trashedRows[0].App != "alpha" || !trashedRows[0].DeletedAt.Equal(trashedAt) {
		t.Errorf("TrashedAppsIn = %+v, want alpha trashed at %v", trashedRows, trashedAt)
	}

	// Idempotent: convergence has converged.
	if err := MaterializeApps(ctx, rel, st); err != nil {
		t.Fatal(err)
	}
	apps, _ := st.AppsIn(id)
	if len(apps) != 2 || apps[0] != "beta" || apps[1] != "delta" {
		t.Errorf("AppsIn after a second pass = %v, want [beta delta]", apps)
	}
}

// A MISTYPED BUCKET NAME MUST NOT FREE NAMES.
//
// Case (c) is the only branch that reasons from absence, and an empty listing
// is exactly what CARLOS_DEPLOYMENT_BUCKET pointing one character wrong
// produces: a listing that SUCCEEDS and returns nothing. Purging on that reads
// "everything was purged everywhere" from what is really "I am looking in the
// wrong place" — and it is unrecoverable, because this pass is row-driven: once
// the rows are gone, the markers in the real bucket are invisible to it forever
// and the names are free for anyone to claim.
func TestMaterializeAppsRefusesAnEmptyBucket(t *testing.T) {
	st := openTestStore(t)
	ctx := t.Context()
	now := time.Unix(1700000000, 0).UTC()

	empty := release.NewDirStore(t.TempDir())
	id, sqid, err := domains.MintAccount(ctx, empty, "Testco", "paul@keymail.dev", now)
	if err != nil {
		t.Fatal(err)
	}
	if n, err := MaterializeAccounts(ctx, empty, st); err != nil || n != 1 {
		t.Fatalf("materialized %d (%v), want the account", n, err)
	}
	if _, err := st.CreateAppIn(id, "gamma", "paul@keymail.dev", now); err != nil {
		t.Fatal(err)
	}
	if err := st.TrashAppRow(id, "gamma", now.Add(-48*time.Hour)); err != nil {
		t.Fatal(err)
	}

	// The wrong bucket: account manifests happen to be there, nothing under
	// apps/ is. (A genuinely wrong bucket has neither; this store is the
	// STRONGER test, since it still lists something.)
	var logged bytes.Buffer
	log.SetOutput(&logged)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })
	if err := MaterializeApps(ctx, empty, st); err != nil {
		t.Fatal(err)
	}
	if trashed, _ := st.HasTrashedAppIn(id, "gamma"); !trashed {
		t.Error("an empty apps/ listing released a held name — one wrong environment variable " +
			"is now enough to free every trashed app name in the deployment")
	}
	if accts, _ := st.AppAccounts("gamma"); len(accts) != 1 {
		t.Errorf("AppAccounts(gamma) = %v, want the reservation still held", accts)
	}
	if !strings.Contains(logged.String(), "CARLOS_DEPLOYMENT_BUCKET") {
		t.Errorf("the refusal was silent; log was:\n%s\nan operator has to be told why "+
			"names are not being released", logged.String())
	}

	// And the tripwire is about EMPTINESS, not about this app: one unrelated
	// key under apps/ is enough evidence that the bucket is the right bucket,
	// and gamma — no marker, no keys of its own — purges as before.
	if _, err := empty.Put(ctx, "apps/"+sqid+"/somethingelse/channels/stable.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	if err := MaterializeApps(ctx, empty, st); err != nil {
		t.Fatal(err)
	}
	if accts, _ := st.AppAccounts("gamma"); len(accts) != 0 {
		t.Errorf("AppAccounts(gamma) = %v after a non-empty listing, want the name freed — "+
			"the tripwire must hold back purges only while there is no evidence at all", accts)
	}
}

// rewriteManifest overwrites a live account manifest with one naming owner —
// what a bucket-writer does, and the thing MintAccount's create-only Put does
// NOT prevent anyone else from doing. The etag is read and passed because the
// DirStore emulates compare-and-swap; S3 has no such constraint at all, so a
// caller holding s3:PutObject simply writes the object.
func rewriteManifest(t *testing.T, rel release.Store, id int64, owner string) {
	t.Helper()
	ctx := t.Context()
	sqid := domains.Sqid(id)
	key := domains.AccountManifestKey(sqid)
	_, etag, err := rel.Get(ctx, key)
	if err != nil {
		t.Fatal(err)
	}
	body, err := json.Marshal(domains.AccountManifest{
		ID: id, Sqid: sqid, Name: "Kass", Owner: owner, CreatedAt: time.Unix(1700000000, 0).UTC(),
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := rel.Put(ctx, key, body, etag); err != nil {
		t.Fatal(err)
	}
}

// A deployment with no bucket configured at all still has to work: this is
// best-effort catch-up, not a precondition.
func TestMaterializeWithNoStoreIsANoOp(t *testing.T) {
	st := openTestStore(t)
	if n, err := MaterializeAccounts(t.Context(), nil, st); err != nil || n != 0 {
		t.Errorf("MaterializeAccounts(nil store) = %d %v, want 0 nil", n, err)
	}
	// And app convergence above all: with no bucket to compare against, "no
	// keys" would otherwise read as "purged everywhere".
	if err := MaterializeApps(t.Context(), nil, st); err != nil {
		t.Errorf("MaterializeApps(nil store) = %v, want nil", err)
	}
}