package console

// A keyed run create (service runs spec § 3): admitted once, settled by the
// instance record, and a retry with the same key gets the same run.

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"mime/multipart"
	"net/http"
	"net/http/httptest"
	"net/url"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/runs"
	"github.com/carlosframework/platform/internal/svcrole"
)

const testKey = "fleet:import:acct-42"

// keyedFixture is runFixture with keyed creates turned on.
func keyedFixture(t *testing.T) *runFx {
	t.Helper()
	fx := runFixture(t)
	fx.srv.cfg.RunKeysEnabled = true
	return fx
}

func sha(s string) string {
	h := sha256.Sum256([]byte(s))
	return hex.EncodeToString(h[:])
}

// fileMeta is okMeta with one declared file, x.csv holding body.
func fileMeta(body string) runs.Meta {
	m := okMeta()
	m.Files = []runs.MetaFile{{Name: "x.csv", Size: int64(len(body)), SHA256: sha(body)}}
	return m
}

func xcsv(body string) []struct{ name, body string } {
	return []struct{ name, body string }{{"x.csv", body}}
}

// postKeyed is postRun with an Idempotency-Key header ("" sends none).
func postKeyed(t *testing.T, u, bearer, key string, meta runs.Meta, parts []struct{ name, body string }) (int, string) {
	t.Helper()
	var buf bytes.Buffer
	mw := multipart.NewWriter(&buf)
	mp, err := mw.CreateFormField("meta")
	if err != nil {
		t.Fatal(err)
	}
	if err := json.NewEncoder(mp).Encode(meta); err != nil {
		t.Fatal(err)
	}
	for _, p := range parts {
		fp, err := mw.CreateFormFile("file", p.name)
		if err != nil {
			t.Fatal(err)
		}
		io.WriteString(fp, p.body)
	}
	mw.Close()
	req, err := http.NewRequest(http.MethodPost, u, &buf)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Content-Type", mw.FormDataContentType())
	req.Header.Set("Authorization", "Bearer "+bearer)
	if key != "" {
		req.Header.Set("Idempotency-Key", key)
	}
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer res.Body.Close()
	out, _ := io.ReadAll(res.Body)
	return res.StatusCode, string(out)
}

// keyedCreated is the 202 body, with replayed.
type keyedCreated struct {
	runs.Created
	Replayed *bool `json:"replayed"`
}

func createdOf(t *testing.T, status int, body string) keyedCreated {
	t.Helper()
	if status != http.StatusAccepted {
		t.Fatalf("create: %d %s", status, body)
	}
	var c keyedCreated
	if err := json.Unmarshal([]byte(body), &c); err != nil || !runs.ValidID(c.ID) {
		t.Fatalf("created %s: %v", body, err)
	}
	return c
}

func (c keyedCreated) replayed() bool { return c.Replayed != nil && *c.Replayed }

// createKeyed is a keyed create that must answer 202.
func (fx *runFx) createKeyed(t *testing.T, bearer, key string, meta runs.Meta, parts []struct{ name, body string }) keyedCreated {
	t.Helper()
	status, body := postKeyed(t, fx.runsURL(runTestHost), bearer, key, meta, parts)
	return createdOf(t, status, body)
}

// runIDsWithObjects is every run id with objects under the shop app.
func (fx *runFx) runIDsWithObjects(t *testing.T) []string {
	t.Helper()
	prefix := runs.RunsPrefix(release.OpsAccount, "shop")
	keys, err := fx.rel.List(context.Background(), prefix)
	if err != nil {
		t.Fatal(err)
	}
	var ids []string
	for _, k := range keys {
		id := strings.SplitN(strings.TrimPrefix(k, prefix), "/", 2)[0]
		if !slices.Contains(ids, id) {
			ids = append(ids, id)
		}
	}
	return ids
}

func (fx *runFx) instRecord(t *testing.T) *provision.Record {
	t.Helper()
	rec, err := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost)
	if err != nil {
		t.Fatal(err)
	}
	return rec
}

func (fx *runFx) runStarts(t *testing.T) int {
	t.Helper()
	events, _ := fx.st.Events(OpsAccountID, "shop", 100)
	n := 0
	for _, e := range events {
		if e.Action == "run-start" {
			n++
		}
	}
	return n
}

// finishAndAck takes id from queued to acknowledged through the doors a box
// and the CLI use.
func (fx *runFx) finishAndAck(t *testing.T, bearer, id string) {
	t.Helper()
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0, LastSeq: 0}); err != nil {
		t.Fatal(err)
	}
	fx.transition(t, provision.RunFinish(id))
	if status, body := fx.followAs(t, bearer, "ack", id); status != http.StatusOK {
		t.Fatalf("ack: %d %s", status, body)
	}
}

func TestRunKeyedCreateThenRetryReturnsTheSameRun(t *testing.T) {
	fx := keyedFixture(t)
	first := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
	if first.replayed() || first.State != provision.RunQueued {
		t.Fatalf("a fresh admission: %+v", first)
	}
	again := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
	if again.ID != first.ID || !again.replayed() || again.Host != runTestHost {
		t.Fatalf("the retry: %+v, want %s replayed", again, first.ID)
	}
	if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != first.ID {
		t.Fatalf("runs with objects %v, want only %s", ids, first.ID)
	}
	if n := fx.runStarts(t); n != 1 {
		t.Errorf("%d run-start rows, want 1", n)
	}
	rec := fx.instRecord(t)
	if len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != first.ID || rec.RunKeys[0].Key != runs.ScopedKey("user:alice@keymail.dev", testKey) {
		t.Fatalf("run_keys %+v", rec.RunKeys)
	}
	req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", first.ID)
	if err != nil || len(req.Files) != 1 || req.Files[0].SHA256 != sha("a,b,c") {
		t.Fatalf("request %+v, %v", req, err)
	}
	// request.json links the run to its key by the scoped hash, never the
	// caller's own words.
	if req.IdempotencyKey != runs.ScopedKey("user:alice@keymail.dev", testKey) {
		t.Fatalf("request.json idempotency_key %q, want the scoped hash", req.IdempotencyKey)
	}
	raw, _, err := fx.rel.Get(context.Background(), runs.RequestKey(release.OpsAccount, "shop", first.ID))
	if err != nil || strings.Contains(string(raw), testKey) {
		t.Fatalf("request.json holds the raw key: %s, %v", raw, err)
	}
}

func TestRunKeyedRetryWhileOpenReturnsTheRun(t *testing.T) {
	fx := keyedFixture(t)
	runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
	first := fx.createKeyed(t, runner, testKey, serviceMeta(), nil)
	fx.transition(t, provision.RunPick(first.ID, "box-1", "boot-1", time.Now().Add(time.Hour)))
	again := fx.createKeyed(t, runner, testKey, serviceMeta(), nil)
	if again.ID != first.ID || !again.replayed() || again.State != provision.RunPicked {
		t.Fatalf("the retry while open: %+v, want %s picked and replayed", again, first.ID)
	}
	if n := fx.runStarts(t); n != 1 {
		t.Errorf("%d run-start rows, want 1", n)
	}
}

func TestRunKeyedRetryAfterAckStillReturnsTheRun(t *testing.T) {
	fx := keyedFixture(t)
	first := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	fx.finishAndAck(t, fx.member, first.ID)
	again := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	if again.ID != first.ID || !again.replayed() || again.State != "" {
		t.Fatalf("the retry after ack: %+v, want %s replayed, with no record state", again, first.ID)
	}
	// A later unkeyed run takes the record; the key still answers.
	other := fx.startRun(t)
	again = fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	if again.ID != first.ID || !again.replayed() {
		t.Fatalf("the retry after %s replaced it: %+v", other, again)
	}
	if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != other {
		t.Fatalf("runs with objects %v, want only %s", ids, other)
	}
}

func TestRunKeyReusedWithADifferentRequest(t *testing.T) {
	fx := keyedFixture(t)
	first := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	before := fx.instRecord(t)
	other := okMeta()
	other.Argv = []string{"import", "y.csv"}
	status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, other, nil)
	if status != http.StatusConflict || codeOf(t, body) != runCodeKeyReused || errorOf(t, body) == "" {
		t.Fatalf("the same key, another request: %d %s, want 409 %s", status, body, runCodeKeyReused)
	}
	if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != first.ID {
		t.Fatalf("runs with objects %v, want only %s", ids, first.ID)
	}
	if after := fx.instRecord(t); after.RecordSeq != before.RecordSeq {
		t.Fatalf("the refusal wrote the record: seq %d -> %d", before.RecordSeq, after.RecordSeq)
	}
}

// barrierStore holds every admission's record write until want request.json
// writes have landed, so that concurrent creates are all past the door's key
// lookup and have all written their objects before any is admitted.
type barrierStore struct {
	release.Store
	recordKey string
	want      int
	mu        sync.Mutex
	requests  int
	ready     chan struct{}
}

func (b *barrierStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == b.recordKey {
		select {
		case <-b.ready:
		case <-time.After(10 * time.Second):
			return "", errors.New("barrier: the other create never wrote its objects")
		}
	}
	etag, err := b.Store.Put(ctx, key, body, ifMatch)
	if err == nil && strings.HasSuffix(key, "/request.json") {
		b.mu.Lock()
		b.requests++
		if b.requests == b.want {
			close(b.ready)
		}
		b.mu.Unlock()
	}
	return etag, err
}

func TestRunConcurrentIdenticalKeyedCreatesMakeOneRun(t *testing.T) {
	fx := keyedFixture(t)
	runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
	fx.srv.cfg.Releases = &barrierStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), want: 2, ready: make(chan struct{})}
	meta := serviceMeta()
	meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5, SHA256: sha("a,b,c")}}
	var wg sync.WaitGroup
	statuses, bodies := make([]int, 2), make([]string, 2)
	for i := range 2 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			statuses[i], bodies[i] = postKeyed(t, fx.runsURL(runTestHost), runner, testKey, meta, xcsv("a,b,c"))
		}()
	}
	wg.Wait()
	a, b := createdOf(t, statuses[0], bodies[0]), createdOf(t, statuses[1], bodies[1])
	if a.ID != b.ID {
		t.Fatalf("two runs for one key: %s and %s", a.ID, b.ID)
	}
	if a.replayed() == b.replayed() {
		t.Fatalf("want one fresh admission and one replay: %s / %s", bodies[0], bodies[1])
	}
	if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != a.ID {
		t.Fatalf("runs with objects %v, want only the winner %s: the loser's objects must be gone", ids, a.ID)
	}
	if n := fx.runStarts(t); n != 1 {
		t.Errorf("%d run-start rows, want 1", n)
	}
	if rec := fx.instRecord(t); rec.ActiveRun == nil || rec.ActiveRun.ID != a.ID || len(rec.RunKeys) != 1 {
		t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys)
	}
}

// lookupKey is GET …/runs?key=<key>.
func (fx *runFx) lookupKey(t *testing.T, bearer, key string) (int, string) {
	t.Helper()
	status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+url.QueryEscape(key), bearer, "")
	return status, body
}

func TestRunKeyOfAnotherCallerIsInvisible(t *testing.T) {
	fx := keyedFixture(t)
	runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
	person := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	if status, body := fx.lookupKey(t, runner, testKey); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
		t.Fatalf("the service looking up the person's key: %d %s, want 404", status, body)
	}
	fx.finishAndAck(t, fx.member, person.ID)
	svc := fx.createKeyed(t, runner, testKey, serviceMeta(), nil)
	if svc.ID == person.ID || svc.replayed() {
		t.Fatalf("the service's create with the person's key: %+v, want a fresh run", svc)
	}
	for _, c := range []struct {
		bearer, want string
	}{{fx.member, person.ID}, {runner, svc.ID}} {
		status, body := fx.lookupKey(t, c.bearer, testKey)
		var got struct {
			RunID string `json:"run_id"`
		}
		if status != http.StatusOK || json.Unmarshal([]byte(body), &got) != nil || got.RunID != c.want {
			t.Errorf("lookup: %d %s, want %s", status, body, c.want)
		}
	}
}

func TestRunKeyedHashMismatchWritesNothing(t *testing.T) {
	fx := keyedFixture(t)
	dir := spoolDir(t)
	status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,d"))
	if status != http.StatusBadRequest || codeOf(t, body) != runCodeFileHashMismatch {
		t.Fatalf("a file that is not its declared sha256: %d %s, want 400 %s", status, body, runCodeFileHashMismatch)
	}
	fx.noRunObjects(t, "a hash mismatch")
	assertNoSpool(t, dir, "a hash mismatch")
	if rec := fx.instRecord(t); len(rec.RunKeys) != 0 {
		t.Fatalf("run_keys %+v", rec.RunKeys)
	}

	// A keyed create must declare every file's sha256.
	undeclared := fileMeta("a,b,c")
	undeclared.Files[0].SHA256 = ""
	status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, undeclared, xcsv("a,b,c"))
	if status != http.StatusBadRequest || codeOf(t, body) != runCodeInvalidRequest {
		t.Fatalf("a keyed file with no sha256: %d %s, want 400 %s", status, body, runCodeInvalidRequest)
	}
	// A declared sha256 is checked whether or not the create is keyed.
	status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, "", fileMeta("a,b,c"), xcsv("a,b,d"))
	if status != http.StatusBadRequest || codeOf(t, body) != runCodeFileHashMismatch {
		t.Fatalf("an unkeyed mismatch: %d %s", status, body)
	}
	badShape := fileMeta("a,b,c")
	badShape.Files[0].SHA256 = "ABC"
	status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, "", badShape, xcsv("a,b,c"))
	if status != http.StatusBadRequest || codeOf(t, body) != runCodeInvalidRequest {
		t.Fatalf("a malformed sha256: %d %s", status, body)
	}
	fx.noRunObjects(t, "the hash refusals")
}

// failAfter is a request body that serves head and then fails every read.
type failAfter struct {
	head io.Reader
	hit  bool
}

func (f *failAfter) Read(p []byte) (int, error) {
	if n, err := f.head.Read(p); n > 0 || err != io.EOF {
		return n, err
	}
	f.hit = true
	return 0, errors.New("failAfter: the file part must not be read")
}

// serveBrokenBody sends a create whose body is the meta part and the start
// of a file part, after which every read fails, straight to the handler.
func (fx *runFx) serveBrokenBody(t *testing.T, bearer, key string, meta runs.Meta) *httptest.ResponseRecorder {
	t.Helper()
	var buf bytes.Buffer
	mw := multipart.NewWriter(&buf)
	mp, _ := mw.CreateFormField("meta")
	if err := json.NewEncoder(mp).Encode(meta); err != nil {
		t.Fatal(err)
	}
	fp, _ := mw.CreateFormFile("file", "x.csv")
	io.WriteString(fp, "a")
	req := httptest.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &failAfter{head: &buf})
	req.Header.Set("Content-Type", mw.FormDataContentType())
	req.Header.Set("Authorization", "Bearer "+bearer)
	if key != "" {
		req.Header.Set("Idempotency-Key", key)
	}
	w := httptest.NewRecorder()
	fx.srv.ServeHTTP(w, req)
	return w
}

func TestRunKnownKeyAnsweredWithoutReadingFiles(t *testing.T) {
	fx := keyedFixture(t)
	first := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
	w := fx.serveBrokenBody(t, fx.member, testKey, fileMeta("a,b,c"))
	again := createdOf(t, w.Code, w.Body.String())
	if again.ID != first.ID || !again.replayed() {
		t.Fatalf("the retry: %+v, want %s replayed", again, first.ID)
	}
	// The same broken body with a fresh key reaches the file part and fails.
	fx.finishAndAck(t, fx.member, first.ID)
	w = fx.serveBrokenBody(t, fx.member, "another-key", fileMeta("a,b,c"))
	if w.Code != http.StatusBadRequest || codeOf(t, w.Body.String()) != runCodeMalformed {
		t.Fatalf("a fresh key with a broken file part: %d %s, want 400 malformed", w.Code, w.Body.String())
	}
}

func TestRunRefusedKeyIsReevaluated(t *testing.T) {
	fx := keyedFixture(t)
	open := fx.startRun(t)
	status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, okMeta(), nil)
	if status != http.StatusConflict || codeOf(t, body) != runCodeRunOpen {
		t.Fatalf("a keyed create over an open run: %d %s, want 409 run_open", status, body)
	}
	fx.finishAndAck(t, fx.member, open)
	c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	if c.replayed() || c.ID == open {
		t.Fatalf("the retry once the other run was acked: %+v, want a fresh run", c)
	}
}

func TestRunKeyLookup(t *testing.T) {
	fx := keyedFixture(t)
	if status, body := fx.lookupKey(t, fx.member, testKey); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
		t.Fatalf("an unknown key: %d %s, want 404", status, body)
	}
	c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	status, body := fx.lookupKey(t, fx.member, testKey)
	if status != http.StatusOK || body != fmt.Sprintf("{\"run_id\":%q}\n", c.ID) {
		t.Fatalf("a known key: %d %s, want %s", status, body, c.ID)
	}
	for _, bad := range []string{"", "a b", strings.Repeat("k", 129)} {
		status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+url.QueryEscape(bad), fx.member, "")
		if status != http.StatusBadRequest || codeOf(t, body) != runCodeKeyInvalid {
			t.Errorf("key %q: %d %s, want 400 %s", bad, status, body, runCodeKeyInvalid)
		}
	}
	status, _, body = rawReq(t, http.MethodGet, fx.ts.URL+"/api/cli/apps/"+release.OpsAccount+"/shop/instances/nobody.example.com/runs?key="+testKey, fx.member, "")
	if status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
		t.Fatalf("an unknown host: %d %s", status, body)
	}
	// The shared gates hold here as at every run door.
	stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-key-stranger")
	status, h, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+testKey, stranger, "")
	assertNotFound(t, "a stranger's lookup", canonicalNotFound(t), status, h, body)
	fx.srv.cfg.HostClaim = true
	if status, body := fx.lookupKey(t, fx.member, testKey); status != http.StatusConflict || codeOf(t, body) != runCodeHostClaim {
		t.Fatalf("a lookup on a host-claim console: %d %s", status, body)
	}
}

// The key lookup is a run door: the run role reaches it, every other role
// gets the role gate's 403, Admin included.
func TestRunKeyLookupRoleMatrix(t *testing.T) {
	fx := keyedFixture(t)
	runner := runServiceToken(t, fx.st, "ci", "shop", importClassic)
	u := fx.runsURL(runTestHost) + "?key=" + testKey
	if status, _, body := rawReq(t, http.MethodGet, u, runner, ""); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound {
		t.Fatalf("a run service: %d %s, want the lookup's own 404", status, body)
	}
	for _, r := range svcrole.All() {
		if r == svcrole.Run {
			continue
		}
		app, rungs := "shop", []string(nil)
		if r == svcrole.Admin {
			app = ""
		}
		if r == svcrole.Deploy {
			rungs = []string{"edge"}
		}
		tok, err := fx.st.NewServicePrincipalScoped(OpsAccountID, "role-"+string(r), app, r, rungs, nil, "paul@keymail.dev", time.Now())
		if err != nil {
			t.Fatalf("mint %s: %v", r, err)
		}
		raw, _ := json.Marshal(map[string]string{"error": fmt.Sprintf("this credential holds role %q; this door needs one of [run]", r)})
		if status, _, body := rawReq(t, http.MethodGet, u, tok, ""); status != http.StatusForbidden || body != string(raw)+"\n" {
			t.Errorf("role %s: %d %q, want 403 %s", r, status, body, raw)
		}
	}
}

// TestRunUnkeyedOrderUnchanged: an unkeyed create over an open run is
// refused before its body is read, as before keys. A keyed one reads its
// meta first, so the same broken body is malformed.
func TestRunUnkeyedOrderUnchanged(t *testing.T) {
	fx := keyedFixture(t)
	fx.startRun(t)
	broken := func(key string) *httptest.ResponseRecorder {
		req := httptest.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &failAfter{head: strings.NewReader("")})
		req.Header.Set("Content-Type", "multipart/form-data; boundary=x")
		req.Header.Set("Authorization", "Bearer "+fx.member)
		if key != "" {
			req.Header.Set("Idempotency-Key", key)
		}
		w := httptest.NewRecorder()
		fx.srv.ServeHTTP(w, req)
		return w
	}
	if w := broken(""); w.Code != http.StatusConflict || codeOf(t, w.Body.String()) != runCodeRunOpen {
		t.Fatalf("unkeyed over an open run: %d %s, want 409 run_open before the body", w.Code, w.Body.String())
	}
	if w := broken(testKey); w.Code != http.StatusBadRequest || codeOf(t, w.Body.String()) != runCodeMalformed {
		t.Fatalf("keyed over an open run with a broken body: %d %s, want 400 malformed", w.Code, w.Body.String())
	}
}

// The keyed form of TestAPIRunCreateAdmissionWhoseAnswerWasLost: the retry
// inside updateRecord finds this request's own key, which is a fresh
// admission, and the run's objects stay.
func TestAPIRunCreateAdmissionWhoseAnswerWasLostKeyed(t *testing.T) {
	for name, failure := range map[string]error{
		"retried write":   fmt.Errorf("s3store: record: %w", release.ErrCASConflict),
		"transport error": errors.New("connection reset by peer"),
	} {
		t.Run(name, func(t *testing.T) {
			fx := keyedFixture(t)
			fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), err: failure}
			c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
			if c.replayed() || c.State != provision.RunQueued {
				t.Fatalf("created %+v, want a fresh admission", c)
			}
			rec := fx.instRecord(t)
			if rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != c.ID {
				t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys)
			}
			keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID))
			if len(keys) != 2 {
				t.Fatalf("the admitted run's objects are %q, want request.json and in/x.csv", keys)
			}
			if n := fx.runStarts(t); n != 1 {
				t.Errorf("%d run-start rows, want 1", n)
			}
		})
	}
}

// admitRun itself answers the retry inside updateRecord that finds this
// request's own key: admitted, with the record's state, and no error for the
// handler to settle.
func TestAdmitRunOwnKeyReplayIsAdmitted(t *testing.T) {
	fx := keyedFixture(t)
	fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost),
		err: fmt.Errorf("s3store: record: %w", release.ErrCASConflict)}
	id, err := runs.MintID()
	if err != nil {
		t.Fatal(err)
	}
	state, err := fx.srv.admitRun(context.Background(), release.OpsAccount, "shop", runTestHost, id, "alice@keymail.dev", "K", "D", "C")
	if err != nil || state != provision.RunQueued {
		t.Fatalf("admitRun = %q, %v; want queued and no error", state, err)
	}
}

func TestKeyedCreateRefusedUntilKeysAreOn(t *testing.T) {
	fx := runFixture(t)
	status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
	if status != http.StatusServiceUnavailable || codeOf(t, body) != runCodeKeysNotReady || errorOf(t, body) == "" {
		t.Fatalf("a keyed create with keys off: %d %s, want 503 %s", status, body, runCodeKeysNotReady)
	}
	fx.noRunObjects(t, "a keyed create with keys off")
	if rec := fx.instRecord(t); len(rec.RunKeys) != 0 {
		t.Fatalf("run_keys %+v", rec.RunKeys)
	}
	// Even an invalid key is refused as not ready, never run unkeyed.
	if status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, "a b", okMeta(), nil); status != http.StatusServiceUnavailable {
		t.Fatalf("an invalid key with keys off: %d %s", status, body)
	}
	c := fx.createKeyed(t, fx.member, "", okMeta(), nil)
	if c.replayed() {
		t.Fatalf("an unkeyed create: %+v", c)
	}
	if req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", c.ID); err != nil || req.IdempotencyKey != "" {
		t.Fatalf("an unkeyed request: %+v, %v", req, err)
	}
	if rec := fx.instRecord(t); len(rec.RunKeys) != 0 {
		t.Fatalf("an unkeyed create recorded a key: %+v", rec.RunKeys)
	}
	// With keys on, an invalid key is refused by its shape.
	fx.srv.cfg.RunKeysEnabled = true
	if status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, "a/b", okMeta(), nil); status != http.StatusBadRequest || codeOf(t, body) != runCodeKeyInvalid {
		t.Fatalf("an invalid key: %d %s, want 400 %s", status, body, runCodeKeyInvalid)
	}
}

// The codes this task adds are in the one list.
func TestRunKeyCodesAreListed(t *testing.T) {
	for _, c := range []string{runCodeKeyReused, runCodeKeyInvalid, runCodeKeysNotReady, runCodeFileHashMismatch} {
		if !slices.Contains(runRefusalCodes, c) {
			t.Errorf("%s is not in runRefusalCodes", c)
		}
	}
	for c, want := range map[string]string{
		runCodeKeyReused: "idempotency_key_reused", runCodeKeyInvalid: "idempotency_key_invalid",
		runCodeKeysNotReady: "keys_not_ready", runCodeFileHashMismatch: "file_hash_mismatch",
	} {
		if c != want {
			t.Errorf("code %s, want %s", c, want)
		}
	}
}

// movedOnStore commits the admission's record write, then has the run picked,
// finished and cleared (a box and an ack, faster than the answer), and
// answers err: the record's ActiveRun no longer names the run when the door
// settles the lost answer.
type movedOnStore struct {
	release.Store
	recordKey string
	err       error
	once      sync.Once
	t         *testing.T
}

func (m *movedOnStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := m.Store.Put(ctx, key, body, ifMatch)
	if err != nil || key != m.recordKey {
		return etag, err
	}
	m.once.Do(func() {
		rec, gerr := provision.DecodeRecord(body)
		if gerr != nil || rec.ActiveRun == nil {
			m.t.Errorf("movedOnStore: %v %+v", gerr, rec.ActiveRun)
			return
		}
		id := rec.ActiveRun.ID
		for _, mut := range []func(*provision.Record) error{
			provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)),
			provision.RunFinish(id),
			provision.RunClear(id),
		} {
			if _, e := provision.SetRecordRun(context.Background(), m.Store, release.OpsAccount, "shop", runTestHost, mut); e != nil {
				m.t.Errorf("movedOnStore: %v", e)
			}
		}
		err = m.err
	})
	return etag, err
}

// A keyed admission whose answer was lost is settled by its key, not by
// ActiveRun: the record has moved on, the key still names this run, so the
// door answers it admitted and keeps its objects.
func TestRunKeyedLostAdmissionSettledByItsKey(t *testing.T) {
	fx := keyedFixture(t)
	fx.srv.cfg.Releases = &movedOnStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost),
		err: errors.New("connection reset by peer"), t: t}
	c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
	if c.replayed() || c.State != "" {
		t.Fatalf("created %+v, want a fresh admission whose run the record no longer names", c)
	}
	rec := fx.instRecord(t)
	if rec.ActiveRun != nil || len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != c.ID {
		t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys)
	}
	if keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID)); len(keys) != 2 {
		t.Fatalf("the committed run's objects are %q, want request.json and in/x.csv", keys)
	}
	if n := fx.runStarts(t); n != 1 {
		t.Errorf("%d run-start rows, want 1", n)
	}
}

// keysDroppedStore commits the admission's record write, then has a record
// writer drop run_keys (a box rolled back to code that does not know them)
// while the record still holds the run, and answers err.
type keysDroppedStore struct {
	release.Store
	recordKey string
	err       error
	once      sync.Once
	t         *testing.T
}

func (m *keysDroppedStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := m.Store.Put(ctx, key, body, ifMatch)
	if err != nil || key != m.recordKey {
		return etag, err
	}
	m.once.Do(func() {
		drop := func(r *provision.Record) error { r.RunKeys = nil; return nil }
		if _, e := provision.SetRecordRun(context.Background(), m.Store, release.OpsAccount, "shop", runTestHost, drop); e != nil {
			m.t.Errorf("keysDroppedStore: %v", e)
		}
	})
	return etag, m.err
}

// A keyed admission whose answer was lost, and whose key is gone from the
// record while the record still names the run, is admitted: the objects stay
// and the answer names the run, as the unkeyed settle would have.
func TestRunKeyedSettleKeepsARunTheRecordHoldsWhenItsKeyIsGone(t *testing.T) {
	fx := keyedFixture(t)
	fx.srv.cfg.Releases = &keysDroppedStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost),
		err: errors.New("connection reset by peer"), t: t}
	c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
	rec := fx.instRecord(t)
	if rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || len(rec.RunKeys) != 0 {
		t.Fatalf("answer %+v; record run %+v keys %+v, want the run held with no key", c, rec.ActiveRun, rec.RunKeys)
	}
	if c.State != provision.RunQueued {
		t.Errorf("answered %+v, want the record's queued state", c)
	}
	if keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID)); len(keys) != 2 {
		t.Fatalf("the held run's objects are %q, want request.json and in/x.csv", keys)
	}
}

// twinOnRetryStore loses both of updateRecord's attempts: the first record
// write is refused as a conflict without landing, and before the second a
// twin's keyed admission lands, so the second conflicts for real.
type twinOnRetryStore struct {
	release.Store
	recordKey string
	twin      provision.RunAdmission
	puts      int
	t         *testing.T
}

func (s *twinOnRetryStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.recordKey {
		s.puts++
		switch s.puts {
		case 1:
			return "", fmt.Errorf("s3store: record: %w", release.ErrCASConflict)
		case 2:
			if _, err := provision.SetRecordRun(context.Background(), s.Store, release.OpsAccount, "shop", runTestHost, provision.RunAdmit(s.twin)); err != nil {
				s.t.Errorf("twin admission: %v", err)
			}
		}
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// A keyed create that loses both CAS attempts to a twin answers with the
// twin's run, replayed, and deletes only its own objects: not a 502.
func TestRunKeyedCreateLosingBothAttemptsAnswersTheTwin(t *testing.T) {
	fx := keyedFixture(t)
	scope := "user:alice@keymail.dev"
	twinID, err := runs.MintID()
	if err != nil {
		t.Fatal(err)
	}
	st := &twinOnRetryStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), t: t,
		twin: provision.RunAdmission{ID: twinID, By: "alice@keymail.dev", Deadline: time.Now().Add(time.Minute),
			Key: runs.ScopedKey(scope, testKey), Digest: runs.MetaDigest(okMeta()), Caller: runs.CallerOf(scope)}}
	fx.srv.cfg.Releases = st
	c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
	if st.puts != 2 {
		t.Fatalf("%d record writes, want the two lost attempts", st.puts)
	}
	if c.ID != twinID || !c.replayed() || c.State != provision.RunQueued {
		t.Fatalf("created %+v, want the twin %s, replayed and queued", c, twinID)
	}
	if ids := fx.runIDsWithObjects(t); len(ids) != 0 {
		t.Fatalf("runs with objects %v: the loser's objects must be gone", ids)
	}
	if n := fx.runStarts(t); n != 0 {
		t.Errorf("%d run-start rows, want none from the loser", n)
	}
}

// A declared sha256 may be in either case: it is lowercased before it is
// compared or digested, so an uppercase declaration is not refused and a
// retry in the other case is the same request.
func TestRunDeclaredHashIsCaseInsensitive(t *testing.T) {
	fx := keyedFixture(t)
	upper := fileMeta("a,b,c")
	upper.Files[0].SHA256 = strings.ToUpper(upper.Files[0].SHA256)
	plain := fx.createKeyed(t, fx.member, "", upper, xcsv("a,b,c"))
	fx.finishAndAck(t, fx.member, plain.ID)
	first := fx.createKeyed(t, fx.member, testKey, upper, xcsv("a,b,c"))
	again := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
	if again.ID != first.ID || !again.replayed() {
		t.Fatalf("a lowercase retry of an uppercase create: %+v, want %s replayed", again, first.ID)
	}
	req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", first.ID)
	if err != nil || req.Files[0].SHA256 != sha("a,b,c") {
		t.Fatalf("request %+v, %v", req, err)
	}
}