package dbdoor

import (
	"context"
	"errors"
	"io"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

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

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

// metaSpy records every useMetadata the listing passes down: the box must
// list LastModified as CreatedAt, exactly as the console's door does
// (listReplicaLevel), never S3 metadata timestamps (spec §6, T6).
type metaSpy struct {
	ReplicaReader
	sawMetadata bool
}

func (m *metaSpy) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
	if useMetadata {
		m.sawMetadata = true
	}
	return m.ReplicaReader.LTXFiles(ctx, level, seek, useMetadata)
}

func TestListLevelIsTheDoorsListing(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	spy := &metaSpy{ReplicaReader: lsfile.NewReplicaClient(dir)}
	got, err := ListLevel(context.Background(), spy, 0)
	if err != nil {
		t.Fatal(err)
	}
	want := collect(t, NewClient(newTestDoor(t, dir).do), 0, 0)
	if len(got.Files) != len(want) || spy.sawMetadata {
		t.Fatalf("ListLevel = %d files (metadata %v), the door lists %d", len(got.Files), spy.sawMetadata, len(want))
	}
	for i := range want {
		f := got.Files[i]
		if f.MinTXID != uint64(want[i].MinTXID) || f.MaxTXID != uint64(want[i].MaxTXID) || f.Size != want[i].Size || !f.CreatedAt.Equal(want[i].CreatedAt) {
			t.Fatalf("file %d: %+v, the door lists %+v", i, f, want[i])
		}
	}
	empty, err := ListLevel(context.Background(), spy, 5)
	if err != nil || empty.Files == nil || len(empty.Files) != 0 {
		t.Fatalf("an empty level = %+v, %v; want an empty list", empty, err)
	}
}

// PlanFrom plans what RestoreAsOf through the door restores: the same
// transaction, the same time, the same size, for the newest copy and as of
// a time.
func TestPlanFromMatchesTheDoorsRestore(t *testing.T) {
	dir := t.TempDir()
	src := dbdoortest.Replicate(t, dir, 2)
	at := pause()
	addBatch(t, src, dir, 2)
	for _, when := range []time.Time{{}, at} {
		plan, err := PlanFrom(context.Background(), lsfile.NewReplicaClient(dir), when)
		if err != nil {
			t.Fatal(err)
		}
		got, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), when)
		if err != nil {
			t.Fatal(err)
		}
		st, err := os.Stat(got.Path)
		if err != nil {
			t.Fatal(err)
		}
		os.RemoveAll(got.Dir)
		if plan.TXID != got.TXID || !plan.At.Equal(got.At) || plan.RestoredBytes != st.Size() || len(plan.Planned) != plan.Files || plan.Files == 0 {
			t.Fatalf("as of %v: plan %+v; the door restored txid %d at %v, %d bytes", when, plan, got.TXID, got.At, st.Size())
		}
	}
}

func TestPlanFromBeforeHistoryNamesTheOldestPoint(t *testing.T) {
	t.Parallel()
	before := time.Now().Add(-time.Hour)
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 2)
	_, err := PlanFrom(context.Background(), lsfile.NewReplicaClient(dir), before)
	var bh *BeforeHistoryError
	if !errors.As(err, &bh) {
		t.Fatalf("PlanFrom before history = %v", err)
	}
	_, derr := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), before)
	var dbh *BeforeHistoryError
	if !errors.As(derr, &dbh) || !dbh.Oldest.Equal(bh.Oldest) {
		t.Fatalf("the door says %v, the plan says %v", derr, bh.Oldest)
	}
}

func TestPlanFromAnEmptyReplica(t *testing.T) {
	t.Parallel()
	_, err := PlanFrom(context.Background(), lsfile.NewReplicaClient(t.TempDir()), time.Time{})
	if !errors.Is(err, ErrNothingReplicated) {
		t.Fatalf("PlanFrom on nothing = %v", err)
	}
}

// RestoreLocal restores exactly the planned files, copied into a fresh file
// replica the way the worker writes them, to the plan's transaction.
func TestRestoreLocalRestoresThePlan(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 4)
	src := lsfile.NewReplicaClient(dir)
	plan, err := PlanFrom(context.Background(), src, time.Time{})
	if err != nil {
		t.Fatal(err)
	}
	root := t.TempDir()
	for _, f := range plan.Planned {
		rc, err := src.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
		if err != nil {
			t.Fatal(err)
		}
		path := litestream.LTXFilePath(root, f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID))
		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
			t.Fatal(err)
		}
		out, err := os.Create(path)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := io.Copy(out, rc); err != nil {
			t.Fatal(err)
		}
		rc.Close()
		out.Close()
	}
	db := filepath.Join(t.TempDir(), "copy.db")
	if err := RestoreLocal(context.Background(), lsfile.NewReplicaClient(root), db, plan.TXID); err != nil {
		t.Fatal(err)
	}
	if rows := countRows(t, db); rows != 4*dbdoortest.Rows {
		t.Fatalf("restored %d rows", rows)
	}
	st, _ := os.Stat(db)
	if st.Size() != plan.RestoredBytes {
		t.Fatalf("restored %d bytes, planned %d", st.Size(), plan.RestoredBytes)
	}
}

// countingReads is a litestream.ReplicaClient over a file replica that
// cancels a context on the first byte read and counts every read made after
// it, so a test can see whether a restore keeps reading once cancelled.
type countingReads struct {
	litestream.ReplicaClient
	cancel context.CancelFunc
	mu     sync.Mutex
	reads  int
	after  int
}

func (c *countingReads) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, off, size int64) (io.ReadCloser, error) {
	rc, err := c.ReplicaClient.OpenLTXFile(ctx, level, lo, hi, off, size)
	if err != nil {
		return nil, err
	}
	return &countedReader{ReadCloser: rc, c: c}, nil
}

type countedReader struct {
	io.ReadCloser
	c *countingReads
}

func (r *countedReader) Read(p []byte) (int, error) {
	r.c.mu.Lock()
	r.c.reads++
	if r.c.reads == 1 {
		r.c.cancel()
	} else {
		r.c.after++
	}
	r.c.mu.Unlock()
	return r.ReadCloser.Read(p)
}

// Astra round 1 on plan 3, finding 2: litestream's local compaction checks
// no context, so RestoreLocal's replica reads must: once the restore is
// cancelled (Ctrl-C, or the read's deadline), no further byte is read and
// nothing is left at out. Mutation: RestoreLocal reads the client without
// its context wrapper (red: reads continue after the cancel).
func TestRestoreLocalStopsReadingOnceCancelled(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 4)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	c := &countingReads{ReplicaClient: lsfile.NewReplicaClient(dir), cancel: cancel}
	plan, err := PlanFrom(context.Background(), lsfile.NewReplicaClient(dir), time.Time{})
	if err != nil {
		t.Fatal(err)
	}
	db := filepath.Join(t.TempDir(), "copy.db")
	err = RestoreLocal(ctx, c, db, plan.TXID)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("a cancelled restore answered %v", err)
	}
	if c.after != 0 {
		t.Fatalf("%d replica reads after the cancel", c.after)
	}
	if _, err := os.Stat(db); !os.IsNotExist(err) {
		t.Fatalf("something is at out: %v", err)
	}
}