package console

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"mime/multipart"
	"net/http"
	netmail "net/mail"
	"os"
	"time"

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

// The run handlers' refusal codes (service runs spec §3): every refusal a
// run handler writes carries one beside its sentence, so a caller with no
// person reading the answer can act on it. The shared gates before the
// handler (not signed in, the role mismatch, the membership 404) keep their
// bodies exactly as every other door gives them.
const (
	runCodeNotFound          = "not_found"
	runCodeHostClaim         = "host_claim"
	runCodeRateLimited       = "rate_limited"
	runCodeUnavailable       = "unavailable"
	runCodeRecordFailed      = "record_failed"
	runCodeReadFailed        = "read_failed"
	runCodeNotRunnable       = "not_runnable"
	runCodeMoving            = "moving"
	runCodeRunOpen           = "run_open"
	runCodeMalformed         = "malformed"
	runCodeInvalidRequest    = "invalid_request"
	runCodeFilesTooLarge     = "files_too_large"
	runCodeImportUncommitted = "import_uncommitted"
	runCodeCommandNotGranted = "command_not_granted"
	runCodeOnBehalfOfInvalid = "on_behalf_of_invalid"
	runCodeRunStarted        = "run_started"
	runCodeRunEnded          = "run_ended"
	runCodeRunNotFinished    = "run_not_finished"
	runCodeLastSeqMismatch   = "last_seq_mismatch"
	runCodeKeyInvalid        = "idempotency_key_invalid"
	runCodeKeyReused         = "idempotency_key_reused"
	runCodeKeysNotReady      = "keys_not_ready"
	runCodeFileHashMismatch  = "file_hash_mismatch"
)

// runRefusalCodes is every code above, the one list the docs and the tests
// read.
var runRefusalCodes = []string{
	runCodeNotFound, runCodeHostClaim, runCodeRateLimited, runCodeUnavailable,
	runCodeRecordFailed, runCodeReadFailed, runCodeNotRunnable, runCodeMoving,
	runCodeRunOpen, runCodeMalformed, runCodeInvalidRequest, runCodeFilesTooLarge,
	runCodeImportUncommitted, runCodeCommandNotGranted, runCodeOnBehalfOfInvalid,
	runCodeRunStarted, runCodeRunEnded, runCodeRunNotFinished, runCodeLastSeqMismatch,
	runCodeKeyInvalid, runCodeKeyReused, runCodeKeysNotReady, runCodeFileHashMismatch,
}

// runCaller reports whether p may use the run doors at all: a person (whose
// membership the handler checks next), or a run service with a credential
// id to own its runs by.
func runCaller(p Principal) bool {
	switch p.Kind {
	case PrincipalUser:
		return true
	case PrincipalService:
		return p.Role == svcrole.Run && p.CredentialID != ""
	}
	return false
}

// callerScope names a caller for every comparison between callers: the
// principal's KIND and its stable identity, so a person's address can never
// equal a credential's id. "" for a principal with no identity, which
// matches nothing.
func callerScope(p Principal) string {
	switch {
	case p.Kind == PrincipalService && p.CredentialID != "":
		return "svc:" + p.CredentialID
	case p.Kind == PrincipalUser && p.Address != "":
		return "user:" + p.Address
	}
	return ""
}

// runVisibleTo reports whether p may see and act on req. A person sees every
// run on an account they are a member of (the gate has checked that); a
// service sees only the runs its own credential started, by the credential's
// id, never its name, so a name minted again after a revoke inherits
// nothing.
func (s *Server) runVisibleTo(p Principal, req runs.Request) bool {
	switch p.Kind {
	case PrincipalUser:
		return true
	case PrincipalService:
		return req.ByCredential != "" && callerScope(p) == "svc:"+req.ByCredential
	}
	return false
}

// maxOnBehalfOf is the longest on_behalf_of an address may be.
const maxOnBehalfOf = 254

const onBehalfOfPersonMsg = "on_behalf_of is for service credentials. a person runs as themselves"

const onBehalfOfShapeMsg = "on_behalf_of must be an email address of at most 254 bytes"

// onBehalfOfRefusal is "" when meta's on_behalf_of may stand: empty, or an
// address of the right shape sent by a run service. Its shape is checked,
// never its membership: it is recorded on the service's word and grants
// nothing.
func onBehalfOfRefusal(p Principal, v string) string {
	switch {
	case v == "":
		return ""
	case p.Kind != PrincipalService:
		return onBehalfOfPersonMsg
	case len(v) > maxOnBehalfOf || !ValidAddress(v):
		return onBehalfOfShapeMsg
	}
	if parsed, err := netmail.ParseAddress(v); err != nil || parsed.Address != v || parsed.Name != "" {
		return onBehalfOfShapeMsg
	}
	return ""
}

// claimRunUnsupportedMsg is every run route's answer on a host-claim
// console (index item 12, RUN-C42).
const claimRunUnsupportedMsg = "carlos run is not available on this console yet. it needs the host-claim launch protocol"

const runTooManyMsg = "too many run requests. wait a minute and try again"

// runOpenMsg is refusal 7 (RUN-C24), at the door and at admission. req is
// the open run's request when it could be read: a service's run is named in
// the house wording (runs.Request.ServiceBy), never by its audit address.
func runOpenMsg(host string, run provision.RunState, req *runs.Request) string {
	if req != nil {
		if by := req.ServiceBy(); by != "" {
			return fmt.Sprintf("%s already has a run open: %s, started %s", host, run.ID, by)
		}
	}
	return fmt.Sprintf("%s already has a run open: %s, started by %s", host, run.ID, run.By)
}

// runOpenRedactedMsg is refusal 7 for a service whose open run is not its
// own: neither that run's id nor who started it.
func runOpenRedactedMsg(host string) string {
	return fmt.Sprintf("%s already has a run open. try again once it has finished", host)
}

// runOpenRefusal is refusal 7 as p may read it. A service is told the open
// run's id and requester only when the run is its own; a run whose request
// cannot be read is not shown to be its own.
func (s *Server) runOpenRefusal(ctx context.Context, p Principal, sqid, app, host string, run provision.RunState) string {
	req, err := runs.ReadRequest(ctx, s.cfg.Releases, sqid, app, run.ID)
	if err != nil {
		req = nil
	}
	if p.Kind == PrincipalUser {
		return runOpenMsg(host, run, req)
	}
	if req != nil && s.runVisibleTo(p, *req) {
		return runOpenMsg(host, run, req)
	}
	return runOpenRedactedMsg(host)
}

// runMovingMsg is refusal 6 (RUN-C23), at the door and at admission.
func runMovingMsg(host string) string {
	return fmt.Sprintf("%s is moving. run the command once the move has finished", host)
}

// runImportMsg is refusal 9 (RUN-C33), at the door and at admission.
func runImportMsg(host string) string {
	return fmt.Sprintf("%s is an import that is not committed yet. finish or abort the import first", host)
}

// runFilesTooLargeMsg is refusal 8 for the file total (RUN-C30), whether meta
// declares it or the body grows past it.
const runFilesTooLargeMsg = "the files come to more than 64 MB"

const runMalformedMsg = "the run request is malformed"

const runRecordFailedMsg = "could not record the run. try again"

const runCommandNotGrantedMsg = "this credential may not run that command. it runs only the commands it was minted with"

// runKeyHeader is the name of a create's idempotency key (service runs spec
// § 3).
const runKeyHeader = "Idempotency-Key"

const runKeysNotReadyMsg = "this console does not take an Idempotency-Key on runs yet. try again later"

const runKeyInvalidMsg = "an Idempotency-Key is 1 to 128 characters: letters, digits, dots, underscores, colons and hyphens"

const runKeyReusedMsg = "this Idempotency-Key was used for a different run request. use a new key for a new request"

const runFileHashRequiredMsg = "a run created with an Idempotency-Key must declare the sha256 of each file"

func runFileHashShapeMsg(name string) string {
	return fmt.Sprintf("the sha256 declared for %s is not 64 hex digits", name)
}

func runFileHashMismatchMsg(name string) string {
	return fmt.Sprintf("%s does not match the sha256 declared for it", name)
}

func runNoKeyMsg(host string) string {
	return fmt.Sprintf("%s has no run started with that Idempotency-Key", host)
}

// runKeyOf reads the create's Idempotency-Key. sent reports whether the
// header was there at all, so an empty, repeated or malformed key is
// refused (ok false), never taken for an unkeyed create.
func runKeyOf(r *http.Request) (key string, sent, ok bool) {
	vals := r.Header.Values(runKeyHeader)
	switch {
	case len(vals) == 0:
		return "", false, false
	case len(vals) != 1 || !runs.ValidKey(vals[0]):
		return "", true, false
	}
	return vals[0], true, true
}

// runFileHashRefusal is "" when meta's declared sha256s may stand: each one
// declared is 64 hex digits, and a keyed create declares them all.
func runFileHashRefusal(m runs.Meta, keyed bool) string {
	for _, f := range m.Files {
		switch {
		case f.SHA256 == "" && keyed:
			return runFileHashRequiredMsg
		case f.SHA256 != "" && !runs.ValidFileHash(f.SHA256):
			return runFileHashShapeMsg(f.Name)
		}
	}
	return ""
}

// runReplayed answers a create whose key already admitted run id: 202 with
// that run, replayed, and its state while the record still names it.
func runReplayed(w http.ResponseWriter, rec *provision.Record, host, id string) {
	state := ""
	if rec != nil && rec.ActiveRun != nil && rec.ActiveRun.ID == id {
		state = rec.ActiveRun.State
	}
	writeJSON(w, http.StatusAccepted, runs.Created{ID: id, Host: host, State: state, Replayed: true})
}

// runMetaRefusalCode is the code for a refusal runMetaRefusal named.
func runMetaRefusalCode(msg string) string {
	if msg == runFilesTooLargeMsg {
		return runCodeFilesTooLarge
	}
	return runCodeInvalidRequest
}

// runClassRefusal is spec §5.1 refusal 5 as a sentence, "" for a
// platform-run, hibernating exec instance on a platform box.
func runClassRefusal(rec provision.Record) string {
	switch {
	case rec.Kind != "" && rec.Kind != "instance":
		return fmt.Sprintf("%s is a %s route. carlos run works on instances", rec.Host, rec.Kind)
	case rec.Addr != "" || len(rec.Addrs) > 0:
		return fmt.Sprintf("%s is served from its own address, so the platform does not run its process", rec.Host)
	case !rec.HibernateOrDefault():
		return fmt.Sprintf("%s does not sleep, so the platform does not run its process", rec.Host)
	case rec.Pool != "":
		return fmt.Sprintf("%s is a pooled instance. carlos run works on instances with a process of their own", rec.Host)
	}
	if fleet, ok := registry.FleetPlacement(rec.Region); ok {
		return fmt.Sprintf("%s runs on your fleet %s. carlos run works on platform boxes only", rec.Host, fleet)
	}
	return ""
}

// runMetaRefusal is refusal 8 as a sentence, "" when meta is valid.
func runMetaRefusal(m runs.Meta) string {
	switch err := runs.CheckArgv(m.Argv); {
	case errors.Is(err, runs.ErrNoArgs):
		return "name the command to run after --"
	case errors.Is(err, runs.ErrTooManyArgs):
		return "a run takes at most 256 arguments"
	case errors.Is(err, runs.ErrArgvTooLarge):
		return "the arguments come to more than 64 KB"
	case errors.Is(err, runs.ErrArgNUL):
		return "an argument holds a NUL byte"
	}
	if runs.CheckTimeout(time.Duration(m.TimeoutS)*time.Second) != nil {
		return "the timeout must be between 1 second and 6 hours"
	}
	err := runs.CheckFiles(m.Files)
	var fe *runs.FileNameError
	switch {
	case errors.Is(err, runs.ErrFilesTooLarge):
		return runFilesTooLargeMsg
	case errors.As(err, &fe) && fe.Twice:
		return fmt.Sprintf("two files are both named %s", fe.Name)
	case errors.As(err, &fe):
		return fmt.Sprintf("%s is not a plain file name", fe.Name)
	}
	return ""
}

// spooled is one uploaded file on the console's disk.
type spooled struct {
	runs.File
	path string
}

// spoolRunFiles reads the file parts, one per meta file in meta's order,
// into temp files, checking each name and size and hashing it. The caller
// removes the files. A short, long, misnamed or missing part is
// malformed; a body past the cap is the 64 MB refusal.
func spoolRunFiles(mr *multipart.Reader, files []runs.MetaFile) ([]spooled, error) {
	var out []spooled
	fail := func(err error) ([]spooled, error) {
		for _, f := range out {
			os.Remove(f.path)
		}
		return nil, err
	}
	for _, want := range files {
		part, err := mr.NextPart()
		if err != nil {
			return fail(err)
		}
		if part.FormName() != "file" || part.FileName() != want.Name {
			return fail(errRunMalformed)
		}
		tmp, err := os.CreateTemp("", "carlos-run-*")
		if err != nil {
			return fail(err)
		}
		h := sha256.New()
		n, err := io.Copy(io.MultiWriter(tmp, h), io.LimitReader(part, want.Size+1))
		cerr := tmp.Close()
		out = append(out, spooled{File: runs.File{Name: want.Name, Size: n, SHA256: hex.EncodeToString(h.Sum(nil))}, path: tmp.Name()})
		if err != nil {
			return fail(err)
		}
		if cerr != nil {
			return fail(cerr)
		}
		if n != want.Size {
			return fail(errRunMalformed)
		}
	}
	if _, err := mr.NextPart(); err != io.EOF {
		return fail(errRunMalformed)
	}
	return out, nil
}

var errRunMalformed = errors.New("console: run request disagrees with its meta")

// handleAPIRunCreate is POST …/instances/{host}/runs (run spec §5.1). The
// refusals come in the spec's order; nothing is written before every one
// that needs no body has passed.
//
// A create that carries an Idempotency-Key (service runs spec § 3) reads its
// meta straight after the record, and answers a key the record already
// holds before any other refusal and without reading a file: the same
// request is the run that key admitted, a different one is refused. Only
// then come the record's refusals, the command check and the files, whose
// bytes must match the sha256 meta declares. The admission records the key
// in the same write. An unkeyed create keeps the order it had before keys.
func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
	p, ok := principalFrom(r.Context())
	if !ok || !runCaller(p) {
		http.NotFound(w, r)
		return
	}
	actor, acct, app, ok := s.apiAccountApp(w, r)
	if !ok {
		return
	}
	if s.cfg.HostClaim {
		jsonErrorCode(w, http.StatusConflict, runCodeHostClaim, claimRunUnsupportedMsg)
		return
	}
	if !s.cli.instances.allow(hashToken(bearer(r)), time.Now()) {
		jsonErrorCode(w, http.StatusTooManyRequests, runCodeRateLimited, runTooManyMsg)
		return
	}
	key, keyed, keyOK := runKeyOf(r)
	if keyed && !s.cfg.RunKeysEnabled {
		// Never run as unkeyed: the caller would retry it into a second run.
		jsonErrorCode(w, http.StatusServiceUnavailable, runCodeKeysNotReady, runKeysNotReadyMsg)
		return
	}
	scope := callerScope(p)
	if keyed && (!keyOK || scope == "") {
		jsonErrorCode(w, http.StatusBadRequest, runCodeKeyInvalid, runKeyInvalidMsg)
		return
	}
	if s.cfg.Releases == nil {
		jsonErrorCode(w, http.StatusServiceUnavailable, runCodeUnavailable, runRecordFailedMsg)
		return
	}
	ctx := r.Context()
	host := r.PathValue("host")
	notInstance := fmt.Sprintf("%s is not an instance of %s", host, app)
	if !provision.ValidHostShape(host) {
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, notInstance)
		return
	}
	rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
	if errors.Is(err, os.ErrNotExist) {
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, notInstance)
		return
	}
	if err != nil {
		log.Printf("console: api run %s/%s/%s: read record: %v", acct.Sqid, app, host, err)
		jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
		return
	}
	// recordRefused writes refusals 5 to 7 from the record (the class, a move
	// in flight, an open run) and reports whether it wrote one.
	recordRefused := func() bool {
		if msg := runClassRefusal(*rec); msg != "" {
			jsonErrorCode(w, http.StatusConflict, runCodeNotRunnable, msg)
			return true
		}
		if rec.Move != nil {
			jsonErrorCode(w, http.StatusConflict, runCodeMoving, runMovingMsg(host))
			return true
		}
		if rec.ActiveRun.Open() {
			jsonErrorCode(w, http.StatusConflict, runCodeRunOpen, s.runOpenRefusal(ctx, p, acct.Sqid, app, host, *rec.ActiveRun))
			return true
		}
		return false
	}
	if !keyed && recordRefused() {
		return
	}
	r.Body = http.MaxBytesReader(w, r.Body, runs.MaxFileBytes+runs.MetaAllowance)
	mr, err := r.MultipartReader()
	if err != nil {
		jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg)
		return
	}
	var meta runs.Meta
	part, err := mr.NextPart()
	if err == nil && part.FormName() == "meta" {
		err = json.NewDecoder(io.LimitReader(part, runs.MetaAllowance)).Decode(&meta)
	} else if err == nil {
		err = errRunMalformed
	}
	if err != nil {
		jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg)
		return
	}
	// A declared sha256 is compared and digested in the lowercase form the
	// spool computes, so either case declares the same file.
	for i := range meta.Files {
		meta.Files[i].SHA256 = runs.NormalFileHash(meta.Files[i].SHA256)
	}
	if msg := runMetaRefusal(meta); msg != "" {
		jsonErrorCode(w, http.StatusBadRequest, runMetaRefusalCode(msg), msg)
		return
	}
	if msg := onBehalfOfRefusal(p, meta.OnBehalfOf); msg != "" {
		jsonErrorCode(w, http.StatusBadRequest, runCodeOnBehalfOfInvalid, msg)
		return
	}
	if msg := runFileHashRefusal(meta, keyed); msg != "" {
		jsonErrorCode(w, http.StatusBadRequest, runCodeInvalidRequest, msg)
		return
	}
	var scoped, digest, caller string
	if keyed {
		scoped, digest, caller = runs.ScopedKey(scope, key), runs.MetaDigest(meta), runs.CallerOf(scope)
		if k, ok := rec.RunKeyFor(scoped, time.Now()); ok {
			if k.Digest != digest {
				jsonErrorCode(w, http.StatusConflict, runCodeKeyReused, runKeyReusedMsg)
				return
			}
			// The files are never read: the run that key admitted has them.
			runReplayed(w, rec, host, k.RunID)
			return
		}
		if recordRefused() {
			return
		}
		if rec.ImportUncommitted() {
			jsonErrorCode(w, http.StatusConflict, runCodeImportUncommitted, runImportMsg(host))
			return
		}
	}
	// A run service runs only what it was granted, refused before any
	// object is written. A person runs any command.
	if p.Kind == PrincipalService && !runs.CommandGranted(meta.Argv, p.Commands) {
		jsonErrorCode(w, http.StatusForbidden, runCodeCommandNotGranted, runCommandNotGrantedMsg)
		return
	}
	if !keyed && rec.ImportUncommitted() {
		jsonErrorCode(w, http.StatusConflict, runCodeImportUncommitted, runImportMsg(host))
		return
	}
	files, err := spoolRunFiles(mr, meta.Files)
	defer func() {
		for _, f := range files {
			os.Remove(f.path)
		}
	}()
	var tooLarge *http.MaxBytesError
	switch {
	case errors.As(err, &tooLarge):
		jsonErrorCode(w, http.StatusRequestEntityTooLarge, runCodeFilesTooLarge, runFilesTooLargeMsg)
		return
	case err != nil:
		jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg)
		return
	}
	for i, f := range files {
		if want := meta.Files[i].SHA256; want != "" && f.SHA256 != want {
			jsonErrorCode(w, http.StatusBadRequest, runCodeFileHashMismatch, runFileHashMismatchMsg(f.Name))
			return
		}
	}
	id, err := runs.MintID()
	if err != nil {
		jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
		return
	}
	inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
	if err != nil {
		jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
		return
	}
	now := time.Now().UTC()
	req := runs.Request{Schema: 1, ID: id, Account: acct.Sqid, App: app, Host: host, Incarnation: inc,
		Argv: meta.Argv, TimeoutS: meta.TimeoutS, By: actor, RequestedAt: now, IdempotencyKey: scoped}
	if p.Kind == PrincipalService {
		req.ByService, req.ByCredential, req.OnBehalfOf = p.Name, p.CredentialID, meta.OnBehalfOf
	}
	for _, f := range files {
		req.Files = append(req.Files, f.File)
	}
	if err := s.writeRunObjects(ctx, req, files); err != nil {
		log.Printf("console: api run %s/%s/%s: write %s: %v", acct.Sqid, app, host, id, err)
		_ = runs.DeleteRun(context.WithoutCancel(ctx), s.cfg.Releases, acct.Sqid, app, id)
		jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
		return
	}
	// The pickup window opens at admission, not at the first byte: a slow
	// upload must not eat the box's 60 seconds.
	//
	// The admission runs to its end even if the caller has gone: a conditional
	// Put that is cancelled may or may not have landed, and deleting the
	// objects of a run the record holds would leave a queued run no box can
	// read. The outcome is then settled from the record, not from the error.
	wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
	defer cancel()
	state, err := s.admitRun(wctx, acct.Sqid, app, host, id, actor, scoped, digest, caller)
	// twinWon answers with the run a twin (the same key and request) had
	// admitted first. This request's own objects are no one's.
	twinWon := func(cur *provision.Record, twin string) {
		if derr := runs.DeleteRun(context.WithoutCancel(ctx), s.cfg.Releases, acct.Sqid, app, id); derr != nil {
			log.Printf("console: api run %s/%s/%s: delete %s after its twin %s won: %v", acct.Sqid, app, host, id, twin, derr)
		}
		runReplayed(w, cur, host, twin)
	}
	var replay *provision.RunKeyReplay
	if errors.As(err, &replay) {
		// admitRun answered a replay of this request's own id as its
		// admission, so this one names a twin.
		cur, _ := provision.GetRecord(wctx, s.cfg.Releases, acct.Sqid, app, host)
		twinWon(cur, replay.RunID)
		return
	}
	if err != nil && !runAdmitRefused(err) {
		// The outcome is unknown (a lost response, a cancelled request, a
		// transport error after the write, or both CAS attempts lost): ask
		// the record whose run it holds.
		cur, rerr := provision.GetRecord(wctx, s.cfg.Releases, acct.Sqid, app, host)
		switch {
		case rerr != nil:
			// Still unknown. Keep the objects: if the run was admitted a box
			// needs them, and if not, the orphan sweep ages them out by
			// request.json (Task 5).
			log.Printf("console: api run %s/%s/%s: admit %s: %v; record unreadable: %v", acct.Sqid, app, host, id, err, rerr)
			jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
			return
		case keyed:
			// A keyed admission is settled by its key, which commits with it
			// or not at all, never by ActiveRun, which a box or an ack may
			// already have moved on: a committed run keeps its objects.
			k, found := cur.RunKeyFor(scoped, time.Now())
			switch {
			case found && k.RunID == id:
				state, err = "", nil
				if cur.ActiveRun != nil && cur.ActiveRun.ID == id {
					state = cur.ActiveRun.State
				}
			case found && k.Digest == digest:
				twinWon(cur, k.RunID)
				return
			case found:
				err = provision.ErrRunKeyReused
			case cur.ActiveRun != nil && cur.ActiveRun.ID == id:
				// The key is gone (a record writer dropped run_keys between
				// the admit and now) but the record still holds this run:
				// it is admitted, and deleting its objects would leave a
				// queued run no box can read.
				state, err = cur.ActiveRun.State, nil
			}
		case cur.ActiveRun != nil && cur.ActiveRun.ID == id:
			state, err = cur.ActiveRun.State, nil
		}
	}
	if err != nil {
		_ = runs.DeleteRun(context.WithoutCancel(ctx), s.cfg.Releases, acct.Sqid, app, id)
		var open *provision.RunOpenError
		switch {
		case errors.As(err, &open):
			jsonErrorCode(w, http.StatusConflict, runCodeRunOpen, s.runOpenRefusal(wctx, p, acct.Sqid, app, host, open.Run))
		case errors.Is(err, provision.ErrRunMoving):
			jsonErrorCode(w, http.StatusConflict, runCodeMoving, runMovingMsg(host))
		case errors.Is(err, provision.ErrRunImport):
			jsonErrorCode(w, http.StatusConflict, runCodeImportUncommitted, runImportMsg(host))
		case errors.Is(err, provision.ErrRunKeyReused):
			jsonErrorCode(w, http.StatusConflict, runCodeKeyReused, runKeyReusedMsg)
		case errors.Is(err, os.ErrNotExist):
			jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, notInstance)
		default:
			log.Printf("console: api run %s/%s/%s: admit %s: %v", acct.Sqid, app, host, id, err)
			jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
		}
		return
	}
	// The serial bump and the run-start row belong to the admission, so they
	// run on its context too: a caller that left after it must not cost the
	// box its prompt pickup or the row its version.
	s.bumpSerial(wctx)
	s.auditActivity(acct, app, actor, Activity{Action: "run-start", TargetType: "instance", Target: host,
		Detail: runStartDetail(req, s.runVersion(wctx, acct.Sqid, app, rec.Channel))})
	writeJSON(w, http.StatusAccepted, runs.Created{ID: id, Host: host, State: state})
}

// admitRun is the admission CAS, with the caller's scoped key, request
// digest and runs.CallerOf when the create is keyed. A RunOpenError naming id itself means an
// earlier attempt of this same call committed (updateRecord retries after a
// lost-response conflict, and finds its own run queued): that is admitted.
// So is a RunKeyReplay naming id, the same retry finding its own key; any
// other RunKeyReplay is returned for the caller to answer.
func (s *Server) admitRun(ctx context.Context, sqid, app, host, id, by, key, digest, caller string) (string, error) {
	got, err := provision.SetRecordRun(ctx, s.cfg.Releases, sqid, app, host,
		provision.RunAdmit(provision.RunAdmission{ID: id, By: by, Deadline: time.Now().UTC().Add(runs.PickupWindow), Key: key, Digest: digest, Caller: caller}))
	var open *provision.RunOpenError
	var replay *provision.RunKeyReplay
	switch {
	case err == nil:
		return got.ActiveRun.State, nil
	case errors.As(err, &open) && open.Run.ID == id:
		return open.Run.State, nil
	case errors.As(err, &replay) && replay.RunID == id:
		// This request's own admission committed; delete nothing. The state
		// is the record's, "" if it cannot be read now.
		state := ""
		if cur, rerr := provision.GetRecord(ctx, s.cfg.Releases, sqid, app, host); rerr == nil && cur.ActiveRun != nil && cur.ActiveRun.ID == id {
			state = cur.ActiveRun.State
		}
		return state, nil
	}
	return "", err
}

// runAdmitRefused reports whether err is the record saying no (another run,
// a move, an uncommitted import, a key reused, no record) rather than a
// failure whose outcome the record has to settle.
func runAdmitRefused(err error) bool {
	var open *provision.RunOpenError
	return errors.As(err, &open) || errors.Is(err, provision.ErrRunMoving) ||
		errors.Is(err, provision.ErrRunImport) || errors.Is(err, provision.ErrRunKeyReused) ||
		errors.Is(err, os.ErrNotExist)
}

// handleAPIRunKeyLookup is GET …/instances/{host}/runs?key=<key> (service
// runs spec § 3): the run the caller's own key admitted on this instance,
// from the record, or 404. Another caller's key is never found, because the
// record holds keys scoped to their caller. It is the run doors' gate, then
// a read.
func (s *Server) handleAPIRunKeyLookup(w http.ResponseWriter, r *http.Request) {
	p, ok := principalFrom(r.Context())
	if !ok || !runCaller(p) {
		http.NotFound(w, r)
		return
	}
	_, acct, app, ok := s.apiAccountApp(w, r)
	if !ok {
		return
	}
	if s.cfg.HostClaim {
		jsonErrorCode(w, http.StatusConflict, runCodeHostClaim, claimRunUnsupportedMsg)
		return
	}
	if !s.cli.runs.allow(hashToken(bearer(r)), time.Now()) {
		jsonErrorCode(w, http.StatusTooManyRequests, runCodeRateLimited, runTooManyMsg)
		return
	}
	keys, scope := r.URL.Query()["key"], callerScope(p)
	if len(keys) != 1 || !runs.ValidKey(keys[0]) || scope == "" {
		jsonErrorCode(w, http.StatusBadRequest, runCodeKeyInvalid, runKeyInvalidMsg)
		return
	}
	if s.cfg.Releases == nil {
		jsonErrorCode(w, http.StatusServiceUnavailable, runCodeUnavailable, runReadFailedMsg)
		return
	}
	host := r.PathValue("host")
	if !provision.ValidHostShape(host) {
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, fmt.Sprintf("%s is not an instance of %s", host, app))
		return
	}
	rec, err := provision.GetRecord(r.Context(), s.cfg.Releases, acct.Sqid, app, host)
	switch {
	case errors.Is(err, os.ErrNotExist):
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, fmt.Sprintf("%s is not an instance of %s", host, app))
		return
	case err != nil:
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
		return
	}
	k, found := rec.RunKeyFor(runs.ScopedKey(scope, keys[0]), time.Now())
	if !found {
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, runNoKeyMsg(host))
		return
	}
	writeJSON(w, http.StatusOK, runs.KeyAnswer{RunID: k.RunID})
}

// writeRunObjects writes request.json first (the orphan sweep ages a run
// directory by it, Task 5), then the files. The record CAS comes after
// both, so no box ever sees a run whose objects are partial.
func (s *Server) writeRunObjects(ctx context.Context, req runs.Request, files []spooled) error {
	body, err := json.Marshal(req)
	if err != nil {
		return err
	}
	if _, err := s.cfg.Releases.Put(ctx, runs.RequestKey(req.Account, req.App, req.ID), body, ""); err != nil {
		return err
	}
	for _, f := range files {
		fh, err := os.Open(f.path)
		if err != nil {
			return err
		}
		_, err = release.PutObjectFrom(ctx, s.cfg.Releases, runs.InputKey(req.Account, req.App, req.ID, f.Name), fh, f.Size, "")
		fh.Close()
		if err != nil {
			return err
		}
	}
	return nil
}

// runVersion is the release the instance's channel serves, "" when it
// cannot be read: the audit row's version is a record, not a gate.
func (s *Server) runVersion(ctx context.Context, sqid, app, channel string) string {
	ptr, err := release.Channel(ctx, s.cfg.Releases, s.cfg.ReleasePub, sqid, app, channel)
	if err != nil || ptr == nil {
		return ""
	}
	return ptr.Version
}

// runStartDetail is the run-start row's detail (run spec §6): the run id,
// argv, the file names and sizes and the version, as JSON. Output never
// goes in the audit log; argv does, and the docs say so.
func runStartDetail(req runs.Request, version string) string {
	type file struct {
		Name string `json:"name"`
		Size int64  `json:"size"`
	}
	d := struct {
		ID      string   `json:"id"`
		Argv    []string `json:"argv"`
		Files   []file   `json:"files"`
		Version string   `json:"version"`
		// Requester is a service's run's line (runs.Request.ServiceLine),
		// absent for a person's: the row's actor already names them.
		Requester string `json:"requester,omitempty"`
	}{ID: req.ID, Argv: req.Argv, Version: version, Requester: req.ServiceLine()}
	for _, f := range req.Files {
		d.Files = append(d.Files, file{f.Name, f.Size})
	}
	raw, _ := json.Marshal(d)
	return string(raw)
}