package console

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/aws/smithy-go"
	lsfile "github.com/benbjohnson/litestream/file"
	lss3 "github.com/benbjohnson/litestream/s3"
	"github.com/superfly/ltx"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/svcrole"
)

// Every one of these is refused BEFORE any S3 call. A door that reached the
// bucket and mapped AccessDenied would leak both latency and the fact that
// a bucket was tried. resolveReplica takes no store at all, so it cannot.
func TestResolveReplicaRefusesBeforeTouchingS3(t *testing.T) {
	s := &Server{}
	s.cfg.Region = "eu-west-1"
	s.cfg.HibernatePrefix = "instances"

	cases := []struct {
		name string
		rec  provision.Record
		host string
		want error
	}{
		{
			name: "another region",
			rec:  provision.Record{Host: "a.bab.oncarlos.com", Region: "us-east-1"},
			host: "a.bab.oncarlos.com",
			want: ErrResidency,
		},
		{
			name: "the home region seen from a regional console",
			rec:  provision.Record{Host: "a.bab.oncarlos.com"},
			host: "a.bab.oncarlos.com",
			want: ErrResidency,
		},
		{
			name: "fleet-homed",
			rec:  provision.Record{Host: "a.bab.oncarlos.com", Region: "fleet/acme"},
			host: "a.bab.oncarlos.com",
			want: ErrResidency,
		},
		{
			name: "pooled",
			rec:  provision.Record{Host: "a.bab.oncarlos.com", Region: "eu-west-1", Pool: "shared"},
			host: "a.bab.oncarlos.com",
			want: ErrPooled,
		},
		{
			name: "hostile host",
			rec:  provision.Record{Host: "../../etc", Region: "eu-west-1"},
			host: "../../etc",
			want: ErrBadHost,
		},
		{
			name: "record for another host",
			rec:  provision.Record{Host: "b.bab.oncarlos.com", Region: "eu-west-1"},
			host: "a.bab.oncarlos.com",
			want: ErrBadHost,
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, err := s.resolveReplica(tc.rec, tc.host)
			if !errors.Is(err, tc.want) {
				t.Fatalf("resolveReplica = %v, want %v", err, tc.want)
			}
		})
	}
}

// A fleet placement is refused even on a console whose own region string
// happened to match it: the data is on the customer's box, whatever a
// misconfigured CARLOS_REGION says.
func TestResolveReplicaRefusesAFleetEvenWhenTheRegionMatches(t *testing.T) {
	s := &Server{}
	s.cfg.Region = "fleet/acme"
	s.cfg.HibernatePrefix = "instances"
	if _, err := s.resolveReplica(provision.Record{Host: "a.bab.oncarlos.com", Region: "fleet/acme"}, "a.bab.oncarlos.com"); !errors.Is(err, ErrResidency) {
		t.Fatalf("resolveReplica = %v, want ErrResidency", err)
	}
}

// The happy path is an in-region, non-pooled instance, and the prefix is
// the activator's own layout, not a string this package invents.
func TestResolveReplicaBuildsTheActivatorPrefix(t *testing.T) {
	s := &Server{}
	s.cfg.Region = "eu-west-1"
	s.cfg.HibernatePrefix = "instances"
	got, err := s.resolveReplica(
		provision.Record{Host: "a.bab.oncarlos.com", Region: "eu-west-1"},
		"a.bab.oncarlos.com")
	if err != nil {
		t.Fatal(err)
	}
	if got != "instances/a.bab.oncarlos.com" {
		t.Fatalf("prefix = %q, want instances/a.bab.oncarlos.com", got)
	}
}

// A single-region deployment leaves both empty, and "" == "" must be a
// match rather than a refusal, otherwise the door never works anywhere
// that has not set CARLOS_REGION. Home-region records keep region "" too
// (Config.HomeRegion), so this is also the flagship's home console.
func TestResolveReplicaTreatsEmptyRegionsAsAMatch(t *testing.T) {
	s := &Server{}
	s.cfg.HibernatePrefix = "instances"
	if _, err := s.resolveReplica(
		provision.Record{Host: "a.bab.oncarlos.com"}, "a.bab.oncarlos.com"); err != nil {
		t.Fatalf("single-region deployment refused: %v", err)
	}
}

// ---- the door handlers (Task 4) ----------------------------------------------

const dbTestHost = "acme.go.tito.io"

type dbFx struct {
	*consoleFixture
	member string
	root   string        // the fake data bucket: one file replica per prefix
	opens  *atomic.Int32 // how many times the door reached for a replica client
}

// dbFixture is an in-region, hibernating exec instance of titogo whose
// replica is a real litestream replica on disk, served through the door's
// replica seam exactly as the S3 client would serve it.
func dbFixture(t *testing.T) *dbFx {
	t.Helper()
	fx, member := instanceFixture(t, "code-db")
	root := t.TempDir()
	fx.srv.cfg.HibernateBucket = "data-test"
	fx.srv.cfg.HibernatePrefix = "instances"
	opens := new(atomic.Int32)
	fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
		opens.Add(1)
		return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
	}
	putDBRecord(t, fx, provision.Record{Host: dbTestHost})
	return &dbFx{consoleFixture: fx, member: member, root: root, opens: opens}
}

// putDBRecord writes an instance record for titogo with the fields a
// hibernating exec instance has, overridden by whatever rec sets.
func putDBRecord(t *testing.T, fx *consoleFixture, rec provision.Record) {
	t.Helper()
	rec.App, rec.Account = "titogo", release.OpsAccount
	if rec.Hibernate == nil {
		yes := true
		rec.Hibernate = &yes
	}
	body, err := json.Marshal(rec)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(context.Background(), provision.RecordKey(rec.Account, rec.App, rec.Host), body, ""); err != nil {
		t.Fatal(err)
	}
}

func (fx *dbFx) replicate(t *testing.T, host string, batches int) string {
	t.Helper()
	return dbdoortest.Replicate(t, filepath.Join(fx.root, "instances", host), batches)
}

func (fx *dbFx) url(p string) string { return fx.ts.URL + p }

// A level the activator never writes is EMPTY, not missing. A 404 here
// stops CalcRestorePlan at the first gap and the restore fails.
func TestAPIDBListAnswersEmptyForUnwrittenLevels(t *testing.T) {
	fx := dbFixture(t)
	for level := 0; level <= dbdoor.MaxLevel; level++ {
		status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, level)), fx.member)
		if status != http.StatusOK {
			t.Fatalf("level %d = %d %s, want 200 with an empty list", level, status, out)
		}
		var res dbdoor.ListResponse
		if err := json.Unmarshal([]byte(out), &res); err != nil || res.Files == nil || len(res.Files) != 0 {
			t.Fatalf("level %d = %s, want {\"files\":[]}", level, out)
		}
	}
}

// Wire specifics 2 and 4: ascending by min_txid, and size, min_txid and
// max_txid on every entry.
func TestAPIDBListAscendsAndCarriesTheMandatoryFields(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 3)
	status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member)
	if status != http.StatusOK {
		t.Fatalf("list = %d %s", status, out)
	}
	var res dbdoor.ListResponse
	if err := json.Unmarshal([]byte(out), &res); err != nil {
		t.Fatal(err)
	}
	if len(res.Files) < 2 {
		t.Fatalf("list = %s, want several L0 files", out)
	}
	for i, f := range res.Files {
		if f.Size <= 0 || f.MinTXID == 0 || f.MaxTXID < f.MinTXID || f.Level != 0 {
			t.Errorf("entry %d = %+v, want level 0 with size, min_txid and max_txid", i, f)
		}
		if i > 0 && f.MinTXID <= res.Files[i-1].MinTXID {
			t.Errorf("entries %d and %d are not ascending by min_txid: %s", i-1, i, out)
		}
	}
	for _, field := range []string{`"size"`, `"min_txid"`, `"max_txid"`} {
		if !strings.Contains(out, field) {
			t.Errorf("list body lacks %s: %s", field, out)
		}
	}
}

// get streams the file by coordinate, byte-for-byte, and honours the two
// Range forms litestream's client sends.
func TestAPIDBGetServesTheFileAndHonoursRange(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	c := lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost))
	files := listFiles(t, c, 0)
	f := files[0]
	want, err := os.ReadFile(c.LTXFilePath(0, f.MinTXID, f.MaxTXID))
	if err != nil {
		t.Fatal(err)
	}
	u := fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID)))
	status, _, out := rawReq(t, http.MethodGet, u, fx.member, "")
	if status != http.StatusOK || out != string(want) {
		t.Fatalf("get = %d, %d bytes; want 200 and the file's %d bytes", status, len(out), len(want))
	}
	for _, rc := range []struct{ offset, size int64 }{{10, 0}, {10, 20}} {
		req, _ := http.NewRequest(http.MethodGet, u, nil)
		req.Header.Set("Authorization", "Bearer "+fx.member)
		req.Header.Set("Range", dbdoor.RangeHeader(rc.offset, rc.size))
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		got, _ := io.ReadAll(res.Body)
		res.Body.Close()
		end := int64(len(want))
		if rc.size > 0 {
			end = rc.offset + rc.size
		}
		if res.StatusCode != http.StatusPartialContent || string(got) != string(want[rc.offset:end]) {
			t.Errorf("range %+v = %d, %d bytes; want 206 and bytes %d-%d", rc, res.StatusCode, len(got), rc.offset, end)
		}
	}
}

func listFiles(t *testing.T, c *lsfile.ReplicaClient, level int) []*ltx.FileInfo {
	t.Helper()
	itr, err := c.LTXFiles(context.Background(), level, 0, false)
	if err != nil {
		t.Fatal(err)
	}
	defer itr.Close()
	var out []*ltx.FileInfo
	for itr.Next() {
		out = append(out, itr.Item())
	}
	if len(out) == 0 {
		t.Fatal("no files at that level")
	}
	return out
}

// A planned file that has gone is a 404 the client maps to os.ErrNotExist
// (wire specific 5), which is what makes the whole-Restore retry work.
func TestAPIDBGetOfAMissingFileIsA404(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	status, _ := getRaw(t, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 900, 901)), fx.member)
	if status != http.StatusNotFound {
		t.Fatalf("get of a missing coordinate = %d, want 404", status)
	}
}

// Coordinate addressing exists to make lock.json unreachable. This is the
// test that keeps the door from decaying into an object proxy.
func TestAPIDBServesNoKeyOutsideItsCoordinates(t *testing.T) {
	fx := dbFixture(t)
	dir := filepath.Join(fx.root, "instances", dbTestHost)
	fx.replicate(t, dbTestHost, 1)
	if err := os.WriteFile(filepath.Join(dir, "lock.json"), []byte(`{"owner":"box-secret"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	base := "/api/cli/apps/bab/titogo/db/" + dbTestHost
	for _, bad := range []string{
		base + "/ltx/0/../../lock.json",
		base + "/lock.json",
		base + "/ltx/lock.json",
		base + "/ltx/0/lock.json",
		base + "/ltx/10",
		base + "/ltx/-1",
		base + "/ltx/0/1",
		base + "/ltx/0/0-1",
		base + "/ltx/0/2-1",
		base + "/ltx/10/1-2",
		"/api/cli/apps/bab/titogo/db/..%2F..%2Fetc/ltx/0",
		"/api/cli/apps/bab/titogo/db/" + "b.go.tito.io" + "/ltx/0",
	} {
		status, _, out := rawReq(t, http.MethodGet, fx.url(bad), fx.member, "")
		if status == http.StatusOK || status == http.StatusPartialContent || strings.Contains(out, "box-secret") {
			t.Errorf("%s = %d %s, want a refusal", bad, status, out)
		}
	}
}

// Not a member, no such app, no such instance: the same bytes http.NotFound
// writes, matching instanceGate, so the door is no oracle for any of them.
func TestAPIDBStrangersMeetTheByteIdentical404(t *testing.T) {
	fx := dbFixture(t)
	want := canonicalNotFound(t)
	stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-db-stranger")
	for name, c := range map[string]struct{ tok, path string }{
		"not a member":      {stranger, dbdoor.ListPath("bab", "titogo", dbTestHost, 0)},
		"no such app":       {fx.member, dbdoor.ListPath("bab", "nosuch", dbTestHost, 0)},
		"no such instance":  {fx.member, dbdoor.ListPath("bab", "titogo", "nobody.go.tito.io", 0)},
		"no such account":   {fx.member, dbdoor.ListPath("zzzz", "titogo", dbTestHost, 0)},
		"get, not a member": {stranger, dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)},
	} {
		status, h, body := rawReq(t, http.MethodGet, fx.url(c.path), c.tok, "")
		assertNotFound(t, name, want, status, h, body)
	}
	if n := fx.opens.Load(); n != 0 {
		t.Fatalf("the door reached for a replica %d times for callers it refuses", n)
	}
}

// Every service principal, every role including admin, and the legacy app
// token meet the closed-door refusal before any handler runs: the same bytes
// whether or not the app and the instance exist, so it says nothing about
// THIS app (Ruling, Task 4: the repo's zero-Set closed door, not a 404).
func TestAPIDBRefusesEveryServicePrincipalTheSameWay(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	appTok, err := fx.st.NewAppToken(OpsAccountID, "titogo", "alice@keymail.dev", "signup", time.Now())
	if err != nil {
		t.Fatal(err)
	}
	toks := map[string]string{"legacy app token": appTok}
	for _, role := range []svcrole.Role{svcrole.Publish, svcrole.Instance, svcrole.Operate, svcrole.Admin} {
		toks[string(role)] = serviceToken(t, fx.st, OpsAccountID, "svc-"+string(role), "titogo", role)
	}
	var first string
	for name, tok := range toks {
		for _, p := range []string{
			dbdoor.ListPath("bab", "titogo", dbTestHost, 0),
			dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1),
			dbdoor.ListPath("bab", "nosuch", "nobody.go.tito.io", 0),
		} {
			status, _, body := rawReq(t, http.MethodGet, fx.url(p), tok, "")
			if status != http.StatusForbidden || !strings.Contains(body, "closed to service credentials") {
				t.Errorf("%s at %s = %d %s, want the closed-door refusal", name, p, status, body)
			}
			if first == "" {
				first = body
			} else if body != first {
				t.Errorf("%s at %s: refusal varies (%q vs %q)", name, p, body, first)
			}
		}
	}
	if n := fx.opens.Load(); n != 0 {
		t.Fatalf("the door reached for a replica %d times for service principals", n)
	}
}

// The residency and pooled refusals are answered with no replica client
// ever built: a 409 naming the cause, before the bucket.
func TestAPIDBResidencyRefusalsNeverReachTheBucket(t *testing.T) {
	fx := dbFixture(t)
	fx.srv.cfg.Region = "ie"
	putDBRecord(t, fx.consoleFixture, provision.Record{Host: "uk.go.tito.io", Region: "uk"})
	putDBRecord(t, fx.consoleFixture, provision.Record{Host: "fleet.go.tito.io", Region: "fleet/acme"})
	putDBRecord(t, fx.consoleFixture, provision.Record{Host: "pool.go.tito.io", Region: "ie", Pool: "tito"})
	for host, want := range map[string]string{
		"uk.go.tito.io":    "uk",
		"fleet.go.tito.io": "fleet",
		"pool.go.tito.io":  "pool",
		dbTestHost:         "",
	} {
		status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", host, 0)), fx.member)
		if status != http.StatusConflict || !strings.Contains(out, want) {
			t.Errorf("%s = %d %s, want a 409 naming %q", host, status, out, want)
		}
	}
	if n := fx.opens.Load(); n != 0 {
		t.Fatalf("the door reached for a replica %d times before refusing on residency", n)
	}
}

// Only an activator-managed exec instance has a replica the platform
// writes. Everything else is refused by class, before the bucket.
func TestAPIDBRefusesRoutesTheActivatorDoesNotReplicate(t *testing.T) {
	fx := dbFixture(t)
	no := false
	putDBRecord(t, fx.consoleFixture, provision.Record{Host: "svc.go.tito.io", Kind: "service"})
	putDBRecord(t, fx.consoleFixture, provision.Record{Host: "addr.go.tito.io", Addr: "10.0.0.1:8080"})
	putDBRecord(t, fx.consoleFixture, provision.Record{Host: "awake.go.tito.io", Hibernate: &no})
	for _, host := range []string{"svc.go.tito.io", "addr.go.tito.io", "awake.go.tito.io"} {
		status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", host, 0)), fx.member)
		if status != http.StatusConflict {
			t.Errorf("%s = %d %s, want a 409 refusal", host, status, out)
		}
	}
	if n := fx.opens.Load(); n != 0 {
		t.Fatalf("the door reached for a replica %d times for routes with none", n)
	}
}

// No data bucket configured is a 503, never a door that fails open.
func TestAPIDBWithoutADataBucketIs503(t *testing.T) {
	fx := dbFixture(t)
	fx.srv.cfg.HibernateBucket = ""
	status, out := getRaw(t, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member)
	if status != http.StatusServiceUnavailable {
		t.Fatalf("no bucket = %d %s, want 503", status, out)
	}
	if fx.opens.Load() != 0 {
		t.Fatal("the door reached for a replica with no bucket configured")
	}
}

// S3 refusing the read is the residency refusal as a backstop (spec §6),
// never a generic 502 — and the store's own error text, which carries the
// bucket name, never reaches the caller.
func TestAPIDBAccessDeniedIsTheResidencyRefusal(t *testing.T) {
	fx := dbFixture(t)
	fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) {
		return deniedReplica{}, nil
	}
	for _, p := range []string{
		dbdoor.ListPath("bab", "titogo", dbTestHost, 0),
		dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1),
	} {
		status, out := getRaw(t, fx.url(p), fx.member)
		if status != http.StatusConflict || strings.Contains(out, "data-test") || strings.Contains(out, "secret-bucket") {
			t.Errorf("%s = %d %s, want the residency 409 with no bucket name", p, status, out)
		}
	}
}

type deniedReplica struct{}

func (deniedReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
	return nil, &smithy.GenericAPIError{Code: "AccessDenied", Message: "no access to secret-bucket"}
}

func (deniedReplica) OpenLTXFile(context.Context, int, ltx.TXID, ltx.TXID, int64, int64) (io.ReadCloser, error) {
	return nil, &smithy.GenericAPIError{Code: "AccessDenied", Message: "no access to secret-bucket"}
}

// Every successful open of a file writes an audit row naming the instance
// and the TXID span served (spec §6 "Who may open it").
func TestAPIDBGetIsAudited(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
	if status, _ := getRaw(t, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID))), fx.member); status != http.StatusOK {
		t.Fatalf("get = %d", status)
	}
	evs, err := fx.st.Events(OpsAccountID, "titogo", 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, e := range evs {
		if e.Action == "db-read" && e.Target == dbTestHost && e.Address == "alice@keymail.dev" &&
			strings.Contains(e.Detail, fmt.Sprint(uint64(f.MaxTXID))) {
			return
		}
	}
	t.Fatalf("no db-read audit row naming %s and txid %d: %+v", dbTestHost, f.MaxTXID, evs)
}

// One replica client per prefix: a restore's dozens of requests share one
// transport rather than paying a handshake each. And the production reader
// carries the configured bucket and the resolved prefix, nothing a request
// supplied.
func TestCachedReadReplicaIsOnePerPrefix(t *testing.T) {
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
	s := &Server{}
	s.cfg.HibernateBucket = "data-ie"
	s.cfg.HibernateRegion = "eu-west-1"
	ctx := context.Background()
	a1, err := s.cachedReadReplica(ctx, "instances/a.example.com")
	if err != nil {
		t.Fatal(err)
	}
	a2, _ := s.cachedReadReplica(ctx, "instances/a.example.com")
	b, _ := s.cachedReadReplica(ctx, "instances/b.example.com")
	if a1 != a2 {
		t.Error("the same prefix built two clients")
	}
	if a1 == b {
		t.Error("two prefixes shared one client")
	}
	c, ok := a1.(*lss3.ReplicaClient)
	if !ok || c.Bucket != "data-ie" || c.Path != "instances/a.example.com" || c.Region != "eu-west-1" {
		t.Fatalf("client = %#v, want the configured bucket, region and the resolved prefix", a1)
	}
}

// The door has its own token-keyed budget, separate from the instance
// verbs': spending it does not touch the instance budget, and it applies
// only after membership, so a stranger cannot learn anything from a 429.
func TestAPIDBHasItsOwnBudget(t *testing.T) {
	fx := dbFixture(t)
	fx.srv.cli.db = newRateLimiter(2, cliRateWindow)
	u := fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0))
	for i := 0; i < 2; i++ {
		if status, out := getRaw(t, u, fx.member); status != http.StatusOK {
			t.Fatalf("request %d = %d %s", i, status, out)
		}
	}
	if status, _ := getRaw(t, u, fx.member); status != http.StatusTooManyRequests {
		t.Fatalf("third request = %d, want 429", status)
	}
	if !fx.srv.cli.instances.allow(hashToken(fx.member), time.Now()) {
		t.Fatal("the db door spent the instance verbs' budget")
	}
	stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-db-budget")
	for i := 0; i < 3; i++ {
		if status, _ := getRaw(t, u, stranger); status != http.StatusNotFound {
			t.Fatalf("stranger request %d = %d, want the 404 before any budget", i, status)
		}
	}
}

// End to end: the CLI's read-only client restores a real replica through
// the real door, with a member's token, and gets every row.
func TestAPIDBRestoresThroughTheDoor(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 3)
	// The tightest bound there is, and a restore still completes under it.
	// (Small files finish server-side before the client reads them, so the
	// one-stream-at-a-time property itself is pinned in internal/dbdoor.)
	fx.srv.dbStreams = newDBStreamLimits(1, 1)
	do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
		if err != nil {
			return nil, err
		}
		req.Header = h.Clone()
		req.Header.Set("Authorization", "Bearer "+fx.member)
		return http.DefaultClient.Do(req)
	}
	got, err := dbdoor.RestoreTo(context.Background(), dbdoor.NewClient(do), t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(got.Dir)
	db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	var n int
	if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	if n != 3*dbdoortest.Rows || got.TXID == 0 || got.At.IsZero() {
		t.Fatalf("restored %d rows, txid %d at %v; want %d rows and an as-of", n, got.TXID, got.At, 3*dbdoortest.Rows)
	}
}

// A ranged GET is still a read of the file. Listing the coordinates and
// fetching each one with Range: bytes=1- returns all but one byte of every
// file, so every successful GET writes a row, and a resumed one says where
// it started.
func TestAPIDBRangedGetIsAudited(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
	req, _ := http.NewRequest(http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID))), nil)
	req.Header.Set("Authorization", "Bearer "+fx.member)
	req.Header.Set("Range", "bytes=1-")
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	io.Copy(io.Discard, res.Body)
	res.Body.Close()
	if res.StatusCode != http.StatusPartialContent {
		t.Fatalf("ranged get = %d", res.StatusCode)
	}
	evs, err := fx.st.Events(OpsAccountID, "titogo", 10)
	if err != nil {
		t.Fatal(err)
	}
	for _, e := range evs {
		if e.Action == "db-read" && e.Target == dbTestHost && strings.Contains(e.Detail, "byte 1") {
			return
		}
	}
	t.Fatalf("a ranged read wrote no db-read row naming its start: %+v", evs)
}

// stallReplica serves listings from a file replica but opens files whose
// reads never finish: each blocks until the request's context ends, the way
// a stalled S3 read does, or produces zeros forever when endless is set.
type stallReplica struct {
	replicaReader
	endless bool
	opened  chan struct{}
}

func (s *stallReplica) OpenLTXFile(ctx context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	select {
	case s.opened <- struct{}{}:
	default:
	}
	if s.endless {
		return io.NopCloser(zeros{}), nil
	}
	return io.NopCloser(ctxReader{ctx}), nil
}

type ctxReader struct{ ctx context.Context }

func (r ctxReader) Read([]byte) (int, error) { <-r.ctx.Done(); return 0, r.ctx.Err() }

type zeros struct{}

func (zeros) Read(p []byte) (int, error) { clear(p); return len(p), nil }

func (fx *dbFx) stall(endless bool) *stallReplica {
	sr := &stallReplica{endless: endless, opened: make(chan struct{}, 16)}
	fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
		sr.replicaReader = lsfile.NewReplicaClient(filepath.Join(fx.root, prefix))
		return sr, nil
	}
	return sr
}

func (fx *dbFx) memberToken(t *testing.T, addr, code string) string {
	t.Helper()
	if err := fx.st.AddMember(OpsAccountID, addr, RoleMember, time.Now()); err != nil {
		t.Fatal(err)
	}
	return cliToken(t, fx.consoleFixture, addr, code)
}

// startGet begins a GET and returns once the door has opened the file, so
// the stream holds its slot; cancel ends it.
func (fx *dbFx) startGet(t *testing.T, sr *stallReplica, tok string) (stop func()) {
	t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
	req.Header.Set("Authorization", "Bearer "+tok)
	done := make(chan struct{})
	go func() {
		defer close(done)
		if res, err := http.DefaultClient.Do(req); err == nil {
			io.Copy(io.Discard, res.Body)
			res.Body.Close()
		}
	}()
	select {
	case <-sr.opened:
	case <-time.After(5 * time.Second):
		t.Fatal("the stream never opened")
	}
	return func() { cancel(); <-done }
}

func (fx *dbFx) activeStreams() int {
	fx.srv.dbStreams.mu.Lock()
	defer fx.srv.dbStreams.mu.Unlock()
	return fx.srv.dbStreams.total
}

func (fx *dbFx) waitIdle(t *testing.T, within time.Duration) {
	t.Helper()
	deadline := time.Now().Add(within)
	for fx.activeStreams() != 0 {
		if time.Now().After(deadline) {
			t.Fatalf("%d streams still hold slots after %v", fx.activeStreams(), within)
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// Slow downloads must not accumulate: active file streams are bounded per
// principal and across the console, and a refusal says when to come back.
func TestAPIDBBoundsConcurrentStreams(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	fx.srv.dbStreams = newDBStreamLimits(2, 1)
	sr := fx.stall(false)
	bob := fx.memberToken(t, "bob@keymail.dev", "code-db-bob")
	carol := fx.memberToken(t, "carol@keymail.dev", "code-db-carol")
	u := fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1))

	stopAlice := fx.startGet(t, sr, fx.member)
	status, h, body := rawReq(t, http.MethodGet, u, fx.member, "")
	if status != http.StatusTooManyRequests || h.Get("Retry-After") == "" {
		t.Fatalf("second stream for one person = %d %q %s, want 429 with Retry-After", status, h.Get("Retry-After"), body)
	}
	stopBob := fx.startGet(t, sr, bob)
	status, h, body = rawReq(t, http.MethodGet, u, carol, "")
	if status != http.StatusTooManyRequests || h.Get("Retry-After") == "" {
		t.Fatalf("stream past the console's bound = %d %q %s, want 429 with Retry-After", status, h.Get("Retry-After"), body)
	}
	stopAlice()
	stopBob()
	fx.waitIdle(t, 5*time.Second)
	stopAgain := fx.startGet(t, sr, fx.member)
	stopAgain()
}

// A stalled bucket read ends after the idle deadline and frees its slot.
func TestAPIDBStalledReadIsCut(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	fx.srv.dbStreamIdle = 200 * time.Millisecond
	sr := fx.stall(false)
	start := time.Now()
	status, _, _ := rawReq(t, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), fx.member, "")
	<-sr.opened
	if time.Since(start) > 5*time.Second {
		t.Fatalf("a stalled read held the request for %v", time.Since(start))
	}
	_ = status
	fx.waitIdle(t, 5*time.Second)
}

// A client that stops reading is cut by the renewable write deadline, so a
// trickling or abandoned download cannot hold a slot forever.
func TestAPIDBStalledClientIsCut(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	fx.srv.dbStreamIdle = 300 * time.Millisecond
	sr := fx.stall(true)
	req, _ := http.NewRequest(http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
	req.Header.Set("Authorization", "Bearer "+fx.member)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer res.Body.Close() // never read: the client has stalled
	<-sr.opened
	fx.waitIdle(t, 15*time.Second)
}

// recConn records, in order, every write and every write-deadline change on
// a server-side connection.
type recConn struct {
	net.Conn
	mu     *sync.Mutex
	events *[]string
}

func (c recConn) Write(p []byte) (int, error) {
	c.mu.Lock()
	*c.events = append(*c.events, "write")
	c.mu.Unlock()
	return c.Conn.Write(p)
}

func (c recConn) SetWriteDeadline(t time.Time) error {
	c.mu.Lock()
	if t.IsZero() {
		*c.events = append(*c.events, "clear")
	} else {
		*c.events = append(*c.events, "deadline")
	}
	c.mu.Unlock()
	return c.Conn.SetWriteDeadline(t)
}

type recListener struct {
	net.Listener
	mu     sync.Mutex
	events []string
}

func (l *recListener) Accept() (net.Conn, error) {
	c, err := l.Listener.Accept()
	if err != nil {
		return nil, err
	}
	return recConn{Conn: c, mu: &l.mu, events: &l.events}, nil
}

// net/http flushes what is still buffered (the last bytes, the chunked
// terminator) in finishRequest, AFTER the handler returns, and clears the
// write deadline itself only after that. A handler that clears it on the
// way out leaves that final flush with no deadline, so a client that stops
// reading near the end holds the connection forever, outside the stream
// bound. No byte may ever be written with the deadline cleared.
func TestAPIDBNeverWritesWithoutADeadline(t *testing.T) {
	fx := dbFixture(t)
	fx.replicate(t, dbTestHost, 1)
	f := listFiles(t, lsfile.NewReplicaClient(filepath.Join(fx.root, "instances", dbTestHost)), 0)[0]
	ts := httptest.NewUnstartedServer(fx.srv)
	rl := &recListener{Listener: ts.Listener}
	ts.Listener = rl
	ts.Start()
	defer ts.Close()
	req, _ := http.NewRequest(http.MethodGet, ts.URL+dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(f.MinTXID), uint64(f.MaxTXID)), nil)
	req.Header.Set("Authorization", "Bearer "+fx.member)
	req.Close = true
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	io.Copy(io.Discard, res.Body)
	res.Body.Close()
	if res.StatusCode != http.StatusOK {
		t.Fatalf("get = %d", res.StatusCode)
	}
	deadline := time.Now().Add(5 * time.Second)
	for {
		rl.mu.Lock()
		events := append([]string(nil), rl.events...)
		rl.mu.Unlock()
		if len(events) > 0 && events[len(events)-1] == "clear" || time.Now().After(deadline) {
			armed := false
			for i, e := range events {
				switch e {
				case "deadline":
					armed = true
				case "clear":
					armed = false
				case "write":
					if !armed {
						t.Fatalf("write %d happened with no write deadline: %v", i, events)
					}
				}
			}
			if !armed && events[len(events)-1] != "clear" {
				t.Fatalf("events = %v", events)
			}
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
}