package ingest
import (
"errors"
"fmt"
"os"
"os/exec"
"path/filepath"
"strings"
"testing"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/zitcli"
)
func run(t *testing.T, dir string, args ...string) {
t.Helper()
cmd := exec.Command("zit", args...)
cmd.Dir = dir
if out, err := cmd.CombinedOutput(); err != nil {
t.Fatalf("zit %v: %v\n%s", args, err, out)
}
}
func commitFile(t *testing.T, repo, name, body, msg string) {
t.Helper()
if err := os.WriteFile(filepath.Join(repo, name), []byte(body), 0o644); err != nil {
t.Fatal(err)
}
run(t, repo, "add", name)
run(t, repo, "commit", "-m", msg)
}
func testPipeline(t *testing.T) (*Pipeline, *index.Index, string) {
t.Helper()
if _, err := exec.LookPath("zit"); err != nil {
t.Skip("zit not on PATH")
}
repo := t.TempDir()
run(t, repo, "init")
commitFile(t, repo, "a.txt", "alpha\n", "add a")
commitFile(t, repo, "b.txt", "beta\n", "add b")
ev, err := events.Start(filepath.Join(t.TempDir(), "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
ix, err := index.Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
return NewPipeline(zitcli.New(repo), ev, ix, "test", "n1"), ix, repo
}
func TestPipelineBootstrapAndIncremental(t *testing.T) {
pipe, ix, repo := testPipeline(t)
n, err := pipe.Sync()
if err != nil {
t.Fatal(err)
}
if n == 0 {
t.Fatal("bootstrap published nothing")
}
tl, _, err := ix.Timeline(10, "", "", "ci")
if err != nil {
t.Fatal(err)
}
if len(tl) != 2 {
t.Fatalf("timeline has %d entries: %+v", len(tl), tl)
}
comments := map[string]bool{tl[0].Comment: true, tl[1].Comment: true}
if !comments["add a"] || !comments["add b"] {
t.Fatalf("timeline contents: %+v", tl)
}
// Current cursor: nothing published, nothing re-folded.
if n, err = pipe.Sync(); err != nil || n != 0 {
t.Fatalf("idle sync: n=%d err=%v", n, err)
}
// Append flows store → events → index.
commitFile(t, repo, "c.txt", "gamma\n", "add c")
if n, err = pipe.Sync(); err != nil || n == 0 {
t.Fatalf("incremental sync: n=%d err=%v", n, err)
}
tl, _, _ = ix.Timeline(10, "", "", "ci")
if len(tl) != 3 {
t.Fatalf("timeline after append: %+v", tl)
}
}
func TestPipelineRecoversFromStoreRewrite(t *testing.T) {
pipe, ix, repo := testPipeline(t)
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
before, _, _ := ix.Timeline(10, "", "", "ci")
// Rewrite the store under the cursor: shun a blob and reclaim.
head, err := exec.Command("zit", "ls", "-R", repo, "tip").Output()
if err != nil {
t.Fatal(err)
}
blob := strings.Fields(string(head))[0]
run(t, repo, "shun", blob, "--force")
// The reset event and re-census flow through the stream; the index
// converges to the same timeline.
if _, err := pipe.Sync(); err != nil {
t.Fatalf("sync after rewrite should reset and rebuild, got %v", err)
}
after, _, _ := ix.Timeline(10, "", "", "ci")
if len(after) != len(before) {
t.Fatalf("rebuild diverged: %d vs %d check-ins", len(after), len(before))
}
}
func TestPipelineSurvivesIndexerRestart(t *testing.T) {
// The durable consumer's position lives in the stream, not the
// process: a fresh Pipeline over the same events+index must not
// re-fold or lose anything.
pipe, ix, repo := testPipeline(t)
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
commitFile(t, repo, "d.txt", "delta\n", "add d")
fresh := NewPipeline(pipe.cli, pipe.ev, pipe.ix, "test", "n1")
if _, err := fresh.Sync(); err != nil {
t.Fatal(err)
}
tl, _, _ := ix.Timeline(10, "", "", "ci")
if len(tl) != 3 {
t.Fatalf("timeline after restart: %+v", tl)
}
}
func TestReindexRebuildsFromZero(t *testing.T) {
pipe, ix, _ := testPipeline(t)
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
before, _, _ := ix.Timeline(10, "", "", "ci")
if len(before) == 0 {
t.Fatal("no baseline")
}
n, err := pipe.Reindex()
if err != nil {
t.Fatal(err)
}
if n == 0 {
t.Fatal("reindex republished nothing")
}
after, _, _ := ix.Timeline(10, "", "", "ci")
if len(after) != len(before) {
t.Fatalf("reindex diverged: %d vs %d", len(after), len(before))
}
// A reindexed index answers idle syncs normally afterwards.
if n, err := pipe.Sync(); err != nil || n != 0 {
t.Fatalf("post-reindex idle sync: n=%d err=%v", n, err)
}
}
// ZITEVT is capped and discards OLD messages, so a census larger than
// the cap once deleted its own head while the indexer was still folding
// it: the index came out silently partial and then reported itself
// done. sqlite indexed 41,478 of 131,558 artifacts that way, losing
// every ticket, wiki page and forum post. Publishing must yield to
// folding before the stream can overrun.
func TestLargeCensusSurvivesACappedStream(t *testing.T) {
p, ix, repo := testPipeline(t)
// A stream far smaller than the census, and a flush threshold small
// enough that a test-sized corpus crosses it.
if _, err := p.ev.JS().UpdateStream(&nats.StreamConfig{
Name: events.StreamName,
Subjects: []string{"zit.evt.>"},
Storage: nats.FileStorage,
Discard: nats.DiscardOld,
MaxBytes: 128 << 10,
}); err != nil {
t.Fatal(err)
}
old := flushBytes
flushBytes = 8 << 10
t.Cleanup(func() { flushBytes = old })
const commits = 60
body := strings.Repeat("padding to make each manifest substantial\n", 40)
for i := 0; i < commits; i++ {
commitFile(t, repo, fmt.Sprintf("f%02d.txt", i), body+fmt.Sprint(i), fmt.Sprintf("commit %d", i))
}
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
tl, _, err := ix.Timeline(commits+50, "", "", "ci")
if err != nil {
t.Fatal(err)
}
if want := commits + 2; len(tl) != want { // +2 from testPipeline's fixture
t.Fatalf("indexed %d check-ins, want %d — the stream dropped events the indexer never folded", len(tl), want)
}
}
// The census must be handed over AS IT IS READ, never collected first:
// collecting it holds the whole stream in memory, 13.2GB for sqlite,
// which is the standing barrier to a store of real size.
//
// What this test can prove cheaply is the half that a unit test can
// see — that a caller may STOP mid-census and the read stops with it,
// which a buffering reader cannot offer at any price. The memory claim
// itself is measured against a real corpus, not asserted here.
func TestCensusReadCanBeAbandonedMidStream(t *testing.T) {
p, _, repo := testPipeline(t)
old := flushBytes
flushBytes = 1 // flush after every artifact, so each one is observable
t.Cleanup(func() { flushBytes = old })
for i := 0; i < 12; i++ {
commitFile(t, repo, fmt.Sprintf("s%02d.txt", i), strings.Repeat("x", 200), fmt.Sprintf("c%d", i))
}
// The whole census still arrives when nobody interrupts.
seen := 0
_, err := p.cli.FramesEach(zitcli.Cursor{}, func(a zitcli.Artifact) error {
seen++
return nil
})
if err != nil {
t.Fatal(err)
}
if seen < 10 {
t.Fatalf("only %d artifacts streamed, expected the whole census", seen)
}
// And an error mid-stream must stop the read rather than run to the
// end, so a caller can abandon a 13GB census cheaply.
stopped := 0
_, err = p.cli.FramesEach(zitcli.Cursor{}, func(a zitcli.Artifact) error {
stopped++
if stopped == 2 {
return errors.New("caller gave up")
}
return nil
})
if err == nil {
t.Fatal("an error from the callback did not stop the read")
}
if stopped != 2 {
t.Fatalf("read continued past the callback error: %d artifacts", stopped)
}
}
// The event stream is a bounded live tail, so a sustained fold failure
// can let unacked messages age out; the publisher's cursor then moves
// past them and nothing republishes. Nothing errors at the moment of
// loss, which is why a round that SUCCEEDS must still prove the index
// is whole — and repair it from the store when it is not.
func TestSyncRepairsAGapItFinds(t *testing.T) {
p, ix, _ := testPipeline(t)
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
before, err := ix.DanglingParents(50)
if err != nil {
t.Fatal(err)
}
if len(before) != 0 {
t.Fatalf("the fixture starts with %d dangling parents", len(before))
}
// Punch out a check-in the way a lost artifact does: the row goes,
// the edge that referenced it stays.
var victim string
if err := ix.RawDB().QueryRow(
`SELECT parent FROM checkin_parent LIMIT 1`).Scan(&victim); err != nil {
t.Skipf("fixture has no parent edges: %v", err)
}
if _, err := ix.RawDB().Exec(`DELETE FROM checkin WHERE hash = ?`, victim); err != nil {
t.Fatal(err)
}
if d, _ := ix.DanglingParents(50); len(d) == 0 {
t.Fatal("could not create a gap to repair")
}
// A successful round must notice and rebuild from the store.
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
after, err := ix.DanglingParents(50)
if err != nil {
t.Fatal(err)
}
if len(after) != 0 {
t.Errorf("after a round the index still has %d dangling edges; the gap was not repaired", len(after))
}
}
// A rebuild that cannot close a gap must be reported once, not retried
// forever: replaying a large store costs minutes, and doing it every
// poll would make an unrepairable index consume the machine instead of
// telling anyone. This pins the "try once, then report" contract.
func TestSyncDoesNotRebuildForeverOnAnUnrepairableGap(t *testing.T) {
p, ix, _ := testPipeline(t)
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
// A dangling edge whose parent the store does NOT hold — the shape
// fossil-born histories legitimately have. It must never trigger a
// rebuild at all.
if _, err := ix.RawDB().Exec(
`INSERT INTO checkin_parent (child, parent, ord)
SELECT hash, ?, 9 FROM checkin LIMIT 1`, "ff"+strings.Repeat("00", 19)); err != nil {
t.Fatal(err)
}
before := p.Stats()
if _, err := p.Sync(); err != nil {
t.Fatal(err)
}
after := p.Stats()
if after.Gaps != 0 {
t.Errorf("an ancestor the store never held is counted as loss (gaps=%d); "+
"every fossil-born repository would rebuild on every poll", after.Gaps)
}
_ = before
}