Artifact [51e7578026]

Artifact 51e75780268e08f06a7f0eba2948d41f856ddadcef29777551f312daccfedcad:

• File cmd/quarry/main.go — part of check-in [f70144e381] at 2026-08-22 16:54:36 on branch trunk — quarry users pull: import a remote fossil's users and caps

  Fetch (admin-credentialed) then import: unknown logins are created with their fossil §6.3 credential and the project code it needs, existing accounts keep their own password, categories land as categories. An unmentioned category keeps quarry's default rather than blanking into a lockout. (user: dmestas, size: 50701) [annotate] [blame] [check-ins using] [more...]

// quarry is a multi-repository forge for zit. See docs/rfcs/.
package main

import (
	"bytes"
	"context"
	"craftdesign.group/quarry/internal/errlog"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"sort"
	"strings"
	"sync"
	"time"

	"craftdesign.group/quarry/internal/api"
	"craftdesign.group/quarry/internal/archive"
	"craftdesign.group/quarry/internal/chat"
	"craftdesign.group/quarry/internal/ci"
	"craftdesign.group/quarry/internal/events"
	"craftdesign.group/quarry/internal/index"
	"craftdesign.group/quarry/internal/ingest"
	"craftdesign.group/quarry/internal/lease"
	"craftdesign.group/quarry/internal/notify"
	"craftdesign.group/quarry/internal/oltp"
	"craftdesign.group/quarry/internal/ops"
	"craftdesign.group/quarry/internal/web"
	"craftdesign.group/quarry/internal/webhook"
	"craftdesign.group/quarry/internal/work"
	"craftdesign.group/quarry/internal/xfer"
	"craftdesign.group/quarry/internal/zitcli"
)

// version is stamped at build time; every page footer and the API report it
// (0001 Principle 1).
var version = "0.0.0-dev"

func main() {
	if len(os.Args) > 1 {
		switch os.Args[1] {
		case "version":
			fmt.Println("quarry", version)
			return
		case "spec":
			if err := printSpec(os.Stdout); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "serve":
			if err := serveCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "reindex":
			if err := reindexCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "docs":
			r := ops.NewRegistry()
			if err := api.RegisterAll(r, api.Deps{Version: version}); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			os.Stdout.Write(ops.Docs(r, "quarry"))
			return
		case "users":
			if err := usersCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "user":
			if err := userCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "webhook":
			if err := webhookCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "accounts":
			if err := accountAPICmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "repo":
			if err := repoCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		case "account":
			if err := accountCmd(os.Args[2:]); err != nil {
				fmt.Fprintln(os.Stderr, "quarry:", err)
				os.Exit(1)
			}
			return
		}
	}
	fmt.Fprintln(os.Stderr, "usage: quarry version|spec|serve|reindex|repo|accounts|user|users|account|webhook|docs")
	os.Exit(2)
}

// userCmd manages sync users (0001 §7 v1 auth): `quarry user add`
// derives the RFC §6.3 shared secret from the store's project code and
// stores it in forge OLTP — plaintext is never persisted. A repo with
// at least one sync user requires authenticated pushes.
func userCmd(args []string) error {
	if len(args) < 1 || args[0] != "add" {
		return fmt.Errorf("usage: quarry user add --repo name=path --user U [--password P] [--data DIR]")
	}
	fs := flag.NewFlagSet("user add", flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "repository as name=path")
	user := fs.String("user", "", "sync login name")
	password := fs.String("password", "", "password (omit to read from QUARRY_PASSWORD)")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside the repo)")
	if err := fs.Parse(args[1:]); err != nil {
		return err
	}
	if len(repos) != 1 || *user == "" {
		return fmt.Errorf("user add wants exactly one --repo and a --user")
	}
	pw := *password
	if pw == "" {
		pw = os.Getenv("QUARRY_PASSWORD")
	}
	if pw == "" {
		return fmt.Errorf("no password: use --password or QUARRY_PASSWORD")
	}
	var name, path string
	for n, p := range repos {
		name, path = n, p
	}
	dir := *data
	if dir == "" {
		dir = filepath.Join(path, ".quarry")
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return err
	}
	code, err := xfer.ProjectCode(path)
	if err != nil {
		return fmt.Errorf("project code of %s: %w", name, err)
	}
	db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
	if err != nil {
		return err
	}
	defer db.Close()
	if err := db.SetSyncUser(name, *user, xfer.SharedSecret(code, *user, pw)); err != nil {
		return err
	}
	fmt.Printf("sync user %q added to %s; pushes now require authentication\n", *user, name)
	return nil
}

// usersCmd pulls a remote fossil's user table into this forge
// (RFC 0004 §6). The credential is the OPERATOR's own account on that
// peer: fossil serves the user group only to Admin, and refusing a
// lesser credential is the remote's job, not ours. The password is
// taken from the environment so it never lands in shell history.
func usersCmd(args []string) error {
	if len(args) < 1 || args[0] != "pull" {
		return fmt.Errorf("usage: quarry users pull --repo name=path --url URL --user U [--password P] [--data DIR]")
	}
	fs := flag.NewFlagSet("users pull", flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "repository as name=path")
	peer := fs.String("url", "", "the peer's sync URL")
	user := fs.String("user", "", "an ADMIN login on the peer")
	password := fs.String("password", "", "that login's password (omit to read QUARRY_PASSWORD)")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside the repo)")
	dry := fs.Bool("dry-run", false, "report what would be imported and change nothing")
	if err := fs.Parse(args[1:]); err != nil {
		return err
	}
	if len(repos) != 1 || *peer == "" || *user == "" {
		return fmt.Errorf("users pull wants one --repo, a --url and a --user")
	}
	pw := *password
	if pw == "" {
		pw = os.Getenv("QUARRY_PASSWORD")
	}
	if pw == "" {
		return fmt.Errorf("no password: use --password or QUARRY_PASSWORD")
	}
	var name, path string
	for n, p := range repos {
		name, path = n, p
	}
	// The project code comes from OUR clone: the shared secret is
	// derived from it (§6.3), and a mirror always has one.
	code, err := xfer.ProjectCode(path)
	if err != nil {
		return fmt.Errorf("project code of %s: %w", name, err)
	}
	users, err := xfer.FetchUsers(nil, *peer, code, *user, pw)
	if err != nil {
		return err
	}
	rows := make([]oltp.ImportedUser, 0, len(users))
	for _, u := range users {
		rows = append(rows, oltp.ImportedUser{
			Login: u.Login, Caps: u.Caps, Info: u.Info, PWHash: u.PWHash,
		})
	}
	if *dry {
		for _, u := range rows {
			fmt.Printf("%-20s %s\n", u.Login, u.Caps)
		}
		fmt.Printf("%d users would be imported into %s\n", len(rows), name)
		return nil
	}
	dir := *data
	if dir == "" {
		dir = filepath.Join(path, ".quarry")
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return err
	}
	db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
	if err != nil {
		return err
	}
	defer db.Close()
	created, updated, err := db.ImportUsers(name, code, rows)
	if err != nil {
		return err
	}
	fmt.Printf("imported %d users into %s: %d new, %d updated\n",
		len(rows), name, created, updated)
	return nil
}

// reindexCmd is 0001 §5's full rebuild as a verb: for each --repo, drop
// the derived tables and cursor, then replay the store. Run it with the
// same --repo/--data flags as serve, while serve is stopped (the
// embedded NATS store directory admits one process).
func reindexCmd(args []string) error {
	fs := flag.NewFlagSet("reindex", flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "repository as name=path (repeatable; bare path uses its basename)")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside a SINGLE repo; required for a fleet)")
	natsURL := fs.String("nats", "", "external NATS URL (multi-node); reindex then publishes a cluster-wide reset that every node's fold absorbs")
	if err := fs.Parse(args); err != nil {
		return err
	}
	if len(repos) == 0 {
		return fmt.Errorf("reindex wants at least one --repo")
	}
	dir, err := dataDir(*data, repos)
	if err != nil {
		return err
	}
	// On external NATS the reset republishes the census to the shared
	// stream, so every node re-folds (idempotent) — reindex is a
	// cluster-wide operation there, not a local-only one. Embedded is the
	// offline single-node rebuild (serve must be stopped: one process per
	// store dir).
	ev, err := startEvents(*natsURL, dir)
	if err != nil {
		return fmt.Errorf("%w (is serve still running against this data dir?)", err)
	}
	defer ev.Close()
	for name, path := range repos {
		if err := os.MkdirAll(filepath.Join(dir, name), 0o755); err != nil {
			return err
		}
		ix, err := index.Open(filepath.Join(dir, name, "index.db"))
		if err != nil {
			return err
		}
		n, err := ingest.NewPipeline(zitcli.New(path), ev, ix, name, "reindex").Reindex()
		ix.Close()
		if err != nil {
			return fmt.Errorf("reindex %s: %w", name, err)
		}
		fmt.Printf("reindexed %s: %d artifacts replayed\n", name, n)
	}
	return nil
}

// repoFlags collects repeatable --repo name=path values (bare paths take
// their directory basename as the repository name).
type repoFlags map[string]string

func (r repoFlags) String() string { return fmt.Sprint(map[string]string(r)) }

func (r repoFlags) Set(v string) error {
	name, path, found := strings.Cut(v, "=")
	if !found {
		path = v
		name = filepath.Base(strings.TrimSuffix(v, "/"))
	}
	if name == "" || path == "" {
		return fmt.Errorf("--repo wants name=path or path, got %q", v)
	}
	if !repoName.MatchString(name) {
		return fmt.Errorf("repository name %q: use letters, digits, dot, dash, underscore", name)
	}
	if _, dup := r[name]; dup {
		return fmt.Errorf("repository %q registered twice", name)
	}
	r[name] = path
	return nil
}

var repoName = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$`)

// serveCmd runs the HTTP surface: the op-generated API plus the HTML
// pages, both consumers of one registry. Each --repo gets its own index
// and watcher pipeline; one embedded NATS carries every store's events
// under zit.evt.<name>.* (0001 §4: a watcher per registered store).

// dataDir resolves the derived-data directory. With one repository the
// default is .quarry beside it — deterministic and obviously derived.
// With several there IS no principled default: the old rule took
// "the first --repo", which for a Go map was a coin flip per boot, and
// on 2026-08-16 it planted indexes, NATS state and the OLTP database
// inside a corpus meant to hold only source repositories. 0001 §11
// says derived data is disposable; that only helps if you can tell
// which directory is the derived one — so a fleet must say.
func dataDir(explicit string, repos map[string]string) (string, error) {
	if explicit != "" {
		return explicit, nil
	}
	if len(repos) == 1 {
		for _, p := range repos {
			return filepath.Join(p, ".quarry"), nil
		}
	}
	names := make([]string, 0, len(repos))
	for n := range repos {
		names = append(names, n)
	}
	sort.Strings(names)
	return "", fmt.Errorf("serving %d repositories needs an explicit --data: "+
		"there is no principled default, and the old rule silently wrote "+
		"indexes, event-stream state and the account database into an "+
		"arbitrary one of %v", len(repos), names)
}

func serveCmd(args []string) error {
	fs := flag.NewFlagSet("serve", flag.ContinueOnError)
	addr := fs.String("addr", "127.0.0.1:8080", "listen address")
	repos := repoFlags{}
	fs.Var(repos, "repo", "repository to serve as name=path (repeatable; bare path uses its basename)")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside a SINGLE repo; required for a fleet)")
	interval := fs.Duration("sync-interval", 2*time.Second, "store poll interval")
	leaseTTL := fs.Duration("lease-ttl", lease.DefaultTTL, "single-writer lease lifetime")
	natsURL := fs.String("nats", "", "external NATS URL for multi-node (default: embedded, single node); the server must allow max_payload >= 8MiB")
	notifyURL := fs.String("notify", "", "notify companion base URL for mentions and activity (default: notifications disabled)")
	dagnatsURL := fs.String("dagnats", "", "dagnats control-plane URL for CI (0001 §6); empty disables CI")
	ciVerdicts := fs.Bool("ci-verdicts", true, "record ci-pass/ci-fail as tag artifacts on the check-in, so CI history clones with the repository")
	ciPoll := fs.Duration("ci-poll", 10*time.Second, "how often to refresh unfinished CI runs")
	// How often DUE-ness is evaluated, not how often anything is pulled:
	// each mirror carries its own interval. A minute is fine because the
	// loop does nothing when no mirror is due.
	mirrorTick := fs.Duration("mirror-tick", time.Minute, "how often to check for due mirrors")
	// Bulk work is bounded by memory, not cores: each archive buffers its
	// whole output. Two at a time with a short queue keeps the worst case
	// near half a gigabyte instead of unbounded.
	bulkLimit := fs.Int("bulk-limit", 2, "concurrent bulk operations (archive)")
	// Where repositories created or cloned through the API are placed.
	// Without it an API caller must supply a filesystem path, which only
	// the machine's operator can meaningfully choose.
	reposRoot := fs.String("repos-root", "", "directory for repositories created through the API")
	bulkQueue := fs.Int("bulk-queue", 8, "bulk operations allowed to wait before refusing")
	archiveCache := fs.Int64("archive-cache", 2<<30, "bytes of built archives to keep on disk (0 disables)")
	// Ingest is bounded because it is the one path that streams a WHOLE
	// store: `zit frames` on sqlite peaks at 507 MB and tcl is larger.
	// One goroutine per repository with nothing between them means a
	// cold start, or a mirror round that lands artifacts in several big
	// stores at once, runs that many censuses simultaneously and the
	// memory adds. Observed: zit spiking to gigabytes after pulling
	// sqlite and tcl together.
	ingestLimit := fs.Int("ingest-limit", 3, "stores that may ingest at once")
	errorlogPath := fs.String("errorlog", "", "append server errors to this file (the in-process journal restarts with the server; this survives)")
	forumHook := fs.String("forum-hook", "", "AI moderation command: staged-post JSON on stdin, {action,score,reason} JSON on stdout; failures hold for a human")
	tracker := fs.String("tracker", "", "repository whose tickets are the forge's own; enables /fleet's file-ticket buttons")
	notifyKey := fs.String("notify-key", "", "notify console API key (or QUARRY_NOTIFY_KEY); notify's /v1 routes answer 401 without one")
	if err := fs.Parse(args); err != nil {
		return err
	}
	// Refuse leftovers. Go's flag package stops at the first non-flag
	// argument and hands the rest back, so a malformed invocation — one
	// unquoted shell variable is enough — silently drops every flag past
	// that point. Doing this cost twelve --repo flags once, and quarry
	// started and served the remaining eleven without a word.
	if rest := fs.Args(); len(rest) > 0 {
		return fmt.Errorf("unexpected argument %q: serve takes only flags "+
			"(a shell variable holding flags must be word-split, e.g. \"${args[@]}\")", rest[0])
	}

	deps := api.Deps{Version: version, Repos: api.NewRepos(nil), Errors: errlog.Default}
	var notifier web.Notifier
	var chatLive web.ChatLive
	var verify api.Verify
	var sessions web.Sessions
	// Capability enforcement (RFC 0004 §4). nil until the accounts
	// plane exists — a forge without accounts has no caps to check and
	// keeps the Auth floor as its whole story.
	var capsFor api.CapsFor
	var capsAdmin web.Caps
	var inbox web.Inbox
	var engine *ci.Engine
	var ciCloneBase string
	var fronter *xfer.Fronter
	if len(repos) > 0 {
		fronter = xfer.New()
		defer fronter.Close()
		// Registration happens in the registrar below, which owns the
		// whole lifetime; registering here too would mean two places
		// that must agree about which repositories exist.
	}
	if len(repos) > 0 {
		dir, err := dataDir(*data, repos)
		if err != nil {
			return err
		}
		if err := os.MkdirAll(dir, 0o755); err != nil {
			return err
		}
		odb, err := oltp.Open(filepath.Join(dir, "oltp.db"))
		if err != nil {
			return err
		}
		defer odb.Close()
		fronter.SetAuth(odb)
		// Two credentials, on purpose (0001 §8). Accounts are the forge
		// identity and carry per-repo roles; the RFC §6.3 shared secret
		// stays the SYNC credential, because the protocol fixes its
		// derivation and the server must be able to recompute it. Once
		// any account exists it governs web/API writes; before that,
		// v1's per-repo secret still does, so an existing deployment
		// keeps working and a fresh local one stays open.
		// Cookie sessions for the browser; Basic stays for the API.
		sessions = sessionPlane{db: odb}
		var codeMu sync.Mutex
		codes := map[string]string{}
		verify = func(repo, user, pass string, need ops.Auth) api.Decision {
			if odb.HasAccounts() {
				acct, ok := odb.Authenticate(user, pass)
				if !ok {
					return api.Denied
				}
				// Authenticated. From here a refusal is about capability,
				// not identity, and saying so is the difference between a
				// client that stops and one that retries forever.
				if odb.RoleOf(acct, repo) >= requiredRole(need) {
					return api.Allowed
				}
				return api.Insufficient
			}
			if !odb.Gated(repo) {
				return api.Allowed
			}
			secret, ok := odb.Secret(repo, user)
			if !ok || pass == "" {
				return api.Denied
			}
			codeMu.Lock()
			code, cached := codes[repo]
			codeMu.Unlock()
			if !cached {
				path, ok := repos[repo]
				if !ok {
					return api.Denied
				}
				c, err := xfer.ProjectCode(path)
				if err != nil {
					log.Printf("auth: project code of %s: %v", repo, err)
					return api.Denied
				}
				codeMu.Lock()
				codes[repo], code = c, c
				codeMu.Unlock()
			}
			if secret == xfer.SharedSecret(code, user, pass) {
				return api.Allowed
			}
			return api.Denied
		}

		ev, err := startEvents(*natsURL, dir)
		if err != nil {
			return err
		}
		defer ev.Close()
		notifier = ev

		// Single-writer coordination (0001 §3). node identifies this
		// process; one embedded NATS means one node, but the machinery
		// and its failure posture (coordinator down → writes refuse) run
		// the same on one node as on N.
		leases, err := lease.New(ev.JS(), *addr, *leaseTTL)
		if err != nil {
			return err
		}
		defer leases.Close()
		// Per-repo chat plane (0001 §5): messages fan out over the same
		// NATS the events ride, with a 7-day JetStream window.
		room, err := chat.New(ev.Conn(), ev.JS())
		if err != nil {
			return err
		}
		deps.Chat = room
		chatLive = &chatAdapter{room: room, codes: codes, mu: &codeMu}
		// Notification triggers (0001 §5): fired best-effort to the notify
		// companion; disabled when --notify is unset.
		if *notifyURL != "" {
			key := *notifyKey
			if key == "" {
				key = os.Getenv("QUARRY_NOTIFY_KEY")
			}
			if key == "" {
				// Better to say so at startup than to discover it as 401s
				// in the log after a write nobody was told about.
				log.Printf("notify: no API key (--notify-key or QUARRY_NOTIFY_KEY); triggers will be refused")
			}
			nclient := notify.New(*notifyURL, key)
			deps.Notify = &notifyAdapter{client: nclient}
			inbox = inboxPlane{client: nclient}
			fmt.Println("quarry: notifications ->", *notifyURL)
			// A server error's FIRST sighting notifies; repeats only
			// count in the journal — an alert channel that repeats
			// itself gets muted. Signature dedup makes this cheap.
			np := deps.Notify
			errlog.Default.OnFirst = func(e errlog.Entry) {
				_ = np.Trigger(context.Background(), api.NotifyEvent{
					Workflow: "quarry-error", To: "admins",
					Payload: map[string]any{
						"surface": e.Surface, "path": e.Path,
						"summary": e.Message,
					},
					TxnID: "err:" + e.First.Format(time.RFC3339Nano),
				})
			}
		}
		// CI (0001 §6): dagnats is the engine, quarry the trigger and
		// display. Disabled unless --dagnats names one, so a forge with
		// no engine simply has no CI rather than a broken one.
		engine = ci.New(*dagnatsURL)
		if engine.Enabled() {
			// The clone URL handed to a pipeline: quarry's own sync
			// fronting, on the address quarry was told to listen on.
			ciCloneBase = "http://" + *addr + "/r/"
			deps.CI = ciRuns{db: odb}
			fmt.Println("quarry: CI ->", *dagnatsURL)
			go func() {
				for range time.Tick(*ciPoll) {
					pollCI(engine, odb, deps.Repos, *ciVerdicts)
				}
			}()
		}
		// Mirroring: always available, because it costs nothing until an
		// admin configures an upstream — the scheduler wakes, finds no
		// rows, and sleeps. The tick is how often DUE-ness is checked,
		// not how often anything is pulled; each mirror's own interval
		// decides that.
		// One bulk lane for the whole process, not per repository: the
		// resource being protected is this machine's memory, and eleven
		// per-repo lanes of two would be a bound of twenty-two.
		deps.Bulk = work.NewLane("bulk", *bulkLimit, *bulkQueue)
		// Archives are reproducible, so a cached one is never stale and
		// the budget is purely a disk-space choice. Without it every
		// reader of a popular tag pays the same seconds of compression.
		if *archiveCache > 0 {
			ac, err := archive.New(filepath.Join(dir, "archives"), *archiveCache)
			if err != nil {
				return err
			}
			deps.Archives = ac
		}
		// The queue is per-repository and generous: ingest must WAIT,
		// never be shed. Dropping a round would leave an index behind
		// its store until the next poll, which is a correctness surprise
		// in exchange for a memory saving the limit already delivers.
		ingestLane := work.NewLane("ingest", *ingestLimit, len(repos)+1)
		deps.Mirrors = mirrorStore{db: odb}
		// After Mirrors is set: Writable reads it, and deps is a value.
		fronter.SetWriteGate(deps.Writable)
		go mirrorScheduler(&deps, odb, *mirrorTick)
		// Webhook dispatch (Principle 6): always on, because it costs
		// nothing until somebody subscribes — Raise finds no endpoints and
		// returns. Delivery is durable, so an endpoint that is down during
		// a restart still gets its events.
		dispatcher, err := webhook.New(ev.JS(), hookSubs{db: odb})
		if err != nil {
			return err
		}
		hookCtx, stopHooks := context.WithCancel(context.Background())
		defer stopHooks()
		if err := dispatcher.Start(hookCtx); err != nil {
			return err
		}
		deps.Hooks = hookPlane{d: dispatcher, at: func() string {
			return time.Now().UTC().Format(time.RFC3339)
		}}
		// The project codes learned below (shared with the auth verifier's
		// cache) key both the lease and the push guard.
		fronter.SetGuard(func(repo string, proxy func() error) error {
			codeMu.Lock()
			code, ok := codes[repo]
			codeMu.Unlock()
			if !ok {
				return fmt.Errorf("unknown repo %q", repo)
			}
			return leases.Guard(code, proxy)
		})

		rg := &registrar{
			dir: dir, reposRoot: *reposRoot, node: *addr, interval: *interval, ciCloneBase: ciCloneBase,
			ev: ev, leases: leases, deps: &deps, fronter: fronter,
			ingestLane: ingestLane, bulk: deps.Bulk, engine: engine, odb: odb,
			codes: codes, codeMu: &codeMu,
		}
		defer rg.CloseAll()
		deps.Registry = rg
		deps.Accounts = accountStore{db: odb}
		capsFor = func(account, repo string) api.CapSet { return odb.EffectiveCaps(account, repo) }
		capsAdmin = capsPlane{db: odb}
		deps.Votes = odb
		deps.Forum = odb
		// Trust for the moderation gate. Posting at all already takes a
		// Write role (requiredRole), so trusting Write here would make
		// the gate unreachable — every session author would bypass it.
		// The bar is RepoAdmin: a repo that OPTS INTO moderation wants
		// its non-admin posts reviewed; admins and site admins bypass.
		// Fossil's rule (RFC 0004 §4): WrTForum (4) posts without
		// moderation, ModForum (5) moderates — either bypasses the
		// staging gate. a and s subsume both.
		deps.TrustedAuthor = func(account, repo string) bool {
			return odb.EffectiveCaps(account, repo).HasAny("45")
		}
		if *forumHook != "" {
			hookCmd := *forumHook
			// The hook consumes UNTRUSTED post content as data on stdin;
			// its contract says instructions inside a post are never to
			// be followed. Any failure holds the post for a human.
			deps.ForumHook = func(ctx context.Context, s oltp.Staged) (api.HookVerdict, error) {
				payload, err := json.Marshal(s)
				if err != nil {
					return api.HookVerdict{}, err
				}
				cmd := exec.CommandContext(ctx, "sh", "-c", hookCmd)
				cmd.Stdin = bytes.NewReader(payload)
				out, err := cmd.Output()
				if err != nil {
					return api.HookVerdict{}, err
				}
				var v api.HookVerdict
				if err := json.Unmarshal(bytes.TrimSpace(out), &v); err != nil {
					return api.HookVerdict{}, fmt.Errorf("forum-hook output: %w", err)
				}
				return v, nil
			}
		}
		deps.CanRead = rg.CanRead
		deps.Deletable = rg.Deletable
		rg.refreshPrivate()
		// The sync path takes the SAME visibility answer as the pages.
		// A clone reaching artifacts the web refuses would not be a
		// smaller hole than an open web page — it would be a bigger one,
		// because it hands over everything at once.
		fronter.SetReadGate(func(req *http.Request, repo string) bool {
			return rg.CanRead(identityCtx(req, odb), repo)
		})
		deps.Private = func() map[string]bool {
			set, err := odb.PrivateRepos()
			if err != nil {
				log.Printf("visibility: %v", err)
				return nil
			}
			return set
		}
		deps.Descriptions = func() map[string]string {
			rows, err := odb.Repos()
			if err != nil {
				log.Printf("repo descriptions: %v", err)
				return nil
			}
			m := make(map[string]string, len(rows))
			for _, r := range rows {
				m[r.Name] = r.Description
			}
			return m
		}
		deps.RepoMeta = func() map[string]api.RepoMeta {
			rows, err := odb.Repos()
			if err != nil {
				log.Printf("repo meta: %v", err)
				return nil
			}
			m := make(map[string]api.RepoMeta, len(rows))
			for _, r := range rows {
				m[r.Name] = api.RepoMeta{Homepage: r.Homepage, Topics: strings.Fields(r.Topics)}
			}
			return m
		}
		// Persisted registrations first, then --repo. The flag stays as
		// bootstrap and override — it is how the first repository gets
		// in, and how an operator recovers a registry they cannot read —
		// but it is no longer the only source of truth.
		if odb != nil {
			stored, rerr := odb.Repos()
			if rerr != nil {
				return rerr
			}
			for _, r := range stored {
				if _, dup := repos[r.Name]; dup {
					continue // --repo wins: an explicit flag is the override
				}
				if err := rg.Register(r.Name, r.Path); err != nil {
					// One unreadable stored repository must not stop the
					// server: the rest are fine and the operator needs a
					// running forge to fix this one.
					log.Printf("registered repository %s (%s): %v", r.Name, r.Path, err)
				}
			}
		}
		for name, path := range repos {
			if err := rg.Register(name, path); err != nil {
				return err
			}
			// A flag-named repository still belongs in the registry, or
			// it cannot be described, taken private, or managed at all —
			// it would be served and simultaneously invisible to every
			// admin surface.
			if odb != nil {
				if err := odb.EnsureRepo(name, path); err != nil {
					log.Printf("registry: recording %s: %v", name, err)
				}
			}
		}
		rg.refreshPrivate()
	}

	if *errorlogPath != "" {
		f, ferr := os.OpenFile(*errorlogPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
		if ferr != nil {
			return fmt.Errorf("errorlog: %w", ferr)
		}
		errlog.Default.SetOutput(f)
	}
	mux, err := buildHandler(deps, notifier, chatLive, verify, capsFor, capsAdmin, sessions, inbox, *tracker)
	if err != nil {
		return err
	}
	if fronter != nil {
		mount := http.NewServeMux()
		// Sync fronting (0001 §7): zit clients POST the repo URL
		// verbatim; /xfer is the alias zit serve itself answers on.
		mount.Handle("POST /r/{repo}", fronter)
		mount.Handle("POST /r/{repo}/xfer", fronter)
		mount.Handle("/", mux)
		mux = mount
	}
	fmt.Println("quarry", version, "listening on", *addr)
	return http.ListenAndServe(*addr, mux)
}

// chatAdapter bridges the chat data plane to web.ChatLive: it maps a
// repo name to its project code and converts messages to the view type,
// keeping web free of the data-plane package.
type chatAdapter struct {
	room  *chat.Room
	codes map[string]string
	mu    *sync.Mutex
}

func (a *chatAdapter) SubscribeChat(repo string) (<-chan web.ChatMessage, func()) {
	a.mu.Lock()
	code, ok := a.codes[repo]
	a.mu.Unlock()
	if !ok {
		ch := make(chan web.ChatMessage)
		close(ch)
		return ch, func() {}
	}
	src, cancel := a.room.Subscribe(code)
	out := make(chan web.ChatMessage, 32)
	go func() {
		defer close(out)
		for m := range src {
			select {
			case out <- web.ChatMessage{User: m.User, Text: m.Text, At: m.At}:
			default:
			}
		}
	}()
	return out, cancel
}

// notifyAdapter bridges api.NotifyPlane to the notify client, mapping a
// quarry event to a Novu-shaped trigger. The event's To is the
// subscriber id (a user or a "repo:<name>" topic key).
type notifyAdapter struct{ client *notify.Client }

func (a *notifyAdapter) Trigger(ctx context.Context, e api.NotifyEvent) error {
	// NotifyEvent.To is documented as "a user, or a repo:<name> topic".
	// notify addresses those through different fields — subscriberId
	// names one person, topicKey fans out — so the convention has to be
	// honoured here or an activity event reaches nobody: it lands on an
	// invented subscriber literally called "repo:synthetic" while every
	// watcher of that repository hears nothing.
	to := notify.Recipient{SubscriberID: e.To}
	if repo, ok := strings.CutPrefix(e.To, "repo:"); ok && repo != "" {
		to = notify.Recipient{TopicKey: repoTopicKey(repo)}
	}
	return a.client.Trigger(ctx, e.Workflow, to, e.Payload, e.TxnID)
}

// startEvents opens the event plane: an external NATS when a URL is
// given (multi-node), else an embedded server under dir (single node).
func startEvents(url, dir string) (*events.Server, error) {
	if url != "" {
		fmt.Println("quarry: connecting to external NATS", url)
		return events.Connect(url)
	}
	return events.Start(filepath.Join(dir, "nats"))
}

// buildHandler wires the two registry consumers onto one mux. notifier
// may be nil (no live pages).
func buildHandler(deps api.Deps, notifier web.Notifier, chatLive web.ChatLive, verify api.Verify, capsFor api.CapsFor, capsAdmin web.Caps, sessions web.Sessions, inbox web.Inbox, tracker string) (http.Handler, error) {
	r := ops.NewRegistry()
	if err := api.RegisterAll(r, deps); err != nil {
		return nil, err
	}
	// The router refines each op's Auth floor with the letters that
	// op's surface needs.
	apiRouter, err := api.RouterWithCaps(r, deps.Version, capsFor, verify)
	if err != nil {
		return nil, err
	}
	mux := http.NewServeMux()
	mux.Handle("/api/", apiRouter)
	var opts []web.Option
	if sessions != nil {
		opts = append(opts, web.WithSessions(sessions))
		if capsAdmin != nil {
			opts = append(opts, web.WithCaps(capsAdmin))
		}
	}
	if tracker != "" {
		opts = append(opts, web.WithTracker(tracker))
	}
	if inbox != nil {
		opts = append(opts, web.WithInbox(inbox))
	}
	mux.Handle("/", web.HandlerWith(r, notifier, chatLive, opts, web.Verify(verify)))
	return mux, nil
}

// printSpec emits the OpenAPI document generated from the full registry —
// the same registry serve will run, so the spec cannot drift (0001 §9).
func printSpec(w *os.File) error {
	r := ops.NewRegistry()
	if err := api.RegisterAll(r, api.Deps{Version: version}); err != nil {
		return err
	}
	spec, err := ops.Spec(r, "quarry", version)
	if err != nil {
		return err
	}
	enc := json.NewEncoder(w)
	enc.SetIndent("", "  ")
	return enc.Encode(spec)
}

// requiredRole maps an op's declared capability onto the role an account
// must hold on that repo. This is the only place the ops vocabulary and
// the ACL vocabulary meet, so adding a level is a change here and
// nowhere else.
func requiredRole(need ops.Auth) oltp.Role {
	switch need {
	case ops.Admin:
		return oltp.RepoAdmin
	case ops.Token:
		return oltp.Write
	}
	return oltp.NoRole
}

// accountCmd manages forge accounts and per-repo roles (0001 §8's v2
// tables). Distinct from `quarry user add`, which writes the SYNC
// credential: that one is the RFC §6.3 shared secret a push presents,
// this one is the login a person uses. --repo-name names a REGISTERED
// repository, because roles are held against served names, not paths.
func accountCmd(args []string) error {
	const usage = `usage:
  quarry account add    --user U [--password P] [--admin]
  quarry account passwd --user U [--password P]
  quarry account grant  --user U --repo-name N --role read|write|admin
  quarry account revoke --user U --repo-name N
  quarry account delete --user U
  quarry account list
every form also takes --data DIR, or one --repo name=path to locate it`
	if len(args) < 1 {
		return errors.New(usage)
	}
	sub := args[0]
	fs := flag.NewFlagSet("account "+sub, flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "a registered repository as name=path, used only to locate the data directory")
	data := fs.String("data", "", "derived-data directory (default: .quarry beside the repo)")
	user := fs.String("user", "", "account name")
	password := fs.String("password", "", "password (omit to read from QUARRY_PASSWORD)")
	repoTarget := fs.String("repo-name", "", "repository the role applies to")
	role := fs.String("role", "", "read, write or admin")
	admin := fs.Bool("admin", false, "site admin: every repository, every capability")
	if err := fs.Parse(args[1:]); err != nil {
		return err
	}

	dir := *data
	if dir == "" {
		if len(repos) != 1 {
			return errors.New("need --data DIR, or one --repo name=path to locate it\n" + usage)
		}
		for _, p := range repos {
			dir = filepath.Join(p, ".quarry")
		}
	}
	db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
	if err != nil {
		return err
	}
	defer db.Close()

	// Password sources in order: the flag, then the environment, so a
	// password need not reach the process table or shell history.
	readPassword := func() (string, error) {
		if *password != "" {
			return *password, nil
		}
		if pw := os.Getenv("QUARRY_PASSWORD"); pw != "" {
			return pw, nil
		}
		return "", errors.New("no password: use --password or QUARRY_PASSWORD")
	}

	switch sub {
	case "add":
		if *user == "" {
			return errors.New("account add wants --user")
		}
		pw, err := readPassword()
		if err != nil {
			return err
		}
		first := !db.HasAccounts()
		if err := db.AddAccount(*user, pw, *admin); err != nil {
			return err
		}
		if *admin {
			fmt.Printf("account %q created (site admin)\n", *user)
		} else {
			fmt.Printf("account %q created\n", *user)
		}
		if first {
			fmt.Println("this is the first account: web and API writes now require one")
		}
		return nil

	case "passwd":
		if *user == "" {
			return errors.New("account passwd wants --user")
		}
		pw, err := readPassword()
		if err != nil {
			return err
		}
		if err := db.SetPassword(*user, pw); err != nil {
			return err
		}
		fmt.Printf("password changed for %q\n", *user)
		return nil

	case "grant":
		if *user == "" || *repoTarget == "" || *role == "" {
			return errors.New("account grant wants --user, --repo-name and --role")
		}
		r, err := oltp.ParseRole(*role)
		if err != nil {
			return err
		}
		if err := db.Grant(*user, *repoTarget, r); err != nil {
			return err
		}
		fmt.Printf("%q now has %s on %s\n", *user, r, *repoTarget)
		return nil

	case "revoke":
		if *user == "" || *repoTarget == "" {
			return errors.New("account revoke wants --user and --repo-name")
		}
		if err := db.Revoke(*user, *repoTarget); err != nil {
			return err
		}
		fmt.Printf("%q no longer has a role on %s\n", *user, *repoTarget)
		return nil

	case "delete":
		if *user == "" {
			return errors.New("account delete wants --user")
		}
		if err := db.DeleteAccount(*user); err != nil {
			return err
		}
		fmt.Printf("account %q deleted, with every role it held\n", *user)
		return nil

	case "list":
		accts, err := db.Accounts()
		if err != nil {
			return err
		}
		if len(accts) == 0 {
			fmt.Println("no accounts: web and API writes are open (single-user local serving)")
			return nil
		}
		for _, a := range accts {
			line := a.Name
			if a.Admin {
				line += " (site admin)"
			}
			grants, err := db.GrantsFor(a.Name)
			if err != nil {
				return err
			}
			for _, repo := range oltp.SortedRepos(grants) {
				line += fmt.Sprintf("  %s=%s", repo, grants[repo])
			}
			fmt.Println(line)
		}
		return nil
	}
	return errors.New(usage)
}

// hookSubs adapts the OLTP database to the dispatcher's Subscriptions
// view. The two packages keep their own types on purpose: oltp owns the
// stored row (secrets, timestamps, paused flag), webhook owns what a
// delivery needs, and this is the only place they meet.
type hookSubs struct{ db *oltp.DB }

func (h hookSubs) ActiveWebhooks(repo string) ([]webhook.Sub, error) {
	rows, err := h.db.ActiveWebhooks(repo)
	if err != nil {
		return nil, err
	}
	out := make([]webhook.Sub, 0, len(rows))
	for _, w := range rows {
		out = append(out, webhook.Sub{ID: w.ID, URL: w.URL, Secret: w.Secret, Active: true})
	}
	return out, nil
}

func (h hookSubs) SubByID(id int64) (webhook.Sub, bool) {
	w, ok := h.db.WebhookByID(id)
	if !ok {
		return webhook.Sub{}, false
	}
	return webhook.Sub{ID: w.ID, URL: w.URL, Secret: w.Secret, Active: w.Active}, true
}

// hookPlane adapts the dispatcher to the ops layer's WebhookPlane, which
// speaks primitives so the ops never import the delivery machinery.
type hookPlane struct {
	d  *webhook.Dispatcher
	at func() string
}

func (h hookPlane) Raise(id, repo, kind, actor, summary, link string) error {
	return h.d.Raise(webhook.Event{
		ID: id, Repo: repo, Kind: kind, Actor: actor,
		Summary: summary, Link: link, At: h.at(),
	})
}

// webhookCmd manages subscriptions to quarry's outbound event surface
// (Principle 6). The signing secret is printed once at creation and
// never again — a key a listing can read back is a key that leaks
// through logs, screenshots and shoulders.
func webhookCmd(args []string) error {
	const usage = `usage:
  quarry webhook add    --repo-name N --url https://...
  quarry webhook list   --repo-name N
  quarry webhook pause  --id N
  quarry webhook resume --id N
  quarry webhook delete --id N
every form also takes --data DIR, or one --repo name=path to locate it`
	if len(args) < 1 {
		return errors.New(usage)
	}
	sub := args[0]
	fs := flag.NewFlagSet("webhook "+sub, flag.ContinueOnError)
	repos := repoFlags{}
	fs.Var(repos, "repo", "a registered repository as name=path, used only to locate the data directory")
	data := fs.String("data", "", "derived-data directory")
	repoTarget := fs.String("repo-name", "", "repository whose events to deliver")
	endpoint := fs.String("url", "", "endpoint to POST events to")
	id := fs.Int64("id", 0, "subscription id")
	if err := fs.Parse(args[1:]); err != nil {
		return err
	}
	dir := *data
	if dir == "" {
		if len(repos) != 1 {
			return errors.New("need --data DIR, or one --repo name=path to locate it\n" + usage)
		}
		for _, p := range repos {
			dir = filepath.Join(p, ".quarry")
		}
	}
	db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
	if err != nil {
		return err
	}
	defer db.Close()

	switch sub {
	case "add":
		if *repoTarget == "" || *endpoint == "" {
			return errors.New("webhook add wants --repo-name and --url")
		}
		w, err := db.AddWebhook(*repoTarget, *endpoint)
		if err != nil {
			return err
		}
		fmt.Printf("webhook %d: %s -> %s\n", w.ID, w.Repo, w.URL)
		fmt.Printf("signing secret (shown once): %s\n", w.Secret)
		fmt.Println("verify X-Quarry-Signature as sha256=HMAC-SHA256(secret, body)")
		return nil
	case "list":
		if *repoTarget == "" {
			return errors.New("webhook list wants --repo-name")
		}
		hooks, err := db.Webhooks(*repoTarget)
		if err != nil {
			return err
		}
		if len(hooks) == 0 {
			fmt.Printf("no webhooks on %s\n", *repoTarget)
			return nil
		}
		for _, w := range hooks {
			state := "active"
			if !w.Active {
				state = "paused"
			}
			fmt.Printf("%d  %-7s %s  (added %s)\n", w.ID, state, w.URL, w.Created)
		}
		return nil
	case "pause", "resume":
		if *id == 0 {
			return fmt.Errorf("webhook %s wants --id", sub)
		}
		if err := db.SetWebhookActive(*id, sub == "resume"); err != nil {
			return err
		}
		fmt.Printf("webhook %d %sd\n", *id, sub)
		return nil
	case "delete":
		if *id == 0 {
			return errors.New("webhook delete wants --id")
		}
		if err := db.DeleteWebhook(*id); err != nil {
			return err
		}
		fmt.Printf("webhook %d deleted\n", *id)
		return nil
	}
	return errors.New(usage)
}

// sessionPlane adapts the OLTP database to the web's Sessions seam. The
// authorization half deliberately reuses the same role lookup the Basic
// path uses, so a cookie can never be worth more than a password.
type sessionPlane struct{ db *oltp.DB }

func (s sessionPlane) Login(user, password string) (string, time.Time, bool) {
	return s.db.NewSession(user, password)
}

func (s sessionPlane) Resolve(token string) (string, bool) {
	acct, ok := s.db.SessionAccount(token)
	return acct.Name, ok
}

func (s sessionPlane) Logout(token string) { s.db.EndSession(token) }

// CapsOf implements web.Caps: the page layer gates affordances on the
// same expanded capability set the API enforces.
func (s sessionPlane) CapsOf(account, repo string) web.CapSet {
	return s.db.EffectiveCaps(account, repo)
}

func (s sessionPlane) Allow(user, repo string, need ops.Auth) bool {
	if !s.db.HasAccounts() {
		return true
	}
	// The caller has already proven the cookie, so this is authorization,
	// not authentication — but it runs the SAME role lookup the password
	// path does, so a cookie is never worth more than a password.
	a, found := s.db.AccountByName(user)
	if !found {
		return false
	}
	return s.db.RoleOf(a, repo) >= requiredRole(need)
}

// capsPlane is the web layer's window onto capabilities: the resolver
// every page gate reads, and the administration seam the settings
// editor writes through (RFC 0004 §5).
type capsPlane struct{ db *oltp.DB }

func (c capsPlane) CapsOf(account, repo string) web.CapSet {
	return c.db.EffectiveCaps(account, repo)
}

func (c capsPlane) CapHolders(repo string) ([]web.CapHolder, error) {
	rows, err := c.db.CapHolders(repo)
	if err != nil {
		return nil, err
	}
	out := make([]web.CapHolder, 0, len(rows))
	for _, r := range rows {
		out = append(out, web.CapHolder{Account: r.Account, Caps: r.Caps})
	}
	return out, nil
}

func (c capsPlane) CategoryCaps(repo, category string) string {
	return c.db.CategoryCaps(repo, category)
}

func (c capsPlane) SetCaps(account, repo, caps string) error {
	return c.db.SetCaps(account, repo, caps)
}

func (c capsPlane) SetCategoryCaps(repo, category, caps string) error {
	return c.db.SetCategoryCaps(repo, category, caps)
}

// inboxPlane adapts the notify client to the web's Inbox seam, keeping
// notify's message shape out of the page layer.
type inboxPlane struct{ client *notify.Client }

func (i inboxPlane) Messages(ctx context.Context, subscriber string) ([]web.InboxMessage, error) {
	msgs, err := i.client.Inbox(ctx, subscriber)
	if err != nil {
		return nil, err
	}
	out := make([]web.InboxMessage, 0, len(msgs))
	for _, m := range msgs {
		out = append(out, web.InboxMessage{
			Subject: m.Subject, Content: m.Content, CreatedAt: m.CreatedAt, Read: m.Read,
		})
	}
	return out, nil
}

// topicUnsafe matches everything notify refuses in a topic key: it
// accepts lowercase letters, digits, hyphens and underscores only.
var topicUnsafe = regexp.MustCompile(`[^a-z0-9_-]+`)

// repoTopicKey turns a repository name into a topic key notify accepts.
// quarry's internal convention is "repo:<name>" (0001 §5) and its repo
// names allow dots, dashes, underscores and uppercase — none of which a
// colon-bearing key can carry into notify, which rejects the whole
// trigger rather than mangling it.
//
// Two names that differ only in punctuation or case collapse to one key
// ("My.Repo" and "my-repo" both give repo-my-repo), so two repositories
// could share an activity topic. That is a real limitation, accepted
// because the alternative — hashing — produces keys no operator can
// read when subscribing someone to a repository's activity.
func repoTopicKey(repo string) string {
	return "repo-" + topicUnsafe.ReplaceAllString(strings.ToLower(repo), "-")
}

// ciStore adapts the OLTP database to the CI trigger's record-keeping.
// mirrorStore adapts the OLTP mirror rows to the ops layer's view. The
// interval crosses as seconds because that is the JSON surface's shape;
// the duration is reconstituted here rather than leaked outward.
type mirrorStore struct{ db *oltp.DB }

func (m mirrorStore) Mirrors() ([]api.Mirror, error) {
	rows, err := m.db.Mirrors()
	if err != nil {
		return nil, err
	}
	out := make([]api.Mirror, 0, len(rows))
	for _, r := range rows {
		out = append(out, toAPIMirror(r))
	}
	return out, nil
}

func (m mirrorStore) MirrorFor(repo string) (api.Mirror, bool, error) {
	row, ok, err := m.db.MirrorFor(repo)
	if err != nil || !ok {
		return api.Mirror{}, ok, err
	}
	return toAPIMirror(row), true, nil
}

func (m mirrorStore) SetMirror(repo, url string, everySeconds int) error {
	return m.db.SetMirror(repo, url, time.Duration(everySeconds)*time.Second)
}

func (m mirrorStore) DetachMirror(repo string) error { return m.db.DetachMirror(repo) }

func (m mirrorStore) SetMirrorWrites(repo string, accept bool) error {
	return m.db.SetMirrorWrites(repo, accept)
}

func (m mirrorStore) RecordMirrorRun(repo string, err error) error {
	return m.db.RecordMirrorRun(repo, time.Now(), err)
}

func toAPIMirror(r oltp.Mirror) api.Mirror {
	return api.Mirror{
		Repo: r.Repo, URL: r.URL, Every: int(r.Every.Seconds()), Enabled: r.Enabled,
		LastAt: r.LastAt, LastOK: r.LastOK, LastErr: r.LastErr, Failures: r.Failures,
		AcceptWrites: r.AcceptWrites, DetachedFrom: r.DetachedFrom,
	}
}

// mirrorScheduler pulls every due mirror on a tick. It is one goroutine
// for the whole fleet rather than one per store on purpose: pulls are
// network-bound against OTHER PEOPLE'S servers, and eleven goroutines
// each free to hit fossil-scm.org whenever their own timer fires is a
// worse neighbour than a single loop that walks the list.
func mirrorScheduler(deps *api.Deps, odb *oltp.DB, tick time.Duration) {
	for range time.Tick(tick) {
		rows, err := odb.Mirrors()
		if err != nil {
			log.Printf("mirror: listing: %v", err)
			continue
		}
		now := time.Now()
		for _, row := range rows {
			if !row.Due(now) {
				continue
			}
			changed, err := deps.PullMirror(toAPIMirror(row))
			switch {
			case err != nil:
				log.Printf("mirror: %s from %s: %v", row.Repo, row.URL, err)
			case changed:
				log.Printf("mirror: %s pulled new artifacts from %s", row.Repo, row.URL)
			}
			// A quiet pull is not logged. Almost every one is quiet, and
			// a log line per store per interval buries the ones that matter.
		}
	}
}

type ciStore struct{ db *oltp.DB }

// ciRuns adapts the OLTP rows to the ops layer's view.
type ciRuns struct{ db *oltp.DB }

func (c ciRuns) CIRuns(repo string, limit int) ([]api.CIRun, error) {
	rows, err := c.db.CIRuns(repo, limit)
	return toAPIRuns(rows), err
}

func (c ciRuns) CIRunsFor(repo, checkin string) ([]api.CIRun, error) {
	rows, err := c.db.CIRunsFor(repo, checkin)
	return toAPIRuns(rows), err
}

func toAPIRuns(rows []oltp.CIRun) []api.CIRun {
	out := make([]api.CIRun, 0, len(rows))
	for _, r := range rows {
		out = append(out, api.CIRun{RunID: r.RunID, Checkin: r.Checkin,
			Workflow: r.Workflow, Pipeline: r.Pipeline, Status: r.Status, Started: r.Started})
	}
	return out
}

func (s ciStore) CIWatermark(repo string) (string, bool) { return s.db.CIWatermark(repo) }
func (s ciStore) SetCIWatermark(repo, c string) error    { return s.db.SetCIWatermark(repo, c) }
func (s ciStore) AddRun(runID, repo, checkin, workflow, pipeline, status string) error {
	return s.db.AddCIRun(oltp.CIRun{RunID: runID, Repo: repo, Checkin: checkin,
		Workflow: workflow, Pipeline: pipeline, Status: status})
}

// ciRepo adapts one repository's index and store to the trigger's read
// side: the index answers "what is new", zit answers "what is in it".
type ciRepo struct {
	ix  *index.Index
	zit *zitcli.Client
}

func (r ciRepo) NewCheckins(after string, limit int) ([]ci.Checkin, error) {
	entries, err := r.ix.CheckinsAfter(after, limit)
	if err != nil {
		return nil, err
	}
	out := make([]ci.Checkin, 0, len(entries))
	for _, e := range entries {
		out = append(out, ci.Checkin{Hash: e.Hash, Branch: e.Branch,
			Date: e.Date, User: e.User, Comment: e.Comment})
	}
	return out, nil
}

func (r ciRepo) Files(version string) ([]string, error) {
	entries, err := r.zit.Ls(context.Background(), version)
	if err != nil {
		return nil, err
	}
	out := make([]string, 0, len(entries))
	for _, e := range entries {
		out = append(out, e.Name)
	}
	return out, nil
}

func (r ciRepo) Read(path, version string) ([]byte, error) {
	return r.zit.Cat(context.Background(), path, version)
}

// pollCI refreshes the status of runs quarry still thinks are going and
// writes the durable verdict when one finishes.
//
// dagnats owns the run; quarry caches only the last status it saw, so a
// missed poll costs nothing but freshness. The verdict is the part that
// must not be lost: it becomes a tag artifact on the check-in, which
// clones with the repository (0001 §6's no-hostage rule).
func pollCI(engine *ci.Engine, db *oltp.DB, repos *api.Repos, verdicts bool) {
	runs, err := db.UnfinishedCIRuns()
	if err != nil {
		log.Printf("ci: poll: %v", err)
		return
	}
	for _, r := range runs {
		run, err := engine.Status(r.RunID)
		if err != nil {
			continue // the engine may be restarting; try again next tick
		}
		if run.Status == r.Status {
			continue
		}
		if err := db.SetCIStatus(r.RunID, run.Status); err != nil {
			log.Printf("ci: status %s: %v", r.RunID, err)
			continue
		}
		if !run.Terminal() || !verdicts {
			continue
		}
		repo, ok := repos.Get(r.Repo)
		if !ok {
			continue
		}
		verdict := "ci-fail"
		if run.Passed() {
			verdict = "ci-pass"
		}
		// Written through the lease holder, like every other artifact
		// quarry mints: a verdict is a write, and two nodes must not
		// both record one.
		if err := repo.WriteTag(verdict, r.Checkin, r.RunID); err != nil {
			log.Printf("ci: verdict %s on %.12s: %v", verdict, r.Checkin, err)
			continue
		}
		log.Printf("ci: %s %.12s %s (run %s)", r.Repo, r.Checkin, verdict, r.RunID)
	}
}