Artifact [5000553dbe]

Artifact 5000553dbebc3e75292925f4df2e53d87bdd1ce892d2e8c9c9d001416f3ba950:

• File internal/ingest/ingest.go — part of check-in [b0dc69c9a7] at 2026-08-15 11:14:05 on branch trunk — events: embedded NATS leg of the ingest pipeline (0001 §4)

  Watcher publishes parsed artifacts to zit.evt..artifact; durable pull consumer folds; cursors in KV; reset event on store rewrite. (user: dmestas, size: 4679) [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 (
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"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>

	sub *nats.Subscription // durable pull consumer, created lazily
}

func NewPipeline(cli *zitcli.Client, ev *events.Server, ix *index.Index, code string) *Pipeline {
	return &Pipeline{cli: cli, ev: ev, ix: ix, code: code}
}

// Sync runs one full round: watch (store → events) then fold (events →
// index). Returns the number of artifacts published this round.
func (p *Pipeline) Sync() (int, error) {
	n, err := p.watch()
	if err != nil {
		return 0, err
	}
	if err := p.fold(); err != nil {
		return n, err
	}
	return n, nil
}

// watch polls the store and publishes new structural artifacts.
func (p *Pipeline) watch() (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 := 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, err
		}
		published++
	}
	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(">"), "indexer-"+p.code)
		if err != nil {
			return err
		}
		p.sub = sub
	}
	for {
		msgs, err := p.sub.Fetch(256, nats.MaxWait(50*time.Millisecond))
		if errors.Is(err, nats.ErrTimeout) {
			return nil // 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
				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})
		}
		if err := p.ix.Fold(parsed); err != nil {
			return err
		}
		for _, m := range msgs {
			if err := m.Ack(); err != nil {
				return err
			}
		}
	}
}

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
}