package console

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"mime/multipart"
	"net/http"
	"net/http/httptest"
	"os"
	"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/seeds"
	"github.com/carlosframework/platform/internal/svcrole"
)

const runTestHost = "acme.example.com"

type runFx struct {
	*consoleFixture
	member string
}

func runFixture(t *testing.T) *runFx {
	t.Helper()
	fx := newConsoleFixture(t)
	now := time.Now()
	if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", now); err != nil {
		t.Fatal(err)
	}
	if err := fx.st.AddMember(OpsAccountID, "alice@keymail.dev", RoleMember, now); err != nil {
		t.Fatal(err)
	}
	member := cliToken(t, fx, "alice@keymail.dev", "code-run")
	putShopRecord(t, fx, provision.Record{Host: runTestHost, Channel: "stable", CreatedAt: time.Now().UTC()})
	return &runFx{consoleFixture: fx, member: member}
}

// putShopRecord writes an instance record for the shop app with the fields
// a hibernating exec instance has, overridden by whatever rec sets.
func putShopRecord(t *testing.T, fx *consoleFixture, rec provision.Record) {
	t.Helper()
	rec.App, rec.Account = "shop", release.OpsAccount
	if rec.Hibernate == nil {
		yes := true
		rec.Hibernate = &yes
	}
	body, err := json.Marshal(rec)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(context.Background(), provision.RecordKey(rec.Account, rec.App, rec.Host), body, ""); err != nil {
		t.Fatal(err)
	}
}

func (fx *runFx) runsURL(host string) string {
	return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/instances/" + host + "/runs"
}

// postRun sends the multipart create; parts lets a test send parts that
// disagree with meta.
func postRun(t *testing.T, u, bearer string, meta runs.Meta, parts []struct{ name, body string }) (int, string) {
	t.Helper()
	status, _, body := postRunH(t, u, bearer, meta, parts)
	return status, body
}

// postRunH is postRun that also returns the response headers.
func postRunH(t *testing.T, u, bearer string, meta runs.Meta, parts []struct{ name, body string }) (int, http.Header, 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())
	if bearer != "" {
		req.Header.Set("Authorization", "Bearer "+bearer)
	}
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer res.Body.Close()
	out, _ := io.ReadAll(res.Body)
	return res.StatusCode, res.Header, string(out)
}

func okMeta() runs.Meta { return runs.Meta{Argv: []string{"import", "x.csv"}, TimeoutS: 60} }

func errorOf(t *testing.T, body string) string {
	t.Helper()
	var e struct{ Error string }
	if err := json.Unmarshal([]byte(body), &e); err != nil {
		t.Fatalf("body %q: %v", body, err)
	}
	return e.Error
}

// spoolDir points the door's spool files at a directory of the test's own
// and returns it, so a test can see that none is left behind.
func spoolDir(t *testing.T) string {
	t.Helper()
	dir := t.TempDir()
	t.Setenv("TMPDIR", dir)
	return dir
}

func assertNoSpool(t *testing.T, dir, when string) {
	t.Helper()
	if left, _ := os.ReadDir(dir); len(left) != 0 {
		t.Errorf("%s left %d spool file(s) behind", when, len(left))
	}
}

func TestAPIRunCreateAdmitsAndRecords(t *testing.T) {
	fx := runFixture(t)
	spool := spoolDir(t)
	meta := okMeta()
	meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
	status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}})
	if status != http.StatusAccepted {
		t.Fatalf("status %d %s", status, body)
	}
	assertNoSpool(t, spool, "an admitted run")
	var c runs.Created
	if err := json.Unmarshal([]byte(body), &c); err != nil || !runs.ValidID(c.ID) || c.State != provision.RunQueued {
		t.Fatalf("created %+v, %v", c, err)
	}
	ctx := context.Background()
	rec, err := provision.GetRecord(ctx, fx.rel, release.OpsAccount, "shop", runTestHost)
	if err != nil || rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || rec.ActiveRun.By != "alice@keymail.dev" {
		t.Fatalf("record run %+v, %v", rec.ActiveRun, err)
	}
	if d := time.Until(rec.ActiveRun.Deadline); d < 50*time.Second || d > 61*time.Second {
		t.Errorf("pickup deadline %s away, want about 60 s", d)
	}
	req, err := runs.ReadRequest(ctx, fx.rel, release.OpsAccount, "shop", c.ID)
	if err != nil || req.Host != runTestHost || len(req.Incarnation) != 64 || req.TimeoutS != 60 || len(req.Files) != 1 || len(req.Files[0].SHA256) != 64 {
		t.Fatalf("request %+v, %v", req, err)
	}
	raw, _, err := fx.rel.Get(ctx, runs.InputKey(release.OpsAccount, "shop", c.ID, "x.csv"))
	if err != nil || string(raw) != "a,b,c" {
		t.Fatalf("input %q, %v", raw, err)
	}
	events, _ := fx.st.Events(OpsAccountID, "shop", 10)
	found := false
	for _, e := range events {
		if e.Action == "run-start" && e.Target == runTestHost && e.Address == "alice@keymail.dev" && strings.Contains(e.Detail, c.ID) && strings.Contains(e.Detail, "x.csv") {
			found = true
		}
	}
	if !found {
		t.Errorf("no run-start audit row in %+v", events)
	}
}

// Each refusal in spec §5.1's order: every case is valid in every later
// respect, so the status and message show which check refused it.
func TestAPIRunCreateRefusesInOrder(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	svc := serviceToken(t, fx.st, OpsAccountID, "ci", "shop", svcrole.Instance)
	stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-stranger")

	if status, _ := postRun(t, fx.runsURL(runTestHost), svc, okMeta(), nil); status != http.StatusForbidden {
		t.Errorf("service credential: %d, want 403", status)
	}
	status, h, body := postRunH(t, fx.runsURL(runTestHost), stranger, okMeta(), nil)
	assertNotFound(t, "stranger", canonicalNotFound(t), status, h, body)
	if status, body := postRun(t, fx.runsURL("nobody.example.com"), fx.member, okMeta(), nil); status != http.StatusNotFound || errorOf(t, body) != "nobody.example.com is not an instance of shop" {
		t.Errorf("no record: %d %s", status, body)
	}
	for host, c := range map[string]struct {
		rec provision.Record
		msg string
	}{
		"static.example.com": {provision.Record{Kind: "static"}, "static.example.com is a static route. carlos run works on instances"},
		"addr.example.com":   {provision.Record{Addr: "127.0.0.1:9"}, "addr.example.com is served from its own address, so the platform does not run its process"},
		"pool.example.com":   {provision.Record{Pool: "shared"}, "pool.example.com is a pooled instance. carlos run works on instances with a process of their own"},
		"fleet.example.com":  {provision.Record{Region: "fleet/pi"}, "fleet.example.com runs on your fleet pi. carlos run works on platform boxes only"},
	} {
		c.rec.Host = host
		putShopRecord(t, fx.consoleFixture, c.rec)
		if status, body := postRun(t, fx.runsURL(host), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != c.msg {
			t.Errorf("%s: %d %s", host, status, body)
		}
	}
	no := false
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "awake.example.com", Hibernate: &no})
	if status, body := postRun(t, fx.runsURL("awake.example.com"), fx.member, okMeta(), nil); status != http.StatusConflict || !strings.Contains(errorOf(t, body), "does not sleep") {
		t.Errorf("always-on: %d %s", status, body)
	}
	if _, err := provision.SetRecordMove(ctx, fx.rel, release.OpsAccount, "shop", runTestHost, provision.MoveRequest("uk", time.Now())); err != nil {
		t.Fatal(err)
	}
	if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != runTestHost+" is moving. run the command once the move has finished" {
		t.Errorf("moving: %d %s", status, body)
	}
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "busy.example.com", ActiveRun: &provision.RunState{ID: "run_aaaaaaaaaaaaaaaaaaaaaaaaaa", State: provision.RunPicked, By: "bob@keymail.dev"}})
	if status, body := postRun(t, fx.runsURL("busy.example.com"), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != "busy.example.com already has a run open: run_aaaaaaaaaaaaaaaaaaaaaaaaaa, started by bob@keymail.dev" {
		t.Errorf("open run: %d %s", status, body)
	}
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "fresh.example.com"})
	bad := []struct {
		meta runs.Meta
		msg  string
	}{
		{runs.Meta{TimeoutS: 60}, "name the command to run after --"},
		{runs.Meta{Argv: make([]string, 257), TimeoutS: 60}, "a run takes at most 256 arguments"},
		{runs.Meta{Argv: []string{strings.Repeat("a", 64<<10+1)}, TimeoutS: 60}, "the arguments come to more than 64 KB"},
		{runs.Meta{Argv: []string{"a\x00"}, TimeoutS: 60}, "an argument holds a NUL byte"},
		{runs.Meta{Argv: []string{"a"}, TimeoutS: 6*3600 + 1}, "the timeout must be between 1 second and 6 hours"},
		{runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "a", Size: 65 << 20}}}, "the files come to more than 64 MB"},
		{runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "../x", Size: 1}}}, "../x is not a plain file name"},
		{runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "a", Size: 1}, {Name: "a", Size: 1}}}, "two files are both named a"},
	}
	for i := range bad[1].meta.Argv {
		bad[1].meta.Argv[i] = "x"
	}
	for _, c := range bad {
		if status, body := postRun(t, fx.runsURL("fresh.example.com"), fx.member, c.meta, nil); status != http.StatusBadRequest || errorOf(t, body) != c.msg {
			t.Errorf("%q: %d %s", c.msg, status, body)
		}
	}
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "import.example.com", Seed: &provision.SeedSource{Kind: seeds.KindImport, State: seeds.StatePlaced}})
	if status, body := postRun(t, fx.runsURL("import.example.com"), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != "import.example.com is an import that is not committed yet. finish or abort the import first" {
		t.Errorf("import: %d %s", status, body)
	}
}

func TestAPIRunCreateSpendsTheInstanceBudget(t *testing.T) {
	fx := runFixture(t)
	// A window of one: the first hit in a window is always allowed
	// (rateLimiter.allow, authcli.go:226), the second is over. The budget
	// is refusal 3, so it comes before refusal 7 (the run the first opened).
	fx.srv.cli.instances = newRateLimiter(1, cliRateWindow)
	if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusAccepted {
		t.Fatalf("first: %d %s", status, body)
	}
	if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusTooManyRequests || errorOf(t, body) != "too many run requests. wait a minute and try again" {
		t.Fatalf("over budget: %d %s", status, body)
	}
}

func TestAPIRunCreateRefusesAPartThatDisagreesWithMeta(t *testing.T) {
	fx := runFixture(t)
	spool := spoolDir(t)
	meta := okMeta()
	meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
	for name, parts := range map[string][]struct{ name, body string }{
		"short":      {{"x.csv", "abc"}},
		"long":       {{"x.csv", "abcdefg"}},
		"wrong name": {{"y.csv", "abcde"}},
		"missing":    nil,
		"extra":      {{"x.csv", "abcde"}, {"y.csv", "abcde"}},
	} {
		status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, parts)
		if status != http.StatusBadRequest || errorOf(t, body) != "the run request is malformed" {
			t.Errorf("%s: %d %s", name, status, body)
		}
		assertNoSpool(t, spool, name)
	}
	if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "shop")); len(left) != 0 {
		t.Fatalf("a refused upload left %q", left)
	}
	if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost); rec.ActiveRun != nil {
		t.Fatal("a refused upload admitted a run")
	}
}

// Two members at once: one run is admitted, the other is told who holds the
// instance, and its objects are gone (codex R1-1).
func TestAPIRunCreateRaceAdmitsOne(t *testing.T) {
	fx := runFixture(t)
	var wg sync.WaitGroup
	statuses := make([]int, 2)
	bodies := make([]string, 2)
	for i := range 2 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			statuses[i], bodies[i] = postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil)
		}()
	}
	wg.Wait()
	won, lost := 0, 0
	for i := range 2 {
		switch statuses[i] {
		case http.StatusAccepted:
			won++
		case http.StatusConflict:
			lost++
			if !strings.Contains(errorOf(t, bodies[i]), "already has a run open") {
				t.Errorf("loser said %s", bodies[i])
			}
		}
	}
	if won != 1 || lost != 1 {
		t.Fatalf("statuses %v %v", statuses, bodies)
	}
	// DirStore keeps emptied directories, so count the runs that still own
	// objects rather than the directories.
	prefix := runs.RunsPrefix(release.OpsAccount, "shop")
	keys, _ := fx.rel.List(context.Background(), prefix)
	owners := map[string]bool{}
	for _, k := range keys {
		owners[strings.SplitN(strings.TrimPrefix(k, prefix), "/", 2)[0]] = true
	}
	if len(owners) != 1 {
		t.Fatalf("runs with objects %v, want only the winner's", owners)
	}
}

// rivalStore admits another run on the instance record the moment the door
// writes request.json: after the door's own checks passed and its objects
// began to land, before its admission. That is the race, made certain.
type rivalStore struct {
	release.Store
	t    *testing.T
	once sync.Once
}

func (r *rivalStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if strings.HasSuffix(key, "/request.json") {
		r.once.Do(func() {
			_, err := provision.SetRecordRun(ctx, r.Store, release.OpsAccount, "shop", runTestHost,
				provision.RunAdmit(provision.RunAdmission{ID: "run_aaaaaaaaaaaaaaaaaaaaaaaaaa", By: "bob@keymail.dev", Deadline: time.Now().Add(time.Minute)}))
			if err != nil {
				r.t.Error(err)
			}
		})
	}
	return r.Store.Put(ctx, key, body, ifMatch)
}

// The loser of an admission that happens after its objects began to land is
// told who holds the instance, and nothing of its own is left in the bucket.
func TestAPIRunCreateLoserAfterUploadLeavesNothing(t *testing.T) {
	fx := runFixture(t)
	fx.srv.cfg.Releases = &rivalStore{Store: fx.rel, t: t}
	meta := okMeta()
	meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
	status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}})
	if want := "already has a run open: run_aaaaaaaaaaaaaaaaaaaaaaaaaa, started by bob@keymail.dev"; status != http.StatusConflict || !strings.HasSuffix(errorOf(t, body), want) {
		t.Fatalf("%d %s", status, body)
	}
	if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "shop")); len(left) != 0 {
		t.Fatalf("the loser left %q", left)
	}
}

func TestAPIRunCreateOnAHostClaimConsole(t *testing.T) {
	fx := runFixture(t)
	fx.srv.cfg.HostClaim = true
	if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg {
		t.Fatalf("%d %s", status, body)
	}
	// RUN-C42 comes right after membership: a stranger still gets the
	// byte-identical 404, and a member with no budget left gets RUN-C42.
	stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-stranger")
	status, h, body := postRunH(t, fx.runsURL(runTestHost), stranger, okMeta(), nil)
	assertNotFound(t, "stranger on a claim console", canonicalNotFound(t), status, h, body)
	fx.srv.cli.instances = newRateLimiter(1, cliRateWindow)
	fx.srv.cli.instances.allow(hashToken(fx.member), time.Now())
	if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg {
		t.Fatalf("over budget on a claim console: %d %s", status, body)
	}
	fx.srv.cfg.HostClaim = false
	if status, _ := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusTooManyRequests {
		t.Fatalf("the same member off a claim console: %d, want the 429 the claim console skipped", status)
	}
}

// The body cap (file bytes plus the meta allowance) ends a stream that grows
// past it with the 64 MB refusal: here a full-size declared file behind a
// meta part padded past the allowance.
func TestAPIRunCreateRefusesABodyPastTheCap(t *testing.T) {
	fx := runFixture(t)
	spool := spoolDir(t)
	meta := okMeta()
	meta.Files = []runs.MetaFile{{Name: "big", Size: runs.MaxFileBytes}}
	var mb bytes.Buffer
	json.NewEncoder(&mb).Encode(meta)
	pad := strings.Repeat(" ", runs.MetaAllowance) // trailing white space: valid JSON, counted by the cap
	var buf bytes.Buffer
	mw := multipart.NewWriter(&buf)
	mp, _ := mw.CreateFormField("meta")
	mp.Write(mb.Bytes())
	mp.Write([]byte(pad))
	fp, _ := mw.CreateFormFile("file", "big")
	fp.Write(make([]byte, runs.MaxFileBytes))
	mw.Close()
	if buf.Len() <= runs.MaxFileBytes+runs.MetaAllowance {
		t.Fatalf("body is %d bytes, under the cap; the test needs a longer pad", buf.Len())
	}
	req, _ := http.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &buf)
	req.Header.Set("Content-Type", mw.FormDataContentType())
	req.Header.Set("Authorization", "Bearer "+fx.member)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	out, _ := io.ReadAll(res.Body)
	res.Body.Close()
	if res.StatusCode != http.StatusRequestEntityTooLarge || errorOf(t, string(out)) != "the files come to more than 64 MB" {
		t.Fatalf("%d %s", res.StatusCode, out)
	}
	assertNoSpool(t, spool, "a body past the cap")
	if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "shop")); len(left) != 0 {
		t.Fatalf("left %q", left)
	}
}

// commitThenFailStore commits the first write to the instance record and then
// answers err, as a lost response or a cancelled request would: the record
// holds the run, the caller was told nothing.
type commitThenFailStore struct {
	release.Store
	recordKey string
	err       error
	once      sync.Once
}

func (c *commitThenFailStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := c.Store.Put(ctx, key, body, ifMatch)
	if err == nil && key == c.recordKey {
		c.once.Do(func() { err = c.err })
	}
	return etag, err
}

// An admission whose answer was lost is still an admission: the door says
// 202 and keeps the run's objects, because the record holds the run and a
// box will read them.
func TestAPIRunCreateAdmissionWhoseAnswerWasLost(t *testing.T) {
	for name, failure := range map[string]error{
		// The SDK re-sent a committed write: 412, then updateRecord's retry
		// finds this very run queued.
		"retried write": fmt.Errorf("s3store: record: %w", release.ErrCASConflict),
		// A transport error, or the request's context cancelled, after the
		// write landed: the door asks the record.
		"transport error": errors.New("connection reset by peer"),
	} {
		t.Run(name, func(t *testing.T) {
			fx := runFixture(t)
			fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), err: failure}
			meta := okMeta()
			meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
			status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}})
			if status != http.StatusAccepted {
				t.Fatalf("%d %s", status, body)
			}
			var c runs.Created
			if err := json.Unmarshal([]byte(body), &c); err != nil || c.State != provision.RunQueued {
				t.Fatalf("created %+v, %v", c, err)
			}
			ctx := context.Background()
			rec, err := provision.GetRecord(ctx, fx.rel, release.OpsAccount, "shop", runTestHost)
			if err != nil || rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID {
				t.Fatalf("record run %+v, %v", rec.ActiveRun, err)
			}
			keys, _ := fx.rel.List(ctx, 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)
			}
		})
	}
}

// leavingStore answers every call whose context has ended with that
// context's error, as a real bucket client does, and ends the caller's
// request once the admission's record write has landed: the client went
// away right after its run was admitted.
type leavingStore struct {
	release.Store
	recordKey string
	leave     context.CancelFunc
}

func (l *leavingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if err := ctx.Err(); err != nil {
		return nil, "", err
	}
	return l.Store.Get(ctx, key)
}

func (l *leavingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if err := ctx.Err(); err != nil {
		return "", err
	}
	etag, err := l.Store.Put(ctx, key, body, ifMatch)
	if err == nil && key == l.recordKey {
		l.leave()
	}
	return etag, err
}

// A run admitted for a client that then went away still bumps the serial,
// so a box picks it up at once, and its run-start row still names the
// version the instance follows: both run on the admission's own context.
func TestAPIRunCreateFinishesTheAdmissionForAClientThatLeft(t *testing.T) {
	fx := runFixture(t)
	shipVersion(t, fx.consoleFixture, "shop", "v1")
	seedChannel(t, fx.consoleFixture, "shop", "stable", "v1")
	before, _, _ := release.ReadSerial(context.Background(), fx.rel)

	var buf bytes.Buffer
	mw := multipart.NewWriter(&buf)
	mp, _ := mw.CreateFormField("meta")
	if err := json.NewEncoder(mp).Encode(okMeta()); err != nil {
		t.Fatal(err)
	}
	mw.Close()
	ctx, leave := context.WithCancel(context.Background())
	defer leave()
	fx.srv.cfg.Releases = &leavingStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), leave: leave}
	req := httptest.NewRequestWithContext(ctx, http.MethodPost, fx.runsURL(runTestHost), &buf)
	req.Header.Set("Content-Type", mw.FormDataContentType())
	req.Header.Set("Authorization", "Bearer "+fx.member)
	w := httptest.NewRecorder()
	fx.srv.ServeHTTP(w, req)
	if w.Code != http.StatusAccepted {
		t.Fatalf("%d %s", w.Code, w.Body.String())
	}
	if ctx.Err() == nil {
		t.Fatal("the client never left: the test proves nothing")
	}
	if after, _, err := release.ReadSerial(context.Background(), fx.rel); err != nil || after.Serial <= before.Serial {
		t.Errorf("serial %d -> %d, %v: not bumped", before.Serial, after.Serial, err)
	}
	events, _ := fx.st.Events(OpsAccountID, "shop", 10)
	found := false
	for _, e := range events {
		if e.Action == "run-start" && strings.Contains(e.Detail, `"version":"v1"`) {
			found = true
		}
	}
	if !found {
		t.Errorf("no run-start row naming v1 in %+v", events)
	}
}