package main

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

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

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbreads"
)

// These feed readDBForward crafted line streams, sealed with real keys:
// unit tests of the CLI's half (dbforward_e2e_test.go runs the binary).

const fwdID = "dbr_aaaaaaaaaaaaaaaaaaaaaaaaaa"

// stream builds a reads-route body: admitted, picked, the chunks, then end.
type stream struct {
	t   *testing.T
	pub *ecdh.PublicKey
	buf bytes.Buffer
}

func newStream(t *testing.T, key *ecdh.PrivateKey) *stream {
	s := &stream{t: t, pub: key.PublicKey()}
	s.line(dbreads.Line{Admitted: fwdID})
	s.line(dbreads.Line{Picked: "uk-1"})
	return s
}

func (s *stream) line(l dbreads.Line) {
	b, _ := json.Marshal(l)
	s.buf.Write(append(b, '\n'))
}

func (s *stream) chunk(seq int64, last bool, plain []byte) {
	sealed, err := dbreads.SealChunk(s.pub, fwdID, seq, dbreads.StreamRows, last, plain)
	if err != nil {
		s.t.Fatal(err)
	}
	s.line(dbreads.Line{Chunk: seq, Data: sealed})
}

func (s *stream) result(state, code string, last int64, sum dbreads.Summary) {
	sealed, _ := dbreads.SealSummary(s.pub, fwdID, sum)
	s.line(dbreads.Line{Result: &dbreads.Result{State: state, Code: code, LastSeq: last, Sealed: sealed}})
}

func summaryOf(rows string) dbreads.Summary {
	h := sha256.Sum256([]byte(rows))
	return dbreads.Summary{TXID: 7, At: time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC), Bytes: int64(len(rows)), SHA256: hex.EncodeToString(h[:])}
}

func readFwd(t *testing.T, s *stream, key *ecdh.PrivateKey) (dbForwardAnswer, string, error) {
	t.Helper()
	var status bytes.Buffer
	sink := &memSink{}
	id, res, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfReading, &status, sink, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes))
	if err != nil {
		return dbForwardAnswer{}, status.String(), err
	}
	ans, err := openDBForward(id, key, sink, *res, "shop.example.com", "uk")
	return ans, status.String(), err
}

func TestForwardedQueryPrintsOnlyACheckedAnswer(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, false, []byte("n\n"))
	s.chunk(2, true, []byte("30\n"))
	s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n"))
	ans, status, err := readFwd(t, s, key)
	if err != nil || string(ans.rows) != "n\n30\n" || ans.summary.TXID != 7 {
		t.Fatalf("%q %v", ans.rows, err)
	}
	if status != fmt.Sprintf(dbfReading, "shop.example.com", "uk")+"\n" {
		t.Fatalf("status %q", status)
	}
}

// T11's unit half (codex 6): a dropped or bad last chunk and a summary that
// does not add up each print nothing.
func TestForwardedQueryRefusesAnAnswerThatDoesNotCheckOut(t *testing.T) {
	t.Parallel()
	unchecked := fmt.Sprintf(dbfUnchecked, "uk")
	for name, build := range map[string]func(s *stream){
		"the last chunk dropped": func(s *stream) {
			s.chunk(1, false, []byte("n\n"))
			s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n"))
		},
		"the last chunk not sealed as last": func(s *stream) {
			s.chunk(1, false, []byte("n\n"))
			s.chunk(2, false, []byte("30\n"))
			s.result(dbreads.StateDone, "", 2, summaryOf("n\n30\n"))
		},
		"a hash mismatch": func(s *stream) {
			s.chunk(1, true, []byte("n\n31\n"))
			s.result(dbreads.StateDone, "", 1, summaryOf("n\n30\n"))
		},
		"a chunk out of order": func(s *stream) {
			s.chunk(2, true, []byte("30\n"))
			s.result(dbreads.StateDone, "", 1, summaryOf("30\n"))
		},
	} {
		key, _ := dbreads.NewReplyKey()
		s := newStream(t, key)
		build(s)
		ans, _, err := readFwd(t, s, key)
		if err == nil || err.Error() != unchecked || ans.rows != nil {
			t.Errorf("%s: %q, %v", name, ans.rows, err)
		}
	}
}

func TestForwardedQueryEndsWithoutAResult(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, false, []byte("n\n"))
	if _, _, err := readFwd(t, s, key); err == nil || err.Error() != fmt.Sprintf(dbfUnfinished, "shop.example.com", "uk") {
		t.Fatalf("no result: %v", err)
	}
	key2, _ := dbreads.NewReplyKey()
	s2 := &stream{t: t, pub: key2.PublicKey()}
	s2.line(dbreads.Line{Admitted: fwdID})
	s2.line(dbreads.Line{Wait: dbreads.WaitPickup})
	s2.line(dbreads.Line{Ended: dbreads.EndedNotPicked})
	if _, _, err := readFwd(t, s2, key2); err == nil || err.Error() != fmt.Sprintf(dbfNotPicked, "uk") {
		t.Fatalf("not picked: %v", err)
	}
}

func TestForwardedQueryMapsEveryCode(t *testing.T) {
	t.Parallel()
	host := "shop.example.com"
	for code, want := range map[string]string{
		dbreads.CodeTooLarge:          fmt.Sprintf(dbfTooLarge, host, "3.0GB", "2.0GB", "uk"),
		dbreads.CodeNoRoom:            fmt.Sprintf(dbfNoRoom, "uk", host),
		dbreads.CodeTimeout:           dbfTimeout,
		dbreads.CodeSQLError:          "no such table: nope",
		dbreads.CodeNothingReplicated: dbRestoreError(host, dbdoorErrNothing()).Error(),
		dbreads.CodeChangedTwice:      dbRestoreError(host, errVanished()).Error(),
		dbreads.CodeInterrupted:       fmt.Sprintf(dbfUnfinished, host, "uk"),
	} {
		key, _ := dbreads.NewReplyKey()
		s := newStream(t, key)
		sum := dbreads.Summary{Restored: 3 << 30, Cap: 2 << 30, Error: "no such table: nope"}
		s.result(dbreads.StateFailed, code, 0, sum)
		if _, _, err := readFwd(t, s, key); err == nil || err.Error() != want {
			t.Errorf("%s: %v, want %q", code, err, want)
		}
	}
	for code, want := range map[string]string{
		dbreads.CodeRecordChanged: fmt.Sprintf(dbfChanged, host),
		dbreads.CodeMoving:        fmt.Sprintf(dbfMovingCLI, host),
	} {
		key, _ := dbreads.NewReplyKey()
		s := newStream(t, key)
		s.result(dbreads.StateRefused, code, 0, dbreads.Summary{})
		if _, _, err := readFwd(t, s, key); err == nil || err.Error() != want {
			t.Errorf("%s: %v", code, err)
		}
	}
}

func TestForwardedQueryOutputCapPrintsTheRows(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, true, []byte("cut"))
	s.result(dbreads.StateDone, dbreads.CodeOutputCap, 1, summaryOf("cut"))
	ans, _, err := readFwd(t, s, key)
	if err != nil || !ans.capped || string(ans.rows) != "cut" {
		t.Fatalf("%+v %v", ans, err)
	}
}

// Review Focus 2: a full-size chunk line, about 350 KiB, is read.
func TestForwardedQueryReadsAFullSizeChunkLine(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	rows := strings.Repeat("x", dbreads.ChunkBytes)
	s := newStream(t, key)
	s.chunk(1, false, []byte(rows))
	s.chunk(2, true, nil)
	s.result(dbreads.StateDone, "", 2, summaryOf(rows))
	ans, _, err := readFwd(t, s, key)
	if err != nil || len(ans.rows) != dbreads.ChunkBytes {
		t.Fatalf("%d bytes, %v", len(ans.rows), err)
	}
}

// Review Focus 1: Ctrl-C mid-stream is "stopped", with nothing printed.
func TestForwardedQueryCtrlCPrintsNothing(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	s := newStream(t, key)
	s.chunk(1, false, []byte("n\n"))
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	var status bytes.Buffer
	_, _, err := readDBForward(ctx, &cutReader{r: &s.buf}, key, "shop.example.com", "uk", dbfReading, &status, &memSink{}, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes))
	if err == nil || err.Error() != "stopped" {
		t.Fatalf("%v", err)
	}
}

// Plan review finding 37: what the CLI holds before a result is bounded by
// count as well as bytes, and a malformed chunk is refused before it is
// kept. Each stream is otherwise one the CLI would accept, so each case is
// red under its own mutation of dbfChunkBound.admit: drop the size floor
// ("a chunk shorter than its seal" then ends DBF-12, the stream having no
// result, instead of DBF-13 at once); drop the empty-chunk rule or the
// count bound (the stream then opens and prints).
func TestForwardedQueryBoundsChunkCountAndSize(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	unchecked := fmt.Sprintf(dbfUnchecked, "uk")
	many := newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes).maxChunks + 1
	for name, write := range map[string]func(s *stream){
		"a chunk shorter than its seal": func(s *stream) {
			s.line(dbreads.Line{Chunk: 1, Data: []byte("short")})
		},
		"a chunk after an empty one": func(s *stream) {
			s.chunk(1, false, nil)
			s.chunk(2, true, []byte("n\n"))
			s.result(dbreads.StateDone, "", 2, summaryOf("n\n"))
		},
		"more chunks than a read can make": func(s *stream) {
			for seq := int64(1); seq <= many; seq++ {
				s.chunk(seq, seq == many, []byte("x"))
			}
			s.result(dbreads.StateDone, "", many, summaryOf(strings.Repeat("x", int(many))))
		},
	} {
		s := newStream(t, key)
		write(s)
		var status bytes.Buffer
		sink := &memSink{}
		id, res, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfReading, &status, sink, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes))
		if err == nil {
			_, err = openDBForward(id, key, sink, *res, "shop.example.com", "uk")
		}
		if err == nil || err.Error() != unchecked {
			t.Errorf("%s: %v", name, err)
		}
	}
}

// Plan review finding 36: SQL that JSON would rewrite is refused before
// anything is sent. Mutation: drop the check (red: the request goes out).
func TestForwardedQueryRefusesSQLThatIsNotUTF8(t *testing.T) {
	t.Parallel()
	var sent atomic.Bool
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		sent.Store(true)
		http.Error(w, "no", http.StatusTeapot)
	}))
	defer ts.Close()
	target := dbTarget{sess: &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}, acct: apiAcct{Seg: "acme"},
		app: "shop", host: "shop.example.com"}
	for _, q := range []string{"SELECT '\xff'", "SELECT '�'"} {
		err := dbQueryForward(context.Background(), target, q, dbdoor.Forward{Region: "uk", Ops: []string{dbreads.OpQuery}})
		if err == nil || err.Error() != fmt.Sprintf(dbfNotUTF8, "uk") {
			t.Errorf("%q: %v", q, err)
		}
	}
	if sent.Load() {
		t.Fatal("a query JSON would rewrite was sent")
	}
}

// cutReader ends like a connection the context closed.
type cutReader struct{ r *bytes.Buffer }

func (c *cutReader) Read(p []byte) (int, error) {
	if c.r.Len() == 0 {
		return 0, context.Canceled
	}
	return c.r.Read(p)
}

// The errors the in-region restore answers with, for the mapping test.
func dbdoorErrNothing() error { return dbdoor.ErrNothingReplicated }

func errVanished() error { return os.ErrNotExist }

// pullStream is a pull's body for the files of dir's replica, in one chunk.
func pullStream(t *testing.T, key *ecdh.PrivateKey, dir string, mutate func(s *dbreads.Summary, files [][]byte)) *stream {
	t.Helper()
	return pullStreamN(t, key, dir, 1, mutate)
}

// pullStreamN is pullStream with the files' bytes split across n chunks.
func pullStreamN(t *testing.T, key *ecdh.PrivateKey, dir string, n int, mutate func(s *dbreads.Summary, files [][]byte)) *stream {
	t.Helper()
	rd := lsfile.NewReplicaClient(dir)
	plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{})
	if err != nil {
		t.Fatal(err)
	}
	var files [][]byte
	var all bytes.Buffer
	for _, f := range plan.Planned {
		rc, err := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
		if err != nil {
			t.Fatal(err)
		}
		b, _ := io.ReadAll(rc)
		rc.Close()
		files = append(files, b)
		all.Write(b)
	}
	sum := sha256.Sum256(all.Bytes())
	s := dbreads.Summary{TXID: plan.TXID, At: plan.At, Files: plan.Planned, Bytes: int64(all.Len()), SHA256: hex.EncodeToString(sum[:])}
	if mutate != nil {
		mutate(&s, files)
	}
	st := newStream(t, key)
	b := all.Bytes()
	per := (len(b) + n - 1) / n
	for seq := 1; seq <= n; seq++ {
		part := b[min((seq-1)*per, len(b)):min(seq*per, len(b))]
		sealed, err := dbreads.SealChunk(key.PublicKey(), fwdID, int64(seq), dbreads.StreamLTX, seq == n, part)
		if err != nil {
			t.Fatal(err)
		}
		st.line(dbreads.Line{Chunk: int64(seq), Data: sealed})
	}
	st.result(dbreads.StateDone, "", int64(n), s)
	return st
}

func readPull(t *testing.T, st *stream, key *ecdh.PrivateKey) (*dbdoor.Restored, error) {
	t.Helper()
	return readPullIn(t, st, key, t.TempDir())
}

func readPullIn(t *testing.T, st *stream, key *ecdh.PrivateKey, dir string) (*dbdoor.Restored, error) {
	t.Helper()
	sink := &fileSink{dir: filepath.Join(dir, "sealed")}
	id, res, err := readDBForward(context.Background(), &st.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes))
	if err != nil {
		return nil, err
	}
	return openDBPull(context.Background(), id, key, sink, *res, "shop.example.com", "uk", dir)
}

func TestForwardedPullRestoresTheFiles(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	for _, n := range []int{1, 4} {
		key, _ := dbreads.NewReplyKey()
		out := t.TempDir()
		r, err := readPullIn(t, pullStreamN(t, key, dir, n, nil), key, out)
		if err != nil {
			t.Fatalf("%d chunks: %v", n, err)
		}
		db, _ := sql.Open("sqlite", "file:"+r.Path+"?mode=ro")
		var rows int
		err = db.QueryRow("SELECT count(*) FROM t").Scan(&rows)
		db.Close()
		if err != nil || rows != 30 {
			t.Fatalf("%d chunks: %d rows, %v", n, rows, err)
		}
		// Only the copy is left: each spooled chunk went once its bytes were
		// in the files (so the two are never both whole on the disk), and
		// the files went once the copy was restored. dbPullForward removes
		// the empty spool directory.
		if left, _ := os.ReadDir(filepath.Join(out, "sealed")); len(left) != 0 {
			t.Errorf("%d chunks: %d spooled chunks are still there", n, len(left))
		}
		if _, err := os.Stat(filepath.Join(out, "replica")); !os.IsNotExist(err) {
			t.Errorf("%d chunks: the files are still there (%v)", n, err)
		}
	}
}

// Review Focus 3 and plan review findings 22 and 24: a file list that does
// not add up is DBF-13, never a panic, and no file is created before the
// list is checked whole. Mutations: drop checkPullManifest's per-file size
// bounds (red: "sizes -1 and N+1" panics or creates files); drop its sum
// (red: "sizes that do not sum" creates the first file before failing);
// drop the duplicate check (red: "a file twice" creates a file); drop the
// level bound (red: "level 12").
func TestForwardedPullRefusesFilesThatDoNotAddUp(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	for name, mutate := range map[string]func(s *dbreads.Summary, files [][]byte){
		"sizes that do not sum":      func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Size++ },
		"a list that ends elsewhere": func(s *dbreads.Summary, _ [][]byte) { s.TXID++ },
		"no files":                   func(s *dbreads.Summary, _ [][]byte) { s.Files = nil },
		"sizes -1 and N+1": func(s *dbreads.Summary, _ [][]byte) {
			s.Files[1].Size += s.Files[0].Size + 1
			s.Files[0].Size = -1
		},
		"a file twice": func(s *dbreads.Summary, _ [][]byte) {
			s.Files[1] = s.Files[0]
			s.Bytes = 0
			for _, f := range s.Files {
				s.Bytes += f.Size
			}
		},
		"level 12":                   func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Level = 12 },
		"a range that runs back":     func(s *dbreads.Summary, _ [][]byte) { s.Files[0].MinTXID = s.Files[0].MaxTXID + 1 },
		"a size past the pull's cap": func(s *dbreads.Summary, _ [][]byte) { s.Files[0].Size = dbreads.MaxPullBytes + 1 },
	} {
		key, _ := dbreads.NewReplyKey()
		out := t.TempDir()
		func() {
			defer func() {
				if p := recover(); p != nil {
					t.Errorf("%s: panic %v", name, p)
				}
			}()
			if _, err := readPullIn(t, pullStream(t, key, dir, mutate), key, out); err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") {
				t.Errorf("%s: %v", name, err)
			}
		}()
		if _, err := os.Stat(filepath.Join(out, "replica")); !os.IsNotExist(err) {
			t.Errorf("%s: a file was created before the list was checked (%v)", name, err)
		}
	}
}

// The other endings of a pull: too-large is DBF-17 with the planned bytes
// and the cap, no-room DBF-24, and the rest the query's mapping.
func TestForwardedPullEndings(t *testing.T) {
	t.Parallel()
	for name, c := range map[string]struct {
		code string
		sum  dbreads.Summary
		want string
	}{
		"too large": {dbreads.CodeTooLarge, dbreads.Summary{Restored: 3 << 30, Cap: 2 << 30},
			fmt.Sprintf(dbfPullTooLarge, "shop.example.com", humanBytes(ptr(int64(3<<30))), humanBytes(ptr(int64(2<<30))), "uk")},
		"no room":            {dbreads.CodeNoRoom, dbreads.Summary{}, fmt.Sprintf(dbfPullNoRoom, "uk", "shop.example.com")},
		"nothing replicated": {dbreads.CodeNothingReplicated, dbreads.Summary{}, dbRestoreError("shop.example.com", dbdoor.ErrNothingReplicated).Error()},
	} {
		key, _ := dbreads.NewReplyKey()
		st := newStream(t, key)
		st.result(dbreads.StateFailed, c.code, 0, c.sum)
		if _, err := readPull(t, st, key); err == nil || err.Error() != c.want {
			t.Errorf("%s: %v, want %q", name, err, c.want)
		}
	}
}

func ptr[T any](v T) *T { return &v }

// Plan review finding 37, at the spool: a chunk shorter than a sealed
// empty one creates no spool file, and a stream of tiny chunks stops at the
// bound's count, so the spool never holds more files than a read can make.
// (TestForwardedQueryBoundsChunkCountAndSize pins the count's derivation; a
// bound of 10 keeps this test small.) Mutation: add to the sink before
// admit (red: a file for the short chunk, eleven files for the tiny ones).
func TestForwardedPullSpoolsNoMalformedOrExcessChunk(t *testing.T) {
	t.Parallel()
	key, _ := dbreads.NewReplyKey()
	for name, write := range map[string]func(s *stream) int{
		"a chunk shorter than its seal": func(s *stream) int {
			s.line(dbreads.Line{Chunk: 1, Data: []byte("short")})
			return 0
		},
		"tiny chunks past the count": func(s *stream) int {
			for seq := int64(1); seq <= 11; seq++ {
				sealed, _ := dbreads.SealChunk(key.PublicKey(), fwdID, seq, dbreads.StreamLTX, false, []byte("x"))
				s.line(dbreads.Line{Chunk: seq, Data: sealed})
			}
			return 10
		},
	} {
		s := newStream(t, key)
		want := write(s)
		dir := filepath.Join(t.TempDir(), "sealed")
		sink := &fileSink{dir: dir}
		_, _, err := readDBForward(context.Background(), &s.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink,
			&dbfChunkBound{maxBytes: dbreads.MaxPullBytes, maxChunks: 10})
		files, _ := os.ReadDir(dir)
		if err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") || len(files) != want {
			t.Errorf("%s: %v, %d spool files, want %d", name, err, len(files), want)
		}
	}
}

// standInConsole answers the list door with the forward marker offering a
// pull, and the reads route with body (then holds the request open until
// the client goes when hold is set). It is a stand-in for the console, so
// that these tests run dbPull itself.
func standInConsole(t *testing.T, body []byte, hold bool) dbTarget {
	t.Helper()
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !strings.HasSuffix(r.URL.Path, "/reads") {
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusConflict)
			json.NewEncoder(w).Encode(dbdoor.ForwardBody{Error: "shop.example.com keeps its data in region uk.",
				Forward: 1, Region: "uk", ForwardOps: []string{dbreads.OpQuery, dbreads.OpPull}})
			return
		}
		io.Copy(io.Discard, r.Body)
		w.Header().Set("Content-Type", "application/x-ndjson")
		w.WriteHeader(http.StatusOK)
		w.Write(body)
		w.(http.Flusher).Flush()
		if hold {
			<-r.Context().Done()
		}
	}))
	t.Cleanup(ts.Close)
	return dbTarget{sess: &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}, acct: apiAcct{Seg: "acme"},
		app: "shop", host: "shop.example.com"}
}

// leftBehind is what a pull left in its temp dir and at --out.
func leftBehind(t *testing.T, tmp, out string) {
	t.Helper()
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Errorf("something is at --out: %v", err)
	}
	if left, _ := filepath.Glob(filepath.Join(tmp, "carlos-db-*")); len(left) != 0 {
		t.Errorf("the pull's directory is still there: %v", left)
	}
}

// Review Focus 1 and plan review finding 23: a spool that fails part way,
// through dbPull itself, leaves nothing at --out and removes its directory.
// Not parallel: it sets TMPDIR and dbPullSpoolWrite. Mutation: drop
// dbPullForward's deferred RemoveAll (red: carlos-db-* is left).
func TestForwardedPullLeavesNothingWhenTheSpoolFails(t *testing.T) {
	tmp := t.TempDir()
	t.Setenv("TMPDIR", tmp)
	writes := 0
	dbPullSpoolWrite = func(name string, b []byte, mode os.FileMode) error {
		if writes++; writes == 3 {
			return errors.New("no space left on device")
		}
		return os.WriteFile(name, b, mode)
	}
	t.Cleanup(func() { dbPullSpoolWrite = os.WriteFile })
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	// The stand-in's chunks are sealed to a key the CLI does not hold; the
	// pull fails at the spool, before anything is opened.
	key, _ := dbreads.NewReplyKey()
	st := pullStreamN(t, key, dir, 4, nil)
	out := filepath.Join(t.TempDir(), "copy.db")
	if err := dbPull(context.Background(), standInConsole(t, st.buf.Bytes(), false), out); err == nil {
		t.Fatal("a spool that failed was accepted")
	}
	if writes != 3 {
		t.Fatalf("%d spool writes: the failure was not part way", writes)
	}
	leftBehind(t, tmp, out)
}

// Review Focus 4 and plan review finding 23: Ctrl-C while chunks are
// spooling, through dbPull itself, is "stopped", with nothing at --out and
// the directory removed. Not parallel: it sets TMPDIR and dbPullSpoolWrite.
// Mutation: drop dbPullForward's deferred RemoveAll (red). The binary's
// SIGINT is Task 9's TestDBForwardPullInterruptedLeavesNothing.
func TestForwardedPullCtrlCLeavesNothing(t *testing.T) {
	tmp := t.TempDir()
	t.Setenv("TMPDIR", tmp)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	dbPullSpoolWrite = func(name string, b []byte, mode os.FileMode) error {
		err := os.WriteFile(name, b, mode)
		cancel() // Ctrl-C once the first chunk is on disk
		return err
	}
	t.Cleanup(func() { dbPullSpoolWrite = os.WriteFile })
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	key, _ := dbreads.NewReplyKey()
	st := pullStreamN(t, key, dir, 4, nil)
	// Admitted, picked and the first chunk only (newStream writes admitted
	// and picked first), then the stand-in holds the request.
	lines := bytes.SplitAfter(st.buf.Bytes(), []byte("\n"))
	out := filepath.Join(t.TempDir(), "copy.db")
	err := dbPull(ctx, standInConsole(t, bytes.Join(lines[:3], nil), true), out)
	if err == nil || err.Error() != "stopped" {
		t.Fatalf("%v", err)
	}
	leftBehind(t, tmp, out)
}

// pullStreamOf is a pull's body for the given bytes and summary, one chunk.
func pullStreamOf(t *testing.T, key *ecdh.PrivateKey, b []byte, s dbreads.Summary) *stream {
	t.Helper()
	st := newStream(t, key)
	sealed, err := dbreads.SealChunk(key.PublicKey(), fwdID, 1, dbreads.StreamLTX, true, b)
	if err != nil {
		t.Fatal(err)
	}
	st.line(dbreads.Line{Chunk: 1, Data: sealed})
	st.result(dbreads.StateDone, "", 1, s)
	return st
}

// Astra round 1 on plan 3, finding 1, at the CLI: a valid snapshot of
// transaction 1, listed as [1,2] with a summary at 2 that adds up in every
// other way, is DBF-13 and restores nothing: the CLI checks each rebuilt
// file's LTX header against its manifest entry. Mutation: drop the header
// comparison (red: it restores and reports transaction 2).
func TestForwardedPullRefusesARelabelledSnapshot(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 2)
	rc, err := lsfile.NewReplicaClient(dir).OpenLTXFile(context.Background(), 0, 1, 1, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	b, _ := io.ReadAll(rc)
	rc.Close()
	sum := sha256.Sum256(b)
	s := dbreads.Summary{TXID: 2, Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 2, Size: int64(len(b))}},
		Bytes: int64(len(b)), SHA256: hex.EncodeToString(sum[:])}
	key, _ := dbreads.NewReplyKey()
	out := t.TempDir()
	r, err := readPullIn(t, pullStreamOf(t, key, b, s), key, out)
	if err == nil || err.Error() != fmt.Sprintf(dbfUnchecked, "uk") {
		t.Fatalf("a relabelled snapshot gave %+v, %v", r, err)
	}
	if _, err := os.Stat(filepath.Join(out, "replica.db")); !os.IsNotExist(err) {
		t.Fatalf("a copy was restored: %v", err)
	}
}

// Astra round 1 on plan 3, finding 2, after the download: a pull cancelled
// once every chunk has arrived is "stopped" and rebuilds no file. Mutation:
// drop the rebuild loop's context check (red: the files are written and
// restored before the cancel is seen).
func TestForwardedPullStopsWhenCancelledAfterTheDownload(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	key, _ := dbreads.NewReplyKey()
	st := pullStreamN(t, key, dir, 4, nil)
	out := t.TempDir()
	sink := &fileSink{dir: filepath.Join(out, "sealed")}
	id, res, err := readDBForward(context.Background(), &st.buf, key, "shop.example.com", "uk", dbfFetching, io.Discard, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if _, err := openDBPull(ctx, id, key, sink, *res, "shop.example.com", "uk", out); err == nil || err.Error() != "stopped" {
		t.Fatalf("a cancelled pull answered %v", err)
	}
	for _, gone := range []string{"replica", "replica.db"} {
		if _, err := os.Stat(filepath.Join(out, gone)); !os.IsNotExist(err) {
			t.Errorf("%s was written after the cancel (%v)", gone, err)
		}
	}
}