go-cart-actor/pkg/actor/disk_storage_test.go

package actor

import (
	"context"
	"sync"
	"testing"
	"time"

	cart_messages "git.k6n.net/mats/go-cart-actor/proto/cart"
	"google.golang.org/protobuf/proto"
)

type diskTestGrain struct {
	testState
	id       uint64
	accessed time.Time
}

func (g *diskTestGrain) GetId() uint64            { return g.id }
func (g *diskTestGrain) GetLastAccess() time.Time { return g.accessed }
func (g *diskTestGrain) GetLastChange() time.Time { return g.accessed }
func (g *diskTestGrain) GetCurrentState() (*testState, error) {
	return &g.testState, nil
}

func diskTestRegistry(t *testing.T) MutationRegistry {
	t.Helper()
	reg := NewMutationRegistry()
	reg.RegisterMutations(NewMutation(
		func(_ *diskTestGrain, _ *cart_messages.AddItem) error { return nil },
	))
	return reg
}

func TestDiskStorageAppendAndReplay(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	const id = uint64(99)
	events := []proto.Message{
		&cart_messages.AddItem{ItemId: 1, Quantity: 2},
		&cart_messages.AddItem{ItemId: 2, Quantity: 1},
	}
	for _, evt := range events {
		if err := storage.AppendMutations(id, evt); err != nil {
			t.Fatalf("append: %v", err)
		}
	}

	grain := &diskTestGrain{id: id, accessed: time.Now()}
	if err := storage.LoadEvents(context.Background(), id, grain); err != nil {
		t.Fatalf("replay: %v", err)
	}
}

func TestDiskStorageReusesOpenWriter(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	const id = uint64(7)
	msg := &cart_messages.AddItem{ItemId: 1, Quantity: 1}
	if err := storage.AppendMutations(id, msg); err != nil {
		t.Fatal(err)
	}
	if got := storage.openWriterCount(); got != 1 {
		t.Fatalf("open writers after first append = %d, want 1", got)
	}
	if err := storage.AppendMutations(id, msg); err != nil {
		t.Fatal(err)
	}
	if got := storage.openWriterCount(); got != 1 {
		t.Fatalf("open writers after second append = %d, want 1", got)
	}
}

func TestDiskStorageIdleWriterClosed(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	storage.writerIdleTTL = 20 * time.Millisecond
	t.Cleanup(storage.Close)

	if err := storage.AppendMutations(1, &cart_messages.AddItem{ItemId: 1, Quantity: 1}); err != nil {
		t.Fatal(err)
	}
	if storage.openWriterCount() != 1 {
		t.Fatal("expected one open writer")
	}

	time.Sleep(30 * time.Millisecond)
	storage.purgeIdleWriters()
	if storage.openWriterCount() != 0 {
		t.Fatalf("expected idle writer to be closed, got %d open", storage.openWriterCount())
	}
}

func TestDiskStorageSaveLoopUsesWriterCache(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	go storage.SaveLoop(20 * time.Millisecond)

	const id = uint64(55)
	if err := storage.AppendMutations(id, &cart_messages.AddItem{ItemId: 3, Quantity: 4}); err != nil {
		t.Fatal(err)
	}

	deadline := time.After(500 * time.Millisecond)
	for {
		if storage.openWriterCount() >= 1 {
			break
		}
		select {
		case <-deadline:
			t.Fatal("timed out waiting for SaveLoop flush")
		default:
			time.Sleep(10 * time.Millisecond)
		}
	}

	grain := &diskTestGrain{id: id, accessed: time.Now()}
	if err := storage.LoadEvents(context.Background(), id, grain); err != nil {
		t.Fatalf("replay after SaveLoop: %v", err)
	}
}

func TestDiskStorageConcurrentAppendSameID(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	const id = uint64(88)
	const workers = 16
	var wg sync.WaitGroup
	wg.Add(workers)
	for i := range workers {
		go func(itemID uint32) {
			defer wg.Done()
			_ = storage.AppendMutations(id, &cart_messages.AddItem{ItemId: itemID, Quantity: 1})
		}(uint32(i + 1))
	}
	wg.Wait()

	grain := &diskTestGrain{id: id, accessed: time.Now()}
	if err := storage.LoadEvents(context.Background(), id, grain); err != nil {
		t.Fatalf("replay: %v", err)
	}
}

func TestDiskStorageCloseFlushesSaveLoop(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)

	// Start the buffered SaveLoop but give it a long interval so it does NOT
	// flush during the test — only Close() should trigger the flush. A brief
	// sleep lets the goroutine schedule and set s.queue before we append.
	go storage.SaveLoop(10 * time.Minute)
	time.Sleep(5 * time.Millisecond)

	const id = uint64(101)
	if err := storage.AppendMutations(id, &cart_messages.AddItem{ItemId: 7, Quantity: 3}); err != nil {
		t.Fatal(err)
	}

	// Verify event is queued but NOT yet on disk (no writer should be open
	// because the buffered path only opens files inside SaveLoop → save()).
	if got := storage.openWriterCount(); got != 0 {
		t.Fatalf("expected 0 open writers before flush, got %d", got)
	}

	// Close flushes the queue then closes writers.
	storage.Close()

	// After Close(), the event should be persisted. Open a fresh storage and
	// replay to verify.
	fresh := NewDiskStorage[testState](storage.path, reg)
	t.Cleanup(fresh.Close)
	grain := &diskTestGrain{id: id, accessed: time.Now()}
	if err := fresh.LoadEvents(context.Background(), id, grain); err != nil {
		t.Fatalf("replay after Close flush: %v", err)
	}
}

func TestDiskStorageMultipleGrainsIndependent(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	ids := []uint64{10, 20, 30}
	for i, id := range ids {
		evt := &cart_messages.AddItem{ItemId: uint32(i + 1), Quantity: int32(i + 1)}
		if err := storage.AppendMutations(id, evt); err != nil {
			t.Fatalf("append id=%d: %v", id, err)
		}
	}

	// Verify each grain has exactly one writer open (shared per id).
	if got := storage.openWriterCount(); got != len(ids) {
		t.Fatalf("open writers = %d, want %d", got, len(ids))
	}

	// Replay each grain and verify the correct event was stored.
	for _, id := range ids {
		ctx := context.Background()
		grain := &diskTestGrain{id: id, accessed: time.Now()}
		if err := storage.LoadEvents(ctx, id, grain); err != nil {
			t.Fatalf("replay id=%d: %v", id, err)
		}
	}
}

func TestDiskStorageLoadEventsNoFile(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	// No AppendMutations — no file should exist.
	grain := &diskTestGrain{id: 999, accessed: time.Now()}
	if err := storage.LoadEvents(context.Background(), 999, grain); err != nil {
		t.Fatalf("LoadEvents on non-existent file: %v", err)
	}
}

func TestDiskStorageWriteAfterIdlePurge(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	// Set a very short idle TTL so the writer is evicted quickly.
	storage.writerIdleTTL = 10 * time.Millisecond

	const id = uint64(202)
	msg := &cart_messages.AddItem{ItemId: 1, Quantity: 1}

	// First write opens a writer.
	if err := storage.AppendMutations(id, msg); err != nil {
		t.Fatal(err)
	}
	if got := storage.openWriterCount(); got != 1 {
		t.Fatalf("open writers after first write = %d, want 1", got)
	}

	// Wait for idle TTL to expire and purge.
	time.Sleep(20 * time.Millisecond)
	storage.purgeIdleWriters()

	if got := storage.openWriterCount(); got != 0 {
		t.Fatalf("open writers after purge = %d, want 0", got)
	}

	// Second write must create a new writer through the retry loop in getWriter.
	if err := storage.AppendMutations(id, msg); err != nil {
		t.Fatalf("write after purge: %v", err)
	}
	if got := storage.openWriterCount(); got != 1 {
		t.Fatalf("open writers after write-before-close = %d, want 1", got)
	}

	// Replay must see both events.
	grain := &diskTestGrain{id: id, accessed: time.Now()}
	if err := storage.LoadEvents(context.Background(), id, grain); err != nil {
		t.Fatalf("replay after purge: %v", err)
	}
}

func TestDiskStorageCloseWithoutSaveLoop(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)

	// Verify Close() on a storage that never started SaveLoop works cleanly.
	const id = uint64(303)
	if err := storage.AppendMutations(id, &cart_messages.AddItem{ItemId: 1, Quantity: 1}); err != nil {
		t.Fatal(err)
	}

	// Close() — queue is nil so it skips save(), goes straight to closeWriters().
	storage.Close()

	// A fresh storage should be able to replay the persisted event.
	fresh := NewDiskStorage[testState](storage.path, reg)
	t.Cleanup(fresh.Close)
	grain := &diskTestGrain{id: id, accessed: time.Now()}
	if err := fresh.LoadEvents(context.Background(), id, grain); err != nil {
		t.Fatalf("replay after close: %v", err)
	}
}

// TestDiskStorageConcurrentWritesDifferentIDs verifies that concurrent writes to
// different grains don't interfere with each other, and all events are persisted.
func TestDiskStorageConcurrentWritesDifferentIDs(t *testing.T) {
	reg := diskTestRegistry(t)
	storage := NewDiskStorage[testState](t.TempDir(), reg)
	t.Cleanup(storage.Close)

	const grainCount = 32
	var wg sync.WaitGroup
	wg.Add(grainCount)
	for id := range uint64(grainCount) {
		go func(grainID uint64) {
			defer wg.Done()
			for i := range 5 {
				_ = storage.AppendMutations(grainID, &cart_messages.AddItem{ItemId: uint32(i + 1), Quantity: 1})
			}
		}(id + 1)
	}
	wg.Wait()

	// Replay each grain and verify.
	for id := range uint64(grainCount) {
		grainID := id + 1
		ctx := context.Background()
		grain := &diskTestGrain{id: grainID, accessed: time.Now()}
		if err := storage.LoadEvents(ctx, grainID, grain); err != nil {
			t.Fatalf("replay grain %d: %v", grainID, err)
		}
	}
}