package console

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	lsfile "github.com/benbjohnson/litestream/file"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbquery"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/dbreads/dbreadstest"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// The console forwards only with the switch on, a root key whose public
// half is the deployment's, and no host claims (spec §5.5, §9.1, §11.1).
func TestDBForwardSignerNeedsTheSwitchAndTheRootKey(t *testing.T) {
	t.Parallel()
	pub, priv, _ := ed25519.GenerateKey(rand.Reader)
	_, other, _ := ed25519.GenerateKey(rand.Reader)
	for name, c := range map[string]struct {
		cfg  Config
		want bool
	}{
		"on, root key":       {Config{DBForward: true, Key: priv, ReleasePub: pub}, true},
		"off":                {Config{Key: priv, ReleasePub: pub}, false},
		"no key":             {Config{DBForward: true, ReleasePub: pub}, false},
		"not the root's key": {Config{DBForward: true, Key: other, ReleasePub: pub}, false},
		"host claims":        {Config{DBForward: true, Key: priv, ReleasePub: pub, HostClaim: true}, false},
	} {
		s := &Server{cfg: c.cfg}
		if _, ok := s.dbForwardSigner(); ok != c.want {
			t.Errorf("%s: %v", name, ok)
		}
	}
}

func TestNewHTTPServerKeepsTheConsolesTimeouts(t *testing.T) {
	t.Parallel()
	srv := NewHTTPServer(nil)
	if srv.ReadHeaderTimeout != 10*time.Second || srv.ReadTimeout != 5*time.Minute || srv.WriteTimeout != 0 {
		t.Fatalf("%+v", srv)
	}
}

// putFixtureObject creates key, or replaces it under its current etag: a
// fixture may republish a box's files, as a box does every pass. Protocol
// writes stay create-only; this helper is for fixtures only (plan review
// finding 6: DirStore.Put refuses a create over an existing key).
func putFixtureObject(t *testing.T, s release.Store, key string, body []byte) {
	t.Helper()
	etag := ""
	if _, cur, err := s.Get(context.Background(), key); err == nil {
		etag = cur
	}
	if _, err := s.Put(context.Background(), key, body, etag); err != nil {
		t.Fatal(err)
	}
}

// putEdgesFile writes control/edges/<box>.json as a box publishes it.
func putEdgesFile(t *testing.T, s release.Store, box, placement string, caps []string, draining bool) {
	t.Helper()
	body, _ := json.Marshal(map[string]any{"v": 1, "box": box, "eip": "203.0.113.10", "placement": placement,
		"capabilities": caps, "draining": draining, "hosts": map[string]any{}})
	putFixtureObject(t, s, "control/edges/"+box+".json", body)
}

// putBoxStatus writes status/boxes/<box>.json generated at at.
func putBoxStatus(t *testing.T, s release.Store, box string, at time.Time) {
	t.Helper()
	body, _ := json.Marshal(map[string]any{"generated_at": at.UTC(), "label": box})
	putFixtureObject(t, s, "status/boxes/"+box+".json", body)
}

// T14 (codex 10): only complete evidence, both flags, refuses up front.
// Incomplete evidence of any kind admits the read and lets the pickup
// window decide. Mutation: check Complete and not BoxesComplete (red: the
// missing, unmatched and unreadable status cases refuse).
func TestDBForwardEvidenceRefusesOnlyOnCompleteEvidence(t *testing.T) {
	t.Parallel()
	now := time.Now()
	capable := []string{dbreads.Capability}
	for name, c := range map[string]struct {
		setup func(t *testing.T, s release.Store)
		want  string
	}{
		"a capable live box": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", capable, false)
			putBoxStatus(t, s, "uk-1", now)
		}, ""},
		"no capable box": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
			putBoxStatus(t, s, "uk-1", now)
		}, dbForwardNone},
		"a capable box that is draining": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", capable, true)
			putBoxStatus(t, s, "uk-1", now)
		}, dbForwardNone},
		"capable but stale": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", capable, false)
			putBoxStatus(t, s, "uk-1", now.Add(-time.Hour))
		}, dbForwardDead},
		"capable in another region only": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "ie-1", "ie", capable, false)
			putBoxStatus(t, s, "ie-1", now)
		}, dbForwardNone},
		"a status snapshot missing": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
		}, ""},
		"an edges file unreadable": {func(t *testing.T, s release.Store) {
			s.Put(context.Background(), "control/edges/uk-1.json", []byte("{not json"), "")
			putBoxStatus(t, s, "uk-1", now)
		}, ""},
		"a fresh snapshot with no edges file": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
			putBoxStatus(t, s, "uk-1", now)
			putBoxStatus(t, s, "uk-2", now)
		}, ""},
		"status snapshots unreadable": {func(t *testing.T, s release.Store) {
			putEdgesFile(t, s, "uk-1", "uk", nil, false)
			s.Put(context.Background(), "status/boxes/uk-1.json", []byte("{not json"), "")
		}, ""},
	} {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			s := release.NewDirStore(t.TempDir())
			c.setup(t, s)
			snap, err := readLiveEdgesSnapshot(context.Background(), s, now)
			if got := dbForwardEvidence(snap, err, "uk"); got != c.want {
				t.Fatalf("evidence %q, want %q (complete %v, boxes %v)", got, c.want, snap.Complete, snap.BoxesComplete)
			}
		})
	}
}

const dbfHost = "shop.example.com"

// dbfFixture is a console in ie with forwarding on or off, a member, and
// shop.example.com's record in uk with a capable, live uk box published.
type dbfFixture struct {
	*consoleFixture
	member string
}

func newDBFFixture(t *testing.T, forward bool) *dbfFixture {
	t.Helper()
	fx := newConsoleFixture(t)
	fx.srv.cfg.DBForward = forward
	fx.srv.cfg.Region = "ie"
	fx.srv.cfg.HibernateBucket = "data-ie"
	if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := fx.st.AddMember(OpsAccountID, "member@example.com", RoleMember, time.Now()); err != nil {
		t.Fatal(err)
	}
	member := cliToken(t, fx, "member@example.com", "code-dbf")
	putShopRecord(t, fx, provision.Record{Host: dbfHost, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)})
	putEdgesFile(t, fx.rel, "uk-1", "uk", []string{dbreads.Capability}, false)
	putBoxStatus(t, fx.rel, "uk-1", time.Now())
	return &dbfFixture{consoleFixture: fx, member: member}
}

func (fx *dbfFixture) listURL() string {
	return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/" + dbfHost + "/ltx/9"
}

// T8, the console's half: forwarding off is main's answer byte for byte.
// The want is main's dbResidencyRefusal for this fixture, confirmed against
// main's code before this branch changed dbGate.
func TestListDoorWithForwardingOffIsMains(t *testing.T) {
	fx := newDBFFixture(t, false)
	status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	want := `{"error":"shop.example.com keeps its data in region uk. This console can only read data kept in region ie."}`
	if status != http.StatusConflict || strings.TrimSpace(body) != want {
		t.Fatalf("%d %s", status, body)
	}
}

func TestListDoorMarksACapableRegion(t *testing.T) {
	fx := newDBFFixture(t, true)
	status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	var got dbdoor.ForwardBody
	json.Unmarshal([]byte(body), &got)
	if status != http.StatusConflict || got.Forward != 1 || got.Region != "uk" ||
		!slices.Equal(got.ForwardOps, []string{dbreads.OpQuery, dbreads.OpPull}) ||
		got.Error != fmt.Sprintf(dbfUpdateCarlos, dbfHost, "uk") {
		t.Fatalf("%d %s", status, body)
	}
}

func TestListDoorRefusesOnCompleteEvidence(t *testing.T) {
	fx := newDBFFixture(t, true)
	putEdgesFile(t, fx.rel, "uk-1", "uk", nil, false)
	status, _, body := rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoCapable, dbfHost, "uk")) || strings.Contains(body, `"forward"`) {
		t.Fatalf("no capable box: %d %s", status, body)
	}
	putEdgesFile(t, fx.rel, "uk-2", "uk", []string{dbreads.Capability}, false)
	putBoxStatus(t, fx.rel, "uk-2", time.Now().Add(-time.Hour))
	status, _, body = rawReq(t, http.MethodGet, fx.listURL(), fx.member, "")
	if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoneLive, dbfHost, "uk")) {
		t.Fatalf("none live: %d %s", status, body)
	}
}

// Review Focus 5: a home-region record seen from a regional console keeps
// main's sentence; there is no region to forward to by name.
func TestListDoorKeepsMainsSentenceForTheHomeRegion(t *testing.T) {
	fx := newDBFFixture(t, true)
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "home.example.com", CreatedAt: time.Now().UTC()})
	u := fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/home.example.com/ltx/9"
	status, _, body := rawReq(t, http.MethodGet, u, fx.member, "")
	if status != http.StatusConflict || strings.Contains(body, `"forward"`) || !strings.Contains(body, "This console can only read data kept in") {
		t.Fatalf("%d %s", status, body)
	}
}

func (fx *dbfFixture) readURL(host string) string {
	return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/shop/db/" + host + "/reads"
}

func queryBody(t *testing.T, sqlText string) (dbreads.ReadBody, *ecdh.PrivateKey) {
	t.Helper()
	key, err := dbreads.NewReplyKey()
	if err != nil {
		t.Fatal(err)
	}
	return dbreads.ReadBody{Op: dbreads.OpQuery, SQL: sqlText, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())}, key
}

// openRead posts a read and returns the open response; the caller reads its
// lines or closes it.
func (fx *dbfFixture) openRead(t *testing.T, host, bearer string, body dbreads.ReadBody) *http.Response {
	t.Helper()
	raw, _ := json.Marshal(body)
	req, _ := http.NewRequest(http.MethodPost, fx.readURL(host), bytes.NewReader(raw))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+bearer)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	return res
}

func readLines(t *testing.T, r io.Reader) []dbreads.Line {
	t.Helper()
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes)
	var out []dbreads.Line
	for sc.Scan() {
		var l dbreads.Line
		if err := json.Unmarshal(sc.Bytes(), &l); err != nil {
			t.Fatalf("not a line: %s", sc.Bytes())
		}
		out = append(out, l)
	}
	return out
}

func refusal(t *testing.T, res *http.Response) (int, string) {
	t.Helper()
	defer res.Body.Close()
	b, _ := io.ReadAll(res.Body)
	return res.StatusCode, string(b)
}

func TestDBReadRouteRefusesBeforeAdmitting(t *testing.T) {
	fx := newDBFFixture(t, true)
	good, _ := queryBody(t, "SELECT 1")
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "here.example.com", Region: "ie", CreatedAt: time.Now().UTC()})
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "moving.example.com", Region: "uk", CreatedAt: time.Now().UTC(),
		Move: &provision.MoveState{ID: "mv_x", To: "ie", State: "requested"}})
	two := good
	two.SQL = "SELECT 1; SELECT 2"
	shell := good
	shell.Op = "shell"
	pullSQL := good
	pullSQL.Op = dbreads.OpPull
	badKey := good
	badKey.ReplyKey = "AAAA"
	future := good
	future.At = time.Now().Add(time.Hour).UTC().Format(time.RFC3339)
	for name, c := range map[string]struct {
		host   string
		body   dbreads.ReadBody
		status int
		want   string
	}{
		"an in-region instance":               {"here.example.com", good, http.StatusNotFound, ""},
		"no such instance":                    {"nope.example.com", good, http.StatusNotFound, ""},
		"moving":                              {"moving.example.com", good, http.StatusConflict, fmt.Sprintf(dbfMoving, "moving.example.com")},
		"two statements":                      {dbfHost, two, http.StatusBadRequest, dbquery.ErrOneStatement.Error()},
		"an op this console does not forward": {dbfHost, shell, http.StatusBadRequest, ""},
		"a pull carrying SQL":                 {dbfHost, pullSQL, http.StatusBadRequest, ""},
		"a reply key that is not a key":       {dbfHost, badKey, http.StatusBadRequest, ""},
		"a time in the future":                {dbfHost, future, http.StatusBadRequest, ""},
	} {
		status, body := refusal(t, fx.openRead(t, c.host, fx.member, c.body))
		if status != c.status || !strings.Contains(body, c.want) {
			t.Errorf("%s: %d %s", name, status, body)
		}
	}
	// SQL that JSON decoding would have rewritten (plan review finding 36):
	// a byte that is not UTF-8, and a lone surrogate escape.
	for name, sqlJSON := range map[string]string{
		"a byte that is not UTF-8": "\"SELECT '\xff'\"",
		"a lone surrogate":         `"SELECT '\ud800'"`,
	} {
		raw := `{"op":"query","sql":` + sqlJSON + `,"reply_key":"` + good.ReplyKey + `"}`
		req, _ := http.NewRequest(http.MethodPost, fx.readURL(dbfHost), strings.NewReader(raw))
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Authorization", "Bearer "+fx.member)
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		if status, body := refusal(t, res); status != http.StatusBadRequest {
			t.Errorf("%s: %d %s", name, status, body)
		}
	}
	if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a refused read wrote %v", keys)
	}
	fx.srv.cfg.DBForward = false
	status, body := refusal(t, fx.openRead(t, dbfHost, fx.member, good))
	if status != http.StatusConflict || !strings.Contains(body, "This console can only read data kept in region ie.") {
		t.Fatalf("forwarding off: %d %s", status, body)
	}
	fx.srv.cfg.DBForward, fx.srv.cfg.HostClaim = true, true
	status, body = refusal(t, fx.openRead(t, dbfHost, fx.member, good))
	if status != http.StatusConflict || !strings.Contains(body, "This console can only read data kept in") {
		t.Fatalf("a host-claim console: %d %s", status, body)
	}
}

func TestDBReadRouteRefusesOnCompleteEvidence(t *testing.T) {
	fx := newDBFFixture(t, true)
	putEdgesFile(t, fx.rel, "uk-1", "uk", nil, false)
	good, _ := queryBody(t, "SELECT 1")
	status, body := refusal(t, fx.openRead(t, dbfHost, fx.member, good))
	if status != http.StatusConflict || !strings.Contains(body, fmt.Sprintf(dbfNoCapable, dbfHost, "uk")) {
		t.Fatalf("%d %s", status, body)
	}
}

// Admission, then the request, the queue entry, the serial and the audit
// row, all before the 200.
func TestDBReadQueuesASignedRequestAndAudits(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.pickup = 200 * time.Millisecond
	body, _ := queryBody(t, "SELECT count(*) FROM orders")
	res := fx.openRead(t, dbfHost, fx.member, body)
	lines := readLines(t, res.Body)
	res.Body.Close()
	if res.StatusCode != http.StatusOK || len(lines) == 0 || lines[0].Admitted == "" {
		t.Fatalf("%d %+v", res.StatusCode, lines)
	}
	id := lines[0].Admitted
	raw, _, err := fx.rel.Get(context.Background(), dbreads.RequestKey(id))
	if err != nil {
		t.Fatal(err)
	}
	req, err := dbreads.Verify(fx.pub, raw, id, "uk", time.Now())
	if err != nil || req.SQL != body.SQL || req.Host != dbfHost || req.By != "member@example.com" || req.Account != release.OpsAccount {
		t.Fatalf("request %+v: %v", req, err)
	}
	issued, _ := req.Issued()
	if _, _, err := fx.rel.Get(context.Background(), dbreads.QueueKey("uk", issued, id)); err != nil {
		t.Fatalf("no queue entry: %v", err)
	}
	if _, _, err := fx.rel.Get(context.Background(), release.SerialKey); err != nil {
		t.Fatalf("the serial was not bumped: %v", err)
	}
	evs, _ := fx.st.Events(OpsAccountID, "shop", 10)
	want := dbForwardAuditDetail(dbreads.OpQuery, "uk", body.SQL, time.Time{})
	found := false
	for _, e := range evs {
		found = found || (e.Action == "db-read" && e.Target == dbfHost && e.Detail == want)
	}
	if !found || strings.Contains(want, "orders") {
		t.Fatalf("audit %q in %+v", want, evs)
	}
}

// dbfPutHook runs hook after each create of a key with suffix, once the
// create has landed.
type dbfPutHook struct {
	release.Store
	suffix string
	hook   func()
}

func (h *dbfPutHook) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	etag, err := h.Store.Put(ctx, key, body, ifMatch)
	if strings.HasSuffix(key, h.suffix) || (h.suffix == "queue" && strings.HasPrefix(key, dbreads.Prefix+"queue/")) {
		h.hook()
	}
	return etag, err
}

// dbReadAudited is safe from a handler's goroutine: it never calls t.Fatal.
func dbReadAudited(t *testing.T, st *Store) bool {
	t.Helper()
	evs, err := st.Events(OpsAccountID, "shop", 10)
	if err != nil {
		t.Error(err)
		return false
	}
	for _, e := range evs {
		if e.Action == "db-read" && e.Target == dbfHost {
			return true
		}
	}
	return false
}

// dbfServe drives the handler in process with ctx as the request's own
// context, so a test can end the request at an exact step.
func (fx *dbfFixture) dbfServe(t *testing.T, ctx context.Context, sqlText string) *httptest.ResponseRecorder {
	t.Helper()
	body, _ := queryBody(t, sqlText)
	raw, _ := json.Marshal(body)
	req := httptest.NewRequest(http.MethodPost, fx.readURL(dbfHost), bytes.NewReader(raw)).WithContext(ctx)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+fx.member)
	rec := httptest.NewRecorder()
	fx.srv.ServeHTTP(rec, req)
	return rec
}

// dbfBucket is what a read left under db-reads/: request objects, queue
// entries and cancels, by count.
func dbfBucket(t *testing.T, s release.Store) (requests, queued, cancels int) {
	t.Helper()
	keys, _ := s.List(context.Background(), dbreads.Prefix)
	for _, k := range keys {
		switch {
		case strings.HasPrefix(k, dbreads.Prefix+"queue/"):
			queued++
		case strings.HasSuffix(k, "/request.json"):
			requests++
		case strings.HasSuffix(k, "/cancel"):
			cancels++
		}
	}
	return
}

// Plan review findings 11, 28 and 33 (J1, spec §5.5 step 3 and §9.3 rev 6):
// the request's context is checked immediately before every write of
// admission. Ended before the audit row: nothing at all. Before
// request.json: the row only. Before the queue entry: the row, request.json
// and the cancel. Before the serial: the row, both objects and the cancel.
// Never a serial bump, never a 200. And the bump itself runs on the
// request's context, so a request that ends while it reads the serial does
// not write it.
// Mutations, each red in its own case: drop any one of the four ended()
// checks; drop the cancel on a published step; write the audit row after
// the creates ("the audit row is down before the request"); call
// s.bumpSerial instead of announceDBRead ("ended during the serial read":
// the detached write lands).
func TestDBReadEndedDuringAdmissionWritesOnlyTheCancel(t *testing.T) {
	t.Run("the audit row is down before the request", func(t *testing.T) {
		fx := newDBFFixture(t, true)
		auditedFirst := false
		fx.srv.cfg.Releases = &dbfPutHook{Store: fx.rel, suffix: "/request.json", hook: func() {
			auditedFirst = dbReadAudited(t, fx.st)
		}}
		fx.srv.dbf.pickup = 100 * time.Millisecond
		body, _ := queryBody(t, "SELECT 1")
		res := fx.openRead(t, dbfHost, fx.member, body)
		io.Copy(io.Discard, res.Body)
		res.Body.Close()
		if !auditedFirst {
			t.Fatal("request.json was written before the audit row")
		}
	})
	for _, c := range []struct {
		step                      string
		audited                   bool
		requests, queued, cancels int
	}{
		{"audit", false, 0, 0, 0},
		{"request", true, 0, 0, 0},
		{"queue", true, 1, 0, 1},
		{"serial", true, 1, 1, 1},
	} {
		t.Run("ended before "+c.step, func(t *testing.T) {
			fx := newDBFFixture(t, true)
			_, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey)
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			fx.srv.dbf.beforeWrite = func(step string) {
				if step == c.step {
					cancel()
				}
			}
			rec := fx.dbfServe(t, ctx, "SELECT 1")
			// The ended path writes no response (nobody is there to read
			// it), and a recorder never written reports 200: read the body.
			if strings.Contains(rec.Body.String(), "admitted") {
				t.Fatalf("an ended request was admitted: %s", rec.Body)
			}
			if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore {
				t.Fatal("the serial was bumped after the request ended")
			}
			requests, queued, cancels := dbfBucket(t, fx.rel)
			if got := dbReadAudited(t, fx.st); got != c.audited || requests != c.requests || queued != c.queued || cancels != c.cancels {
				t.Fatalf("audited %v, requests %d, queue entries %d, cancels %d", got, requests, queued, cancels)
			}
		})
	}
	// Over the real DirStore, whose Put ignores its context (Astra round 1,
	// P2): only the read of the serial is hooked, so nothing but the bump's
	// own check can stop the write. The request is then re-checked after the
	// bump: the read was published, so it gets its cancel and no 200.
	t.Run("ended during the serial read", func(t *testing.T) {
		fx := newDBFFixture(t, true)
		_, serialBefore, _ := fx.rel.Get(context.Background(), release.SerialKey)
		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		fx.srv.cfg.Releases = &dbfGetHook{Store: fx.rel, key: release.SerialKey, onGet: cancel}
		rec := fx.dbfServe(t, ctx, "SELECT 1")
		if _, serialAfter, _ := fx.rel.Get(context.Background(), release.SerialKey); serialAfter != serialBefore {
			t.Fatal("the serial was written after the request ended during the bump")
		}
		if strings.Contains(rec.Body.String(), "admitted") {
			t.Fatalf("a request that ended during the bump was admitted: %s", rec.Body)
		}
		if requests, queued, cancels := dbfBucket(t, fx.rel); requests != 1 || queued != 1 || cancels != 1 {
			t.Fatalf("requests %d, queue entries %d, cancels %d", requests, queued, cancels)
		}
	})
}

// dbfGetHook runs onGet once on a read of key and is otherwise the store
// beneath it, its context handling included: a DirStore's Put ignores its
// context, as CARLOS_DEPLOYMENT_DIR's store does in production.
type dbfGetHook struct {
	release.Store
	key   string
	onGet func()
	once  sync.Once
}

func (h *dbfGetHook) Get(ctx context.Context, key string) ([]byte, string, error) {
	if key == h.key {
		h.once.Do(h.onGet)
	}
	return h.Store.Get(ctx, key)
}

// Plan review finding 13: a body delivered slowly is cut at s.dbf.preHeader,
// well inside backhaul's 30 s header wait, and admits nothing.
// Mutation: drop the preHeader read deadline, keeping only the bound's (red:
// the connection is still open at 3 s).
func TestDBReadStalledBodyIsCutAtThePreHeaderBound(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.preHeader = 300 * time.Millisecond
	conn, err := net.Dial("tcp", strings.TrimPrefix(fx.ts.URL, "http://"))
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	path := strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL)
	fmt.Fprintf(conn, "POST %s HTTP/1.1\r\nHost: console\r\nAuthorization: Bearer %s\r\n"+
		"Content-Type: application/json\r\nContent-Length: 4096\r\n\r\n{\"op\":", path, fx.member)
	conn.SetReadDeadline(time.Now().Add(3 * time.Second))
	start := time.Now()
	_, err = io.ReadAll(conn) // a refusal, then the close; or the close alone
	if ne, ok := err.(net.Error); ok && ne.Timeout() {
		t.Fatalf("the stalled body still held the request after %s", time.Since(start))
	}
	if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a stalled body wrote %v", keys)
	}
}

// startBox runs plan 1's real agent and worker code in process for uk,
// passing every 20 ms: a stand-in for the box's serial poll, not acceptance.
func (fx *dbfFixture) startBox(t *testing.T, replicaDir string) *dbreadstest.Box {
	t.Helper()
	box := dbreadstest.Start(t, dbreadstest.Options{Store: fx.rel, Region: "uk", Box: "uk-1", Pub: fx.pub, HibernatePrefix: "instances",
		OpenReplica: func(context.Context, string) (dbdoor.ReplicaReader, error) {
			return lsfile.NewReplicaClient(replicaDir), nil
		}})
	runBox(t, box)
	return box
}

func runBox(t *testing.T, box *dbreadstest.Box) {
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		for ctx.Err() == nil {
			box.Agent.Pass(ctx)
			time.Sleep(20 * time.Millisecond)
		}
	}()
	t.Cleanup(func() { cancel(); <-done; box.Agent.Wait() })
}

// openAnswer opens a done result's chunks and summary with the CLI's key.
func openAnswer(t *testing.T, key *ecdh.PrivateKey, lines []dbreads.Line) ([]byte, dbreads.Summary) {
	t.Helper()
	id := lines[0].Admitted
	res := lines[len(lines)-1].Result
	if res == nil || res.State != dbreads.StateDone {
		t.Fatalf("no done result: %+v", lines[len(lines)-1])
	}
	var rows bytes.Buffer
	for _, l := range lines {
		if l.Chunk == 0 {
			continue
		}
		plain, err := dbreads.OpenChunk(key, id, l.Chunk, dbreads.StreamRows, l.Chunk == res.LastSeq, l.Data)
		if err != nil {
			t.Fatalf("chunk %d: %v", l.Chunk, err)
		}
		rows.Write(plain)
	}
	s, err := dbreads.OpenSummary(key, id, res.Sealed)
	if err != nil {
		t.Fatal(err)
	}
	return rows.Bytes(), s
}

// A pull through the console: the route admits it, relays the box's ltx
// stream and its summary's file list, and audits it with DBF-19; with --at,
// DBF-20.
func TestDBReadPullStreamsTheFiles(t *testing.T) {
	fx := newDBFFixture(t, true)
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	fx.startBox(t, dir)
	key, _ := dbreads.NewReplyKey()
	res := fx.openRead(t, dbfHost, fx.member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	lines := readLines(t, res.Body)
	res.Body.Close()
	last := lines[len(lines)-1].Result
	if last == nil || last.State != dbreads.StateDone {
		t.Fatalf("lines %+v", lines)
	}
	s, err := dbreads.OpenSummary(key, lines[0].Admitted, last.Sealed)
	if err != nil || len(s.Files) == 0 || s.TXID != s.Files[len(s.Files)-1].MaxTXID {
		t.Fatalf("summary %+v %v", s, err)
	}
	h := sha256.New()
	for _, l := range lines {
		if l.Chunk == 0 {
			continue
		}
		plain, err := dbreads.OpenChunk(key, lines[0].Admitted, l.Chunk, dbreads.StreamLTX, l.Chunk == last.LastSeq, l.Data)
		if err != nil {
			t.Fatalf("chunk %d: %v", l.Chunk, err)
		}
		h.Write(plain)
	}
	if hex.EncodeToString(h.Sum(nil)) != s.SHA256 {
		t.Fatal("the pulled bytes do not add up to the summary")
	}
	at := time.Now().Add(-time.Second).UTC().Truncate(time.Second)
	res = fx.openRead(t, dbfHost, fx.member, dbreads.ReadBody{Op: dbreads.OpPull, At: at.Format(time.RFC3339),
		ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	readLines(t, res.Body)
	res.Body.Close()
	evs, _ := fx.st.Events(OpsAccountID, "shop", 10)
	for _, want := range []string{fmt.Sprintf(dbfAuditPull, "uk"), fmt.Sprintf(dbfAuditPullAt, "uk", at.Format(time.RFC3339))} {
		found := false
		for _, e := range evs {
			found = found || (e.Action == "db-read" && e.Target == dbfHost && e.Detail == want)
		}
		if !found {
			t.Errorf("no %q audit row in %+v", want, evs)
		}
	}
}

func TestDBReadStreamsAQueryEndToEnd(t *testing.T) {
	fx := newDBFFixture(t, true)
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	fx.startBox(t, dir)
	body, key := queryBody(t, "SELECT count(*) AS n FROM t")
	res := fx.openRead(t, dbfHost, fx.member, body)
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK || res.Header.Get("Content-Type") != "application/x-ndjson" ||
		res.Header.Get("Cache-Control") != "no-store" || res.ContentLength != -1 {
		t.Fatalf("%d %v", res.StatusCode, res.Header)
	}
	lines := readLines(t, res.Body)
	if len(lines) < 2 || lines[0].Admitted == "" || lines[1].Picked != "uk-1" {
		t.Fatalf("lines %+v", lines)
	}
	rows, _ := openAnswer(t, key, lines)
	if string(rows) != "n\n30\n" {
		t.Fatalf("rows %q", rows)
	}
}

// T1's second tenant, locally (spec §11.4: the host or prefix taken from
// anywhere but the record): two instances in uk, each with its own replica
// at the prefix the box derives for it, served in turn by one box. Each read
// gets its own rows. Mutation: the agent opens the first host it ever
// checked (red: the second read gets the first instance's rows).
// TestDBForwardMinio makes the same check against real S3 with per-region
// credentials, in CI's MinIO step.
func TestDBReadEachTenantGetsItsOwnRows(t *testing.T) {
	fx := newDBFFixture(t, true)
	const other = "other.example.com"
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: other, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)})
	root := t.TempDir()
	dbdoortest.Replicate(t, filepath.Join(root, "instances", dbfHost), 3)
	dbdoortest.Replicate(t, filepath.Join(root, "instances", other), 1)
	box := dbreadstest.Start(t, dbreadstest.Options{Store: fx.rel, Region: "uk", Box: "uk-1", Pub: fx.pub, HibernatePrefix: "instances",
		OpenReplica: func(_ context.Context, prefix string) (dbdoor.ReplicaReader, error) {
			return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
		}})
	runBox(t, box)
	for _, c := range []struct{ host, want string }{{dbfHost, "n\n30\n"}, {other, "n\n10\n"}} {
		body, key := queryBody(t, "SELECT count(*) AS n FROM t")
		res := fx.openRead(t, c.host, fx.member, body)
		lines := readLines(t, res.Body)
		res.Body.Close()
		if rows, _ := openAnswer(t, key, lines); string(rows) != c.want {
			t.Fatalf("%s: %q, want %q", c.host, rows, c.want)
		}
	}
}

// No box: wait lines while the console waits, then ended not-picked after
// the pickup window, a cancel, and the slot free again.
func TestDBReadEndsNotPickedAndCancels(t *testing.T) {
	fx := newDBFFixture(t, true)
	// issued_at is truncated to the second (production's semantics), so the
	// pickup window seen from admission is pickup minus up to a second: 1.4 s
	// leaves at least 400 ms, eight wait periods (plan review finding 30).
	fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, 1400*time.Millisecond
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	lines := readLines(t, res.Body)
	res.Body.Close()
	waits := 0
	for _, l := range lines {
		if l.Wait == dbreads.WaitPickup {
			waits++
		}
	}
	last := lines[len(lines)-1]
	if waits < 3 || last.Ended != dbreads.EndedNotPicked || last.Result != nil {
		t.Fatalf("lines %+v", lines)
	}
	if _, _, err := fx.rel.Get(context.Background(), dbreads.CancelKey(lines[0].Admitted)); err != nil {
		t.Fatalf("no cancel: %v", err)
	}
}

// T7, the console's half: admitted leaves with the headers, and every line
// is flushed as it is written.
func TestDBReadLinesAreFlushedAsWritten(t *testing.T) {
	fx := newDBFFixture(t, true)
	// At least a second of pickup is left after issued_at's truncation: room
	// for the five lines below at 100 ms each (plan review finding 30).
	fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 100*time.Millisecond, 2*time.Second
	body, _ := queryBody(t, "SELECT 1")
	start := time.Now()
	res := fx.openRead(t, dbfHost, fx.member, body)
	defer res.Body.Close()
	br := bufio.NewReader(res.Body)
	prev := start
	for i := 0; i < 5; i++ {
		if _, err := br.ReadBytes('\n'); err != nil {
			t.Fatal(err)
		}
		if gap := time.Since(prev); gap > 400*time.Millisecond {
			t.Fatalf("line %d arrived %s after the last: not flushed", i, gap)
		}
		prev = time.Now()
	}
}

// dbfOpen is one open read: its response, a reader over its lines, and its
// id ("" when it was refused).
type dbfOpen struct {
	res *http.Response
	br  *bufio.Reader
	id  string
}

func (fx *dbfFixture) openHeld(t *testing.T, base, bearer string) dbfOpen {
	t.Helper()
	body, _ := queryBody(t, "SELECT 1")
	raw, _ := json.Marshal(body)
	req, _ := http.NewRequest(http.MethodPost, base+"/api/cli/apps/"+release.OpsAccount+"/shop/db/"+dbfHost+"/reads", bytes.NewReader(raw))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+bearer)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	o := dbfOpen{res: res, br: bufio.NewReader(res.Body)}
	if res.StatusCode == http.StatusOK {
		line, _ := o.br.ReadBytes('\n')
		var l dbreads.Line
		json.Unmarshal(line, &l)
		o.id = l.Admitted
	}
	return o
}

// untilPicked reads lines until the console has written picked, so the
// handler is in the owner's phase.
func (o dbfOpen) untilPicked(t *testing.T) {
	t.Helper()
	for {
		line, err := o.br.ReadBytes('\n')
		if err != nil {
			t.Fatalf("%s ended before picked: %v", o.id, err)
		}
		var l dbreads.Line
		json.Unmarshal(line, &l)
		if l.Picked != "" {
			return
		}
	}
}

func refusedTooMany(t *testing.T, o dbfOpen, what string) {
	t.Helper()
	status, body := refusal(t, o.res)
	if status != http.StatusTooManyRequests || o.res.Header.Get("Retry-After") == "" || !strings.Contains(body, dbfTooMany) {
		t.Fatalf("%s: %d %s", what, status, body)
	}
}

// admitWithin retries until a read is admitted: a handler frees its slot when
// it notices its client went, which takes up to a poll.
func (fx *dbfFixture) admitWithin(t *testing.T, base, bearer string) dbfOpen {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		o := fx.openHeld(t, base, bearer)
		if o.id != "" {
			return o
		}
		o.res.Body.Close()
		if time.Now().After(deadline) {
			t.Fatal("a disconnected read still holds its slot")
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// DBF-9 and T15 (J5; plan review findings 7 and 18): two open reads per
// person and sixteen per console, held by the open requests. Both of a
// person's slots come back when both CLIs leave before pickup, and again
// when both leave after it; the seventeenth read across nine people is
// refused; each disconnected read writes cancel; and a console restarted
// under open reads recovers nothing and starts with empty counts.
// Mutations: drop handleAPIDBRead's defer release (red: the second pair is
// never admitted); release at the end of admitDBRead (red: the third read of
// a pair is admitted); skip cancelDBRead on a client that went (red: no
// cancel); a startup pass that writes under db-reads/ (red: the listing
// changes across NewServer).
func TestDBReadAdmissionBounds(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.pickup = 10*time.Millisecond, 50*time.Millisecond, time.Minute
	base := fx.ts.URL
	var left []string

	// Before pickup.
	a, b := fx.openHeld(t, base, fx.member), fx.openHeld(t, base, fx.member)
	if a.id == "" || b.id == "" {
		t.Fatal("the first two reads were refused")
	}
	refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read")
	a.res.Body.Close()
	b.res.Body.Close()
	left = append(left, a.id, b.id)

	// Both slots come back; the pair is picked, then both CLIs leave.
	c, d := fx.admitWithin(t, base, fx.member), fx.admitWithin(t, base, fx.member)
	refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read after the first pair left")
	for _, o := range []dbfOpen{c, d} {
		scriptFor(t, fx.rel, o.id).picked(time.Now().Add(time.Minute))
		o.untilPicked(t)
		o.res.Body.Close()
		left = append(left, o.id)
	}
	e, f := fx.admitWithin(t, base, fx.member), fx.admitWithin(t, base, fx.member)
	refusedTooMany(t, fx.openHeld(t, base, fx.member), "a third read after the picked pair left")
	e.res.Body.Close()
	f.res.Body.Close()
	left = append(left, e.id, f.id)
	for _, id := range left {
		deadline := time.Now().Add(5 * time.Second)
		for {
			if _, _, err := fx.rel.Get(context.Background(), dbreads.CancelKey(id)); err == nil {
				break
			}
			if time.Now().After(deadline) {
				t.Fatalf("%s: no cancel after its CLI left", id)
			}
			time.Sleep(20 * time.Millisecond)
		}
	}

	// Sixteen per console, across people.
	var held []dbfOpen
	people := dbreads.OpenPerConsole/dbreads.OpenPerPerson + 1
	for i := 0; i < people; i++ {
		tok := fx.member
		if i > 0 {
			addr := fmt.Sprintf("member%d@example.com", i)
			if err := fx.st.AddMember(OpsAccountID, addr, RoleMember, time.Now()); err != nil {
				t.Fatal(err)
			}
			tok = cliToken(t, fx.consoleFixture, addr, fmt.Sprintf("code-dbf-%d", i))
		}
		for j := 0; j < dbreads.OpenPerPerson; j++ {
			if len(held) == dbreads.OpenPerConsole {
				refusedTooMany(t, fx.openHeld(t, base, tok), "the seventeenth read on the console")
				continue
			}
			o := fx.admitWithin(t, base, tok)
			held = append(held, o)
		}
	}
	if len(held) != dbreads.OpenPerConsole {
		t.Fatalf("held %d reads", len(held))
	}

	// A restart under open reads: nothing to recover, nothing written, and
	// the new console's counts start empty.
	before, _ := fx.rel.List(context.Background(), dbreads.Prefix)
	srv2, err := NewServer(fx.srv.cfg, fx.st)
	if err != nil {
		t.Fatal(err)
	}
	srv2.dbf = fx.srv.dbf
	ts2 := httptest.NewServer(srv2)
	defer ts2.Close()
	after, _ := fx.rel.List(context.Background(), dbreads.Prefix)
	if !slices.Equal(before, after) {
		t.Fatalf("starting a console wrote under db-reads/: %v", after)
	}
	g := fx.openHeld(t, ts2.URL, fx.member)
	if g.id == "" {
		status, body := refusal(t, g.res)
		t.Fatalf("the restarted console refused a read: %d %s", status, body)
	}
	g.res.Body.Close()
	for _, o := range held {
		o.res.Body.Close()
	}
}

// slowStore holds every call whose key contains match until free closes,
// ignoring its context: the worst store, for plan review finding 12.
type slowStore struct {
	release.Store
	match string
	delay time.Duration // 0: until free closes
	free  chan struct{}
}

func (s *slowStore) hold(key string) {
	if !strings.Contains(key, s.match) {
		return
	}
	if s.delay > 0 {
		time.Sleep(s.delay)
		return
	}
	<-s.free
}

func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.hold(key)
	return s.Store.Get(ctx, key)
}

func (s *slowStore) List(ctx context.Context, prefix string) ([]string, error) {
	s.hold(prefix)
	return s.Store.List(ctx, prefix)
}

// Plan review finding 12: a store that stops answering, in either phase,
// neither silences the stream nor holds it past the phase's deadline.
// Mutations: call the store on r.Context() without dbfCall (red: no wait
// line and no end until free closes); drop the pickup deadline context (red:
// same); drop the owner's deadline context (red: same in the owner case).
func TestDBReadStalledStoreKeepsTheStreamAndItsBound(t *testing.T) {
	for _, phase := range []string{"pickup", "owner"} {
		t.Run(phase, func(t *testing.T) {
			fx := newDBFFixture(t, true)
			fx.srv.dbf.poll, fx.srv.dbf.wait, fx.srv.dbf.call = 10*time.Millisecond, 100*time.Millisecond, time.Minute
			fx.srv.dbf.pickup, fx.srv.dbf.grace = 1500*time.Millisecond, 0
			ss := &slowStore{Store: fx.rel, match: "/picked", free: make(chan struct{})}
			want := dbreads.EndedNotPicked
			if phase == "owner" {
				ss.match, want = "/result.json", dbreads.EndedNoResult
			}
			defer close(ss.free)
			fx.srv.cfg.Releases = ss
			o := fx.openHeld(t, fx.ts.URL, fx.member)
			defer o.res.Body.Close()
			if phase == "owner" {
				scriptFor(t, fx.rel, o.id).picked(time.Now().Add(time.Second))
			}
			start, prev := time.Now(), time.Now()
			var last dbreads.Line
			for {
				line, err := o.br.ReadBytes('\n')
				if err != nil {
					break
				}
				if gap := time.Since(prev); gap > 500*time.Millisecond {
					t.Fatalf("a %s gap with the store stalled", gap)
				}
				prev = time.Now()
				json.Unmarshal(line, &last)
				if time.Since(start) > 5*time.Second {
					t.Fatal("the stalled store held the request past its deadline")
				}
			}
			if last.Ended != want {
				t.Fatalf("last line %+v, want %s", last, want)
			}
		})
	}
}

// Plan review finding 12: a relay slower than the time left ends at the
// request's bound, without the rest of the chunks and without the result,
// though both are in the bucket. issued_at is truncated to the second, so
// the bound is 2 to 3 s after the request; relaying all forty chunks takes 6.
// Mutation: relay without the owner's deadline, the chunk Gets on
// r.Context() and no ownCtx check after them (red: all forty are relayed
// and the request lasts 6 s).
// Not witnessed alone: the ownCtx check before the result line; dbfCall
// already cuts every call after the deadline, so that check guards only a
// line write that outlasts it, which TestDBReadResultAfterTheDeadlineIsNotRelayed
// forces with a held writer.
func TestDBReadSlowRelayEndsAtTheBound(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.bound, fx.srv.dbf.grace = 10*time.Millisecond, 3*time.Second, 0
	fx.srv.cfg.Releases = &slowStore{Store: fx.rel, match: "/out/", delay: 150 * time.Millisecond}
	start := time.Now()
	o := fx.openHeld(t, fx.ts.URL, fx.member)
	defer o.res.Body.Close()
	box := scriptFor(t, fx.rel, o.id)
	const n = 40
	for seq := int64(1); seq <= n; seq++ {
		box.chunk(seq, seq == n, "row\n")
	}
	box.result(n, strings.Repeat("row\n", n))
	box.picked(time.Now().Add(time.Minute))
	lines := readLines(t, o.br)
	chunks := 0
	for _, l := range lines {
		if l.Result != nil {
			t.Fatalf("a result relayed past the bound: %+v", l.Result)
		}
		if l.Chunk != 0 {
			chunks++
		}
	}
	if took := time.Since(start); took > 4500*time.Millisecond || chunks >= n {
		t.Fatalf("relayed %d chunks in %s", chunks, took)
	}
	if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
		t.Fatalf("last line %+v", last)
	}
}

// scriptedBox writes a read's objects itself: a stand-in for a box, for the
// orderings T18 needs. It seals real chunks to the request's reply key.
type scriptedBox struct {
	t   *testing.T
	s   release.Store
	pub *ecdh.PublicKey
	id  string
}

func scriptFor(t *testing.T, s release.Store, id string) *scriptedBox {
	t.Helper()
	raw, _, err := s.Get(context.Background(), dbreads.RequestKey(id))
	if err != nil {
		t.Fatal(err)
	}
	var w struct{ Payload string }
	json.Unmarshal(raw, &w)
	payload, _ := base64.StdEncoding.DecodeString(w.Payload)
	var req dbreads.Request
	json.Unmarshal(payload, &req)
	pub, err := dbreads.ParseReplyKey(req.ReplyKey)
	if err != nil {
		t.Fatal(err)
	}
	return &scriptedBox{t: t, s: s, pub: pub, id: id}
}

func (b *scriptedBox) picked(deadline time.Time) {
	p, _ := json.Marshal(dbreads.Picked{Box: "uk-1", Nonce: "n", Deadline: deadline})
	b.s.Put(context.Background(), dbreads.PickedKey(b.id), p, "")
}

func (b *scriptedBox) chunk(seq int64, last bool, plain string) {
	sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamRows, last, []byte(plain))
	b.s.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "")
}

func (b *scriptedBox) result(last int64, rows string) {
	sum := sha256.Sum256([]byte(rows))
	sealed, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: 9, Bytes: int64(len(rows)), SHA256: hex.EncodeToString(sum[:])})
	r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: last, Box: "uk-1", Sealed: sealed})
	b.s.Put(context.Background(), dbreads.ResultKey(b.id), r, "")
}

// dbReadHookStore runs once on the console's first listing of a read's
// chunks, after the listing is taken: a box committing between the console's
// listing and its next result read. It also records, in order, each result
// read and each chunk listing. (hookStore is temphook_test.go's: plan review
// finding 2.)
type dbReadHookStore struct {
	release.Store
	once sync.Once
	hook func()
	mu   sync.Mutex
	ops  []string
}

func (h *dbReadHookStore) record(op string) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.ops = append(h.ops, op)
}

func (h *dbReadHookStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if strings.HasSuffix(key, "/result.json") {
		h.record("result")
	}
	return h.Store.Get(ctx, key)
}

func (h *dbReadHookStore) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := h.Store.List(ctx, prefix)
	if strings.HasSuffix(prefix, "/out/") {
		h.record("list")
		h.once.Do(h.hook)
	}
	return keys, err
}

// T18 (codex 20): the box commits its last chunk and result.json between the
// console's listing and its result read, in both orders. The result line
// never comes before every chunk through last_seq, the read succeeds, and in
// every round the console reads result.json before it lists the chunks.
// Mutations, each red on its own (plan review finding 20): list out/ before
// reading result.json (the operation-order check), or write the result line
// without waiting for next > last_seq (the chunk-count check).
func TestDBReadResultWaitsForItsLastChunk(t *testing.T) {
	for _, resultFirst := range []bool{false, true} {
		fx := newDBFFixture(t, true)
		fx.srv.dbf.poll = 10 * time.Millisecond
		hs := &dbReadHookStore{Store: fx.rel}
		fx.srv.cfg.Releases = hs
		body, key := queryBody(t, "SELECT 1")
		res := fx.openRead(t, dbfHost, fx.member, body)
		br := bufio.NewReader(res.Body)
		first, _ := br.ReadBytes('\n')
		var adm dbreads.Line
		json.Unmarshal(first, &adm)
		box := scriptFor(t, fx.rel, adm.Admitted)
		hs.hook = func() {
			if resultFirst {
				box.result(2, "a\nb\n")
				box.chunk(2, true, "b\n")
			} else {
				box.chunk(2, true, "b\n")
				box.result(2, "a\nb\n")
			}
		}
		box.chunk(1, false, "a\n")
		box.picked(time.Now().Add(time.Minute))
		lines := append([]dbreads.Line{adm}, readLines(t, br)...)
		res.Body.Close()
		chunks := 0
		for _, l := range lines {
			if l.Chunk != 0 {
				chunks++
			}
			if l.Result != nil && chunks != 2 {
				t.Fatalf("result-first %v: the result line came after %d chunks", resultFirst, chunks)
			}
		}
		if rows, _ := openAnswer(t, key, lines); string(rows) != "a\nb\n" {
			t.Fatalf("rows %q", rows)
		}
		hs.mu.Lock()
		ops := append([]string(nil), hs.ops...)
		hs.mu.Unlock()
		for i, op := range ops {
			if op == "list" && (i == 0 || ops[i-1] != "result") {
				t.Fatalf("result-first %v: a chunk listing not preceded by its round's result read: %v", resultFirst, ops)
			}
		}
	}
}

// dbfHideStore leaves chunk seq out of the first hide listings of a read's
// chunks: a listing that lags the box's writes.
type dbfHideStore struct {
	release.Store
	seq  int64
	mu   sync.Mutex
	hide int
}

func (h *dbfHideStore) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := h.Store.List(ctx, prefix)
	if !strings.HasSuffix(prefix, "/out/") {
		return keys, err
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.hide == 0 {
		return keys, err
	}
	h.hide--
	var out []string
	for _, k := range keys {
		if seq, ok := dbreads.ParseChunkKey(k, prefix); !ok || seq != h.seq {
			out = append(out, k)
		}
	}
	return out, err
}

// T18's completeness guard on its own (plan review finding 31): result.json
// says last_seq 2 from the first round, and the first three listings miss
// chunk 2. The result line waits until chunk 2 is relayed. Mutation: write
// the result line without waiting for next > res.LastSeq (red: the result
// comes after one chunk).
func TestDBReadResultWaitsForAChunkTheListingMissed(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll = 10 * time.Millisecond
	fx.srv.cfg.Releases = &dbfHideStore{Store: fx.rel, seq: 2, hide: 3}
	body, key := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	defer res.Body.Close()
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	box := scriptFor(t, fx.rel, adm.Admitted)
	box.chunk(1, false, "a\n")
	box.chunk(2, true, "b\n")
	box.result(2, "a\nb\n")
	box.picked(time.Now().Add(time.Minute))
	lines := append([]dbreads.Line{adm}, readLines(t, br)...)
	chunks := 0
	for _, l := range lines {
		if l.Chunk != 0 {
			chunks++
		}
		if l.Result != nil && chunks != 2 {
			t.Fatalf("the result line came after %d chunks", chunks)
		}
	}
	if rows, _ := openAnswer(t, key, lines); string(rows) != "a\nb\n" {
		t.Fatalf("rows %q", rows)
	}
}

// dbfHeldWriter is a ResponseWriter that holds the write of a line
// containing holdOn until until: a client that takes one line slowly. A
// stand-in for a network client, in a unit test.
type dbfHeldWriter struct {
	header http.Header
	holdOn string
	until  time.Time
	mu     sync.Mutex
	buf    bytes.Buffer
}

func (w *dbfHeldWriter) Header() http.Header { return w.header }
func (w *dbfHeldWriter) WriteHeader(int)     {}
func (w *dbfHeldWriter) Flush()              {}

func (w *dbfHeldWriter) Write(p []byte) (int, error) {
	if bytes.Contains(p, []byte(w.holdOn)) {
		time.Sleep(time.Until(w.until))
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.buf.Write(p)
}

// Plan review finding 35: the owner's deadline holds for the result line
// too. Every chunk and result.json are in the bucket, but the client takes
// the last chunk's line until past the deadline: the read ends no-result,
// and the result is not relayed. Mutation: drop the ownCtx check before the
// result line (red: the result is relayed).
func TestDBReadResultAfterTheDeadlineIsNotRelayed(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0
	key, err := dbreads.NewReplyKey()
	if err != nil {
		t.Fatal(err)
	}
	id, err := dbreads.MintID()
	if err != nil {
		t.Fatal(err)
	}
	box := &scriptedBox{t: t, s: fx.rel, pub: key.PublicKey(), id: id}
	deadline := time.Now().Add(800 * time.Millisecond)
	box.chunk(1, true, "a\n")
	box.result(1, "a\n")
	box.picked(deadline)
	w := &dbfHeldWriter{header: http.Header{}, holdOn: `"chunk":1,`, until: deadline.Add(200 * time.Millisecond)}
	r := httptest.NewRequest(http.MethodPost, "/", nil)
	fx.srv.streamDBRead(w, r, http.NewResponseController(w), dbReadAdmitted{id: id, region: "uk", issued: time.Now().UTC().Truncate(time.Second)})
	w.mu.Lock()
	lines := readLines(t, bytes.NewReader(w.buf.Bytes()))
	w.mu.Unlock()
	chunks := 0
	for _, l := range lines {
		if l.Result != nil {
			t.Fatalf("a result relayed past the owner's deadline: %+v", l.Result)
		}
		if l.Chunk != 0 {
			chunks++
		}
	}
	if last := lines[len(lines)-1]; chunks != 1 || last.Ended != dbreads.EndedNoResult {
		t.Fatalf("%d chunks, last line %+v", chunks, last)
	}
}

// dbfDeadlineListener records, for every read deadline the server sets on
// its connections, how far ahead it was set.
type dbfDeadlineListener struct {
	net.Listener
	mu    sync.Mutex
	ahead []time.Duration
}

type dbfDeadlineConn struct {
	net.Conn
	l *dbfDeadlineListener
}

func (l *dbfDeadlineListener) Accept() (net.Conn, error) {
	c, err := l.Listener.Accept()
	if err != nil {
		return nil, err
	}
	return dbfDeadlineConn{Conn: c, l: l}, nil
}

func (c dbfDeadlineConn) SetReadDeadline(t time.Time) error {
	if !t.IsZero() {
		c.l.mu.Lock()
		c.l.ahead = append(c.l.ahead, time.Until(t))
		c.l.mu.Unlock()
	}
	return c.Conn.SetReadDeadline(t)
}

// Astra round 1, P1: complete query JSON followed by padding, with the rest
// of the declared body withheld. A decoder that stops after one JSON value
// would admit the read, and the server would later drain the tail under the
// extended read deadline. The whole body is read under s.dbf.preHeader
// instead, so the request is cut there and nothing is published.
// Mutation: decode one value without reading to EOF (red: the read is
// queued and the connection is still held at 3 s).
func TestDBReadWithheldBodyTailIsCutAtThePreHeaderBound(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.preHeader = 300 * time.Millisecond
	body, _ := queryBody(t, "SELECT 1")
	raw, _ := json.Marshal(body)
	conn, err := net.Dial("tcp", strings.TrimPrefix(fx.ts.URL, "http://"))
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	path := strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL)
	fmt.Fprintf(conn, "POST %s HTTP/1.1\r\nHost: console\r\nAuthorization: Bearer %s\r\n"+
		"Content-Type: application/json\r\nContent-Length: 16384\r\n\r\n%s%s", path, fx.member, raw, strings.Repeat(" ", 64))
	conn.SetReadDeadline(time.Now().Add(3 * time.Second))
	start := time.Now()
	_, err = io.ReadAll(conn)
	if ne, ok := err.(net.Error); ok && ne.Timeout() {
		t.Fatalf("the withheld body tail still held the request after %s", time.Since(start))
	}
	if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a request with a withheld body tail wrote %v", keys)
	}
}

// A body holding more than one JSON value is refused, not half-read.
func TestDBReadRefusesTrailingJSON(t *testing.T) {
	fx := newDBFFixture(t, true)
	body, _ := queryBody(t, "SELECT 1")
	raw, _ := json.Marshal(body)
	req, _ := http.NewRequest(http.MethodPost, fx.readURL(dbfHost), strings.NewReader(string(raw)+`{"op":"query"}`))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+fx.member)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	if status, out := refusal(t, res); status != http.StatusBadRequest {
		t.Fatalf("%d %s", status, out)
	}
	if keys, _ := fx.rel.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a body with trailing JSON wrote %v", keys)
	}
}

// Plan review findings 13 and 38: admission reads under s.dbf.preHeader,
// and once admitted the read deadline moves out to the read's bound plus a
// minute. Observed on the connection itself: once a body has been read to
// its end, net/http's background read clears the read deadline on its own,
// so a stream that outlives ReadTimeout does not show the extension.
// Mutations: drop the extension (red: no deadline past 22 minutes); drop the
// preHeader deadline (red: none between 15 and 20 s; the server's own are
// 10 s and 5 minutes).
func TestDBReadSetsBothReadDeadlines(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.pickup = 10*time.Millisecond, 1500*time.Millisecond
	raw, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	ln := &dbfDeadlineListener{Listener: raw}
	hs := NewHTTPServer(fx.srv)
	go hs.Serve(ln)
	t.Cleanup(func() { hs.Close() })
	body, _ := queryBody(t, "SELECT 1")
	rawBody, _ := json.Marshal(body)
	req, _ := http.NewRequest(http.MethodPost, "http://"+raw.Addr().String()+strings.TrimPrefix(fx.readURL(dbfHost), fx.ts.URL), bytes.NewReader(rawBody))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+fx.member)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	io.Copy(io.Discard, res.Body)
	res.Body.Close()
	ln.mu.Lock()
	defer ln.mu.Unlock()
	var pre, extended bool
	for _, d := range ln.ahead {
		pre = pre || (d > 15*time.Second && d <= 20*time.Second)
		extended = extended || d > 22*time.Minute
	}
	if !pre || !extended {
		t.Fatalf("read deadlines set this far ahead: %v", ln.ahead)
	}
}

// Review Focus 3: a picked deadline far in the future does not hold the
// request past its own bound.
func TestDBReadRequestBoundClampsAForgedDeadline(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.bound = 10*time.Millisecond, 1500*time.Millisecond
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	scriptFor(t, fx.rel, adm.Admitted).picked(time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC))
	done := make(chan []dbreads.Line)
	go func() { done <- readLines(t, br) }()
	select {
	case lines := <-done:
		if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
			t.Fatalf("last line %+v", last)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("the request outlived its bound")
	}
	res.Body.Close()
}

// Review Focus 4: a result.json that is not a result is never relayed.
func TestDBReadIgnoresAMalformedResult(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	box := scriptFor(t, fx.rel, adm.Admitted)
	box.picked(time.Now().Add(500 * time.Millisecond))
	fx.rel.Put(context.Background(), dbreads.ResultKey(adm.Admitted), []byte(`{"state":"maybe","last_seq":-1}`), "")
	lines := readLines(t, br)
	res.Body.Close()
	for _, l := range lines {
		if l.Result != nil {
			t.Fatalf("a malformed result was relayed: %+v", l.Result)
		}
	}
	if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
		t.Fatalf("last line %+v", last)
	}
}

// strictStore fails the test on any delete or replacing write under
// db-reads/ (T16, the console's half).
type strictStore struct {
	release.Store
	t *testing.T
}

func (s strictStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if strings.HasPrefix(key, dbreads.Prefix) && ifMatch != "" {
		s.t.Errorf("a replacing write of %s", key)
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

func (s strictStore) Delete(ctx context.Context, key string) error {
	if strings.HasPrefix(key, dbreads.Prefix) {
		s.t.Errorf("a delete of %s", key)
	}
	return s.Store.Delete(ctx, key)
}

// T16 (codex 4, 13, 14): a box that writes after the request ended changes
// nothing any reader sees, and the console never deletes or replaces.
func TestDBReadLateWritesChangeNothing(t *testing.T) {
	fx := newDBFFixture(t, true)
	fx.srv.cfg.Releases = strictStore{Store: fx.rel, t: t}
	fx.srv.dbf.poll, fx.srv.dbf.grace = 10*time.Millisecond, 0
	body, _ := queryBody(t, "SELECT 1")
	res := fx.openRead(t, dbfHost, fx.member, body)
	br := bufio.NewReader(res.Body)
	first, _ := br.ReadBytes('\n')
	var adm dbreads.Line
	json.Unmarshal(first, &adm)
	box := scriptFor(t, fx.rel, adm.Admitted)
	box.picked(time.Now().Add(200 * time.Millisecond))
	lines := readLines(t, br) // ends no-result at the deadline
	res.Body.Close()
	box.chunk(1, true, "late\n")
	box.result(1, "late\n")
	if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
		t.Fatalf("last line %+v", last)
	}
	keys, _ := fx.rel.List(context.Background(), dbreads.Prefix+adm.Admitted+"/")
	for _, k := range keys {
		if strings.HasSuffix(k, "/cancel") {
			return
		}
	}
	t.Fatalf("no cancel among %v", keys)
}