// End-to-end acceptance for `carlos db` (spec 2026-09-02): a REAL
// console.Server whose door reads a fake S3 holding a real litestream
// replica, through the production replica client (lease.ReadReplicaClient
// with CARLOS_S3_ENDPOINT), a device-flow login, then the real runDB verbs
// driven in-process. The member's whole path, with no box access and no
// AWS credentials of their own. TestE2ELogsThroughConsole's shape.
package main

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"encoding/xml"
	"fmt"
	"io/fs"
	"net/http"
	"net/http/httptest"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/superfly/ltx"

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

// replicaS3 is just enough S3 for litestream's read path (ListObjectsV2 and
// a ranged GetObject), serving file replicas under root in the S3 key
// layout <prefix>/<level %04x>/<file>.ltx. It records every key asked for,
// so a test can assert nothing outside the coordinates was ever requested.
type replicaS3 struct {
	t      *testing.T
	bucket string
	root   string
	mu     sync.Mutex
	keys   []string
}

func (s *replicaS3) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	rest := strings.TrimPrefix(r.URL.Path, "/"+s.bucket)
	if r.Method != http.MethodGet {
		s.t.Errorf("fake S3: %s %s; the door must only read", r.Method, r.URL.Path)
		w.WriteHeader(http.StatusForbidden)
		return
	}
	if rest == "" || rest == "/" {
		s.list(w, r.URL.Query().Get("prefix"))
		return
	}
	key := strings.TrimPrefix(rest, "/")
	s.mu.Lock()
	s.keys = append(s.keys, key)
	s.mu.Unlock()
	f, err := os.Open(s.file(key))
	if err != nil {
		w.Header().Set("Content-Type", "application/xml")
		w.WriteHeader(http.StatusNotFound)
		w.Write([]byte(`<Error><Code>NoSuchKey</Code></Error>`))
		return
	}
	defer f.Close()
	http.ServeContent(w, r, "", time.Time{}, f)
}

// file maps <p>/<level hex>/<name> to the file client's <p>/ltx/<level>/<name>.
func (s *replicaS3) file(key string) string {
	dir, name := path.Split(key)
	prefix, levelHex := path.Split(strings.TrimSuffix(dir, "/"))
	level, err := strconv.ParseInt(levelHex, 16, 32)
	if err != nil {
		return filepath.Join(s.root, "nonexistent")
	}
	return filepath.Join(s.root, filepath.FromSlash(prefix), "ltx", strconv.Itoa(int(level)), name)
}

func (s *replicaS3) list(w http.ResponseWriter, prefix string) {
	type obj struct {
		Key          string
		Size         int64
		LastModified string
	}
	var objs []obj
	dir, _ := path.Split(prefix)
	p, levelHex := path.Split(strings.TrimSuffix(dir, "/"))
	if level, err := strconv.ParseInt(levelHex, 16, 32); err == nil {
		local := filepath.Join(s.root, filepath.FromSlash(p), "ltx", strconv.Itoa(int(level)))
		ents, _ := os.ReadDir(local)
		for _, e := range ents {
			if _, _, err := ltx.ParseFilename(e.Name()); err != nil {
				continue
			}
			fi, err := e.Info()
			if err != nil {
				continue
			}
			objs = append(objs, obj{Key: prefix + e.Name(), Size: fi.Size(), LastModified: fi.ModTime().UTC().Format(time.RFC3339)})
		}
	}
	type result struct {
		XMLName     xml.Name `xml:"ListBucketResult"`
		Name        string
		Prefix      string
		KeyCount    int
		IsTruncated bool
		Contents    []obj
	}
	w.Header().Set("Content-Type", "application/xml")
	xml.NewEncoder(w).Encode(result{Name: s.bucket, Prefix: prefix, KeyCount: len(objs), Contents: objs})
}

func TestE2EDBThroughConsole(t *testing.T) {
	useTempCredentials(t)
	clearDeploymentEnv(t)
	checkNoRegistry := requireNoRegistryTouch(t)

	tmp := t.TempDir()
	const host = "acme.go.tito.io"
	replicaRoot := filepath.Join(tmp, "data")
	// Two batches, then a third in a later whole second, so --at can
	// name a time between them in the fake S3's second-resolution listing.
	replicaDir := filepath.Join(replicaRoot, "instances", host)
	src := dbdoortest.Replicate(t, replicaDir, 2)
	time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 10*time.Millisecond)))
	between := time.Now()
	time.Sleep(time.Until(between.Truncate(time.Second).Add(time.Second + 10*time.Millisecond)))
	{
		db, err := sql.Open("sqlite", src)
		if err != nil {
			t.Fatal(err)
		}
		for i := 0; i < dbdoortest.Rows; i++ {
			if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("row-2-%d", i)); err != nil {
				t.Fatal(err)
			}
		}
		db.Close()
		dbdoortest.Sync(t, src, replicaDir)
	}
	// lock.json sits under the same prefix in production. Nothing may fetch it.
	if err := os.WriteFile(filepath.Join(replicaRoot, "instances", host, "lock.json"), []byte(`{"owner":"box"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	fake := &replicaS3{t: t, bucket: "data-test", root: replicaRoot}
	s3srv := httptest.NewServer(fake)
	t.Cleanup(s3srv.Close)
	t.Setenv("CARLOS_S3_ENDPOINT", s3srv.URL)
	t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
	t.Setenv("AWS_EC2_METADATA_DISABLED", "true")

	relStore := release.NewDirStore(filepath.Join(tmp, "store"))
	seed, pubB64, err := release.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, err := release.ParseSigningKey(seed, "e2e")
	if err != nil {
		t.Fatal(err)
	}
	pub, err := release.ParsePublicKey(pubB64, "e2e")
	if err != nil {
		t.Fatal(err)
	}
	cst, err := console.Open(filepath.Join(tmp, "console.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { cst.Close() })
	var srv *console.Server
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		srv.ServeHTTP(w, r)
	}))
	t.Cleanup(ts.Close)
	srv, err = console.NewServer(console.Config{
		Origin:          ts.URL,
		Keymail:         "http://keymail.invalid",
		Registry:        filepath.Join(tmp, "registry.db"),
		Releases:        relStore,
		ReleasePub:      pub,
		Key:             priv,
		AppScheme:       "http",
		HibernateBucket: "data-test",
		HibernateRegion: "us-east-1",
		HibernatePrefix: "instances",
	}, cst)
	if err != nil {
		t.Fatal(err)
	}
	const addr = "paul@keymail.dev"
	if err := cst.AddMember(console.OpsAccountID, addr, console.RoleOwner, time.Now()); err != nil {
		t.Fatal(err)
	}
	if _, err := cst.CreateAppIn(console.OpsAccountID, "titogo", addr, time.Now()); err != nil {
		t.Fatal(err)
	}
	yes := true
	rec, _ := json.Marshal(provision.Record{Host: host, App: "titogo", Account: release.OpsAccount, Hibernate: &yes})
	if _, err := relStore.Put(context.Background(), provision.RecordKey(release.OpsAccount, "titogo", host), rec, ""); err != nil {
		t.Fatal(err)
	}
	token, expiresAt := e2eDeviceLogin(t, ts.URL, cst, addr, "e2e@laptop")
	credFile, err := loadCredentials()
	if err != nil {
		t.Fatal(err)
	}
	credFile.put(credential{Console: ts.URL, Address: addr, Token: token, ExpiresAt: expiresAt})
	if err := credFile.save(); err != nil {
		t.Fatal(err)
	}
	t.Setenv("TMPDIR", t.TempDir())

	// pull: the copy lands at --out, 0600, with every row, and the handover
	// says it is a copy and names its newest transaction.
	out := filepath.Join(t.TempDir(), "pulled.db")
	buf := captureStdout(t)
	runDB([]string{"pull", "--console", ts.URL, "--app", "titogo", "--account", "ops", "--instance", host, "--out", out})
	said := buf.String()
	for _, want := range []string{out, "copy", "transaction", host} {
		if !strings.Contains(said, want) {
			t.Errorf("pull said %q, want it to mention %q", said, want)
		}
	}
	fi, err := os.Stat(out)
	if err != nil {
		t.Fatal(err)
	}
	if fi.Mode().Perm() != 0o600 {
		t.Errorf("pulled copy is %04o, want 0600", fi.Mode().Perm())
	}
	var rows bytes.Buffer
	if err := runQuery(&rows, out, "SELECT count(*) AS n FROM t"); err != nil || rows.String() != "n\n"+strconv.Itoa(3*dbdoortest.Rows)+"\n" {
		t.Fatalf("pulled copy = %q %v", rows.String(), err)
	}

	// pull --at: the copy as of a time between the second and third
	// batches holds the first two, and says which time it was asked for.
	outAt := filepath.Join(t.TempDir(), "pulled-at.db")
	buf.Reset()
	runDB([]string{"pull", "--console", ts.URL, "--app", "titogo", "--account", "ops", "--instance", host, "--out", outAt, "--at", between.UTC().Format(time.RFC3339Nano)})
	if want := "the last one replicated before " + dbTime(between); !strings.Contains(buf.String(), want) {
		t.Errorf("pull --at said %q, want it to mention %q", buf.String(), want)
	}
	rows.Reset()
	if err := runQuery(&rows, outAt, "SELECT count(*) AS n FROM t"); err != nil || rows.String() != "n\n"+strconv.Itoa(2*dbdoortest.Rows)+"\n" {
		t.Fatalf("copy as of %v = %q %v, want %d rows", between, rows.String(), err, 2*dbdoortest.Rows)
	}

	// query: rows on stdout, nothing else.
	buf.Reset()
	runDB([]string{"query", "--console", ts.URL, "--app", "titogo", "--account", "ops", "--instance", host, "--sql", "SELECT body FROM t ORDER BY id LIMIT 2"})
	if got := buf.String(); got != "body\nrow-0-0\nrow-0-1\n" {
		t.Errorf("query stdout = %q, want the header and two rows only", got)
	}

	// Nothing private is left in the temp dir, and nothing but LTX files
	// under this instance's prefix was ever asked of the bucket.
	_ = filepath.WalkDir(os.Getenv("TMPDIR"), func(p string, d fs.DirEntry, err error) error {
		if err == nil && strings.HasPrefix(d.Name(), "carlos-db-") {
			t.Errorf("restore directory left behind: %s", p)
		}
		return nil
	})
	fake.mu.Lock()
	defer fake.mu.Unlock()
	if len(fake.keys) == 0 {
		t.Fatal("the door never read the bucket")
	}
	for _, k := range fake.keys {
		if !strings.HasPrefix(k, "instances/"+host+"/") || !strings.HasSuffix(k, ".ltx") {
			t.Errorf("the door asked the bucket for %q", k)
		}
	}
	checkNoRegistry()
}