// 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/evt"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/notify"
"craftdesign.group/quarry/internal/obs"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/timing"
"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)")
oltpFlag := fs.String("oltp", "", "path to oltp.db (default: <data>/oltp.db); point several nodes at one file to share accounts and grants")
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(oltpPath(*oltpFlag, dir))
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]\n without --url the table is read from the store itself (zit config pull user)")
}
fs := flag.NewFlagSet("users pull", flag.ContinueOnError)
repos := repoFlags{}
fs.Var(repos, "repo", "repository as name=path")
peer := fs.String("url", "", "fetch from this peer instead of reading the store's own table")
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)")
oltpFlag := fs.String("oltp", "", "path to oltp.db (default: <data>/oltp.db); point several nodes at one file to share accounts and grants")
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 {
return fmt.Errorf("users pull wants exactly one --repo")
}
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 repository we serve always has one.
code, err := xfer.ProjectCode(path)
if err != nil {
return fmt.Errorf("project code of %s: %w", name, err)
}
var rows []oltp.ImportedUser
if *peer == "" {
// Store path: the table is already local, brought across by
// `zit import-fossil` or `zit config pull user` (zit#124/#125).
// Nothing is fetched and no credential is needed — quarry is
// reading a repository it already serves.
users, err := zitcli.New(path).Users(context.Background())
if err != nil {
return err
}
if len(users) == 0 {
return fmt.Errorf("%s carries no user table — run `zit config pull user URL` in the store, or pass --url to fetch it here", name)
}
for _, u := range users {
rows = append(rows, oltp.ImportedUser{
Login: u.Login, Caps: u.Caps, Info: u.Info, PWHash: u.PWHash,
})
}
} else {
// Wire path: ask the peer ourselves, with the operator's own
// admin credential there. The peer enforces Admin, not us.
if *user == "" {
return fmt.Errorf("--url needs a --user that is an admin on that peer")
}
pw := *password
if pw == "" {
pw = os.Getenv("QUARRY_PASSWORD")
}
if pw == "" {
return fmt.Errorf("no password: use --password or QUARRY_PASSWORD")
}
users, err := xfer.FetchUsers(nil, *peer, code, *user, pw)
if err != nil {
return err
}
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(oltpPath(*oltpFlag, dir))
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.
// oltpPath is where the forge's own records live: accounts, capability
// grants, webhooks, praise, mirrors, saved views, CI runs.
//
// It sits under --data by default, and that default is the whole reason
// this function exists. --data is DERIVED state — per-repository
// indexes, built archives, the event store — all of it rebuildable from
// the stores (0001 §5) and therefore private to one node. oltp.db is
// none of those things: it is forge-local TRUTH (0001 §8), derivable
// from nothing, and two nodes each holding their own copy do not have a
// forge between them, they have two forges that disagree about who has
// an account.
//
// So a deployment that runs more than one node points --oltp at storage
// both nodes share, while --data stays local and fast. A single-node
// deployment passes neither and is unaffected.
func oltpPath(explicit, dir string) string {
if explicit != "" {
return explicit
}
return filepath.Join(dir, "oltp.db")
}
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")
var dataRoot string // derived-data root, for the web's drained-log reads
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)")
oltpFlag := fs.String("oltp", "", "path to oltp.db (default: <data>/oltp.db). --data is derived state and is private to a node; the OLTP is forge truth and must be SHARED by every node of one forge")
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)")
// Analytics defaults ON, in the data directory, because the live
// panels are a snapshot and a forge that restarts loses exactly the
// window worth having. Empty disables it; the panels still work.
analyticsDir := fs.String("analytics", "", `directory for the operational analytics store (default: beside the derived data; "off" disables it)`)
analyticsKeep := fs.Duration("analytics-keep", 14*24*time.Hour, "how long to keep observations; 0 keeps them for ever")
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 protectAdmin web.Protect
var inbox web.Inbox
var webHooks web.Hooks
var webTallies web.Tallies
var webPraise web.Praise
var webEvents web.EventLog
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)
dataRoot = dir
if err != nil {
return err
}
if err := os.MkdirAll(dir, 0o755); err != nil {
return err
}
// Derived, disposable, and beside the other derived data. On by
// default because the live panels are a snapshot and a forge
// that restarts loses exactly the window worth having; "off"
// disables it and the panels still work.
adir := dir
if *analyticsDir != "" {
adir = *analyticsDir
}
if adir == "off" {
adir = ""
}
defer openAnalytics(adir, *analyticsKeep)()
odb, err := oltp.Open(oltpPath(*oltpFlag, dir))
if err != nil {
return err
}
defer odb.Close()
fronter.SetAuth(odb)
// RFC 0007: the identity the gate verified becomes the rule the
// child enforces. Refused lists the protected globs a pusher
// does not own.
fronter.SetProtect(odb.Refused)
// 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}
webHooks = hookStore{db: odb}
webPraise = odb
deps.Praise = odb
deps.Protect = odb
deps.Checks = odb // RFC 0009: required checks beside the protections
deps.External = odb // RFC 0009: external checks and reporter tokens
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.
// RFC 0008: the floor is signed-in or admin; the op's own
// letters decide the rest, categories included.
if odb.MayAt(acct, repo, need == ops.Admin) {
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 = ¬ifyAdapter{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),
})
}
// A route crossing the render budget is news the FIRST time
// and noise after: the same page is slow on every request
// until somebody fixes it. Same rule as the error journal
// above, and for the same reason.
//
// This is the difference between an operator watching the
// forge and the forge watching itself — the ops page has to
// be looked at; this arrives.
obs.Default.OnSlow = func(v obs.Render) {
_ = np.Trigger(context.Background(), api.NotifyEvent{
Workflow: "quarry-slow", To: "admins",
Payload: map[string]any{
"path": v.Path, "repo": v.Repo,
"elapsedMs": v.Elapsed.Milliseconds(),
"opsMs": v.Ops.Milliseconds(),
"zitMs": v.Zit.Milliseconds(),
"budgetMs": obs.Budget.Milliseconds(),
"summary": fmt.Sprintf("%s took %s (budget %s)",
v.Path, timing.Format(v.Elapsed), timing.Format(obs.Budget)),
},
// One per route per process: a restart may report it
// again, which is right — a fresh process has not
// told anyone yet.
TxnID: "slow:" + v.Path,
})
}
}
// 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, dir)
}
}()
}
// 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)
// Clones and fetches are the one kind of forge activity that
// leaves NO trace in the DAG — a clone reads and writes nothing,
// so without this it is invisible. Two sinks, deliberately: the
// event says it happened once, for anything reacting; the
// counter says how many, and converges across nodes instead of
// losing increments to a read-modify-write.
webTallies = ev
webEvents = eventLog{ev: ev}
fronter.SetCounted(func(repo string, clone bool) {
kind, name := evt.CloneFetched, "fetch"
if clone {
kind, name = evt.CloneCloned, "clone"
}
if err := ev.Raise(kind, repo, map[string]any{"repo": repo}); err != nil {
log.Printf("events: raise %s: %v", kind, err)
}
if err := ev.Count(repo, name, 1); err != nil {
log.Printf("events: count %s/%s: %v", repo, name, err)
}
})
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
}
// Deliveries come from the vocabulary stream, not from the ops
// layer: one consumer turns every raised event into webhooks,
// so a change notifies the same subscribers whether it arrived
// through the forge or on a push.
if err := dispatcher.Fanout(hookCtx); err != nil {
return err
}
// RFC 0005's vocabulary on quarry.evt.<repo>.<noun>.<action>.
// Same reasoning as the dispatcher above: it costs nothing until
// something subscribes, and a subject nobody listens to is a
// message the stream ages out.
deps.Events = ev
// 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 := ®istrar{
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,
ciTagsDefault: *ciVerdicts,
codes: codes, codeMu: &codeMu,
}
defer rg.CloseAll()
deps.Registry = rg
deps.CITrigger = rg.TriggerCI // RFC 0009: the re-run path
deps.Accounts = accountStore{db: odb}
capsFor = func(account, repo string) api.CapSet { return odb.EffectiveCaps(account, repo) }
// The category strings ARE the private flag (RFC 0004 §Private), so
// a save through the settings page must refresh the read gate's
// cache the way repo.update does.
capsAdmin = capsPlane{db: odb, changed: rg.refreshPrivate}
deps.Visibility = odb.VisibilityName
protectAdmin = odb // RFC 0007: the settings page edits the rules under `a`
deps.Votes = odb
deps.Forum = odb
deps.Views = 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, protectAdmin, sessions, inbox, webHooks, webTallies, webPraise, webEvents, *tracker, webEngine(engine), ciLogStore{dir: dataRoot})
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
}
log.Printf("startup complete; binding %s", *addr)
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, protectAdmin web.Protect, sessions web.Sessions, inbox web.Inbox, hooks web.Hooks, tallies web.Tallies, praise web.Praise, evlog web.EventLog, tracker string, eng web.EngineViews, cilogs web.LogStore) (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 eng != nil {
opts = append(opts, web.WithEngine(eng)) // RFC 0009 §logs
}
if cilogs != nil {
opts = append(opts, web.WithCILogs(cilogs))
}
if sessions != nil {
opts = append(opts, web.WithSessions(sessions))
if capsAdmin != nil {
opts = append(opts, web.WithCaps(capsAdmin))
}
if protectAdmin != nil {
opts = append(opts, web.WithProtect(protectAdmin))
}
}
if tracker != "" {
opts = append(opts, web.WithTracker(tracker))
}
if inbox != nil {
opts = append(opts, web.WithInbox(inbox))
}
if hooks != nil {
opts = append(opts, web.WithHooks(hooks))
}
if tallies != nil {
opts = append(opts, web.WithTallies(tallies))
}
if praise != nil {
opts = append(opts, web.WithPraise(praise))
}
if evlog != nil {
opts = append(opts, web.WithEventLog(evlog))
}
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.
// 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 reader|developer|maintainer|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)")
oltpFlag := fs.String("oltp", "", "path to oltp.db (default: <data>/oltp.db); point several nodes at one file to share accounts and grants")
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", "", "reader, developer, maintainer 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(oltpPath(*oltpFlag, dir))
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, Events: w.Events})
}
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,
Events: w.Events}, true
}
// 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 events --id N --events "ticket landing.accepted"
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")
oltpFlag := fs.String("oltp", "", "path to oltp.db (default: <data>/oltp.db); point several nodes at one file to share accounts and grants")
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")
events := fs.String("events", "", "vocabulary patterns to deliver, space separated\n"+
" \t(default \"*\" = everything; e.g. \"ticket landing.accepted\")")
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(oltpPath(*oltpFlag, dir))
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")
}
if err := checkEventPatterns(*events); err != nil {
return err
}
w, err := db.AddWebhook(*repoTarget, *endpoint, *events)
if err != nil {
return err
}
fmt.Printf("webhook %d: %s -> %s [%s]\n", w.ID, w.Repo, w.URL, w.Events)
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 %-40s %s (added %s)\n", w.ID, state, w.URL, w.Events, w.Created)
}
return nil
case "events":
if *id == 0 {
return errors.New("webhook events wants --id and --events")
}
if err := checkEventPatterns(*events); err != nil {
return err
}
if err := db.SetWebhookEvents(*id, *events); err != nil {
return err
}
sel := *events
if strings.TrimSpace(sel) == "" {
sel = oltp.DefaultEvents
}
fmt.Printf("webhook %d now delivers %s\n", *id, sel)
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)
}
// eventLog adapts the event plane to the web layer's EventLog seam. The
// web layer names what a log row shows; the plane owns where it is read
// from, which is the stream itself rather than any stored copy.
type eventLog struct{ ev *events.Server }
func (e eventLog) Recent(repo string, limit int) ([]web.LoggedEvent, error) {
rows, err := e.ev.Recent(repo, limit)
if err != nil {
return nil, err
}
out := make([]web.LoggedEvent, 0, len(rows))
for _, r := range rows {
out = append(out, web.LoggedEvent{Kind: string(r.Kind), Actor: r.Actor,
Summary: r.Summary, At: r.At})
}
return out, nil
}
// hookStore adapts the OLTP database to the web layer's Hooks seam. The
// secret is carried out of AddWebhook and nowhere else — Webhooks does
// not select it, so a listing cannot leak what it never read.
type hookStore struct{ db *oltp.DB }
func (h hookStore) Webhooks(repo string) ([]web.HookSub, error) {
rows, err := h.db.Webhooks(repo)
if err != nil {
return nil, err
}
out := make([]web.HookSub, 0, len(rows))
for _, w := range rows {
out = append(out, web.HookSub{ID: w.ID, URL: w.URL, Events: w.Events,
Active: w.Active, Created: w.Created})
}
return out, nil
}
func (h hookStore) AddWebhook(repo, endpoint, events string) (web.HookSub, error) {
w, err := h.db.AddWebhook(repo, endpoint, events)
if err != nil {
return web.HookSub{}, err
}
return web.HookSub{ID: w.ID, URL: w.URL, Events: w.Events, Active: w.Active,
Created: w.Created, Secret: w.Secret}, nil
}
func (h hookStore) SetWebhookEvents(id int64, events string) error {
return h.db.SetWebhookEvents(id, events)
}
func (h hookStore) SetWebhookActive(id int64, active bool) error {
return h.db.SetWebhookActive(id, active)
}
func (h hookStore) DeleteWebhook(id int64) error { return h.db.DeleteWebhook(id) }
// checkEventPatterns refuses a pattern no event can ever match.
//
// A webhook that silently delivers nothing is the worst outcome here:
// the owner configured something, saw no error, and will conclude the
// forge is broken rather than that they typed "tickets". Every pattern
// is checked against the vocabulary at the moment it is set.
func checkEventPatterns(patterns string) error {
for _, p := range strings.Fields(patterns) {
if p == "*" {
continue
}
matched := false
for _, k := range evt.All {
if k.Matches(p) {
matched = true
break
}
}
if !matched {
return fmt.Errorf("no event matches %q; try one of: %s, or a full kind like ticket.closed",
p, strings.Join(evt.Nouns(), " "))
}
}
return nil
}
// 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.MayAt(a, repo, need == ops.Admin)
}
// 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
changed func() // after a category write; nil is fine
}
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) GuardCaps(actor, account, repo, caps string) (string, error) {
return c.db.GuardCaps(actor, account, repo, caps)
}
func (c capsPlane) SetCategoryCaps(repo, category, caps string) error {
if err := c.db.SetCategoryCaps(repo, category, caps); err != nil {
return err
}
if c.changed != nil {
c.changed()
}
return nil
}
func (c capsPlane) ReachOnlyByCategory(repo string) ([]string, error) {
return c.db.ReachOnlyByCategory(repo)
}
func (c capsPlane) EffectiveWith(account, repo, own string) web.CapSet {
return c.db.EffectiveCapsWith(account, repo, own)
}
// 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 (c ciRuns) CIRunsForPair(repo, pair string) ([]api.CIRun, error) {
rows, err := c.db.CIRunsForPair(repo, pair)
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, Pair: r.Pair,
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) {
// Names only: this returns []string of names.
entries, err := r.zit.LsNames(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).
// webEngine hands the web the engine's read side, or nil when no
// engine is configured — the pages say so in words.
func webEngine(e *ci.Engine) web.EngineViews {
if e == nil || !e.Enabled() {
return nil
}
return engineViews{e}
}
// engineViews adapts the engine's read side to the web's seam.
type engineViews struct{ e *ci.Engine }
func (v engineViews) RunDetail(runID string) (ci.Run, error) { return v.e.Status(runID) }
func (v engineViews) StepLogs(runID, step string, afterSeq int, fromFailure bool) ([]ci.LogChunk, error) {
return v.e.Logs(runID, step, afterSeq, fromFailure)
}
func (v engineViews) Workers() ([]ci.Worker, error) { return v.e.Workers() }
func pollCI(engine *ci.Engine, db *oltp.DB, repos *api.Repos, verdicts bool, dataDir string) {
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() {
continue
}
// Retention (RFC 0009 §logs): the hot lane expires; the record
// beside the verdict does not.
steps := make([]string, 0, len(run.Steps))
for id := range run.Steps {
steps = append(steps, id)
}
if err := drainLogs(engine, dataDir, r.Repo, r.RunID, steps); err != nil {
log.Printf("ci: drain %s: %v", r.RunID, err)
}
// RFC 0009 §verdict-tags: the repository chooses; '' follows
// the fleet default the flag sets.
switch db.CITagsMode(r.Repo) {
case "off":
continue
case "on":
default:
if !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)
}
}