Skip to content
quarry Artifact [b442691cbc]
Login

Artifact [b442691cbc]

Check-ins Using Download Hex

Artifact b442691cbc2103c7eb574b9beaa75269a60f57c8957ba1340cfb0129bbe3f7db:

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

// 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.
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, fmt.Errorf("publish %s (%d bytes): %w", a.Hash, len(msg), 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
	}
	branchy := false
	// A short wait detects the idle case cheaply; after a non-empty
	// fetch the wait stretches, so mid-backlog delivery lag is not
	// mistaken for "drained".
	wait := 50 * time.Millisecond
	for {
		msgs, err := p.sub.Fetch(256, nats.MaxWait(wait))
		if errors.Is(err, nats.ErrTimeout) {
			break // drained
		}
		if err != nil {
			return err
		}
		wait = 500 * time.Millisecond
		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
}