// 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, Chat: docsChatPlane{}}); 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 {
// changes is the event plane, when one is running: the live pages
// wait on it instead of each keeping its own timer.
var changes web.Changes
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")
dagnatsToken := fs.String("dagnats-token", os.Getenv("DAGNATS_BRIDGE_TOKEN"), "admin bearer for the engine's worker-token routes (dagnats#627); defaults to $DAGNATS_BRIDGE_TOKEN")
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")
// The attention stream costs one connection per open tab. Plain
// HTTP/1.1 allows a browser six per origin and a repository page
// already spends one or two, so the default is repository pages
// alone; a deployment with h2 in front (TLS, or h2c at the proxy)
// has no such limit and can afford it everywhere. docs/running.md
// says so.
attnEverywhere := fs.Bool("attention-everywhere", false, "open the attention stream on every page, not repository pages alone; needs HTTP/2 in front (see docs/running.md)")
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 webPrefs web.Prefs
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, provision: &deps.Provision}
webHooks = hookStore{db: odb}
webPraise = odb
webPrefs = prefStore{db: odb} // RFC 0011 §5: the desktop-notification switch
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
deps.Auto = odb // RFC 0009: armed auto-accepts beside them
deps.Gate = odb // RFC 0007: the bypass tripwire's memory
deps.AttnRead = odb // RFC 0011 §5: the attention read pointer
deps.ChatRobots = odb // RFC 0011 §10: the robot's event patterns
deps.Prefs = odb // RFC 0011 §6: toasts, desktop, per-repo mute
deps.PushDevices = pushDevices{db: odb}
// The editor's save: a scratch checkout against the repository's
// own store — the serving tree is never touched (RFC 0009).
deps.SavePipeline = func(repo, branch, user, message, path string, body []byte) (string, error) {
dir, ok := repos[repo]
if !ok {
return "", fmt.Errorf("unknown repository %q", repo)
}
return zitcli.CommitFile(filepath.Join(dir, ".zit"), "trunk", branch, user, message, path, body)
}
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
// Carried out of this block so the web layer's live streams can
// wait on it (see extraOpts below).
changes = 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
// RFC 0011 §10: the robot's lines are remembered as they are
// posted, so an unread count subtracts them from the broker's
// own number instead of reading the window to find them.
room.Robots = odb
chatLive = &chatAdapter{room: room, codes: codes, mu: &codeMu}
// RFC 0011 §5: what is addressed to a person, on its own small
// stream, written before notify and independently of it.
deps.Attention = attentionAdapter{ev: ev}
// RFC 0011 §10: the timeline talking in the channel. One
// consumer for the fleet; a repository with no stored patterns
// is silent, so this costs nothing until somebody opts in.
robotCtx, stopRobot := context.WithCancel(context.Background())
defer stopRobot()
if err := chat.NewRobot(room, ev.JS(),
func(repo string) string { return odb.ChatRobot(repo) },
func(repo string) (string, bool) {
codeMu.Lock()
defer codeMu.Unlock()
code, ok := codes[repo]
return code, ok
}).Start(robotCtx); err != nil {
return err
}
// 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}
// With notify wired, the count people can act on is the
// notify inbox's; without it, inbox.count falls back to the
// local attention tail on its own.
deps.Inbox = inboxCounter{client: nclient, names: notify.NewWorkflowNames(nclient)}
deps.Provision = &provisionAdapter{client: nclient}
inbox = inboxPlane{client: nclient}
fmt.Println("quarry: notifications ->", *notifyURL)
// Every trigger reaches nobody until its workflow exists
// (RFC 0011 §7): create the six quarry workflows at startup,
// never overwriting one an operator already edited (notify's
// upsert is asked for with onConflict=keep). Best-effort and
// logged, like everything else notify-shaped — a notify outage
// at startup must not fail quarry's own startup. Run off the
// startup path entirely: EnsureWorkflow is one upsert per
// workflow, each with its own 5s timeout, and a notify that accepts TCP
// but never answers turns "provision six workflows" into a
// ~60s stall before ListenAndServe — quarry would refuse
// every request until notify was reachable, though nothing
// notify-shaped is on the startup critical path otherwise.
go func() {
for _, wf := range notify.Workflows() {
if err := nclient.EnsureWorkflow(context.Background(), wf); err != nil {
log.Printf("notify: ensure workflow %s: %v", wf.Name, err)
}
}
}()
// 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
evs := ev
errlog.Default.OnFirst = func(e errlog.Entry) {
// Onto the event plane as well, so a subscriber can act
// on a failure the way it acts on anything else. Only
// when the failure names a repository: the subject IS
// the repo, and a fleet-wide fault — a poll for a store
// that no longer exists, a tick with nothing in scope —
// has no honest subject to publish under. Those still
// reach the admin notify below and the panel.
if e.Repo != "" {
if rerr := evs.Raise(evt.ErrorRaised, e.Repo, map[string]any{
"surface": e.Surface, "path": e.Path,
"message": e.Message, "count": e.Count,
}); rerr != nil {
log.Printf("errors: raise: %v", rerr)
}
}
_ = 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)
engine.SetAdminToken(*dagnatsToken)
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}
deps.ValidateCIYAML = engine.ValidateYAML
fmt.Println("quarry: CI ->", *dagnatsURL)
go func() {
for range time.Tick(*ciPoll) {
pollCI(engine, odb, deps.Repos, *ciVerdicts, dir, ev)
}
}()
}
// 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) }
deps.CapsFor = capsFor
// 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.ChatRefs = 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)
}
// RFC 0009 §verdicts, the follow-through: armed landings land
// themselves once their checks go green. Swept on a fixed beat —
// the deferred accept re-checks everything, so a wasted try costs
// one index read.
if sweeper, ok := deps.Auto.(api.AutoSweeper); ok {
go func() {
for range time.Tick(30 * time.Second) {
for _, id := range deps.AutoAcceptSweep(sweeper) {
log.Printf("landing: %s auto-accepted — checks green", id)
}
}
}()
}
// RFC 0011 §14: the chat backlink and reply rows die with the
// channel window. The window ages out on the stream's retention,
// which moves whether or not anybody posts — so a quiet repository
// needs the sweep, not just the post path.
if deps.ChatRefs != nil && deps.Chat != nil {
// The reply rows a window predating them never got, written
// once so the log, the permalink and the preview card count the
// same replies. Idempotent, so a restart repeats it for free.
deps.ChatReplyBackfill()
sweepCtx, stopSweep := context.WithCancel(context.Background())
defer stopSweep()
go func() {
t := time.NewTicker(5 * time.Minute)
defer t.Stop()
for {
select {
case <-t.C:
deps.ChatWindowSweep()
case <-sweepCtx.Done():
return
}
}
}()
}
var extraOpts []web.Option
// The live streams wait on the event plane rather than their own
// timers: one poller learns a run moved, every reader is told.
if changes != nil {
extraOpts = append(extraOpts, web.WithChanges(changes))
}
if webPrefs != nil {
extraOpts = append(extraOpts, web.WithPrefs(webPrefs))
}
if *attnEverywhere {
extraOpts = append(extraOpts, web.WithAttentionEverywhere())
}
mux, err := buildHandler(deps, notifier, chatLive, verify, capsFor, capsAdmin, protectAdmin, sessions, inbox, webHooks, webTallies, webPraise, webEvents, *tracker, webEngine(engine), ciLogStore{dir: dataRoot}, extraOpts...)
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)
}
// docsChatPlane is a non-nil api.ChatPlane that exists only so `quarry
// docs` and `quarry spec` register the chat ops (RegisterAll gates them
// on Deps.Chat != nil, since without a real plane there is nothing for
// chat.post et al. to call). Neither subcommand ever calls an op's
// Run — they reflect over the registry's descriptors — so every method
// here is unreachable; each returns the zero value rather than
// panicking, in case that ever changes.
type docsChatPlane struct{}
func (docsChatPlane) Post(string, chat.Message) (uint64, error) { return 0, nil }
func (docsChatPlane) History(string, uint64, int) ([]chat.Message, uint64, error) { return nil, 0, nil }
func (docsChatPlane) Tail(string, uint64) (<-chan chat.Message, func()) { return nil, func() {} }
func (docsChatPlane) Range(string, uint64, uint64) ([]chat.Message, error) { return nil, nil }
func (docsChatPlane) Bounds(string) (uint64, uint64, error) { return 0, 0, nil }
func (docsChatPlane) Unread(string, uint64) (int, error) { return 0, nil }
// 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
}
// Tail satisfies web.ChatLive: the never-drops resumable stream RFC
// 0011 §2 asks for (unlike the old SubscribeChat, which rode core
// Subscribe's 32-slot drop-on-backpressure buffer). Rendering here has
// no markup pipeline to hand (that lives behind repo.Index in
// internal/api) — chatMsgHTML falls back to plainly-escaped text when
// HTML is empty, so a live line still shows, unformatted.
func (a *chatAdapter) Tail(repo string, from uint64) (<-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, stop := a.room.Tail(code, from)
out := make(chan web.ChatMessage) // unbuffered: mirrors chat.Tail's own never-drops contract
done := make(chan struct{})
go func() {
// chat.Tail closes src on cancel, so the range ends; done
// releases a bridge that is parked on a send nobody will ever
// receive, which is the other way this goroutine could outlive
// the request that opened it.
defer close(out)
for m := range src {
select {
case out <- web.ChatMessage{
Seq: m.Seq, Kind: m.Kind, Ref: m.Ref,
User: m.User, Text: m.Text, At: m.At,
Edited: m.Edited, Deleted: m.Deleted,
}:
case <-done:
return
}
}
}()
var once sync.Once
return out, func() {
once.Do(func() {
close(done)
stop()
})
}
}
// 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)
}
// provisionAdapter bridges api.ProvisionPlane to the notify client
// (RFC 0011 §7): quarry decides who belongs to which repo's activity
// topic, notify remembers it. The topic key uses the same
// repoTopicKey sanitization notifyAdapter's "repo:" addressing does, so
// a grant's topic and an activity trigger's topic are always the same
// key.
type provisionAdapter struct{ client *notify.Client }
func (a *provisionAdapter) EnsureSubscriber(ctx context.Context, account string) error {
return a.client.EnsureSubscriber(ctx, account)
}
func (a *provisionAdapter) SetTopicMember(ctx context.Context, repo, account string, member bool) error {
key := repoTopicKey(repo)
if err := a.client.EnsureTopic(ctx, key, repo+" activity"); err != nil {
return err
}
if member {
// notify's POST /v1/topics/{key}/subscribers answers 422
// "subscriber not found" for a subscriberId that was never
// created — which every account that existed before this
// feature shipped, or was inserted straight into OLTP rather
// than through account.add, is. Upsert it here too: cheap
// (idempotent, same as every other Ensure* call), and the only
// way a grant on such an account ever becomes a real topic
// membership instead of a logged 422.
if err := a.client.EnsureSubscriber(ctx, account); err != nil {
return err
}
return a.client.AddTopicMember(ctx, key, account)
}
return a.client.RemoveTopicMember(ctx, key, account)
}
// 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, extra ...web.Option) (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)
// The service worker must be served from the ROOT to claim the
// site's scope (RFC 0011 §6). Mounted here rather than left to the
// api router, which this server mounts under /api/.
mux.Handle("GET /sw.js", api.ServiceWorker())
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))
}
// Options the composition root decided rather than derived: the
// preference store and the deployment's --attention-everywhere.
opts = append(opts, extra...)
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, Chat: docsChatPlane{}}); 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 := evt.CheckPatterns(*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 := evt.CheckPatterns(*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) }
// 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.
//
// provision points at deps.Provision rather than holding a copy: it is
// captured before notify's client is wired up (deps.Provision is set
// later, only when --notify is given), and reading through the pointer
// at Login time — long after main's setup runs — sees whatever the
// field ends up holding. web must not import notify (RFC 0011 §7), so
// this indirection lives here in main rather than in internal/web.
type sessionPlane struct {
db *oltp.DB
provision *api.ProvisionPlane
}
// Login authenticates and, best-effort, ensures the account is a notify
// subscriber (RFC 0011 §7): account.add covers accounts created through
// quarry, but an account that pre-dates this feature, or was inserted
// straight into OLTP, never went through that path, and a grant for it
// fails with notify's 422 "subscriber not found" until something
// creates the subscriber first. A successful login is that something —
// cheap (idempotent upsert) and reached by every account eventually.
func (s sessionPlane) Login(user, password string) (string, time.Time, bool) {
token, expires, ok := s.db.NewSession(user, password)
if ok && s.provision != nil {
if p := *s.provision; p != nil {
account := user
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := p.EnsureSubscriber(ctx, account); err != nil {
log.Printf("notify: provision: ensure subscriber on login %q: %v", account, err)
}
}()
}
}
return token, expires, ok
}
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)
errlog.Default.ReportIn("mirror", "listing", "", "", 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)
errlog.Default.ReportIn("mirror", row.URL, row.Repo, "", 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,
Completed: r.Completed})
}
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 (v engineViews) WorkerTokens() ([]ci.WorkerToken, error) { return v.e.WorkerTokens() }
func (v engineViews) MintWorkerToken(label string, prefixes []string) (ci.WorkerToken, error) {
return v.e.MintWorkerToken(label, prefixes)
}
func (v engineViews) RevokeWorkerToken(id string) error { return v.e.RevokeWorkerToken(id) }
func pollCI(engine *ci.Engine, db *oltp.DB, repos *api.Repos, verdicts bool, dataDir string, ev *events.Server) {
runs, err := db.UnfinishedCIRuns()
if err != nil {
log.Printf("ci: poll: %v", err)
errlog.Default.ReportIn("ci", "poll", "", "", 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
}
// The status MOVED. dagnats does not push, so this loop is the
// one place that learns it — and every live page used to learn
// the same thing by asking on its own timer, multiplying the
// question by the number of readers. Raise it once here and let
// them wait on the event instead (RFC 0005).
raise := func(k evt.Kind) {
if ev == nil || r.Repo == "" {
return
}
if rerr := ev.Raise(k, r.Repo, map[string]any{
"runId": r.RunID, "checkin": r.Checkin,
"pipeline": r.Pipeline, "status": run.Status,
}); rerr != nil {
log.Printf("ci: raise %s: %v", k, rerr)
}
}
if !run.Terminal() {
if err := db.SetCIStatus(r.RunID, run.Status); err != nil {
log.Printf("ci: status %s: %v", r.RunID, err)
}
raise(evt.CIStarted)
continue
}
if err := db.CompleteCIRun(r.RunID, run.Status, run.CompletedAt); err != nil {
log.Printf("ci: status %s: %v", r.RunID, err)
continue
}
raise(evt.CICompleted)
// 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)
errlog.Default.ReportIn("ci", "drain", r.Repo, "", 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)
errlog.Default.ReportIn("ci", "verdict", r.Repo, "", err)
continue
}
log.Printf("ci: %s %.12s %s (run %s)", r.Repo, r.Checkin, verdict, r.RunID)
}
}
// attentionAdapter carries RFC 0011 §5's stream across the ops seam:
// internal/api names what it needs and the composition root says which
// transport answers, exactly as notifyAdapter does for triggers.
type attentionAdapter struct{ ev *events.Server }
func (a attentionAdapter) Attend(account string, e api.AttentionEvent) (uint64, error) {
return a.ev.Attend(account, events.AttentionEvent{
Kind: e.Kind, Repo: e.Repo, From: e.From,
Summary: e.Summary, Link: e.Link, At: e.At,
})
}
func (a attentionAdapter) TailAttention(account string, from uint64) (<-chan api.AttentionEvent, func()) {
src, stop := a.ev.TailAttention(account, from)
out := make(chan api.AttentionEvent) // unbuffered: mirrors TailAttention's own never-drops contract
done := make(chan struct{})
go func() {
// TailAttention closes src on cancel, so the range ends; done
// releases a bridge parked on a send nobody will ever receive,
// which is the other way this goroutine could outlive the
// request that opened it.
defer close(out)
for e := range src {
select {
case out <- api.AttentionEvent{
Seq: e.Seq, Kind: e.Kind, Repo: e.Repo, From: e.From,
Summary: e.Summary, Link: e.Link, At: e.At, Ref: e.Ref,
}:
case <-done:
return
}
}
}()
var once sync.Once
return out, func() {
once.Do(func() {
stop()
close(done)
})
}
}
func (a attentionAdapter) AttentionPending(account string, since uint64) (int, error) {
return a.ev.AttentionPending(account, since)
}
func (a attentionAdapter) AttentionTip(account string) (uint64, error) {
return a.ev.AttentionTip(account)
}
func (a attentionAdapter) RecentAttention(account string, limit int) ([]api.AttentionEvent, error) {
evs, err := a.ev.RecentAttention(account, limit)
if err != nil {
return nil, err
}
out := make([]api.AttentionEvent, 0, len(evs))
for _, e := range evs {
out = append(out, api.AttentionEvent{
Seq: e.Seq, Kind: e.Kind, Repo: e.Repo, From: e.From,
Summary: e.Summary, Link: e.Link, At: e.At, Ref: e.Ref,
})
}
return out, nil
}
// inboxCounter is api.InboxPlane over the notify companion: the inbox
// people can actually open, page, clear and configure. Every method is
// a straight relay — the mapping between quarry's words and notify's
// (a "done" row is an "archived" message, an "inapp" channel is
// "in_app") is the OP's job, so this layer never has two vocabularies
// to keep in step.
type inboxCounter struct {
client *notify.Client
// names turns the workflow ULID a message carries into the trigger
// identifier quarry knows it by. Cached here rather than in the ops
// layer because it is a fact about the TRANSPORT: internal/api asks
// for a workflow, and what notify hands back happens to be a row id.
names *notify.WorkflowNames
}
// remote maps notify's "this route has not shipped" onto the ops
// layer's own word for it, so internal/api can degrade around a missing
// route (notify#305) without importing a client. Every other error
// passes through untouched — an outage must stay an outage.
func remote(err error) error {
if err != nil && errors.Is(err, notify.ErrRouteMissing) {
return fmt.Errorf("%w: %w", api.ErrRemoteUnsupported, err)
}
return err
}
func (i inboxCounter) Unread(ctx context.Context, account string) (int, error) {
return i.client.UnreadCount(ctx, account)
}
func (i inboxCounter) List(ctx context.Context, account, before string, limit int) ([]api.InboxMessage, string, error) {
msgs, next, _, _, err := i.client.Messages(ctx, account, before, limit)
if err != nil {
return nil, "", err
}
// Message.WorkflowID is the workflow ROW id. Resolving the whole
// page at once costs at most one extra call, and an id that will
// not resolve is passed through verbatim — the row still lists.
ids := make([]string, 0, len(msgs))
for _, m := range msgs {
ids = append(ids, m.WorkflowID)
}
byID := i.names.Identifiers(ctx, ids)
out := make([]api.InboxMessage, 0, len(msgs))
for _, m := range msgs {
out = append(out, api.InboxMessage{
ID: m.ID, Workflow: byID[m.WorkflowID], Subject: m.Subject, Body: m.Content,
Link: m.Link(), At: m.CreatedAt, Read: m.Read, Seen: m.Seen,
})
}
return out, next, nil
}
func (i inboxCounter) Mark(ctx context.Context, account string, ids []string, all bool, as string) error {
return i.client.MarkMessages(ctx, account, ids, all, as)
}
func (i inboxCounter) Preferences(ctx context.Context, account string) (map[string]map[string]bool, bool, error) {
rows, err := i.client.Preferences(ctx, account)
if err != nil {
return nil, false, remote(err)
}
// Only the per-workflow layer is surfaced: quarry's matrix has a
// row per family, and folding notify's global layer into it would
// show a person a cell they cannot separately set.
ids := make([]string, 0, len(rows))
for _, row := range rows {
if row.Type == notify.PreferenceWorkflow && row.WorkflowID != "" {
ids = append(ids, row.WorkflowID)
}
}
// One resolution for the whole set, not one per row.
byID := i.names.Identifiers(ctx, ids)
out := map[string]map[string]bool{}
for _, row := range rows {
if row.Type != notify.PreferenceWorkflow || row.WorkflowID == "" {
continue
}
// Keyed BOTH ways: by the trigger identifier the ops layer asks
// under, and by the companion's own id, so a row that arrived
// under a key quarry could not translate is still findable.
out[byID[row.WorkflowID]] = row.Channels
out[row.WorkflowID] = row.Channels
}
return out, i.names.Resolved(ctx), nil
}
func (i inboxCounter) SetPreferences(ctx context.Context, account string, prefs map[string]map[string]bool) error {
return remote(i.client.SetPreferences(ctx, account, preferenceRows(ctx, i.names, prefs)))
}
// preferenceRows builds the preference payload, keyed BOTH ways.
//
// This is the one function that knows the ambiguity, and it is isolated
// so exactly one body changes when it is resolved. notify#305: quarry
// needs a subscriber-preferences route and there is none today, so the
// key the future route will read is not yet decided by anything quarry
// can observe. Everything else in the codebase keys a workflow by its
// trigger identifier ("quarry-mention"), while every workflow-keyed row
// notify already stores (domain.Preference) uses the workflow ULID.
//
// Sending both is not hedging — it is the only shape that is correct
// under either answer AND readable under the other: a route keyed by
// identifier finds its rows, a route keyed by ULID finds its rows, and
// neither sees a key it must guess about. When the route lands, delete
// the half that is wrong.
func preferenceRows(ctx context.Context, names *notify.WorkflowNames, prefs map[string]map[string]bool) []notify.Preference {
rows := make([]notify.Preference, 0, 2*len(prefs))
for identifier, channels := range prefs {
rows = append(rows, notify.Preference{
Type: notify.PreferenceWorkflow, WorkflowID: identifier, Channels: channels,
})
if id := names.ID(ctx, identifier); id != "" && id != identifier {
rows = append(rows, notify.Preference{
Type: notify.PreferenceWorkflow, WorkflowID: id, Channels: channels,
})
}
}
sort.Slice(rows, func(a, b int) bool { return rows[a].WorkflowID < rows[b].WorkflowID })
return rows
}
func (i inboxCounter) PushKey(ctx context.Context) (string, error) {
return i.client.WebPushKey(ctx)
}
func (i inboxCounter) PushSubscribe(ctx context.Context, account, endpoint, p256dh, auth string) error {
return i.client.PutWebPush(ctx, account, notify.PushSubscription{
Endpoint: endpoint, Keys: notify.PushKeys{P256dh: p256dh, Auth: auth},
})
}
func (i inboxCounter) PushUnsubscribe(ctx context.Context, account, endpoint string) error {
return remote(i.client.DeleteWebPush(ctx, account, endpoint))
}
// pushDevices adapts forge OLTP to api.PushDeviceStore, mapping the
// storage row to the ops layer's own shape so internal/api keeps
// naming what it needs.
type pushDevices struct{ db *oltp.DB }
func (p pushDevices) AddPushDevice(account, endpoint, ua string) error {
return p.db.AddPushDevice(account, endpoint, ua)
}
func (p pushDevices) DeletePushDevice(account, endpoint string) error {
return p.db.DeletePushDevice(account, endpoint)
}
func (p pushDevices) MarkPushDevicePendingDelete(account, endpoint string) error {
return p.db.MarkPushDevicePendingDelete(account, endpoint)
}
func (p pushDevices) PendingPushDeletes(account string) ([]string, error) {
return p.db.PendingPushDeletes(account)
}
func (p pushDevices) PushDevices(account string) ([]api.PushDevice, error) {
rows, err := p.db.PushDevices(account)
if err != nil {
return nil, err
}
out := make([]api.PushDevice, 0, len(rows))
for _, r := range rows {
out = append(out, api.PushDevice{EndpointHash: r.EndpointHash, UA: r.UA, At: r.At,
PendingDelete: r.PendingDelete})
}
return out, nil
}
// prefStore adapts the OLTP to web's per-account preference seam. The
// web layer names "a preference" and knows nothing about which table it
// lands in, the same separation every other store here keeps.
type prefStore struct{ db *oltp.DB }
func (p prefStore) Pref(account, key string) string { return p.db.AccountPref(account, key) }
func (p prefStore) SetPref(account, key, value string) error {
return p.db.SetAccountPref(account, key, value)
}