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

package actor

import (
	"context"
	"os"
	"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)
		}
	}
}

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

	ctx := context.Background()
	msg := &cart_messages.AddItem{ItemId: 1, Quantity: 1}

	// Write mutations to create event logs for 1, 2, 3, 4
	ids := []uint64{1, 2, 3, 4}
	for _, id := range ids {
		if err := storage.AppendMutations(id, msg); err != nil {
			t.Fatalf("append %d: %v", id, err)
		}
	}

	// Force save to disk
	storage.save()

	// 1. Grain 1: Old modification time, inactive, no writer -> should be purged
	path1 := storage.logPath(1)
	oldTime := time.Now().Add(-10 * 24 * time.Hour)
	if err := os.Chtimes(path1, oldTime, oldTime); err != nil {
		t.Fatal(err)
	}

	// We must close the writer for grain 1, 2, 4 to make them eligible for purging (no open writer)
	storage.writersMu.Lock()
	for _, id := range []uint64{1, 2, 4} {
		if w, ok := storage.writers[id]; ok {
			w.purged = true
			w.file.Close()
			delete(storage.writers, id)
		}
	}
	storage.writersMu.Unlock()

	// 2. Grain 2: Old mod time, but marked active -> should NOT be purged
	path2 := storage.logPath(2)
	if err := os.Chtimes(path2, oldTime, oldTime); err != nil {
		t.Fatal(err)
	}

	// 3. Grain 3: Old mod time, but has open writer -> should NOT be purged
	path3 := storage.logPath(3)
	if err := os.Chtimes(path3, oldTime, oldTime); err != nil {
		t.Fatal(err)
	}

	// 4. Grain 4: Recent mod time -> should NOT be purged
	path4 := storage.logPath(4)

	activeGrains := map[uint64]bool{2: true}
	isGrainActive := func(id uint64) bool {
		return activeGrains[id]
	}

	purged, err := storage.PurgeOlderThan(ctx, 5*24*time.Hour, isGrainActive)
	if err != nil {
		t.Fatalf("PurgeOlderThan failed: %v", err)
	}

	if purged != 1 {
		t.Errorf("Expected exactly 1 purged file, got %d", purged)
	}

	// Verify file status
	if _, err := os.Stat(path1); !os.IsNotExist(err) {
		t.Error("Expected grain 1 log to be deleted")
	}
	if _, err := os.Stat(path2); err != nil {
		t.Error("Expected grain 2 log to exist (active)")
	}
	if _, err := os.Stat(path3); err != nil {
		t.Error("Expected grain 3 log to exist (active writer)")
	}
	if _, err := os.Stat(path4); err != nil {
		t.Error("Expected grain 4 log to exist (recent)")
	}
}