package ingest
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"sourcecraft.dev/bigbes/lethe/internal/collector/config"
"sourcecraft.dev/bigbes/lethe/internal/collector/parser"
"sourcecraft.dev/bigbes/lethe/internal/collector/state"
"sourcecraft.dev/bigbes/lethe/internal/shared/wire"
)
func TestRunOnce_AllAcceptedPersistsNewOffsetAndDoesNotEnqueueOutbox(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b")}, newOffset: 300},
})
sender := acceptingSender(t, []Result{{Accepted: 2}})
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 300)
assertOutboxCount(t, ctx, store, 0)
}
func TestRunOnce_TrimsPreexistingOutboxOverLimitEvenWithNoEnqueue(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
// Pre-populate outbox with two 10-byte items.
items := []state.OutboxItem{
{Tool: "claude-code", Host: "h", SourceFile: file, Payload: []byte("{\"seq\":1}\n")},
{Tool: "claude-code", Host: "h", SourceFile: file, Payload: []byte("{\"seq\":2}\n")},
}
for _, it := range items {
if err := store.Enqueue(ctx, it); err != nil {
t.Fatalf("Enqueue: %v", err)
}
}
cfg := testConfig()
cfg.Outbox.MaxBytes = 10
// No new events => no new enqueue.
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{}, newOffset: 0},
})
// Accepted:0 so replay does not delete rows; enforce must trim.
sender := acceptingSender(t, nil)
err := RunOnce(ctx, cfg, source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
rows, err := store.Oldest(ctx, 10)
if err != nil {
t.Fatalf("Oldest: %v", err)
}
if len(rows) != 1 {
t.Errorf("expected 1 row after trim, got %d", len(rows))
}
if len(rows) > 0 && string(rows[0].Payload) != "{\"seq\":2}\n" {
t.Errorf("remaining payload = %q, want %q", string(rows[0].Payload), "{\"seq\":2}\n")
}
}
func TestRunOnce_PartialAcceptWithErrorSkipsFailedEvent(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b"), turnEvent(300, "c")}, newOffset: 400},
})
sender := acceptingSender(t, []Result{{Accepted: 1, Errors: []LineError{{Line: 2, Err: "bad"}}}})
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 300)
assertOutboxCount(t, ctx, store, 0)
}
func TestRunOnce_ZeroAcceptedWithErrorLineTwoPostsPrefixAndSkipsFailedEvent(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
events := []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b"), turnEvent(300, "c")}
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: events, newOffset: 400},
})
var posted [][]wire.TurnEvent
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
var batch []wire.TurnEvent
for _, line := range strings.Split(string(body), "\n") {
if line == "" {
continue
}
var ev wire.TurnEvent
if err := json.Unmarshal([]byte(line), &ev); err != nil {
t.Fatalf("unmarshal event: %v", err)
}
batch = append(batch, ev)
}
posted = append(posted, batch)
w.Header().Set("Content-Type", "application/json")
if len(posted) == 1 {
_, _ = w.Write([]byte(resultJSON(Result{Accepted: 0, Errors: []LineError{{Line: 2, Err: "bad"}}})))
} else {
_, _ = w.Write([]byte(resultJSON(Result{Accepted: len(batch)})))
}
}))
defer ts.Close()
sender := NewSender(ts.URL, ts.Client())
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce first run: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 300)
if len(posted) != 2 {
t.Fatalf("first run: expected 2 POSTs, got %d", len(posted))
}
if len(posted[0]) != 3 {
t.Errorf("first run batch: expected 3 events, got %d", len(posted[0]))
}
if len(posted[1]) != 1 {
t.Errorf("prefix batch: expected 1 event, got %d", len(posted[1]))
}
if posted[1][0].Content != "a" {
t.Errorf("prefix batch: expected event 'a', got %q", posted[1][0].Content)
}
err = RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce second run: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 400)
if len(posted) != 3 {
t.Fatalf("second run: expected 3 POSTs total, got %d", len(posted))
}
if len(posted[2]) != 1 {
t.Errorf("second run batch: expected 1 event, got %d", len(posted[2]))
}
if posted[2][0].Content != "c" {
t.Errorf("second run batch: expected event 'c', got %q", posted[2][0].Content)
}
}
func TestRunOnce_MalformedErrorLineOutOfRangeReturnsError(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b")}, newOffset: 300},
})
sender := acceptingSender(t, []Result{{Accepted: 0, Errors: []LineError{{Line: 5, Err: "bad"}}}})
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err == nil {
t.Fatal("expected RunOnce error for malformed error line")
}
assertOffset(t, ctx, store, "claude-code", file, 0)
}
func TestRunOnce_ZeroAcceptedWithErrorSkipsFirstEvent(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b")}, newOffset: 300},
})
sender := acceptingSender(t, []Result{{Accepted: 0, Errors: []LineError{{Line: 1, Err: "bad"}}}})
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 200)
assertOutboxCount(t, ctx, store, 0)
}
func TestRunOnce_PartialAcceptSkipsBadRowAndRerunDoesNotRepostIt(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
events := []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b"), turnEvent(300, "c")}
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: events, newOffset: 400},
})
var posted [][]wire.TurnEvent
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
var batch []wire.TurnEvent
for _, line := range strings.Split(string(body), "\n") {
if line == "" {
continue
}
var ev wire.TurnEvent
if err := json.Unmarshal([]byte(line), &ev); err != nil {
t.Fatalf("unmarshal event: %v", err)
}
batch = append(batch, ev)
}
posted = append(posted, batch)
w.Header().Set("Content-Type", "application/json")
if len(posted) == 1 {
_, _ = w.Write([]byte(resultJSON(Result{Accepted: 1, Errors: []LineError{{Line: 2, Err: "bad"}}})))
} else {
_, _ = w.Write([]byte(resultJSON(Result{Accepted: 1})))
}
}))
defer ts.Close()
sender := NewSender(ts.URL, ts.Client())
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce first run: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 300)
if len(posted) != 1 {
t.Fatalf("first run: expected 1 POST, got %d", len(posted))
}
if len(posted[0]) != 3 {
t.Errorf("first run: expected 3 events in batch, got %d", len(posted[0]))
}
err = RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce second run: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 400)
if len(posted) != 2 {
t.Fatalf("second run: expected 2 POSTs total, got %d", len(posted))
}
if len(posted[1]) != 1 {
t.Errorf("second run: expected 1 event in batch, got %d", len(posted[1]))
}
if posted[1][0].Content != "c" {
t.Errorf("second run: expected event 'c', got %q", posted[1][0].Content)
}
}
func TestRunOnce_FullAcceptWithErrorsReturnsError(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b")}, newOffset: 300},
})
sender := acceptingSender(t, []Result{{Accepted: 2, Errors: []LineError{{Line: 2, Err: "bad"}}}})
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err == nil {
t.Fatal("expected RunOnce error when server accepts all but reports errors")
}
assertOffset(t, ctx, store, "claude-code", file, 0)
}
func TestRunOnce_HardErrorEnqueuesUnsentBatchAndDoesNotAdvanceOffset(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
events := []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b")}
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: events, newOffset: 300},
})
sender := failingSender(t)
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err == nil {
t.Fatal("expected RunOnce error on hard POST failure")
}
assertOffset(t, ctx, store, "claude-code", file, 0)
rows, err := store.Oldest(ctx, 10)
if err != nil {
t.Fatalf("Oldest: %v", err)
}
if len(rows) != 1 {
t.Fatalf("len(rows) = %d, want 1", len(rows))
}
wantPayload, err := EncodeNDJSON(events)
if err != nil {
t.Fatalf("EncodeNDJSON: %v", err)
}
if string(rows[0].Payload) != string(wantPayload) {
t.Errorf("payload = %q, want %q", string(rows[0].Payload), string(wantPayload))
}
}
func TestRunOnce_ContinuesToOtherFilesWhenOneFileFails(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
badFile := filepath.Join(source.Path, "bad.jsonl")
goodFile := filepath.Join(source.Path, "good.jsonl")
p := newFakeParser("claude-code", []parser.SourceFile{{Path: badFile}, {Path: goodFile}}, map[string]parseResult{
badFile: {err: errParseBoom},
goodFile: {events: []wire.TurnEvent{turnEvent(100, "ok")}, newOffset: 200},
})
sender := acceptingSender(t, []Result{{Accepted: 1}})
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err == nil {
t.Fatal("expected RunOnce to return collected parse error")
}
assertOffset(t, ctx, store, "claude-code", badFile, 0)
assertOffset(t, ctx, store, "claude-code", goodFile, 200)
}
func TestRunDaemon_ExitsWhenContextCanceled(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
store := openTestStore(t, context.Background())
source := testSource(t, "claude-code", 10, 4096)
source.PollInterval = time.Hour
p := newFakeParser("claude-code", nil, nil)
sender := acceptingSender(t, nil)
cfg := testConfig()
cfg.Sources = []config.SourceConfig{source}
done := make(chan error, 1)
go func() {
done <- RunDaemon(ctx, cfg, map[string]parser.Parser{"claude-code": p}, store, sender)
}()
cancel()
select {
case err := <-done:
if err != nil {
t.Fatalf("RunDaemon: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("RunDaemon did not exit after context cancellation")
}
}
func TestRunDaemon_CancelWaitsForActiveRunOnce(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
store := openTestStore(t, context.Background())
source := testSource(t, "claude-code", 10, 4096)
source.PollInterval = time.Hour
file := filepath.Join(source.Path, "one.jsonl")
postStarted := make(chan struct{})
postContinue := make(chan struct{})
var postOnce sync.Once
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{turnEvent(100, "a")}, newOffset: 200},
})
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
postOnce.Do(func() { close(postStarted) })
<-postContinue
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(resultJSON(Result{Accepted: 1})))
}))
defer ts.Close()
sender := NewSender(ts.URL, ts.Client())
cfg := testConfig()
cfg.Sources = []config.SourceConfig{source}
cfg.HTTP.Timeout = 5 * time.Second
done := make(chan error, 1)
go func() {
done <- RunDaemon(ctx, cfg, map[string]parser.Parser{"claude-code": p}, store, sender)
}()
select {
case <-postStarted:
case <-time.After(2 * time.Second):
t.Fatal("POST did not start")
}
cancel()
select {
case <-done:
t.Fatal("RunDaemon should wait for active RunOnce, not exit immediately")
case <-time.After(100 * time.Millisecond):
// expected
}
close(postContinue)
select {
case err := <-done:
if err != nil {
t.Fatalf("RunDaemon: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("RunDaemon did not exit after active run completed")
}
assertOffset(t, context.Background(), store, "claude-code", file, 200)
}
func TestRunBackfillOnce_IgnoresSavedOffsetAndPersistsProgress(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
if err := store.SaveOffset(ctx, "claude-code", file, 150); err != nil {
t.Fatalf("SaveOffset: %v", err)
}
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b")}, newOffset: 300},
})
sender := acceptingSender(t, []Result{{Accepted: 2}})
err := RunBackfillOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunBackfillOnce: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 300)
assertOutboxCount(t, ctx, store, 0)
}
func TestRunBackfillOnce_InterruptedProgressIsResumable(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
if err := store.SaveOffset(ctx, "claude-code", file, 150); err != nil {
t.Fatalf("SaveOffset: %v", err)
}
events := []wire.TurnEvent{turnEvent(100, "a"), turnEvent(200, "b")}
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: events, newOffset: 300},
})
sender := acceptingSender(t, []Result{{Accepted: 1}})
err := RunBackfillOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunBackfillOnce: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 200)
sender2 := acceptingSender(t, []Result{{Accepted: 1}})
err = RunOnce(ctx, testConfig(), source, p, store, sender2)
if err != nil {
t.Fatalf("RunOnce resume: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 300)
}
func TestRunOnce_SkippedOnlyParseResultPersistsNewOffsetAndDoesNotPost(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
var posted int
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
posted++
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(resultJSON(Result{Accepted: 0})))
}))
defer ts.Close()
sender := NewSender(ts.URL, ts.Client())
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{}, newOffset: 250},
})
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 250)
if posted != 0 {
t.Errorf("POST count = %d, want 0", posted)
}
}
func TestRunOnce_EmptyParseResultNoProgressDoesNotSaveOffset(t *testing.T) {
ctx := context.Background()
store := openTestStore(t, ctx)
source := testSource(t, "claude-code", 10, 4096)
file := filepath.Join(source.Path, "one.jsonl")
p := newFakeParser("claude-code", []parser.SourceFile{{Path: file}}, map[string]parseResult{
file: {events: []wire.TurnEvent{}, newOffset: 0},
})
sender := acceptingSender(t, nil)
err := RunOnce(ctx, testConfig(), source, p, store, sender)
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
assertOffset(t, ctx, store, "claude-code", file, 0)
}
type parseResult struct {
events []wire.TurnEvent
newOffset int64
err error
}
type fakeParser struct {
tool string
files []parser.SourceFile
results map[string]parseResult
}
func newFakeParser(tool string, files []parser.SourceFile, results map[string]parseResult) *fakeParser {
return &fakeParser{tool: tool, files: files, results: results}
}
func (p *fakeParser) Tool() string { return p.tool }
func (p *fakeParser) Discover(root string) ([]parser.SourceFile, error) { return p.files, nil }
func (p *fakeParser) Parse(path string, since int64) ([]wire.TurnEvent, int64, error) {
res := p.results[path]
if res.err != nil {
return nil, since, res.err
}
var filtered []wire.TurnEvent
for _, ev := range res.events {
if ev.Seq >= since {
filtered = append(filtered, ev)
}
}
return filtered, res.newOffset, nil
}
type roundTripFunc func(*http.Request) (*http.Response, error)
func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
func acceptingSender(t *testing.T, results []Result) *Sender {
t.Helper()
var mu sync.Mutex
idx := 0
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = io.ReadAll(r.Body)
mu.Lock()
defer mu.Unlock()
result := Result{Accepted: 0}
if idx < len(results) {
result = results[idx]
idx++
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(resultJSON(result)))
}))
t.Cleanup(ts.Close)
return NewSender(ts.URL, ts.Client())
}
func failingSender(t *testing.T) *Sender {
t.Helper()
client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
return nil, errPostBoom
})}
return NewSender("http://127.0.0.1", client)
}