// Acceptance for the forwarded `carlos db query` through the real CLI
// binary and the real console server: T6 (parity with the in-region door,
// through the fake S3 the db door's own e2e uses), T7 (the long request
// through the real edge and the console's real http.Server), T8 (mixed
// versions) and T11 (nothing printed for an answer that does not check out).
// The box is plan 1's real agent and worker code joined in process
// (dbreadstest): not acceptance for the units, which plan 1's root-runner
// tests are. A scripted box (e2eScript) is a stand-in and says so.
package main

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

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

	"github.com/carlosframework/platform/internal/console"
	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/dbreads/dbreadstest"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

const (
	dbfE2EHere = "shop.example.com" // in the console's region, ie
	dbfE2EFar  = "far.example.com"  // in uk, with the same database bytes
)

type dbfE2EOptions struct {
	forward bool
	// serve serves the console's handler and returns the URL the CLI uses
	// and any extra environment it needs; nil is the httptest server.
	serve func(t *testing.T, h http.Handler) (consoleURL string, env []string)
	// worker is the in-process box's worker options (a slow restore for T7).
	worker dbreads.WorkerOptions
	// noBox leaves the uk box out; the test scripts its own.
	noBox bool
}

type dbfE2E struct {
	bin, creds, url, tok, root string
	env                        []string
	store                      release.Store
	pub                        ed25519.PublicKey
	priv                       ed25519.PrivateKey
}

// newDBFE2E is a console in ie over a DirStore and the fake S3: shop in ie
// and far in uk with byte-identical replicas (the same files with the same
// modification times, so the fake S3 lists the same LastModified for both),
// a capable live uk box, and a signed-in CLI.
func newDBFE2E(t *testing.T, o dbfE2EOptions) *dbfE2E {
	t.Helper()
	creds := useTempCredentials(t)
	clearDeploymentEnv(t)
	tmp := t.TempDir()
	e := &dbfE2E{bin: buildCarlos(t), creds: creds, store: release.NewDirStore(filepath.Join(tmp, "store")), root: filepath.Join(tmp, "data")}
	var err error
	if e.pub, e.priv, err = ed25519.GenerateKey(rand.Reader); err != nil {
		t.Fatal(err)
	}
	here := filepath.Join(e.root, "instances", dbfE2EHere)
	dbfE2EFixture(t, here)
	copyTreeKeepingTimes(t, here, filepath.Join(e.root, "instances", dbfE2EFar))
	s3 := httptest.NewServer(&replicaS3{t: t, bucket: "data", root: e.root})
	t.Cleanup(s3.Close)
	t.Setenv("CARLOS_S3_ENDPOINT", s3.URL)
	t.Setenv("AWS_ACCESS_KEY_ID", "test")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")

	cst, err := console.Open(filepath.Join(tmp, "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { cst.Close() })
	var srv *console.Server
	h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { srv.ServeHTTP(w, r) })
	direct := httptest.NewServer(h)
	t.Cleanup(direct.Close)
	e.url = direct.URL
	if o.serve != nil {
		e.url, e.env = o.serve(t, h)
	}
	srv, err = console.NewServer(console.Config{
		Origin: direct.URL, Keymail: "http://keymail.invalid", Registry: filepath.Join(tmp, "registry.db"),
		Releases: e.store, ReleasePub: e.pub, Key: e.priv, AppScheme: "http",
		Region: "ie", HibernateBucket: "data", HibernateRegion: "us-east-1", HibernatePrefix: "instances",
		DBForward: o.forward,
	}, cst)
	if err != nil {
		t.Fatal(err)
	}
	const addr = "member@example.com"
	if err := cst.AddMember(console.OpsAccountID, addr, console.RoleOwner, time.Now()); err != nil {
		t.Fatal(err)
	}
	if _, err := cst.CreateAppIn(console.OpsAccountID, "shop", addr, time.Now()); err != nil {
		t.Fatal(err)
	}
	created := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
	for host, region := range map[string]string{dbfE2EHere: "ie", dbfE2EFar: "uk"} {
		yes := true
		body, _ := json.Marshal(provision.Record{Host: host, App: "shop", Account: release.OpsAccount, Region: region, Hibernate: &yes, CreatedAt: created})
		if _, err := e.store.Put(context.Background(), provision.RecordKey(release.OpsAccount, "shop", host), body, ""); err != nil {
			t.Fatal(err)
		}
	}
	edges, _ := json.Marshal(map[string]any{"v": 1, "box": "uk-1", "eip": "203.0.113.10", "placement": "uk",
		"capabilities": []string{dbreads.Capability}, "hosts": map[string]any{}})
	e.store.Put(context.Background(), "control/edges/uk-1.json", edges, "")
	status, _ := json.Marshal(map[string]any{"generated_at": time.Now().UTC(), "label": "uk-1"})
	e.store.Put(context.Background(), "status/boxes/uk-1.json", status, "")
	token, expiresAt := e2eDeviceLogin(t, direct.URL, cst, addr, "e2e@laptop")
	e.tok = token
	cf, err := loadCredentials()
	if err != nil {
		t.Fatal(err)
	}
	cf.put(credential{Console: e.url, Address: addr, Token: token, ExpiresAt: expiresAt})
	if err := cf.save(); err != nil {
		t.Fatal(err)
	}
	if !o.noBox {
		box := dbreadstest.Start(t, dbreadstest.Options{Store: e.store, Region: "uk", Box: "uk-1", Pub: e.pub,
			HibernatePrefix: "instances", Worker: o.worker,
			OpenReplica: func(ctx context.Context, prefix string) (dbdoor.ReplicaReader, error) {
				return lease.ReadReplicaClient(ctx, "data", prefix, "us-east-1")
			}})
		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(50 * time.Millisecond)
			}
		}()
		t.Cleanup(func() { cancel(); <-done; box.Agent.Wait() })
	}
	return e
}

// dbfE2EFixture is T6's database (NULL, a tab and a newline in one value, a
// blob, a time) in two syncs a whole second apart.
func dbfE2EFixture(t *testing.T, dir string) {
	t.Helper()
	src := filepath.Join(t.TempDir(), "shop.db")
	db, err := sql.Open("sqlite", src)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	for _, q := range []string{
		`PRAGMA journal_mode=WAL`,
		`CREATE TABLE orders (id INTEGER PRIMARY KEY, note TEXT, data BLOB, at DATETIME)`,
	} {
		if _, err := db.Exec(q); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`INSERT INTO orders (note, data, at) VALUES (?, ?, ?), (NULL, NULL, NULL)`,
		"a\ttab\nline", []byte{0, 0xff}, time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC)); err != nil {
		t.Fatal(err)
	}
	dbdoortest.Sync(t, src, dir)
	time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(1100 * time.Millisecond)))
	if _, err := db.Exec(`INSERT INTO orders (note) VALUES ('later')`); err != nil {
		t.Fatal(err)
	}
	dbdoortest.Sync(t, src, dir)
}

func copyTreeKeepingTimes(t *testing.T, from, to string) {
	t.Helper()
	err := filepath.Walk(from, func(p string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}
		rel, _ := filepath.Rel(from, p)
		dst := filepath.Join(to, rel)
		if info.IsDir() {
			return os.MkdirAll(dst, 0o755)
		}
		b, err := os.ReadFile(p)
		if err != nil {
			return err
		}
		if err := os.WriteFile(dst, b, 0o644); err != nil {
			return err
		}
		return os.Chtimes(dst, info.ModTime(), info.ModTime())
	})
	if err != nil {
		t.Fatal(err)
	}
}

// run runs the real CLI. Callers name the account (--account ops), as the
// db door's own e2e does: a defaulted account prints its note on stdout,
// which is main's behaviour for every db verb and not this plan's to change.
func (e *dbfE2E) run(t *testing.T, args ...string) (string, string, error) {
	t.Helper()
	cmd := exec.Command(e.bin, args...)
	cmd.Env = carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...)
	cmd.Dir = t.TempDir()
	var stdout, stderr bytes.Buffer
	cmd.Stdout, cmd.Stderr = &stdout, &stderr
	err := cmd.Run()
	return stdout.String(), stderr.String(), err
}

// betweenTheSyncs is an --at that holds the first sync and not the second:
// half a second into the second after the first L0 file was written. The
// second sync is at least 1.1 s after the first (dbfE2EFixture), so its
// listed second is never before this time's.
func (e *dbfE2E) betweenTheSyncs(t *testing.T) string {
	t.Helper()
	ents, err := os.ReadDir(filepath.Join(e.root, "instances", dbfE2EHere, "ltx", "0"))
	if err != nil || len(ents) < 2 {
		t.Fatalf("the fixture has %d L0 files: %v", len(ents), err)
	}
	var first time.Time
	for _, ent := range ents {
		info, err := ent.Info()
		if err != nil {
			t.Fatal(err)
		}
		if first.IsZero() || info.ModTime().Before(first) {
			first = info.ModTime()
		}
	}
	return first.Truncate(time.Second).Add(1500 * time.Millisecond).UTC().Format(time.RFC3339Nano)
}

// doer is an old CLI's view of the db door: GETs with the saved token.
func (e *dbfE2E) doer(host string) dbdoor.Doer {
	return func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
		req, err := http.NewRequestWithContext(ctx, http.MethodGet, e.url+dbdoor.Base(release.OpsAccount, "shop", host)+tail, nil)
		if err != nil {
			return nil, err
		}
		for k, vs := range h {
			req.Header[k] = vs
		}
		req.Header.Set("Authorization", "Bearer "+e.tok)
		return http.DefaultClient.Do(req)
	}
}

// e2eScript writes one read's objects itself: a stand-in for a box, sealing
// real chunks and a real summary to the request's reply key.
type e2eScript struct {
	store release.Store
	id    string
	pub   *ecdh.PublicKey
}

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

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

func (b *e2eScript) 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.store.Put(context.Background(), dbreads.ResultKey(b.id), r, "")
}

// scriptBox runs script once for each new request in uk's queue, after
// verifying it as a box would.
func (e *dbfE2E) scriptBox(t *testing.T, script func(*e2eScript)) func() {
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		seen := map[string]bool{}
		for ctx.Err() == nil {
			keys, _ := e.store.List(ctx, dbreads.QueueRegionPrefix("uk"))
			for _, k := range keys {
				q, ok := dbreads.ParseQueueKey(k)
				if !ok || seen[q.ID] {
					continue
				}
				seen[q.ID] = true
				raw, _, err := e.store.Get(ctx, dbreads.RequestKey(q.ID))
				if err != nil {
					continue
				}
				req, err := dbreads.Verify(e.pub, raw, q.ID, "uk", time.Now())
				if err != nil {
					t.Errorf("the scripted box refused %s: %v", q.ID, err)
					continue
				}
				pub, _ := dbreads.ParseReplyKey(req.ReplyKey)
				go script(&e2eScript{store: e.store, id: q.ID, pub: pub})
			}
			time.Sleep(50 * time.Millisecond)
		}
	}()
	return func() { cancel(); <-done }
}

// throughTheEdge serves the console's handler with console.NewHTTPServer on
// 127.0.0.1, its ReadTimeout set to readTimeout, behind a real edge route
// for console.test. The CLI reaches it by that name through HTTP_PROXY,
// which stands in for DNS.
func throughTheEdge(readTimeout time.Duration, edgeURL *string, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) {
	return func(t *testing.T, h http.Handler) (string, []string) {
		ln, err := net.Listen("tcp", "127.0.0.1:0")
		if err != nil {
			t.Fatal(err)
		}
		hs := console.NewHTTPServer(h)
		hs.ReadTimeout = readTimeout
		go hs.Serve(ln)
		t.Cleanup(func() { hs.Close() })
		reg, err := registry.Open(filepath.Join(t.TempDir(), "edge.db"))
		if err != nil {
			t.Fatal(err)
		}
		if err := reg.Upsert(registry.Route{Host: "console.test", Kind: "service", Addr: ln.Addr().String(), Status: "ready"}); err != nil {
			t.Fatal(err)
		}
		ed, err := edge.New(reg, nil)
		if err != nil {
			t.Fatal(err)
		}
		if err := ed.Reload(); err != nil {
			t.Fatal(err)
		}
		es := httptest.NewServer(ed)
		t.Cleanup(es.Close)
		*edgeURL = es.URL
		// The CLI's proxy is a tap in front of the edge, so the test sees
		// what reaches the CLI's own socket (plan review finding 34).
		tp := newDBFTap(t, es.Listener.Addr().String(), false)
		*tap = tp
		return "http://console.test", []string{"HTTP_PROXY=http://" + tp.addr, "NO_PROXY="}
	}
}

// dbfTap stands between the CLI and the first edge, forwarding bytes both
// ways and noting when each read from the edge's side was passed to the
// CLI's socket: what the CLI process received, and when. With connect it is
// an HTTPS proxy (CONNECT), which is how the CLI reaches a TLS edge by name
// here; without, a plain forwarder for HTTP_PROXY.
type dbfTap struct {
	addr string
	mu   sync.Mutex
	down []time.Time
}

func newDBFTap(t *testing.T, target string, connect bool) *dbfTap {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { ln.Close() })
	tp := &dbfTap{addr: ln.Addr().String()}
	go func() {
		for {
			c, err := ln.Accept()
			if err != nil {
				return
			}
			go tp.serve(c, target, connect)
		}
	}()
	return tp
}

func (tp *dbfTap) serve(c net.Conn, target string, connect bool) {
	defer c.Close()
	var in io.Reader = c
	if connect {
		br := bufio.NewReader(c)
		req, err := http.ReadRequest(br)
		if err != nil || req.Method != http.MethodConnect {
			return
		}
		if _, err := io.WriteString(c, "HTTP/1.1 200 Connection established\r\n\r\n"); err != nil {
			return
		}
		in = br
	}
	up, err := net.Dial("tcp", target)
	if err != nil {
		return
	}
	defer up.Close()
	go func() {
		io.Copy(up, in)
		up.(*net.TCPConn).CloseWrite()
	}()
	buf := make([]byte, 32<<10)
	for {
		n, err := up.Read(buf)
		if n > 0 {
			tp.mu.Lock()
			tp.down = append(tp.down, time.Now())
			tp.mu.Unlock()
			if _, werr := c.Write(buf[:n]); werr != nil {
				return
			}
		}
		if err != nil {
			return
		}
	}
}

// longestGap is the longest silence at the CLI's socket between start and
// end, both of which count as arrivals.
func (tp *dbfTap) longestGap(start, end time.Time) time.Duration {
	tp.mu.Lock()
	defer tp.mu.Unlock()
	prev, longest := start, time.Duration(0)
	for _, at := range append(append([]time.Time(nil), tp.down...), end) {
		if at.Before(start) || at.After(end) {
			continue
		}
		if d := at.Sub(prev); d > longest {
			longest = d
		}
		prev = at
	}
	return longest
}

// checkProgressAtTheCLI is T7's witness at the CLI process itself (plan
// review finding 34): it runs the real CLI and needs DBF-11 on its stderr
// within a second of the process starting (stricter than the spec's "of the
// request": the process starts before it sends anything), no silence over
// 16 s (15 s and a poll) at its socket, the read lasting at least atLeast, and
// its rows. atLeast is the quiet restore's length.
func checkProgressAtTheCLI(t *testing.T, e *dbfE2E, tap *dbfTap, atLeast time.Duration) {
	t.Helper()
	cmd := exec.Command(e.bin, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders")
	cmd.Env = carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...)
	cmd.Dir = t.TempDir()
	var out bytes.Buffer
	cmd.Stdout = &out
	pipe, err := cmd.StderrPipe()
	if err != nil {
		t.Fatal(err)
	}
	start := time.Now()
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	var stderr []string
	var reading time.Time
	sc := bufio.NewScanner(pipe)
	for sc.Scan() {
		if reading.IsZero() && strings.Contains(sc.Text(), fmt.Sprintf(dbfReading, dbfE2EFar, "uk")) {
			reading = time.Now()
		}
		stderr = append(stderr, sc.Text())
	}
	err = cmd.Wait()
	end := time.Now()
	if err != nil || out.String() != "n\n3\n" {
		t.Fatalf("%q %v\n%s", out.String(), err, strings.Join(stderr, "\n"))
	}
	if reading.IsZero() || reading.Sub(start) > time.Second {
		t.Fatalf("DBF-11 after %s: %q", reading.Sub(start), stderr)
	}
	if gap := tap.longestGap(start, end); gap > 16*time.Second {
		t.Fatalf("%s of silence at the CLI's socket", gap)
	}
	if took := end.Sub(start); took < atLeast {
		t.Fatalf("the read took %s, under %s: it did not hold the request", took, atLeast)
	}
}

// slowRestore holds the box's restore for d before restoring for real.
func slowRestore(d time.Duration) dbreads.WorkerOptions {
	return dbreads.WorkerOptions{Restore: func(ctx context.Context, c litestream.ReplicaClient, out string, target uint64) error {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(d):
		}
		return dbdoor.RestoreLocal(ctx, c, out, target)
	}}
}

// T6: the forwarded query and the in-region query print byte-identical
// stdout, for the newest copy and as of a time between the two syncs.
func TestDBForwardParityThroughTheCLI(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true})
	const q = "SELECT id, note, data, at FROM orders ORDER BY id"
	for _, at := range []string{"", e.betweenTheSyncs(t)} {
		args := []string{"db", "query", "--account", "ops", "--app", "shop", "--sql", q}
		if at != "" {
			args = append(args, "--at", at)
		}
		here, hereErr, err := e.run(t, append(args, "--instance", dbfE2EHere)...)
		if err != nil {
			t.Fatalf("in region: %v\n%s", err, hereErr)
		}
		far, farErr, err := e.run(t, append(args, "--instance", dbfE2EFar)...)
		if err != nil {
			t.Fatalf("forwarded: %v\n%s", err, farErr)
		}
		if here != far || here == "" {
			t.Fatalf("at %q: in region %q, forwarded %q", at, here, far)
		}
		if !strings.Contains(farErr, fmt.Sprintf(dbfReading, dbfE2EFar, "uk")) {
			t.Fatalf("no DBF-11 on stderr: %s", farErr)
		}
		if at != "" && strings.Contains(far, "later") {
			t.Fatalf("as of %s the copy holds the second sync: %q", at, far)
		}
	}
}

// T7, scaled: the console's real http.Server with ReadTimeout 2 s instead of
// 5 minutes, behind the real edge, and a read held 20 s by a slow restore.
// The connection outlives ReadTimeout tenfold, admitted arrives within a
// second, a line arrives at least every 15 s (plus a poll), and the real CLI
// gets its rows through the same path.
func TestDBForwardLongRequestThroughTheEdge(t *testing.T) {
	var edgeURL string
	var tap *dbfTap
	e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughTheEdge(2*time.Second, &edgeURL, &tap), worker: slowRestore(20 * time.Second)})
	key, _ := dbreads.NewReplyKey()
	body, _ := json.Marshal(dbreads.ReadBody{Op: dbreads.OpQuery, SQL: "SELECT count(*) AS n FROM orders", ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	req, _ := http.NewRequest(http.MethodPost, edgeURL+"/api/cli/apps/"+release.OpsAccount+"/shop/db/"+dbfE2EFar+"/reads", bytes.NewReader(body))
	req.Host = "console.test"
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+e.tok)
	start := time.Now()
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer res.Body.Close()
	br := bufio.NewReader(res.Body)
	prev, first, sawResult := start, true, false
	for {
		line, err := br.ReadBytes('\n')
		if err == io.EOF {
			break
		}
		if err != nil {
			t.Fatal(err)
		}
		gap := time.Since(prev)
		if first && gap > time.Second {
			t.Fatalf("admitted took %s", gap)
		}
		if gap > 16*time.Second {
			t.Fatalf("%s with no line", gap)
		}
		prev, first = time.Now(), false
		sawResult = sawResult || bytes.Contains(line, []byte(`"result"`))
	}
	if !sawResult || time.Since(start) < 20*time.Second {
		t.Fatalf("result %v after %s", sawResult, time.Since(start))
	}
	// The same through the real CLI process, observed at its own stderr and
	// socket (plan review finding 34).
	checkProgressAtTheCLI(t, e, tap, 20*time.Second)
}

// T7, unscaled: a read that holds the request for 21 minutes through the
// real edge and the console's unmodified http.Server (5-minute ReadTimeout),
// inside the CLI's 23-minute context. The box is scripted (a stand-in):
// picked at 1 minute 50 seconds with a 20-minute deadline, one chunk at 10
// minutes, the last and the result at 21. It waits; it does not compute.
// Run once before handoff with CARLOS_DBF_LONG=1.
func TestDBForwardTwentyOneMinuteRead(t *testing.T) {
	if os.Getenv("CARLOS_DBF_LONG") != "1" {
		t.Skip("set CARLOS_DBF_LONG=1 to run the unscaled 21-minute read, once, before handoff")
	}
	var edgeURL string
	var tap *dbfTap
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: throughTheEdge(5*time.Minute, &edgeURL, &tap)})
	stop := e.scriptBox(t, func(b *e2eScript) {
		time.Sleep(110 * time.Second)
		b.picked(time.Now().Add(20 * time.Minute))
		time.Sleep(8 * time.Minute)
		b.chunk(1, false, "n\n")
		time.Sleep(11*time.Minute + 10*time.Second)
		b.chunk(2, true, "3\n")
		b.result(2, "n\n3\n")
	})
	defer stop()
	start := time.Now()
	out, stderr, err := e.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders")
	if err != nil || out != "n\n3\n" || time.Since(start) < 21*time.Minute {
		t.Fatalf("after %s: %q %v\n%s", time.Since(start), out, err, stderr)
	}
}

// T8: the rows of spec §9.2 the CLI sees.
func TestDBForwardMixedVersions(t *testing.T) {
	off := newDBFE2E(t, dbfE2EOptions{forward: false, noBox: true})
	// A new CLI and a console with forwarding off (or an old console): the
	// residency sentence as it is, and nothing on stdout.
	out, stderr, err := off.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT 1")
	if err == nil || out != "" || !strings.Contains(stderr, "far.example.com keeps its data in region uk. This console can only read data kept in region ie.") {
		t.Fatalf("new CLI, forwarding off: %q %q %v", out, stderr, err)
	}
	on := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	// An old CLI and a console with forwarding on: every CLI from before
	// this plan runs dbRestoreError on a door refusal, which prints the
	// door's sentence. That sentence is DBF-14.
	_, derr := dbdoor.NewClient(on.doer(dbfE2EFar)).LTXFiles(context.Background(), 9, 0, false)
	want := "far.example.com keeps its data in region uk. this version of carlos cannot read it from there. run carlos update, then try again."
	if got := dbRestoreError(dbfE2EFar, derr).Error(); got != want {
		t.Fatalf("old CLI, new console: %q", got)
	}
}

// T11 (codex 6): the real CLI prints nothing for a dropped last chunk, a
// summary hash mismatch, or a response that ends without a result. The box
// is scripted: a stand-in that writes picked, chunks and result itself.
func TestDBForwardPrintsNothingForABadAnswer(t *testing.T) {
	for name, c := range map[string]struct {
		script func(b *e2eScript)
		want   string
	}{
		// The bucket lost the real last chunk: the result names chunk 1,
		// which was sealed as not the last.
		"the last chunk dropped": {func(b *e2eScript) {
			b.picked(time.Now().Add(time.Minute))
			b.chunk(1, false, "n\n")
			b.result(1, "n\n")
		}, "was invalid, so carlos db printed nothing"},
		"a summary hash mismatch": {func(b *e2eScript) {
			b.picked(time.Now().Add(time.Minute))
			b.chunk(1, true, "n\n31\n")
			b.result(1, "n\n30\n")
		}, "was invalid, so carlos db printed nothing"},
		// The console waits the owner's deadline plus its 1-minute grace.
		"no result": {func(b *e2eScript) {
			b.picked(time.Now().Add(2 * time.Second))
			b.chunk(1, false, "n\n")
		}, "did not finish, so nothing was printed"},
	} {
		t.Run(name, func(t *testing.T) {
			e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
			stop := e.scriptBox(t, c.script)
			defer stop()
			out, stderr, err := e.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT count(*) AS n FROM orders")
			if err == nil || out != "" || !strings.Contains(stderr, c.want) {
				t.Fatalf("%q %q %v", out, stderr, err)
			}
		})
	}
}

// Plan review finding 36, through the binary: a query holding a byte that
// is not UTF-8 is refused with DBF-32 before anything is sent, so no altered
// query reaches the bucket. (An argv may carry any bytes on Linux, where
// this runs.)
func TestDBForwardRefusesSQLThatIsNotUTF8(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	out, stderr, err := e.run(t, "db", "query", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--sql", "SELECT 'caf\xe9'")
	if err == nil || out != "" || !strings.Contains(stderr, fmt.Sprintf(dbfNotUTF8, "uk")) {
		t.Fatalf("%q %q %v", out, stderr, err)
	}
	if keys, _ := e.store.List(context.Background(), dbreads.Prefix); len(keys) != 0 {
		t.Fatalf("a query JSON would rewrite reached the bucket: %v", keys)
	}
}

// throughBackhaul serves the console's handler with console.NewHTTPServer
// (its ReadTimeout set to readTimeout) on 127.0.0.1, behind two real edges
// that package edge's test binary runs as a helper process: the CLI reaches
// betty, which backhauls to annie, the console's home. The CLI finds betty
// through a CONNECT proxy (a dbfTap, standing in for DNS) and trusts the
// edges' test CA through SSL_CERT_FILE. Building the helper needs the go
// command on PATH, as every CI step has.
func throughBackhaul(readTimeout time.Duration, tap **dbfTap) func(t *testing.T, h http.Handler) (string, []string) {
	return func(t *testing.T, h http.Handler) (string, []string) {
		ln, err := net.Listen("tcp", "127.0.0.1:0")
		if err != nil {
			t.Fatal(err)
		}
		hs := console.NewHTTPServer(h)
		hs.ReadTimeout = readTimeout
		go hs.Serve(ln)
		t.Cleanup(func() { hs.Close() })
		dir := t.TempDir()
		bin := filepath.Join(dir, "edge.test")
		// From the module root with the environment TestMain kept for the
		// carlos build (the real HOME and Go cache), as sharedCarlos builds.
		build := exec.Command("go", "test", "-c", "-o", bin, "./internal/edge")
		build.Dir, build.Env = moduleRoot, carlosBuildEnv
		if out, err := build.CombinedOutput(); err != nil {
			t.Fatalf("build the edge test binary: %v\n%s", err, out)
		}
		cmd := exec.Command(bin, "-test.run", "^TestDBForwardBackhaulPairHelper$", "-test.v")
		cmd.Env = append(os.Environ(), "DBF_BACKHAUL_HELPER=1", "DBF_BACKHAUL_UPSTREAM="+ln.Addr().String(), "DBF_BACKHAUL_DIR="+dir)
		stdin, err := cmd.StdinPipe()
		if err != nil {
			t.Fatal(err)
		}
		pr, pw := io.Pipe()
		cmd.Stdout = pw
		var helperErr bytes.Buffer
		cmd.Stderr = &helperErr
		if err := cmd.Start(); err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() {
			stdin.Close()
			cmd.Wait()
			pw.Close()
			if t.Failed() {
				t.Logf("backhaul helper stderr:\n%s", helperErr.String())
			}
		})
		lines := bufio.NewScanner(pr)
		sender := ""
		for lines.Scan() {
			if addr, ok := strings.CutPrefix(lines.Text(), "DBF-BACKHAUL "); ok {
				sender = addr
				break
			}
		}
		if sender == "" {
			t.Fatalf("the backhaul helper never came up:\n%s", helperErr.String())
		}
		go func() {
			for lines.Scan() {
			}
		}()
		tp := newDBFTap(t, sender, true)
		*tap = tp
		return "https://console.backhaul.test", []string{"HTTPS_PROXY=http://" + tp.addr, "NO_PROXY=", "SSL_CERT_FILE=" + filepath.Join(dir, "ca.pem")}
	}
}

// T7, the backhaul half, end to end (plan review finding 34): the real CLI
// binary reaches the real console through two real edges, the second hop a
// backhaul with its 30 s header bound, during a 40 s quiet restore. At the
// CLI's own stderr and socket: DBF-11 within a second, no silence over 16 s,
// and its rows after more than the hop's 30 s. The console's real
// http.Server runs with ReadTimeout 2 s, as in
// TestDBForwardLongRequestThroughTheEdge.
// Mutations: drop the stream's flush (red: the wait lines sit in the
// console's buffer and the CLI's socket is silent); stream nothing before
// the box's first chunk (red: no headers inside the hop's 30 s).
func TestDBForwardThroughTheBackhaulHop(t *testing.T) {
	var tap *dbfTap
	e := newDBFE2E(t, dbfE2EOptions{forward: true, serve: throughBackhaul(2*time.Second, &tap), worker: slowRestore(40 * time.Second)})
	checkProgressAtTheCLI(t, e, tap, 40*time.Second)
}

func sha256File(t *testing.T, path string) string {
	t.Helper()
	b, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	sum := sha256.Sum256(b)
	return hex.EncodeToString(sum[:])
}

// The pull's parity (plan 3): a forwarded pull and an in-region pull of the
// same replica save byte-identical copies, for the newest copy and as of a
// time, and print the same lines but for the host and the path.
func TestDBForwardPullParityThroughTheCLI(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true})
	for _, at := range []string{"", e.betweenTheSyncs(t)} {
		dir := t.TempDir()
		args := []string{"db", "pull", "--account", "ops", "--app", "shop"}
		if at != "" {
			args = append(args, "--at", at)
		}
		hereOut, farOut := filepath.Join(dir, "here.db"), filepath.Join(dir, "far.db")
		here, hereErr, err := e.run(t, append(args, "--instance", dbfE2EHere, "--out", hereOut)...)
		if err != nil {
			t.Fatalf("in region: %v\n%s", err, hereErr)
		}
		far, farErr, err := e.run(t, append(args, "--instance", dbfE2EFar, "--out", farOut)...)
		if err != nil {
			t.Fatalf("forwarded: %v\n%s", err, farErr)
		}
		if sha256File(t, hereOut) != sha256File(t, farOut) {
			t.Fatalf("at %q the copies differ", at)
		}
		if here == "" || strings.ReplaceAll(strings.ReplaceAll(far, dbfE2EFar, dbfE2EHere), farOut, hereOut) != here {
			t.Fatalf("at %q: in region %q, forwarded %q", at, here, far)
		}
		if !strings.Contains(farErr, fmt.Sprintf(dbfFetching, dbfE2EFar, "uk")) {
			t.Fatalf("no DBF-18: %s", farErr)
		}
		db, _ := sql.Open("sqlite", "file:"+farOut+"?mode=ro")
		var later int
		err = db.QueryRow(`SELECT count(*) FROM orders WHERE note = 'later'`).Scan(&later)
		db.Close()
		if err != nil || (at == "") != (later == 1) {
			t.Fatalf("at %q the copy holds %d of the second sync's rows (%v)", at, later, err)
		}
	}
}

// T11 for a pull: a hash mismatch saves nothing at --out. Everything else
// is a pull the CLI would accept: far's real planned files, whole and
// valid, a manifest that passes checkPullManifest, and a summary whose only
// fault is its SHA-256 (plan review finding 32). Mutation: openDBPull skips
// the hash comparison (red: the files restore and a copy is saved).
func TestDBForwardPullSavesNothingForABadAnswer(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	stop := e.scriptBox(t, func(b *e2eScript) {
		ctx := context.Background()
		rd := lsfile.NewReplicaClient(filepath.Join(e.root, "instances", dbfE2EFar))
		plan, err := dbdoor.PlanFrom(ctx, rd, time.Time{})
		if err != nil {
			t.Errorf("plan far's replica: %v", err)
			return
		}
		var all bytes.Buffer
		for _, f := range plan.Planned {
			rc, err := rd.OpenLTXFile(ctx, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
			if err != nil {
				t.Errorf("open %+v: %v", f, err)
				return
			}
			io.Copy(&all, rc)
			rc.Close()
		}
		b.picked(time.Now().Add(time.Minute))
		sealed, _ := dbreads.SealChunk(b.pub, b.id, 1, dbreads.StreamLTX, true, all.Bytes())
		b.store.Put(ctx, dbreads.ChunkKey(b.id, 1), sealed, "")
		wrong := sha256.Sum256([]byte("not these bytes"))
		summary, _ := dbreads.SealSummary(b.pub, b.id, dbreads.Summary{TXID: plan.TXID, At: plan.At, Files: plan.Planned,
			Bytes: int64(all.Len()), SHA256: hex.EncodeToString(wrong[:])})
		r, _ := json.Marshal(dbreads.Result{State: dbreads.StateDone, LastSeq: 1, Box: "uk-1", Sealed: summary})
		b.store.Put(ctx, dbreads.ResultKey(b.id), r, "")
	})
	defer stop()
	out := filepath.Join(t.TempDir(), "far.db")
	_, stderr, err := e.run(t, "db", "pull", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
	if err == nil || !strings.Contains(stderr, fmt.Sprintf(dbfUnchecked, "uk")) {
		t.Fatalf("%q %v", stderr, err)
	}
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Fatalf("something was saved at --out: %v", err)
	}
}

// Review Focus 4 and plan review finding 23, through the binary: SIGINT
// while the CLI is spooling a pull prints "stopped", saves nothing at --out
// and removes the pull's directory. The scripted box (a stand-in) writes two
// chunks and never a result, so the CLI is mid-pull when the signal lands.
// Mutation: drop dbPullForward's deferred RemoveAll (red: carlos-db-* is
// left in the CLI's TMPDIR).
func TestDBForwardPullInterruptedLeavesNothing(t *testing.T) {
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true})
	stop := e.scriptBox(t, func(b *e2eScript) {
		b.picked(time.Now().Add(time.Minute))
		for seq := int64(1); seq <= 2; seq++ {
			sealed, _ := dbreads.SealChunk(b.pub, b.id, seq, dbreads.StreamLTX, false, bytes.Repeat([]byte{'x'}, 4096))
			b.store.Put(context.Background(), dbreads.ChunkKey(b.id, seq), sealed, "")
		}
	})
	defer stop()
	spool := t.TempDir()
	out := filepath.Join(t.TempDir(), "far.db")
	cmd := exec.Command(e.bin, "db", "pull", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
	// The last TMPDIR wins (os/exec keeps the last of a duplicated key).
	cmd.Env = append(carlosTestEnv(t, append([]string{"CARLOS_CREDENTIALS=" + e.creds}, e.env...)...), "TMPDIR="+spool)
	cmd.Dir = t.TempDir()
	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	deadline := time.Now().Add(30 * time.Second)
	for {
		if got, _ := filepath.Glob(filepath.Join(spool, "carlos-db-*", "sealed", "2")); len(got) == 1 {
			break
		}
		if time.Now().After(deadline) {
			cmd.Process.Kill()
			cmd.Wait()
			t.Fatalf("the CLI never spooled two chunks: %s", stderr.String())
		}
		time.Sleep(50 * time.Millisecond)
	}
	if err := cmd.Process.Signal(os.Interrupt); err != nil {
		t.Fatal(err)
	}
	if err := cmd.Wait(); err == nil || !strings.Contains(stderr.String(), "stopped") {
		t.Fatalf("%v: %s", err, stderr.String())
	}
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Fatalf("something was saved at --out: %v", err)
	}
	if left, _ := filepath.Glob(filepath.Join(spool, "carlos-db-*")); len(left) != 0 {
		t.Fatalf("the pull's directory is still there: %v", left)
	}
}

// A console from plan 2 forwards queries and not pulls. Its marker is
// played here by a wrapper that drops "pull" from forward_ops on the way
// out: the new CLI says DBF-8 and saves nothing. The wrapper buffers each
// response, so it serves only where no response streams (the list door's
// 409 is all this test reads).
func TestDBForwardPullOnAQueryOnlyConsole(t *testing.T) {
	queryOnly := func(t *testing.T, h http.Handler) (string, []string) {
		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			rec := httptest.NewRecorder()
			h.ServeHTTP(rec, r)
			body := rec.Body.Bytes()
			if rec.Code == http.StatusConflict {
				body = bytes.Replace(body, []byte(`"forward_ops":["query","pull"]`), []byte(`"forward_ops":["query"]`), 1)
			}
			for k, vs := range rec.Header() {
				w.Header()[k] = vs
			}
			w.Header().Del("Content-Length")
			w.WriteHeader(rec.Code)
			w.Write(body)
		}))
		t.Cleanup(ts.Close)
		return ts.URL, nil
	}
	e := newDBFE2E(t, dbfE2EOptions{forward: true, noBox: true, serve: queryOnly})
	out := filepath.Join(t.TempDir(), "far.db")
	_, stderr, err := e.run(t, "db", "pull", "--account", "ops", "--app", "shop", "--instance", dbfE2EFar, "--out", out)
	if err == nil || !strings.Contains(stderr, fmt.Sprintf(dbfNoPull, dbfE2EFar, "uk")) {
		t.Fatalf("%q %v", stderr, err)
	}
	if _, err := os.Stat(out); !os.IsNotExist(err) {
		t.Fatalf("something was saved at --out: %v", err)
	}
}