Artifact [2c710d6e7b]

Artifact 2c710d6e7bb83f1ecd5b9dbd4cb3fae72da162190944d563e1465f29eaf4c275:

• File internal/ingest/ingest_test.go — part of check-in [7124d5e366] at 2026-08-15 18:22:38 on branch trunk — reindex verb: drop derived state and replay stores (0001 §5)

  Pipeline.Reindex resets tables + deletes the KV cursor, then syncs; refolding streamed events is idempotent. CLI: quarry reindex --repo. (user: dmestas, size: 4388) [annotate] [blame] [check-ins using]

package ingest

import (
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"

	"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"), 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")
	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)
	}
}