package main

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"io/fs"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"testing"
)

// newMirroredTestDB builds a real DB whose store is primary+mirror over two
// fake S3 servers — the exact construction openDB performs from env.
func newMirroredTestDB(t *testing.T) (*DB, *fakeS3, *fakeS3) {
	t.Helper()
	primary := &fakeS3{objects: map[string][]byte{}}
	mirror := &fakeS3{objects: map[string][]byte{}}
	psrv := httptest.NewServer(http.HandlerFunc(primary.handler))
	msrv := httptest.NewServer(http.HandlerFunc(mirror.handler))
	t.Cleanup(psrv.Close)
	t.Cleanup(msrv.Close)

	t.Setenv("AWS_ACCESS_KEY_ID", "test")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
	t.Setenv("AWS_REGION", "auto")
	t.Setenv("LCHAT_BLOB_S3_BUCKET", "prim")
	t.Setenv("LCHAT_BLOB_S3_ENDPOINT", psrv.URL)
	t.Setenv("LCHAT_BLOB_S3_MIRROR_BUCKET", "mirr")
	t.Setenv("LCHAT_BLOB_S3_MIRROR_ENDPOINT", msrv.URL)
	// The mirror carries its own key pair (a different provider) — exercises
	// the static-credentials path.
	t.Setenv("LCHAT_BLOB_S3_MIRROR_ACCESS_KEY", "mirror-key")
	t.Setenv("LCHAT_BLOB_S3_MIRROR_SECRET_KEY", "mirror-secret")

	db, err := openDB(filepath.Join(t.TempDir(), "space.example.com.db"))
	if err != nil {
		t.Fatalf("openDB: %v", err)
	}
	t.Cleanup(func() { db.sql.Close() })
	if _, ok := db.blobs.(*mirroredBlobStore); !ok {
		t.Fatalf("store is %T, want mirroredBlobStore", db.blobs)
	}
	return db, primary, mirror
}

// The Phase A loop: a put lands on the primary only; the sweep copies it to
// the mirror, records mirrored_at, and a second pass is a no-op (idempotence
// = zero further requests). Reads survive a lost primary object via the
// mirror, and deletes reap both stores.
func TestBlobMirrorSweep(t *testing.T) {
	db, primary, mirror := newMirroredTestDB(t)
	payload := []byte("ciphertext-bytes-0123456789")

	if err := db.putBlob("blobA", "t1", payload); err != nil {
		t.Fatalf("putBlob: %v", err)
	}
	if got := len(keysOf(mirror)); got != 0 {
		t.Fatalf("mirror has %d objects before the sweep", got)
	}

	n, err := db.mirrorOnce(32)
	if err != nil || n != 1 {
		t.Fatalf("mirrorOnce = (%d, %v), want (1, nil)", n, err)
	}
	mbytes, ok := mirror.objects["/mirr/space.example.com/blobA"]
	if !ok || !bytes.Equal(mbytes, payload) {
		t.Fatalf("mirror copy wrong: ok=%v have %v", ok, keysOf(mirror))
	}

	// Idempotent: nothing pending → zero puts on a second pass.
	basePuts := mirror.puts
	if n, err := db.mirrorOnce(32); err != nil || n != 0 {
		t.Fatalf("second mirrorOnce = (%d, %v), want (0, nil)", n, err)
	}
	if mirror.puts != basePuts {
		t.Fatalf("idle sweep PUT %d times", mirror.puts-basePuts)
	}

	// Redundancy pays out: primary object lost → the read serves from the
	// mirror.
	delete(primary.objects, "/prim/space.example.com/blobA")
	r, _, _, err := db.blobs.open("blobA")
	if err != nil {
		t.Fatalf("open after primary loss: %v", err)
	}
	got, _ := io.ReadAll(r)
	r.Close()
	if !bytes.Equal(got, payload) {
		t.Fatalf("mirror served %q", got)
	}

	// GC reaps both stores.
	if err := db.deleteBlobs("t1", []string{"blobA"}); err != nil {
		t.Fatalf("deleteBlobs: %v", err)
	}
	if len(keysOf(mirror)) != 0 {
		t.Fatalf("mirror kept %v after delete", keysOf(mirror))
	}
}

// A pre-hash row (sha256 = ”) gets its hash backfilled by the sweep — the
// migration path for every blob uploaded before this feature.
func TestBlobMirrorBackfillsHash(t *testing.T) {
	db, _, _ := newMirroredTestDB(t)
	payload := []byte("old blob, no hash on record")
	if err := db.putBlob("oldblob", "t1", payload); err != nil {
		t.Fatal(err)
	}
	if _, err := db.sql.Exec("UPDATE blobs SET sha256 = '' WHERE id = 'oldblob'"); err != nil {
		t.Fatal(err)
	}
	if _, err := db.mirrorOnce(32); err != nil {
		t.Fatal(err)
	}
	var sha string
	if err := db.sql.QueryRow("SELECT sha256 FROM blobs WHERE id = 'oldblob'").Scan(&sha); err != nil {
		t.Fatal(err)
	}
	want := sha256.Sum256(payload)
	if sha != hex.EncodeToString(want[:]) {
		t.Fatalf("backfilled hash = %.12s, want %.12s", sha, hex.EncodeToString(want[:]))
	}
}

// A primary whose bytes no longer match the recorded hash is NEVER mirrored —
// corruption must not spread to the copy — and the row stays unmirrored so it
// keeps being noticed.
func TestBlobMirrorRefusesCorruptPrimary(t *testing.T) {
	db, primary, mirror := newMirroredTestDB(t)
	if err := db.putBlob("blobC", "t1", []byte("genuine bytes")); err != nil {
		t.Fatal(err)
	}
	primary.objects["/prim/space.example.com/blobC"] = []byte("tampered bytes")

	n, err := db.mirrorOnce(32)
	if err != nil || n != 0 {
		t.Fatalf("mirrorOnce = (%d, %v), want (0, nil) — corrupt blob must not count as mirrored", n, err)
	}
	if len(keysOf(mirror)) != 0 {
		t.Fatalf("corruption reached the mirror: %v", keysOf(mirror))
	}
	var mirroredAt int64
	if err := db.sql.QueryRow("SELECT mirrored_at FROM blobs WHERE id = 'blobC'").Scan(&mirroredAt); err != nil || mirroredAt != 0 {
		t.Fatalf("mirrored_at = %d, want 0 (still pending)", mirroredAt)
	}
}

// Legacy in-DB rows (bytes in the data column) never enter the sweep — the
// DB's own replication carries them.
func TestBlobMirrorSkipsLegacyRows(t *testing.T) {
	db, _, mirror := newMirroredTestDB(t)
	if _, err := db.sql.Exec(
		"INSERT INTO blobs(id, data, size, thread_id, created_at) VALUES('legacy', X'AABB', 2, 't1', 1)"); err != nil {
		t.Fatal(err)
	}
	if n, err := db.mirrorOnce(32); err != nil || n != 0 {
		t.Fatalf("mirrorOnce = (%d, %v), want (0, nil)", n, err)
	}
	if len(keysOf(mirror)) != 0 {
		t.Fatalf("legacy row reached the mirror: %v", keysOf(mirror))
	}
}

// The audit sweep catches a corrupted or missing copy on either store and
// says which.
func TestBlobAudit(t *testing.T) {
	db, primary, mirror := newMirroredTestDB(t)
	if err := db.putBlob("blobD", "t1", []byte("audited bytes")); err != nil {
		t.Fatal(err)
	}
	if _, err := db.mirrorOnce(32); err != nil {
		t.Fatal(err)
	}

	// All healthy: a clean pass.
	res, err := db.auditBlobsOnce(5)
	if err != nil || res.Checked != 1 || res.BadPrimary+res.BadMirror+res.MissingPrimary+res.MissingMirror != 0 {
		t.Fatalf("healthy audit = %+v, %v", res, err)
	}

	// Corrupt the mirror; lose the primary. Both verdicts land.
	mirror.objects["/mirr/space.example.com/blobD"] = []byte("rot")
	delete(primary.objects, "/prim/space.example.com/blobD")
	res, err = db.auditBlobsOnce(5)
	if err != nil {
		t.Fatal(err)
	}
	if res.BadMirror != 1 {
		t.Fatalf("mirror corruption missed: %+v", res)
	}
	// The primary miss falls through to the DISK fallback, which still holds
	// nothing for an S3-era blob → missing.
	if res.MissingPrimary != 1 {
		t.Fatalf("primary loss missed: %+v", res)
	}
}

// A mirror configured without a primary is a config mistake openDB refuses.
func TestBlobMirrorRequiresPrimary(t *testing.T) {
	t.Setenv("LCHAT_BLOB_S3_BUCKET", "")
	t.Setenv("LCHAT_BLOB_S3_MIRROR_BUCKET", "mirr")
	if _, err := openDB(filepath.Join(t.TempDir(), "x.db")); err == nil {
		t.Fatal("openDB accepted a mirror without a primary")
	}
}

// The upload handler records the streamed sha256 — putBlob's twin assertion
// (the handler itself is exercised in handlers tests; this pins the hash).
func TestPutBlobRecordsHash(t *testing.T) {
	db, _, _ := newMirroredTestDB(t)
	payload := []byte("hash me")
	if err := db.putBlob("hashed", "", payload); err != nil {
		t.Fatal(err)
	}
	var sha string
	if err := db.sql.QueryRow("SELECT sha256 FROM blobs WHERE id = 'hashed'").Scan(&sha); err != nil {
		t.Fatal(err)
	}
	want := sha256.Sum256(payload)
	if sha != hex.EncodeToString(want[:]) {
		t.Fatalf("recorded hash = %.12s, want %.12s", sha, hex.EncodeToString(want[:]))
	}
}

// The fs.ErrNotExist contract survives composition: an id nobody holds errors
// identically through the mirrored store.
func TestMirroredStoreNotExist(t *testing.T) {
	db, _, _ := newMirroredTestDB(t)
	if _, _, _, err := db.blobs.open("ghost"); !errors.Is(err, fs.ErrNotExist) {
		t.Fatalf("open(ghost) = %v, want fs.ErrNotExist", err)
	}
}