// The forwarded `carlos db` read (db region forward spec
// docs/superpowers/specs/2026-10-22-db-region-forward-design.md): the list
// door's marker for a cross-region instance, and the one route that admits a
// read, queues it for a box in the data's region, and relays the box's sealed
// answer inside the same request.
package console

import (
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"slices"
	"strings"
	"time"
	"unicode/utf8"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbquery"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// dbForwardTimes is the forwarded read's timing (spec §5.2, §5.5). Tests
// shrink it; production uses defaultDBForwardTimes.
type dbForwardTimes struct {
	poll      time.Duration // how often the console reads picked, result and chunks
	wait      time.Duration // a progress line at least this often
	stall     time.Duration // a write or flush that blocks this long ends the request
	preHeader time.Duration // everything before the 200 (backhaul waits 30 s for headers)
	pickup    time.Duration // picked must appear this long after issued_at
	grace     time.Duration // after the owner's deadline
	bound     time.Duration // the request's own bound after issued_at
	call      time.Duration // one store call inside the stream (plan review finding 12)

	// beforeWrite, nil in production, runs just before admission checks the
	// request's context ahead of each write ("audit", "request", "queue",
	// "serial"): a test ends the request there (plan review finding 28).
	beforeWrite func(step string)
}

func defaultDBForwardTimes() dbForwardTimes {
	return dbForwardTimes{
		poll: time.Second, wait: 15 * time.Second, stall: 30 * time.Second, preHeader: 20 * time.Second,
		pickup: dbreads.ConsoleWaitsPicked, grace: dbreads.OwnerGrace, bound: dbreads.RequestBound,
		call: 5 * time.Second,
	}
}

// dbForwardSigner is the root signer when this console forwards: the switch
// on, no host claims (a host-claim console answers with the residency
// sentence, as it refuses carlos run), and a key whose public half is the
// deployment root a box verifies against. A console holding only a scoped
// key cannot sign a request, and so cannot forward (spec §5.5).
func (s *Server) dbForwardSigner() (release.Signer, bool) {
	if !s.cfg.DBForward || s.cfg.HostClaim || s.cfg.Key == nil {
		return release.Signer{}, false
	}
	pub, ok := s.cfg.Key.Public().(ed25519.PublicKey)
	if !ok || !pub.Equal(s.cfg.ReleasePub) {
		return release.Signer{}, false
	}
	return release.RootSigner(s.cfg.Key), true
}

const (
	dbForwardNone = "none" // DBF-1: no capable, non-draining box publishes in the region
	dbForwardDead = "dead" // DBF-2: capable boxes, none live
)

// dbForwardEvidence is spec §7.1's rule. It reads Complete and
// BoxesComplete, never the error alone: ReadLiveSnapshot answers a nil
// error when status snapshots cannot be loaded and reports the gap in those
// flags (codex 10). Incomplete evidence comes first and wins over both
// refusals: the request is written and the pickup window decides.
func dbForwardEvidence(snap edgesSnapshot, err error, region string) string {
	if err != nil || !snap.Complete || !snap.BoxesComplete {
		return ""
	}
	capable, live := 0, 0
	for _, f := range snap.Files {
		if f.Placement != region || f.Draining || !slices.Contains(f.Capabilities, dbreads.Capability) {
			continue
		}
		capable++
		if f.Fresh {
			live++
		}
	}
	switch {
	case capable == 0:
		return dbForwardNone
	case live == 0:
		return dbForwardDead
	}
	return ""
}

// The console's forwarded-read sentences (copy record DBF-1, DBF-2, DBF-14),
// approved by Luna on 2026-10-23.
const (
	dbfNoCapable    = "%s keeps its data in region %s. no server in that region can read databases yet, so carlos db cannot read it for now."
	dbfNoneLive     = "%s keeps its data in region %s. no server in that region is answering right now. try again in a few minutes."
	dbfUpdateCarlos = "%s keeps its data in region %s. this version of carlos cannot read it from there. run carlos update, then try again."
)

// forwardable is whether this console forwards reads of rec: it can sign,
// and rec is in a named platform region (a fleet placement keeps its own
// sentence, and a home-region record seen from a regional console has no
// region to name).
func (s *Server) forwardable(rec provision.Record) bool {
	_, fleet := registry.FleetPlacement(rec.Region)
	_, ok := s.dbForwardSigner()
	return ok && !fleet && dbreads.ValidRegion(rec.Region)
}

// dbResidency answers a record homed outside this console's region (spec
// §5.5 step 1, §9.2). Not forwardable: main's refusal, byte for byte.
// Forwardable: DBF-1 or DBF-2 on complete evidence, else the forward marker,
// whose sentence an old CLI prints (DBF-14).
func (s *Server) dbResidency(w http.ResponseWriter, r *http.Request, rec provision.Record) {
	if !s.forwardable(rec) {
		jsonError(w, http.StatusConflict, s.dbResidencyRefusal(rec))
		return
	}
	snap, err := s.liveEdges(r.Context())
	switch dbForwardEvidence(snap, err, rec.Region) {
	case dbForwardNone:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoCapable, rec.Host, rec.Region))
		return
	case dbForwardDead:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoneLive, rec.Host, rec.Region))
		return
	}
	writeJSON(w, http.StatusConflict, dbdoor.ForwardBody{
		Error: fmt.Sprintf(dbfUpdateCarlos, rec.Host, rec.Region), Forward: 1,
		Region: rec.Region, ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull},
	})
}

// The forwarded read's other console sentences (copy record DBF-9, DBF-10,
// DBF-21, DBF-22), approved by Luna on 2026-10-23.
const (
	dbfTooMany = "too many database reads are already running. wait a minute and try again."
	dbfMoving  = "%s is moving between regions. read its database once the move has finished."
	dbfAudit   = "forwarded a database query to region %s. the query was %d bytes, sha256 %s (cli)"
	dbfAuditAt = "forwarded a database query to region %s, as of %s. the query was %d bytes, sha256 %s (cli)"
)

// The pull's audit details (copy record DBF-19, DBF-20), approved by Luna
// on 2026-10-23.
const (
	dbfAuditPull   = "forwarded a database pull to region %s (cli)"
	dbfAuditPullAt = "forwarded a database pull to region %s, as of %s (cli)"
)

const dbReadRetryAfter = "30"

// dbForwardAuditDetail is the db-read row's detail (spec §9.3): for a query
// the SQL's length and the first 16 hex digits of its SHA-256, never its
// text; a pull carries no SQL.
func dbForwardAuditDetail(op, region, sqlText string, at time.Time) string {
	if op == dbreads.OpPull {
		if at.IsZero() {
			return fmt.Sprintf(dbfAuditPull, region)
		}
		return fmt.Sprintf(dbfAuditPullAt, region, at.UTC().Format(time.RFC3339))
	}
	sum := sha256.Sum256([]byte(sqlText))
	sha := hex.EncodeToString(sum[:])[:16]
	if at.IsZero() {
		return fmt.Sprintf(dbfAudit, region, len(sqlText), sha)
	}
	return fmt.Sprintf(dbfAuditAt, region, at.UTC().Format(time.RFC3339), len(sqlText), sha)
}

// dbReadAdmitted is a read this console has admitted, signed and queued.
type dbReadAdmitted struct {
	id, region string
	issued     time.Time
}

// handleAPIDBRead is POST /api/cli/apps/{account}/{app}/db/{host}/reads
// (spec §9.1): one forwarded read, which exists only inside this request
// (§5.2, J1). The admission counts are released by this handler's return and
// nowhere else (J5).
func (s *Server) handleAPIDBRead(w http.ResponseWriter, r *http.Request) {
	rc := http.NewResponseController(w)
	// Everything before the 200, the body included, gets s.dbf.preHeader:
	// a body delivered slowly must not hold the request past backhaul's
	// 30 s header wait (plan review finding 13).
	if err := rc.SetReadDeadline(time.Now().Add(s.dbf.preHeader)); err != nil && !errors.Is(err, http.ErrNotSupported) {
		log.Printf("console: db read: set the admission read deadline: %v", err)
	}
	a, actor, ok := s.admitDBRead(w, r)
	if !ok {
		return
	}
	defer s.dbReads.release(actor)
	// Then the server's ReadTimeout would end this connection under a read
	// that lasts longer; the read's own bound holds it instead, as the
	// import door's read deadline does (apiinstanceimport.go).
	if err := rc.SetReadDeadline(time.Now().Add(s.dbf.bound + time.Minute)); err != nil && !errors.Is(err, http.ErrNotSupported) {
		log.Printf("console: db read: extend the read deadline: %v", err)
	}
	s.streamDBRead(w, r, rc, a)
}

// readDBReadBody reads the whole body, to EOF, under the admission read
// deadline handleAPIDBRead set, and requires it to be exactly one JSON value
// (Astra round 1, P1). readJSONLimit decodes one value and leaves any tail
// unread; net/http would drain that tail when the response is first
// flushed, under the read deadline the stream extends to the read's bound,
// so a withheld tail could hold a request and its admission slot. The
// refusals are readJSONLimit's own sentences.
func readDBReadBody(w http.ResponseWriter, r *http.Request, out *dbreads.ReadBody) bool {
	ct, _, _ := strings.Cut(r.Header.Get("Content-Type"), ";")
	if strings.TrimSpace(strings.ToLower(ct)) != "application/json" {
		jsonError(w, http.StatusUnsupportedMediaType, "send a JSON body with Content-Type: application/json")
		return false
	}
	raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, dbreads.MaxSQLBytes+8<<10))
	if err != nil {
		var tooLarge *http.MaxBytesError
		if errors.As(err, &tooLarge) {
			jsonError(w, http.StatusRequestEntityTooLarge, "request body too large")
			return false
		}
		jsonError(w, http.StatusBadRequest, "malformed JSON body")
		return false
	}
	// json.Unmarshal refuses anything but whitespace after the one value.
	if json.Unmarshal(raw, out) != nil {
		jsonError(w, http.StatusBadRequest, "malformed JSON body")
		return false
	}
	return true
}

// cancelDBRead is the one write J1 allows after a read's request ends
// without a result: best effort, so the owner stops early and a box that
// has not picked it skips it.
func (s *Server) cancelDBRead(id string) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if _, err := s.cfg.Releases.Put(ctx, dbreads.CancelKey(id), nil, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
		log.Printf("console: db read %s: cancel: %v", id, err)
	}
}

// announceDBRead bumps the serial on the request's own context, so a
// request that has already ended announces nothing (J1; plan review finding
// 11). bumpSerial detaches on purpose, for callers whose writes outlive
// them; this read does not outlive its request.
func (s *Server) announceDBRead(ctx context.Context) {
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	if err := release.BumpSerial(ctx, s.cfg.Releases, time.Now()); err != nil && ctx.Err() == nil {
		log.Printf("console: db read: serial bump: %v (boxes list every 30 s anyway)", err)
	}
}

// admitDBRead is spec §5.5 step 3, all of it bounded by s.dbf.preHeader so
// the 200 leaves well inside backhaul's 30 s header wait: dbGate's order
// (person, membership with the byte-identical 404, the db budget,
// configuration, the record, the class rail, pooled, fleet), then a moving
// record, the SQL guard, capability on complete evidence, the admission
// counts, and the audit row, the signed request, its queue entry and the
// serial. Any refusal is an ordinary status and JSON error.
func (s *Server) admitDBRead(w http.ResponseWriter, r *http.Request) (dbReadAdmitted, string, bool) {
	none := dbReadAdmitted{}
	ctx, cancel := context.WithTimeout(r.Context(), s.dbf.preHeader)
	defer cancel()
	if p, ok := principalFrom(r.Context()); !ok || p.Kind != PrincipalUser {
		http.NotFound(w, r)
		return none, "", false
	}
	actor, acct, app, ok := s.apiAccountApp(w, r)
	if !ok {
		return none, "", false
	}
	if !s.cli.db.allow(hashToken(bearer(r)), time.Now()) {
		jsonError(w, http.StatusTooManyRequests, "too many database requests. Wait a minute and try again.")
		return none, "", false
	}
	if s.cfg.Releases == nil || s.cfg.HibernateBucket == "" {
		jsonError(w, http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot read databases.")
		return none, "", false
	}
	var body dbreads.ReadBody
	if !readDBReadBody(w, r, &body) {
		return none, "", false
	}
	host := r.PathValue("host")
	src, err := s.resolveDBSource(ctx, acct, actor, app, host)
	switch {
	case errors.Is(err, os.ErrNotExist), errors.Is(err, ErrBadHost):
		http.NotFound(w, r)
		return none, "", false
	case errors.Is(err, errSourceClass):
		jsonError(w, http.StatusConflict, dbClassRefusal(src.rec))
		return none, "", false
	case errors.Is(err, ErrPooled):
		jsonError(w, http.StatusConflict, fmt.Sprintf(
			"%s runs in the pool %s, so its data is on the pool's own server. carlos db cannot read it.", host, src.rec.Pool))
		return none, "", false
	case errors.Is(err, ErrResidency):
		// The case this route exists for.
	case err == nil:
		// In region: the db door serves it directly. There is no forwarded
		// read for it until in-region reads move to the box path (plan 4).
		http.NotFound(w, r)
		return none, "", false
	case errors.Is(err, errSourceResolve):
		log.Printf("console: db read %s/%s/%s: resolve: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
		return none, "", false
	default:
		log.Printf("console: db read %s/%s/%s: read record: %v", acct.Sqid, app, host, err)
		jsonError(w, http.StatusBadGateway, "could not read the instance. Try again.")
		return none, "", false
	}
	rec := src.rec
	signer, ok := s.dbForwardSigner()
	if !ok || !s.forwardable(rec) {
		jsonError(w, http.StatusConflict, s.dbResidencyRefusal(rec))
		return none, "", false
	}
	if rec.Move != nil {
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfMoving, host))
		return none, "", false
	}
	switch body.Op {
	case dbreads.OpQuery:
	case dbreads.OpPull:
		// A pull runs nothing: SQL in one is a request no CLI builds.
		if body.SQL != "" {
			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
			return none, "", false
		}
	default:
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	// encoding/json turns bytes that are not UTF-8, and a lone surrogate
	// escape, into U+FFFD without a word, so the SQL decoded here may not be
	// the SQL sent. The box must run exactly the bytes typed or be told no
	// (plan 1's Review Focus 1), so any U+FFFD is refused; a query that
	// meant one literally is refused too, and says so the same way (plan
	// review finding 36). The CLI refuses first, with DBF-32.
	if strings.ContainsRune(body.SQL, utf8.RuneError) {
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	if body.Op == dbreads.OpQuery {
		if err := dbquery.CheckSQL(body.SQL); err != nil || body.SQL == "" {
			msg := http.StatusText(http.StatusBadRequest)
			if err != nil {
				msg = err.Error()
			}
			jsonError(w, http.StatusBadRequest, msg)
			return none, "", false
		}
	}
	var at time.Time
	if body.At != "" {
		if at, err = time.Parse(time.RFC3339, body.At); err != nil || at.After(time.Now().Add(dbreads.Skew)) {
			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
			return none, "", false
		}
	}
	if _, err := dbreads.ParseReplyKey(body.ReplyKey); err != nil {
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	snap, serr := s.liveEdges(ctx)
	switch dbForwardEvidence(snap, serr, rec.Region) {
	case dbForwardNone:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoCapable, host, rec.Region))
		return none, "", false
	case dbForwardDead:
		jsonError(w, http.StatusConflict, fmt.Sprintf(dbfNoneLive, host, rec.Region))
		return none, "", false
	}
	if !s.dbReads.acquire(actor) {
		w.Header().Set("Retry-After", dbReadRetryAfter)
		jsonError(w, http.StatusTooManyRequests, dbfTooMany)
		return none, "", false
	}
	admitted := false
	defer func() {
		if !admitted {
			s.dbReads.release(actor)
		}
	}()
	fail := func(what string, err error) (dbReadAdmitted, string, bool) {
		log.Printf("console: db read %s/%s/%s: %s: %v", acct.Sqid, app, host, what, err)
		jsonError(w, http.StatusBadGateway, "could not read the instance. Try again.")
		return none, "", false
	}
	id, err := dbreads.MintID()
	if err != nil {
		return fail("mint id", err)
	}
	inc, err := hostclaim.Incarnation(acct.Sqid, app, host, rec.CreatedAt)
	if err != nil {
		return fail("incarnation", err)
	}
	issued := time.Now().UTC().Truncate(time.Second)
	atRaw := ""
	if !at.IsZero() {
		atRaw = at.UTC().Format(time.RFC3339Nano)
	}
	wire, err := dbreads.Sign(signer, dbreads.Request{ID: id, Account: acct.Sqid, App: app, Host: host, Incarnation: inc,
		Region: rec.Region, Op: body.Op, SQL: body.SQL, At: atRaw, ReplyKey: body.ReplyKey, By: actor,
		IssuedAt: issued.Format(time.RFC3339)})
	if err != nil {
		// SQL the envelope cannot carry intact (invalid UTF-8).
		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return none, "", false
	}
	// J1 at every write (plan review finding 28): the request's context is
	// checked immediately before each one, because the stores do not stop a
	// write for it (DirStore.Put ignores its context). The audit row comes
	// first, while the request is open and before any box can see the read,
	// so a read a box runs always has its row (spec §9.3 rev 6). A request
	// that ends before request.json is written leaves at most that row; one
	// that ends after leaves request.json, perhaps the queue entry, and the
	// cancel, never a serial bump. A row the store cannot write refuses the
	// read rather than run it unaudited.
	ended := func(step string, published bool) bool {
		if s.dbf.beforeWrite != nil {
			s.dbf.beforeWrite(step)
		}
		if ctx.Err() == nil {
			return false
		}
		if published {
			s.cancelDBRead(id)
		}
		return true
	}
	if ended("audit", false) {
		return none, "", false
	}
	if err := s.store.RecordActivityContext(ctx, acct.ID, app, actor, Activity{Action: dbAuditAction,
		TargetType: "instance", Target: host, Detail: dbForwardAuditDetail(body.Op, rec.Region, body.SQL, at)}, time.Now()); err != nil {
		return fail("audit", err)
	}
	if ended("request", false) {
		return none, "", false
	}
	if _, err := s.cfg.Releases.Put(ctx, dbreads.RequestKey(id), wire, ""); err != nil {
		// An ambiguous create may have landed: cancel, as for the queue.
		s.cancelDBRead(id)
		return fail("write the request", err)
	}
	if ended("queue", true) {
		return none, "", false
	}
	if _, err := s.cfg.Releases.Put(ctx, dbreads.QueueKey(rec.Region, issued, id), nil, ""); err != nil {
		// The queue entry may have landed on an ambiguous answer: cancel, so
		// a box that lists it skips it (J1's one write).
		s.cancelDBRead(id)
		return fail("write the queue entry", err)
	}
	if ended("serial", true) {
		return none, "", false
	}
	// The bump runs on the request's context, not bumpSerial's detached one,
	// so a request that ends while it runs does not announce (finding 33).
	s.announceDBRead(ctx)
	// The request may have ended during the bump (Astra round 1, P2): the
	// read is published, so it gets its cancel, and the caller no 200.
	if ctx.Err() != nil {
		s.cancelDBRead(id)
		return none, "", false
	}
	admitted = true
	return dbReadAdmitted{id: id, region: rec.Region, issued: issued}, actor, true
}

// errDBClientGone is a line the client did not take within s.dbf.stall.
var errDBClientGone = errors.New("the client went or stalled")

// dbfEndSlack is how far past the request's bound the last line may still be
// written: room for the ended line, and no more.
const dbfEndSlack = 10 * time.Second

// dbLineWriter writes one line at a time, each under a fresh write deadline
// and flushed under the same deadline (spec §5.5 step 6, codex 19). The
// deadline never passes until, the request's bound plus dbfEndSlack (plan
// review finding 12). A write or flush error ends the request as a stalled
// client.
type dbLineWriter struct {
	w     http.ResponseWriter
	rc    *http.ResponseController
	stall time.Duration
	until time.Time
	last  time.Time
}

func (l *dbLineWriter) write(line dbreads.Line) error {
	b, err := json.Marshal(line)
	if err != nil {
		return err
	}
	dl := time.Now().Add(l.stall)
	if dl.After(l.until) {
		dl = l.until
	}
	if err := l.rc.SetWriteDeadline(dl); err != nil && !errors.Is(err, http.ErrNotSupported) {
		return err
	}
	if _, err := l.w.Write(append(b, '\n')); err != nil {
		return err
	}
	if err := l.rc.Flush(); err != nil {
		return err
	}
	l.last = time.Now()
	return nil
}

// tick waits one poll and writes a wait line when nothing has been written
// for every. It answers ctx's error when ctx ends first, and errDBClientGone
// when the line cannot be written.
func (l *dbLineWriter) tick(ctx context.Context, phase string, poll, every time.Duration) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(poll):
	}
	if time.Since(l.last) >= every && l.write(dbreads.Line{Wait: phase}) != nil {
		return errDBClientGone
	}
	return nil
}

// dbReadRelay is one admitted read's relay, with what its last log line
// counts. (dbStream is apidb.go's streaming function: plan review finding 25.)
type dbReadRelay struct {
	s      *Server
	lw     *dbLineWriter
	chunks int
	bytes  int
}

// dbfCall runs one store call under s.dbf.call and the phase's context, and
// keeps writing wait lines while it blocks, so a stalled store neither holds
// the request past its phase nor silences the stream (plan review finding
// 12). A store that ignores its context is left behind: its goroutine ends
// when the call does, into a buffered channel nobody reads.
func dbfCall[T any](st *dbReadRelay, ctx context.Context, phase string, op func(context.Context) (T, error)) (T, error) {
	var zero T
	cctx, cancel := context.WithTimeout(ctx, st.s.dbf.call)
	defer cancel()
	type outcome struct {
		v   T
		err error
	}
	done := make(chan outcome, 1)
	go func() {
		v, err := op(cctx)
		done <- outcome{v, err}
	}()
	t := time.NewTicker(st.s.dbf.poll)
	defer t.Stop()
	for {
		select {
		case o := <-done:
			return o.v, o.err
		case <-cctx.Done():
			return zero, cctx.Err()
		case <-t.C:
			if time.Since(st.lw.last) >= st.s.dbf.wait && st.lw.write(dbreads.Line{Wait: phase}) != nil {
				return zero, errDBClientGone
			}
		}
	}
}

// readDBResult is result.json, or nil while there is none, and nil for one
// that is not a result in the vocabulary (Review Focus 4).
func (s *Server) readDBResult(ctx context.Context, id string) *dbreads.Result {
	raw, _, err := s.cfg.Releases.Get(ctx, dbreads.ResultKey(id))
	if err != nil {
		return nil
	}
	var res dbreads.Result
	if json.Unmarshal(raw, &res) != nil || res.Valid() != nil {
		return nil
	}
	return &res
}

// streamDBRead is spec §5.5 steps 6 and 7 inside the admitted request:
// admitted at once, picked once, then about once a second result.json first
// and the chunk listing second (codex 20; the run follow door's order), each
// chunk in sequence with no gaps, a wait line whenever s.dbf.wait passes
// with nothing else, and the result last, only once every chunk through its
// last_seq has been written. Each phase runs on its own deadline context:
// pickup on issued_at plus s.dbf.pickup, the owner's on its deadline plus
// s.dbf.grace, never later than s.dbf.bound after issued_at. It ends without
// a result, and writes cancel, when either deadline passes, a result
// included, or when the client goes or stalls. Its last log line is the
// relay's own measure (plan review finding 21).
func (s *Server) streamDBRead(w http.ResponseWriter, r *http.Request, rc *http.ResponseController, a dbReadAdmitted) {
	ctx := r.Context()
	bound := a.issued.Add(s.dbf.bound)
	w.Header().Set("Content-Type", "application/x-ndjson")
	w.Header().Set("Cache-Control", "no-store")
	w.WriteHeader(http.StatusOK)
	st := &dbReadRelay{s: s, lw: &dbLineWriter{w: w, rc: rc, stall: s.dbf.stall, until: bound.Add(dbfEndSlack)}}
	start, why := time.Now(), "client gone"
	defer func() {
		log.Printf("console: db read %s: relayed %d chunks (%d bytes) in %s, ended %s",
			a.id, st.chunks, st.bytes, time.Since(start).Round(time.Millisecond), why)
	}()
	end := func(ended string) {
		if ended != "" {
			why = ended
			st.lw.write(dbreads.Line{Ended: ended})
		}
		s.cancelDBRead(a.id)
	}
	// stop ends a phase: a client that went or stalled gets no line, a phase
	// whose deadline passed gets its ended word.
	stop := func(err error, ended string) {
		if errors.Is(err, errDBClientGone) || ctx.Err() != nil {
			end("")
			return
		}
		end(ended)
	}
	if st.lw.write(dbreads.Line{Admitted: a.id}) != nil {
		end("")
		return
	}
	store := s.cfg.Releases

	pickCtx, cancelPick := context.WithDeadline(ctx, a.issued.Add(s.dbf.pickup))
	defer cancelPick()
	var picked dbreads.Picked
	for {
		raw, err := dbfCall(st, pickCtx, dbreads.WaitPickup, func(c context.Context) ([]byte, error) {
			b, _, err := store.Get(c, dbreads.PickedKey(a.id))
			return b, err
		})
		if err == nil && json.Unmarshal(raw, &picked) == nil {
			break
		}
		if pickCtx.Err() != nil || errors.Is(err, errDBClientGone) {
			stop(err, dbreads.EndedNotPicked)
			return
		}
		if err := st.lw.tick(pickCtx, dbreads.WaitPickup, s.dbf.poll, s.dbf.wait); err != nil {
			stop(err, dbreads.EndedNotPicked)
			return
		}
	}
	if st.lw.write(dbreads.Line{Picked: picked.Box}) != nil {
		end("")
		return
	}
	deadline := picked.Deadline.Add(s.dbf.grace)
	if deadline.After(bound) {
		deadline = bound
	}
	ownCtx, cancelOwn := context.WithDeadline(ctx, deadline)
	defer cancelOwn()
	next := int64(1)
	for {
		res, err := dbfCall(st, ownCtx, dbreads.WaitReading, func(c context.Context) (*dbreads.Result, error) {
			return s.readDBResult(c, a.id), nil
		}) // first
		if ownCtx.Err() != nil || errors.Is(err, errDBClientGone) {
			stop(err, dbreads.EndedNoResult)
			return
		}
		keys, err := dbfCall(st, ownCtx, dbreads.WaitReading, func(c context.Context) ([]string, error) {
			return store.List(c, dbreads.OutPrefix(a.id))
		})
		if ownCtx.Err() != nil || errors.Is(err, errDBClientGone) {
			stop(err, dbreads.EndedNoResult)
			return
		}
		if err == nil {
			listed := map[int64]bool{}
			for _, k := range keys {
				if seq, ok := dbreads.ParseChunkKey(k, dbreads.OutPrefix(a.id)); ok {
					listed[seq] = true
				}
			}
			for listed[next] {
				seq := next
				sealed, err := dbfCall(st, ownCtx, dbreads.WaitReading, func(c context.Context) ([]byte, error) {
					b, _, err := store.Get(c, dbreads.ChunkKey(a.id, seq))
					return b, err
				})
				if ownCtx.Err() != nil || errors.Is(err, errDBClientGone) {
					stop(err, dbreads.EndedNoResult)
					return
				}
				if err != nil {
					break
				}
				if st.lw.write(dbreads.Line{Chunk: seq, Data: sealed}) != nil {
					end("")
					return
				}
				st.chunks++
				st.bytes += len(sealed)
				next++
			}
		}
		if res != nil && next > res.LastSeq {
			// The bound holds for a result too: one that is complete only
			// after the deadline is not relayed (plan review finding 12).
			if ownCtx.Err() != nil {
				stop(nil, dbreads.EndedNoResult)
				return
			}
			why = res.State
			st.lw.write(dbreads.Line{Result: res})
			return
		}
		if err := st.lw.tick(ownCtx, dbreads.WaitReading, s.dbf.poll, s.dbf.wait); err != nil {
			stop(err, dbreads.EndedNoResult)
			return
		}
	}
}