package main

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strconv"
	"sync"
	"testing"
	"time"
)

// fakeS3 is just enough S3 for the store: path-style PUT/HEAD/GET(+Range)/
// DELETE over an in-memory map, counting requests per verb so tests can pin
// the request economy (#836's lesson: requests ARE the bill).
type fakeS3 struct {
	mu      sync.Mutex
	objects map[string][]byte // full path-style key: /bucket/prefix/id
	gets    int
	puts    int
	heads   int
}

func (f *fakeS3) handler(w http.ResponseWriter, r *http.Request) {
	f.mu.Lock()
	defer f.mu.Unlock()
	switch r.Method {
	case http.MethodPut:
		f.puts++
		b, _ := io.ReadAll(r.Body)
		f.objects[r.URL.Path] = b
		w.WriteHeader(http.StatusOK)
	case http.MethodHead:
		f.heads++
		b, ok := f.objects[r.URL.Path]
		if !ok {
			w.WriteHeader(http.StatusNotFound)
			return
		}
		w.Header().Set("Content-Length", strconv.Itoa(len(b)))
		w.Header().Set("Last-Modified", time.Now().UTC().Format(http.TimeFormat))
		w.WriteHeader(http.StatusOK)
	case http.MethodGet:
		f.gets++
		b, ok := f.objects[r.URL.Path]
		if !ok {
			w.Header().Set("Content-Type", "application/xml")
			w.WriteHeader(http.StatusNotFound)
			fmt.Fprint(w, `<Error><Code>NoSuchKey</Code></Error>`)
			return
		}
		if rng := r.Header.Get("Range"); rng != "" {
			var start, end int
			if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &start, &end); err == nil {
				if end >= len(b) {
					end = len(b) - 1
				}
				w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(b)))
				w.WriteHeader(http.StatusPartialContent)
				w.Write(b[start : end+1])
				return
			}
		}
		w.Write(b)
	case http.MethodDelete:
		delete(f.objects, r.URL.Path)
		w.WriteHeader(http.StatusNoContent)
	default:
		w.WriteHeader(http.StatusNotImplemented)
	}
}

// newTestS3Store wires an s3BlobStore at a fake endpoint over a real (temp)
// disk store — the same construction openDB performs when the env is set.
func newTestS3Store(t *testing.T) (*s3BlobStore, *fakeS3, *diskBlobStore) {
	t.Helper()
	fake := &fakeS3{objects: map[string][]byte{}}
	srv := httptest.NewServer(http.HandlerFunc(fake.handler))
	t.Cleanup(srv.Close)

	// Static credentials + region so the SDK's signer is satisfied offline.
	t.Setenv("AWS_ACCESS_KEY_ID", "test")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
	t.Setenv("AWS_REGION", "auto")

	disk, err := newDiskBlobStore(filepath.Join(t.TempDir(), "blobs"))
	if err != nil {
		t.Fatal(err)
	}
	store, err := newS3BlobStore(&s3BlobConfig{
		Bucket:   "test-bucket",
		Endpoint: srv.URL,
	}, disk.dir, disk, "space.example.com")
	if err != nil {
		t.Fatal(err)
	}
	return store, fake, disk
}

// The blobStore contract (TestDiskBlobStoreRoundTrip's sequence) against
// implementation #2. Any drift between the two implementations here is a bug
// in the new one — the contract came first.
func TestS3BlobStoreRoundTrip(t *testing.T) {
	store, fake, _ := newTestS3Store(t)
	payload := []byte("pretend-this-is-encrypted-bytes\x00\x01\x02")

	w, err := store.create()
	if err != nil {
		t.Fatalf("create: %v", err)
	}
	if _, err := w.Write(payload[:10]); err != nil {
		t.Fatalf("write: %v", err)
	}
	// Staged bytes are invisible until commit — S3 empty, fallback empty.
	if _, _, _, err := store.open("blobid_123-abc"); !errors.Is(err, fs.ErrNotExist) {
		t.Fatalf("open before commit: want fs.ErrNotExist, got %v", err)
	}
	if _, err := w.Write(payload[10:]); err != nil {
		t.Fatalf("write: %v", err)
	}
	if err := w.commit("blobid_123-abc"); err != nil {
		t.Fatalf("commit: %v", err)
	}
	if fake.puts != 1 {
		t.Fatalf("puts = %d, want exactly 1 (the commit)", fake.puts)
	}
	// The object landed under the derived per-instance prefix.
	if _, ok := fake.objects["/test-bucket/space.example.com/blobid_123-abc"]; !ok {
		t.Fatalf("object key wrong; have %v", keysOf(fake))
	}

	r, size, modTime, err := store.open("blobid_123-abc")
	if err != nil {
		t.Fatalf("open: %v", err)
	}
	got, err := io.ReadAll(r)
	r.Close()
	if err != nil || !bytes.Equal(got, payload) {
		t.Fatalf("round-trip mismatch: err=%v got=%q", err, got)
	}
	if size != int64(len(payload)) {
		t.Fatalf("size = %d, want %d", size, len(payload))
	}
	if modTime.IsZero() {
		t.Fatal("zero modtime")
	}

	if err := store.remove("blobid_123-abc"); err != nil {
		t.Fatalf("remove: %v", err)
	}
	if _, _, _, err := store.open("blobid_123-abc"); !errors.Is(err, fs.ErrNotExist) {
		t.Fatalf("open after remove: want fs.ErrNotExist, got %v", err)
	}
	if err := store.remove("blobid_123-abc"); err != nil {
		t.Fatalf("remove of missing id: %v", err)
	}
}

func keysOf(f *fakeS3) []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := make([]string, 0, len(f.objects))
	for k := range f.objects {
		out = append(out, k)
	}
	return out
}

// An aborted staged write publishes nothing and leaves no spool file behind.
func TestS3BlobStoreAbortDiscards(t *testing.T) {
	store, fake, disk := newTestS3Store(t)
	w, err := store.create()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := w.Write([]byte("half an upload")); err != nil {
		t.Fatal(err)
	}
	w.abort()
	if len(keysOf(fake)) != 0 {
		t.Fatalf("abort published: %v", keysOf(fake))
	}
	entries, err := filepath.Glob(filepath.Join(disk.dir, "*"))
	if err != nil || len(entries) != 0 {
		t.Fatalf("spool not cleaned: %v %v", entries, err)
	}
}

// Pre-flip blobs live on disk; the s3 store serves and reaps them through its
// fallback, so a flipped instance never strands its history.
func TestS3BlobStoreFallbackServesDiskBlobs(t *testing.T) {
	store, _, disk := newTestS3Store(t)
	w, err := disk.create()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := w.Write([]byte("pre-flip bytes")); err != nil {
		t.Fatal(err)
	}
	if err := w.commit("oldblob"); err != nil {
		t.Fatal(err)
	}

	r, size, _, err := store.open("oldblob")
	if err != nil {
		t.Fatalf("open via fallback: %v", err)
	}
	got, _ := io.ReadAll(r)
	r.Close()
	if string(got) != "pre-flip bytes" || size != 14 {
		t.Fatalf("fallback read = %q (%d)", got, size)
	}

	if err := store.remove("oldblob"); err != nil {
		t.Fatalf("remove via fallback: %v", err)
	}
	if _, _, _, err := disk.open("oldblob"); !errors.Is(err, fs.ErrNotExist) {
		t.Fatalf("disk bytes survived remove: %v", err)
	}
}

// The reader's request economy: a size probe costs nothing, a scrub to the
// middle of a video costs ONE window fetch, and a sequential read costs
// ceil(size/window) — never a request per 32KB ServeContent slice.
func TestS3ReaderRangedWindows(t *testing.T) {
	store, fake, _ := newTestS3Store(t)
	payload := bytes.Repeat([]byte("0123456789abcdef"), (s3ReadChunk*3)/16/2) // 1.5 windows
	w, _ := store.create()
	w.Write(payload)
	if err := w.commit("video"); err != nil {
		t.Fatal(err)
	}

	r, size, _, err := store.open("video")
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()
	if size != int64(len(payload)) {
		t.Fatalf("size = %d, want %d", size, len(payload))
	}

	// http.ServeContent's size probe: seeks only — zero GETs.
	base := fake.gets
	if _, err := r.Seek(0, io.SeekEnd); err != nil {
		t.Fatal(err)
	}
	if _, err := r.Seek(0, io.SeekStart); err != nil {
		t.Fatal(err)
	}
	if fake.gets != base {
		t.Fatalf("seeks cost %d GETs", fake.gets-base)
	}

	// A Range request landing mid-object (the video scrub): one window.
	if _, err := r.Seek(int64(s3ReadChunk+100), io.SeekStart); err != nil {
		t.Fatal(err)
	}
	slice := make([]byte, 64<<10)
	if _, err := io.ReadFull(r, slice); err != nil {
		t.Fatal(err)
	}
	if fake.gets-base != 1 {
		t.Fatalf("mid-object 64KB read cost %d GETs, want 1", fake.gets-base)
	}
	if !bytes.Equal(slice, payload[s3ReadChunk+100:s3ReadChunk+100+64<<10]) {
		t.Fatal("scrub read returned wrong bytes")
	}

	// Full sequential read from the top: exactly ceil(1.5 windows) = 2 GETs
	// (the mid-object window above already cached one of them — reset first).
	r2, _, _, err := store.open("video")
	if err != nil {
		t.Fatal(err)
	}
	defer r2.Close()
	base = fake.gets
	got, err := io.ReadAll(r2)
	if err != nil || !bytes.Equal(got, payload) {
		t.Fatalf("sequential read mismatch: %v", err)
	}
	if fake.gets-base != 2 {
		t.Fatalf("sequential 1.5-window read cost %d GETs, want 2", fake.gets-base)
	}
}

// The env switch: unset = disk store untouched; set = s3 store with the
// prefix derived from the DB basename when none is named.
func TestS3BlobConfigFromEnv(t *testing.T) {
	t.Setenv("LCHAT_BLOB_S3_BUCKET", "")
	if cfg := s3BlobConfigFromEnv(); cfg != nil {
		t.Fatalf("empty bucket produced config %+v", cfg)
	}
	t.Setenv("LCHAT_BLOB_S3_BUCKET", "media")
	t.Setenv("LCHAT_BLOB_S3_PREFIX", "/custom/")
	cfg := s3BlobConfigFromEnv()
	if cfg == nil || cfg.Bucket != "media" || cfg.Prefix != "custom" {
		t.Fatalf("config = %+v", cfg)
	}
}