// The grant walk (§5): the provisioning sequence every other verb trusts.
// Each numbered step is check-then-create, so the walk is re-runnable after a
// failure at any point, and the secret's ONLY landing place is the app's
// sealed config (§7) — it is delivered before it is recorded, never after.
package objstore

import (
	"context"
	"crypto/ecdh"
	"errors"
	"fmt"
	"regexp"
	"time"

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

// ErrNotDeclared is a grant for an app whose member never asked for a store.
// The operator's verb never invents the declaration: the env prefix is the
// member's choice, and a store nobody declared has nowhere to be delivered.
var ErrNotDeclared = errors.New("objstore: no store is declared for this app")

// ErrEnvPrefixLocked is a re-declaration that tries to move the env prefix of
// an already-provisioned store. Renaming the delivered set would leave the
// old names in config.json holding live credentials while the app reads
// nothing — so the prefix is fixed at first provision (the console door
// raises this; the grant walk itself never re-reads a prefix choice).
var ErrEnvPrefixLocked = errors.New("objstore: store already provisioned — env prefix is fixed")

// ErrNotProvisioned is a rotation verb aimed at a store that was never
// provisioned — declared-only, granted-but-half-walked, or absent entirely.
// Rotation edits an existing credential estate; it never creates one, because
// the coordinates it reseals (bucket, region, endpoint, IAM user) come from
// the record, and there is no defensible guess for them. The caller's repair
// for this state is Grant, not a rotate.
var ErrNotProvisioned = errors.New("objstore: store is not provisioned")

// ErrNoPendingFinish is a finish with nothing to finish. Finish is the one
// destructive step in the rotation state machine — it kills a key that may
// still be in use — so it refuses rather than succeeding vacuously: a member
// who believes they just retired the old key, when in truth an earlier finish
// already did, has a different picture of the estate than the estate does.
var ErrNoPendingFinish = errors.New("objstore: no pending key to finish")

// UserPath is the IAM path every store principal lives under, so an operator
// (or a policy) can name the whole population in one place.
const UserPath = "/carlos-store/"

// Deployment is the per-deployment backend configuration the walks need: who
// provisions (Backend), where (Region), an optional non-AWS Endpoint
// delivered to the app, and the prefix minted bucket names carry.
type Deployment struct {
	Backend      string
	Region       string
	Endpoint     string
	BucketPrefix string
}

// MintedBucket is the name a store gets when the deployment mints one — the
// adopt case (an existing bucket named in the grant) never comes through here.
func MintedBucket(d Deployment, acct, app string) string {
	return MintedBucketIn(d, acct, app, "")
}

// MintedBucketIn is MintedBucket for one store of an app. An environment
// store carries the environment as a third component, so two stores of one
// app can never resolve to one bucket.
func MintedBucketIn(d Deployment, acct, app, env string) string {
	name := d.BucketPrefix + acct + "-" + app
	if env != "" {
		name += "-" + env
	}
	return name
}

// UserName is the IAM principal that owns an app's store credentials. One
// user per app: a key can be reaped or rotated without touching any other
// app, and the user's inline policy names exactly one bucket.
func UserName(acct, app string) string { return UserNameIn(acct, app, "") }

// UserNameIn is UserName for one store of an app — one principal per STORE,
// which is what keeps an environment store's key reapable and rotatable
// without touching the app's own, and its inline policy naming exactly one
// bucket.
func UserNameIn(acct, app, env string) string {
	name := "carlos-store-" + acct + "-" + app
	if env != "" {
		name += "-" + env
	}
	return name
}

// MaxIAMUserName is AWS's hard ceiling on an IAM user name. A name over it is
// refused by CreateUser, deep in the walk and against live AWS — the worst
// place to learn a store cannot be named, since the grant block is already
// written by then and the caller is left with a record naming an estate that
// was never created.
const MaxIAMUserName = 64

// ValidUserName gates the principal name the same way ValidBucketName gates
// the bucket: in front of the walk, before anything is created or recorded.
//
// The bucket half needs no separate ceiling — bucketNameRe's own length bound
// IS S3's 63 — but nothing bounded the IAM name before environment stores
// added a third component to it.
func ValidUserName(name string) bool { return name != "" && len(name) <= MaxIAMUserName }

// EnvDelivered reports which of the delivered variables are NOT present in
// the app's config manifest AS THIS PACKAGE WRITES THEM — an empty result
// means fully delivered. _ENDPOINT counts only when wantEndpoint (a
// deployment with no endpoint never delivers one). A manifest that does not
// exist reads as every name missing, which is what an undelivered app is.
//
// Kind is part of "present": the coordinates must be plain "env" vars and
// BOTH key halves must be "secret". A name occupied by the wrong kind counts
// as missing, because that is what it is to the app — a key half stored as a
// plaintext "env" var is a credential the app cannot use as one (and a
// coordinate stored as a sealed "secret" renders from an envelope nobody
// sealed for it). Overwriting drift like that is exactly what §7 promises a
// grant or rotate does; reading it as healthy would leave the walk refusing
// to repair the one state it exists to repair. Values are deliberately NOT
// compared: this seam answers "is the set delivered", and the walk rewrites
// every value it owns whenever it runs anyway.
//
// The grant walk's idempotence short-circuit and the console's missing-vars
// report are the same question asked twice, so they ask it here: a member who
// unsets or clobbers a delivered var must see the same "not delivered" the
// walk acts on.
func EnvDelivered(ctx context.Context, s release.Store, acct, app, prefix string, wantEndpoint bool) (missing []string, err error) {
	return EnvDeliveredIn(ctx, s, acct, app, "", prefix, wantEndpoint)
}

// EnvDeliveredIn is EnvDelivered against the bundle a store actually delivers
// into: the app's default bundle for the app's own store, and the named
// environment's bundle for an environment store. Reading the default bundle
// for an environment store would report every name missing on every walk —
// the short-circuit would never hold and every re-run would mint a key.
func EnvDeliveredIn(ctx context.Context, s release.Store, acct, app, env, prefix string, wantEndpoint bool) (missing []string, err error) {
	var m *appconfig.Manifest
	if env == "" {
		m, err = appconfig.Get(ctx, s, acct, app)
	} else {
		m, err = appconfig.LoadEnvironment(ctx, s, acct, app, env)
	}
	if err != nil {
		return nil, err
	}
	bucket, region, endpoint, accessKey, secretKey := EnvNames(prefix)
	type wantVar struct{ name, kind string }
	want := []wantVar{{bucket, "env"}, {region, "env"}}
	if wantEndpoint {
		want = append(want, wantVar{endpoint, "env"})
	}
	want = append(want, wantVar{accessKey, "secret"}, wantVar{secretKey, "secret"})
	for _, w := range want {
		if v, ok := m.Vars[w.name]; !ok || v.Kind != w.kind {
			missing = append(missing, w.name)
		}
	}
	return missing, nil
}

// bucketNameRe is the conservative intersection of S3 bucket naming rules:
// 3-63 characters of lowercase letters, digits, dots and hyphens, starting
// and ending alphanumeric. It deliberately refuses some names AWS would take
// (uppercase and underscores are legal in some legacy us-east-1 buckets)
// rather than admit any name that is not safely embeddable.
var bucketNameRe = regexp.MustCompile(`^[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$`)

// ValidBucketName gates every bucket name this package acts on — the minted
// one and, more importantly, an operator-supplied adopt name, which is
// interpolated straight into the IAM policy ARN the credential is scoped by.
// An unchecked name there is a scope the operator did not mean to grant, so
// the rail sits in front of the walk and in front of the console door (Task
// 9 reuses this function, so the API's refusal and the walk's cannot drift).
func ValidBucketName(name string) bool { return bucketNameRe.MatchString(name) }

// Grant provisions the declared store and delivers its credentials, in spec
// §5 order. It is idempotent: a healthy provisioned store makes a re-run a
// no-op (no fresh key, no rewrite), and an unhealthy one — a crash anywhere
// in the sequence — is repaired by walking it again. adoptBucket names an
// existing bucket to adopt instead of minting one; it is honoured only on the
// FIRST grant, since the recorded bucket is what the app's data is in.
//
// source is GrantSourceOperator or GrantSourceAccountApproval — why this grant
// is authorised — and lands in the grant block beside actor. It is recorded
// only on the FIRST grant, with the block itself: a re-run repairs an estate
// whose authority was already established and must not restate it, exactly as
// it must not re-choose the bucket.
func claimRawGrant(ctx context.Context, s release.Store, p Provisioner, d Deployment,
	sealPub *ecdh.PublicKey, acct, app, actor, source, adoptBucket string, now time.Time) (*Record, error) {
	return GrantIn(ctx, s, p, d, sealPub, acct, app, "", actor, source, adoptBucket, now)
}

// GrantIn is Grant for one store of an app: the app's own when env is "",
// else the store belonging to that named config environment, whose bucket,
// IAM principal and delivered bundle are all its own. Every property Grant's
// doc comment states holds here unchanged — the environment is another
// coordinate of WHICH estate, never a change to how the walk behaves.
//
// seed is the seed id this grant is for (GrantInFor), "" for every other
// grant. Any grant of a record whose declaration carries another seed's
// mark (Declared.Seed), a first grant or a re-run, clears the mark in the
// walk's first write (with the grant block, on a first grant): the store
// is then held by this grant, not the seed, and the seed's finish must
// never tear it down (seeded-instances spec §4 Storage). An aborted
// seed's bucket can leak that way, which is the safe direction. A mark
// therefore survives only while its own seed is the only one granting.
// A record with no mark is written exactly as before.
func claimRawGrantIn(ctx context.Context, s release.Store, p Provisioner, d Deployment,
	sealPub *ecdh.PublicKey, acct, app, env, actor, source, adoptBucket, seed string, now time.Time) (*Record, error) {
	// 1. resolve + write the grant block
	rec, etag, err := GetIn(ctx, s, acct, app, env)
	if err != nil {
		return nil, err
	}
	if rec == nil || rec.Declared == nil {
		return nil, ErrNotDeclared
	}
	// The key this walk started from, for the step-6 retry's guard below. ""
	// is the ordinary case: a first grant, or a re-run of one that died before
	// it recorded anything.
	startKeyID := ""
	if rec.Provisioned != nil {
		startKeyID = rec.Provisioned.KeyID
	}
	bucket := adoptBucket
	if rec.Grant != nil && rec.Grant.Bucket != "" {
		bucket = rec.Grant.Bucket // a re-run never re-chooses the bucket
	}
	if bucket == "" {
		bucket = MintedBucketIn(d, acct, app, env)
	}
	// Before anything is created or recorded: the name is about to become the
	// IAM policy's resource ARN, so it is checked whether it was minted from a
	// deployment prefix or handed over by an operator adopting a bucket.
	if !ValidBucketName(bucket) {
		return nil, fmt.Errorf("objstore: invalid bucket name %q", bucket)
	}
	// The principal's name is checked in the same breath, and for the same
	// reason: an environment store's name carries a third component, so a
	// long enough (account, app, environment) is unnameable to IAM. Learning
	// that at CreateUser would mean a grant block already written over an
	// estate that was never created — the shape Teardown has to treat as
	// live. The sentence names the parts, since the member can shorten one.
	user := UserNameIn(acct, app, env)
	if !ValidUserName(user) {
		return nil, fmt.Errorf("objstore: IAM user name %q is %d characters, over AWS's %d — shorten the account, app or environment name",
			user, len(user), MaxIAMUserName)
	}
	// Both names must be this store's alone (#218), and that is settled here,
	// before the grant block names them and before EnsureUser, EnsureBucket or
	// the reap below act on them: either of those, run against a name another
	// store holds, hands this caller that store's estate and kills its key.
	// Every run re-checks, which is also how a store granted before claims
	// existed gets its claims.
	if err := claimStoreNames(ctx, s, acct, app, env, bucket, user, rec.Grant == nil, now); err != nil {
		return nil, err
	}
	// A grant for any seed but the one that marked the record clears the
	// mark, first grant or re-run, in this same write: whoever grants the
	// store now holds it, and the seed's finish must not tear it down.
	clearMark := rec.Declared.Seed != "" && rec.Declared.Seed != seed
	if clearMark {
		rec.Declared.Seed = ""
	}
	if rec.Grant == nil || clearMark {
		if rec.Grant == nil {
			rec.Grant = &GrantBlock{Backend: d.Backend, Bucket: bucket, Source: source, At: now, By: actor}
		}
		if etag, err = PutIn(ctx, s, acct, app, env, rec, etag); err != nil {
			return nil, err
		}
	}
	// 2. the principal and its policy. EnsureUser runs UNCONDITIONALLY, before
	// the idempotence short-circuit below — the same fix internal/mail's walk
	// made as "C2", for the same reason.
	//
	// EnsureUser is the ONLY place this user's policy is written
	// (PutUserPolicy overwrites wholesale), and the deny in it is recomputed
	// from the CURRENT record on every call. Behind the short-circuit, turning
	// enforcement on for an already-provisioned store — which is every store
	// it would ever be turned on for — changed the record and nothing else,
	// forever. There is no other convergence path: Rotate never writes a
	// policy either.
	//
	// It needs only the bucket ARN string, resolved above, not an existing
	// bucket, so EnsureBucket stays behind the short-circuit. EnsureUser is
	// cheap to re-run (idempotent CreateUser, one PutUserPolicy); it is the
	// reap-and-mint below that spends IAM's two-key ceiling, and THAT is what
	// the short-circuit still has to gate.
	var deny *DenyTag
	if rec.Scanning != nil {
		deny = rec.Scanning.Enforce
	}
	arn, err := p.EnsureUser(ctx, user, UserPath, "arn:aws:s3:::"+bucket, deny)
	if err != nil {
		return nil, fmt.Errorf("objstore: ensure user %s: %w", user, err)
	}
	// Idempotence short-circuit: "healthy" means the record says provisioned
	// with THIS deployment's endpoint AND every delivered var is present, of
	// the right kind, in config.json. Anything less and the walk runs, because
	// a missing var means the app cannot reach its store — and an endpoint the
	// deployment has since changed or dropped means it is reaching for the
	// wrong one, which only a re-walk (which rewrites the set, and unsets
	// _ENDPOINT when there is no longer an endpoint) can put right.
	if rec.Provisioned != nil && rec.Provisioned.Endpoint == d.Endpoint {
		missing, err := EnvDeliveredIn(ctx, s, acct, app, env, rec.Declared.EnvPrefix, d.Endpoint != "")
		if err != nil {
			return nil, err
		}
		if len(missing) == 0 {
			return rec, nil
		}
	}
	// 3. the bucket
	region, err := p.EnsureBucket(ctx, bucket)
	if err != nil {
		return nil, fmt.Errorf("objstore: ensure bucket %s: %w", bucket, err)
	}
	// 4. key. Reap first: reaching here means delivery is missing, so any
	// existing CURRENT key is either undelivered (its secret died with a
	// failed config write — unrecoverable) or stranded by a crash between mint
	// and record, which no record can name. Only list-and-delete finds both; a
	// pending-finish key from an old rotation is kept for its explicit finish.
	keep := []string(nil)
	if rec.Provisioned != nil && rec.Provisioned.PrevKeyID != "" {
		keep = append(keep, rec.Provisioned.PrevKeyID)
	}
	if err := p.ReapKeys(ctx, user, keep); err != nil {
		return nil, fmt.Errorf("objstore: reap keys for %s: %w", user, err)
	}
	id, secret, err := p.MintKey(ctx, user)
	if err != nil {
		return nil, fmt.Errorf("objstore: mint key for %s: %w", user, err)
	}
	// 5. deliver — the secret's ONLY landing place (§7)
	if err := writeEnv(ctx, s, sealPub, acct, app, env, rec.Declared.EnvPrefix, bucket, region, d.Endpoint, id, secret); err != nil {
		return nil, err
	}
	// 6. record success — after the secret is recorded, never before (§5).
	// PrevKeyID carries over untouched: the reap above already removed the
	// only stale current key, and a pending-finish key from an old rotation
	// still deserves its explicit finish.
	prev := ""
	if rec.Provisioned != nil {
		prev = rec.Provisioned.PrevKeyID
	}
	rec.Provisioned = &Provisioned{Bucket: bucket, Region: region, Endpoint: d.Endpoint,
		IAMUser: user, UserARN: arn, KeyID: id, PrevKeyID: prev, At: now}
	if _, err := PutIn(ctx, s, acct, app, env, rec, etag); err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			return nil, err
		}
		// Same window, same rule, same retry as Rotate's step five — see its
		// doc comment for the full argument. A lost CAS here is the one shape
		// of the gap that is NOT survivable: the secret has already been
		// delivered (step 5), so the app is running on `id`, while the record
		// would keep naming whatever it named before. The next rotation reaps
		// every key the record does not name, and would delete the key in use.
		//
		// Before self-serve object storage (#381) this needed two operators
		// running `grant` at once. Now a member's own `carlos store create`
		// drives this walk, and the console's declare door writes this very
		// object, so the window is ordinary rather than exotic.
		//
		// Re-read and try once more, carrying THEIR blocks and only our
		// Provisioned one — but only if the record still names the key this
		// walk started from. If it names another, a concurrent walk minted and
		// DELIVERED a credential of its own; overwriting it would strand
		// exactly what this retry protects against, so that conflict is the
		// caller's to see.
		fresh, freshEtag, gerr := GetIn(ctx, s, acct, app, env)
		if gerr != nil {
			return nil, err
		}
		freshKeyID := ""
		if fresh != nil && fresh.Provisioned != nil {
			freshKeyID = fresh.Provisioned.KeyID
		}
		if fresh == nil || freshKeyID != startKeyID {
			return nil, err
		}
		fresh.Provisioned = rec.Provisioned
		if _, perr := PutIn(ctx, s, acct, app, env, fresh, freshEtag); perr != nil {
			return nil, perr
		}
		return fresh, nil
	}
	return rec, nil
}

// Rotate mints a fresh key, delivers it, and leaves the OLD key active (§8).
// That overlap is the whole point: adoption swaps a symlink, it does not
// restart processes, so at the instant the new credentials land there are
// still hosts, in-flight requests and hibernating tenants holding the old
// pair. Killing it here would break them; RotateFinish is the member's
// explicit "nothing is using it any more".
//
// The overlap runs into AWS's hard ceiling of two access keys per user, so
// the estate is a two-state machine — one live key, or one live plus one
// pending — and every step below exists to keep it there:
//
// Step one deletes a pending key from an UNFINISHED earlier rotation.
// Rotating twice without finishing is ordinary member behaviour, and the
// second rotation is itself the statement that the oldest key is done.
//
// Step two reaps every key the record does not name. A crash between mint and
// reseal (step four failing after step three) leaves a key no record can name;
// without the reap the next mint wedges at the ceiling forever, and re-running
// the walk — the platform's whole repair story — would stop working.
//
// Steps three and four mint, then reseal into the app's config. Coordinates
// come from the record: a rotation changes credentials and nothing else, so
// the bucket an app's data is in is never re-derived from a deployment that
// may have moved on.
//
// Step five records, last, as in Grant. The invariant it buys is one-way: the
// record never names a key whose secret was never delivered. A record that
// claimed the new key after a FAILED delivery would point every reader at a
// key whose secret died with the write, and the next run would reap the ONE
// key still working. The reverse gap is survivable and deliberate — after a
// successful delivery whose Put then fails, the record names the old key
// while the app already holds the new one, and both are live, so the app
// keeps working and a re-run converges.
//
// The one shape of that gap which is NOT survivable is a lost CAS: the record
// would keep naming the old key permanently, and the next rotation's reap
// would delete the delivered key the app is actually running on. So the final
// Put retries once against a re-read record — but only when that re-read still
// names the key this walk started from, i.e. no other rotation landed in
// between. A blind retry would clobber a concurrent rotate and strand ITS
// delivered key, which is the exact failure the retry exists to prevent.
func claimRawRotate(ctx context.Context, s release.Store, p Provisioner, sealPub *ecdh.PublicKey,
	acct, app string, now time.Time) (*Record, error) {
	return RotateIn(ctx, s, p, sealPub, acct, app, "", now)
}

// RotateIn is Rotate for one store of an app. Coordinates come from that
// store's own record, so a rotation of an environment store never reaches for
// the app store's bucket, principal or bundle.
func claimRawRotateIn(ctx context.Context, s release.Store, p Provisioner, sealPub *ecdh.PublicKey,
	acct, app, env string, now time.Time) (*Record, error) {
	rec, etag, err := GetIn(ctx, s, acct, app, env)
	if err != nil {
		return nil, err
	}
	if rec == nil || rec.Provisioned == nil {
		return nil, ErrNotProvisioned
	}
	if rec.Declared == nil {
		return nil, ErrNotDeclared // no prefix to deliver into
	}
	pr := rec.Provisioned
	startKeyID := pr.KeyID // the key the app is holding, and the retry guard
	// 1. finish the pending rotation late. The reap below would collect this
	// key too, but doing it by name first keeps the intent legible and means
	// a backend that refuses the delete stops the walk here, before a mint.
	if pr.PrevKeyID != "" {
		if err := p.DeleteKey(ctx, pr.IAMUser, pr.PrevKeyID); err != nil {
			return nil, fmt.Errorf("objstore: delete pending key %s for %s: %w", pr.PrevKeyID, pr.IAMUser, err)
		}
	}
	// 2. collect anything no record names — only list-and-delete finds it.
	if err := p.ReapKeys(ctx, pr.IAMUser, []string{pr.KeyID}); err != nil {
		return nil, fmt.Errorf("objstore: reap keys for %s: %w", pr.IAMUser, err)
	}
	// 3. mint into the now-guaranteed free slot
	id, secret, err := p.MintKey(ctx, pr.IAMUser)
	if err != nil {
		return nil, fmt.Errorf("objstore: mint key for %s: %w", pr.IAMUser, err)
	}
	// 4. deliver — same names, same single CAS write, same validation of the
	// credential halves before sealing (writeEnv owns that rail).
	if err := writeEnv(ctx, s, sealPub, acct, app, env, rec.Declared.EnvPrefix,
		pr.Bucket, pr.Region, pr.Endpoint, id, secret); err != nil {
		return nil, err
	}
	// 5. record: the key just retired from live becomes the pending one.
	pr.PrevKeyID, pr.KeyID, pr.At = pr.KeyID, id, now
	if _, err := PutIn(ctx, s, acct, app, env, rec, etag); err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			return nil, err
		}
		// Someone else wrote this record between our Get and our Put — the
		// console's declare/grant doors write the same object. Re-read and
		// try once more, carrying THEIR blocks and only our Provisioned one,
		// but only if the record still names the key we started from. If it
		// names another key, a concurrent rotation delivered a credential of
		// its own; overwriting it would strand exactly what we are protecting
		// against, so the conflict is the caller's to see.
		fresh, freshEtag, gerr := GetIn(ctx, s, acct, app, env)
		if gerr != nil {
			return nil, err
		}
		if fresh == nil || fresh.Provisioned == nil || fresh.Provisioned.KeyID != startKeyID {
			return nil, err
		}
		fresh.Provisioned = pr
		if _, perr := PutIn(ctx, s, acct, app, env, fresh, freshEtag); perr != nil {
			return nil, perr
		}
		return fresh, nil
	}
	return rec, nil
}

// RotateFinish closes the two-key window: the pending key is deleted and the
// record stops naming it. Nothing is delivered — the app has been reading the
// current key since the rotation, so there is no config change to make, and
// making one would restart nothing and risk everything.
//
// This is the only walk that destroys a credential an app might still be
// using, which is why it is a separate verb a member asks for rather than a
// timer: only the member knows every host, worker and hibernating tenant has
// picked the new pair up.
func claimRawRotateFinish(ctx context.Context, s release.Store, p Provisioner, acct, app string) (*Record, error) {
	return RotateFinishIn(ctx, s, p, acct, app, "")
}

// RotateFinishIn is RotateFinish for one store of an app — the pending key it
// deletes is that store's own, named by that store's own record.
func claimRawRotateFinishIn(ctx context.Context, s release.Store, p Provisioner, acct, app, env string) (*Record, error) {
	rec, etag, err := GetIn(ctx, s, acct, app, env)
	if err != nil {
		return nil, err
	}
	if rec == nil || rec.Provisioned == nil {
		return nil, ErrNotProvisioned
	}
	pr := rec.Provisioned
	if pr.PrevKeyID == "" {
		return nil, ErrNoPendingFinish
	}
	if err := p.DeleteKey(ctx, pr.IAMUser, pr.PrevKeyID); err != nil {
		return nil, fmt.Errorf("objstore: delete pending key %s for %s: %w", pr.PrevKeyID, pr.IAMUser, err)
	}
	// Recorded after the delete: a record that forgot the key before the
	// backend did would strand a live credential nothing names, which is the
	// one thing the reap exists to clean up and the estate exists to avoid.
	pr.PrevKeyID = ""
	if _, err := PutIn(ctx, s, acct, app, env, rec, etag); err != nil {
		return nil, err
	}
	return rec, nil
}

// Teardown dismantles a store: bucket contents, bucket, IAM principal (keys
// and inline policy with it), the delivered config vars, and finally the
// record. It is the shared engine under the console's "Delete now" purge and
// its store-destroy door (§9), and it is idempotent at every step, so a
// half-failed run is repaired by running it again.
//
// The ordering rule §9 states once lives here: the record is READ FIRST and
// DELETED LAST. Everything this walk destroys is named only by the record —
// the bucket an app's data is in, the principal holding its keys — so a
// record deleted before the backend is a bucket nobody can find again, still
// billing, with a live credential attached. Any failure therefore returns with
// the record intact and provisioned; the caller (a purge) aborts on that error
// rather than sweeping the prefix over an estate that is still standing.
//
// An absent record is nil, not an error: "no store" is the state teardown is
// trying to reach, and both callers race other operators to it.
//
// What decides the path is the GRANT block, not the provisioned one. A record
// carrying a grant but no `provisioned` is not "nothing to tear down": Grant
// creates the bucket, the user and a key, and delivers the credentials, all
// BEFORE its step-6 Put — so a lost CAS (or any crash) at that last write
// leaves exactly this shape over a live, billing, credentialed estate. Reading
// it as trivially deletable would erase the only record naming that estate,
// which is precisely the failure §9's ordering rule exists to prevent. Only a
// record with NO grant at all — declared, never acted on — dies trivially.
func claimRawTeardown(ctx context.Context, s release.Store, p Provisioner, sc Scanner, acct, app string) error {
	return TeardownIn(ctx, s, p, sc, acct, app, "")
}

// TeardownIn is Teardown for one store of an app. Tearing an environment
// store down never touches the app's own store, or any other environment's:
// the bucket, the principal and the bundle the vars come out of are all named
// by this record alone.
func claimRawTeardownIn(ctx context.Context, s release.Store, p Provisioner, sc Scanner, acct, app, env string) error {
	rec, _, err := GetIn(ctx, s, acct, app, env)
	if err != nil {
		return err
	}
	if rec == nil {
		return nil
	}
	// What this walk believes it is destroying — the guard on the final delete.
	startKeyID, startBucket := "", ""
	if rec.Provisioned != nil {
		startKeyID = rec.Provisioned.KeyID
	}
	if rec.Grant != nil {
		startBucket = rec.Grant.Bucket
	}
	if rec.Grant == nil && rec.Provisioned == nil {
		// A grant that claimed its names and died before writing its block
		// leaves both claims behind. The user's name is computable; the
		// bucket it chose is recorded nowhere but in the claim itself, so
		// the bucket claims are searched for this store's own.
		bucket, err := claimedBucket(ctx, s, acct, app, env)
		if err != nil {
			return err
		}
		return deleteRecordIfUnchanged(ctx, s, acct, app, env, startKeyID, startBucket, bucket, UserNameIn(acct, app, env))
	}
	// Everything below spends backend authority, and a nil Provisioner is a
	// deployment that has none — an ordinary shape (console Config.
	// StoreProvisioner documents it), so this is a refusal, not an assertion.
	// The console's destroy door already refuses an estate it cannot tear down,
	// but it decides from a record it read a round trip earlier: a grant landing
	// in between turns a declaration-only teardown into an estate teardown, and
	// calling through a nil backend would panic the request rather than refuse
	// it. The record is left exactly as found, which is what a refusal owes.
	if p == nil {
		return fmt.Errorf("objstore: %s/%s has backend estate and this deployment has no storage backend configured", acct, app)
	}
	// Coordinates: the provisioned block when there is one, since the bucket
	// an app's data is in is whatever was recorded at provision time. Falling
	// back to the grant covers the crashed-grant shape above — and the user
	// name there is not a guess: EnsureUser is called with exactly
	// UserName(acct, app), so a principal created by a grant that never
	// recorded itself is still findable by name.
	bucket, user := "", UserNameIn(acct, app, env)
	switch {
	case rec.Provisioned != nil:
		bucket, user = rec.Provisioned.Bucket, rec.Provisioned.IAMUser
	default:
		bucket = rec.Grant.Bucket
	}
	// A bucket or user another store holds is left standing (#218, #191). Two
	// shapes reach here: a pair whose joined names collided before claims
	// existed, both granted; and a teardown re-run after its bucket was
	// deleted, its name given to another store in between. Either way this
	// record names an estate that is no longer — or never was — only its own,
	// and deleting it would destroy the other store's data or credentials.
	// The rest of the walk goes on: the vars and the record are this store's.
	bucketShared, userShared, err := sharedNames(ctx, s, acct, app, env, bucket, user)
	if err != nil {
		return fmt.Errorf("objstore: check names for %s/%s: %w", acct, app, err)
	}
	estateBucket := bucket
	if bucketShared {
		bucket = ""
	}
	// The scanner goes BEFORE the bucket, while the record that names it still
	// exists. A Malware Protection plan (and its managed EventBridge rule)
	// outlives the bucket it protects, counting against a non-adjustable
	// 25-buckets-per-region quota, and once the record is gone nothing names
	// it any more — the read-first, delete-last failure this walk's ordering
	// exists to prevent.
	//
	// A nil scanner over an enabled record REFUSES, for the same reason a nil
	// provisioner does above: a teardown that cannot reach the engine must not
	// go on to delete the only record of what it should have disabled.
	if rec.Scanning != nil && rec.Scanning.Enabled != nil && bucket != "" {
		if sc == nil {
			return fmt.Errorf("objstore: %s/%s has scanning enabled and this deployment has no scanner configured (CARLOS_STORE_SCANNER)", acct, app)
		}
		if err := sc.EnsureDisabled(ctx, bucket); err != nil {
			return fmt.Errorf("objstore: disable scanning on %s: %w", bucket, err)
		}
	}
	// A grant block with no bucket cannot come from the walk (Grant resolves
	// and validates a name before writing the block), so there is nothing
	// nameable to destroy — but the deterministic user might exist, and the
	// delivered vars might too, so only the bucket steps are skipped.
	if bucket != "" {
		// Empty before delete: a real backend refuses to delete a bucket with
		// objects in it, and EmptyBucket is paginated and resumable for exactly
		// the bucket too big to clear in one call.
		if err := p.EmptyBucket(ctx, bucket); err != nil {
			return fmt.Errorf("objstore: empty bucket %s: %w", bucket, err)
		}
		if err := p.DeleteBucket(ctx, bucket); err != nil {
			return fmt.Errorf("objstore: delete bucket %s: %w", bucket, err)
		}
	}
	// DeleteUser reaps the keys and the inline policy, so the credential dies
	// with the estate it was scoped to — no orphan principal keeps a key that
	// names a bucket that no longer exists.
	//
	// A user another store holds stays, but the keys THIS store minted on it
	// do not: the tenant has seen them, and the shared user's policy covers
	// the other store's bucket. Deleting a key that is already gone is
	// success, so a re-run converges.
	if userShared {
		if rec.Provisioned != nil {
			for _, id := range []string{rec.Provisioned.KeyID, rec.Provisioned.PrevKeyID} {
				if id == "" {
					continue
				}
				if err := p.DeleteKey(ctx, user, id); err != nil {
					return fmt.Errorf("objstore: delete key %s for %s: %w", id, user, err)
				}
			}
		}
	} else if err := p.DeleteUser(ctx, user); err != nil {
		return fmt.Errorf("objstore: delete user %s: %w", user, err)
	}
	// The delivered set goes even though the credentials behind it are already
	// dead: leaving them would hand the app's next render a bucket name and a
	// key that authenticate against nothing, which is harder to debug than an
	// unset variable. The prefix is the member's, read back off the record;
	// Declared should never be nil on a provisioned store (Grant refuses
	// without it), but this is bucket data an operator can hand-edit, so the
	// fallback is the default set rather than a panic or a stranded estate.
	prefix := DefaultEnvPrefix
	if rec.Declared != nil {
		prefix = rec.Declared.EnvPrefix
	}
	bucketName, regionName, endpointName, accessKeyName, secretKeyName := EnvNames(prefix)
	unsets := []string{bucketName, regionName, endpointName, accessKeyName, secretKeyName}
	// _ENDPOINT is unset unconditionally: whether it was delivered depends on
	// what the deployment looked like at grant time, and unsetting a var that
	// is not there is a no-op.
	//
	// The guard is existence, not the unset itself: appconfig REFUSES an
	// unset-only apply against a config object that does not exist (creating
	// one would mint an empty manifest, which means "configured, then fully
	// unset" to Sync, which converges every route's env file to empty as
	// root). An app that was never configured has nothing of ours to remove
	// and must not have that refusal strand its backend teardown — so if none
	// of the names are present, this step is already converged.
	if err := unsetDelivered(ctx, s, acct, app, env, unsets); err != nil {
		return err
	}
	// The scan sidecar, before the record. It is the only other object under
	// this app's prefix that this package writes, and an orphan left behind
	// makes the app NAME unusable: the claim gate refuses a name whose prefix
	// still holds any key, and the box-side auto-purge never reaches it
	// because the tombstone goes last. Delete is idempotent on an absent key,
	// so a store that was never swept costs one no-op.
	//
	// Only the app's own store has one. Scanning is (acct, app)-scoped
	// throughout — Sweep's cursor, ScanResultKey, the queue — so an
	// environment store has no sidecar to orphan, and deleting ScanKey here
	// would destroy the APP store's sweep cursor while tearing an
	// environment's estate down (2026-09-19 spec §9).
	if env == "" {
		if err := s.Delete(ctx, ScanKey(acct, app)); err != nil {
			return fmt.Errorf("objstore: delete scan sidecar for %s/%s: %w", acct, app, err)
		}
	}
	// Last (§9). Everything it named is gone.
	return deleteRecordIfUnchanged(ctx, s, acct, app, env, startKeyID, startBucket, estateBucket, user)
}

// unsetDelivered removes the delivered set from the bundle a store delivers
// INTO — the app's default bundle, or the named environment's.
//
// The presence check is why this is one helper rather than two call sites:
// appconfig REFUSES an unset-only apply against a bundle that does not exist,
// because creating one would mint an empty manifest, which means "configured,
// then fully unset" to Sync — which converges every route's env file to
// empty, as root. A store whose bundle was never configured has nothing of
// ours to remove and must not have that refusal strand its backend teardown.
// An environment store is the likelier shape to meet it: its bundle exists
// only because delivery created it, so a crashed grant can leave an estate
// with no bundle at all.
func unsetDelivered(ctx context.Context, s release.Store, acct, app, env string, unsets []string) error {
	var (
		m   *appconfig.Manifest
		err error
	)
	if env == "" {
		m, err = appconfig.Get(ctx, s, acct, app)
	} else {
		m, err = appconfig.LoadEnvironment(ctx, s, acct, app, env)
	}
	if err != nil {
		return fmt.Errorf("objstore: read config for %s/%s: %w", acct, app, err)
	}
	present := false
	for _, name := range unsets {
		if _, ok := m.Vars[name]; ok {
			present = true
			break
		}
	}
	if !present {
		return nil
	}
	if env == "" {
		if _, err := appconfig.Apply(ctx, s, acct, app, nil, unsets); err != nil {
			return fmt.Errorf("objstore: remove store config for %s/%s: %w", acct, app, err)
		}
		return nil
	}
	if _, err := appconfig.ApplyEnvironment(ctx, s, acct, app, env, nil, unsets); err != nil {
		return fmt.Errorf("objstore: remove store config for %s/%s environment %s: %w", acct, app, env, err)
	}
	return nil
}

// deleteRecordIfUnchanged is the final step of Teardown: delete the record,
// but only if it still describes the estate the walk just destroyed.
//
// Store.Delete has no compare-and-swap, so this is the same shape Rotate's
// retry uses — re-read, and act only on a record that still names what we
// started from. The race it closes is a Grant landing mid-teardown: the
// console's grant door writes this same object, and a grant that completes
// after our unset has minted a NEW bucket, user and key and recorded them. A
// blind delete there erases the record of a live estate — the §9 failure,
// arrived at from the other end.
//
// The refusal wraps release.ErrCASConflict so a caller can tell it apart from
// a backend failure: the estate the walk was aimed at IS gone, and the right
// response is to re-run against the record as it now stands, not to retry the
// same delete. A record that has vanished under us is success, not conflict —
// another teardown got there first, which is the state we wanted.
//
// This narrows the window; it does not close it. A grant landing between the
// re-read and the Delete still loses its record, and only a CAS delete in the
// Store interface could fix that. The console door serialises these verbs per
// app (internal/console/storelock.go), so the remaining window is a fraction
// of one round trip.
//
// That serialisation was asserted here long before it existed — no lock of any
// kind guarded these walks until #381 — so if this reference ever stops
// resolving, the sentence above is a claim about nothing and this window is a
// whole walk wide again.
//
// The store's name claims go in the same breath, after the re-read and before
// the delete: given back any earlier, a grant landing mid-teardown would be
// left holding names nothing claims; any later, a failed delete of the record
// would leave its names claimed by a store that no longer exists.
func deleteRecordIfUnchanged(ctx context.Context, s release.Store, acct, app, env, startKeyID, startBucket, claimBucket, claimUser string) error {
	fresh, _, err := GetIn(ctx, s, acct, app, env)
	if err != nil {
		return fmt.Errorf("objstore: re-read store record for %s/%s: %w", acct, app, err)
	}
	if fresh == nil {
		return nil
	}
	keyID, bucket := "", ""
	if fresh.Provisioned != nil {
		keyID = fresh.Provisioned.KeyID
	}
	if fresh.Grant != nil {
		bucket = fresh.Grant.Bucket
	}
	if keyID != startKeyID || bucket != startBucket {
		return fmt.Errorf("objstore: store record for %s/%s changed during teardown — re-run: %w", acct, app, release.ErrCASConflict)
	}
	if err := releaseStoreNames(ctx, s, acct, app, env, claimBucket, claimUser); err != nil {
		return err
	}
	if err := s.Delete(ctx, KeyIn(acct, app, env)); err != nil {
		return fmt.Errorf("objstore: delete store record for %s/%s: %w", acct, app, err)
	}
	return nil
}

// writeEnv delivers the whole coordinate set in ONE appconfig.Apply — one CAS
// write, so an app never reads a bucket name paired with the previous key.
// Both key halves are sealed for the deployment's secrets key; the plaintext
// secret exists nowhere else, in no other object and no log line.
//
// A deployment with no endpoint UNSETS the endpoint variable rather than
// leaving whatever is there: dropping an endpoint override is a real change
// (the app must go back to the backend's own address), and a stale
// _ENDPOINT would keep pointing every host at an address the deployment no
// longer uses. The set half is always non-empty, so this never trips
// applyAt's refusal to create a bundle on an unset-only call.
func writeEnv(ctx context.Context, s release.Store, sealPub *ecdh.PublicKey,
	acct, app, env, prefix, bucket, region, endpoint, keyID, secret string) error {
	bucketName, regionName, endpointName, accessKeyName, secretKeyName := EnvNames(prefix)
	sets := map[string]appconfig.Var{
		bucketName: {Kind: "env", Value: bucket},
		regionName: {Kind: "env", Value: region},
	}
	var unsets []string
	if endpoint != "" {
		sets[endpointName] = appconfig.Var{Kind: "env", Value: endpoint}
	} else {
		unsets = append(unsets, endpointName)
	}
	// Validate the credential halves BEFORE sealing. appconfig checks "env"
	// values at set time for exactly this reason, but a sealed secret reaches
	// it as an opaque envelope, so an unrenderable byte would sail through
	// here and stall the app's WHOLE config render on every host, hours later
	// and out of sight. Refusing now costs one wasted key, which the next
	// walk's reap collects. The errors name the variable and NEVER the value:
	// one half is a live secret, and these strings reach logs and HTTP bodies.
	for _, half := range []struct{ name, plain string }{
		{accessKeyName, keyID}, {secretKeyName, secret},
	} {
		if err := appconfig.ValidVarValue(half.plain); err != nil {
			return fmt.Errorf("objstore: minted credential for %s is not EnvironmentFile-safe: %w", half.name, err)
		}
		env, err := appconfig.Seal(sealPub, appconfig.Binding{Account: acct, App: app}, []byte(half.plain))
		if err != nil {
			return fmt.Errorf("objstore: seal %s: %w", half.name, err)
		}
		sets[half.name] = appconfig.Var{Kind: "secret", Envelope: env}
	}
	// The bundle the store belongs to, which is the bundle the instances
	// bound to it already read. ApplyEnvironment is the same one-write
	// contract as Apply, touches the same per-app generation object, and
	// applies the same refusal to create a bundle on an unset-only call —
	// the set half here is always non-empty, so that refusal is never met.
	if env == "" {
		if _, err := appconfig.Apply(ctx, s, acct, app, sets, unsets); err != nil {
			return fmt.Errorf("objstore: deliver store config for %s/%s: %w", acct, app, err)
		}
		return nil
	}
	if _, err := appconfig.ApplyEnvironment(ctx, s, acct, app, env, sets, unsets); err != nil {
		return fmt.Errorf("objstore: deliver store config for %s/%s environment %s: %w", acct, app, env, err)
	}
	return nil
}
func Grant(ctx context.Context, s release.Store, p Provisioner, d Deployment, sealPub *ecdh.PublicKey, acct, app, actor, source, adoptBucket string, now time.Time) (claimResult0 *Record, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.Grant", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawGrant(claimContext, s, p, d, sealPub, acct, app, actor, source, adoptBucket, now)
		return claimResult1
	})
	return
}
func GrantIn(ctx context.Context, s release.Store, p Provisioner, d Deployment, sealPub *ecdh.PublicKey, acct, app, env, actor, source, adoptBucket string, now time.Time) (claimResult0 *Record, claimResult1 error) {
	return GrantInFor(ctx, s, p, d, sealPub, acct, app, env, actor, source, adoptBucket, "", now)
}

// GrantInFor is GrantIn for a seeded create's store: seed is the seed id
// whose mark the grant keeps. Every other grant is GrantIn, seed "".
func GrantInFor(ctx context.Context, s release.Store, p Provisioner, d Deployment, sealPub *ecdh.PublicKey, acct, app, env, actor, source, adoptBucket, seed string, now time.Time) (claimResult0 *Record, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.GrantIn", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawGrantIn(claimContext, s, p, d, sealPub, acct, app, env, actor, source, adoptBucket, seed, now)
		return claimResult1
	})
	return
}
func Rotate(ctx context.Context, s release.Store, p Provisioner, sealPub *ecdh.PublicKey, acct, app string, now time.Time) (claimResult0 *Record, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.Rotate", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawRotate(claimContext, s, p, sealPub, acct, app, now)
		return claimResult1
	})
	return
}
func RotateFinish(ctx context.Context, s release.Store, p Provisioner, acct, app string) (claimResult0 *Record, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.RotateFinish", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawRotateFinish(claimContext, s, p, acct, app)
		return claimResult1
	})
	return
}
func RotateFinishIn(ctx context.Context, s release.Store, p Provisioner, acct, app, env string) (claimResult0 *Record, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.RotateFinishIn", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawRotateFinishIn(claimContext, s, p, acct, app, env)
		return claimResult1
	})
	return
}
func RotateIn(ctx context.Context, s release.Store, p Provisioner, sealPub *ecdh.PublicKey, acct, app, env string, now time.Time) (claimResult0 *Record, claimResult1 error) {
	claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.RotateIn", func(claimContext context.Context) error {
		claimResult0, claimResult1 = claimRawRotateIn(claimContext, s, p, sealPub, acct, app, env, now)
		return claimResult1
	})
	return
}
func Teardown(ctx context.Context, s release.Store, p Provisioner, sc Scanner, acct, app string) (claimResult0 error) {
	claimResult0 = release.WithAppMutation(ctx, s, acct, app, "objstore.Teardown", func(claimContext context.Context) error {
		claimResult0 = claimRawTeardown(claimContext, s, p, sc, acct, app)
		return claimResult0
	})
	return
}
func TeardownIn(ctx context.Context, s release.Store, p Provisioner, sc Scanner, acct, app, env string) (claimResult0 error) {
	claimResult0 = release.WithAppMutation(ctx, s, acct, app, "objstore.TeardownIn", func(claimContext context.Context) error {
		claimResult0 = claimRawTeardownIn(claimContext, s, p, sc, acct, app, env)
		return claimResult0
	})
	return
}