Skip to content
quarry Artifact [6ba849678e]
Login

Artifact [6ba849678e]

Check-ins Using Download Hex

Artifact 6ba849678e2dee0eca2b10211997b9c08c1f9b8c4b23c945124b91da956161dc:

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

	// 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 {
			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, 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, err
	}
	if err := saveCursor(kv, p.code, next); 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
}