// dbforward.go is the CLI's half of a forwarded `carlos db query` and
// `carlos db pull` (spec
// docs/superpowers/specs/2026-10-22-db-region-forward-design.md §5.5, §6):
// when the db door answers with the forward marker, the CLI makes an
// ephemeral key, holds one request to the console's reads route open, keeps
// every sealed chunk (in memory for a query, spooled to a private directory
// for a pull), and prints or saves nothing until a done result's summary
// checks out against the bytes (codex 6).
package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

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

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

// The forwarded read's CLI sentences (copy record DBF-3 to DBF-8, DBF-11 to
// DBF-13, DBF-15, DBF-16, DBF-32), approved by Luna on 2026-10-23.
const (
	dbfNotPicked  = "no server in region %s started the read within 60 seconds. try again in a few minutes."                                                         // DBF-3
	dbfTooLarge   = "the database for %s is %s once restored. carlos db can query at most %s in region %s."                                                          // DBF-4
	dbfNoRoom     = "the server in region %s does not have room to restore %s's database right now. try again later."                                                // DBF-5
	dbfOutputCap  = "the result was cut off at 64 MB. narrow the query with WHERE or LIMIT and run it again."                                                        // DBF-6
	dbfTimeout    = "the query ran for 5 minutes and was stopped. narrow it and run it again."                                                                       // DBF-7
	dbfNoPull     = "%s keeps its data in region %s. carlos db pull cannot fetch a copy from there yet. carlos db query can read it."                                // DBF-8
	dbfReading    = "reading a copy of %s's database in region %s."                                                                                                  // DBF-11
	dbfUnfinished = "the read of %s's database in region %s did not finish, so nothing was printed. try again."                                                      // DBF-12
	dbfUnchecked  = "the answer from region %s was invalid, so carlos db printed nothing. try again."                                                                // DBF-13
	dbfChanged    = "%s changed while the read was waiting. try again."                                                                                              // DBF-15
	dbfMovingCLI  = "%s is moving between regions. read its database once the move has finished."                                                                    // DBF-16
	dbfNotUTF8    = "this query cannot be sent to region %s, because it contains invalid UTF-8 or the replacement character U+FFFD. fix the query and run it again." // DBF-32
)

// The forwarded pull's CLI sentences (copy record DBF-17, DBF-18, DBF-24),
// approved by Luna on 2026-10-23.
const (
	dbfPullTooLarge = "the database for %s has %s of files to fetch. carlos db pull can fetch at most %s from region %s."         // DBF-17
	dbfFetching     = "fetching a copy of %s's database from region %s."                                                          // DBF-18
	dbfPullNoRoom   = "the server in region %s does not have room to prepare a copy of %s's database right now. try again later." // DBF-24
)

// errDBForward is the db door's forward marker, handed back by dbRestore so
// a verb can switch to the forwarded read.
type errDBForward struct {
	door *dbdoor.DoorError
}

func (e *errDBForward) Error() string { return e.door.Error() }

type dbForwardAnswer struct {
	rows    []byte
	summary dbreads.Summary
	capped  bool
}

// dbQueryForward is spec §5.5 steps 2 and 8 for a query.
func dbQueryForward(ctx context.Context, t dbTarget, sqlText string, fwd dbdoor.Forward) error {
	// json.Marshal would send U+FFFD for bytes that are not UTF-8, and the
	// box would run SQL nobody typed; the console refuses U+FFFD for the
	// same reason (plan review finding 36). First, so no other check can
	// read the bytes differently.
	if !utf8.ValidString(sqlText) || strings.ContainsRune(sqlText, utf8.RuneError) {
		return fmt.Errorf(dbfNotUTF8, fwd.Region)
	}
	if err := checkQuerySQL(sqlText); err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(ctx, dbreads.CLIBound)
	defer cancel()
	key, err := dbreads.NewReplyKey()
	if err != nil {
		return err
	}
	at := ""
	if !t.at.IsZero() {
		at = t.at.UTC().Format(time.RFC3339Nano)
	}
	body, err := json.Marshal(dbreads.ReadBody{Op: dbreads.OpQuery, SQL: sqlText, At: at, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	if err != nil {
		return err
	}
	res, err := t.sess.c.doStream(ctx, http.MethodPost, apiPath(t.acct, t.app, "/db/"+url.PathEscape(t.host)+"/reads"),
		bytes.NewReader(body), int64(len(body)), http.Header{"Content-Type": {"application/json"}})
	if err != nil {
		if errors.Is(ctx.Err(), context.Canceled) {
			return errors.New("stopped")
		}
		return fmt.Errorf(dbfUnfinished, t.host, fwd.Region)
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10))
		return apiErrorFromBody(res.StatusCode, raw)
	}
	sink := &memSink{}
	id, result, err := readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfReading, os.Stderr, sink, newDBFChunkBound(dbreads.MaxOutputBytes, dbreads.ChunkBytes))
	if err != nil {
		return err
	}
	ans, err := openDBForward(id, key, sink, *result, t.host, fwd.Region)
	if err != nil {
		return err
	}
	fmt.Fprint(os.Stderr, dbQueryStatus(t.host, &dbdoor.Restored{TXID: ans.summary.TXID, At: ans.summary.At}, t.at))
	if _, err := stdout.Write(ans.rows); err != nil {
		return err
	}
	if ans.capped {
		return errors.New(dbfOutputCap)
	}
	return nil
}

// dbfChunkBound is what a reader keeps before a result arrives (plan review
// finding 37). Every chunk is at least a sealed empty one; only the last
// may be empty, so nothing may follow an empty chunk; there are at most as
// many chunks as a read can make (full ones for the payload, plus one per
// flush interval over the owner's deadline, plus the last); and their
// plaintext, ciphertext less each seal, stays within the payload cap.
type dbfChunkBound struct {
	maxBytes, maxChunks int64
	n, plain            int64
	sawEmpty            bool
}

func newDBFChunkBound(maxBytes, chunkBytes int64) *dbfChunkBound {
	return &dbfChunkBound{maxBytes: maxBytes,
		maxChunks: (maxBytes+chunkBytes-1)/chunkBytes + int64(dbreads.OwnerDeadline/dbreads.ChunkEvery) + 1}
}

// admit says whether a chunk may be kept; false is DBF-13.
func (b *dbfChunkBound) admit(sealed []byte) bool {
	size := int64(len(sealed)) - int64(dbreads.SealOverhead)
	if size < 0 || b.sawEmpty || b.n >= b.maxChunks || b.plain+size > b.maxBytes {
		return false
	}
	b.n++
	b.plain += size
	b.sawEmpty = size == 0
	return true
}

// dbChunkSink holds a read's sealed chunks until its result arrives: in
// memory for a query (at most 64 MiB), in the pull's private directory for a
// pull (up to its cap).
type dbChunkSink interface {
	add(sealed []byte) error
	count() int
	chunk(i int) ([]byte, error) // 0-based
}

type memSink struct{ chunks [][]byte }

func (m *memSink) add(b []byte) error          { m.chunks = append(m.chunks, b); return nil }
func (m *memSink) count() int                  { return len(m.chunks) }
func (m *memSink) chunk(i int) ([]byte, error) { return m.chunks[i], nil }

// fileSink spools sealed chunks to dir, one file each, 0600 in a 0700
// directory the caller removes.
type fileSink struct {
	dir string
	n   int
}

func (f *fileSink) add(b []byte) error {
	if f.n == 0 {
		if err := os.MkdirAll(f.dir, 0o700); err != nil {
			return err
		}
	}
	f.n++
	return dbPullSpoolWrite(filepath.Join(f.dir, strconv.Itoa(f.n)), b, 0o600)
}

func (f *fileSink) count() int { return f.n }

func (f *fileSink) chunk(i int) ([]byte, error) {
	return os.ReadFile(filepath.Join(f.dir, strconv.Itoa(i+1)))
}

// drop removes chunk i once its bytes are in the files, so the spool and
// the files are never both whole on the member's disk.
func (f *fileSink) drop(i int) { os.Remove(filepath.Join(f.dir, strconv.Itoa(i+1))) }

// readDBForward reads the reads route's lines (spec §5.5 steps 6 to 8). It
// says the reading sentence once admitted, keeps every chunk in sequence in
// sink (each one checked against bound before the sink keeps anything, plan
// review finding 37), and returns the read's id and its result unopened.
func readDBForward(ctx context.Context, body io.Reader, key *ecdh.PrivateKey, host, region, reading string, status io.Writer, sink dbChunkSink, bound *dbfChunkBound) (string, *dbreads.Result, error) {
	unchecked := fmt.Errorf(dbfUnchecked, region)
	sc := bufio.NewScanner(body)
	sc.Buffer(make([]byte, 64<<10), dbreads.MaxLineBytes)
	var id, ended string
	for sc.Scan() {
		var l dbreads.Line
		if err := json.Unmarshal(sc.Bytes(), &l); err != nil {
			return "", nil, unchecked
		}
		switch {
		case l.Admitted != "":
			if id != "" || !dbreads.ValidID(l.Admitted) {
				return "", nil, unchecked
			}
			id = l.Admitted
			fmt.Fprintln(status, fmt.Sprintf(reading, host, region))
		case l.Chunk != 0:
			if id == "" || l.Chunk != int64(sink.count())+1 || !bound.admit(l.Data) {
				return "", nil, unchecked
			}
			if err := sink.add(l.Data); err != nil {
				return "", nil, err
			}
		case l.Result != nil:
			if id == "" {
				return "", nil, unchecked
			}
			return id, l.Result, nil
		case l.Ended != "":
			ended = l.Ended
		}
	}
	if errors.Is(ctx.Err(), context.Canceled) {
		return "", nil, errors.New("stopped")
	}
	if ended == dbreads.EndedNotPicked {
		return "", nil, fmt.Errorf(dbfNotPicked, region)
	}
	return "", nil, fmt.Errorf(dbfUnfinished, host, region)
}

// openDBForward opens a result: a done one's every chunk and its summary,
// checked against each other before anything is returned; any other state
// mapped to its sentence (spec §5.6, §10).
func openDBForward(id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region string) (dbForwardAnswer, error) {
	unchecked := fmt.Errorf(dbfUnchecked, region)
	if res.Valid() != nil {
		return dbForwardAnswer{}, unchecked
	}
	s, err := dbreads.OpenSummary(key, id, res.Sealed)
	if err != nil {
		return dbForwardAnswer{}, unchecked
	}
	switch res.State {
	case dbreads.StateDone:
		if res.LastSeq != int64(sink.count()) {
			return dbForwardAnswer{}, unchecked
		}
		var rows bytes.Buffer
		for i := 0; i < sink.count(); i++ {
			c, err := sink.chunk(i)
			if err != nil {
				return dbForwardAnswer{}, unchecked
			}
			seq := int64(i + 1)
			plain, err := dbreads.OpenChunk(key, id, seq, dbreads.StreamRows, seq == res.LastSeq, c)
			if err != nil {
				return dbForwardAnswer{}, unchecked
			}
			rows.Write(plain)
		}
		sum := sha256.Sum256(rows.Bytes())
		if int64(rows.Len()) != s.Bytes || hex.EncodeToString(sum[:]) != s.SHA256 {
			return dbForwardAnswer{}, unchecked
		}
		return dbForwardAnswer{rows: rows.Bytes(), summary: s, capped: res.Code == dbreads.CodeOutputCap}, nil
	case dbreads.StateRefused:
		if res.Code == dbreads.CodeMoving {
			return dbForwardAnswer{}, fmt.Errorf(dbfMovingCLI, host)
		}
		return dbForwardAnswer{}, fmt.Errorf(dbfChanged, host)
	case dbreads.StateFailed:
		switch res.Code {
		case dbreads.CodeNothingReplicated:
			return dbForwardAnswer{}, dbRestoreError(host, dbdoor.ErrNothingReplicated)
		case dbreads.CodeHistoryGone:
			if !s.Oldest.IsZero() {
				return dbForwardAnswer{}, dbRestoreError(host, &dbdoor.BeforeHistoryError{Oldest: s.Oldest})
			}
		case dbreads.CodeCheckFailed:
			return dbForwardAnswer{}, dbRestoreError(host, fmt.Errorf("%w: %s", dbdoor.ErrCheckFailed, s.Error))
		case dbreads.CodeChangedTwice:
			return dbForwardAnswer{}, dbRestoreError(host, os.ErrNotExist)
		case dbreads.CodeTooLarge:
			return dbForwardAnswer{}, fmt.Errorf(dbfTooLarge, host, humanBytes(&s.Restored), humanBytes(&s.Cap), region)
		case dbreads.CodeNoRoom:
			return dbForwardAnswer{}, fmt.Errorf(dbfNoRoom, region, host)
		case dbreads.CodeSQLError:
			return dbForwardAnswer{}, errors.New(s.Error)
		case dbreads.CodeTimeout:
			return dbForwardAnswer{}, errors.New(dbfTimeout)
		}
	}
	return dbForwardAnswer{}, fmt.Errorf(dbfUnfinished, host, region)
}

// openDBPull opens a pull's result (spec §6, phase 2): a done one's chunks,
// opened in order, are the planned files concatenated; they are checked
// against the summary's count and hash and split by its file list into a
// litestream file replica under dir, which today's restore and check turn
// into the copy. Any other state maps to its sentence.
func openDBPull(ctx context.Context, id string, key *ecdh.PrivateKey, sink dbChunkSink, res dbreads.Result, host, region, dir string) (*dbdoor.Restored, error) {
	unchecked := fmt.Errorf(dbfUnchecked, region)
	if res.Valid() != nil {
		return nil, unchecked
	}
	s, err := dbreads.OpenSummary(key, id, res.Sealed)
	if err != nil {
		return nil, unchecked
	}
	if res.State != dbreads.StateDone {
		switch res.Code {
		case dbreads.CodeTooLarge:
			return nil, fmt.Errorf(dbfPullTooLarge, host, humanBytes(&s.Restored), humanBytes(&s.Cap), region)
		case dbreads.CodeNoRoom:
			return nil, fmt.Errorf(dbfPullNoRoom, region, host)
		}
		_, err := openDBForward(id, key, sink, res, host, region)
		return nil, err
	}
	// A done pull is the whole answer: no output cap applies to one.
	if res.Code != "" || res.LastSeq != int64(sink.count()) || checkPullManifest(s) != nil {
		return nil, unchecked
	}
	root := filepath.Join(dir, "replica")
	h := sha256.New()
	fi, written := 0, int64(0)
	var head []byte // the current file's first ltx.HeaderSize bytes
	var out *os.File
	defer func() {
		if out != nil {
			out.Close()
		}
	}()
	drop, _ := sink.(interface{ drop(int) })
	for i := 0; i < sink.count(); i++ {
		// Ctrl-C and the read's deadline stop the rebuild too (Astra round
		// 1 on plan 3, finding 2); RestoreLocal's reads check ctx after it.
		if err := ctx.Err(); err != nil {
			return nil, dbPullStopped(err, host, region)
		}
		sealed, err := sink.chunk(i)
		if err != nil {
			return nil, unchecked
		}
		seq := int64(i + 1)
		plain, err := dbreads.OpenChunk(key, id, seq, dbreads.StreamLTX, seq == res.LastSeq, sealed)
		if err != nil {
			return nil, unchecked
		}
		h.Write(plain)
		for len(plain) > 0 {
			if fi >= len(s.Files) {
				return nil, unchecked
			}
			f := s.Files[fi]
			if out == nil {
				p := litestream.LTXFilePath(root, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID))
				if err := os.MkdirAll(filepath.Dir(p), 0o700); err != nil {
					return nil, err
				}
				if out, err = os.OpenFile(p, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600); err != nil {
					return nil, err
				}
			}
			n := min(int64(len(plain)), f.Size-written)
			if _, err := out.Write(plain[:n]); err != nil {
				return nil, err
			}
			if len(head) < ltx.HeaderSize {
				head = append(head, plain[:min(n, int64(ltx.HeaderSize-len(head)))]...)
			}
			plain, written = plain[n:], written+n
			if written == f.Size {
				if err := out.Close(); err != nil {
					out = nil
					return nil, err
				}
				// litestream's restore picks files by name and compacts them
				// by header: a file whose header is not its manifest entry
				// would restore as a transaction it does not hold (Astra
				// round 1 on plan 3, finding 1).
				var lh ltx.Header
				if lh.UnmarshalBinary(head) != nil || uint64(lh.MinTXID) != f.MinTXID || uint64(lh.MaxTXID) != f.MaxTXID {
					out = nil
					return nil, unchecked
				}
				out, fi, written, head = nil, fi+1, 0, nil
			}
		}
		if drop != nil {
			drop.drop(i)
		}
	}
	if fi != len(s.Files) || hex.EncodeToString(h.Sum(nil)) != s.SHA256 {
		return nil, unchecked
	}
	db := filepath.Join(dir, "replica.db")
	if err := dbdoor.RestoreLocal(ctx, lsfile.NewReplicaClient(root), db, s.TXID); err != nil {
		if cerr := ctx.Err(); cerr != nil {
			return nil, dbPullStopped(cerr, host, region)
		}
		return nil, dbRestoreError(host, err)
	}
	os.RemoveAll(root)
	return &dbdoor.Restored{Path: db, Dir: dir, TXID: s.TXID, At: s.At}, nil
}

// dbPullStopped is a pull that ended on its context: "stopped" for Ctrl-C,
// DBF-12 for the read's own deadline.
func dbPullStopped(err error, host, region string) error {
	if errors.Is(err, context.Canceled) {
		return errors.New("stopped")
	}
	return fmt.Errorf(dbfUnfinished, host, region)
}

// checkPullManifest is the summary's file list checked whole before any
// file is created (plan review finding 22): every size a possible LTX file
// and within the cap, every level one litestream writes, every range a
// range, no file twice, sizes that sum without overflow to the summary's
// byte count, and a last file that ends at its transaction.
func checkPullManifest(s dbreads.Summary) error {
	bad := errors.New("the file list does not add up")
	if len(s.Files) == 0 || s.Bytes <= 0 || s.Bytes > dbreads.MaxPullBytes {
		return bad
	}
	type coord struct {
		level    int
		min, max uint64
	}
	seen := make(map[coord]bool, len(s.Files))
	var sum int64
	for _, f := range s.Files {
		c := coord{f.Level, f.MinTXID, f.MaxTXID}
		switch {
		case f.Size < ltx.HeaderSize+ltx.TrailerSize, f.Size > dbreads.MaxPullBytes,
			f.Level < 0, f.Level > litestream.SnapshotLevel,
			f.MinTXID < 1, f.MaxTXID < f.MinTXID, seen[c],
			sum > dbreads.MaxPullBytes-f.Size: // sum + f.Size would pass the cap
			return bad
		}
		seen[c] = true
		sum += f.Size
	}
	if sum != s.Bytes || s.Files[len(s.Files)-1].MaxTXID != s.TXID {
		return bad
	}
	return nil
}

// dbPullSpoolWrite writes one spooled chunk: os.WriteFile, a seam so a test
// can fail the spool part way (plan review finding 23).
var dbPullSpoolWrite = os.WriteFile

// dbPullForward is a pull through the forwarded read (spec §6). The spool,
// the files and the copy live in one private 0700 directory under the system
// temp dir, removed on every path but success, where the caller removes it
// after copyOut, as it does today's restore directory.
func dbPullForward(ctx context.Context, t dbTarget, fwd dbdoor.Forward) (*dbdoor.Restored, error) {
	ctx, cancel := context.WithTimeout(ctx, dbreads.CLIBound)
	defer cancel()
	key, err := dbreads.NewReplyKey()
	if err != nil {
		return nil, err
	}
	at := ""
	if !t.at.IsZero() {
		at = t.at.UTC().Format(time.RFC3339Nano)
	}
	body, err := json.Marshal(dbreads.ReadBody{Op: dbreads.OpPull, At: at, ReplyKey: dbreads.EncodeReplyKey(key.PublicKey())})
	if err != nil {
		return nil, err
	}
	res, err := t.sess.c.doStream(ctx, http.MethodPost, apiPath(t.acct, t.app, "/db/"+url.PathEscape(t.host)+"/reads"),
		bytes.NewReader(body), int64(len(body)), http.Header{"Content-Type": {"application/json"}})
	if err != nil {
		if errors.Is(ctx.Err(), context.Canceled) {
			return nil, errors.New("stopped")
		}
		return nil, fmt.Errorf(dbfUnfinished, t.host, fwd.Region)
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10))
		return nil, apiErrorFromBody(res.StatusCode, raw)
	}
	dir, err := os.MkdirTemp("", "carlos-db-")
	if err != nil {
		return nil, err
	}
	// MkdirTemp already creates 0700; say it anyway, as the in-region
	// restore does, because the copy's privacy rests on it.
	if err := os.Chmod(dir, 0o700); err != nil {
		os.RemoveAll(dir)
		return nil, err
	}
	done := false
	defer func() {
		if !done {
			os.RemoveAll(dir)
		}
	}()
	sink := &fileSink{dir: filepath.Join(dir, "sealed")}
	id, result, err := readDBForward(ctx, res.Body, key, t.host, fwd.Region, dbfFetching, os.Stderr, sink, newDBFChunkBound(dbreads.MaxPullBytes, dbreads.PullChunkBytes))
	if err != nil {
		return nil, err
	}
	r, err := openDBPull(ctx, id, key, sink, *result, t.host, fwd.Region, dir)
	if err != nil {
		return nil, err
	}
	os.RemoveAll(sink.dir)
	done = true
	return r, nil
}