Skip to content
quarry Artifact [ad6f342835]
Login

Artifact [ad6f342835]

Check-ins Using Download Hex

Artifact ad6f342835c890b0e5260ccdab73cf9e9d57aab236072dca23166c4303b18a82:

// Package ingest is the 0001 §4 pipeline: a watcher validates the store
// prefix and publishes the new tail as events; a per-node indexer (a
// durable JetStream consumer) folds events into SQLite. The store is the
// event stream's source of truth — cursor loss (pack, shun reclaim) is
// detected, a reset event tells indexers to drop derived state, and the
// census republishes from zero. `quarry reindex` is the same path taken
// on purpose.
package ingest

import (
	"context"
	"crypto/sha1"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/nats-io/nats.go"

	"craftdesign.group/quarry/internal/artifact"
	"craftdesign.group/quarry/internal/events"
	"craftdesign.group/quarry/internal/evt"
	"craftdesign.group/quarry/internal/index"
	"craftdesign.group/quarry/internal/zitcli"
)

// Event is one artifact on the stream: the hash plus its parsed form.
// Opaque blobs never ride the stream — content is served on demand
// through zit, only structure is indexed.
type Event struct {
	Hash string             `json:"hash"`
	Art  *artifact.Artifact `json:"artifact"`
}

// Stats is a snapshot of one store's ingest health. A fleet of eleven
// repositories makes the server log useless as a diagnostic — a store
// that has been failing its round every two seconds for an hour reads
// exactly like a healthy one on any page that shows only counts. This
// is the state that tells them apart.
type Stats struct {
	// LastOK is when a round last completed without error; zero means
	// none ever has.
	LastOK time.Time
	// LastRound is how long that round took.
	LastRound time.Duration
	// LastCount is the artifacts published in it, Total across all
	// rounds since start.
	LastCount int
	Total     int
	// Fallbacks counts artifacts that failed to parse and were treated
	// as opaque content. Nonzero is not fatal — most stores carry a few
	// — but a jump is how a parser regression announces itself.
	//
	// Scope is THIS PROCESS, like Total: an artifact is only parsed when
	// it is ingested, so a store whose index is already warm reports
	// zero however many unparseable artifacts it holds. The number to
	// watch is the one a `quarry reindex` produces, not the one a
	// long-running server accumulates. Making it a property of the store
	// rather than the run means counting fallbacks in the index, which
	// is a schema change and a separate piece of work.
	Fallbacks int
	// LastErr is the most recent round's failure, empty once a round
	// succeeds. Failures counts CONSECUTIVE failures, so a flapping
	// store and a wedged one are distinguishable.
	LastErr   string
	LastErrAt time.Time
	Failures  int
	// Stalled means the pipeline has stopped retrying on its own
	// schedule: stallAfter consecutive failures, each retry
	// republishing the store's unfolded tail, is a wedge, not a blip.
	// RetryAt is when it will try again — backoff, not a latch, so a
	// fixed cause heals without operator ceremony. `quarry reindex`
	// bypasses it deliberately.
	Stalled bool
	RetryAt time.Time
	// Gaps counts DAG edges pointing at a check-in this index does
	// not hold. Non-zero means artifacts were lost from the event
	// window and a rebuild has not closed it — the one health signal
	// that says the index is LYING rather than merely behind.
	Gaps int
}

// Pipeline owns one store's watcher and indexer ends.
type Pipeline struct {
	cli  *zitcli.Client
	ev   *events.Server
	ix   *index.Index
	code string // project code in subjects: zit.evt.<code>.<kind>
	node string // this node's durable-consumer suffix (multi-node fan-out)

	sub *nats.Subscription // durable pull consumer, created lazily
	mu  sync.Mutex         // serializes Sync rounds

	// lists resolves check-in file changes during fold (changes.go).
	lists *fileLists

	// healed records that a repair has been attempted, and healedGaps
	// what it left behind, so a rebuild that cannot close a gap reports
	// it once rather than replaying the whole store every few minutes
	// forever. Written only from Sync, which is not concurrent with
	// itself for a given pipeline.
	healed     bool
	healedGaps int

	// tipChecked: the first round of each process asks zit for the
	// store's tip and rebuilds IMMEDIATELY if the index lacks it —
	// before any bus draining. Repair outranks catch-up: a reader is
	// looking at a wrong index NOW, and the rebuild is the fast path.
	tipChecked bool

	// statMu guards stats and is DELIBERATELY not mu: a round holds mu
	// for its whole duration, and a store's first fold runs for minutes.
	// Reading health through mu would block every fleet page render on
	// the slowest store in the fleet.
	statMu sync.Mutex
	stats  Stats
}

// Stats returns a snapshot of this store's ingest health.
func (p *Pipeline) Stats() Stats {
	p.statMu.Lock()
	defer p.statMu.Unlock()
	return p.stats
}

// record folds one round's outcome into the stats.
func (p *Pipeline) record(n int, d time.Duration, err error) {
	p.statMu.Lock()
	defer p.statMu.Unlock()
	p.stats.LastRound = d
	if err != nil {
		p.stats.LastErr = err.Error()
		p.stats.LastErrAt = time.Now()
		p.stats.Failures++
		if p.stats.Failures >= stallAfter {
			backoff := min(stallBase<<uint(p.stats.Failures-stallAfter), stallMax)
			p.stats.Stalled = true
			p.stats.RetryAt = time.Now().Add(backoff)
			if p.stats.Failures == stallAfter {
				log.Printf("ingest: %s: STALLED after %d consecutive failures (%v); backing off — the store is intact and the cursor is held, so nothing is lost, but this index stops advancing until the cause is fixed", p.code, p.stats.Failures, err)
			}
		}
		return
	}
	p.stats.LastOK = time.Now()
	p.stats.LastCount = n
	p.stats.Total += n
	p.stats.LastErr = ""
	p.stats.Failures = 0
	p.stats.Stalled = false
	p.stats.RetryAt = time.Time{}
}

// fallback counts one artifact that would not parse.
func (p *Pipeline) fallback() {
	p.statMu.Lock()
	defer p.statMu.Unlock()
	p.stats.Fallbacks++
}

// ErrStalled is Sync's answer while the pipeline is backing off after
// repeated failures. It is a sentinel so callers can tell "still broken,
// deliberately not retrying yet" from a fresh failure: the poller skips
// re-logging it, and a write op treats it as "the surface lags" rather
// than failing a write the store has already recorded.
var ErrStalled = errors.New("ingest: stalled; backing off")

// stallAfter is how many consecutive failures make a stall. One is a
// blip, two may be a race with a writer; three rounds of republishing a
// store's tail into a failing fold is a wedge.
const stallAfter = 3

// stallBase scales the backoff: base<<(failures-stallAfter), capped at
// stallMax. A var so tests can compress time.
var stallBase = 30 * time.Second

const stallMax = 10 * time.Minute

// ackWait is how long an unacked message stays invisible before the
// stream redelivers it. A round after a failed fold BLOCKS up to this
// long in the drain loop waiting for that redelivery — the wait is
// honest (breaking early would let the cursor save with messages still
// unfolded), so the knob is here rather than a shortcut there. A var so
// tests can compress it.
var ackWait = 30 * time.Second

// stallWait reports how long Sync must still hold off, zero when it may
// run.
func (p *Pipeline) stallWait() time.Duration {
	p.statMu.Lock()
	defer p.statMu.Unlock()
	if !p.stats.Stalled {
		return 0
	}
	return time.Until(p.stats.RetryAt)
}

// NewPipeline builds a store's pipeline for one node. node distinguishes
// this node's durable consumer from other nodes' on a shared stream, so
// each node folds the FULL stream (a shared durable name would
// load-balance one copy across nodes, leaving indexes partial). Empty
// node is the single-node default.
func NewPipeline(cli *zitcli.Client, ev *events.Server, ix *index.Index, code, node string) *Pipeline {
	return &Pipeline{cli: cli, ev: ev, ix: ix, code: code, node: node}
}

// durable is this node's consumer name for the store — valid NATS token
// (the node id, which may be an addr with dots/colons, is hashed).
func (p *Pipeline) durable() string {
	if p.node == "" {
		return "indexer-" + p.code
	}
	h := sha1.Sum([]byte(p.node))
	return "indexer-" + p.code + "-" + hex.EncodeToString(h[:6])
}

// Reindex is 0001 §5's full rebuild: drop the derived tables and the
// store cursor, then replay — the same convergence path cursor loss
// takes, invoked on purpose. Refolding events already in the stream is
// safe because folds are idempotent.
func (p *Pipeline) Reindex() (int, error) {
	// A rebuild is LOCAL DERIVATION, not synchronization: it reads the
	// census straight from zit into the fold, skipping the event bus
	// entirely. The bus (0001 §4) exists so OTHER consumers can follow
	// incrementally; routing a full replay through it cost JSON and
	// JetStream persistence per artifact and took hours where fossil's
	// in-process rebuild of the same store takes about a minute — and
	// its bounded live tail (1GiB, one hour) silently dropped whatever
	// out-published the fold. At the end the publisher cursor is set to
	// the stream's own trailer, so incremental sync resumes exactly
	// where the rebuild left the index. Folds stay idempotent, so any
	// stale messages a durable consumer still holds are absorbed.
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.reindexLocked()
}

func (p *Pipeline) reindexLocked() (int, error) {
	if err := p.ix.Reset(); err != nil {
		return 0, err
	}
	kv, err := p.ev.Cursors()
	if err != nil {
		return 0, err
	}
	if err := kv.Delete(p.code); err != nil && !errors.Is(err, nats.ErrKeyNotFound) {
		return 0, err
	}
	const batchSize = 512
	n := 0
	start := time.Now()
	if err := p.ix.SuspendDerivedMaintenance(); err != nil {
		return 0, err
	}
	// The census reader starts FIRST and fills batches while the
	// crosslink fold below owns the database.
	framesBatches := make(chan []index.Parsed, 8)
	var framesErr error
	var framesCursor zitcli.Cursor
	go func() {
		defer close(framesBatches)
		pending := make([]index.Parsed, 0, batchSize)
		framesCursor, framesErr = p.cli.FramesEachStructural(zitcli.Cursor{}, func(a zitcli.Artifact) error {
			if isM, ok := artifact.QuickIsManifest(a.Bytes); ok && isM {
				return nil
			}
			art, perr := artifact.Parse(a.Bytes)
			if perr != nil {
				p.fallback()
				log.Printf("ingest: %s: treating as content: %v", a.Hash, perr)
			}
			if art == nil || art.Type == artifact.Manifest {
				return nil
			}
			pending = append(pending, index.Parsed{Hash: a.Hash, Art: art})
			if len(pending) >= batchSize {
				framesBatches <- pending
				pending = make([]index.Parsed, 0, batchSize)
			}
			return nil
		})
		if len(pending) > 0 {
			framesBatches <- pending
		}
	}()
	// Check-in state comes from `zit crosslink` (zit#109): identity,
	// author, comment, tags and CHANGED FILES, already derived and
	// cache-served — 0.2s for tcl's 42K check-ins. This is what ended
	// the era of quarry re-parsing and re-expanding every manifest:
	// measured 85-163s of pass-2 work, deleted rather than optimised.
	records := make([]index.CheckinRecord, 0, batchSize)
	flushRecords := func() error {
		if len(records) == 0 {
			return nil
		}
		if err := p.ix.FoldCheckins(records); err != nil {
			return err
		}
		records = records[:0]
		return nil
	}
	if err := p.cli.Crosslink(func(r zitcli.CrosslinkRecord) error {
		rec := index.CheckinRecord{
			Hash: r.Hash, Date: crosslinkDate(r.MtimeMillis),
			User: r.User, Comment: r.Comment,
			Parents: r.Parents, Tags: r.Tags,
		}
		if r.FilesKnown {
			rec.Changes = r.Changes
		}
		records = append(records, rec)
		n++
		if len(records) >= batchSize {
			return flushRecords()
		}
		return nil
	}); err != nil {
		return n, err
	}
	if err := flushRecords(); err != nil {
		return n, err
	}
	crosslinkDone := time.Now()
	// Everything that is NOT a check-in still comes from the census:
	// tickets, wiki, forum, technotes, attachments, control artifacts.
	// Manifests are skipped outright — crosslink already carried their
	// every derived fact. Streaming and parsing runs CONCURRENTLY with
	// the crosslink fold above having warmed nothing — this goroutine
	// was started before it, so the census read and the check-in SQL
	// overlap; only the folds themselves serialise on the database.
	for pb := range framesBatches {
		n += len(pb)
		if err := p.ix.FoldBulk(pb); err != nil {
			return n, err
		}
	}
	if framesErr != nil {
		return n, framesErr
	}
	cur := framesCursor
	streamDone := time.Now()
	changesDone := streamDone
	if err := p.ix.ResumeDerivedMaintenance(); err != nil {
		return n, err
	}
	if err := p.ix.FinishBulk(); err != nil {
		return n, err
	}
	if err := p.ix.RebuildSearch(); err != nil {
		return n, err
	}
	if err := p.ix.RecomputeBranches(); err != nil {
		return n, err
	}
	if err := saveCursor(kv, p.code, cur); err != nil {
		return n, err
	}
	log.Printf("ingest: %s: rebuilt %d records in %.1fs (crosslink %.1fs, frames %.1fs, finish %.1fs)",
		p.code, n, time.Since(start).Seconds(), crosslinkDone.Sub(start).Seconds(),
		streamDone.Sub(crosslinkDone).Seconds(), time.Since(changesDone).Seconds())
	return n, nil
}

// Sync runs one full round: watch (store → events) then fold (events →
// index). Returns the number of artifacts published this round. Safe
// for concurrent callers (the background poll and write ops both call
// it); rounds serialize.
func (p *Pipeline) Sync() (int, error) {
	if wait := p.stallWait(); wait > 0 {
		return 0, fmt.Errorf("%w for %s more", ErrStalled, wait.Round(time.Second))
	}
	n, err := p.round()
	if err != nil {
		return n, err
	}
	// A round can succeed and still leave the index short. The stream
	// is a bounded live tail (1 GiB, one hour, discard-old): a fold
	// that keeps failing lets its own messages age out unacked, and the
	// publisher's cursor then advances past them on a later healthy
	// round. Nothing errors at the moment of loss — the artifacts are
	// simply never republished, and every surface built on the index
	// renders the hole as if it were the truth. A missing check-in read
	// as two branches on the timeline is how this was found.
	//
	// flushBytes already keeps a single burst inside the size cap; this
	// covers the other dimension, where the backlog is small but the
	// failure is long.
	// Shape before completeness: an index written under a different
	// derived shape cannot be repaired by folding forward, so there is
	// no point asking whether it is complete first.
	if p.ix.Stale() {
		log.Printf("ingest: %s: index was written by a different schema; rebuilding", p.code)
		if _, err := p.Reindex(); err != nil {
			log.Printf("ingest: %s: rebuild failed: %v", p.code, err)
			return n, nil
		}
		p.ix.ClearStale()
		return n, nil
	}

	// A dangling parent is only damage if the STORE still holds that
	// artifact. Fossil-born histories reference ancestors the store
	// never had — shunned artifacts, or an import that began partway
	// down the DAG; fossil, tcl and sqlite all carry some, and a
	// rebuild cannot conjure them. Counting those as damage would make
	// five of this corpus's repositories rebuild themselves on every
	// poll, forever.
	lost, gerr := p.lostArtifacts()
	if gerr != nil {
		log.Printf("ingest: %s: gap check: %v", p.code, gerr)
		return n, nil
	}
	p.statMu.Lock()
	p.stats.Gaps = lost
	p.statMu.Unlock()
	if lost == 0 {
		return n, nil
	}
	// One attempt, then report. A rebuild that does not close the gap
	// will not close it on the next poll either, and retrying forever
	// costs a full store replay every few minutes — on tk that is
	// 69,182 artifacts and about two minutes of work, permanently. An
	// index that cannot repair itself is a thing an operator must be
	// TOLD about, not something to grind at.
	if p.healed && lost >= p.healedGaps {
		return n, nil // already tried; no worse than last time
	}
	if p.healed {
		log.Printf("ingest: %s: %d artifacts still missing (was %d) — rebuilding again as it is getting worse", p.code, lost, p.healedGaps)
	} else {
		log.Printf("ingest: %s: %d artifacts the store holds are missing from the index; rebuilding", p.code, lost)
	}
	if _, err := p.Reindex(); err != nil {
		log.Printf("ingest: %s: rebuild failed: %v", p.code, err)
	}
	after, aerr := p.lostArtifacts()
	if aerr == nil {
		p.statMu.Lock()
		p.stats.Gaps = after
		p.statMu.Unlock()
		if after > 0 {
			log.Printf("ingest: %s: STILL %d artifacts missing after a full rebuild — the store holds them and the index does not; NOT retrying", p.code, after)
		}
		p.healedGaps = after
	}
	p.healed = true
	return n, nil
}

// lostArtifacts counts dangling parents the STORE still holds AS
// CHECK-INS — the repairable kind. Holding the hash is not enough: tk's
// history carries a P card naming a file blob (tkMacOSXEvent.c, CVS-
// conversion residue from 2011), and fossil's own database records the
// same edge and draws the same orphan. Counting that as damage sent tk
// through a full 69,182-artifact rebuild that could not possibly close
// the gap — no rebuild conjures a check-in from a C file — and then
// reported the store permanently broken. An artifact that is not a
// manifest is the same category as a shunned ancestor: inherent
// history, not loss.
//
// Bounded: a handful of probes is a fair price per round, and a
// repository with more dangling edges than this is already telling us
// what we need to know.
func (p *Pipeline) lostArtifacts() (int, error) {
	dangling, err := p.ix.DanglingParents(danglingProbe)
	if err != nil || len(dangling) == 0 {
		return 0, err
	}
	lost := 0
	for _, h := range dangling {
		raw, err := p.cli.ArtifactRaw(context.Background(), h, 1<<20)
		if err != nil {
			continue // the store does not hold it: inherent, not loss
		}
		if art, perr := artifact.Parse(raw); perr != nil || art == nil || art.Type != artifact.Manifest {
			continue // held, but not a check-in: no rebuild can index it as one
		}
		lost++
	}
	return lost, nil
}

// danglingProbe bounds the per-round cost of the check above: each probe
// is a zit subprocess, and the point is to notice loss, not to audit.
const danglingProbe = 25

// round is one publish-then-fold pass. Rounds serialize; the gap check
// in Sync deliberately runs OUTSIDE this lock, because healing re-enters
// the pipeline.
func (p *Pipeline) round() (int, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	// No cursor means bootstrap or repair, and both mean the same
	// thing: derive the index from the census directly. Publishing the
	// whole census through the bus "to catch up" is the hours-long
	// path the rebuild was moved off of.
	if kv, kerr := p.ev.Cursors(); kerr == nil {
		if _, gerr := kv.Get(p.code); gerr != nil {
			return p.reindexLocked()
		}
	}
	// Once per process: if the store's own tip is not in the index, the
	// index is wrong RIGHT NOW, and no amount of draining old bus
	// traffic fixes that faster than deriving from the census does.
	if !p.tipChecked {
		p.tipChecked = true
		if hash, _, err := p.cli.Info(context.Background(), "trunk"); err == nil {
			if _, _, ok, ierr := p.ix.CheckinByPrefix(hash); ierr == nil && !ok {
				log.Printf("ingest: %s: index lacks the store tip %.12s; rebuilding", p.code, hash)
				return p.reindexLocked()
			}
		}
	}
	start := time.Now()
	n, save, err := p.watch(p.fold)
	if err == nil {
		err = p.fold()
	}
	// The cursor is saved ONLY once everything published this round has
	// also been folded. The stream is a bounded live tail: messages a
	// failing fold leaves unacked age out of it, and a cursor that has
	// already moved past them means nothing ever republishes those
	// artifacts — the loss is silent and permanent (the 2026-08-19
	// schema incident). Held back, a fold failure costs only the re-read
	// and republish of the same tail next round, which idempotent folds
	// absorb.
	if err == nil && save != nil {
		err = save()
	}
	p.record(n, time.Since(start), err)
	return n, err
}

// flushBytes bounds how much a single publish burst may put on the
// stream before the caller is made to fold. ZITEVT is capped and
// discards OLD messages (events.go), so a census larger than the cap
// used to delete its own head while the indexer was still folding it —
// the index came out silently partial and then declared itself done.
// Keeping each burst far inside the window makes the stream apply
// backpressure instead of dropping. Well under the 1 GiB ceiling even
// with every repo ingesting at once.
var flushBytes = 64 << 20

// watch polls the store and publishes new structural artifacts. flush
// is called whenever the un-folded burst reaches flushBytes; it must
// drain the consumer (see Sync). The returned save advances the
// publisher cursor; round calls it only after the final fold succeeds.
func (p *Pipeline) watch(flush func() error) (int, func() error, error) {
	kv, err := p.ev.Cursors()
	if err != nil {
		return 0, nil, err
	}
	cur := loadCursor(kv, p.code)

	// Publish AS THE CENSUS IS READ. Collecting it first would hold the
	// whole thing in memory — sqlite's stream is 13.2GB and tcl's is
	// larger — and the peak is now one artifact plus the in-flight
	// backlog, whatever the store's size.
	published, unfolded := 0, 0
	each := func(a zitcli.Artifact) error {
		art, perr := artifact.Parse(a.Bytes)
		if perr != nil {
			p.fallback()
			log.Printf("ingest: %s: treating as content: %v", a.Hash, perr)
		}
		if art == nil {
			return nil
		}
		msg, err := json.Marshal(Event{Hash: a.Hash, Art: art})
		if err != nil {
			return err
		}
		if _, err := p.ev.JS().Publish(p.subject("artifact"), msg); err != nil {
			return fmt.Errorf("publish %s (%d bytes): %w", a.Hash, len(msg), err)
		}
		published++
		if unfolded += len(msg); unfolded >= flushBytes {
			if err := flush(); err != nil {
				return err
			}
			unfolded = 0
		}
		return nil
	}

	next, err := p.cli.FramesEach(cur, each)
	if errors.Is(err, zitcli.ErrCursorInvalid) {
		// Store rewritten: tell indexers to drop derived state, then
		// republish the census from zero.
		if _, err := p.ev.JS().Publish(p.subject("reset"), nil); err != nil {
			return published, nil, err
		}
		next, err = p.cli.FramesEach(zitcli.Cursor{}, each)
	}
	if err != nil {
		// The cursor is NOT saved on a failed round. Whatever was already
		// published gets seen again next time, which idempotent folds
		// absorb; saving it would skip the remainder for good.
		return published, nil, err
	}
	return published, func() error { return saveCursor(kv, p.code, next) }, nil
}

// fold drains the durable consumer into the index, one transaction per
// fetched batch. At-least-once redelivery is safe: folds are idempotent
// upserts keyed by artifact hash.
func (p *Pipeline) fold() error {
	if p.sub == nil {
		sub, err := p.ev.JS().PullSubscribe(p.subject(">"), p.durable(),
			nats.AckWait(ackWait))
		if err != nil {
			return err
		}
		p.sub = sub
	}
	branchy := false
	// Check-ins new to this index in this drain. branch.moved cannot be
	// raised inline with them: the branch is computed, not carried, and
	// the answer only exists after RecomputeBranches below.
	var freshCheckins []string
	for {
		msgs, err := p.sub.Fetch(256, nats.MaxWait(50*time.Millisecond))
		if errors.Is(err, nats.ErrTimeout) {
			// An empty fetch is not proof of drained — delivery can lag
			// the backlog. The consumer's pending count is authoritative.
			info, ierr := p.sub.ConsumerInfo()
			if ierr != nil {
				return ierr
			}
			if info.NumPending > 0 || info.NumAckPending > 0 {
				continue
			}
			break // drained
		}
		if err != nil {
			return err
		}
		if p.lists == nil {
			p.lists = newFileLists(p.cli)
		}
		var parsed []index.Parsed
		for _, m := range msgs {
			if m.Subject == p.subject("reset") {
				// Order matters: fold what preceded the reset, drop,
				// then continue with what follows.
				p.lists.annotate(parsed)
				if _, err := p.foldAndRaise(parsed, &freshCheckins); err != nil {
					return err
				}
				parsed = nil
				branchy = false
				if err := p.ix.Reset(); err != nil {
					return err
				}
				continue
			}
			var e Event
			if err := json.Unmarshal(m.Data, &e); err != nil {
				return fmt.Errorf("ingest: bad event: %w", err)
			}
			parsed = append(parsed, index.Parsed{Hash: e.Hash, Art: e.Art})
		}
		p.lists.annotate(parsed)
		b, err := p.foldAndRaise(parsed, &freshCheckins)
		if err != nil {
			return err
		}
		branchy = branchy || b
		for _, m := range msgs {
			if err := m.Ack(); err != nil {
				return err
			}
		}
	}
	if branchy {
		if err := p.ix.RecomputeBranches(); err != nil {
			return err
		}
	}
	p.raiseBranchMoved(freshCheckins)
	return nil
}

// raiseBranchMoved fires one branch.moved per new check-in, after the
// branches are computed.
//
// Per check-in, not per push, and not deduplicated per branch. RFC 0005
// §2.3.2 settled this: dagnats triggers carry a debounce, so a pipeline
// that wants one build per ten-commit push configures one, and a
// pipeline that wants ten gets ten. Coalescing here would take the
// choice away from both.
func (p *Pipeline) raiseBranchMoved(hashes []string) {
	if p.ev == nil || len(hashes) == 0 {
		return
	}
	branches, err := p.ix.BranchOf(hashes)
	if err != nil {
		log.Printf("ingest: %s: branch lookup: %v", p.code, err)
		return
	}
	for _, h := range hashes {
		b, ok := branches[h]
		if !ok {
			continue
		}
		if err := p.ev.Raise(evt.BranchMoved, p.code, map[string]any{
			"branch": b, "hash": h,
		}); err != nil {
			log.Printf("ingest: %s: raise %s: %v", p.code, evt.BranchMoved, err)
		}
	}
}

// foldAndRaise folds a batch and then raises RFC 0005's derived events
// for what was actually new.
//
// Two orderings matter and neither is arbitrary. Newness is asked BEFORE
// the fold, because the fold is what makes an artifact seen — asking
// afterwards would find everything already known and raise nothing. The
// events go out AFTER the commit, because an event says a thing is now
// true and a subscriber that reacts by querying must find it; raising
// first would offer a window where the forge says one thing and answers
// another.
//
// Only the DERIVED events come from here (§2.1). The operational ones —
// and the ticket, wiki and forum events that have an op behind them —
// are raised at that op, which knows more than the artifact does: an
// artifact records a ticket's new status, but only the op knows the old
// one, and without that there is no telling ticket.closed from
// ticket.changed. An artifact arriving from a clone's push has no op,
// and so today raises nothing beyond the check-in it rides on; that gap
// is real and named in the RFC rather than papered over here.
//
// A raise that fails is logged, never returned: the index is the record
// and it is already committed. Failing the fold over an undelivered
// notification would turn a missed reaction into a stalled store.
func (p *Pipeline) foldAndRaise(parsed []index.Parsed, freshCheckins *[]string) (bool, error) {
	var fresh map[string]bool
	if p.ev != nil && len(parsed) > 0 {
		hashes := make([]string, 0, len(parsed))
		for _, x := range parsed {
			hashes = append(hashes, x.Hash)
		}
		var err error
		if fresh, err = p.ix.Unseen(hashes); err != nil {
			return false, err
		}
	}
	branchy, err := p.ix.Fold(parsed)
	if err != nil {
		return branchy, err
	}
	for _, x := range parsed {
		if !fresh[x.Hash] {
			continue
		}
		if x.Art != nil && x.Art.Type == artifact.Manifest {
			*freshCheckins = append(*freshCheckins, x.Hash)
		}
		for _, r := range derive(x.Hash, x.Art) {
			if err := p.ev.Raise(r.Kind, p.code, r.Payload); err != nil {
				log.Printf("ingest: %s: raise %s: %v", p.code, r.Kind, err)
			}
		}
	}
	return branchy, nil
}

func (p *Pipeline) subject(kind string) string {
	return "zit.evt." + p.code + "." + kind
}

func loadCursor(kv nats.KeyValue, code string) zitcli.Cursor {
	entry, err := kv.Get(code)
	if err != nil {
		return zitcli.Cursor{} // advisory: absent means bootstrap
	}
	var c zitcli.Cursor
	if json.Unmarshal(entry.Value(), &c) != nil {
		return zitcli.Cursor{}
	}
	return c
}

// crosslinkDate renders epoch milliseconds in the canonical D-card
// millisecond form, matching what fold() stores from artifacts.
func crosslinkDate(ms int64) string {
	return time.UnixMilli(ms).UTC().Format("2006-01-02T15:04:05.000")
}

func saveCursor(kv nats.KeyValue, code string, c zitcli.Cursor) error {
	raw, err := json.Marshal(c)
	if err != nil {
		return err
	}
	_, err = kv.Put(code, raw)
	return err
}