golangLAKEHOUSE/internal/pathway/persistor_test.go

package pathway

import (
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
)

// persistor_test covers the corruption-recovery contract per
// Sprint 2 row 7: malformed JSONL lines must not halt replay.

func TestPersistor_MissingFileIsNotError(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "nonexistent.jsonl")
	p, err := NewPersistor(path)
	if err != nil {
		t.Fatalf("NewPersistor on missing file should not error, got %v", err)
	}
	n, err := p.Replay(func(event) error { return nil })
	if err != nil {
		t.Errorf("Replay on missing file should be 0,nil; got %d, %v", n, err)
	}
	if n != 0 {
		t.Errorf("Replay on missing file replayed %d events, want 0", n)
	}
}

func TestPersistor_AppendThenReplay(t *testing.T) {
	p := mustPersistor(t)

	if err := p.Append(event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}}); err != nil {
		t.Fatalf("Append: %v", err)
	}
	if err := p.Append(event{Op: opAdd, Trace: &Trace{UID: "B", Content: json.RawMessage(`{}`)}}); err != nil {
		t.Fatalf("Append: %v", err)
	}

	var seen []string
	n, err := p.Replay(func(e event) error {
		if e.Trace != nil {
			seen = append(seen, e.Trace.UID)
		}
		return nil
	})
	if err != nil {
		t.Fatalf("Replay: %v", err)
	}
	if n != 2 {
		t.Errorf("Replay applied %d events, want 2", n)
	}
	if len(seen) != 2 || seen[0] != "A" || seen[1] != "B" {
		t.Errorf("seen = %v, want [A B]", seen)
	}
}

func TestPersistor_CorruptedLines_Skipped(t *testing.T) {
	p := mustPersistor(t)

	// Mix of valid and corrupted lines.
	good1 := mustMarshal(t, event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}})
	bad := []byte(`{this is not json}`)
	good2 := mustMarshal(t, event{Op: opAdd, Trace: &Trace{UID: "B", Content: json.RawMessage(`{}`)}})
	emptyLine := []byte(``)
	good3 := mustMarshal(t, event{Op: opAdd, Trace: &Trace{UID: "C", Content: json.RawMessage(`{}`)}})

	contents := []byte{}
	for _, line := range [][]byte{good1, bad, good2, emptyLine, good3} {
		contents = append(contents, line...)
		contents = append(contents, '\n')
	}
	if err := os.WriteFile(p.Path(), contents, 0o644); err != nil {
		t.Fatalf("write file: %v", err)
	}

	var applied []string
	n, err := p.Replay(func(e event) error {
		if e.Trace != nil {
			applied = append(applied, e.Trace.UID)
		}
		return nil
	})
	if err != nil {
		t.Fatalf("Replay: %v", err)
	}
	// 3 valid + 1 bad + 1 empty (skipped silently) = 3 applied.
	if n != 3 {
		t.Errorf("Replay applied %d, want 3 (1 corrupt line skipped)", n)
	}
	if len(applied) != 3 || applied[0] != "A" || applied[1] != "B" || applied[2] != "C" {
		t.Errorf("applied = %v, want [A B C]", applied)
	}
}

func TestPersistor_ApplyError_Skipped(t *testing.T) {
	// If the apply function returns error for an event, replay
	// should keep going (the error is logged, not raised).
	p := mustPersistor(t)
	_ = p.Append(event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}})
	_ = p.Append(event{Op: opAdd, Trace: &Trace{UID: "B", Content: json.RawMessage(`{}`)}})
	_ = p.Append(event{Op: opAdd, Trace: &Trace{UID: "C", Content: json.RawMessage(`{}`)}})

	count := 0
	n, err := p.Replay(func(e event) error {
		if e.Trace != nil && e.Trace.UID == "B" {
			return errors.New("simulated apply error on B")
		}
		count++
		return nil
	})
	if err != nil {
		t.Fatalf("Replay: %v", err)
	}
	if n != 2 || count != 2 {
		t.Errorf("Replay applied %d (callback called %d), want 2 each (B's error skipped)", n, count)
	}
}

func TestPersistor_NewPersistor_EmptyPath_Errors(t *testing.T) {
	_, err := NewPersistor("")
	if err == nil {
		t.Error("NewPersistor with empty path should error")
	}
}

func TestPersistor_CreatesParentDir(t *testing.T) {
	dir := t.TempDir()
	nested := filepath.Join(dir, "nested", "deep", "pathway.jsonl")
	p, err := NewPersistor(nested)
	if err != nil {
		t.Fatalf("NewPersistor: %v", err)
	}
	if err := p.Append(event{Op: opAdd, Trace: &Trace{UID: "A", Content: json.RawMessage(`{}`)}}); err != nil {
		t.Fatalf("Append after creating nested dir: %v", err)
	}
}

func TestPersistor_LongLine_HandlesUpTo1MiB(t *testing.T) {
	p := mustPersistor(t)

	// Build a content blob ~750 KiB so the JSON line is ~800 KiB
	// (under the 1 MiB scanner cap).
	blob := strings.Repeat("x", 750*1024)
	bigContent, _ := json.Marshal(map[string]string{"data": blob})
	tr := &Trace{UID: "BIG", Content: bigContent}
	if err := p.Append(event{Op: opAdd, Trace: tr}); err != nil {
		t.Fatalf("Append big trace: %v", err)
	}

	count := 0
	n, _ := p.Replay(func(e event) error {
		if e.Trace != nil && e.Trace.UID == "BIG" {
			count++
		}
		return nil
	})
	if n != 1 || count != 1 {
		t.Errorf("big-line replay: got %d events / %d matches, want 1 each", n, count)
	}
}

// ── helpers ──

func mustPersistor(t *testing.T) *Persistor {
	t.Helper()
	dir := t.TempDir()
	path := filepath.Join(dir, "test.jsonl")
	p, err := NewPersistor(path)
	if err != nil {
		t.Fatalf("NewPersistor: %v", err)
	}
	return p
}

func mustMarshal(t *testing.T, e event) []byte {
	t.Helper()
	b, err := json.Marshal(e)
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	return b
}