Artifact [bc1d18f8f0]

Artifact bc1d18f8f09f7e4397516a0a6579f1167a40c038bd5a7bf064637d7b5e2aa873:

• File internal/ingest/ingest.go — part of check-in [44c86387b9] at 2026-08-16 15:57:41 on branch trunk — ingest: fold as we publish so a big census can't overrun the stream

  ZITEVT is capped with DiscardOld, and watch() published the whole census in one burst — so a repo larger than the cap deleted its own head while the indexer was still folding, then reported itself done. sqlite indexed 41,478 of 131,558 artifacts and lost every ticket, wiki page and forum post. Publishing now yields to folding every 64 MiB. (user: dmestas, size: 7479) [annotate] [blame] [check-ins using]

// 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 (
	"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/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"`
}

// 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
}

// 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) {
	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
	}
	return p.Sync()
}

// 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) {
	p.mu.Lock()
	defer p.mu.Unlock()
	n, err := p.watch(p.fold)
	if err != nil {
		return 0, err
	}
	if err := p.fold(); err != nil {
		return n, err
	}
	return n, nil
}

// 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).
func (p *Pipeline) watch(flush func() error) (int, error) {
	kv, err := p.ev.Cursors()
	if err != nil {
		return 0, err
	}
	cur := loadCursor(kv, p.code)
	batch, err := p.cli.Frames(cur)
	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 0, err
		}
		batch, err = p.cli.Frames(zitcli.Cursor{})
	}
	if err != nil {
		return 0, err
	}
	published, unfolded := 0, 0
	for _, a := range batch.Artifacts {
		art, perr := artifact.Parse(a.Bytes)
		if perr != nil {
			log.Printf("ingest: %s: treating as content: %v", a.Hash, perr)
		}
		if art == nil {
			continue
		}
		msg, err := json.Marshal(Event{Hash: a.Hash, Art: art})
		if err != nil {
			return published, err
		}
		if _, err := p.ev.JS().Publish(p.subject("artifact"), msg); err != nil {
			return published, 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 published, err
			}
			unfolded = 0
		}
	}
	if err := saveCursor(kv, p.code, batch.Cursor); err != nil {
		return published, err
	}
	return published, 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())
		if err != nil {
			return err
		}
		p.sub = sub
	}
	branchy := false
	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
		}
		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.
				if _, err := p.ix.Fold(parsed); 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})
		}
		b, err := p.ix.Fold(parsed)
		if err != nil {
			return err
		}
		branchy = branchy || b
		for _, m := range msgs {
			if err := m.Ack(); err != nil {
				return err
			}
		}
	}
	if branchy {
		return p.ix.RecomputeBranches()
	}
	return 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
}

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
}