// The import door (seeded-instances spec §5): `carlos instances import`
// moves an instance the platform does not host yet onto it, from an
// uploaded snapshot. Begin is the create door with "import" in its body:
// handleAPIInstanceCreate runs every ordinary check, then hands over here.
// Begin writes the record awaiting-upload, provisions an empty store for a
// private-storage app, and answers 201 with the verify token, once.

package console

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"runtime/debug"
	"strconv"
	"time"

	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// seedImportReq is the create body's "import" (spec §5.1): the snapshot's
// length and hash, which the upload must match, and the address of the
// addr-backed row the box swaps out.
type seedImportReq struct {
	SHA256   string `json:"sha256"`
	Bytes    int64  `json:"bytes"`
	Replaces string `json:"replaces"`
}

// importFailed is the import door's retry sentence (SD-N67).
const importFailed = "could not store the upload. Try again."

// importStepFailure is the retry sentence of every import call that is not
// storing the upload: the record read at the gate, and the stopped, commit
// and abort steps (SD-N120).
func importStepFailure(host string) string {
	return fmt.Sprintf("could not update the import of %s. Try again.", host)
}

// importSizeRule is SD-N48, answered 400 below the range and 413 above it.
const importSizeRule = "bytes must be more than 0 and at most 4 GiB"

var (
	sha256Hex = regexp.MustCompile(`^[0-9a-f]{64}$`)
	portShape = regexp.MustCompile(`^[0-9]{1,5}$`)
)

// seedOneDoorRefusal is the create door's rule that a body takes at most
// one seed door: "from" and "import" together are refused.
func seedOneDoorRefusal(from, imp bool) *doorRefusal {
	if from && imp {
		return &doorRefusal{http.StatusBadRequest, "a create takes from or import, not both."}
	}
	return nil
}

// importShapeRefusal checks begin's own fields: the size against the
// limit, the hash's spelling, and replaces as host:port with a port in
// 1 to 65535.
func importShapeRefusal(imp seedImportReq) *doorRefusal {
	switch {
	case imp.Bytes <= 0:
		return &doorRefusal{http.StatusBadRequest, importSizeRule}
	case imp.Bytes > seedMaxBytes:
		return &doorRefusal{http.StatusRequestEntityTooLarge, importSizeRule}
	case !sha256Hex.MatchString(imp.SHA256):
		return &doorRefusal{http.StatusBadRequest, "sha256 must be 64 lowercase hex characters"}
	case !validReplaces(imp.Replaces):
		return &doorRefusal{http.StatusBadRequest, "replaces must name the address the import takes over, as host:port"}
	}
	return nil
}

func validReplaces(addr string) bool {
	host, port, err := net.SplitHostPort(addr)
	if err != nil || host == "" || !portShape.MatchString(port) {
		return false
	}
	n, err := strconv.Atoi(port)
	return err == nil && n > 0 && n < 65536
}

// createImport is the import door's begin (spec §5.1). Its answer is held
// until the door is done, so that a begin which fails after writing its
// record has rolled that record back and finished it (freeing the host)
// before the refusal reaches the caller (preflight M2).
func (s *Server) createImport(w http.ResponseWriter, r *http.Request, c seedCreate, imp seedImportReq) {
	held := &heldResponse{header: w.Header()}
	s.beginImport(held, r, c, imp)
	held.flush(w, importFailed)
}

// beginImport runs begin's checks after the create door's own, then mints
// the seed id and the verify token, writes the record awaiting-upload with
// only the token's hash, provisions the seed's store, and answers 201 with
// the token. The token is in that one body and nowhere else: never logged,
// audited or stored.
func (s *Server) beginImport(w http.ResponseWriter, r *http.Request, c seedCreate, imp seedImportReq) {
	ctx := r.Context()
	acct, app, host := c.acct, c.app, c.rec.Host
	if ref := seedPlacementRefusal(c); ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" {
		ref := seedNoDataRefusal()
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	if ref := importShapeRefusal(imp); ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	// The snapshot goes to this console's data bucket, and the import's
	// rollback reads it there: the record is homed in this region.
	if c.rec.Region != s.cfg.Region {
		jsonError(w, http.StatusConflict, fmt.Sprintf("an import runs in this console's region, %s. To run %s elsewhere, import it here, commit it, and then run carlos move.",
			s.dbRegionName(s.cfg.Region), host))
		return
	}
	checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
	defer cancel()
	if ref := s.seedRegionReady(checks, c.rec.Region, true); ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	id, err := seeds.NewID()
	if err != nil {
		log.Printf("console: import %s: seed id: %v", host, err)
		jsonError(w, http.StatusBadGateway, importFailed)
		return
	}
	token, hash, err := seeds.NewVerifyToken()
	if err != nil {
		log.Printf("console: import %s: verify token: %v", host, err)
		jsonError(w, http.StatusBadGateway, importFailed)
		return
	}
	if c.m.PrivateStorage {
		// The store gates that write nothing, before the record (ruling
		// R3): their refusals leave nothing holding the host.
		if _, ok := s.instanceStorePreflight(ctx, w, acct, app); !ok {
			return
		}
		storeCtx, storeCancel := context.WithTimeout(ctx, instanceStoreTimeout)
		ref := s.seedStoreRefusal(storeCtx, c, id, importFailed)
		storeCancel()
		if ref != nil {
			jsonError(w, ref.Status, ref.Msg)
			return
		}
	}

	rec := c.rec
	rec.Seed = &provision.SeedSource{ID: id, Kind: seeds.KindImport, SHA256: imp.SHA256, Bytes: imp.Bytes,
		State: seeds.StateAwaitingUpload, By: c.principal, At: time.Now().UTC(), Replaces: imp.Replaces, VerifyHash: hash}
	// Not the request's cancellation: a client leaving now must not turn a
	// write that landed into an error.
	writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(ctx), instanceStoreTimeout)
	err = provision.CreateRecord(writeCtx, s.cfg.Releases, rec)
	writeCancel()
	if err != nil {
		if errors.Is(err, release.ErrCASConflict) {
			jsonError(w, http.StatusConflict, "instance "+host+" already exists")
			return
		}
		// The write may have landed all the same: undo it by its seed id,
		// which touches nothing if it did not.
		log.Printf("console: import %s: write the record: %v", host, err)
		s.unbeginImport(ctx, c, id, "a failed record write")
		jsonError(w, http.StatusBadGateway, importFailed)
		return
	}

	// From here every failure rolls the import back and finishes it
	// before the answer goes out.
	privateBucket := ""
	if c.m.PrivateStorage {
		// The request's own context, as the ordinary create's walk: the
		// walk takes storeGrantTimeout, and a client that leaves cancels a
		// begin whose token it can no longer receive. A refusal is held on
		// w already, except errSeedStoreTaken and errSeedEnvironmentUnread,
		// which are spoken here.
		storeRec, ok, err := s.instanceStoreProvisionFor(ctx, w, acct, app, rec.Environment, c.principal, id)
		if !ok {
			s.unbeginImport(ctx, c, id, "store provisioning")
			switch {
			case errors.Is(err, errSeedStoreTaken):
				ref := seedStoreTakenRefusal(rec.Environment, app, storeRec)
				jsonError(w, ref.Status, ref.Msg)
			case errors.Is(err, errSeedEnvironmentUnread):
				jsonError(w, http.StatusBadGateway, importFailed)
			}
			return
		}
		privateBucket = storeBucketName(storeRec)
	}
	// The create door's re-check after its write (apiinstances.go): an app
	// created under the host's label since the door asked owns its apex.
	if c.siblingLabel != "" {
		if live, trashed, err := s.store.AppNameState(acct.ID, c.siblingLabel); err == nil && (live || trashed) {
			s.unbeginImport(ctx, c, id, "a sibling app took the label")
			jsonError(w, http.StatusConflict, seedSiblingApexReason(host, c.siblingLabel))
			return
		}
	}

	s.auditActivity(acct, app, c.principal, Activity{
		Action: "instance-seed", TargetType: "instance", Target: host, Message: c.message,
		Detail: fmt.Sprintf("began an import of %s over %s (%s, %d bytes)", host, imp.Replaces, imp.SHA256[:12], imp.Bytes),
	})
	viewCtx, viewCancel := context.WithTimeout(context.WithoutCancel(ctx), instanceStoreTimeout)
	defer viewCancel()
	out := struct {
		apiRecordView
		VerifyToken string `json:"verify_token"`
	}{recordView(rec), token}
	out.PrivateStoreBucket = privateBucket
	out.RegionFrom = c.regionFrom
	out.RegionName = s.placeOf(rec.Region, s.candsOrHome(viewCtx)) // never fleet-placed: refused above
	writeJSON(w, http.StatusCreated, out)
}

// unbeginImport undoes a begin that failed after its record was written:
// the pre-swap rollback, then the console's finish, which tears down the
// store this seed made and deletes the record (spec §5.6's pre-swap path,
// preflight M2). Neither is cut by the client leaving. A failure is
// logged, and the record is left rolled-back for the sweep to finish.
func (s *Server) unbeginImport(ctx context.Context, c seedCreate, id, why string) {
	sqid, app, host := c.acct.Sqid, c.app, c.rec.Host
	bg := context.WithoutCancel(ctx)
	rbCtx, rbCancel := context.WithTimeout(bg, instanceStoreTimeout)
	_, err := provision.SetRecordSeed(rbCtx, s.cfg.Releases, sqid, app, host, provision.SeedRolledBackBeforeSwap(id))
	rbCancel()
	if errors.Is(err, os.ErrNotExist) {
		return // the record was never written
	}
	if err != nil {
		log.Printf("console: import %s: roll back seed %s after %s: %v", host, id, why, err)
	}
	finCtx, finCancel := context.WithTimeout(bg, storeTeardownTimeout+instanceStoreTimeout)
	defer finCancel()
	if err := s.finishSeed(finCtx, sqid, app, host, id); err != nil {
		log.Printf("console: import %s: finish seed %s after %s: %v", host, id, why, err)
	}
}

// Seams for the tests.
var (
	// checkSeedDBConsole is the upload's integrity check (spec §5.2 step 3).
	checkSeedDBConsole = lease.CheckDBContext
	// importCheckDeadline bounds that check.
	importCheckDeadline = 30 * time.Minute
	// importSpoolHook, when set, sees the spool once the body is read and
	// before it is checked or removed.
	importSpoolHook func(path string)
	// importReserveHook, when set, runs after the upload's first read of
	// the record and before its reservation.
	importReserveHook func(host string)
)

// importUploadReadDeadline is how long the upload's body may take to
// arrive: up to 4 GiB, past the server's ReadTimeout. It covers the body
// read only; the check and the put run after the answer.
const importUploadReadDeadline = 60 * time.Minute

// importGate is every import call's gate: the instance gate (cliRoute has
// already applied the instance role), the claim console's refusal, the
// host, the record, that the record is an import (SD-N59), and that it is
// the import the caller names (SD-N60).
func (s *Server) importGate(w http.ResponseWriter, r *http.Request) (principal string, acct *Account, app, host string, rec *provision.Record, ok bool) {
	principal, acct, app, ok = s.instanceGate(w, r)
	if !ok {
		return "", nil, "", "", nil, false
	}
	// A claim console begins no import (seedRegionReady), and its claim
	// protocol has no place for one yet (#250): every import door refuses
	// there before it reads the record.
	if s.cfg.HostClaim {
		ref := seedClaimConsoleRefusal()
		jsonError(w, ref.Status, ref.Msg)
		return "", nil, "", "", nil, false
	}
	host = r.PathValue("host")
	notInstance := notAnInstanceOf(host, app)
	if !provision.ValidHostShape(host) {
		jsonError(w, http.StatusNotFound, notInstance)
		return "", nil, "", "", nil, false
	}
	ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
	defer cancel()
	rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
	if errors.Is(err, os.ErrNotExist) {
		jsonError(w, http.StatusNotFound, notInstance)
		return "", nil, "", "", nil, false
	}
	if err != nil {
		log.Printf("console: import %s/%s/%s: read the record: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, importStepFailure(host))
		return "", nil, "", "", nil, false
	}
	if !rec.Seed.Import() {
		ref := notAnImportRefusal(host)
		jsonError(w, ref.Status, ref.Msg)
		return "", nil, "", "", nil, false
	}
	// The caller names the import it means (final review F1). A delayed
	// stopped, commit, abort or upload sent for one import must never land
	// on a later import of the same host: the record's id is what the
	// transitions below are guarded on, so it has to be the caller's id
	// too. A request that names another seed, or none, changed nothing:
	// SD-N60, and the caller reads the state.
	if r.URL.Query().Get(importSeedParam) != rec.Seed.ID {
		s.importTransitionFailure(w, app, host, provision.ErrSeedConflict)
		return "", nil, "", "", nil, false
	}
	return principal, acct, app, host, rec, true
}

// importSeedParam is the query parameter every import mutation names its
// seed id in: stopped, commit, abort and the upload.
const importSeedParam = "seed"

// importUpload is an upload whose spool matched: everything
// runImportUpload needs once the request has been answered. It owns the
// spool's directory and the scratch reservation from the hand-over.
type importUpload struct {
	ctx           context.Context // the request's values, never its cancellation
	acct          *Account
	app, host, by string
	id            string
	dir, spool    string
	release       func()
}

// handleAPIImportUpload is spec §5.2, in two halves (preflight M6, a
// deviation from the spec's synchronous 422 and 504 for Paul, following
// Plan 5 ruling R2). The request half reads the body into a private spool
// and checks its length and sha256 against begin's: a mismatch answers 422
// and leaves the import awaiting-upload, so the right file can be sent.
// Once the spool matches it answers 202 {"state":"checking"} and the rest
// runs detached (runImportUpload): an edge backhaul's 30-second
// response-header timeout and idle NATs would cut a response that waited
// for a 30-minute check and a 4 GiB put. The caller follows the record
// until it is uploaded, or aborted with the reason in Seed.Reason.
func (s *Server) handleAPIImportUpload(w http.ResponseWriter, r *http.Request) {
	principal, acct, app, host, rec, ok := s.importGate(w, r)
	if !ok {
		return
	}
	ctx := r.Context()
	seed := rec.Seed
	if ref := staleUploadRefusal(host, rec, seed.ID); ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	// The snapshot key is built from the record (preflight M8): an id that
	// cannot name a key must never reach the put.
	if err := seeds.Check(host, seed.ID); err != nil {
		log.Printf("console: import %s: seed %q: %v", host, seed.ID, err)
		jsonError(w, http.StatusBadGateway, importFailed)
		return
	}
	if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" {
		ref := seedNoDataRefusal()
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
	ref := s.seedRegionReady(checks, rec.Region, true)
	cancel()
	if ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	if importReserveHook != nil {
		importReserveHook(host)
	}
	releaseScratch, ref := s.seedScratch.reserve(acct.Sqid+"/"+app, seed.Bytes)
	if ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	// The read above came before the reservation: another upload of this
	// import may have landed in between (fix round 1). Under the
	// reservation, the record must still wait for this seed's upload.
	fresh, err := s.readImportRecord(ctx, acct.Sqid, app, host)
	switch {
	case errors.Is(err, os.ErrNotExist):
		releaseScratch()
		jsonError(w, http.StatusNotFound, notAnInstanceOf(host, app))
		return
	case err != nil:
		releaseScratch()
		log.Printf("console: import %s/%s/%s: re-read the record: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, importFailed)
		return
	}
	if ref := staleUploadRefusal(host, fresh, seed.ID); ref != nil {
		releaseScratch()
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	dir, err := os.MkdirTemp(s.seedScratch.dir(), seedScratchPrefix)
	if err != nil {
		releaseScratch()
		log.Printf("console: import %s: scratch: %v", host, err)
		ref := seedScratchFull()
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	handedOver := false
	defer func() {
		if !handedOver {
			_ = os.RemoveAll(dir) // with any -shm or -wal beside the spool
			releaseScratch()
		}
	}()
	spool := filepath.Join(dir, host+".db")
	if !s.spoolImport(w, r, spool, seed) {
		return
	}
	writeJSON(w, http.StatusAccepted, map[string]string{"state": "checking"})
	handedOver = true
	s.seedCopies.Add(1)
	go s.runImportUpload(&importUpload{ctx: context.WithoutCancel(ctx), acct: acct, app: app, host: host, by: principal,
		id: seed.ID, dir: dir, spool: spool, release: releaseScratch})
}

// readImportRecord reads host's record under instanceStoreTimeout.
func (s *Server) readImportRecord(ctx context.Context, sqid, app, host string) (*provision.Record, error) {
	ctx, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
	defer cancel()
	return provision.GetRecord(ctx, s.cfg.Releases, sqid, app, host)
}

// awaitingUpload reports whether rec's seed is the import id, still
// waiting for its upload.
func awaitingUpload(rec *provision.Record, id string) bool {
	return rec.Seed.Import() && rec.Seed.ID == id && rec.Seed.State == seeds.StateAwaitingUpload
}

// notAnImportRefusal is SD-N59.
func notAnImportRefusal(host string) *doorRefusal {
	return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is not an import.", host)}
}

// staleUploadRefusal is the upload's answer to a record that no longer
// waits for seed id's upload: SD-N53 naming its state, or SD-N59 when it
// is no longer an import at all.
func staleUploadRefusal(host string, rec *provision.Record, id string) *doorRefusal {
	switch {
	case awaitingUpload(rec, id):
		return nil
	case !rec.Seed.Import():
		return notAnImportRefusal(host)
	}
	return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is %s. It takes an upload only while it waits for one.", host, rec.Seed.State)}
}

// spoolImport reads the body into spool, never more than the declared
// length and one byte, hashing as it writes, and reports whether it
// matches begin's length and sha256. It has answered w when it reports
// false.
func (s *Server) spoolImport(w http.ResponseWriter, r *http.Request, spool string, seed *provision.SeedSource) bool {
	host := r.PathValue("host")
	// The server's ReadTimeout would cut a large body; this read gets an
	// hour. Go drops the deadline once the body has been read.
	if err := http.NewResponseController(w).SetReadDeadline(time.Now().Add(importUploadReadDeadline)); err != nil {
		log.Printf("console: import %s: extend the read deadline: %v", host, err)
	}
	f, err := os.OpenFile(spool, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		log.Printf("console: import %s: spool: %v", host, err)
		ref := seedScratchFull()
		jsonError(w, ref.Status, ref.Msg)
		return false
	}
	h := sha256.New()
	n, cerr := io.Copy(io.MultiWriter(f, h), http.MaxBytesReader(w, r.Body, seed.Bytes+1))
	ferr := f.Close()
	if importSpoolHook != nil {
		importSpoolHook(spool)
	}
	var tooLong *http.MaxBytesError
	switch {
	case cerr != nil && !errors.As(cerr, &tooLong):
		// The caller went away, the body broke off, or the spool's disk
		// failed: nothing to keep, and the import still waits.
		log.Printf("console: import %s: read the upload: %v", host, cerr)
		jsonError(w, http.StatusBadGateway, importFailed)
		return false
	case ferr != nil:
		log.Printf("console: import %s: spool: %v", host, ferr)
		jsonError(w, http.StatusBadGateway, importFailed)
		return false
	case tooLong != nil || n != seed.Bytes || hex.EncodeToString(h.Sum(nil)) != seed.SHA256:
		jsonError(w, http.StatusUnprocessableEntity, "the upload does not match the length and sha256 this import began with, so it was not kept.")
		return false
	}
	return true
}

// runImportUpload is the detached half: the integrity check, the put to
// the seed's snapshot key, and the move to uploaded. It never depends on
// the request, which has been answered. A failure moves the record to
// aborted with the sentence the door would have answered in Seed.Reason
// (SD-N52, SD-N61 or SD-N67), but only while the record still waits for
// this upload (SeedAbortUploadBecause): an import that has moved on, by
// another upload, an abort or a delete, is never touched, and neither is
// a snapshot already at its key (fix round 1). On every exit it removes
// the spool's directory and then releases the reservation, as runFrom
// does.
//
// The console has no shutdown hook to cancel a running check: a restart
// leaves the record awaiting-upload, its spool is cleared at start, and
// the sweep aborts it (Plan 8).
func (s *Server) runImportUpload(j *importUpload) {
	defer s.seedCopies.Done()
	defer j.release()
	ctx := j.ctx
	sqid, app, host := j.acct.Sqid, j.app, j.host
	abort := func(reason, why string) {
		bg, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
		defer cancel()
		_, err := provision.SetRecordSeed(bg, s.cfg.Releases, sqid, app, host, provision.SeedAbortUploadBecause(j.id, reason))
		switch {
		case errors.Is(err, provision.ErrSeedConflict), errors.Is(err, os.ErrNotExist):
			log.Printf("console: import %s: seed %s moved on before its upload's %s was recorded; not aborted: %v", host, j.id, why, err)
			return
		case err != nil:
			log.Printf("console: import %s: abort seed %s after %s: %v", host, j.id, why, err)
		}
		s.bumpSerial(bg)
	}
	// A panic in the check or the put would end the whole console: it
	// aborts this seed instead. Registered after release and Done, so it
	// runs before them; the spool's removal, registered later, has
	// already run.
	defer func() {
		if p := recover(); p != nil {
			log.Printf("console: import %s: panic in the upload of seed %s: %v\n%s", host, j.id, p, debug.Stack())
			abort(importFailed, "a panic")
		}
	}()
	defer os.RemoveAll(j.dir) // with any -shm or -wal the check left

	cctx, ccancel := context.WithTimeout(ctx, importCheckDeadline)
	err := checkSeedDBConsole(cctx, j.spool)
	timedOut := errors.Is(cctx.Err(), context.DeadlineExceeded)
	ccancel()
	if err != nil {
		if timedOut {
			abort("checking the upload took longer than 30 minutes, so it was not kept.", "the check's deadline")
			return
		}
		log.Printf("console: import %s: check: %v", host, err)
		abort("the upload is not a database that passes an integrity check, so it was not kept.", "the check")
		return
	}
	// The check can take half an hour: put nothing for an import that has
	// moved on meanwhile (finished, aborted, or landed by another upload).
	rec, err := s.readImportRecord(ctx, sqid, app, host)
	switch {
	case errors.Is(err, os.ErrNotExist):
		log.Printf("console: import %s: the record went while seed %s was checked; nothing put", host, j.id)
		return
	case err != nil:
		log.Printf("console: import %s: re-read before the put: %v", host, err)
		abort(importFailed, "the re-read")
		return
	case !awaitingUpload(rec, j.id):
		log.Printf("console: import %s: seed %s moved on while it was checked (%+v); nothing put", host, j.id, rec.Seed)
		return
	}
	uctx, ucancel := context.WithTimeout(ctx, seedUploadDeadline)
	err = putSnapshot(uctx, s.cfg.SeedData, j.spool, seeds.SnapshotKey(host, j.id))
	ucancel()
	if errors.Is(err, release.ErrCASConflict) {
		// Another writer put this seed's snapshot first, and owns the
		// outcome. Nothing is overwritten and nothing aborted.
		log.Printf("console: import %s: the snapshot of seed %s is already there; left as it is", host, j.id)
		return
	}
	if err != nil {
		log.Printf("console: import %s: put the snapshot: %v", host, err)
		abort(importFailed, "the put")
		return
	}
	pctx, pcancel := context.WithTimeout(ctx, instanceStoreTimeout)
	defer pcancel()
	_, err = provision.SetRecordSeed(pctx, s.cfg.Releases, sqid, app, host, provision.SeedUploaded(j.id))
	switch {
	case errors.Is(err, provision.ErrSeedConflict), errors.Is(err, os.ErrNotExist):
		log.Printf("console: import %s: seed %s moved on before it was marked uploaded; not aborted: %v", host, j.id, err)
		return
	case err != nil:
		log.Printf("console: import %s: record uploaded: %v", host, err)
		abort(importFailed, "the record write")
		return
	}
	s.auditActivity(j.acct, app, j.by, Activity{Action: "import-uploaded", TargetType: "instance", Target: host,
		Detail: fmt.Sprintf("uploaded the snapshot for %s", host)})
}

// putSnapshot writes path to key with one create-only PutFrom (spec §5.2
// step 4). It never replaces an object already there: that answers
// release.ErrCASConflict.
func putSnapshot(ctx context.Context, data release.Store, path, key string) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()
	st, err := f.Stat()
	if err != nil {
		return err
	}
	_, err = release.PutObjectFrom(ctx, data, key, f, st.Size(), "")
	return err
}

// notAnInstanceOf is the 404 sentence for a record that is not there.
func notAnInstanceOf(host, app string) string {
	return fmt.Sprintf("%s is not an instance of %s", host, app)
}

// lostTransition reports whether err is a transition that found the
// import elsewhere, or a record write that lost its race twice.
func lostTransition(err error) bool {
	return errors.Is(err, provision.ErrSeedConflict) || errors.Is(err, release.ErrCASConflict)
}

// importNow re-reads host's record once, after a transition lost, and
// returns its seed when it is still seed id. Nil means the import is gone,
// is another seed, or could not be read: the caller answers SD-N60.
func (s *Server) importNow(ctx context.Context, sqid, app, host, id string) *provision.SeedSource {
	rec, err := s.readImportRecord(ctx, sqid, app, host)
	if err != nil || rec.Seed == nil || rec.Seed.ID != id {
		return nil
	}
	return rec.Seed
}

// importTransitionFailure answers a synchronous transition of the import's
// record that did not land. A guarded transition that finds the import
// somewhere else (provision.ErrSeedConflict), or a record write that lost
// its race twice, means another request got there first: SD-N60, and the
// caller reads the state and decides. A record that is gone is the 404.
// Anything else is the retry sentence, with the cause logged.
func (s *Server) importTransitionFailure(w http.ResponseWriter, app, host string, err error) {
	switch {
	case lostTransition(err):
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s changed while this request was made. Read its state with carlos instances list and try again.", host))
	case errors.Is(err, os.ErrNotExist):
		jsonError(w, http.StatusNotFound, notAnInstanceOf(host, app))
	default:
		log.Printf("console: import %s: write the record: %v", host, err)
		jsonError(w, http.StatusBadGateway, importStepFailure(host))
	}
}

// handleAPIImportStopped is the app saying its own process is down (spec
// §5.3): uploaded to ready, and a serial bump so the box swaps. The
// transition is guarded on the seed id and on uploaded, so a stopped that
// races an abort, a delete or a second stopped changes nothing but its own
// answer; the check below only picks the sentence.
func (s *Server) handleAPIImportStopped(w http.ResponseWriter, r *http.Request) {
	principal, acct, app, host, rec, ok := s.importGate(w, r)
	if !ok {
		return
	}
	if rec.Seed.State != seeds.StateUploaded {
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s. Mark it stopped once, after its upload.", host, rec.Seed.State))
		return
	}
	checks, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
	ref := s.seedRegionReady(checks, rec.Region, true)
	cancel()
	if ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return
	}
	// Not the request's cancellation: a client leaving now must not turn a
	// transition that landed into a missing bump.
	wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout)
	defer wcancel()
	if _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedStopped(rec.Seed.ID)); err != nil {
		// An overlapping stopped may have won: this import is past stopped
		// now, and a sequential second stopped is told so (SD-N54). That
		// is not success: the caller reads 409 naming ready or later as
		// the stop having landed. Any other loss is SD-N60.
		if lostTransition(err) {
			if now := s.importNow(wctx, acct.Sqid, app, host, rec.Seed.ID); now != nil && pastStopped(now.State) {
				jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s. Mark it stopped once, after its upload.", host, now.State))
				return
			}
		}
		s.importTransitionFailure(w, app, host, err)
		return
	}
	s.auditActivity(acct, app, principal, Activity{Action: "import-stopped", TargetType: "instance", Target: host,
		Detail: fmt.Sprintf("marked the old process of %s stopped", host)})
	s.bumpSerial(wctx)
	writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateReady})
}

// handleAPIImportCommit is the app's verdict that the instance serves
// (spec §5.4): placed to committed, then the box clears the gate. A commit
// is a record transition and nothing else: the verify token is what the
// edge checks on the app's own requests (spec §5.5), and this door never
// sees it. Committed already answers 200, so a retried commit is safe.
func (s *Server) handleAPIImportCommit(w http.ResponseWriter, r *http.Request) {
	principal, acct, app, host, rec, ok := s.importGate(w, r)
	if !ok {
		return
	}
	switch rec.Seed.State {
	case seeds.StateCommitted:
		writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted})
		return
	case seeds.StatePlaced:
	case seeds.StateAwaitingUpload, seeds.StateUploaded, seeds.StateReady, seeds.StateSwapped:
		jsonError(w, http.StatusConflict, "nothing has been verified yet. Send a request with the verify token first, then commit.")
		return
	default:
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s, so there is nothing to commit.", host, rec.Seed.State))
		return
	}
	wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout)
	defer wcancel()
	if _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedCommit(rec.Seed.ID)); err != nil {
		// An overlapping commit may have won, between the gate's read and
		// this write: the import is committed, and this is a retry like any
		// other. The winner audited and bumped, so this writes nothing.
		if lostTransition(err) {
			if now := s.importNow(wctx, acct.Sqid, app, host, rec.Seed.ID); now != nil && now.State == seeds.StateCommitted {
				writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted})
				return
			}
		}
		s.importTransitionFailure(w, app, host, err)
		return
	}
	s.auditActivity(acct, app, principal, Activity{Action: "import-committed", TargetType: "instance", Target: host,
		Detail: fmt.Sprintf("committed the import of %s", host)})
	s.bumpSerial(wctx)
	writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted})
}

// pastStopped reports whether an import's state is one it only reaches
// after a stopped: the box has the stop, or has gone further.
func pastStopped(state string) bool {
	switch state {
	case seeds.StateReady, seeds.StateSwapped, seeds.StatePlaced, seeds.StateCommitted:
		return true
	}
	return false
}

// importDeleteFailed is the delete door's retry sentence when its import
// rule cannot read the record or finish the import (preflight M15). Main's
// own delete sentence (apiinstances.go) is unchanged.
const importDeleteFailed = "could not delete the instance. Try again."

// importFinishTimeout bounds a finish run inline for a request: the
// store's teardown has its own storeTeardownTimeout, and the reads, the
// snapshot delete and the record delete around it need a little more.
const importFinishTimeout = storeTeardownTimeout + instanceStoreTimeout

// committedRefusal is abort after commit (SD-N56).
func committedRefusal(host string) string {
	return fmt.Sprintf("%s is committed. Delete it with carlos instances delete instead.", host)
}

// finishImport finishes a rolled-back import inline for a request: the
// request's values, never its cancellation, so a client that leaves cannot
// cut a teardown midway (preflight M15).
func (s *Server) finishImport(r *http.Request, sqid, app, host, id string) error {
	ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), importFinishTimeout)
	defer cancel()
	return s.finishSeed(ctx, sqid, app, host, id)
}

// readDeletedRecord is the delete door's read of host's record for its
// import rule, with the etag the ordinary delete is then bound to. Only
// the store read can fail it (preflight M15, review I1): a body that will
// not decode, or that disagrees with its own path, is no import this
// console can finish, so it answers nil and the ordinary delete removes
// it as it always has.
func (s *Server) readDeletedRecord(ctx context.Context, sqid, app, host string) (*provision.Record, string, error) {
	ctx, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
	defer cancel()
	raw, etag, err := s.cfg.Releases.Get(ctx, provision.RecordKey(sqid, app, host))
	if err != nil {
		return nil, "", err
	}
	rec, err := provision.DecodeRecord(raw)
	if err == nil && (rec.Account != sqid || rec.App != app || rec.Host != host) {
		err = errors.New("the record disagrees with its own path")
	}
	if err != nil {
		log.Printf("console: api instances delete %s/%s/%s: the record could not be read as an import (%v), so it is deleted the ordinary way", sqid, app, host, err)
		return nil, etag, nil
	}
	return &rec, etag, nil
}

// handleAPIImportAbort abandons an import before it is committed (spec
// §5.6): a record CAS to aborted, then the box rolls it back. The
// transition is provision.SeedAbort, which allows every state before
// committed and refuses while a run is open, in the same CAS (preflight
// M14: one rule, in provision). The no-reason form keeps a refused
// import's reason.
//
// It is safe to repeat. An import already aborted answers 200 aborted; one
// the box has rolled back is finished here (its store, its snapshot, then
// its record) and answers 200 rolled-back, even when the finish fails and
// is left to the next abort or the sweep. An abort that loses its race
// re-reads once and answers by what won.
func (s *Server) handleAPIImportAbort(w http.ResponseWriter, r *http.Request) {
	principal, acct, app, host, rec, ok := s.importGate(w, r)
	if !ok {
		return
	}
	id := rec.Seed.ID
	if s.abortAnswered(w, r, acct.Sqid, app, host, rec.Seed) {
		return
	}
	wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout)
	defer wcancel()
	_, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedAbort(id))
	switch {
	case errors.Is(err, provision.ErrRunOpen):
		jsonError(w, http.StatusConflict, fmt.Sprintf("carlos run is busy on %s. Wait for it to finish, then abort.", host))
		return
	case lostTransition(err):
		// Another request got there first. Answer as a sequential abort
		// would be answered after it; anything but this import is SD-N60.
		if now := s.importNow(wctx, acct.Sqid, app, host, id); now != nil && s.abortAnswered(w, r, acct.Sqid, app, host, now) {
			return
		}
		s.importTransitionFailure(w, app, host, err)
		return
	case err != nil:
		s.importTransitionFailure(w, app, host, err)
		return
	}
	s.auditActivity(acct, app, principal, Activity{Action: "import-aborted", TargetType: "instance", Target: host,
		Detail: fmt.Sprintf("aborted the import of %s", host)})
	s.bumpSerial(wctx)
	writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateAborted})
}

// abortAnswered answers an abort of seed that needs no transition, and
// reports whether it did: committed is refused (SD-N56), aborted is done
// already, and rolled-back is finished here. A finish that fails is
// logged and still answers 200: the abort itself has landed, and the next
// abort or the sweep finishes it.
func (s *Server) abortAnswered(w http.ResponseWriter, r *http.Request, sqid, app, host string, seed *provision.SeedSource) bool {
	switch seed.State {
	case seeds.StateCommitted:
		jsonError(w, http.StatusConflict, committedRefusal(host))
	case seeds.StateAborted:
		writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateAborted})
	case seeds.StateRolledBack:
		if err := s.finishImport(r, sqid, app, host, seed.ID); err != nil {
			log.Printf("console: import %s: finish seed %s on abort: %v", host, seed.ID, err)
		}
		writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateRolledBack})
	default:
		return false
	}
	return true
}

// refuseImportDelete is the delete door's import rule (spec §5.6, codex
// R2-7), run before the ordinary delete. answered: the delete must not go
// the ordinary way, and either this function has answered it (a refusal or
// a failure), or, with finished, the delete was done here (the import
// finished, or the ordinary delete bound to this read) and the caller
// writes the ordinary delete's audit row and answer. (false, false) is
// left only to a claim console, whose ordinary delete is the claim's.
//
//   - awaiting-upload, uploaded, refused: no box has swapped, so the record
//     moves straight to rolled-back and is finished here (its store, its
//     snapshot under seeds/<host>/<id>/, then the record).
//   - rolled-back: finished here.
//   - any other uncommitted state: 409, abort the import instead (SD-N57).
//   - committed, a --from seed, an unseeded record, or a record that will
//     not decode: the ordinary delete, conditional on the etag read here
//     (deleteRead), answered as finished.
//   - no record at all: finished, with nothing deleted (final review F7),
//     except on a claim console, where the claim's retirement answers.
//
// It fails closed (preflight M15): a store read that fails, a seed id that
// cannot name a key (M8) and a finish that fails answer main's delete
// sentence, so a read blip never lets the ordinary delete drop an
// uncommitted import's record and leak what its seed made. A body that
// was read but will not decode, or disagrees with its path, is no import
// anything can finish, so the ordinary delete removes it as before
// (review I1). On a claim
// console, where the finish runs nowhere, an uncommitted import is refused
// as every import door refuses there.
func (s *Server) refuseImportDelete(w http.ResponseWriter, r *http.Request, acct *Account, app, host string) (answered, finished bool) {
	return s.importDeleteRule(w, r, acct, app, host, true)
}

// importDeleteRule is refuseImportDelete; again is whether a record that
// changed under the ordinary delete may be read and decided once more.
//
// The ordinary delete is bound to what the rule read (final review F7):
// absence is the delete's own idempotent answer, so no delete runs that
// could remove an import begun since; and otherwise the record is deleted
// only while it is the one the rule read (deleteRead).
func (s *Server) importDeleteRule(w http.ResponseWriter, r *http.Request, acct *Account, app, host string, again bool) (answered, finished bool) {
	sqid := acct.Sqid
	rec, etag, err := s.readDeletedRecord(r.Context(), sqid, app, host)
	if errors.Is(err, os.ErrNotExist) {
		// With host claims on, a record can be gone while its retirement
		// is not (parking, with obligations left): the claim's own flow
		// answers that, pending or done (re-review finding 10).
		if s.cfg.HostClaim {
			return false, false
		}
		return true, true
	}
	if err != nil {
		log.Printf("console: api instances delete %s/%s/%s: read the record: %v", sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, importDeleteFailed)
		return true, false
	}
	if rec == nil || !rec.Seed.Import() || rec.Seed.State == seeds.StateCommitted {
		return s.deleteRead(w, r, acct, app, host, etag, again)
	}
	if s.cfg.HostClaim {
		ref := seedClaimConsoleRefusal()
		jsonError(w, ref.Status, ref.Msg)
		return true, false
	}
	id := rec.Seed.ID
	if err := seeds.Check(host, id); err != nil {
		log.Printf("console: api instances delete %s/%s/%s: import seed %q: %v", sqid, app, host, id, err)
		jsonError(w, http.StatusBadGateway, importDeleteFailed)
		return true, false
	}
	switch rec.Seed.State {
	case seeds.StateAwaitingUpload, seeds.StateUploaded, seeds.StateRefused:
		wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout)
		_, err := provision.SetRecordSeed(wctx, s.cfg.Releases, sqid, app, host, provision.SeedRolledBackBeforeSwap(id))
		var now *provision.Record
		var nowErr error
		if lostTransition(err) {
			now, nowErr = s.readImportRecord(wctx, sqid, app, host)
		}
		wcancel()
		switch {
		case err == nil:
		case errors.Is(err, os.ErrNotExist), errors.Is(nowErr, os.ErrNotExist):
			// Gone already, by this CAS's read or by another delete that
			// finished first: the delete's idempotent answer, with no
			// ordinary delete run after it (final review F7).
			return true, true
		case now != nil && now.Seed != nil && now.Seed.ID == id && now.Seed.State == seeds.StateRolledBack:
			// Another delete rolled it back first: finish it all the same.
		case lostTransition(err):
			s.importTransitionFailure(w, app, host, err) // SD-N60
			return true, false
		default:
			log.Printf("console: api instances delete %s/%s/%s: roll back import seed %s: %v", sqid, app, host, id, err)
			jsonError(w, http.StatusBadGateway, importDeleteFailed)
			return true, false
		}
	case seeds.StateRolledBack:
	case seeds.StateAborted:
		// Aborting is done: the box puts the old route back, and then a
		// delete finishes the import.
		jsonError(w, http.StatusConflict, fmt.Sprintf("the import of %s is already aborted. Once the box puts the old route back on its next pass, delete %s again.", host, host))
		return true, false
	default:
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s is an import that is not committed. Abort the import instead.", host))
		return true, false
	}
	if err := s.finishImport(r, sqid, app, host, id); err != nil {
		log.Printf("console: api instances delete %s/%s/%s: finish import seed %s: %v", sqid, app, host, id, err)
		jsonError(w, http.StatusBadGateway, importDeleteFailed)
		return true, false
	}
	return true, true
}

// deleteRead is the ordinary delete of host's record, conditional on the
// etag the import rule read it under (final review F7). A record gone
// since is deleted already. A record that changed since may now be an
// import begun after the rule's read: it is read and decided again, once,
// and a second change answers main's retry sentence.
//
// A claim console retires records through the claim
// (retireClaimHostRecord) and begins no import, so there the ordinary
// delete runs as before: (false, false).
func (s *Server) deleteRead(w http.ResponseWriter, r *http.Request, acct *Account, app, host, etag string, again bool) (answered, finished bool) {
	if s.cfg.HostClaim {
		return false, false
	}
	ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout)
	defer cancel()
	err := release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(acct.Sqid, app, host), etag)
	switch {
	case err == nil:
		return true, true
	case errors.Is(err, release.ErrCASConflict) && again:
		return s.importDeleteRule(w, r, acct, app, host, false)
	}
	log.Printf("console: api instances delete %s/%s/%s: %v", acct.Sqid, app, host, err)
	jsonError(w, http.StatusBadGateway, importDeleteFailed)
	return true, false
}