package objstore

import (
	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"crypto/rand"
	"errors"
	"os"
	"strings"
	"testing"
	"time"

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

// testSealKeypair mints the secrets keypair the walk seals with and the test
// opens with — the console holds the public half, a host the private one.
func testSealKeypair(t *testing.T) (*ecdh.PrivateKey, *ecdh.PublicKey) {
	t.Helper()
	privB64, pubB64, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := appconfig.ParsePrivateKey(privB64, "test")
	if err != nil {
		t.Fatal(err)
	}
	pub, err := appconfig.ParsePublicKey(pubB64, "test")
	if err != nil {
		t.Fatal(err)
	}
	return priv, pub
}

// failingStore fails the FIRST Put to failKeyOnce, then behaves normally —
// the crash between mint and delivery the walk has to repair.
type failingStore struct {
	release.Store
	failKeyOnce string
}

func (f *failingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key != "" && key == f.failKeyOnce {
		f.failKeyOnce = ""
		return "", errors.New("injected: put failed")
	}
	return f.Store.Put(ctx, key, body, ifMatch)
}

func testDeployment() Deployment {
	return Deployment{Backend: "aws", Region: "eu-west-1", BucketPrefix: "carlos-store-tito-"}
}

// declare writes the member half of the record: a store is asked for.
func declare(t *testing.T, s release.Store, acct, app, prefix string) {
	t.Helper()
	rec := &Record{Declared: &Declared{EnvPrefix: prefix, At: time.Now().UTC(), By: "member@example.com"}}
	if _, err := Put(context.Background(), s, acct, app, rec, ""); err != nil {
		t.Fatal(err)
	}
}

func mustVars(t *testing.T, s release.Store, acct, app string) map[string]appconfig.Var {
	t.Helper()
	m, err := appconfig.Get(context.Background(), s, acct, app)
	if err != nil {
		t.Fatal(err)
	}
	return m.Vars
}

func TestGrantUndeclaredRefused(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); !errors.Is(err, ErrNotDeclared) {
		t.Fatalf("Grant on undeclared app: %v, want ErrNotDeclared", err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("refused grant touched the backend: %v %v", p.Buckets, p.Users)
	}
}

func TestGrantMintsAndDelivers(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	priv, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	now := time.Now().UTC()
	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op@example.com", GrantSourceOperator, "", now)
	if err != nil {
		t.Fatal(err)
	}

	const wantBucket = "carlos-store-tito-bab-titogo"
	const wantUser = "carlos-store-bab-titogo"
	if _, ok := p.Buckets[wantBucket]; !ok {
		t.Fatalf("bucket %q not created: %v", wantBucket, p.Buckets)
	}
	u, ok := p.Users[wantUser]
	if !ok {
		t.Fatalf("user %q not created: %v", wantUser, p.Users)
	}
	if rec.Provisioned == nil {
		t.Fatal("record has no provisioned block")
	}
	if rec.Provisioned.Bucket != wantBucket || rec.Provisioned.IAMUser != wantUser ||
		rec.Provisioned.Region != "eu-west-1" || rec.Provisioned.UserARN != u.ARN {
		t.Fatalf("provisioned = %+v", rec.Provisioned)
	}
	if rec.Grant == nil || rec.Grant.Backend != "aws" || rec.Grant.By != "op@example.com" {
		t.Fatalf("grant block = %+v", rec.Grant)
	}
	secret, ok := u.Keys[rec.Provisioned.KeyID]
	if !ok {
		t.Fatalf("recorded key %q is not in the fake: %v", rec.Provisioned.KeyID, u.Keys)
	}

	// the record is durable, not just the returned value
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored == nil || stored.Provisioned == nil ||
		stored.Provisioned.KeyID != rec.Provisioned.KeyID {
		t.Fatalf("stored record = %+v, %v", stored, err)
	}

	vars := mustVars(t, s, "bab", "titogo")
	if v := vars["TITO_S3_BUCKET"]; v.Kind != "env" || v.Value != wantBucket {
		t.Fatalf("TITO_S3_BUCKET = %+v", v)
	}
	if v := vars["TITO_S3_REGION"]; v.Kind != "env" || v.Value != "eu-west-1" {
		t.Fatalf("TITO_S3_REGION = %+v", v)
	}
	if v := vars["TITO_S3_ACCESS_KEY"]; v.Kind != "secret" || v.Envelope == "" {
		t.Fatalf("TITO_S3_ACCESS_KEY = %+v", v)
	}
	gotID, err := appconfig.Open(priv, appconfig.Binding{Account: "bab", App: "titogo"}, vars["TITO_S3_ACCESS_KEY"].Envelope)
	if err != nil || string(gotID) != rec.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, %v; want %q", gotID, err, rec.Provisioned.KeyID)
	}
	sk := vars["TITO_S3_SECRET_KEY"]
	if sk.Kind != "secret" || sk.Value != "" {
		t.Fatalf("TITO_S3_SECRET_KEY = %+v (secret must never be plaintext)", sk)
	}
	gotSecret, err := appconfig.Open(priv, appconfig.Binding{Account: "bab", App: "titogo"}, sk.Envelope)
	if err != nil || string(gotSecret) != secret {
		t.Fatalf("delivered secret = %q, %v; want %q", gotSecret, err, secret)
	}
	if _, ok := vars["TITO_S3_ENDPOINT"]; ok {
		t.Fatalf("endpoint delivered for an empty endpoint: %+v", vars)
	}
}

func TestGrantDeliversEndpointWhenSet(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-central-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	d := testDeployment()
	d.Endpoint = "https://s3.example.net"
	rec, err := Grant(ctx, s, p, d, pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if rec.Provisioned.Endpoint != d.Endpoint {
		t.Fatalf("provisioned endpoint = %q", rec.Provisioned.Endpoint)
	}
	vars := mustVars(t, s, "bab", "titogo")
	if v := vars["TITO_S3_ENDPOINT"]; v.Kind != "env" || v.Value != d.Endpoint {
		t.Fatalf("TITO_S3_ENDPOINT = %+v", v)
	}
}

func TestGrantAdoptsExistingBucket(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "titogo-uploads", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if rec.Provisioned.Bucket != "titogo-uploads" || rec.Grant.Bucket != "titogo-uploads" {
		t.Fatalf("adopt: grant=%+v provisioned=%+v", rec.Grant, rec.Provisioned)
	}
	if _, ok := p.Buckets["carlos-store-tito-bab-titogo"]; ok {
		t.Fatalf("minted bucket created despite adoption: %v", p.Buckets)
	}
	if v := mustVars(t, s, "bab", "titogo")["TITO_S3_BUCKET"]; v.Value != "titogo-uploads" {
		t.Fatalf("delivered bucket = %+v", v)
	}
}

func TestGrantResumesAfterEachStepFailure(t *testing.T) {
	for _, method := range []string{"EnsureBucket", "EnsureUser", "MintKey"} {
		t.Run(method, func(t *testing.T) {
			ctx := context.Background()
			s := release.NewDirStore(t.TempDir())
			priv, pub := testSealKeypair(t)
			p := NewMemProvisioner("eu-west-1")
			declare(t, s, "bab", "titogo", "TITO_S3")

			p.FailNext[method] = errors.New("injected: " + method + " failed")
			if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); err == nil {
				t.Fatalf("%s failure: Grant returned no error", method)
			}
			// The grant block is written before any backend call, so a
			// half-failed walk leaves the bucket choice pinned, not provisioned.
			half, _, err := Get(ctx, s, "bab", "titogo")
			if err != nil || half == nil || half.Grant == nil || half.Provisioned != nil {
				t.Fatalf("record after %s failure = %+v, %v", method, half, err)
			}

			rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
			if err != nil {
				t.Fatalf("resume after %s failure: %v", method, err)
			}
			if len(p.Buckets) != 1 {
				t.Fatalf("buckets after resume = %v", p.Buckets)
			}
			if len(p.Users) != 1 {
				t.Fatalf("users after resume = %v", p.Users)
			}
			u := p.Users["carlos-store-bab-titogo"]
			if len(u.Keys) != 1 {
				t.Fatalf("keys after resume = %v", u.Keys)
			}
			if _, ok := u.Keys[rec.Provisioned.KeyID]; !ok {
				t.Fatalf("recorded key %q not live: %v", rec.Provisioned.KeyID, u.Keys)
			}
			env := mustVars(t, s, "bab", "titogo")["TITO_S3_SECRET_KEY"].Envelope
			got, err := appconfig.Open(priv, appconfig.Binding{Account: "bab", App: "titogo"}, env)
			if err != nil || string(got) != u.Keys[rec.Provisioned.KeyID] {
				t.Fatalf("delivered secret = %q, %v; want %q", got, err, u.Keys[rec.Provisioned.KeyID])
			}
		})
	}

	t.Run("delivery", func(t *testing.T) {
		ctx := context.Background()
		priv, pub := testSealKeypair(t)
		s := &failingStore{Store: release.NewDirStore(t.TempDir()), failKeyOnce: appconfig.Key("bab", "titogo")}
		p := NewMemProvisioner("eu-west-1")
		declare(t, s, "bab", "titogo", "TITO_S3")

		if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); err == nil {
			t.Fatal("delivery failure: Grant returned no error")
		}
		// The record must NOT claim provisioned: the secret never landed.
		if rec, _, err := Get(ctx, s, "bab", "titogo"); err != nil || rec == nil || rec.Provisioned != nil {
			t.Fatalf("record after failed delivery = %+v, %v", rec, err)
		}

		rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
		if err != nil {
			t.Fatalf("resume after delivery failure: %v", err)
		}
		u := p.Users["carlos-store-bab-titogo"]
		if len(u.Keys) != 1 {
			t.Fatalf("undelivered key not reaped: %v", u.Keys)
		}
		if rec.Provisioned.KeyID != "MEMKEY2" {
			t.Fatalf("recorded key = %q, want the SECOND mint MEMKEY2", rec.Provisioned.KeyID)
		}
		vars := mustVars(t, s, "bab", "titogo")
		gotID, err := appconfig.Open(priv, appconfig.Binding{Account: "bab", App: "titogo"}, vars["TITO_S3_ACCESS_KEY"].Envelope)
		if err != nil || string(gotID) != "MEMKEY2" {
			t.Fatalf("delivered access key = %q, %v; want MEMKEY2", gotID, err)
		}
		gotSecret, err := appconfig.Open(priv, appconfig.Binding{Account: "bab", App: "titogo"}, vars["TITO_S3_SECRET_KEY"].Envelope)
		if err != nil || string(gotSecret) != u.Keys["MEMKEY2"] {
			t.Fatalf("delivered secret = %q, %v; want %q", gotSecret, err, u.Keys["MEMKEY2"])
		}
	})
}

func TestGrantRerunKeepsBucketChoice(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "titogo-uploads", time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	// Wipe delivery so the re-run walks the whole sequence again.
	if _, err := appconfig.Apply(ctx, s, "bab", "titogo", nil, []string{"TITO_S3_BUCKET"}); err != nil {
		t.Fatal(err)
	}
	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if rec.Provisioned.Bucket != "titogo-uploads" || rec.Grant.Bucket != "titogo-uploads" {
		t.Fatalf("re-run re-chose the bucket: grant=%+v provisioned=%+v", rec.Grant, rec.Provisioned)
	}
	if len(p.Buckets) != 1 {
		t.Fatalf("buckets = %v", p.Buckets)
	}
	if v := mustVars(t, s, "bab", "titogo")["TITO_S3_BUCKET"]; v.Value != "titogo-uploads" {
		t.Fatalf("re-delivered bucket = %+v", v)
	}
}

func TestGrantIdempotentWhenHealthy(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	first, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	before := mustVars(t, s, "bab", "titogo")

	second, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC().Add(time.Hour))
	if err != nil {
		t.Fatal(err)
	}
	if second.Provisioned.KeyID != first.Provisioned.KeyID {
		t.Fatalf("re-run minted a fresh key: %q then %q", first.Provisioned.KeyID, second.Provisioned.KeyID)
	}
	if u := p.Users["carlos-store-bab-titogo"]; len(u.Keys) != 1 {
		t.Fatalf("keys after healthy re-run = %v", u.Keys)
	}
	after := mustVars(t, s, "bab", "titogo")
	if len(before) != len(after) {
		t.Fatalf("var set changed: %v then %v", before, after)
	}
	for k, v := range before {
		if after[k] != v {
			t.Fatalf("var %s rewritten: %+v then %+v", k, v, after[k])
		}
	}
}

// A member (or an older tool) can occupy a delivered NAME with the wrong
// KIND — a key half stored as a plaintext env var is the motivating case. The
// app cannot use it, so the store is not delivered and the walk must repair.
func TestGrantRepairsWrongKindVar(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	priv, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	// Clobber the secret half with a plaintext env var of the same name.
	if _, err := appconfig.Apply(ctx, s, "bab", "titogo",
		map[string]appconfig.Var{"TITO_S3_SECRET_KEY": {Kind: "env", Value: "oops"}}, nil); err != nil {
		t.Fatal(err)
	}
	missing, err := EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", false)
	if err != nil || len(missing) != 1 || missing[0] != "TITO_S3_SECRET_KEY" {
		t.Fatalf("wrong-kind secret: missing = %v, %v", missing, err)
	}
	// Same for a coordinate occupied by a sealed secret.
	sealed, err := appconfig.Seal(pub, appconfig.Binding{Account: "bab", App: "titogo"}, []byte("carlos-store-tito-bab-titogo"))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := appconfig.Apply(ctx, s, "bab", "titogo",
		map[string]appconfig.Var{"TITO_S3_BUCKET": {Kind: "secret", Envelope: sealed}}, nil); err != nil {
		t.Fatal(err)
	}
	if missing, err = EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", false); err != nil || len(missing) != 2 {
		t.Fatalf("wrong-kind coordinate: missing = %v, %v", missing, err)
	}

	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	vars := mustVars(t, s, "bab", "titogo")
	if v := vars["TITO_S3_BUCKET"]; v.Kind != "env" || v.Value != "carlos-store-tito-bab-titogo" {
		t.Fatalf("bucket not repaired: %+v", v)
	}
	sk := vars["TITO_S3_SECRET_KEY"]
	if sk.Kind != "secret" || sk.Value != "" {
		t.Fatalf("secret not repaired: %+v", sk)
	}
	live := p.Users["carlos-store-bab-titogo"].Keys[rec.Provisioned.KeyID]
	got, err := appconfig.Open(priv, appconfig.Binding{Account: "bab", App: "titogo"}, sk.Envelope)
	if err != nil || string(got) != live {
		t.Fatalf("repaired secret = %q, %v; want %q", got, err, live)
	}
	if missing, err = EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", false); err != nil || len(missing) != 0 {
		t.Fatalf("after repair: missing = %v, %v", missing, err)
	}
}

func TestValidBucketName(t *testing.T) {
	for name, want := range map[string]bool{
		"carlos-store-tito-bab-titogo": true,
		"titogo-uploads":               true,
		"a.b-c9":                       true,
		"abc":                          true,
		"ab":                           false, // too short
		"":                             false,
		"UPPER-case":                   false,
		"has_underscore":               false,
		"-leading":                     false,
		"trailing-":                    false,
		"has space":                    false,
		"arn:aws:s3:::other/*":         false,
		"x" + string(make([]byte, 70)): false,
	} {
		if ValidBucketName(name) != want {
			t.Errorf("ValidBucketName(%q) != %v", name, want)
		}
	}
	// 63 legal, 64 not
	if !ValidBucketName("a" + strings.Repeat("b", 61) + "c") {
		t.Error("63-char name refused")
	}
	if ValidBucketName("a" + strings.Repeat("b", 62) + "c") {
		t.Error("64-char name accepted")
	}
}

// An adopt name is interpolated into the IAM policy ARN, so a bad one must
// never reach the backend — nor be recorded as this app's bucket.
func TestGrantRefusesInvalidBucketName(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	_, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "Other-Bucket/*", time.Now().UTC())
	if err == nil || !strings.Contains(err.Error(), "invalid bucket name") {
		t.Fatalf("adopt of an invalid name: %v", err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("refused grant touched the backend: %v %v", p.Buckets, p.Users)
	}
	rec, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || rec == nil || rec.Grant != nil {
		t.Fatalf("refused grant was recorded: %+v, %v", rec, err)
	}
	// A deployment whose prefix mints an illegal name is refused the same way.
	d := testDeployment()
	d.BucketPrefix = "Carlos_Store-"
	if _, err := Grant(ctx, s, p, d, pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); err == nil ||
		!strings.Contains(err.Error(), "invalid bucket name") {
		t.Fatalf("minted invalid name: %v", err)
	}
}

// The backend picks the secret's bytes. One that cannot be rendered into an
// EnvironmentFile must be refused HERE, where the operator sees it, not hours
// later on every host — and the refusal must not echo the secret.
func TestGrantRefusesUnrenderableSecret(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	p.MintSecret = "bad\"secret\nwith\\newline"
	declare(t, s, "bab", "titogo", "TITO_S3")

	_, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err == nil {
		t.Fatal("unrenderable secret was delivered")
	}
	if !strings.Contains(err.Error(), "TITO_S3_SECRET_KEY") {
		t.Fatalf("error does not name the variable: %v", err)
	}
	if strings.Contains(err.Error(), p.MintSecret) || strings.Contains(err.Error(), "bad") {
		t.Fatalf("error echoed the secret: %v", err)
	}
	if rec, _, gerr := Get(ctx, s, "bab", "titogo"); gerr != nil || rec.Provisioned != nil {
		t.Fatalf("record claims provisioned after a refused delivery: %+v, %v", rec, gerr)
	}
	if v := mustVars(t, s, "bab", "titogo"); len(v) != 0 {
		t.Fatalf("config written despite refusal: %v", v)
	}
	// The wasted key is collected by the next walk's reap.
	p.MintSecret = ""
	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if u := p.Users["carlos-store-bab-titogo"]; len(u.Keys) != 1 || u.Keys[rec.Provisioned.KeyID] == "" {
		t.Fatalf("keys after recovery = %v (recorded %q)", u.Keys, rec.Provisioned.KeyID)
	}
}

// Dropping a deployment's endpoint override must converge: the stale
// _ENDPOINT is unset, not left pointing hosts at an address nobody uses.
func TestGrantUnsetsStaleEndpoint(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	withEndpoint := testDeployment()
	withEndpoint.Endpoint = "https://s3.example.net"
	if _, err := Grant(ctx, s, p, withEndpoint, pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	if _, ok := mustVars(t, s, "bab", "titogo")["TITO_S3_ENDPOINT"]; !ok {
		t.Fatal("endpoint never delivered")
	}
	// Same app, deployment no longer has an endpoint: the record's endpoint no
	// longer matches, so the walk must run rather than short-circuit.
	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if rec.Provisioned.Endpoint != "" {
		t.Fatalf("record still carries the endpoint: %+v", rec.Provisioned)
	}
	if v, ok := mustVars(t, s, "bab", "titogo")["TITO_S3_ENDPOINT"]; ok {
		t.Fatalf("stale endpoint left behind: %+v", v)
	}
	missing, err := EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", false)
	if err != nil || len(missing) != 0 {
		t.Fatalf("after endpoint drop: missing = %v, %v", missing, err)
	}
}

func TestEnvDeliveredReportsMissing(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")

	missing, err := EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", false)
	if err != nil {
		t.Fatal(err)
	}
	if len(missing) != 4 {
		t.Fatalf("undelivered app: missing = %v, want the 4 non-endpoint names", missing)
	}

	declare(t, s, "bab", "titogo", "TITO_S3")
	if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	if missing, err = EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", false); err != nil || len(missing) != 0 {
		t.Fatalf("after grant: missing = %v, %v", missing, err)
	}
	// endpoint wanted but never delivered (deployment has no endpoint)
	if missing, err = EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", true); err != nil ||
		len(missing) != 1 || missing[0] != "TITO_S3_ENDPOINT" {
		t.Fatalf("endpoint wanted: missing = %v, %v", missing, err)
	}
	if _, err := appconfig.Apply(ctx, s, "bab", "titogo", nil, []string{"TITO_S3_SECRET_KEY"}); err != nil {
		t.Fatal(err)
	}
	if missing, err = EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", false); err != nil ||
		len(missing) != 1 || missing[0] != "TITO_S3_SECRET_KEY" {
		t.Fatalf("after unset: missing = %v, %v", missing, err)
	}
}

// rotateFixture grants a store and returns the provisioned record, so each
// rotation test starts from the one state Rotate is defined against.
func rotateFixture(t *testing.T, s release.Store, pub *ecdh.PublicKey, p *MemProvisioner) *Record {
	t.Helper()
	declare(t, s, "bab", "titogo", "TITO_S3")
	rec, err := Grant(context.Background(), s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	return rec
}

// openVar decrypts a delivered sealed var, so a test asserts on what the app
// would actually read rather than on the record alone.
func openVar(t *testing.T, priv *ecdh.PrivateKey, s release.Store, acct, app, name string) string {
	t.Helper()
	v := mustVars(t, s, acct, app)[name]
	if v.Kind != "secret" || v.Envelope == "" {
		t.Fatalf("%s = %+v, want a sealed secret", name, v)
	}
	plain, err := appconfig.Open(priv, appconfig.Binding{Account: acct, App: app}, v.Envelope)
	if err != nil {
		t.Fatalf("open %s: %v", name, err)
	}
	return string(plain)
}

// The whole point of the two-key window: the app keeps running on the key it
// already holds (in-flight requests, un-restarted processes) while the new one
// is delivered. The old key stays ACTIVE until an explicit finish.
func TestRotateKeepsOldKeyActive(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	priv, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	granted := rotateFixture(t, s, pub, p)
	oldID := granted.Provisioned.KeyID

	now := time.Now().UTC().Add(time.Hour)
	rec, err := Rotate(ctx, s, p, pub, "bab", "titogo", now)
	if err != nil {
		t.Fatal(err)
	}
	if rec.Provisioned.KeyID == oldID {
		t.Fatalf("rotate did not mint a new key: still %q", rec.Provisioned.KeyID)
	}
	if rec.Provisioned.PrevKeyID != oldID {
		t.Fatalf("PrevKeyID = %q, want the old key %q", rec.Provisioned.PrevKeyID, oldID)
	}
	if !rec.Provisioned.At.Equal(now) {
		t.Fatalf("At = %v, want %v", rec.Provisioned.At, now)
	}
	// coordinates are carried, never re-derived
	if rec.Provisioned.Bucket != granted.Provisioned.Bucket ||
		rec.Provisioned.Region != granted.Provisioned.Region ||
		rec.Provisioned.IAMUser != granted.Provisioned.IAMUser ||
		rec.Provisioned.UserARN != granted.Provisioned.UserARN {
		t.Fatalf("rotate moved the store: %+v", rec.Provisioned)
	}

	u := p.Users["carlos-store-bab-titogo"]
	if len(u.Keys) != 2 {
		t.Fatalf("keys after rotate = %v, want both old and new", u.Keys)
	}
	if _, ok := u.Keys[oldID]; !ok {
		t.Fatalf("old key %q was deleted by rotate: %v", oldID, u.Keys)
	}
	if _, ok := u.Keys[rec.Provisioned.KeyID]; !ok {
		t.Fatalf("new key %q is not live: %v", rec.Provisioned.KeyID, u.Keys)
	}

	// the app reads the NEW pair
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != rec.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, want %q", got, rec.Provisioned.KeyID)
	}
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_SECRET_KEY"); got != u.Keys[rec.Provisioned.KeyID] {
		t.Fatalf("delivered secret = %q, want %q", got, u.Keys[rec.Provisioned.KeyID])
	}
	// coordinates still delivered alongside, in the same config
	if v := mustVars(t, s, "bab", "titogo")["TITO_S3_BUCKET"]; v.Kind != "env" || v.Value != granted.Provisioned.Bucket {
		t.Fatalf("TITO_S3_BUCKET = %+v", v)
	}

	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored == nil || stored.Provisioned == nil ||
		stored.Provisioned.KeyID != rec.Provisioned.KeyID || stored.Provisioned.PrevKeyID != oldID {
		t.Fatalf("stored record = %+v, %v", stored, err)
	}
}

// AWS allows two access keys per user, full stop. A member who rotates twice
// without finishing must not wedge at the ceiling: the second rotate finishes
// the pending one first, so the estate is never more than two keys.
func TestRotateTwiceReapsPending(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	priv, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	first := rotateFixture(t, s, pub, p).Provisioned.KeyID
	user := "carlos-store-bab-titogo"

	second, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if len(p.Users[user].Keys) != 2 {
		t.Fatalf("keys after first rotate = %v", p.Users[user].Keys)
	}

	third, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	keys := p.Users[user].Keys
	if len(keys) != 2 {
		t.Fatalf("keys after second rotate = %v, want exactly 2", keys)
	}
	if _, ok := keys[first]; ok {
		t.Fatalf("the finished key %q is still live: %v", first, keys)
	}
	if _, ok := keys[second.Provisioned.KeyID]; !ok {
		t.Fatalf("pending key %q was deleted: %v", second.Provisioned.KeyID, keys)
	}
	if _, ok := keys[third.Provisioned.KeyID]; !ok {
		t.Fatalf("new key %q is not live: %v", third.Provisioned.KeyID, keys)
	}
	if third.Provisioned.PrevKeyID != second.Provisioned.KeyID {
		t.Fatalf("PrevKeyID = %q, want %q", third.Provisioned.PrevKeyID, second.Provisioned.KeyID)
	}
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != third.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, want %q", got, third.Provisioned.KeyID)
	}
}

// A rotate that mints and then fails to reseal leaves a key the record does
// not name. Re-running must collect it (the reap) rather than mint into a full
// estate, and the app must end up holding the key the record names.
func TestRotateResumesAfterResealFailure(t *testing.T) {
	ctx := context.Background()
	priv, pub := testSealKeypair(t)
	s := &failingStore{Store: release.NewDirStore(t.TempDir())}
	p := NewMemProvisioner("eu-west-1")
	granted := rotateFixture(t, s, pub, p)
	oldID := granted.Provisioned.KeyID
	user := "carlos-store-bab-titogo"

	s.failKeyOnce = appconfig.Key("bab", "titogo")
	if _, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC()); err == nil {
		t.Fatal("reseal failure: Rotate returned no error")
	}
	// The record must still name the key the app is holding — the new one was
	// never delivered, so claiming it would strand the app on a dead name.
	half, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || half == nil || half.Provisioned == nil ||
		half.Provisioned.KeyID != oldID || half.Provisioned.PrevKeyID != "" {
		t.Fatalf("record after failed reseal = %+v, %v", half, err)
	}
	if len(p.Users[user].Keys) > 2 {
		t.Fatalf("ceiling breached mid-failure: %v", p.Users[user].Keys)
	}
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != oldID {
		t.Fatalf("failed rotate moved the app's key to %q, want %q", got, oldID)
	}

	rec, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC())
	if err != nil {
		t.Fatalf("resume after reseal failure: %v", err)
	}
	keys := p.Users[user].Keys
	if len(keys) != 2 {
		t.Fatalf("keys after resume = %v, want exactly 2 (stray reaped)", keys)
	}
	if rec.Provisioned.PrevKeyID != oldID {
		t.Fatalf("PrevKeyID = %q, want %q", rec.Provisioned.PrevKeyID, oldID)
	}
	if _, ok := keys[rec.Provisioned.KeyID]; !ok {
		t.Fatalf("recorded key %q is not live: %v", rec.Provisioned.KeyID, keys)
	}
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != rec.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, want %q", got, rec.Provisioned.KeyID)
	}
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_SECRET_KEY"); got != keys[rec.Provisioned.KeyID] {
		t.Fatalf("delivered secret = %q, want %q", got, keys[rec.Provisioned.KeyID])
	}
}

// Finish is the member saying "nothing is using the old key any more". It
// deletes exactly that key and nothing else, delivers nothing, and is not
// repeatable — a second finish has no key to name and must say so.
func TestRotateFinishDeletesPrev(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	priv, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	oldID := rotateFixture(t, s, pub, p).Provisioned.KeyID
	user := "carlos-store-bab-titogo"

	rotated, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	before := mustVars(t, s, "bab", "titogo")

	rec, err := RotateFinish(ctx, s, p, "bab", "titogo")
	if err != nil {
		t.Fatal(err)
	}
	if rec.Provisioned.PrevKeyID != "" {
		t.Fatalf("PrevKeyID after finish = %q", rec.Provisioned.PrevKeyID)
	}
	if rec.Provisioned.KeyID != rotated.Provisioned.KeyID {
		t.Fatalf("finish changed the live key: %q", rec.Provisioned.KeyID)
	}
	keys := p.Users[user].Keys
	if len(keys) != 1 {
		t.Fatalf("keys after finish = %v, want just the live one", keys)
	}
	if _, ok := keys[oldID]; ok {
		t.Fatalf("finished key %q still live: %v", oldID, keys)
	}
	if _, ok := keys[rotated.Provisioned.KeyID]; !ok {
		t.Fatalf("live key %q deleted by finish: %v", rotated.Provisioned.KeyID, keys)
	}
	// Finish delivers nothing: the app's config is untouched.
	after := mustVars(t, s, "bab", "titogo")
	if len(before) != len(after) {
		t.Fatalf("finish rewrote the var set: %v then %v", before, after)
	}
	for k, v := range before {
		if after[k] != v {
			t.Fatalf("finish rewrote %s: %+v then %+v", k, v, after[k])
		}
	}
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != rec.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, want %q", got, rec.Provisioned.KeyID)
	}

	if _, err := RotateFinish(ctx, s, p, "bab", "titogo"); !errors.Is(err, ErrNoPendingFinish) {
		t.Fatalf("second finish: %v, want ErrNoPendingFinish", err)
	}
	if len(p.Users[user].Keys) != 1 {
		t.Fatalf("refused finish touched the backend: %v", p.Users[user].Keys)
	}

	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored == nil || stored.Provisioned == nil || stored.Provisioned.PrevKeyID != "" {
		t.Fatalf("stored record = %+v, %v", stored, err)
	}
}

// Neither walk invents a store: there is nothing to rotate until a grant has
// provisioned one, and both must refuse without touching the backend.
func TestRotateUnprovisionedRefused(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	if _, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC()); !errors.Is(err, ErrNotProvisioned) {
		t.Fatalf("Rotate on a declared-only store: %v, want ErrNotProvisioned", err)
	}
	if _, err := RotateFinish(ctx, s, p, "bab", "titogo"); !errors.Is(err, ErrNotProvisioned) {
		t.Fatalf("RotateFinish on a declared-only store: %v, want ErrNotProvisioned", err)
	}
	// and on an app with no record at all
	if _, err := Rotate(ctx, s, p, pub, "bab", "nostore", time.Now().UTC()); !errors.Is(err, ErrNotProvisioned) {
		t.Fatalf("Rotate on a store-less app: %v, want ErrNotProvisioned", err)
	}
	if _, err := RotateFinish(ctx, s, p, "bab", "nostore"); !errors.Is(err, ErrNotProvisioned) {
		t.Fatalf("RotateFinish on a store-less app: %v, want ErrNotProvisioned", err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("refused rotation touched the backend: %v %v", p.Buckets, p.Users)
	}
	if v := mustVars(t, s, "bab", "titogo"); len(v) != 0 {
		t.Fatalf("refused rotation wrote config: %v", v)
	}
}

// A rotation changes credentials and NOTHING else. The coordinates it
// redelivers come from the record, so an endpoint-bearing store must come out
// of a rotate still pointing at its endpoint — a rotate that re-derived them
// (or dropped them) would silently move every host onto the backend's own
// address, which is a data-plane outage with a green rotate next to it.
func TestRotateKeepsRecordedCoordinates(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	priv, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-central-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	d := testDeployment()
	d.Endpoint = "https://s3.example.net"
	granted, err := Grant(ctx, s, p, d, pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if granted.Provisioned.Endpoint != d.Endpoint || granted.Provisioned.Region != "eu-central-1" {
		t.Fatalf("fixture did not record the coordinates: %+v", granted.Provisioned)
	}

	// Rotate takes no Deployment: the only source for these values is the
	// record, so this is the assertion that it reads them from there.
	rec, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if rec.Provisioned.Endpoint != d.Endpoint || rec.Provisioned.Region != "eu-central-1" ||
		rec.Provisioned.Bucket != granted.Provisioned.Bucket {
		t.Fatalf("rotate moved the coordinates: %+v", rec.Provisioned)
	}
	vars := mustVars(t, s, "bab", "titogo")
	if v := vars["TITO_S3_ENDPOINT"]; v.Kind != "env" || v.Value != d.Endpoint {
		t.Fatalf("TITO_S3_ENDPOINT after rotate = %+v, want %q", v, d.Endpoint)
	}
	if v := vars["TITO_S3_REGION"]; v.Kind != "env" || v.Value != "eu-central-1" {
		t.Fatalf("TITO_S3_REGION after rotate = %+v", v)
	}
	if v := vars["TITO_S3_BUCKET"]; v.Kind != "env" || v.Value != granted.Provisioned.Bucket {
		t.Fatalf("TITO_S3_BUCKET after rotate = %+v", v)
	}
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != rec.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, want the new id %q", got, rec.Provisioned.KeyID)
	}
	if missing, err := EnvDelivered(ctx, s, "bab", "titogo", "TITO_S3", true); err != nil || len(missing) != 0 {
		t.Fatalf("after rotate: missing = %v, %v", missing, err)
	}
}

// racingStore lands a competing write on the record immediately BEFORE the
// walk's own Put reaches the store — the interleaving that turns the walk's
// etag stale. One shot, so only the write under test races.
type racingStore struct {
	release.Store
	raceKeyOnce string
	race        func(inner release.Store)
}

func (r *racingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key != "" && key == r.raceKeyOnce {
		r.raceKeyOnce = ""
		r.race(r.Store)
	}
	return r.Store.Put(ctx, key, body, ifMatch)
}

// The window between a rotate's Get and its Put is real: the console's
// declare/grant doors write the same object. Losing that CAS outright would
// leave the record naming the OLD key while the app runs on the new one —
// and the next rotate's reap would then delete the key actually in use. So
// the walk retries against a re-read record, and the competitor's change to
// the blocks it does not own must survive.
func TestRotateRetriesLostCAS(t *testing.T) {
	ctx := context.Background()
	priv, pub := testSealKeypair(t)
	inner := release.NewDirStore(t.TempDir())
	s := &racingStore{Store: inner}
	p := NewMemProvisioner("eu-west-1")
	oldID := rotateFixture(t, s, pub, p).Provisioned.KeyID

	// A competing writer touches a block this walk does not own.
	s.raceKeyOnce = Key("bab", "titogo")
	s.race = func(in release.Store) {
		other, etag, err := Get(ctx, in, "bab", "titogo")
		if err != nil {
			t.Fatal(err)
		}
		other.Declared.By = "someone-else@example.com"
		if _, err := Put(ctx, in, "bab", "titogo", other, etag); err != nil {
			t.Fatal(err)
		}
	}

	rec, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC())
	if err != nil {
		t.Fatalf("rotate lost its CAS and gave up: %v", err)
	}
	if rec.Provisioned.PrevKeyID != oldID || rec.Provisioned.KeyID == oldID {
		t.Fatalf("provisioned = %+v", rec.Provisioned)
	}
	if rec.Declared.By != "someone-else@example.com" {
		t.Fatalf("retry clobbered the competing write: Declared.By = %q", rec.Declared.By)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored.Declared.By != "someone-else@example.com" ||
		stored.Provisioned.KeyID != rec.Provisioned.KeyID || stored.Provisioned.PrevKeyID != oldID {
		t.Fatalf("stored record = %+v / %+v, %v", stored.Declared, stored.Provisioned, err)
	}
	// The record and the app agree on the live key — the whole point.
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != stored.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, record says %q", got, stored.Provisioned.KeyID)
	}
}

// Rotate-vs-rotate is the case the retry must NOT paper over: the competitor
// delivered a credential of its own, so overwriting its record would strand
// that key exactly as a lost CAS would strand ours. The conflict is surfaced.
func TestRotateRefusesRetryAgainstAnotherRotation(t *testing.T) {
	ctx := context.Background()
	_, pub := testSealKeypair(t)
	inner := release.NewDirStore(t.TempDir())
	s := &racingStore{Store: inner}
	p := NewMemProvisioner("eu-west-1")
	oldID := rotateFixture(t, s, pub, p).Provisioned.KeyID

	s.raceKeyOnce = Key("bab", "titogo")
	s.race = func(in release.Store) {
		other, etag, err := Get(ctx, in, "bab", "titogo")
		if err != nil {
			t.Fatal(err)
		}
		other.Provisioned.KeyID = "MEMKEY99" // another rotation landed first
		other.Provisioned.PrevKeyID = oldID
		if _, err := Put(ctx, in, "bab", "titogo", other, etag); err != nil {
			t.Fatal(err)
		}
	}

	if _, err := Rotate(ctx, s, p, pub, "bab", "titogo", time.Now().UTC()); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("rotate over a concurrent rotation: %v, want ErrCASConflict", err)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored.Provisioned.KeyID != "MEMKEY99" || stored.Provisioned.PrevKeyID != oldID {
		t.Fatalf("the competing rotation was clobbered: %+v, %v", stored.Provisioned, err)
	}
}

// teardownFixture grants a fresh store and returns its record — the estate a
// teardown has to dismantle.
func teardownFixture(t *testing.T, s release.Store, pub *ecdh.PublicKey, p *MemProvisioner) *Record {
	t.Helper()
	declare(t, s, "bab", "titogo", "TITO_S3")
	rec, err := Grant(context.Background(), s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	return rec
}

func TestTeardownFullSweep(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	rec := teardownFixture(t, s, pub, p)
	// A bucket with data in it is the case that matters: DeleteBucket alone
	// would fail on a real backend, so the walk must empty it first.
	p.Buckets[rec.Provisioned.Bucket]["uploads/one.txt"] = []byte("payload")

	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil {
		t.Fatal(err)
	}
	if len(p.Buckets) != 0 {
		t.Fatalf("buckets after teardown = %v", p.Buckets)
	}
	if len(p.Users) != 0 {
		t.Fatalf("users after teardown = %v", p.Users)
	}
	for name := range mustVars(t, s, "bab", "titogo") {
		if strings.HasPrefix(name, "TITO_S3_") {
			t.Fatalf("delivered var %q survived teardown", name)
		}
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored != nil {
		t.Fatalf("record after teardown = %+v, %v", stored, err)
	}
}

func TestTeardownResumable(t *testing.T) {
	for _, method := range []string{"EmptyBucket", "DeleteBucket", "DeleteUser"} {
		t.Run(method, func(t *testing.T) {
			ctx := context.Background()
			s := release.NewDirStore(t.TempDir())
			_, pub := testSealKeypair(t)
			p := NewMemProvisioner("eu-west-1")
			rec := teardownFixture(t, s, pub, p)
			p.Buckets[rec.Provisioned.Bucket]["uploads/one.txt"] = []byte("payload")

			p.FailNext[method] = errors.New("injected: " + method + " failed")
			if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err == nil {
				t.Fatalf("%s failure: Teardown returned no error", method)
			}
			// The record is the ONLY name for what is left standing (§9), so a
			// half-failed teardown must leave it intact and still provisioned.
			half, _, err := Get(ctx, s, "bab", "titogo")
			if err != nil || half == nil || half.Provisioned == nil {
				t.Fatalf("record after %s failure = %+v, %v", method, half, err)
			}
			if half.Provisioned.Bucket != rec.Provisioned.Bucket || half.Provisioned.IAMUser != rec.Provisioned.IAMUser {
				t.Fatalf("record coordinates changed after %s failure: %+v", method, half.Provisioned)
			}

			if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil {
				t.Fatalf("resume after %s failure: %v", method, err)
			}
			if len(p.Buckets) != 0 || len(p.Users) != 0 {
				t.Fatalf("estate after resume: buckets %v users %v", p.Buckets, p.Users)
			}
			for name := range mustVars(t, s, "bab", "titogo") {
				if strings.HasPrefix(name, "TITO_S3_") {
					t.Fatalf("delivered var %q survived teardown", name)
				}
			}
			stored, _, err := Get(ctx, s, "bab", "titogo")
			if err != nil || stored != nil {
				t.Fatalf("record after resume = %+v, %v", stored, err)
			}
		})
	}
}

func TestTeardownDeclaredOnly(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil {
		t.Fatal(err)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored != nil {
		t.Fatalf("declared-only record after teardown = %+v, %v", stored, err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("declared-only teardown touched the backend: %v %v", p.Buckets, p.Users)
	}
}

// A deployment with no backend may still tear down a DECLARATION — there is
// nothing outside the bucket to act on — but an estate refuses instead of
// dereferencing the nil backend. The console's destroy door decides from a
// record it read a round trip earlier, so a grant landing in between arrives
// here as exactly this shape, and a panic would be the difference between a
// refusal and a 500 over a live bucket.
func TestTeardownWithNoProvisioner(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	declare(t, s, "bab", "titogo", "TITO_S3")

	if err := Teardown(ctx, s, nil, nil, "bab", "titogo"); err != nil {
		t.Fatalf("declared-only teardown with no provisioner: %v", err)
	}

	// Now the same walk over a grant that DID create backend estate.
	declare(t, s, "bab", "titogo", "TITO_S3")
	rec, etag, err := Get(ctx, s, "bab", "titogo")
	if err != nil {
		t.Fatal(err)
	}
	rec.Grant = &GrantBlock{Backend: "mem", Bucket: "carlos-bab-titogo", At: time.Now(), By: "op@keymail.dev"}
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}
	if err := Teardown(ctx, s, nil, nil, "bab", "titogo"); err == nil {
		t.Fatal("teardown of a granted store with no provisioner returned nil — it cannot have destroyed anything")
	}
	if stored, _, err := Get(ctx, s, "bab", "titogo"); err != nil || stored == nil || stored.Grant == nil {
		t.Fatalf("the refused teardown did not leave the record intact: %+v %v", stored, err)
	}
}

func TestTeardownIdempotentOnAbsent(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	p := NewMemProvisioner("eu-west-1")
	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil {
		t.Fatalf("teardown with no record: %v", err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("teardown with no record touched the backend: %v %v", p.Buckets, p.Users)
	}
}

func TestTeardownConfigObjectAbsent(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	p := NewMemProvisioner("eu-west-1")
	// A provisioned record whose app was never configured — hand-crafted, so
	// no config.json exists at all. appconfig REFUSES an unset-only apply on a
	// nonexistent object, and that refusal must not strand the estate.
	if _, err := p.EnsureBucket(ctx, "carlos-store-tito-bab-titogo"); err != nil {
		t.Fatal(err)
	}
	arn, err := p.EnsureUser(ctx, "carlos-store-bab-titogo", UserPath, "arn:aws:s3:::carlos-store-tito-bab-titogo", nil)
	if err != nil {
		t.Fatal(err)
	}
	rec := &Record{
		Declared: &Declared{EnvPrefix: "TITO_S3", At: time.Now().UTC(), By: "member@example.com"},
		Grant:    &GrantBlock{Backend: "aws", Bucket: "carlos-store-tito-bab-titogo", At: time.Now().UTC(), By: "op"},
		Provisioned: &Provisioned{Bucket: "carlos-store-tito-bab-titogo", Region: "eu-west-1",
			IAMUser: "carlos-store-bab-titogo", UserARN: arn, KeyID: "MEMKEY1", At: time.Now().UTC()},
	}
	if _, err := Put(ctx, s, "bab", "titogo", rec, ""); err != nil {
		t.Fatal(err)
	}

	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil {
		t.Fatalf("teardown with no config object: %v", err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("estate after teardown: buckets %v users %v", p.Buckets, p.Users)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored != nil {
		t.Fatalf("record after teardown = %+v, %v", stored, err)
	}
	// And no phantom config object was minted on the way through.
	if _, _, err := s.Get(ctx, appconfig.Key("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("teardown minted a config object: %v", err)
	}
}

// crashedGrant reproduces a grant that created and delivered everything and
// then lost its step-6 record Put: bucket, user, key and delivered vars all
// live, record carrying Declared+Grant and NO provisioned block.
func crashedGrant(t *testing.T, s release.Store, p *MemProvisioner) (bucket, user string) {
	t.Helper()
	ctx := context.Background()
	bucket, user = "carlos-store-tito-bab-titogo", UserName("bab", "titogo")
	if _, err := p.EnsureBucket(ctx, bucket); err != nil {
		t.Fatal(err)
	}
	if _, err := p.EnsureUser(ctx, user, UserPath, "arn:aws:s3:::"+bucket, nil); err != nil {
		t.Fatal(err)
	}
	id, _, err := p.MintKey(ctx, user)
	if err != nil {
		t.Fatal(err)
	}
	// writeEnv had already landed when the record write lost — the credentials
	// are in config.json under names no record now explains.
	if _, err := appconfig.Apply(ctx, s, "bab", "titogo", map[string]appconfig.Var{
		"TITO_S3_BUCKET":     {Kind: "env", Value: bucket},
		"TITO_S3_REGION":     {Kind: "env", Value: "eu-west-1"},
		"TITO_S3_ACCESS_KEY": {Kind: "env", Value: id},
	}, nil); err != nil {
		t.Fatal(err)
	}
	rec := &Record{
		Declared: &Declared{EnvPrefix: "TITO_S3", At: time.Now().UTC(), By: "member@example.com"},
		Grant:    &GrantBlock{Backend: "aws", Bucket: bucket, At: time.Now().UTC(), By: "op"},
	}
	if _, err := Put(ctx, s, "bab", "titogo", rec, ""); err != nil {
		t.Fatal(err)
	}
	return bucket, user
}

// A grant block with no provisioned block is NOT "nothing to tear down" — it
// is the shape a crash at Grant's final record write leaves over a live
// estate. Deleting the record alone would strand a billing bucket and a live
// credential nothing names.
func TestTeardownCrashedGrantSweep(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	p := NewMemProvisioner("eu-west-1")
	bucket, _ := crashedGrant(t, s, p)
	p.Buckets[bucket]["uploads/one.txt"] = []byte("payload")

	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil {
		t.Fatal(err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("crashed grant left standing: buckets %v users %v", p.Buckets, p.Users)
	}
	for name := range mustVars(t, s, "bab", "titogo") {
		if strings.HasPrefix(name, "TITO_S3_") {
			t.Fatalf("delivered var %q survived teardown", name)
		}
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored != nil {
		t.Fatalf("record after teardown = %+v, %v", stored, err)
	}
}

func TestTeardownCrashedGrantResumable(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	p := NewMemProvisioner("eu-west-1")
	bucket, _ := crashedGrant(t, s, p)
	p.Buckets[bucket]["uploads/one.txt"] = []byte("payload")

	p.FailNext["DeleteBucket"] = errors.New("injected: DeleteBucket failed")
	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err == nil {
		t.Fatal("DeleteBucket failure: Teardown returned no error")
	}
	half, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || half == nil || half.Grant == nil || half.Grant.Bucket != bucket {
		t.Fatalf("record after DeleteBucket failure = %+v, %v", half, err)
	}

	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); err != nil {
		t.Fatalf("resume after DeleteBucket failure: %v", err)
	}
	if len(p.Buckets) != 0 || len(p.Users) != 0 {
		t.Fatalf("estate after resume: buckets %v users %v", p.Buckets, p.Users)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored != nil {
		t.Fatalf("record after resume = %+v, %v", stored, err)
	}
}

// raceOnReadStore runs race() the FIRST time onGet is read, before the read
// returns — a competing writer landing at a chosen point in a walk.
type raceOnReadStore struct {
	release.Store
	onGet string
	race  func(release.Store)
}

func (r *raceOnReadStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if r.onGet != "" && key == r.onGet {
		r.onGet = ""
		r.race(r.Store)
	}
	return r.Store.Get(ctx, key)
}

// A grant that lands mid-teardown has re-provisioned a live estate and
// recorded it. The final delete must not erase that record just because this
// walk read an older one.
func TestTeardownRefusesConcurrentRegrant(t *testing.T) {
	ctx := context.Background()
	inner := release.NewDirStore(t.TempDir())
	_, pub := testSealKeypair(t)
	p := NewMemProvisioner("eu-west-1")
	s := &raceOnReadStore{Store: inner}
	teardownFixture(t, s, pub, p)

	// Fire on the config probe — the last read Teardown does before its
	// re-read guard, so the competing record is in place by the time the
	// guard looks.
	s.onGet = appconfig.Key("bab", "titogo")
	s.race = func(in release.Store) {
		other, etag, err := Get(ctx, in, "bab", "titogo")
		if err != nil {
			t.Error(err)
			return
		}
		other.Provisioned.KeyID = "MEMKEY99" // a fresh grant landed and recorded
		if _, err := Put(ctx, in, "bab", "titogo", other, etag); err != nil {
			t.Error(err)
		}
	}

	if err := Teardown(ctx, s, p, nil, "bab", "titogo"); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("teardown over a concurrent grant: %v, want ErrCASConflict", err)
	}
	stored, _, err := Get(ctx, inner, "bab", "titogo")
	if err != nil || stored == nil || stored.Provisioned == nil || stored.Provisioned.KeyID != "MEMKEY99" {
		t.Fatalf("the competing grant's record was erased: %+v, %v", stored, err)
	}
}

// TestPurgeGuardSeesOurRecord pins the PARSE AGREEMENT between this package
// and release.PurgeApp. release cannot import objstore (objstore imports
// release), so PurgeApp re-parses `grant` and `provisioned` out of store.json
// with a local struct of its own. That guard is what stops a box's auto-purge
// from sweeping away the only record of a paid bucket and a live IAM user.
// Records written HERE, by the real writer, must trip it — for the finished
// store and for the crashed grant alike.
func TestPurgeGuardSeesOurRecord(t *testing.T) {
	ctx := context.Background()
	_, pub := testSealKeypair(t)
	relPub, _, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	// Past the cooloff, so nothing but the store guard stands between the
	// purge and destruction.
	backdated := time.Now().Add(-2 * release.TrashCooloff)

	t.Run("provisioned", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		p := NewMemProvisioner("eu-west-1")
		rec := rotateFixture(t, s, pub, p) // declare + Grant: a real provisioned record
		if rec.Provisioned == nil {
			t.Fatal("fixture produced no provisioned block")
		}
		// Drop the grant block so `provisioned` stands ALONE. A real record
		// carries both, which would let this subtest pass on the grant key and
		// leave Provisioned's json tag unpinned — the one thing this test
		// exists to catch.
		stored, etag, err := Get(ctx, s, "bab", "titogo")
		if err != nil || stored == nil {
			t.Fatalf("re-read: %+v, %v", stored, err)
		}
		stored.Grant = nil
		if _, err := Put(ctx, s, "bab", "titogo", stored, etag); err != nil {
			t.Fatal(err)
		}
		if err := release.TrashApp(ctx, s, "bab", "titogo", "paul@keymail.dev", backdated); err != nil {
			t.Fatal(err)
		}
		err = release.PurgeApp(ctx, s, relPub, "bab", "titogo", time.Now())
		if !errors.Is(err, release.ErrStoreProvisioned) {
			t.Fatalf("PurgeApp over a provisioned record = %v, want release.ErrStoreProvisioned", err)
		}
		if _, _, err := s.Get(ctx, Key("bab", "titogo")); err != nil {
			t.Fatalf("the refused purge destroyed store.json: %v", err)
		}
	})

	t.Run("crashed-grant", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		p := NewMemProvisioner("eu-west-1")
		crashedGrant(t, s, p) // grant block, no provisioned block, live estate
		if err := release.TrashApp(ctx, s, "bab", "titogo", "paul@keymail.dev", backdated); err != nil {
			t.Fatal(err)
		}
		err := release.PurgeApp(ctx, s, relPub, "bab", "titogo", time.Now())
		if !errors.Is(err, release.ErrStoreProvisioned) {
			t.Fatalf("PurgeApp over a crashed grant = %v, want release.ErrStoreProvisioned", err)
		}
	})

	// The other direction, so the guard cannot pass by refusing everything: a
	// declared-only record has no backend estate and is purged like any other
	// object. This is also the state Teardown leaves behind (it deletes the
	// record outright), which is how a torn-down app becomes purgeable.
	t.Run("declared-only-purges", func(t *testing.T) {
		s := release.NewDirStore(t.TempDir())
		declare(t, s, "bab", "titogo", "TITO_S3")
		if err := release.TrashApp(ctx, s, "bab", "titogo", "paul@keymail.dev", backdated); err != nil {
			t.Fatal(err)
		}
		if err := release.PurgeApp(ctx, s, relPub, "bab", "titogo", time.Now()); err != nil {
			t.Fatalf("PurgeApp over a declared-only record: %v", err)
		}
		if _, _, err := s.Get(ctx, Key("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("store.json survived the purge: %v", err)
		}
	})
}

// Source records WHY a grant is authorised (#381), so that an approval which
// is later withdrawn can be told apart from an operator's deliberate per-app
// grant — mail.GrantBlock.Source's reasoning, applied to storage.
func TestGrantRecordsItsSource(t *testing.T) {
	ctx := context.Background()
	_, pub := testSealKeypair(t)
	s := release.NewDirStore(t.TempDir())
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo",
		"op@example.com", GrantSourceAccountApproval, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if rec.Grant.Source != GrantSourceAccountApproval {
		t.Fatalf("grant source = %q, want %q", rec.Grant.Source, GrantSourceAccountApproval)
	}
	if rec.Grant.By != "op@example.com" {
		t.Fatalf("grant by = %q — an account-approved grant still names the approving operator", rec.Grant.By)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored.Grant.Source != GrantSourceAccountApproval {
		t.Fatalf("stored grant = %+v, %v", stored.Grant, err)
	}
}

// A re-run repairs an estate whose authority was already established. Restating
// the source would let a repair rewrite the record's account of who authorised
// the spend — the same reason a re-run never re-chooses the bucket.
func TestGrantDoesNotRestateSourceOnRerun(t *testing.T) {
	ctx := context.Background()
	_, pub := testSealKeypair(t)
	s := release.NewDirStore(t.TempDir())
	p := NewMemProvisioner("eu-west-1")
	declare(t, s, "bab", "titogo", "TITO_S3")

	if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo",
		"op@example.com", GrantSourceOperator, "", time.Now().UTC()); err != nil {
		t.Fatal(err)
	}
	// Re-run claiming a different authority; the first one must stand.
	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo",
		"someone@example.com", GrantSourceAccountApproval, "", time.Now().UTC())
	if err != nil {
		t.Fatal(err)
	}
	if rec.Grant.Source != GrantSourceOperator || rec.Grant.By != "op@example.com" {
		t.Fatalf("a re-run rewrote the grant's authority: %+v", rec.Grant)
	}
}

// grantBlockOnlyFixture leaves the record exactly as a walk that died mid-flight
// leaves it: the grant block written (Grant's step 1) and no provisioned block.
// A re-run then skips step 1's Put, so the next write it makes to the record is
// step 6's — the one these two tests race.
func grantBlockOnlyFixture(t *testing.T, s release.Store) {
	t.Helper()
	ctx := context.Background()
	declare(t, s, "bab", "titogo", "TITO_S3")
	rec, etag, err := Get(ctx, s, "bab", "titogo")
	if err != nil {
		t.Fatal(err)
	}
	rec.Grant = &GrantBlock{
		Backend: testDeployment().Backend,
		Bucket:  MintedBucket(testDeployment(), "bab", "titogo"),
		At:      time.Now().UTC(),
		By:      "op",
	}
	if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
		t.Fatal(err)
	}
}

// Grant's final Put has the same window Rotate's does — the console's declare
// door writes this very object, and under self-serve storage (#381) a member's
// own `carlos store create` drives the walk, so the window is ordinary rather
// than rare. Losing that CAS outright leaves the record naming no key (or an
// older one) while the app already holds the delivered credential, and the next
// rotate's reap then deletes the key actually in use — the outcome Rotate's own
// doc comment calls NOT survivable. So Grant retries against a re-read record,
// and a competitor's change to blocks it does not own must survive.
func TestGrantRetriesLostCAS(t *testing.T) {
	ctx := context.Background()
	priv, pub := testSealKeypair(t)
	inner := release.NewDirStore(t.TempDir())
	s := &racingStore{Store: inner}
	p := NewMemProvisioner("eu-west-1")
	grantBlockOnlyFixture(t, s)

	// A competing writer touches a block this walk does not own — exactly what
	// a re-declare through the console's create door does.
	s.raceKeyOnce = Key("bab", "titogo")
	s.race = func(in release.Store) {
		other, etag, err := Get(ctx, in, "bab", "titogo")
		if err != nil {
			t.Fatal(err)
		}
		other.Declared.By = "someone-else@example.com"
		if _, err := Put(ctx, in, "bab", "titogo", other, etag); err != nil {
			t.Fatal(err)
		}
	}

	rec, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC())
	if err != nil {
		t.Fatalf("grant lost its CAS and gave up: %v", err)
	}
	if rec.Provisioned == nil || rec.Provisioned.KeyID == "" {
		t.Fatalf("provisioned = %+v", rec.Provisioned)
	}
	if rec.Declared.By != "someone-else@example.com" {
		t.Fatalf("retry clobbered the competing write: Declared.By = %q", rec.Declared.By)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored.Declared.By != "someone-else@example.com" ||
		stored.Provisioned == nil || stored.Provisioned.KeyID != rec.Provisioned.KeyID {
		t.Fatalf("stored record = %+v / %+v, %v", stored.Declared, stored.Provisioned, err)
	}
	// The record and the app agree on the live key — the whole point.
	if got := openVar(t, priv, s, "bab", "titogo", "TITO_S3_ACCESS_KEY"); got != stored.Provisioned.KeyID {
		t.Fatalf("delivered access key = %q, record says %q", got, stored.Provisioned.KeyID)
	}
}

// Grant-vs-grant is the case the retry must NOT paper over: a competing walk
// minted and DELIVERED a key of its own, so overwriting its record would strand
// that key exactly as a lost CAS would strand ours. The conflict is surfaced.
func TestGrantRefusesRetryAgainstAnotherProvision(t *testing.T) {
	ctx := context.Background()
	_, pub := testSealKeypair(t)
	inner := release.NewDirStore(t.TempDir())
	s := &racingStore{Store: inner}
	p := NewMemProvisioner("eu-west-1")
	grantBlockOnlyFixture(t, s)

	s.raceKeyOnce = Key("bab", "titogo")
	s.race = func(in release.Store) {
		other, etag, err := Get(ctx, in, "bab", "titogo")
		if err != nil {
			t.Fatal(err)
		}
		other.Provisioned = &Provisioned{
			Bucket:  other.Grant.Bucket,
			Region:  "eu-west-1",
			IAMUser: UserName("bab", "titogo"),
			KeyID:   "MEMKEY99", // another grant landed first, and delivered
			At:      time.Now().UTC(),
		}
		if _, err := Put(ctx, in, "bab", "titogo", other, etag); err != nil {
			t.Fatal(err)
		}
	}

	if _, err := Grant(ctx, s, p, testDeployment(), pub, "bab", "titogo", "op", GrantSourceOperator, "", time.Now().UTC()); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("grant over a concurrent grant: %v, want ErrCASConflict", err)
	}
	stored, _, err := Get(ctx, s, "bab", "titogo")
	if err != nil || stored.Provisioned == nil || stored.Provisioned.KeyID != "MEMKEY99" {
		t.Fatalf("the competing grant was clobbered: %+v, %v", stored.Provisioned, err)
	}
}