// The --from seed door (seeded-instances spec §4): `carlos instances create
// --from <host>` creates an instance whose database starts as a copy of a
// sibling's replica. handleAPIInstanceCreate runs every ordinary check,
// then hands over here. The door answers 202 once the record is reserved,
// and the copy finishes on its own (createFrom).

package console

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"log"
	"math"
	"net/http"
	"os"
	"path/filepath"
	"runtime/debug"
	"strconv"
	"strings"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/objstore"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// Seams for the tests.
var (
	seedMaxBytes    = seeds.MaxBytes
	planRestore     = dbdoor.PlanRestore
	restoreToTarget = dbdoor.RestoreToTarget
	// seedRestoreDeadline bounds the restore and its check (spec §4 step 5).
	seedRestoreDeadline = 30 * time.Minute
	// restoreWatchEvery is how often the size watcher looks (spec §4 step 5).
	restoreWatchEvery = time.Second
)

// seedUploadDeadline bounds the snapshot's one PUT, after the restore's own
// deadline: up to 4 GiB to the region's own bucket. The record is reserving
// for the whole restore, store walk and upload, and the sweep aborts a
// --from that has been reserving for an hour (spec §3.1), so the three
// together stay well under that.
const seedUploadDeadline = 20 * time.Minute

// seedFromReq is the create body's "from" (spec §4). TXID 0 is the latest
// copy; the CLI refuses a non-positive --at.
type seedFromReq struct {
	Host string `json:"host"`
	TXID uint64 `json:"txid,omitempty"`
}

// seedCreate is what handleAPIInstanceCreate has decided by the time it
// hands a seed door over: every ordinary check has passed. siblingLabel is
// the create door's own (set when the host sits at another label of the
// account's platform namespace), for its re-check before publishing.
type seedCreate struct {
	acct                    *Account
	app, principal, message string
	m                       *provision.Manifest
	rec                     provision.Record
	placed, regionFrom      string
	siblingLabel            string
}

// seedCommonRefusal is both seed doors' shape rule: an ordinary instance.
func seedCommonRefusal(kind string, addrs []string, role string, replicas []string) *doorRefusal {
	if (kind != "" && kind != "instance") || len(addrs) > 0 || role != "" || len(replicas) > 0 {
		return seedOrdinaryRefusal()
	}
	return nil
}

// seedOrdinaryRefusal is SD-N36, the shape rule's answer.
func seedOrdinaryRefusal() *doorRefusal {
	return &doorRefusal{http.StatusBadRequest, "a seeded instance is an ordinary instance. Drop --kind, --addr and --role."}
}

// seedNoDataRefusal is both doors' answer on a console with no data bucket
// for seeds.
func seedNoDataRefusal() *doorRefusal {
	return &doorRefusal{http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot create a seeded instance."}
}

// seedPlacementRefusal is both seed doors' placement rule: the platform
// keeps a replica only of an instance it hosts itself.
func seedPlacementRefusal(c seedCreate) *doorRefusal {
	if c.placed != "" || c.m.Pool != "" {
		return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s places its instances in a pool or on a fleet, so it cannot take a seeded instance.", c.app)}
	}
	return nil
}

// seedOwnEnvironment is SD-N33, the rule a seeded instance's environment
// keeps (spec §4 Storage).
const seedOwnEnvironment = "a seeded instance gets its own environment and an empty store"

// refuseSeededEnvironment is both seed doors' answer to an explicit
// --environment on an app with private storage (spec §4 Storage).
func (s *Server) refuseSeededEnvironment(w http.ResponseWriter) {
	jsonError(w, http.StatusConflict, seedOwnEnvironment)
}

// seedStoreTakenRefusal is the refusal of a seeded create whose environment
// is not new: SD-N34 naming the store's bucket, or, when the environment
// has no store with a bucket and is only named by another instance (final
// review F6), the sentence that says so and what to do.
func seedStoreTakenRefusal(env, app string, store *objstore.Record) *doorRefusal {
	if bucket := storeBucketName(store); bucket != "" {
		return &doorRefusal{http.StatusConflict, fmt.Sprintf("environment %s already has a store (%s). A seeded instance needs a new, empty one.", env, bucket)}
	}
	return &doorRefusal{http.StatusConflict, fmt.Sprintf("environment %s is already used by another instance of %s. A seeded instance gets a new environment named after its host, so choose another host.", env, app)}
}

// seedSiblingApexReason is the sentence both seed doors speak when an app
// created under the host's label while the instance was being written has
// made the host its platform apex. Main's create door keeps its own.
func seedSiblingApexReason(host, label string) string {
	return fmt.Sprintf("%s is app %s's platform apex. That app was created while this instance was being written, so the instance was not kept.", host, label)
}

// errSeedEnvironmentUnread is instanceStoreProvisionFor's answer when it
// cannot tell whether another instance names a seed's environment. It
// answers nothing on w: the seed door speaks its own retry sentence.
var errSeedEnvironmentUnread = errors.New("console: the instances naming the environment could not be read")

// seedEnvironmentUser names an instance of app, other than seed seedID's
// own, whose record names env, or "" when none does (final review F6). A
// seeded instance's derived environment must be unused: its store is new
// and empty and its seed's finish may tear it down, which another
// instance's environment must never be. It reads strictly
// (provision.ListRecordsStrict): a listing it cannot complete, too many
// records, or a record that will not decode is an error, since any of them
// may name env.
func (s *Server) seedEnvironmentUser(ctx context.Context, sqid, app, env, seedID string) (string, error) {
	recs, unreadable, err := provision.ListRecordsStrict(ctx, s.cfg.Releases, sqid, app, instanceListMax)
	if err != nil {
		return "", err
	}
	if len(unreadable) > 0 {
		return "", fmt.Errorf("%d records of %s/%s could not be read, so one may name environment %q", len(unreadable), sqid, app, env)
	}
	for _, r := range recs {
		if r.Environment == env && (r.Seed == nil || r.Seed.ID != seedID) {
			return r.Host, nil
		}
	}
	return "", nil
}

// seedStoreRefusal is spec §4 Storage's existing-store rule for both doors,
// checked before the record is reserved. instanceStoreProvisionFor checks
// it again under the store lock. failed is the door's own retry sentence,
// answered with a 502 when the store record cannot be read.
func (s *Server) seedStoreRefusal(ctx context.Context, c seedCreate, seedID, failed string) *doorRefusal {
	if !c.m.PrivateStorage {
		return nil
	}
	rec, _, err := objstore.GetIn(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.rec.Environment)
	if err != nil {
		log.Printf("console: seed store check %s/%s/%s: %v", c.acct.Sqid, c.app, c.rec.Environment, err)
		return &doorRefusal{http.StatusBadGateway, failed}
	}
	if !seedOwnsStore(rec, seedID) {
		return seedStoreTakenRefusal(c.rec.Environment, c.app, rec)
	}
	user, err := s.seedEnvironmentUser(ctx, c.acct.Sqid, c.app, c.rec.Environment, seedID)
	if err != nil {
		log.Printf("console: seed store check %s/%s/%s: who names the environment: %v", c.acct.Sqid, c.app, c.rec.Environment, err)
		return &doorRefusal{http.StatusBadGateway, failed}
	}
	if user != "" {
		log.Printf("console: seed store check %s/%s/%s: instance %s already names the environment", c.acct.Sqid, c.app, c.rec.Environment, user)
		return seedStoreTakenRefusal(c.rec.Environment, c.app, rec)
	}
	return nil
}

// seedGiB is SD-N26's size, rounded up to a tenth so a size over the limit
// never prints as the limit itself.
func seedGiB(n int64) string {
	return fmt.Sprintf("%.1f GiB", math.Ceil(float64(n)/(1<<30)*10)/10)
}

// fromCopy is a --from copy whose record is reserved: everything runFrom
// needs once the request has been answered. It owns the scratch
// reservation (release) from the moment beginFrom hands it over.
type fromCopy struct {
	ctx     context.Context // the request's values, never its cancellation
	c       seedCreate
	from    seedFromReq
	source  string
	id      string
	plan    dbdoor.RestorePlan
	client  *dbdoor.Client
	release func()
}

// createFrom is --from (spec §4). Every refusal up to the record's
// reservation is answered here, synchronously, after the scratch
// reservation is released. Once the record is reserved the door answers
// 202 with the reserving record and the copy runs on, detached from the
// request (runFrom): a copy can take most of an hour, and the edge's
// backhaul, the CLI's client and idle NATs all cut a response that keeps
// its headers longer than 30 seconds (Plan 5 ruling R2, a deviation from
// spec §4 for Paul). The caller follows the record until its seed is ready
// or aborted, with the reason in Seed.Reason.
func (s *Server) createFrom(w http.ResponseWriter, r *http.Request, c seedCreate, from seedFromReq) {
	held := &heldResponse{header: w.Header()}
	job := s.beginFrom(held, r, c, from)
	if job != nil {
		s.seedCopies.Add(1)
		go s.runFrom(job)
	}
	held.flush(w, fmt.Sprintf("could not copy %s. Nothing was created. Try again.", strings.TrimSpace(from.Host)))
}

// beginFrom is the synchronous half: every check, the plan and its
// admission, the store gates that write nothing, and the record's
// reservation. It returns the copy to run, or nil when it has answered a
// refusal on w.
func (s *Server) beginFrom(w http.ResponseWriter, r *http.Request, c seedCreate, from seedFromReq) *fromCopy {
	ctx := r.Context()
	acct, app, host := c.acct, c.app, c.rec.Host
	source := strings.TrimSpace(from.Host)
	failed := fmt.Sprintf("could not copy %s. Nothing was created. Try again.", source)
	if ref := seedPlacementRefusal(c); ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return nil
	}
	if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" {
		ref := seedNoDataRefusal()
		jsonError(w, ref.Status, ref.Msg)
		return nil
	}
	checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
	defer cancel()
	if ref := s.seedRegionReady(checks, c.rec.Region, false); ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return nil
	}
	src, err := s.resolveDBSource(checks, acct, c.principal, app, source)
	if ref := s.fromSourceRefusal(src, source, app, err); ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return nil
	}
	if src.rec.Region != c.rec.Region {
		jsonError(w, http.StatusConflict, "a copy stays in its source's region. To run it elsewhere, create it here and then run carlos move.")
		return nil
	}
	rd, err := s.dbReplica(ctx, src.prefix)
	if err != nil {
		log.Printf("console: from %s: replica client: %v", source, err)
		jsonError(w, http.StatusBadGateway, failed)
		return nil
	}
	client := dbdoor.NewClient(replicaDoer(rd))
	planCtx, planCancel := context.WithTimeout(ctx, instanceStoreTimeout)
	plan, err := planRestore(planCtx, client, from.TXID)
	planCancel()
	switch {
	case from.TXID != 0 && (errors.Is(err, dbdoor.ErrHistoryGone) || errors.Is(err, dbdoor.ErrNothingReplicated)):
		jsonError(w, http.StatusConflict, fmt.Sprintf("history before %d is no longer kept", from.TXID))
		return nil
	case errors.Is(err, dbdoor.ErrNothingReplicated):
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s has not replicated anything yet. Wake it once, then try again.", source))
		return nil
	case err != nil:
		// A file that vanished under the plan, a header that does not
		// describe its file, a bucket read: a retry plans again.
		log.Printf("console: from %s: plan: %v", source, err)
		jsonError(w, http.StatusBadGateway, failed)
		return nil
	case from.TXID > plan.TXID:
		// The plan clamps an --at past the latest transaction to the
		// latest copy. A caller who named a point later than anything the
		// source has replicated is told so, and nothing is reserved.
		jsonError(w, http.StatusConflict, fmt.Sprintf("%s has replicated up to txid %d, so there is no txid %d to copy yet. Leave out --at to copy the latest.", source, plan.TXID, from.TXID))
		return nil
	}
	if plan.RestoredBytes > seedMaxBytes {
		jsonError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("a copy of %s would be %s, over the 4 GiB limit.", source, seedGiB(plan.RestoredBytes)))
		return nil
	}
	releaseScratch, ref := s.seedScratch.reserve(acct.Sqid+"/"+app, plan.RestoredBytes+plan.LTXBytes)
	if ref != nil {
		jsonError(w, ref.Status, ref.Msg)
		return nil
	}
	handedOver := false
	defer func() {
		if !handedOver {
			releaseScratch()
		}
	}()
	id, err := seeds.NewID()
	if err != nil {
		log.Printf("console: from %s: seed id: %v", source, err)
		jsonError(w, http.StatusBadGateway, failed)
		return nil
	}
	if c.m.PrivateStorage {
		// The store gates that write nothing run before the reservation,
		// so their refusals leave no aborted record holding the host
		// (ruling R3). The walk runs them again, after it.
		if _, ok := s.instanceStorePreflight(ctx, w, acct, app); !ok {
			return nil
		}
		storeCtx, storeCancel := context.WithTimeout(ctx, instanceStoreTimeout)
		ref = s.seedStoreRefusal(storeCtx, c, id, failed)
		storeCancel()
		if ref != nil {
			jsonError(w, ref.Status, ref.Msg)
			return nil
		}
	}

	// Reserve the record (spec §4 step 7.1). It is reserved here, after the
	// size, scratch and store checks rather than straight after resolving
	// the source: a refusal before this point leaves nothing behind, where
	// an earlier record would turn each into an aborted record holding the
	// host until it is finished (a recorded deviation, Plan 5 M10).
	rec := c.rec
	rec.Seed = &provision.SeedSource{ID: id, Kind: seeds.KindReplica, From: source, State: seeds.StateReserving,
		By: c.principal, At: time.Now().UTC()}
	// Not the request's cancellation: a client leaving now must not turn a
	// write that landed into an error, which would answer 502 and start no
	// copy while the reserving record held the host.
	writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(ctx), instanceStoreTimeout)
	defer writeCancel()
	if err := provision.CreateRecord(writeCtx, s.cfg.Releases, rec); err != nil {
		if errors.Is(err, release.ErrCASConflict) {
			jsonError(w, http.StatusConflict, "instance "+host+" already exists")
			return nil
		}
		log.Printf("console: from %s: reserve %s: %v", source, host, err)
		jsonError(w, http.StatusBadGateway, failed)
		return nil
	}
	out := recordView(rec)
	out.RegionFrom = c.regionFrom
	out.RegionName = s.placeOf(rec.Region, s.candsOrHome(writeCtx)) // never fleet-placed: refused above
	writeJSON(w, http.StatusAccepted, out)
	handedOver = true
	return &fromCopy{ctx: context.WithoutCancel(ctx), c: c, from: from, source: source,
		id: id, plan: plan, client: client, release: releaseScratch}
}

// runFrom is the detached half: the seed's store, the restore, the upload
// and the publish. It never depends on the request, which has been
// answered. Every failure moves the record to aborted with the sentence
// the synchronous door would have answered in Seed.Reason, for the box
// and the console's finish to clean up (spec §4 step 7). On every exit it
// removes its scratch and then releases the reservation.
//
// The console has no shutdown hook to cancel a running copy: a restart
// leaves the record reserving, its scratch is cleared at start, and the
// sweep aborts it (Plan 8).
func (s *Server) runFrom(j *fromCopy) {
	defer s.seedCopies.Done()
	defer j.release()
	ctx := j.ctx
	c := j.c
	acct, app, host, source := c.acct, c.app, c.rec.Host, j.source
	// The reason a copy that failed after the answer leaves on its record.
	// The aborted record holds the host until the box and the next sweep
	// finish it, so a retry before then finds the host taken.
	failed := fmt.Sprintf("could not copy %s, so no instance was created. %s is free again within about 5 minutes. Try again then.", source, host)
	abort := func(reason, why string) {
		bg, cancel := context.WithTimeout(ctx, instanceStoreTimeout)
		defer cancel()
		if _, err := provision.SetRecordSeed(bg, s.cfg.Releases, acct.Sqid, app, host, provision.SeedAbortBecause(j.id, reason)); err != nil {
			log.Printf("console: from %s: abort %s (seed %s) after %s: %v", source, host, j.id, why, err)
		}
		s.bumpSerial(bg)
	}
	// Nothing above this copy contains a panic any more: net/http's
	// handler recover covered the synchronous door, and a panic here (in
	// the restore's decode of replica data, the check, the store walk, the
	// upload) would end the whole console. It aborts this seed instead.
	// Registered after release and Done, so it runs before them; scratch
	// removal, registered later, has already run.
	defer func() {
		if p := recover(); p != nil {
			log.Printf("console: from %s: panic in the copy of %s (seed %s): %v\n%s", source, host, j.id, p, debug.Stack())
			abort(failed, "a panic")
		}
	}()

	if c.m.PrivateStorage {
		// Its refusals are written to a held answer nobody reads but this:
		// the sentence becomes the seed's reason.
		held := &heldResponse{header: http.Header{}}
		storeRec, ok, err := s.instanceStoreProvisionFor(ctx, held, acct, app, c.rec.Environment, c.principal, j.id)
		if !ok {
			reason := held.message(failed)
			switch {
			case errors.Is(err, errSeedStoreTaken):
				reason = seedStoreTakenRefusal(c.rec.Environment, c.app, storeRec).Msg
			case errors.Is(err, errSeedEnvironmentUnread):
				reason = failed
			}
			abort(reason, "store provisioning")
			return
		}
	}

	dir, err := os.MkdirTemp(s.seedScratch.dir(), seedScratchPrefix)
	if err != nil {
		log.Printf("console: from %s: scratch: %v", source, err)
		abort(seedScratchFull().Msg, "scratch")
		return
	}
	defer os.RemoveAll(dir)
	// Pinned at the planned TXID, never "latest" again: only that restore
	// produces the size that was admitted, however far the source has moved
	// on since it was planned.
	plan := j.plan
	rctx, rcancel := context.WithTimeout(ctx, seedRestoreDeadline)
	defer rcancel()
	grew := restoreWatch(rctx, rcancel, dir, plan.RestoredBytes, plan.RestoredBytes+plan.LTXBytes)
	restored, err := restoreToTarget(rctx, j.client, dir, plan.TXID)
	timedOut := errors.Is(rctx.Err(), context.DeadlineExceeded)
	rcancel() // stops the watcher
	overgrown := grew()
	if err == nil && !overgrown {
		// The watcher's tick can come after a fast restore has finished.
		st, serr := os.Stat(restored.Path)
		switch {
		case serr != nil:
			err = serr
		case st.Size() > plan.RestoredBytes:
			overgrown = true
		}
	}
	if err != nil || overgrown {
		reason := failed
		switch {
		case overgrown:
			reason = fmt.Sprintf("the copy of %s grew past its planned size, so it was stopped and no instance was created.", source)
		case timedOut:
			reason = fmt.Sprintf("copying %s took longer than 30 minutes, so no instance was created.", source)
		case errors.Is(err, dbdoor.ErrCheckFailed):
			reason = fmt.Sprintf("the copy of %s failed its integrity check, so no instance was created.", source)
		case errors.Is(err, dbdoor.ErrHistoryGone) && plan.Target != 0:
			reason = fmt.Sprintf("history before %d is no longer kept", j.from.TXID)
		default:
			// Including history lost under a latest copy between its plan
			// and its restore: a retry plans again.
			log.Printf("console: from %s: restore: %v", source, err)
		}
		abort(reason, "restore")
		return
	}

	uctx, ucancel := context.WithTimeout(ctx, seedUploadDeadline)
	sha, size, err := uploadSnapshot(uctx, s.cfg.SeedData, restored.Path, seeds.SnapshotKey(host, j.id))
	ucancel()
	if err != nil {
		log.Printf("console: from %s: upload %s: %v", source, host, err)
		abort(failed, "upload")
		return
	}
	// The create door's re-check after its write (apiinstances.go), made
	// here before the publish: a reserving record mints nothing, so the
	// publish is the point of no return.
	if c.siblingLabel != "" {
		if live, trashed, err := s.store.AppNameState(acct.ID, c.siblingLabel); err == nil && (live || trashed) {
			abort(seedSiblingApexReason(host, c.siblingLabel), "a sibling app took the label")
			return
		}
	}
	pctx, pcancel := context.WithTimeout(ctx, instanceStoreTimeout)
	defer pcancel()
	if _, err := provision.SetRecordSeed(pctx, s.cfg.Releases, acct.Sqid, app, host,
		provision.SeedPublishReplica(j.id, strconv.FormatUint(restored.TXID, 10), sha, size)); err != nil {
		log.Printf("console: from %s: publish %s: %v", source, host, err)
		abort(failed, "publish")
		return
	}
	s.auditActivity(acct, app, c.principal, Activity{
		Action: "instance-seed", TargetType: "instance", Target: host, Message: c.message,
		Detail: fmt.Sprintf("seeded %s from %s at txid %d", host, source, restored.TXID),
	})
	s.bumpSerial(pctx)
}

// fromSourceRefusal maps resolveDBSource's answer to --from's sentences.
func (s *Server) fromSourceRefusal(src dbSource, source, app string, err error) *doorRefusal {
	switch {
	case err == nil:
		return nil
	case errors.Is(err, os.ErrNotExist), errors.Is(err, ErrBadHost):
		return &doorRefusal{http.StatusConflict, fmt.Sprintf("no instance %s in %s. A copy can only come from an instance of the same app in the same account.", source, app)}
	case errors.Is(err, errSourceClass):
		return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is not an instance the platform keeps a replica of, so it cannot be copied.", source)}
	case errors.Is(err, ErrPooled):
		return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s runs in the pool %s, so the platform keeps no replica of it to copy.", source, src.rec.Pool)}
	case errors.Is(err, ErrResidency):
		if fleet, ok := registry.FleetPlacement(src.rec.Region); ok {
			return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s runs on your fleet %s, so the platform keeps no replica of it to copy.", source, fleet)}
		}
		return &doorRefusal{http.StatusConflict, s.dbResidencyRefusal(src.rec)}
	case errors.Is(err, errSourceClaim):
		// Unreachable while seedRegionReady refuses a claim console first.
		return seedClaimConsoleRefusal()
	}
	log.Printf("console: from %s: resolve: %v", source, err)
	return &doorRefusal{http.StatusBadGateway, fmt.Sprintf("could not copy %s. Nothing was created. Try again.", source)}
}

// uploadSnapshot hashes path and writes it to key with one create-only
// PutFrom (spec §4 step 7.2: seekable, under S3's 5 GB single-PUT limit).
func uploadSnapshot(ctx context.Context, data release.Store, path, key string) (sha string, size int64, err error) {
	f, err := os.Open(path)
	if err != nil {
		return "", 0, err
	}
	defer f.Close()
	h := sha256.New()
	if size, err = io.Copy(h, f); err != nil {
		return "", 0, err
	}
	if _, err := f.Seek(0, io.SeekStart); err != nil {
		return "", 0, err
	}
	if _, err := release.PutObjectFrom(ctx, data, key, f, size, ""); err != nil {
		return "", 0, err
	}
	return hex.EncodeToString(h.Sum(nil)), size, nil
}

// restoreWatch is the size watcher (spec §4 step 5). Every
// restoreWatchEvery it measures the restore under dir and cancels it when
// the database (litestream's replica.db.tmp while it decodes, replica.db
// after its rename) grows past dbLimit, or everything the restore holds
// (the database and its downloaded LTX files) grows past totalLimit, the
// reservation. grew reports whether it fired.
func restoreWatch(ctx context.Context, cancel context.CancelFunc, dir string, dbLimit, totalLimit int64) (grew func() bool) {
	var fired atomic.Bool
	every := restoreWatchEvery
	go func() {
		t := time.NewTicker(every)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
			}
			if db, total := restoreScratchUse(dir); db > dbLimit || total > totalLimit {
				fired.Store(true)
				cancel()
				return
			}
		}
	}()
	return fired.Load
}

// restoreScratchUse measures a restore under dir: the largest database file
// (replica.db or replica.db.tmp, dbdoor's names) and every regular file
// together. SQLite's shared-memory index, which the read-only check opens
// beside the copy, is not the restore's and is not counted.
func restoreScratchUse(dir string) (db, total int64) {
	_ = filepath.WalkDir(dir, func(_ string, d fs.DirEntry, err error) error {
		if err != nil || !d.Type().IsRegular() || strings.HasSuffix(d.Name(), "-shm") {
			return nil // a file the restore removed between listing and stat is not there
		}
		info, err := d.Info()
		if err != nil {
			return nil
		}
		total += info.Size()
		if n := d.Name(); n == "replica.db" || n == "replica.db.tmp" {
			db = max(db, info.Size())
		}
		return nil
	})
	return db, total
}

// heldResponse holds a door's answer until its cleanup has run.
type heldResponse struct {
	header http.Header
	status int
	body   bytes.Buffer
}

func (h *heldResponse) Header() http.Header { return h.header }

func (h *heldResponse) WriteHeader(code int) {
	if h.status == 0 {
		h.status = code
	}
}

func (h *heldResponse) Write(b []byte) (int, error) {
	h.WriteHeader(http.StatusOK)
	return h.body.Write(b)
}

// flush writes what was held. A door that held nothing answers failed
// with a 502, never an empty 200.
func (h *heldResponse) flush(w http.ResponseWriter, failed string) {
	if h.status == 0 {
		jsonError(w, http.StatusBadGateway, failed)
		return
	}
	w.WriteHeader(h.status)
	_, _ = w.Write(h.body.Bytes())
}

// message is the held refusal's sentence, or fallback when nothing usable
// was held.
func (h *heldResponse) message(fallback string) string {
	var out struct {
		Error string `json:"error"`
	}
	if err := json.Unmarshal(h.body.Bytes(), &out); err != nil || out.Error == "" {
		return fallback
	}
	return out.Error
}