package dbdoor

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

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

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

var _ litestream.ReplicaClient = (*Client)(nil)

// noNetwork fails the test if the client ever reaches for the wire.
func noNetwork(t *testing.T) Doer {
	return func(context.Context, string, http.Header) (*http.Response, error) {
		t.Error("the client made a request")
		return nil, errors.New("no network in this test")
	}
}

// The claim the whole design rests on: this client cannot corrupt a
// replica, because it has no code that writes to one. And it refuses
// without touching the network.
func TestClientMutatingMethodsAlwaysError(t *testing.T) {
	c := NewClient(noNetwork(t))
	ctx := context.Background()
	if _, err := c.WriteLTXFile(ctx, 0, 1, 2, strings.NewReader("x")); err == nil {
		t.Error("WriteLTXFile returned nil; this client must never write")
	}
	if err := c.DeleteLTXFiles(ctx, []*ltx.FileInfo{{Level: 0, MinTXID: 1, MaxTXID: 1}}); err == nil {
		t.Error("DeleteLTXFiles returned nil; this client must never delete")
	}
	if err := c.DeleteAll(ctx); err == nil {
		t.Error("DeleteAll returned nil; this client must never delete")
	}
}

// Restore type-asserts for ReplicaClientV3 and would take the v0.3 path if
// this satisfied it. It must not.
func TestClientIsNotAV3ReplicaClient(t *testing.T) {
	var c any = NewClient(noNetwork(t))
	if _, ok := c.(litestream.ReplicaClientV3); ok {
		t.Fatal("client satisfies ReplicaClientV3; Restore would take the v0.3 path")
	}
}

// testDoor serves a file replica over the door's wire shape: the same
// coordinates, the same Range forms, the same 404 for a missing file. It
// stands in for the console here; internal/console tests the real door
// with this package's client.
type testDoor struct {
	t      *testing.T
	rc     *lsfile.ReplicaClient
	mu     sync.Mutex
	lists  map[int]int // level -> listing count
	gone   map[string]bool
	opened int
}

func newTestDoor(t *testing.T, dir string) *testDoor {
	return &testDoor{t: t, rc: lsfile.NewReplicaClient(dir), lists: map[int]int{}, gone: map[string]bool{}}
}

func (d *testDoor) do(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
	rec := httptest.NewRecorder()
	d.serve(ctx, rec, tail, h)
	return rec.Result(), nil
}

func (d *testDoor) serve(ctx context.Context, w http.ResponseWriter, tail string, h http.Header) {
	parts := strings.Split(strings.TrimPrefix(tail, "/ltx/"), "/")
	level, err := strconv.Atoi(parts[0])
	if err != nil {
		http.NotFound(w, nil)
		return
	}
	if len(parts) == 1 {
		d.mu.Lock()
		d.lists[level]++
		d.mu.Unlock()
		itr, _ := d.rc.LTXFiles(ctx, level, 0, false)
		out := ListResponse{Files: []LTXFile{}}
		for itr.Next() {
			f := itr.Item()
			out.Files = append(out.Files, LTXFile{Level: level, MinTXID: uint64(f.MinTXID), MaxTXID: uint64(f.MaxTXID), Size: f.Size, CreatedAt: f.CreatedAt})
		}
		itr.Close()
		json.NewEncoder(w).Encode(out)
		return
	}
	min, max, err := ParseSpan(parts[1])
	if err != nil {
		http.NotFound(w, nil)
		return
	}
	d.mu.Lock()
	gone := d.gone[parts[1]]
	if gone {
		delete(d.gone, parts[1]) // gone once: a fresh plan finds it again
	}
	d.opened++
	d.mu.Unlock()
	if gone {
		http.NotFound(w, nil)
		return
	}
	off, size, err := ParseRange(h.Get("Range"))
	if err != nil {
		w.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
		return
	}
	rc, err := d.rc.OpenLTXFile(ctx, level, ltx.TXID(min), ltx.TXID(max), off, size)
	if errors.Is(err, os.ErrNotExist) {
		http.NotFound(w, nil)
		return
	}
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	defer rc.Close()
	w.WriteHeader(http.StatusPartialContent)
	io.Copy(w, rc)
}

// counting wraps the client and counts every method litestream calls on it.
type counting struct {
	*Client
	mu     sync.Mutex
	counts map[string]int
}

func (c *counting) inc(m string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.counts[m]++
}

func (c *counting) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
	c.inc("WriteLTXFile")
	return c.Client.WriteLTXFile(ctx, level, min, max, r)
}

func (c *counting) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error {
	c.inc("DeleteLTXFiles")
	return c.Client.DeleteLTXFiles(ctx, a)
}

func (c *counting) DeleteAll(ctx context.Context) error {
	c.inc("DeleteAll")
	return c.Client.DeleteAll(ctx)
}

func (c *counting) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
	c.inc("LTXFiles")
	return c.Client.LTXFiles(ctx, level, seek, useMetadata)
}

func (c *counting) OpenLTXFile(ctx context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	c.inc("OpenLTXFile")
	return c.Client.OpenLTXFile(ctx, level, min, max, offset, size)
}

// The methods erroring is not enough: a restore that CALLED one would be
// relying on an error path. This asserts a full restore never reaches them,
// which is what catches a litestream upgrade changing the restore path.
func TestRestoreNeverCallsAMutatingMethod(t *testing.T) {
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	door := newTestDoor(t, dir)
	c := &counting{Client: NewClient(door.do), counts: map[string]int{}}
	out := filepath.Join(t.TempDir(), "out.db")
	opt := litestream.NewRestoreOptions()
	opt.OutputPath = out
	if err := litestream.NewReplicaWithClient(nil, c).Restore(context.Background(), opt); err != nil {
		t.Fatal(err)
	}
	if n := c.counts["WriteLTXFile"] + c.counts["DeleteLTXFiles"] + c.counts["DeleteAll"]; n != 0 {
		t.Fatalf("restore called a mutating method: %v", c.counts)
	}
	if c.counts["LTXFiles"] == 0 || c.counts["OpenLTXFile"] == 0 {
		t.Fatalf("restore did not go through the client at all: %v", c.counts)
	}
	if rows := countRows(t, out); rows != 3*dbdoortest.Rows {
		t.Fatalf("restored %d rows, want %d", rows, 3*dbdoortest.Rows)
	}
}

func countRows(t *testing.T, path string) int {
	t.Helper()
	db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	var n int
	if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

// seek is not on the wire (wire specific 3): the client filters locally,
// with litestream's own rule (MinTXID >= seek).
func TestClientFiltersSeekLocally(t *testing.T) {
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	c := NewClient(newTestDoor(t, dir).do)
	all := collect(t, c, 0, 0)
	if len(all) < 2 {
		t.Fatalf("want several files, got %d", len(all))
	}
	seek := all[1].MinTXID
	got := collect(t, c, 0, seek)
	if len(got) != len(all)-1 || got[0].MinTXID != seek {
		t.Fatalf("seek %d = %d files starting %d, want %d starting %d", seek, len(got), got[0].MinTXID, len(all)-1, seek)
	}
}

func collect(t *testing.T, c *Client, level int, seek ltx.TXID) []*ltx.FileInfo {
	t.Helper()
	itr, err := c.LTXFiles(context.Background(), level, seek, false)
	if err != nil {
		t.Fatal(err)
	}
	defer itr.Close()
	var out []*ltx.FileInfo
	for itr.Next() {
		out = append(out, itr.Item())
	}
	return out
}

// A 404 on a file is os.ErrNotExist, which is what the whole-Restore retry
// keys on; any other refusal carries the console's own sentence.
func TestClientMapsTheDoorsAnswers(t *testing.T) {
	answer := func(status int, body string) Doer {
		return func(context.Context, string, http.Header) (*http.Response, error) {
			rec := httptest.NewRecorder()
			rec.WriteHeader(status)
			io.WriteString(rec, body)
			return rec.Result(), nil
		}
	}
	ctx := context.Background()
	if _, err := NewClient(answer(404, "404 page not found\n")).OpenLTXFile(ctx, 0, 1, 1, 0, 0); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("open on 404 = %v, want os.ErrNotExist", err)
	}
	_, err := NewClient(answer(409, `{"error":"the data is elsewhere"}`)).LTXFiles(ctx, 0, 0, false)
	var de *DoorError
	if !errors.As(err, &de) || de.Status != 409 || de.Message != "the data is elsewhere" {
		t.Errorf("list on 409 = %v, want a DoorError carrying the sentence", err)
	}
	if _, err := NewClient(answer(409, `{"error":"x"}`)).OpenLTXFile(ctx, 0, 1, 1, 0, 0); errors.Is(err, os.ErrNotExist) || !errors.As(err, &de) {
		t.Errorf("open on 409 = %v, want a DoorError and not ErrNotExist", err)
	}
}

// The client sends the S3 client's exact Range header and the coordinate
// tail, nothing else.
func TestClientSendsCoordinatesAndRange(t *testing.T) {
	var gotTail, gotRange string
	c := NewClient(func(_ context.Context, tail string, h http.Header) (*http.Response, error) {
		gotTail, gotRange = tail, h.Get("Range")
		rec := httptest.NewRecorder()
		rec.WriteHeader(http.StatusPartialContent)
		return rec.Result(), nil
	})
	rc, err := c.OpenLTXFile(context.Background(), 3, 16, 32, 512, 100)
	if err != nil {
		t.Fatal(err)
	}
	rc.Close()
	if gotTail != GetTail(3, 16, 32) || gotRange != RangeHeader(512, 100) {
		t.Fatalf("tail %q range %q, want %q %q", gotTail, gotRange, GetTail(3, 16, 32), RangeHeader(512, 100))
	}
}

// Retry-After is capped at a minute however large the door's number, and a
// number too big to multiply into a Duration must not wrap round to a
// short or negative wait.
func TestClientCapsRetryAfter(t *testing.T) {
	for _, v := range []string{"61", "3600", "18446744074", "9223372036", "9223372037"} {
		do := func(context.Context, string, http.Header) (*http.Response, error) {
			rec := httptest.NewRecorder()
			rec.Header().Set("Retry-After", v)
			rec.WriteHeader(http.StatusTooManyRequests)
			io.WriteString(rec, `{"error":"busy"}`)
			return rec.Result(), nil
		}
		_, err := NewClient(do).OpenLTXFile(context.Background(), 0, 1, 1, 0, 0)
		var de *DoorError
		if !errors.As(err, &de) || de.RetryAfter != time.Minute {
			t.Errorf("Retry-After: %s gave %v, want the one-minute cap", v, de.RetryAfter)
		}
	}
}