Skip to content
quarry Artifact [d854d9b4ef]
Login

Artifact [d854d9b4ef]

Check-ins Using Download Hex

Artifact d854d9b4efbfdd2edb93428551c74a193d52bcc5dcd914d1f11643f78a3eb137:

// Package index is the per-node derived index (0001 §5): SQLite tables
// folded from store artifacts, disposable and rebuildable by design —
// dropping the database file and re-ingesting must converge to identical
// state, and the disposability harness proves it. This package is a data
// plane; the arch gate restricts its importers.
package index

import (
	"database/sql"
	"fmt"
	"strings"

	"github.com/ncruces/go-sqlite3/driver"
	"github.com/ncruces/go-sqlite3/ext/fts5"

	"craftdesign.group/quarry/internal/artifact"
)

const schema = `
CREATE TABLE IF NOT EXISTS checkin (
	hash    TEXT PRIMARY KEY,
	date    TEXT NOT NULL,
	user    TEXT NOT NULL,
	comment TEXT NOT NULL,
	branch  TEXT NOT NULL DEFAULT 'trunk'
);
CREATE TABLE IF NOT EXISTS checkin_tag (
	src     TEXT NOT NULL, -- artifact that carries the T card
	date    TEXT NOT NULL, -- that artifact's D card: tag priority order
	checkin TEXT NOT NULL, -- tag target ("*" resolved to src for manifests)
	name    TEXT NOT NULL, -- without the sign prefix
	sign    TEXT NOT NULL, -- + - *
	value   TEXT NOT NULL,
	PRIMARY KEY (src, checkin, name)
);
CREATE TABLE IF NOT EXISTS checkin_parent (
	child  TEXT NOT NULL,
	parent TEXT NOT NULL,
	ord    INTEGER NOT NULL,
	PRIMARY KEY (child, ord)
);
CREATE TABLE IF NOT EXISTS wiki_edit (
	hash     TEXT PRIMARY KEY,
	title    TEXT NOT NULL,
	date     TEXT NOT NULL,
	user     TEXT NOT NULL,
	mimetype TEXT NOT NULL,
	body     BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS ticket_change (
	hash   TEXT PRIMARY KEY,
	ticket TEXT NOT NULL,
	date   TEXT NOT NULL,
	user   TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS ticket_field (
	hash  TEXT NOT NULL,
	name  TEXT NOT NULL,
	value TEXT NOT NULL,
	PRIMARY KEY (hash, name)
);
CREATE TABLE IF NOT EXISTS search_doc (
	hash  TEXT PRIMARY KEY, -- artifact hash ("ticket:<id>" for folded ticket docs)
	kind  TEXT NOT NULL,    -- checkin | wiki | ticket
	ref   TEXT NOT NULL,    -- link target: checkin hash, wiki title, ticket id
	label TEXT NOT NULL,    -- human-readable result label
	text  TEXT NOT NULL
);
CREATE VIRTUAL TABLE IF NOT EXISTS search_fts USING fts5(
	text, content='search_doc', content_rowid='rowid'
);
CREATE TRIGGER IF NOT EXISTS search_doc_ai AFTER INSERT ON search_doc BEGIN
	INSERT INTO search_fts (rowid, text) VALUES (new.rowid, new.text);
END;
CREATE TRIGGER IF NOT EXISTS search_doc_ad AFTER DELETE ON search_doc BEGIN
	INSERT INTO search_fts (search_fts, rowid, text) VALUES ('delete', old.rowid, old.text);
END;
CREATE TRIGGER IF NOT EXISTS search_doc_au AFTER UPDATE ON search_doc BEGIN
	INSERT INTO search_fts (search_fts, rowid, text) VALUES ('delete', old.rowid, old.text);
	INSERT INTO search_fts (rowid, text) VALUES (new.rowid, new.text);
END;
`

// Index is one repository's derived index.
type Index struct {
	db *sql.DB
}

// Open opens (creating if needed) the index database at path.
func Open(path string) (*Index, error) {
	db, err := driver.Open("file:"+path, fts5.Register)
	if err != nil {
		return nil, err
	}
	if _, err := db.Exec(schema); err != nil {
		db.Close()
		return nil, fmt.Errorf("index: schema: %w", err)
	}
	return &Index{db: db}, nil
}

func (ix *Index) Close() error { return ix.db.Close() }

// Parsed pairs an artifact hash with its parsed form; Art is nil for
// opaque content blobs (folded as nothing — content is served on demand
// through zit, never copied into the index).
type Parsed struct {
	Hash string
	Art  *artifact.Artifact
}

// Fold applies one event batch atomically. Folds are idempotent upserts
// keyed by hash, so at-least-once event delivery converges. Ingest
// position is not stored here: the store cursor lives in NATS KV and the
// stream position in the durable consumer (0001 §4) — both advisory,
// both rebuilt by replaying stores.
func (ix *Index) Fold(batch []Parsed) error {
	tx, err := ix.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	branchy := false
	for _, p := range batch {
		if p.Art == nil {
			continue
		}
		if err := fold(tx, p.Hash, p.Art); err != nil {
			return fmt.Errorf("index: fold %s: %w", p.Hash, err)
		}
		if p.Art.Type == artifact.Manifest || p.Art.Type == artifact.Control {
			branchy = true
		}
	}
	if branchy {
		if err := recomputeBranches(tx); err != nil {
			return fmt.Errorf("index: branches: %w", err)
		}
	}
	return tx.Commit()
}

func fold(tx *sql.Tx, hash string, a *artifact.Artifact) error {
	switch a.Type {
	case artifact.Manifest:
		// INSERT OR REPLACE would clobber the computed branch column on
		// redelivery; upsert only the artifact-borne fields.
		if _, err := tx.Exec(`INSERT INTO checkin (hash, date, user, comment)
			VALUES (?, ?, ?, ?)
			ON CONFLICT (hash) DO UPDATE SET date=excluded.date,
				user=excluded.user, comment=excluded.comment`,
			hash, a.Date, a.User, a.Comment); err != nil {
			return err
		}
		for i, parent := range a.Parents {
			if _, err := tx.Exec(`INSERT OR REPLACE INTO checkin_parent (child, parent, ord)
				VALUES (?, ?, ?)`, hash, parent, i); err != nil {
				return err
			}
		}
		if err := foldTags(tx, hash, hash, a.Date, a.Tags); err != nil {
			return err
		}
	case artifact.Control:
		if err := foldTags(tx, hash, "", a.Date, a.Tags); err != nil {
			return err
		}
	case artifact.Wiki:
		if _, err := tx.Exec(`INSERT OR REPLACE INTO wiki_edit (hash, title, date, user, mimetype, body)
			VALUES (?, ?, ?, ?, ?, ?)`,
			hash, a.WikiTitle, a.Date, a.User, a.Mimetype, a.Body); err != nil {
			return err
		}
	case artifact.Ticket:
		if _, err := tx.Exec(`INSERT OR REPLACE INTO ticket_change (hash, ticket, date, user)
			VALUES (?, ?, ?, ?)`, hash, a.TicketID, a.Date, a.User); err != nil {
			return err
		}
		for _, f := range a.Fields {
			if _, err := tx.Exec(`INSERT OR REPLACE INTO ticket_field (hash, name, value)
				VALUES (?, ?, ?)`, hash, f.Name, f.Value); err != nil {
				return err
			}
		}
	}
	if a.Type == artifact.Ticket {
		// Ticket search docs hold the FOLDED state (one doc per ticket,
		// keyed by id), so snippets never show stale field values from
		// early changes (ticket 050). Order-independent: any change
		// arriving recomputes the whole fold.
		if err := refreshTicketDoc(tx, a.TicketID); err != nil {
			return err
		}
	} else if kind, ref, label, text := searchDoc(hash, a); kind != "" {
		if _, err := tx.Exec(`INSERT OR REPLACE INTO search_doc (hash, kind, ref, label, text)
			VALUES (?, ?, ?, ?, ?)`, hash, kind, ref, label, text); err != nil {
			return err
		}
	}
	// Control, cluster, technote, forum, attachment: folded when their
	// pages land (phase 2); skipping them is safe because a reindex
	// replays every artifact.
	return nil
}

// foldTags records an artifact's T cards as tag facts. self names the
// check-in a "*" target resolves to (the carrying manifest); control
// artifacts have no self and their "*" targets are illegal anyway
// (§5.4), so those rows are skipped.
func foldTags(tx *sql.Tx, src, self, date string, tags []artifact.Tag) error {
	for _, t := range tags {
		target := t.Target
		if target == "*" {
			if self == "" {
				continue
			}
			target = self
		}
		if _, err := tx.Exec(`INSERT OR REPLACE INTO checkin_tag (src, date, checkin, name, sign, value)
			VALUES (?, ?, ?, ?, ?, ?)`,
			src, date, target, strings.TrimLeft(t.Name, "+-*"), t.Name[:1], t.Value); err != nil {
			return err
		}
	}
	return nil
}

// recomputeBranches rewrites every check-in's branch from the recorded
// tag facts: a propagating branch tag names its check-in's branch, and
// the name flows down primary-parent edges until a descendant carries
// its own. A WHOLESALE recompute after each fold batch is what makes
// the column order-independent — the same tag facts and DAG yield the
// same branches no matter what order artifacts arrived in (the
// disposability property). ~4k check-ins recompute in milliseconds.
func recomputeBranches(tx *sql.Tx) error {
	_, err := tx.Exec(`
		WITH RECURSIVE br(hash, branch) AS (
			-- one seed per check-in: the branch tag from the LATEST
			-- carrying artifact wins (bare column with max())
			SELECT t.checkin, t.value FROM (
				SELECT checkin, value, max(date || src)
				FROM checkin_tag WHERE name = 'branch' AND sign = '*'
				GROUP BY checkin
			) t JOIN checkin c ON c.hash = t.checkin
			UNION
			SELECT p.child, br.branch FROM br
			JOIN checkin_parent p ON p.parent = br.hash AND p.ord = 0
			WHERE NOT EXISTS (SELECT 1 FROM checkin_tag t2
				WHERE t2.checkin = p.child AND t2.name = 'branch' AND t2.sign = '*')
		)
		UPDATE checkin SET branch = COALESCE(
			(SELECT branch FROM br WHERE br.hash = checkin.hash LIMIT 1), 'trunk')`)
	return err
}

// refreshTicketDoc rebuilds one ticket's search doc from its folded
// current state, inside the fold transaction.
func refreshTicketDoc(tx *sql.Tx, ticket string) error {
	rows, err := tx.Query(`SELECT f.name, f.value FROM ticket_change c
		JOIN ticket_field f ON f.hash = c.hash
		WHERE c.ticket = ? ORDER BY c.date, c.hash, f.name`, ticket)
	if err != nil {
		return err
	}
	defer rows.Close()
	current := map[string]string{}
	var order []string
	for rows.Next() {
		var name, value string
		if err := rows.Scan(&name, &value); err != nil {
			return err
		}
		if _, seen := current[name]; !seen {
			order = append(order, name)
		}
		current[name] = value // latest assignment wins
	}
	if err := rows.Err(); err != nil {
		return err
	}
	var b strings.Builder
	for _, name := range order {
		b.WriteString(current[name])
		b.WriteByte('\n')
	}
	label := current["title"]
	if label == "" && len(ticket) >= 10 {
		label = ticket[:10]
	}
	_, err = tx.Exec(`INSERT OR REPLACE INTO search_doc (hash, kind, ref, label, text)
		VALUES (?, 'ticket', ?, ?, ?)`, "ticket:"+ticket, ticket, label, b.String())
	return err
}

// searchDoc maps one immutable artifact to its search-corpus row; empty
// kind means "not searched here" (tickets go through refreshTicketDoc).
// Every wiki edit is indexed, so hits can land on historical revisions;
// queries dedupe by ref.
func searchDoc(hash string, a *artifact.Artifact) (kind, ref, label, text string) {
	switch a.Type {
	case artifact.Manifest:
		first, _, _ := strings.Cut(a.Comment, "\n")
		return "checkin", hash, first, a.Comment
	case artifact.Wiki:
		return "wiki", a.WikiTitle, a.WikiTitle, a.WikiTitle + "\n" + string(a.Body)
	}
	return "", "", "", ""
}

// SearchResult is one hit, deduplicated by (kind, ref). Snippet marks
// matches with ⟦…⟧ — characters that cannot collide with markdown or
// bracket-bearing content (ticket 060); renderers turn them into <mark>.
type SearchResult struct {
	Kind    string `json:"kind"`
	Ref     string `json:"ref"`
	Label   string `json:"label"`
	Snippet string `json:"snippet"`
}

// Search runs an FTS5 query over the corpus. The user string is
// re-quoted token-by-token so FTS operator syntax cannot inject or
// error. Ranking is bm25 with ref as the deterministic tiebreak.
func (ix *Index) Search(query string, limit int) ([]SearchResult, error) {
	tokens := strings.Fields(query)
	if len(tokens) == 0 {
		return nil, nil
	}
	for i, tok := range tokens {
		tokens[i] = `"` + strings.ReplaceAll(tok, `"`, `""`) + `"`
	}
	// The MATERIALIZED hint stops SQLite flattening the CTE into the
	// aggregate query, where bm25() is not usable.
	rows, err := ix.db.Query(`
		WITH hits AS MATERIALIZED (
			SELECT d.kind AS kind, d.ref AS ref, d.label AS label,
				snippet(search_fts, 0, '⟦', '⟧', '…', 12) AS snip,
				bm25(search_fts) AS rank
			FROM search_fts JOIN search_doc d ON d.rowid = search_fts.rowid
			WHERE search_fts MATCH ?
		)
		SELECT kind, ref, label, snip, min(rank) AS best FROM hits
		GROUP BY kind, ref ORDER BY best, kind, ref LIMIT ?`,
		strings.Join(tokens, " "), limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []SearchResult
	for rows.Next() {
		var r SearchResult
		var rank float64
		if err := rows.Scan(&r.Kind, &r.Ref, &r.Label, &r.Snippet, &rank); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	return out, rows.Err()
}

// Reset drops all derived state — the reindex path (0001 §5). The next
// Fold rebuilds from a zero cursor.
func (ix *Index) Reset() error {
	for _, table := range []string{
		"checkin", "checkin_parent", "checkin_tag", "wiki_edit",
		"ticket_change", "ticket_field", "search_doc",
	} {
		if _, err := ix.db.Exec("DELETE FROM " + table); err != nil {
			return err
		}
	}
	return nil
}

// TimelineEntry is one check-in for the timeline op.
type TimelineEntry struct {
	Hash    string `json:"hash"`
	Date    string `json:"date"`
	User    string `json:"user"`
	Comment string `json:"comment"`
	Branch  string `json:"branch"`
}

// CheckinByPrefix resolves a hash prefix to exactly one check-in and its
// parents. ok is false when the prefix matches zero or several — an
// ambiguous prefix never resolves (the fossil rule).
func (ix *Index) CheckinByPrefix(prefix string) (TimelineEntry, []string, bool, error) {
	var e TimelineEntry
	rows, err := ix.db.Query(`SELECT hash, date, user, comment, branch FROM checkin
		WHERE hash LIKE ? || '%' LIMIT 2`, prefix)
	if err != nil {
		return e, nil, false, err
	}
	defer rows.Close()
	n := 0
	for rows.Next() {
		n++
		if n > 1 {
			return TimelineEntry{}, nil, false, nil // ambiguous
		}
		if err := rows.Scan(&e.Hash, &e.Date, &e.User, &e.Comment, &e.Branch); err != nil {
			return e, nil, false, err
		}
	}
	if err := rows.Err(); err != nil || n == 0 {
		return e, nil, false, err
	}
	prows, err := ix.db.Query(`SELECT parent FROM checkin_parent WHERE child = ? ORDER BY ord`, e.Hash)
	if err != nil {
		return e, nil, false, err
	}
	defer prows.Close()
	var parents []string
	for prows.Next() {
		var p string
		if err := prows.Scan(&p); err != nil {
			return e, nil, false, err
		}
		parents = append(parents, p)
	}
	return e, parents, true, prows.Err()
}

// WikiTitle is one page in the wiki listing: its name and latest edit.
type WikiTitle struct {
	Title string `json:"title"`
	Date  string `json:"date"`
	User  string `json:"user"`
}

// WikiTitles lists every wiki page by its most recent edit (fossil
// semantics: latest wins), sorted by title for determinism.
func (ix *Index) WikiTitles() ([]WikiTitle, error) {
	rows, err := ix.db.Query(`SELECT title, max(date || ' ' || hash), user FROM wiki_edit
		GROUP BY title ORDER BY title`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []WikiTitle
	for rows.Next() {
		var w WikiTitle
		var dateHash string
		if err := rows.Scan(&w.Title, &dateHash, &w.User); err != nil {
			return nil, err
		}
		w.Date, _, _ = strings.Cut(dateHash, " ")
		out = append(out, w)
	}
	return out, rows.Err()
}

// WikiEdit is one page's current content.
type WikiEdit struct {
	Title    string
	Date     string
	User     string
	Mimetype string
	Body     []byte
}

// WikiPage returns the latest edit of one page; ok is false when the
// page does not exist. Date ties break by hash, deterministically.
func (ix *Index) WikiPage(title string) (WikiEdit, bool, error) {
	var w WikiEdit
	err := ix.db.QueryRow(`SELECT title, date, user, mimetype, body FROM wiki_edit
		WHERE title = ? ORDER BY date DESC, hash DESC LIMIT 1`, title).
		Scan(&w.Title, &w.Date, &w.User, &w.Mimetype, &w.Body)
	if err == sql.ErrNoRows {
		return w, false, nil
	}
	return w, err == nil, err
}

// TicketChange is one recorded change with the fields it set.
type TicketChange struct {
	Hash   string           `json:"hash"`
	Date   string           `json:"date"`
	User   string           `json:"user"`
	Fields []TicketFieldSet `json:"fields"`
}

// TicketFieldSet is one field assignment within a change.
type TicketFieldSet struct {
	Name  string `json:"name"`
	Value string `json:"value"`
}

// TicketState is a ticket's folded state: every field's latest value
// (fossil semantics), plus the change history in order.
type TicketState struct {
	ID      string           `json:"id"`
	Created string           `json:"created"`
	Updated string           `json:"updated"`
	Fields  []TicketFieldSet `json:"fields"`
	Changes []TicketChange   `json:"changes"`
}

// Tickets folds every ticket's change history into current state,
// sorted by ticket id. Change order is (date, hash) so the fold is
// deterministic across rebuilds.
func (ix *Index) Tickets() ([]TicketState, error) {
	rows, err := ix.db.Query(`
		SELECT c.ticket, c.hash, c.date, c.user, f.name, f.value
		FROM ticket_change c JOIN ticket_field f ON f.hash = c.hash
		ORDER BY c.ticket, c.date, c.hash, f.name`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []TicketState
	var cur *TicketState
	for rows.Next() {
		var ticket, hash, date, user, name, value string
		if err := rows.Scan(&ticket, &hash, &date, &user, &name, &value); err != nil {
			return nil, err
		}
		if cur == nil || cur.ID != ticket {
			out = append(out, TicketState{ID: ticket, Created: date})
			cur = &out[len(out)-1]
		}
		cur.Updated = date
		if n := len(cur.Changes); n == 0 || cur.Changes[n-1].Hash != hash {
			cur.Changes = append(cur.Changes, TicketChange{Hash: hash, Date: date, User: user})
		}
		ch := &cur.Changes[len(cur.Changes)-1]
		ch.Fields = append(ch.Fields, TicketFieldSet{Name: name, Value: value})
		// Fold: latest assignment wins.
		set := false
		for i := range cur.Fields {
			if cur.Fields[i].Name == name {
				cur.Fields[i].Value = value
				set = true
			}
		}
		if !set {
			cur.Fields = append(cur.Fields, TicketFieldSet{Name: name, Value: value})
		}
	}
	return out, rows.Err()
}

// TicketByPrefix folds one ticket resolved by id prefix; ok is false on
// zero or multiple matches.
func (ix *Index) TicketByPrefix(prefix string) (TicketState, bool, error) {
	all, err := ix.Tickets()
	if err != nil {
		return TicketState{}, false, err
	}
	var found *TicketState
	for i := range all {
		if strings.HasPrefix(all[i].ID, prefix) {
			if found != nil {
				return TicketState{}, false, nil // ambiguous
			}
			found = &all[i]
		}
	}
	if found == nil {
		return TicketState{}, false, nil
	}
	return *found, true, nil
}

// Timeline lists check-ins newest first; ties break by hash so the
// listing is deterministic across rebuilds (the disposability property).
// A non-empty before (a hash prefix) starts the page strictly after
// that check-in; a non-empty branch filters to it; more reports whether
// older entries remain.
func (ix *Index) Timeline(limit int, before, branch string) ([]TimelineEntry, bool, error) {
	conds, args := []string{}, []any{}
	if before != "" {
		b, _, ok, err := ix.CheckinByPrefix(before)
		if err != nil {
			return nil, false, err
		}
		if !ok {
			return nil, false, nil // unknown cursor: empty page, not an error
		}
		conds = append(conds, `(date < ? OR (date = ? AND hash < ?))`)
		args = append(args, b.Date, b.Date, b.Hash)
	}
	if branch != "" {
		conds = append(conds, `branch = ?`)
		args = append(args, branch)
	}
	where := ""
	if len(conds) > 0 {
		where = "WHERE " + strings.Join(conds, " AND ")
	}
	args = append(args, limit+1)
	rows, err := ix.db.Query(`SELECT hash, date, user, comment, branch FROM checkin `+
		where+` ORDER BY date DESC, hash DESC LIMIT ?`, args...)
	if err != nil {
		return nil, false, err
	}
	defer rows.Close()
	var out []TimelineEntry
	for rows.Next() {
		var e TimelineEntry
		if err := rows.Scan(&e.Hash, &e.Date, &e.User, &e.Comment, &e.Branch); err != nil {
			return nil, false, err
		}
		out = append(out, e)
	}
	if err := rows.Err(); err != nil {
		return nil, false, err
	}
	if len(out) > limit {
		return out[:limit], true, nil
	}
	return out, false, nil
}

// CheckinCount reports the indexed check-in total — the fleet page's
// headline number.
func (ix *Index) CheckinCount() (int, error) {
	var n int
	err := ix.db.QueryRow(`SELECT count(*) FROM checkin`).Scan(&n)
	return n, err
}