fathom/internal/store/store_test.go

package store

import (
	"fmt"
	"os"
	"path/filepath"
	"testing"
	"time"

	"git.z.gd/zzg/fathom/internal/model"
)

func testStore(t *testing.T, opts Options) *Store {
	t.Helper()
	if opts.Dir == "" {
		opts.Dir = t.TempDir()
	}
	// A long interval keeps the background goroutine from racing assertions;
	// tests that want a flush call Flush explicitly.
	if opts.FlushInterval == 0 {
		opts.FlushInterval = time.Hour
	}
	if opts.FlushRows == 0 {
		opts.FlushRows = 1 << 20
	}
	s, err := Open(opts)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { s.Close() })
	return s
}

func spanAt(id string, ts int64) model.Span {
	return model.Span{
		TraceID: "t-" + id, SpanID: id, Name: "span", Service: "svc",
		StartTS: ts, EndTS: ts + 1000, SampleRate: 1,
	}
}

func TestStoreWriteThenRead(t *testing.T) {
	s := testStore(t, Options{})
	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	if err := s.AddSpans([]model.Span{spanAt("a", base), spanAt("b", base+1000)}); err != nil {
		t.Fatal(err)
	}
	// Nothing is on disk until a flush; that is the buffering contract.
	if shards, _ := s.ListShards(model.TableSpans); len(shards) != 0 {
		t.Fatalf("found %d shards before flushing", len(shards))
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	shards, err := s.ListShards(model.TableSpans)
	if err != nil {
		t.Fatal(err)
	}
	if len(shards) != 1 || shards[0].Date != "2026-09-15" {
		t.Fatalf("got shards %+v, want one dated 2026-09-15", shards)
	}

	sh, err := s.OpenShardForRead(model.TableSpans, "2026-09-15")
	if err != nil {
		t.Fatal(err)
	}
	if got := countRows(t, sh); got != 2 {
		t.Errorf("read %d rows, wrote 2", got)
	}
}

// TestFlushSplitsAcrossMidnight covers a batch that straddles a shard boundary.
// Assuming one shard per flush would silently file the late rows in the wrong
// day, which would then make them invisible to a correctly-ranged query.
func TestFlushSplitsAcrossMidnight(t *testing.T) {
	s := testStore(t, Options{})
	before := time.Date(2026, 9, 15, 23, 59, 59, 0, time.UTC).UnixMicro()
	after := time.Date(2026, 9, 16, 0, 0, 1, 0, time.UTC).UnixMicro()

	if err := s.AddSpans([]model.Span{spanAt("a", before), spanAt("b", after)}); err != nil {
		t.Fatal(err)
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	shards, _ := s.ListShards(model.TableSpans)
	if len(shards) != 2 {
		t.Fatalf("got %d shards, want 2 (one per side of midnight)", len(shards))
	}
	for _, sh := range shards {
		h, _ := s.OpenShardForRead(model.TableSpans, sh.Date)
		if got := countRows(t, h); got != 1 {
			t.Errorf("shard %s holds %d rows, want 1", sh.Date, got)
		}
	}
}

func TestFlushRowsTriggersFlush(t *testing.T) {
	s := testStore(t, Options{FlushRows: 3})
	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	for i := 0; i < 2; i++ {
		if err := s.AddSpans([]model.Span{spanAt(fmt.Sprint(i), base)}); err != nil {
			t.Fatal(err)
		}
	}
	if shards, _ := s.ListShards(model.TableSpans); len(shards) != 0 {
		t.Fatal("flushed before reaching the row threshold")
	}
	if err := s.AddSpans([]model.Span{spanAt("2", base)}); err != nil {
		t.Fatal(err)
	}
	sh, err := s.OpenShardForRead(model.TableSpans, "2026-09-15")
	if err != nil {
		t.Fatal(err)
	}
	if got := countRows(t, sh); got != 3 {
		t.Errorf("after crossing the threshold %d rows are durable, want 3", got)
	}
}

func TestShardsOverlapping(t *testing.T) {
	s := testStore(t, Options{})
	for _, d := range []int{13, 14, 15, 16} {
		ts := time.Date(2026, 9, d, 12, 0, 0, 0, time.UTC).UnixMicro()
		if err := s.AddSpans([]model.Span{spanAt(fmt.Sprint(d), ts)}); err != nil {
			t.Fatal(err)
		}
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	cases := []struct {
		name      string
		from, to  time.Time
		wantDates []string
	}{
		{
			"one day", time.Date(2026, 9, 14, 0, 0, 0, 0, time.UTC),
			time.Date(2026, 9, 14, 23, 59, 59, 0, time.UTC), []string{"2026-09-14"},
		},
		{
			"spanning two", time.Date(2026, 9, 14, 6, 0, 0, 0, time.UTC),
			time.Date(2026, 9, 15, 6, 0, 0, 0, time.UTC), []string{"2026-09-14", "2026-09-15"},
		},
		{
			"everything", time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC),
			time.Date(2026, 9, 30, 0, 0, 0, 0, time.UTC),
			[]string{"2026-09-13", "2026-09-14", "2026-09-15", "2026-09-16"},
		},
		{
			"no overlap", time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC),
			time.Date(2026, 8, 2, 0, 0, 0, 0, time.UTC), nil,
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			got, err := s.ShardsOverlapping(model.TableSpans, tc.from.UnixMicro(), tc.to.UnixMicro())
			if err != nil {
				t.Fatal(err)
			}
			var dates []string
			for _, g := range got {
				dates = append(dates, g.Date)
			}
			if fmt.Sprint(dates) != fmt.Sprint(tc.wantDates) {
				t.Errorf("got %v, want %v", dates, tc.wantDates)
			}
		})
	}
}

func TestEvictByAge(t *testing.T) {
	now := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC)
	s := testStore(t, Options{
		RetentionDays: 3,
		Now:           func() time.Time { return now },
	})
	for _, d := range []int{9, 10, 13, 15} {
		ts := time.Date(2026, 9, d, 12, 0, 0, 0, time.UTC).UnixMicro()
		if err := s.AddSpans([]model.Span{spanAt(fmt.Sprint(d), ts)}); err != nil {
			t.Fatal(err)
		}
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	s.Evict()

	shards, _ := s.ListShards(model.TableSpans)
	var dates []string
	for _, sh := range shards {
		dates = append(dates, sh.Date)
	}
	want := "[2026-09-13 2026-09-15]"
	if fmt.Sprint(dates) != want {
		t.Errorf("after eviction shards are %v, want %s", dates, want)
	}

	// The blob sidecar must go with the shard, or the data directory leaks.
	if _, err := os.Stat(filepath.Join(s.opts.Dir, model.TableSpans, "2026-09-09.blob")); !os.IsNotExist(err) {
		t.Error("evicting a shard left its .blob sidecar behind")
	}
}

func TestEvictBySize(t *testing.T) {
	s := testStore(t, Options{MaxBytes: 1}) // force eviction down to the newest
	for _, d := range []int{13, 14, 15} {
		ts := time.Date(2026, 9, d, 12, 0, 0, 0, time.UTC).UnixMicro()
		sp := spanAt(fmt.Sprint(d), ts)
		sp.Input = "some reasonably sized payload to give the shard bytes"
		if err := s.AddSpans([]model.Span{sp}); err != nil {
			t.Fatal(err)
		}
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	before, _ := s.ListShards(model.TableSpans)
	if len(before) != 3 {
		t.Fatalf("setup wrote %d shards, want 3", len(before))
	}
	s.Evict()
	after, _ := s.ListShards(model.TableSpans)

	if len(after) >= len(before) {
		t.Fatalf("size eviction removed nothing: %d shards before, %d after", len(before), len(after))
	}
	// Eviction is oldest-first, so whatever survives must be the newest.
	if len(after) > 0 && after[len(after)-1].Date != "2026-09-15" {
		t.Errorf("newest shard was evicted; survivors %v", after)
	}
}

func TestReopenSeesEarlierData(t *testing.T) {
	dir := t.TempDir()
	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	s1 := testStore(t, Options{Dir: dir})
	if err := s1.AddSpans([]model.Span{spanAt("a", base)}); err != nil {
		t.Fatal(err)
	}
	if err := s1.Close(); err != nil {
		t.Fatal(err)
	}

	// A fresh process must see the data with no index to rebuild: the directory
	// listing is the catalogue.
	s2 := testStore(t, Options{Dir: dir})
	shards, _ := s2.ListShards(model.TableSpans)
	if len(shards) != 1 {
		t.Fatalf("reopened store sees %d shards, want 1", len(shards))
	}
	sh, _ := s2.OpenShardForRead(model.TableSpans, "2026-09-15")
	if got := countRows(t, sh); got != 1 {
		t.Errorf("reopened store reads %d rows, want 1", got)
	}
}

func TestCloseFlushesPending(t *testing.T) {
	dir := t.TempDir()
	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	s, err := Open(Options{Dir: dir, FlushInterval: time.Hour, FlushRows: 1 << 20})
	if err != nil {
		t.Fatal(err)
	}
	if err := s.AddSpans([]model.Span{spanAt("a", base)}); err != nil {
		t.Fatal(err)
	}
	if err := s.Close(); err != nil {
		t.Fatal(err)
	}

	s2 := testStore(t, Options{Dir: dir})
	sh, _ := s2.OpenShardForRead(model.TableSpans, "2026-09-15")
	if got := countRows(t, sh); got != 1 {
		t.Errorf("Close dropped buffered spans: %d rows durable, want 1", got)
	}
}

func TestLogsRoundTrip(t *testing.T) {
	s := testStore(t, Options{})
	ts := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	rec := model.LogRecord{
		TraceID: "t1", SpanID: "s1", EventName: "user_prompt", TS: ts,
		Attrs: model.Attrs{"prompt.id": model.StringValue("p1")},
		Body:  "the actual prompt text",
	}
	if err := s.AddLogs([]model.LogRecord{rec}); err != nil {
		t.Fatal(err)
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	sh, err := s.OpenShardForRead(model.TableLogs, "2026-09-15")
	if err != nil {
		t.Fatal(err)
	}
	var got []model.LogRecord
	var bodies []string
	err = sh.Chunks(func(c *Chunk) error {
		for i := 0; i < c.Rows; i++ {
			got = append(got, readLog(c, i))
			b, err := sh.Blob().Read(blobAt(c, ColBody, i))
			if err != nil {
				return err
			}
			bodies = append(bodies, b)
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 1 {
		t.Fatalf("read %d log records, wrote 1", len(got))
	}
	if got[0].EventName != "user_prompt" || got[0].TraceID != "t1" {
		t.Errorf("log record round-tripped wrong: %+v", got[0])
	}
	if bodies[0] != rec.Body {
		t.Errorf("body = %q, want %q", bodies[0], rec.Body)
	}
}

// TestReadOnlyStoreDoesNotTruncate is the property that makes a CLI safe to run
// against a directory a server is writing to.
//
// Opening for writing runs tail recovery, which truncates. A reader taking that
// path while the server is mid-append cuts off the chunk being written — the
// reader destroying the writer's data. This simulates that exactly: a torn tail
// (what an in-flight append looks like on disk) must survive a read-only open.
func TestReadOnlyStoreDoesNotTruncate(t *testing.T) {
	dir := t.TempDir()
	spans := sampleSpans()

	sh := writeChunks(t, dir, spans[:2])
	c, err := buildSpanChunk(spans[2:], sh.Blob())
	if err != nil {
		t.Fatal(err)
	}
	if err := sh.AppendChunk(c); err != nil {
		t.Fatal(err)
	}
	sh.Close()

	// Cut the last chunk short, which is what a partially flushed append looks
	// like to another process.
	path := filepath.Join(dir, model.TableSpans, "2026-09-15.fathom")
	st, _ := os.Stat(path)
	tornSize := st.Size() - 20
	if err := os.Truncate(path, tornSize); err != nil {
		t.Fatal(err)
	}

	ro, err := store_openReadOnly(dir)
	if err != nil {
		t.Fatal(err)
	}
	// The reader sees only complete chunks...
	rs, err := ro.OpenShardForRead(model.TableSpans, "2026-09-15")
	if err != nil {
		t.Fatal(err)
	}
	if got := countRows(t, rs); got != 2 {
		t.Errorf("read %d rows, want the 2 from the complete chunk", got)
	}
	ro.Close()

	// ...and crucially has not altered the file. A writer resuming its append
	// would otherwise find its own bytes gone.
	after, _ := os.Stat(path)
	if after.Size() != tornSize {
		t.Errorf("a read-only open changed the file from %d to %d bytes",
			tornSize, after.Size())
	}
}

func store_openReadOnly(dir string) (*Store, error) {
	return Open(Options{Dir: dir, ReadOnly: true})
}

func TestReadOnlyStoreRejectsWrites(t *testing.T) {
	dir := t.TempDir()
	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	w := testStore(t, Options{Dir: dir})
	if err := w.AddSpans([]model.Span{spanAt("a", base)}); err != nil {
		t.Fatal(err)
	}
	if err := w.Flush(); err != nil {
		t.Fatal(err)
	}

	ro, err := store_openReadOnly(dir)
	if err != nil {
		t.Fatal(err)
	}
	defer ro.Close()

	// Buffering is harmless; the flush is where it must refuse, and it must
	// refuse quietly rather than corrupt anything.
	ro.AddSpans([]model.Span{spanAt("b", base)})
	if err := ro.Flush(); err != nil {
		t.Errorf("Flush on a read-only store should be a no-op, got %v", err)
	}
	if got := ro.Evict(); got != nil {
		t.Errorf("a read-only store evicted %d shards", len(got))
	}

	// The spans written before must still be all there is.
	sh, _ := ro.OpenShardForRead(model.TableSpans, "2026-09-15")
	if got := countRows(t, sh); got != 1 {
		t.Errorf("read-only store shows %d rows, want 1", got)
	}
}