package store

import (
	"errors"
	"strings"
	"sync"
	"testing"
	"time"
)

// recorder is a persistence that records what it is asked to write and can be told to fail.
type recorder struct {
	mu      sync.Mutex
	items   map[string][]byte
	batches [][]Op
	writes  int
	// fail makes the next N writes fail.
	fail int
	// loads counts Load calls.
	loads int
	// loadErr is returned by Load when set.
	loadErr error
}

// newRecorder builds a recorder holding the given documents.
func newRecorder(items map[string][]byte) *recorder {
	if items == nil {
		items = map[string][]byte{}
	}

	return &recorder{items: items}
}

// Load implements Persistence.
func (r *recorder) Load() (map[string][]byte, error) {
	r.mu.Lock()
	defer r.mu.Unlock()

	r.loads++

	if r.loadErr != nil {
		return nil, r.loadErr
	}

	out := make(map[string][]byte, len(r.items))

	for key, value := range r.items {
		out[key] = value
	}

	return out, nil
}

// Write implements Persistence.
func (r *recorder) Write(batch []Op) error {
	r.mu.Lock()
	defer r.mu.Unlock()

	r.writes++

	if r.fail > 0 {
		r.fail--

		return errors.New("the store is unavailable")
	}

	copied := make([]Op, len(batch))

	for i, op := range batch {
		copied[i] = Op{Key: op.Key, Delete: op.Delete, Value: append([]byte{}, op.Value...)}

		if op.Delete {
			delete(r.items, op.Key)
		} else {
			r.items[op.Key] = append([]byte{}, op.Value...)
		}
	}

	r.batches = append(r.batches, copied)

	return nil
}

// snapshot returns a copy of what the store holds.
func (r *recorder) snapshot() map[string][]byte {
	r.mu.Lock()
	defer r.mu.Unlock()

	out := make(map[string][]byte, len(r.items))

	for key, value := range r.items {
		out[key] = value
	}

	return out
}

// batchCount is how many batches have been written.
func (r *recorder) batchCount() int {
	r.mu.Lock()
	defer r.mu.Unlock()

	return len(r.batches)
}

// open builds a manual-flush cache over a recorder and closes it at the end of the test.
//
// Manual, because a background flush would make a write-count assertion a race: these tests
// count the round trips, so they take the flushes into their own hands.
func open(t *testing.T, persist Persistence) *Cache {
	t.Helper()

	cache, err := OpenManual(persist)

	if err != nil {
		t.Fatal(err)
	}

	t.Cleanup(func() { _ = cache.Close() })

	return cache
}

// TestReadsAndWritesAreMemory: the hot path never touches the persistence.
//
// This is the property the whole package exists for: a wallet matching a filter batch writes
// dozens of times in a row, and a thread-hop per write is the difference between a sync that
// keeps up and one that does not.
func TestReadsAndWritesAreMemory(t *testing.T) {
	persist := newRecorder(nil)
	cache, err := OpenManual(persist)

	if err != nil {
		t.Fatal(err)
	}

	// Manual flushing, so no background write can land between the writes and the count.
	for i := 0; i < 200; i++ {
		if err := cache.Put("addr/"+itoa(i), []byte("value")); err != nil {
			t.Fatal(err)
		}

		if value, ok := cache.Get("addr/" + itoa(i)); !ok || string(value) != "value" {
			t.Fatalf("read back %q, %v", value, ok)
		}
	}

	if cache.Len() != 200 {
		t.Fatalf("the cache holds %d keys", cache.Len())
	}

	// A flush is one round trip for all of them, not one each.
	if err := cache.Flush(); err != nil {
		t.Fatal(err)
	}

	persist.mu.Lock()
	writes := persist.writes
	persist.mu.Unlock()

	if writes != 1 {
		t.Fatalf("%d writes for 200 changes", writes)
	}

	if err := cache.Close(); err != nil {
		t.Fatal(err)
	}

	if len(persist.snapshot()) != 200 {
		t.Fatalf("the store holds %d documents", len(persist.snapshot()))
	}
}

// TestCoalescing: many writes to one key are one document in one batch.
func TestCoalescing(t *testing.T) {
	persist := newRecorder(nil)
	cache := open(t, persist)

	for i := 0; i < 100; i++ {
		if err := cache.Put("utxo/one", []byte("value "+itoa(i))); err != nil {
			t.Fatal(err)
		}
	}

	if err := cache.Flush(); err != nil {
		t.Fatal(err)
	}

	if err := cache.Close(); err != nil {
		t.Fatal(err)
	}

	if persist.batchCount() != 1 {
		t.Fatalf("%d batches", persist.batchCount())
	}

	first := persist.batches[0]

	if len(first) != 1 || first[0].Key != "utxo/one" {
		t.Fatalf("the batch is %+v", first)
	}

	if string(first[0].Value) != "value 99" {
		t.Fatalf("the written value is %q", first[0].Value)
	}
}

// TestFailedWriteKeepsTheDataDirty: a store that is away loses nothing, and the next flush
// carries everything that changed while it was away.
func TestFailedWriteKeepsTheDataDirty(t *testing.T) {
	persist := newRecorder(nil)
	cache, err := OpenManual(persist)

	if err != nil {
		t.Fatal(err)
	}

	defer func() { _ = cache.Close() }()

	// The first write fails, so this change must survive.
	persist.mu.Lock()
	persist.fail = 1
	persist.mu.Unlock()

	if err := cache.Put("addr/one", []byte("first")); err != nil {
		t.Fatal(err)
	}

	if err := cache.Flush(); err == nil {
		t.Fatal("the failing store reported success")
	}

	if failures, last := cache.Failures(); failures != 1 || last == nil {
		t.Fatalf("failures = %d, %v", failures, last)
	}

	// A second change, then a flush that succeeds: both are written.
	if err := cache.Put("addr/two", []byte("second")); err != nil {
		t.Fatal(err)
	}

	if err := cache.Flush(); err != nil {
		t.Fatal(err)
	}

	if err := cache.Close(); err != nil {
		t.Fatal(err)
	}

	stored := persist.snapshot()

	if string(stored["addr/one"]) != "first" || string(stored["addr/two"]) != "second" {
		t.Fatalf("the store holds %v", stored)
	}
}

// TestDeletionIsWritten: a removal is an operation, not an absence.
func TestDeletionIsWritten(t *testing.T) {
	persist := newRecorder(map[string][]byte{
		"addr/gone": []byte("value"),
		"addr/kept": []byte("value"),
	})

	cache := open(t, persist)

	if _, ok := cache.Get("addr/gone"); !ok {
		t.Fatal("the loaded document is not there")
	}

	if err := cache.Delete("addr/gone"); err != nil {
		t.Fatal(err)
	}

	if _, ok := cache.Get("addr/gone"); ok {
		t.Fatal("the deleted document is still readable")
	}

	if err := cache.Close(); err != nil {
		t.Fatal(err)
	}

	stored := persist.snapshot()

	if _, ok := stored["addr/gone"]; ok {
		t.Fatal("the store still holds the deleted document")
	}

	if string(stored["addr/kept"]) != "value" {
		t.Fatal("the other document was lost")
	}
}

// TestReopenSeesWhatWasWritten: the store is what survives a restart.
func TestReopenSeesWhatWasWritten(t *testing.T) {
	persist := newRecorder(nil)

	first, err := OpenManual(persist)

	if err != nil {
		t.Fatal(err)
	}

	for _, key := range []string{"wallet/seed-id", "addr/0", "addr/1", "utxo/a"} {
		if err := first.Put(key, []byte(key)); err != nil {
			t.Fatal(err)
		}
	}

	if err := first.Close(); err != nil {
		t.Fatal(err)
	}

	second, err := OpenManual(persist)

	if err != nil {
		t.Fatal(err)
	}

	defer func() { _ = second.Close() }()

	if second.Len() != 4 {
		t.Fatalf("the reopened cache holds %d keys", second.Len())
	}

	keys := second.Keys("addr/")

	if len(keys) != 2 || keys[0] != "addr/0" || keys[1] != "addr/1" {
		t.Fatalf("keys = %v", keys)
	}
}

// TestLoadFailureIsAnError: a wallet that cannot read its store must say so rather than
// showing a zero balance for a funded wallet.
func TestLoadFailureIsAnError(t *testing.T) {
	persist := newRecorder(nil)
	persist.loadErr = errors.New("the database is corrupt")

	if _, err := Open(persist); err == nil {
		t.Fatal("a store that could not load was opened empty")
	}
}

// TestValuesAreCopied: a caller reusing its buffer cannot change what is stored or written.
func TestValuesAreCopied(t *testing.T) {
	persist := newRecorder(nil)
	cache := open(t, persist)

	buffer := []byte("original")

	if err := cache.Put("addr/one", buffer); err != nil {
		t.Fatal(err)
	}

	buffer[0] = 'X'

	value, _ := cache.Get("addr/one")

	if string(value) != "original" {
		t.Fatalf("the cached value is %q", value)
	}

	if err := cache.Close(); err != nil {
		t.Fatal(err)
	}

	if string(persist.snapshot()["addr/one"]) != "original" {
		t.Fatalf("the written value is %q", persist.snapshot()["addr/one"])
	}
}

// TestKeysAreSorted: iteration order is deterministic on every platform.
func TestKeysAreSorted(t *testing.T) {
	cache := open(t, newRecorder(nil))

	for _, key := range []string{"b", "a", "c", "aa"} {
		if err := cache.Put(key, nil); err != nil {
			t.Fatal(err)
		}
	}

	keys := cache.Keys("")

	if strings.Join(keys, ",") != "a,aa,b,c" {
		t.Fatalf("keys = %v", keys)
	}

	if got := cache.Keys("a"); strings.Join(got, ",") != "a,aa" {
		t.Fatalf("prefixed keys = %v", got)
	}
}

// TestBadKeysAreRefused: an empty or huge key is a caller's mistake, named here.
func TestBadKeysAreRefused(t *testing.T) {
	cache := open(t, newRecorder(nil))

	if err := cache.Put("", nil); err == nil {
		t.Fatal("an empty key was stored")
	}

	if err := cache.Delete(""); err == nil {
		t.Fatal("an empty key was deleted")
	}

	if err := cache.Put(strings.Repeat("k", 300), nil); err == nil {
		t.Fatal("a 300-byte key was stored")
	}
}

// TestAChangeDuringAFlushIsKept: the value written is not necessarily the value the cache
// ends up with, and the newer one must reach the store.
func TestAChangeDuringAFlushIsKept(t *testing.T) {
	persist := &slowRecorder{
		recorder: newRecorder(nil),
		started:  make(chan struct{}),
		release:  make(chan struct{}),
	}

	cache, err := OpenManual(persist)

	if err != nil {
		t.Fatal(err)
	}

	defer func() { _ = cache.Close() }()

	if err := cache.Put("addr/one", []byte("first")); err != nil {
		t.Fatal(err)
	}

	flushed := make(chan error, 1)

	go func() { flushed <- cache.Flush() }()

	// Wait until the write is in progress, then change the value.
	<-persist.started

	if err := cache.Put("addr/one", []byte("second")); err != nil {
		t.Fatal(err)
	}

	close(persist.release)

	if err := <-flushed; err != nil {
		t.Fatal(err)
	}

	if err := cache.Flush(); err != nil {
		t.Fatal(err)
	}

	if err := cache.Close(); err != nil {
		t.Fatal(err)
	}

	if got := string(persist.snapshot()["addr/one"]); got != "second" {
		t.Fatalf("the store holds %q, want the newer value", got)
	}
}

// slowRecorder blocks in Write until it is released, so a change can be made mid-flush.
type slowRecorder struct {
	*recorder
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

// Write implements Persistence.
func (s *slowRecorder) Write(batch []Op) error {
	s.once.Do(func() { close(s.started) })

	<-s.release

	return s.recorder.Write(batch)
}

// TestConcurrentWritersAndTheFlusher: memory operations and the background flush do not
// corrupt each other. It is the one test that wants -race.
func TestConcurrentWritersAndTheFlusher(t *testing.T) {
	// A real interval, because this test is about the flusher running under memory writes.
	cache, err := OpenWith(newRecorder(nil), time.Millisecond)

	if err != nil {
		t.Fatal(err)
	}

	defer func() { _ = cache.Close() }()

	var group sync.WaitGroup

	for worker := 0; worker < 4; worker++ {
		group.Add(1)

		go func(id int) {
			defer group.Done()

			for i := 0; i < 200; i++ {
				key := "addr/" + itoa(id) + "/" + itoa(i)

				if err := cache.Put(key, []byte("value")); err != nil {
					t.Errorf("put: %v", err)

					return
				}

				_ = cache.Keys("addr/")
				_, _ = cache.Get(key)
			}
		}(worker)
	}

	group.Wait()

	if err := cache.Flush(); err != nil {
		t.Fatal(err)
	}

	if cache.Len() != 800 {
		t.Fatalf("the cache holds %d keys", cache.Len())
	}
}

// itoa renders a small integer without pulling in formatting.
func itoa(value int) string {
	if value == 0 {
		return "0"
	}

	var digits []byte

	for value > 0 {
		digits = append([]byte{byte('0' + value%10)}, digits...)
		value /= 10
	}

	return string(digits)
}