package console

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"database/sql"
	"encoding/json"
	"path/filepath"
	"testing"
	"time"

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

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

// The parity half of spec T6 that needs no console server (plan 1): the box
// path (dbreads.Agent through the real worker code, joined in process: not
// acceptance for the units) and the CLI's in-region path (dbdoor.RestoreAsOf
// over this package's own door code, replicaDoer and listReplicaLevel, then
// dbquery.RunContext) print the same bytes, as of the same transaction: the
// newest copy, a time between writes, and a time in the same second as a
// write. Plan 2's T6 runs the same comparison through the console server and
// the CLI binary.
func TestDBReadsParityWithTheInRegionDoor(t *testing.T) {
	const host = "shop.example.com"
	dir := t.TempDir()
	src := filepath.Join(t.TempDir(), "shop.db")
	db, err := sql.Open("sqlite", src)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	exec := func(q string, args ...any) {
		t.Helper()
		if _, err := db.Exec(q, args...); err != nil {
			t.Fatal(err)
		}
	}
	nextSecond := func() time.Time {
		time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 10*time.Millisecond)))
		return time.Now()
	}
	exec(`PRAGMA journal_mode=WAL; CREATE TABLE orders (id INTEGER PRIMARY KEY, note TEXT, data BLOB, at DATETIME)`)
	exec(`INSERT INTO orders (note, data, at) VALUES (?, ?, ?), (NULL, NULL, NULL), (?, ?, ?)`,
		"a\ttab", []byte{0, 0xff}, time.Date(2026, 10, 7, 14, 0, 0, 0, time.UTC),
		"two\nlines", []byte("x"), time.Date(2026, 10, 7, 14, 0, 1, 500, time.UTC))
	dbdoortest.Sync(t, src, dir)
	between := nextSecond()
	nextSecond()
	exec(`INSERT INTO orders (note) VALUES ('later')`)
	dbdoortest.Sync(t, src, dir)
	sameSecond := time.Now() // in the second the last file was written
	nextSecond()

	store := release.NewDirStore(t.TempDir())
	pub, priv, _ := ed25519.GenerateKey(rand.Reader)
	yes := true
	rec := provision.Record{Host: host, App: "shop", Account: "bab", Region: "uk", Hibernate: &yes, CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)}
	body, _ := json.Marshal(rec)
	if _, err := store.Put(context.Background(), provision.RecordKey("bab", "shop", host), body, ""); err != nil {
		t.Fatal(err)
	}
	inc, _ := hostclaim.Incarnation("bab", "shop", host, rec.CreatedAt)
	box := dbreadstest.Start(t, dbreadstest.Options{Store: store, Region: "uk", Box: "uk-1", Pub: pub, HibernatePrefix: "instances",
		OpenReplica: func(context.Context, string) (dbdoor.ReplicaReader, error) { return lsfile.NewReplicaClient(dir), nil }})
	const query = "SELECT id, note, data, at FROM orders ORDER BY id"
	for name, at := range map[string]time.Time{"newest": {}, "between writes": between, "same second as a write": sameSecond} {
		got, err := dbdoor.RestoreAsOf(context.Background(), dbdoor.NewClient(replicaDoer(lsfile.NewReplicaClient(dir))), t.TempDir(), at)
		if err != nil {
			t.Fatalf("%s: in-region: %v", name, err)
		}
		var want bytes.Buffer
		if err := dbquery.RunContext(context.Background(), &want, got.Path, query, dbquery.Options{}); err != nil {
			t.Fatal(err)
		}
		reply, _ := dbreads.NewReplyKey()
		id, _ := dbreads.MintID()
		atRaw := ""
		if !at.IsZero() {
			atRaw = at.UTC().Format(time.RFC3339Nano)
		}
		req := dbreads.Request{ID: id, Account: "bab", App: "shop", Host: host, Incarnation: inc, Region: "uk",
			Op: dbreads.OpQuery, SQL: query, At: atRaw, ReplyKey: dbreads.EncodeReplyKey(reply.PublicKey()),
			By: "member@example.com", IssuedAt: time.Now().UTC().Format(time.RFC3339)}
		dbreadstest.Submit(t, store, priv, req)
		box.Serve(t, store, id, 30*time.Second)
		rows, s, r := dbreadstest.Answer(t, store, id, reply, dbreads.StreamRows)
		if r.State != dbreads.StateDone || !bytes.Equal(rows, want.Bytes()) || s.TXID != got.TXID || !s.At.Equal(got.At) {
			t.Fatalf("%s: box %+v txid %d at %v rows %q; in-region txid %d at %v rows %q",
				name, r, s.TXID, s.At, rows, got.TXID, got.At, want.Bytes())
		}
	}
}