// quarry is a multi-repository forge for zit. See docs/rfcs/.
package main
import (
"bytes"
"context"
"craftdesign.group/quarry/internal/drain"
"craftdesign.group/quarry/internal/errlog"
"craftdesign.group/quarry/internal/fault"
"craftdesign.group/quarry/internal/health"
"craftdesign.group/quarry/internal/runnerdist"
"crypto/rand"
"encoding/hex"
"encoding/json"
"errors"
"flag"
"fmt"
"log"
"net"
"net/http"
"net/url"
"os"
"os/exec"
"os/signal"
"path/filepath"
"regexp"
"sort"
"strings"
"sync"
"syscall"
"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/cibridge"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/evt"
"craftdesign.group/quarry/internal/imgproxy"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/mail"
"craftdesign.group/quarry/internal/markup"
"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 "runner":
if err := runnerCmd(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, stamp, 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)
}
// nodeID is this process's identity across the cluster: the owner its
// leases carry and the salt of its durable consumer names. It is not
// --addr, which is only where to listen: two nodes behind local proxies
// both listen on the default and would share one lease and one
// consumer. The first start writes a random id beside the derived data
// and every later start reads it back, so a restart keeps its identity;
// an explicit id wins and leaves the file alone.
func nodeID(explicit, dir string) (string, error) {
if explicit != "" {
return explicit, nil
}
path := filepath.Join(dir, "node-id")
b, err := os.ReadFile(path)
if err == nil {
id := strings.TrimSpace(string(b))
if id == "" {
return "", fmt.Errorf("%s is empty: delete it to mint a new node id, or pass --node-id", path)
}
return id, nil
}
if !errors.Is(err, os.ErrNotExist) {
return "", err
}
var raw [6]byte
if _, err := rand.Read(raw[:]); err != nil {
return "", err
}
id := hex.EncodeToString(raw[:])
tmp, err := os.CreateTemp(dir, "node-id.*")
if err != nil {
return "", err
}
defer os.Remove(tmp.Name())
_, werr := tmp.WriteString(id + "\n")
if serr := tmp.Sync(); werr == nil {
werr = serr
}
if cerr := tmp.Close(); werr == nil {
werr = cerr
}
if werr == nil {
werr = os.Chmod(tmp.Name(), 0o600)
}
if werr != nil {
return "", werr
}
// Link, not rename: it refuses to replace an id another start wrote
// first, and that one then stands.
if err := os.Link(tmp.Name(), path); errors.Is(err, os.ErrExist) {
return nodeID("", dir)
} else if err != nil {
return "", err
}
return id, nil
}
func serveCmd(args []string) error {
// Deferred first so it runs last: after the drain and after every
// teardown below (ingest, fronter, leases, events, database), the
// stop waits for the zit children still running and logs each end.
defer stopChildren()
// 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")
joinFlag := fs.String("join-notify", os.Getenv("QUARRY_JOIN_NOTIFY"), "accounts emailed when an invited person accepts their invitation, comma-separated (default: nobody; or $QUARRY_JOIN_NOTIFY)")
nodeFlag := fs.String("node-id", os.Getenv("QUARRY_NODE_ID"), "this node's identity for leases and event consumers (default: a random id kept in <data>/node-id; or $QUARRY_NODE_ID)")
ciCloneURL := fs.String("ci-clone-base", "", "base URL CI workers clone from (default: this server, at loopback when --addr names no host)")
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")
runnerDir := fs.String("runner-dir", runnerdist.DefaultDir(), "directory holding the quarry-runner builds served at /runner/ (deploy installs them beside the binary)")
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")
healthEvery := fs.Duration("health-every", 30*time.Second,
"how often to probe dagnats and notify for liveness")
// 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")
allowPrivateBridges := fs.Bool("allow-private-bridges", false,
"permit chat bridges to loopback, private and link-local addresses (RFC 0011 §9)")
// 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")
// Creation is open to any signed-in account (RFC: open repository
// creation); an open door needs a limit, and there is no forge
// settings plane yet to hold it, so it is a flag until there is
// one. A site admin is never subject to it.
repoQuota := fs.Int("repo-quota", 25, "repositories a non-admin account may administer at once")
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")
originFlag := fs.String("origin", os.Getenv("QUARRY_ORIGIN"), "public URL the forge lives at, e.g. https://forge.example — every link in an email, push or notification starts with it (or $QUARRY_ORIGIN)")
// No header is believed by default: a client can write any of them,
// and the login limiter keyed on one could then be dodged or aimed.
proxyHeader := fs.String("trusted-proxy-header", os.Getenv("QUARRY_TRUSTED_PROXY_HEADER"), "the header the proxy in front sets and clients cannot, e.g. CF-Connecting-IP behind Cloudflare; requests carrying it are believed for the client address and X-Forwarded-Proto. Empty believes no forwarded header (or $QUARRY_TRUSTED_PROXY_HEADER)")
zitParallel := fs.Int("zit-parallel", 0, "at most this many store subprocesses at once; callers past it are told busy (503). Default: CPU count, minimum 2")
mailFrom := fs.String("mail-from", os.Getenv("QUARRY_MAIL_FROM"), "sender for invite emails, e.g. 'quarry <[email protected]>'; the domain must be verified at Resend (or $QUARRY_MAIL_FROM)")
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}
// Repositories register after the listener binds; pending is what
// the forge says about a name until its registration finishes.
pending := &pendingSet{}
var registerAll func()
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
// repoPlace answers where a repository appears and whether the name
// is a repository at all. Carried out of the registry's scope the
// same way capsFor is, because the web options are assembled later.
var repoPlace func(name string) (string, bool)
// repoWasAt answers where a repository USED to appear, for the links
// saved before it moved (c12a84b2).
var repoWasAt func(name, ns string) bool
// Whether a repository is private, from the same cached set the sync
// gate reads. The canonical redirect needs it: privacy is a column on
// the repository, not a capability letter, and a repository with no
// per-repository categories inherits a fleet default whose `nobody`
// set carries every read letter.
var repoIsPrivate func(repo string) bool
// markTag answers the uploaded-image lookup for every listing that
// draws a mark. Declared out here, assigned where the database is.
var markTag func(kind, name string) (string, bool)
var markRefresh func()
// operatorOf says which accounts are machine-operated and who
// answers for them; nil without an accounts plane.
var operatorOf func(name string) string
var capsAdmin web.Caps
var protectAdmin web.Protect
var inbox web.Inbox
var webHooks web.Hooks
var webHookSend web.HookSender
var webTallies web.Tallies
var webStamp web.Stamp
var webPrefs web.Prefs
var webEvents web.EventLog
var webSetLog web.SettingsLog
var engine *ci.Engine
// Declared alongside the engine so the web handler can render what
// the watcher believes.
var depWatch *health.Watch
var ciCloneBase string
var fronter *xfer.Fronter
// Which node runs the background jobs: nil, with no repositories and
// so no lease manager, runs them all here.
var jobs *jobGate
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
}
node, err := nodeID(*nodeFlag, dir)
if err != nil {
return err
}
log.Printf("node id %s", node)
deps.Node = node
for _, n := range strings.Split(*joinFlag, ",") {
if n = strings.TrimSpace(n); n != "" {
deps.JoinNotify = append(deps.JoinNotify, n)
}
}
// 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()
// The durable half of the error journal. The ring answers the panel;
// this is what lets it answer for more than the current process.
// Local-first on purpose: recording an error must not depend on any
// service being up, because the moment errors matter most is the
// moment other things are down.
if odb != nil {
// Every occurrence is stamped with the build that saw it, so
// "is this still happening?" survives a deploy.
errlog.Default.SetBuild(version)
// The rate comes from the durable half: the ring counts, the
// table remembers when.
deps.ErrorRates = odb.Rates
if history, herr := odb.Exceptions(500); herr == nil {
seed := make([]errlog.Entry, 0, len(history))
for _, e := range history {
seed = append(seed, errlog.Entry{Message: e.Message, Surface: e.Surface,
Path: e.Path, Count: e.Count, First: e.First, Last: e.Last,
Repo: e.Repo, Trace: e.Trace, Group: e.Sig,
Why: e.Why, Fix: e.Fix, Fields: e.Fields, Build: e.Build, Req: e.Req})
}
errlog.Default.Seed(seed)
} else {
log.Printf("errlog: history unreadable: %v", herr)
}
// Both halves of the journal are retired together, and by
// two different rules. Age retires what has stopped on its
// own; the clear control retires what a maintainer has
// fixed and does not want counted any more. Neither is
// optional on a forge that runs for months: the ring is
// bounded by capacity, but the table it seeds from had no
// bound at all.
deps.ForgetError = func(sig string) error {
return odb.ForgetException(sig)
}
pruneErrors := func() {
cut := time.Now().Add(-oltp.ExceptionKeep)
if n, perr := odb.PruneExceptions(cut); perr != nil {
log.Printf("errlog: prune: %v", perr)
} else if n > 0 {
log.Printf("errlog: retired %d signature(s) last seen over %s ago", n, oltp.ExceptionKeep)
}
errlog.Default.Prune(cut)
}
pruneErrors()
// ci_run had the same shape of defect and the same fix: a row
// written per launch and never removed, in the database 0001
// §8 keeps small by construction. Swept on the same beat,
// because both are retention of derived bookkeeping and a
// second timer would only be a second thing to reason about.
pruneRuns := func() {
cut := time.Now().Add(-oltp.CIRunKeep)
if n, perr := odb.PruneCIRuns(cut); perr != nil {
log.Printf("ci: prune: %v", perr)
} else if n > 0 {
log.Printf("ci: retired %d superseded run(s) older than %s", n, oltp.CIRunKeep)
}
}
pruneRuns()
go func() {
for range time.Tick(6 * time.Hour) {
pruneErrors()
pruneRuns()
}
}()
errlog.Default.SetSink(func(e errlog.Entry) {
if serr := odb.RecordException(oltp.Exception{
Sig: e.Sig(), Message: e.Message, Surface: e.Surface, Path: e.Path,
Repo: e.Repo, Trace: e.Trace, Count: e.Count,
First: e.First, Last: e.Last,
Why: e.Why, Fix: e.Fix, Fields: e.Fields, Build: e.Build, Req: e.Req,
}); serr != nil {
// Logged, never journalled: reporting the failure of the
// journal INTO the journal is how a loop starts.
log.Printf("errlog: persist: %v", serr)
}
})
}
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}
webSetLog = settingsLog{db: odb}
webStamp = odb
webPrefs = prefStore{db: odb} // RFC 0011 §5: the desktop-notification switch
deps.Stamp = odb
deps.SettingsLog = odb.SettingsLog
deps.Companions = odb
deps.Buckets = odb // RFC 0016: where a repository's uploads go
deps.Avatars = odb // uploaded marks, in the forge's own bucket
deps.Attachments = odb // RFC 0016 Phase 3: tickets, holds, the policy switches
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.Local = odb // the local-ticket tier (docs/local-tickets.md)
deps.Idem = odb // a retried filing returns the first ticket
deps.Numbers = odb // ticket numbers (54c5175f)
deps.TicketTier = odb.TicketTier // ...and where each repository keeps its own
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}
var codeMu sync.Mutex
codes := map[string]string{}
// repoPath finds a repository's directory.
//
// The --repo flags are the override and are checked first, but
// they are NOT the whole fleet: every repository the forge
// serves from its repos-root, and every repository created
// through the API, lives in the oltp registry and never enters
// this map. Looking only here meant the project code — and so
// the push credential derived from it — existed for the one
// repository named on the command line and for nothing else.
repoPath := func(repo string) (string, bool) {
if path, ok := repos[repo]; ok {
return path, true
}
if odb == nil {
return "", false
}
stored, err := odb.Repos()
if err != nil {
log.Printf("auth: repository list for %s: %v", repo, err)
return "", false
}
for _, r := range stored {
if r.Name == repo {
return r.Path, true
}
}
return "", false
}
// The editor's save: a scratch checkout against the repository's
// own store — the serving tree is never touched (RFC 0009).
//
// Resolved through repoPath, not the --repo map. That map holds
// only the repositories named on the command line, which on the
// forge is quarry itself; every other repository lives in the
// oltp registry. Looking in the map alone meant the editor could
// save a pipeline for exactly one repository and answered every
// other with a bare 500 that said nothing — measured at 698ms on
// dantest-sandbox, "unknown repository" in the log and "internal
// error" to the caller (a6110cdc, G-64). The same mistake was
// already fixed for the push credential, two closures down; this
// one predates that fix and kept the old lookup.
deps.SavePipeline = func(tok lease.Token, repo, branch, user, message, path string, body []byte) (string, error) {
dir, ok := repoPath(repo)
if !ok {
return "", fmt.Errorf("no repository %q: %w", repo, ops.ErrNotFound)
}
return zitcli.CommitFile(tok, filepath.Join(dir, ".zit"), "trunk", branch, user, message, path, body)
}
projectCode := func(repo string) (string, bool) {
codeMu.Lock()
code, cached := codes[repo]
codeMu.Unlock()
if cached {
return code, true
}
path, ok := repoPath(repo)
if !ok {
return "", false
}
c, err := xfer.ProjectCode(path)
if err != nil {
log.Printf("auth: project code of %s: %v", repo, err)
return "", false
}
codeMu.Lock()
codes[repo] = c
codeMu.Unlock()
return c, true
}
// Without this the sync secret cannot be derived at all, and a
// gated repository admits nobody (internal/oltp/synctoken.go).
odb.SyncProjectCode = projectCode
deps.SyncTokens = odb // minting the push credential (RFC 0001 §7)
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
}
code, ok := projectCode(repo)
if !ok {
return api.Denied
}
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(), node, *leaseTTL)
if err != nil {
return err
}
defer leases.Close()
jobs = &jobGate{leases: leases, code: func(repo string) (string, bool) {
codeMu.Lock()
defer codeMu.Unlock()
code, ok := codes[repo]
return code, ok
}}
// 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
// Who is reading which channel, over the same bus: core NATS,
// not the stream, because presence must not be durable.
pres, err := chat.NewPresence(ev.Conn())
if err != nil {
return err
}
ca := &chatAdapter{room: room, pres: pres, codes: codes, mu: &codeMu}
// Teams are OLTP-only; with no database there are none, and
// TailTeam has nothing to merge.
if odb != nil {
ca.teams = odb
}
chatLive = ca
// 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, book: odb}
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)
// The path is the operation, not the workflow:
// the workflow's name rides in the fault's
// fields, so one unreachable notify is one row.
errlog.Default.Report("notify", "ensure workflows", err)
}
}
// A workflow with an email step and no mail provider
// behind it fails where only somebody reading notify's
// database would find it — and quarry's settings page
// offers an email column the whole time. quarry already
// holds a mail credential for invites, so the companion
// is given the same one rather than an operator being
// asked to POST it by hand.
if key := os.Getenv("RESEND_API_KEY"); key != "" && *mailFrom != "" {
switch made, err := nclient.EnsureEmailIntegration(context.Background(),
notify.EmailSender{Provider: "resend", Identifier: "quarry-mail",
From: *mailFrom, Key: key}); {
case err != nil:
log.Printf("notify: email integration: %v", err)
errlog.Default.Report("notify", "email integration", err)
case made:
log.Printf("notify: email now sends as %s", *mailFrom)
}
}
// A reply to a notification email comes back through
// notify's delivery stream (RFC 0014 Phase 1), which
// needs a signed endpoint at this forge to deliver to.
// notify mints the secret and shows it once, here; the
// forge keeps that copy. Needs --origin: notify cannot
// be told to POST to a host the forge does not know its
// own name at.
if origin := strings.TrimSuffix(*originFlag, "/"); origin != "" {
hookURL := origin + "/api/v1/notify/events"
switch ep, err := nclient.EnsureReplyEndpoint(context.Background(), hookURL); {
case err != nil:
log.Printf("notify: reply endpoint: %v", err)
errlog.Default.Report("notify", "register reply endpoint", err)
case ep.Secret != "":
if err := odb.SetNotifyHook(ep.ID, hookURL, ep.Secret); err != nil {
log.Printf("notify: reply endpoint: %v", err)
} else {
log.Printf("notify: replies now arrive signed at %s", hookURL)
}
}
}
// The workflows exist; the people they address have to
// as well. Every account and grant made before this
// provisioning shipped left notify without the
// subscriber or topic the trigger names, which is not
// visible until an alert is sent to nobody.
sync, err := reconcileNotify(context.Background(), nclient, odb)
if err != nil {
log.Printf("notify: reconcile: %v", err)
errlog.Default.Report("notify", "reconcile subscribers and topics", err)
}
log.Printf("notify: %d subscribers, %d topic memberships current, %d email defaults written",
sync.Subscribers, sync.Members, sync.Quieted)
}()
// 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)
}
}
// The result is NOT discarded: an alert channel that
// silently fails is worse than none, because the
// journal then says the fault was reported. Logged
// rather than reported to the journal itself — that
// call is what is running right now.
err := np.Trigger(context.Background(), api.NotifyEvent{
Workflow: "quarry-error", To: adminsTopic,
Payload: map[string]any{
"surface": e.Surface, "path": e.Path,
"summary": e.Message,
// Where the fault is written up. Without it the
// inbox row is a sentence that leads nowhere.
"link": strings.TrimSuffix(deps.Origin, "/") + "/fleet/exceptions",
},
TxnID: "err:" + e.First.Format(time.RFC3339Nano),
})
if err != nil {
log.Printf("notify: error alert not delivered: %v", err)
}
// And into the forge's own inbox, one row per admin.
// notify is a companion: it can be absent, misconfigured
// or unable to deliver, and none of those should mean an
// operator learns about a fault by reading the journal.
// This half is quarry's own and needs nothing outside it.
alertAdmins(odb, deps.Attention, e)
}
// 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) {
err := np.Trigger(context.Background(), api.NotifyEvent{
Workflow: "quarry-slow", To: adminsTopic,
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)),
// The route's own timing page, not the ops index:
// the alert names one route and so should its link.
"link": strings.TrimSuffix(deps.Origin, "/") +
"/fleet/ops/route?path=" + url.QueryEscape(v.Path),
},
// 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,
})
if err != nil {
log.Printf("notify: slow-page alert not delivered: %v", err)
errlog.Default.Report("notify", "slow-page alert", err)
}
}
}
// The services the forge cannot do its job without. Watched on a
// beat, because their absence is otherwise silent: every feature
// that needs them degrades politely on its own and nobody is
// ever told the fleet is running on three cylinders.
depWatch = health.New([]health.Dep{
{Name: "dagnats", URL: *dagnatsURL,
Why: "CI cannot run: no pipeline will start and no verdict will land"},
{Name: "notify", URL: *notifyURL,
Why: "no mentions, activity or alerts are being delivered"},
// Not a service: a promise. The privacy page says the
// pages load nothing from another host, and only a read
// from OUTSIDE can tell whether that is still true by the
// time a reader gets one — the origin is the half that was
// never in doubt. Watched only where the forge knows its
// own public address.
{Name: "origin", URL: originCheckURL(*originFlag),
Why: "readers are fetching from a third party this forge did not send them to",
Broken: "the page readers receive is not the page this forge sent",
Fix: "something between this forge and its readers is injecting into the HTML — " +
"on Cloudflare this is Web Analytics' automatic beacon; turn it off at the front, " +
"then compare the origin's HTML with the public one to confirm",
Probe: health.ForeignProbe(nil, 15*time.Minute)},
}, func(d health.Dep, up bool, perr error) {
// A recovery is NOT an error. Journalling one put "dagnats
// recovered" in the error list, counted it in the fleet's
// error badge, and split one flapping dependency into two
// rows that had to be read together. The fact worth keeping
// — this has come back before, and when — belongs ON the
// outage, which is the row someone acts on.
flap := depFlaps.seen(d.Name, up)
if up {
log.Printf("health: %s recovered", d.Name)
return
}
log.Printf("health: %s is down: %v", d.Name, perr)
// The journal FIRST: it is local and durable, and it cannot
// be the thing that is broken. Telling the notifier that
// the notifier is down would be a message nobody receives.
headline, fix := d.Broken, d.Fix
if headline == "" {
headline = d.Name + " is unreachable"
}
if fix == "" {
fix = "start " + d.Name + " and it will be picked up on the next health beat; nothing needs restarting here"
}
f := fault.Wrap(perr, headline).
Cost(d.Why).
Do(fix).
With("dependency", d.Name, "endpoint", d.URL, "down since", flap.since)
if flap.outages > 1 {
f.With("outages", flap.outages)
}
if flap.recovered != "" {
f.With("last recovered", flap.recovered)
}
errlog.Default.Report("health", d.Name, f)
})
go depWatch.Run(context.Background(), *healthEvery)
// 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.
//
// A listen address may name no host — ":8950" means every
// interface — and pasting that into a URL produces
// "http://:8950/r/x", which zit rejects as invalid and
// every pipeline fails its clone in under a second. A
// server listening everywhere is reachable at loopback
// from the box it runs on, which is where a worker lives
// unless someone says otherwise, so that is the host we
// name. --ci-clone-base overrides it for a worker
// elsewhere.
ciCloneBase = cloneBaseFor(*addr, *ciCloneURL)
deps.CI = ciRuns{db: odb}
deps.Deferred = odb // the check-ins waiting for a worker slot (029c4730)
// An admin's answer to "does this repository use CI", read on
// every repo page's counts so the navigation can stop
// offering to set CI up once someone has said no.
deps.CIDecline = odb
deps.ValidateCIYAML = engine.ValidateYAML
fmt.Println("quarry: CI ->", *dagnatsURL)
go func() {
for range time.Tick(*ciPoll) {
if jobs.fleet() {
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)
// The panel that exists to explain a refusal can only show the
// lanes it is handed. These two live here — nowhere else holds
// them — and until now the ops page listed the store and
// archive lanes and not the one doing the refusing.
deps.EventWindow = ev.Window
deps.Lanes = func() []zitcli.Lane {
out := make([]zitcli.Lane, 0, 3)
for _, l := range []*work.Lane{ingestLane, deps.Bulk} {
if l == nil {
continue
}
st := l.Stats()
lane := zitcli.Lane{
Name: st.Name, Bound: st.Limit, Running: int(st.InFlight),
Admitted: int(st.Admitted), Refused: int(st.Shed),
Waiting: st.Waiting, Queue: st.Queue,
}
// Age is resolved here rather than sent as a timestamp:
// the page is rendered on this machine, and a duration
// is what a reader is looking for.
for _, it := range l.Items() {
lane.Items = append(lane.Items, zitcli.LaneItem{
Verb: it.Verb, Subject: it.Subject,
AgeMs: it.Age.Milliseconds(), Running: it.Running,
})
}
out = append(out, lane)
}
// Rebuilds run on slots of their own, outside every lane (a
// round that becomes one leaves the ingest lane), so no lane
// above lists them. One more row says what is rebuilding.
rb := ingest.Rebuilds()
lane := zitcli.Lane{Name: "rebuild", Bound: rb.Bound, Running: rb.Running,
Waiting: int64(rb.Queued), Admitted: int(rb.Started)}
for _, it := range rb.Items {
lane.Items = append(lane.Items, zitcli.LaneItem{
Verb: "rebuild", Subject: it.Store,
AgeMs: it.Age.Milliseconds(), Running: it.Running,
})
}
return append(out, lane)
}
deps.Mirrors = mirrorStore{db: odb}
// A fossil repository's users, kept as a reference an admin
// links to accounts rather than as accounts (RFC 0004 §6).
deps.FossilUsers = 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}
// Who is asking, resolved exactly the way the visibility gate
// resolves it. Sharing one answer is the point: "who may read
// this" and "who is recorded as having read it" drifting into
// two notions of identity is how a tally starts naming the
// wrong person.
fronter.SetIdentify(func(req *http.Request) string {
return ops.User(syncIdentityCtx(req, odb))
})
// A push the gate admitted onto a TRACKING store means the store
// now holds check-ins its upstream never saw (5d40f151). Called
// per admitted round, with no hash: the round is a container the
// child ingests. MirrorDiverged is a no-op for anything that is
// not an enabled mirror, so every push can report.
fronter.SetPushed(func(repo string) { deps.MirrorDiverged(repo, "") })
fronter.SetCounted(func(repo string, rd xfer.Reader) {
kind, name := evt.CloneFetched, "fetch"
if rd.Clone {
kind, name = evt.CloneCloned, "clone"
}
// The EVENT stays identity-free, and deliberately so: a
// repository's event log is as public as the repository,
// so a clone event there says how many and never who. The
// identity lives only in the counters below, which no
// ungated page reads.
if err := ev.Raise(kind, repo, map[string]any{"repo": repo}); err != nil {
log.Printf("events: raise %s: %v", kind, err)
}
count := func(suffix string) {
if err := ev.Count(repo, name+suffix, 1); err != nil {
log.Printf("events: count %s/%s%s: %v", repo, name, suffix, err)
}
}
count("")
count(".client." + events.CounterToken(rd.Client) + "." + events.CounterToken(rd.Version))
// The three identity states never share a prefix. One
// prefix holding both accounts and claims would make a
// forged name indistinguishable from a real one the moment
// anybody read the total — and a forgery is free, since
// nothing signs a pull.
switch {
case rd.Account != "":
count(".auth." + events.CounterToken(rd.Account))
case rd.Claimed != "":
count(".claim." + events.CounterToken(rd.Claimed))
default:
count(".anon")
}
})
go mirrorScheduler(&deps, odb, *mirrorTick, jobs)
// RFC 0011 §9: one goroutine per bridged mirror, started and
// stopped by the same tick the mirror pulls run on — a bridge
// configured while the server is up must not need a restart.
deps.ChatBridges = chatBridgeStore{db: odb}
bridgeCtx, stopBridges := context.WithCancel(context.Background())
defer stopBridges()
go (&bridgeRunner{room: room, odb: odb, tick: *mirrorTick,
allowPrivate: *allowPrivateBridges, holds: jobs.store,
code: func(repo string) (string, bool) {
codeMu.Lock()
defer codeMu.Unlock()
code, ok := codes[repo]
return code, ok
}}).run(bridgeCtx)
// 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
}
// The delivery log. Attached before Start so the first attempt
// after boot is recorded like every other one.
dispatcher.Record(hookLog{db: odb})
webHookSend = hookSender{db: odb, d: dispatcher}
// The same three actions through the ops layer, so the CLI and
// an agent can do what the settings page does.
deps.Hooks = hookSender{db: odb, d: dispatcher}
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)
}
// The proxied sync is written by xfer's own zit process, not
// through zitcli, so there is no call to hand the token to;
// the lease is still held for its whole length.
return leases.Guard(code, func(lease.Token) error { return proxy() })
})
rg := ®istrar{
dir: dir, reposRoot: *reposRoot, node: node, interval: *interval, ciCloneBase: ciCloneBase,
ev: ev, leases: leases, jobs: jobs, deps: &deps, fronter: fronter,
ingestLane: ingestLane, bulk: deps.Bulk, engine: engine, odb: odb,
ciTagsDefault: *ciVerdicts,
repoQuota: *repoQuota,
codes: codes, codeMu: &codeMu,
}
defer rg.CloseAll()
deps.Registry = rg
// Waiting check-ins are released on the CI poll tick as well as
// after a fold, or a quiet repository holds them for ever. Each
// store's own lease gates it, inside drainDeferred.
if engine.Enabled() {
go func() {
for range time.Tick(*ciPoll) {
rg.drainDeferred()
}
}()
}
deps.ProjectCode = func(repo string) (string, error) {
codeMu.Lock()
defer codeMu.Unlock()
code, ok := codes[repo]
if !ok {
return "", fmt.Errorf("unknown repo %q: %w", repo, ops.ErrNotFound)
}
return code, nil
}
deps.CITrigger = rg.TriggerCI // RFC 0009: the re-run path
deps.CIRunPair = rg.TriggerPairCI
deps.Origin = strings.TrimSuffix(*originFlag, "/")
if *zitParallel > 0 {
zitcli.SetParallel(*zitParallel)
}
log.Printf("admission: at most %d store subprocesses at once", zitcli.Parallel())
deps.ZitVersion = zitcli.ProbeVersion()
// Said at startup so a deploy onto an old zit is visible before
// anyone presses accept; the accept itself refuses on its own.
if !zitcli.VersionAtLeast(deps.ZitVersion, zitcli.MinVersion) {
zitcli.WarnTooOld(deps.ZitVersion)
}
deps.Tracker = *tracker
// The changelog is CHANGELOG.md on the tracker repository's
// trunk: it lands like any change and is reviewed with the code.
if *tracker != "" {
trackerName := *tracker
deps.Changelog = func(ctx context.Context) ([]byte, error) {
rp, ok := deps.Repos.Get(trackerName)
if !ok || rp.Zit == nil {
return nil, fmt.Errorf("no tracker repository %q", trackerName)
}
return rp.Zit.Cat(ctx, "CHANGELOG.md", "trunk")
}
}
deps.Accounts = accountStore{db: odb}
// Invites by email go straight to Resend when a key is present:
// the person has no account, so no inbox the notify plane could
// reach. The key is only ever read from the environment.
if rs := mail.New(os.Getenv("RESEND_API_KEY"), *mailFrom); rs != nil {
deps.Mail = mailAdapter{rs}
fmt.Println("quarry: invite mail -> resend as", rs.From())
}
capsFor = func(account, repo string) api.CapSet { return odb.EffectiveCaps(account, repo) }
deps.CapsFor = capsFor
deps.Owners = odb
deps.Acts = odb
repoPlace = func(name string) (string, bool) {
if _, ok, err := odb.RepoByName(name); err != nil || !ok {
return "", false
}
ns, err := odb.NamespaceOf(name)
if err != nil {
return "", true
}
return ns, true
}
repoWasAt = odb.WasPlacedAt
// 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}
repoIsPrivate = rg.IsPrivate
marks := newMarkCache(odb)
markTag, markRefresh = marks.tag, marks.refresh
operatorOf = odb.Operator
// The roster's STRING seam, for account.caps and the events it
// raises. Separate from capsPlane because that one's CapsOf
// answers an effective CapSet — the question a gate asks — and
// this one answers the STORED string, which is the question a
// change is described by.
deps.Roster = rosterPlane{db: odb}
deps.Teams = odb
deps.Teams = odb
deps.Visibility = odb.VisibilityName
protectAdmin = odb // RFC 0007: the settings page edits the rules under `a`
deps.Votes = odb
deps.Forum = odb
deps.Replies = 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,
// refined by whether mirroring itself is allowed (rg.SyncReadGate).
fronter.SetReadGate(rg.SyncReadGate)
// Tells "not registered yet" from "no such repository". Stores
// are registered one at a time at startup, and a sync arriving
// before its turn was told the repository did not exist.
fronter.SetKnown(func(name string) bool {
_, ok, err := odb.RepoByName(name)
return err == nil && ok
})
deps.Mirrorable = func(repo string) bool {
ok, err := odb.MayMirror(repo)
if err != nil {
return true
}
return ok
}
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),
Chat: r.Chat, Mirrorable: r.Mirrorable}
}
return m
}
var toRegister []namedPath
// 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 {
// Settle anything a previous run left mid-import BEFORE
// registering, so an interrupted clone is a failed row
// rather than a path that fails to register every boot.
rg.ReconcileImports()
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
}
// A repository whose store never finished copying has
// nothing to serve. The ROW stays — it is the record of
// the attempt, and where the failure is reported.
if r.State != oltp.Ready {
continue
}
toRegister = append(toRegister, namedPath{r.Name, r.Path})
}
}
// A misconfigured --repo is still fatal, and still before the
// bind: the cheap check (is there a store?) happens now, the
// expensive registration after the listener is up.
for name, path := range repos {
if _, err := os.Stat(filepath.Join(path, ".zit")); err != nil {
return fmt.Errorf("--repo %s: %s is not a zit checkout: %w", name, path, err)
}
toRegister = append(toRegister, namedPath{name, path})
// 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)
}
}
}
pending.set(toRegister)
deps.Pending = pending.list
registerAll = func() {
start := time.Now()
// A few at a time, not one after another. Serially, 26
// stores took 46–97s to register and the forge answered
// "still starting" to every sync and every repository page
// for that whole window (a3b9623e, b81ae623 option 3). Each
// registration touches its own index and its own zit; what
// they share — the served set, the fronter, the project
// codes, the pending list — is locked. The window is now the
// slowest store plus a little, rather than the sum.
eachConcurrently(toRegister, registerWorkers, func(np namedPath) {
if err := rg.Register(np.name, np.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", np.name, np.path, err)
}
pending.done(np.name)
})
// No refreshPrivate here: Register does it, because a
// repository that is served and missing from the read gate's
// cache is a private repository that does not exist for
// anyone else being announced to everyone else.
log.Printf("registered %d repositories in %s", len(toRegister), time.Since(start).Round(time.Millisecond))
}
}
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.
// Mergeability is answered before anyone asks: the probe that used
// to run on a reader's first view after a push runs here instead.
go func() {
for range time.Tick(20 * time.Second) {
deps.WarmMergeProbes()
}
}()
if sweeper, ok := deps.Auto.(api.AutoSweeper); ok {
go func() {
for range time.Tick(30 * time.Second) {
for _, id := range deps.AutoAcceptSweep(heldSweeper{AutoSweeper: sweeper, jobs: jobs}) {
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:
if jobs.fleet() {
deps.ChatWindowSweep()
}
case <-sweepCtx.Done():
return
}
}
}()
}
var extraOpts []web.Option
// The sending half of webhooks: present only once a dispatcher
// exists, which is why it is an option rather than a parameter.
if webHookSend != nil {
extraOpts = append(extraOpts, web.WithHookSender(webHookSend))
}
// The durable settings log. A forge with no registry database has
// no log and the section simply does not appear, the same way it
// did not appear before there was one.
if webSetLog != nil {
extraOpts = append(extraOpts, web.WithSettingsLog(webSetLog))
}
// What the watcher believes, so the fleet page can say "dagnats is
// down" instead of leaving a maintainer to infer it from every CI
// run failing.
if deps.Origin != "" {
extraOpts = append(extraOpts, web.WithOrigin(deps.Origin))
}
proxy := web.TrustedProxy(strings.TrimSpace(*proxyHeader))
extraOpts = append(extraOpts, web.WithTrustedProxy(proxy))
// External images in rendered documents load through the forge: the
// renderer signs each one and /img fetches it once, guarded and
// cached, into the data directory. Without a data directory there is
// nowhere to keep the key, and images stay as the author wrote them.
if dataRoot != "" {
proxy, err := imgproxy.New(dataRoot, false)
if err != nil {
return fmt.Errorf("image proxy: %w", err)
}
markup.SetImageProxy(proxy.URL)
extraOpts = append(extraOpts, web.WithImageProxy(proxy))
}
if depWatch != nil {
extraOpts = append(extraOpts, web.WithHealth(depWatch.States))
}
// 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())
}
if repoPlace != nil {
// Where each repository appears, for the URL helper and the
// path rewriter. Two answers from one lookup: the namespace
// segment, and whether the name is a repository at all — the
// second is what lets a path's first segment be read without
// guessing, because a repository name always wins.
extraOpts = append(extraOpts, web.WithNamespaces(repoPlace),
web.WithNamespaceHistory(repoWasAt))
}
if repoIsPrivate != nil {
extraOpts = append(extraOpts, web.WithPrivateRepos(repoIsPrivate))
}
if markTag != nil {
extraOpts = append(extraOpts, web.WithMarks(markTag), web.WithMarksChanged(markRefresh))
}
if operatorOf != nil {
// Who is a machine, and who answers for it. The accounts plane
// is the only thing that can know, and a forge without one has
// no agents because it has no accounts.
extraOpts = append(extraOpts, web.WithAgents(operatorOf))
}
mux, err := buildHandler(deps, notifier, chatLive, verify, capsFor, capsAdmin, protectAdmin, sessions, inbox, webHooks, webTallies, webStamp, webEvents, *tracker, webEngine(engine), ciLogStore{dir: dataRoot}, proxy.ClientAddr, extraOpts...)
if err != nil {
return err
}
// Kill any `zit serve --fronted` children a previous quarry left
// behind. A graceful stop stops its own (CloseAll unregisters each
// repository from the fronter, which stops its child); this is the
// fallback for a quarry that died instead: the stop watchdog, a
// second signal, a crash or a SIGKILL. Before SIGTERM was handled
// every restart was such a death, and one orphan per synced
// repository accumulated, twenty-four by the end of a day of deploys.
if n := xfer.ReapOrphans(); n > 0 {
log.Printf("reaped %d orphaned zit serve children", n)
}
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
}
// Runners on their owners' machines reach the CI engine's worker
// routes through the repository's URL (f24d78d9); the engine itself
// listens only on this box. Tokens are the engine's to check; the
// admin bearer is refused on this path.
if *dagnatsURL != "" {
bridge, berr := cibridge.Handler(*dagnatsURL, *dagnatsToken, ci.GroupFor, func(repo string) bool {
_, ok := deps.Repos.Get(repo)
return ok
})
if berr != nil {
return berr
}
mount := http.NewServeMux()
mount.Handle("POST /r/{repo}/ci/bridge/", bridge)
mount.Handle("/", mux)
mux = mount
}
// The runner builds, for owners running CI on their own machines.
// Mounted whether or not any build is present: a missing one answers
// 404 naming it, which is the truth on a forge built without them.
{
dist := http.NewServeMux()
dist.Handle("GET /runner/", runnerdist.Handler(*runnerDir))
dist.Handle("/", mux)
mux = dist
}
// Bind FIRST, register after. Registration shells out to zit per
// repository and took ~105s on the demo fleet, during which the old
// code was not listening at all — every request a connection
// refusal. Now the port answers at once; a repository that has not
// registered yet answers "still starting" with a refresh.
ln, err := net.Listen("tcp", *addr)
if err != nil {
return err
}
log.Printf("startup complete; bound %s", *addr)
fmt.Println("quarry", version, "listening on", *addr)
if registerAll != nil {
go registerAll()
}
// SIGTERM is how systemd restarts the forge. Without a handler it
// killed the process where it stood, so no deferred teardown above
// ran: streams and pushes were cut mid-response and ingest stopped
// mid-round. Now it returns from here and the defers run in reverse
// order: ingest pipelines stop and are waited on, hooks and bridges
// end, lease renewal stops, then the events bus and the database
// close, and last the zit children still running are waited for.
// A second signal kills at once, and the watchdog below ends a
// teardown that has itself hung.
stopCtx, stopSignals := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stopSignals()
go func() {
<-stopCtx.Done()
stopSignals()
time.AfterFunc(serveStopBound, func() {
log.Printf("stopping: teardown outlived %v; exiting", serveStopBound)
os.Exit(1)
})
}()
dr := drain.New()
srv := newServer(web.RequireHTTPS(*originFlag, proxy, web.PrivateByDefault(dr.Wrap(mux))))
return serveUntil(stopCtx, srv, ln, dr, serveDrainBound)
}
// 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) Forward(string, uint64, int) ([]chat.Message, uint64, error) { return nil, 0, nil }
func (docsChatPlane) Purge(string, ...uint64) error { return 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
pres *chat.Presence
codes map[string]string
mu *sync.Mutex
// teams answers which repositories a team holds, for the merged
// room. nil on a forge with no OLTP, where there are no teams and
// TailTeam has nothing to merge.
teams interface {
TeamRepos(team string) ([]oltp.TeamRepo, error)
}
}
func (a *chatAdapter) code(repo string) (string, bool) {
a.mu.Lock()
defer a.mu.Unlock()
c, ok := a.codes[repo]
return c, ok
}
// Join, Present and PresenceChanged satisfy the presence half of
// web.ChatLive over the same repo-to-code map Tail uses.
func (a *chatAdapter) Join(repo, user string) func() {
code, ok := a.code(repo)
if !ok || a.pres == nil {
return func() {}
}
return a.pres.Join(code, user)
}
func (a *chatAdapter) Present(repo string) []string {
code, ok := a.code(repo)
if !ok || a.pres == nil {
return nil
}
return a.pres.Present(code)
}
func (a *chatAdapter) Activity() (<-chan struct{}, func()) { return a.room.Activity() }
func (a *chatAdapter) PresenceChanged(repo string) (<-chan struct{}, func()) {
code, ok := a.code(repo)
if !ok || a.pres == nil {
return make(chan struct{}), func() {}
}
return a.pres.Changed(code)
}
// 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()
})
}
}
// TailTeam is Tail for a team's merged room: the team's own channel
// and every repository under it, fanned into one stream.
//
// It lives here rather than in the page because the page knows
// repositories by NAME and a team's own channel has no repository —
// only the composition root holds both halves of that. The same reason
// Tail takes a name and translates it.
//
// No resume. A merged stream has no single sequence to resume from:
// each channel numbers its own, so a Last-Event-ID would name a
// position in whichever channel happened to send it. A reader who
// reconnects re-reads the page, which is what the room is for.
//
// Each message carries the repository it was said in, so the row can
// draw its origin; a line said to the TEAM carries none, which is what
// ToTeam means.
func (a *chatAdapter) TailTeam(team string) (<-chan web.ChatMessage, func()) {
out := make(chan web.ChatMessage)
if a.teams == nil {
close(out)
return out, func() {}
}
rows, err := a.teams.TeamRepos(team)
if err != nil {
close(out)
return out, func() {}
}
// code -> the repository name to stamp on its rows. The team's own
// code maps to "", which is a line said to the team.
origin := map[string]string{chat.TeamCode(team): ""}
a.mu.Lock()
for _, r := range rows {
if code, ok := a.codes[r.Repo]; ok {
origin[code] = r.Repo
}
}
a.mu.Unlock()
done := make(chan struct{})
var wg sync.WaitGroup
var stops []func()
for code, repo := range origin {
src, stop := a.room.Tail(code, 0)
stops = append(stops, stop)
wg.Add(1)
go func(src <-chan chat.Message, repo string) {
defer wg.Done()
for m := range src {
select {
case out <- web.ChatMessage{
Seq: m.Seq, Kind: m.Kind, Ref: m.Ref,
Repo: repo, ToTeam: repo == "",
User: m.User, Text: m.Text, At: m.At,
Edited: m.Edited, Deleted: m.Deleted,
}:
case <-done:
return
}
}
}(src, repo)
}
// One closer for N producers: out is closed once, after every
// bridge has stopped, or a send racing the close panics.
go func() {
wg.Wait()
close(out)
}()
var once sync.Once
return out, func() {
once.Do(func() {
close(done)
for _, stop := range stops {
stop()
}
})
}
}
// markCache answers "does this have an uploaded image, and what do its
// bytes hash to" for every listing that draws a mark.
//
// One query per KIND rather than per row: the front page draws a mark
// for every repository, and asking per row would be a query per row.
// Refreshed on a short interval and by the ops that change a mark; a
// stale answer is at worst the previous image for a moment, never the
// wrong image at an address, because the digest is in the URL.
type markCache struct {
db *oltp.DB
mu sync.Mutex
at time.Time
by map[string]map[string]string
}
const markCacheFor = 15 * time.Second
func newMarkCache(db *oltp.DB) *markCache {
return &markCache{db: db, by: map[string]map[string]string{}}
}
// refresh makes the next lookup re-read, so an upload shows at once.
func (m *markCache) refresh() {
m.mu.Lock()
m.at = time.Time{}
m.mu.Unlock()
}
func (m *markCache) tag(kind, name string) (string, bool) {
m.mu.Lock()
defer m.mu.Unlock()
if time.Since(m.at) > markCacheFor {
// People too: a person's own picture shares the table, and
// leaving the kind out here drew every person as their
// identicon however many times they uploaded one.
for _, k := range []oltp.AvatarKind{oltp.AvatarRepo, oltp.AvatarTeam, oltp.AvatarUser} {
tags, err := m.db.AvatarETags(k)
if err != nil {
// Keep what we had rather than blanking every mark on
// the page because one read failed.
continue
}
m.by[string(k)] = tags
}
m.at = time.Now()
}
etag, ok := m.by[kind][name]
return etag, ok
}
// 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 }
// PushOutcome reads how one triggered push went (api.PushOutcomes).
func (a *notifyAdapter) PushOutcome(ctx context.Context, subscriber, txn string) (api.PushOutcome, error) {
r, err := a.client.PushOutcome(ctx, subscriber, txn)
return api.PushOutcome{Found: r.Found, Status: r.Status, Error: r.Error}, err
}
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)}
}
// The admin group is a topic too, and the one alert nobody watches
// for: addressed as a subscriber id it names nobody, and notify
// answers that with a success, so every error and slow-page alert
// was accepted and delivered to no one.
if e.To == adminsTopic {
to = notify.Recipient{TopicKey: adminsTopic}
}
if e.Email != "" {
// An address, not a subscriber: the person may not exist yet.
to = notify.Recipient{Email: e.Email}
}
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.
// namedPath is one repository to register: its forge name and checkout.
type namedPath struct{ name, path string }
// pendingSet is the set of repositories the forge has promised but not
// yet registered — what Deps.Pending answers while startup finishes.
type pendingSet struct {
mu sync.Mutex
names map[string]bool
}
func (p *pendingSet) set(all []namedPath) {
p.mu.Lock()
defer p.mu.Unlock()
p.names = make(map[string]bool, len(all))
for _, np := range all {
p.names[np.name] = true
}
}
func (p *pendingSet) done(name string) {
p.mu.Lock()
defer p.mu.Unlock()
delete(p.names, name)
}
func (p *pendingSet) list() []string {
p.mu.Lock()
defer p.mu.Unlock()
out := make([]string, 0, len(p.names))
for n := range p.names {
out = append(out, n)
}
sort.Strings(out)
return out
}
// mailAdapter is api.MailPlane over the Resend client.
type mailAdapter struct{ rs *mail.Resend }
func (m mailAdapter) Send(ctx context.Context, to, subject, text, html string) (string, error) {
return m.rs.Send(ctx, mail.Message{To: to, Subject: subject, Text: text, HTML: html})
}
// provisionAdapter carries the account book as well as the client
// because notify's subscriber upsert REPLACES the row: every call has to
// send the address, and the ops layer's callers know only a name. One
// place turns an account into a subscriber, so no call site can blank
// an address by not knowing it.
type provisionAdapter struct {
client *notify.Client
book *oltp.DB
}
func (a *provisionAdapter) EnsureSubscriber(ctx context.Context, account string) error {
var email string
if a.book != nil {
if acct, ok := a.book.AccountByName(account); ok {
email = acct.Email
}
}
if err := a.client.EnsureSubscriber(ctx, account, email); err != nil {
return err
}
// A new account is quiet by email until it chooses otherwise, the
// same answer the startup sweep gives everyone else.
_, err := emailOffByDefault(ctx, a.client, account)
return err
}
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.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) {
var ev *events.Server
var err error
if url != "" {
fmt.Println("quarry: connecting to external NATS", url)
ev, err = events.Connect(url)
} else {
ev, err = events.Start(filepath.Join(dir, "nats"))
}
if err == nil {
logEventWindow(ev)
}
return ev, err
}
// logEventWindow says once at boot how much of the event stream
// survived the restart: a store whose own unfolded events fell below the
// first sequence rebuilds in full, so this line is the margin.
func logEventWindow(ev *events.Server) {
w, err := ev.Window()
if err != nil {
log.Printf("events: stream window: %v", err)
return
}
age := "none"
if !w.Oldest.IsZero() {
age = time.Since(w.Oldest).Round(time.Second).String()
}
log.Printf("events: stream window: first %d, last %d, %d events, %d bytes, oldest %s old, kept %s",
w.FirstSeq, w.LastSeq, w.Msgs, w.Bytes, age, w.MaxAge)
}
// 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, stamp web.Stamp, evlog web.EventLog, tracker string, eng web.EngineViews, cilogs web.LogStore, clientAddr func(*http.Request) string, extra ...web.Option) (http.Handler, error) {
r := ops.NewRegistry()
if err := api.RegisterAll(r, deps); err != nil {
return nil, err
}
// One witness for every command, because there is one Invoke. It
// writes a receipt when the forge's operator changes a repository
// somebody else holds, and nothing otherwise.
r.Observe(deps.WitnessOperator())
// The router refines each op's Auth floor with the letters that
// op's surface needs.
//
// canRead comes with it so a refusal can tell "you may not do this
// here" from "there is nothing here for you": the same letters the
// sync gate and the canonical redirect ask, so the three cannot
// disagree about which repositories are hidden.
apiRouter, err := api.RouterWithCaps(r, deps.Version, capsFor, deps.CanRead, verify)
if err != nil {
return nil, err
}
mux := http.NewServeMux()
// The API's stream cap tells clients apart by the same trusted-proxy
// rule the web uses, never by a header the client wrote.
mux.Handle("/api/", api.WithClientAddr(clientAddr, apiRouter))
// notify's delivery stream lands here (RFC 0014 Phase 1). Not an
// op: it authenticates by signature rather than by account, and
// the account it writes as is the one the notification named.
if deps.Replies != nil {
mux.Handle("POST /api/v1/notify/events", api.NewReplyReceiver(deps, r, deps.Replies, capsFor))
}
// The no-script upload (RFC 0016 §Writing) for an API client with
// Basic credentials: a multipart form quarry carries to the bucket
// itself. Not an op — an op's input is a document, and this one is
// a stream — so it is mounted beside the router the way the reply
// receiver is. The page's own form posts to the web route, which
// runs the same function under the session cookie.
api.MountUploadReceiver(mux, deps, r, verify, capsFor, apiRouter)
// Registered HERE, not in the web mux: "/api/" above claims the
// whole subtree, and a browser's error report is not an op — it is
// anonymous, rate-limited and best-effort, and it must stay
// reachable by a reader who has never signed in.
mux.HandleFunc("POST /api/client-error", web.ClientErrorHandler("", extra...))
// 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 deps.Attachments != nil {
// RFC 0016 §Writing: the page's own upload form, carried to the
// bucket by the api layer under the web layer's session.
opts = append(opts, web.WithUploadForm(func(ctx context.Context, repo, target string, req *http.Request) (any, error) {
return api.UploadForm(ctx, deps, r, repo, target, req)
}))
}
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 deps.Accounts != nil {
// RFC 0005's account.caps, told to whoever it happened
// to: web writes caps as raw fossil strings, the
// account seam speaks role names, and accessAdapter is
// where that crossing happens — nowhere else.
opts = append(opts, web.WithAccess(accessAdapter{&deps}))
}
}
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 stamp != nil {
opts = append(opts, web.WithStamp(stamp))
}
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...)
// The page tree carries a Content-Security-Policy and a per-response
// nonce; the API router and the asset routes do not, having no
// scripts to authorise.
// A page under a repository with a bucket bound (RFC 0016) may also
// show bytes from and send bytes to that bucket: img-src, media-src
// and connect-src widen to its origins, since CSP follows the
// redirect every stub attachment is served by.
mux.Handle("/", web.WithCSP(web.HandlerWith(r, notifier, chatLive, opts, web.Verify(verify)),
web.WithRepoOrigins(deps.BucketOrigins), web.WithHandlerNamespaces(opts...)))
// Compression wraps EVERYTHING, including the API router: a
// timeline page is 88% smaller encoded and a highlighted source
// page 92%. It declines server-sent events by content type, which
// matters more than the saving — a buffered stream is a live update
// that never arrives.
return web.Compress(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, Prev: w.PreviousSecret()})
}
return out, nil
}
// Deliveries on the SENDER is repo-scoped, for the API; the store's
// own method below is the page's, which already knows the endpoint
// belongs to the repository it just listed.
func (h hookSender) Deliveries(repo string, sub int64, limit int) ([]api.HookDelivery, error) {
w, ok := h.db.WebhookByID(sub)
if !ok || w.Repo != repo {
return nil, fmt.Errorf("no endpoint %d on %s: %w", sub, repo, ops.ErrNotFound)
}
rows, err := h.db.Deliveries(sub, limit)
if err != nil {
return nil, err
}
out := make([]api.HookDelivery, 0, len(rows))
for _, d := range rows {
out = append(out, api.HookDelivery{ID: d.ID, Sub: d.Sub, Kind: d.Kind, At: d.At,
Status: d.Status, Ms: d.Ms, Err: d.Err, Attempt: d.Attempt})
}
return out, nil
}
// RotateWebhookSecret is repo-scoped for the same reason redelivery is:
// the id names a row the whole forge shares.
func (h hookStore) RotateWebhookSecret(repo string, id int64) (string, error) {
w, ok := h.db.WebhookByID(id)
if !ok || w.Repo != repo {
return "", fmt.Errorf("no endpoint %d on %s: %w", id, repo, ops.ErrNotFound)
}
return h.db.RotateWebhookSecret(id)
}
func (h hookStore) Deliveries(sub int64, limit int) ([]web.HookDelivery, error) {
rows, err := h.db.Deliveries(sub, limit)
if err != nil {
return nil, err
}
out := make([]web.HookDelivery, 0, len(rows))
for _, d := range rows {
out = append(out, web.HookDelivery{ID: d.ID, Kind: d.Kind, At: d.At,
Status: d.Status, Ms: d.Ms, Err: d.Err, Attempt: d.Attempt})
}
return out, nil
}
// hookSender is the SENDING half: a replay of one recorded attempt, and
// a ping so an endpoint can be tested before anything real depends on
// it. It needs the dispatcher, which the store half does not, so the
// two are wired separately.
type hookSender struct {
db *oltp.DB
d *webhook.Dispatcher
}
// Redeliver posts a recorded attempt again, byte for byte. Rebuilding
// the event instead would change the signature, which is the one thing
// somebody debugging a rejected delivery is trying to compare.
func (h hookSender) Redeliver(repo string, id int64) (webhook.Outcome, error) {
rec, ok := h.db.DeliveryByID(id)
if !ok {
return webhook.Outcome{}, fmt.Errorf("no delivery %d: %w", id, ops.ErrNotFound)
}
// The id names a row in a table the whole forge shares. Without
// this an administrator of one repository could replay another's
// deliveries — to another's endpoint — by guessing a number. Not
// found rather than forbidden, so the guess confirms nothing.
if rec.Repo != repo {
return webhook.Outcome{}, fmt.Errorf("no delivery %d on %s: %w", id, repo, ops.ErrNotFound)
}
sub, ok := h.sub(rec.Sub)
if !ok {
return webhook.Outcome{}, fmt.Errorf("the endpoint delivery %d was sent to is gone: %w", id, ops.ErrNotFound)
}
return h.d.Replay(context.Background(), sub,
webhook.Event{ID: rec.Event, Kind: rec.Kind, Repo: rec.Repo},
[]byte(rec.Body)), nil
}
// Ping sends a signed event of its own so an owner can prove an
// endpoint is reachable and its signature check works, without waiting
// for something real to happen.
func (h hookSender) Ping(repo string, subID int64) (webhook.Outcome, error) {
w, ok := h.db.WebhookByID(subID)
if !ok || w.Repo != repo {
return webhook.Outcome{}, fmt.Errorf("no endpoint %d on %s: %w", subID, repo, ops.ErrNotFound)
}
sub, ok := h.sub(subID)
if !ok {
return webhook.Outcome{}, fmt.Errorf("no endpoint %d: %w", subID, ops.ErrNotFound)
}
e := webhook.Event{
ID: fmt.Sprintf("ping-%d", time.Now().UnixNano()),
Kind: "ping",
Repo: w.Repo,
Summary: "a test delivery from quarry; nothing happened",
At: time.Now().UTC().Format(time.RFC3339),
}
body, err := json.Marshal(e)
if err != nil {
return webhook.Outcome{}, err
}
return h.d.Replay(context.Background(), sub, e, body), nil
}
func (h hookSender) sub(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, Prev: w.PreviousSecret()}, true
}
// hookLog is the delivery log's seam: webhook owns what an attempt IS,
// oltp owns keeping it, and this is the only place they meet.
type hookLog struct{ db *oltp.DB }
func (h hookLog) RecordDelivery(a webhook.Attempt) {
h.db.RecordDelivery(oltp.Delivery{Sub: a.Sub, Repo: a.Repo, Event: a.Event,
Kind: a.Kind, Status: a.Status, Ms: a.Ms, Err: a.Err,
Attempt: a.Attempt, Body: a.Body})
}
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
}
// settingsLog adapts the OLTP database to the web layer's SettingsLog
// seam. The web layer names what a row shows; the store owns where it
// is kept and for how long.
type settingsLog struct{ db *oltp.DB }
func (l settingsLog) Changes(repo string, limit int) ([]web.SettingsChange, error) {
rows, err := l.db.SettingsLog(repo, limit)
if err != nil {
return nil, err
}
out := make([]web.SettingsChange, 0, len(rows))
for _, r := range rows {
// A row whose stamp will not parse still has a field, an actor
// and both values, which is most of what the reader came for.
// Dropping it would answer "who changed this" with silence over
// a formatting detail.
at, _ := time.Parse(time.RFC3339, r.At)
out = append(out, web.SettingsChange{At: at, Actor: r.Actor,
Field: r.Field, Was: r.Was, Now: r.Now})
}
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
}
// LoginFrom is Login that records the device (web.SessionsFrom).
func (s sessionPlane) LoginFrom(user, password, ip, agent string) (string, time.Time, bool) {
token, expires, ok := s.db.NewSessionFrom(user, password, ip, agent)
if ok {
s.provisionOnLogin(user)
}
return token, expires, ok
}
func (s sessionPlane) provisionOnLogin(account string) {
if s.provision == nil {
return
}
p := *s.provision
if p == nil {
return
}
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)
}
}()
}
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
}
// The settings page reaches optional seams by type-asserting this plane,
// and a failed assertion is SILENT: the block does not render and its
// POST does nothing. CITagsAdmin was missing for exactly that reason and
// the verdict-tags setting sat unreachable in production. These make a
// missing method a compile error instead of a quiet gap.
//
// Every interface web.go asserts this plane against is listed, not only
// the one that was found missing: a sweep of all six showed the others
// satisfied today, and listing them keeps it that way.
var (
_ web.CapsAdmin = capsPlane{}
_ web.CapsGuard = capsPlane{}
_ web.CapsWhatIf = capsPlane{}
_ web.Reach = capsPlane{}
_ web.CITagsAdmin = capsPlane{}
_ web.CIDeclineAdmin = capsPlane{}
_ web.CapsSources = capsPlane{}
)
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,
Team: r.Team, TeamCaps: r.TeamCaps})
}
return out, nil
}
func (c capsPlane) CategoryCaps(repo, category string) string {
return c.db.CategoryCaps(repo, category)
}
func (c capsPlane) SetCaps(actor, account, repo, caps string) (string, error) {
return c.db.SetCapsGuarded(actor, account, repo, caps)
}
func (c capsPlane) GuardCaps(actor, account, repo, caps string) (string, error) {
return c.db.GuardCaps(actor, account, repo, caps)
}
// rosterPlane is api.CapsWriter: the stored capability string on a
// repository, read, and written under the guard.
type rosterPlane struct{ db *oltp.DB }
func (p rosterPlane) CapsOf(account, repo string) (string, error) {
return p.db.CapsOf(account, repo)
}
func (p rosterPlane) SetCaps(actor, account, repo, caps string) (string, error) {
return p.db.SetCapsGuarded(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)
}
// SourcesWith names where each letter an account would hold comes from,
// so the person editor can say "from developer category" instead of one
// fixed phrase for every inherited ability (web.CapsSources).
func (c capsPlane) SourcesWith(account, repo, own string) map[rune]string {
return c.db.SourcesWith(account, repo, own)
}
// CITagsMode and SetCITagsMode are the verdict-tags setting's seam
// (web.CITagsAdmin). The forge has READ this setting all along — the
// registrar and the verdict writer both consult it — but the settings
// page reaches it through this plane, and without these two methods the
// page's type assertion failed silently: the block never rendered and
// its POST did nothing, so the setting could be honoured but not changed.
func (c capsPlane) CITagsMode(repo string) string { return c.db.CITagsMode(repo) }
func (c capsPlane) SetCITagsMode(repo, mode string) error { return c.db.SetCITagsMode(repo, mode) }
// CIDeclined and SetCIDeclined carry an admin's answer to the CI tab's
// offer to help set CI up (web.CIDeclineAdmin). On capsPlane because
// that is the plane the repository settings page already holds, and the
// answer is a repository setting like any other.
func (c capsPlane) CIDeclined(repo string) (string, string) { return c.db.CIDeclined(repo) }
func (c capsPlane) SetCIDeclined(repo, who string) error { return c.db.SetCIDeclined(repo, who) }
func (c capsPlane) EffectiveWith(account, repo, own string) web.CapSet {
return c.db.EffectiveCapsWith(account, repo, own)
}
// accessAdapter binds web.AccessNotifier to api.Deps.AccessChanged: the
// one place a caps write, crossing from the web package's own vocabulary
// into the account seam's, becomes the mail and the inbox entry RFC
// 0005's account.caps describes. before/after arrive already as role
// names (web.accessRole did the caps-to-role reading) — this only
// crosses the package boundary, nothing more.
type accessAdapter struct{ deps *api.Deps }
func (a accessAdapter) AccessChanged(actor, account, repo, before, after string) {
a.deps.AccessChanged(context.Background(), actor, account, repo, before, after)
}
// 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, took time.Duration, changed bool, err error) (int, error) {
return m.db.RecordMirrorRun(repo, time.Now(), took, changed, err)
}
func (m mirrorStore) MirrorDiverged(repo, checkin string, at time.Time) error {
return m.db.MirrorDiverged(repo, checkin, at)
}
func (m mirrorStore) MirrorPulls(repo string, n int) ([]api.MirrorPull, error) {
rows, err := m.db.MirrorPulls(repo, n)
if err != nil {
return nil, err
}
out := make([]api.MirrorPull, 0, len(rows))
for _, r := range rows {
out = append(out, api.MirrorPull{At: r.At, Ms: r.Ms, Changed: r.Changed, Err: r.Err})
}
return out, nil
}
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,
DivergedAt: r.DivergedAt, DivergedLast: r.DivergedLast,
}
}
// 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, jobs *jobGate) {
for range time.Tick(tick) {
mirrorRound(deps, odb, jobs, time.Now())
}
}
// mirrorRound pulls every mirror due at now.
func mirrorRound(deps *api.Deps, odb *oltp.DB, jobs *jobGate, now time.Time) {
rows, err := odb.Mirrors()
if err != nil {
log.Printf("mirror: listing: %v", err)
errlog.Default.ReportIn("mirror", "listing", "", "",
fault.Wrap(err, "the mirror list could not be read").
Cost("no mirror is pulled this round; every mirrored store falls further behind"))
return
}
for _, row := range rows {
// Checked before PullMirror, which records every attempt: a
// node that does not hold the store must leave no failed run.
if !row.Due(now) || !jobs.store(row.Repo) {
continue
}
changed, err := deps.PullMirror(toAPIMirror(row))
switch {
case errors.Is(err, ops.ErrNotFound):
// A mirror row outliving its repository is a different
// fault from a pull that failed, and it must not be
// filed as one: the upstream is fine, so "check that
// the upstream is reachable" sends an operator to look
// at the one thing that is not broken. Nothing here
// will ever succeed until somebody acts, so it says
// what to do instead of what to check.
log.Printf("mirror: %s: no such repository; the mirror row outlived it", row.Repo)
errlog.Default.ReportIn("mirror", row.URL, row.Repo, "",
fault.Wrap(err, "a mirror points at a repository this forge does not have").
Cost("nothing is pulled for it, and the row is retried for as long as it exists").
Do("remove the mirror, or register the repository it names").
With("repo", row.Repo, "upstream", row.URL))
case err != nil:
log.Printf("mirror: %s from %s: %v", row.Repo, row.URL, err)
errlog.Default.ReportIn("mirror", row.URL, row.Repo, "",
fault.Wrap(err, "a mirror could not be pulled").
Cost("this store is serving what it had at the last successful pull; it is behind, not broken").
Do("check that the upstream is reachable and that its credentials, if any, are still good").
With("repo", row.Repo, "upstream", row.URL))
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) CIRunsPage(repo string, limit, offset int) ([]api.CIRun, error) {
rows, err := c.db.CIRunsPage(repo, limit, offset)
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
}
// LatestPairFor is api.PairLister: the newest pair tested for a tip.
func (c ciRuns) LatestPairFor(repo, branchTip string) (string, error) {
return c.db.LatestPairFor(repo, branchTip)
}
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 {
// A premerge pair's run is filed under its pair, which is how every
// reader of a pair verdict asks for it (CIRunsForPair).
pair := ""
if strings.HasPrefix(checkin, "pair:") {
pair = checkin
}
return s.db.AddCIRun(oltp.CIRun{RunID: runID, Repo: repo, Checkin: checkin, Pair: pair,
Workflow: workflow, Pipeline: pipeline, Status: status})
}
func (s ciStore) Defer(repo string, c ci.Checkin) error {
return s.db.DeferCI(oltp.CIDeferred{Repo: repo, Checkin: c.Hash, Branch: c.Branch, Date: c.Date, User: c.User, Comment: c.Comment})
}
func (s ciStore) Deferred(repo string) ([]ci.Checkin, error) {
rows, err := s.db.CIDeferred(repo)
if err != nil {
return nil, err
}
out := make([]ci.Checkin, 0, len(rows))
for _, d := range rows {
out = append(out, ci.Checkin{Hash: d.Checkin, Branch: d.Branch, Date: d.Date, User: d.User, Comment: d.Comment})
}
return out, nil
}
func (s ciStore) Undefer(repo, checkin string) error { return s.db.UndeferCI(repo, checkin) }
func (s ciStore) RunPipelines(repo, checkin string) ([]string, error) {
runs, err := s.db.CIRunsFor(repo, checkin)
if err != nil {
return nil, err
}
out := make([]string, 0, len(runs))
for _, r := range runs {
// The same rule as the landing's verdictless check: a cancelled
// run never reports and a deferred one has not started.
switch r.Status {
case "cancelled", "canceled", "deferred":
continue
}
out = append(out, r.Pipeline)
}
return out, nil
}
// 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)
}
func (r ciRepo) DefaultBranch() string { return r.ix.DefaultBranch() }
func (r ciRepo) BranchOf(hash string) string {
m, err := r.ix.BranchOf([]string{hash})
if err != nil {
return ""
}
return m[hash]
}
func (r ciRepo) HasOpenLanding(branch string) bool {
_, ok, err := r.ix.OpenLandingOn(branch)
return err == nil && ok
}
// 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, attempt, afterSeq int, fromFailure bool) ([]ci.LogChunk, error) {
return v.e.Logs(runID, step, attempt, afterSeq, fromFailure)
}
func (v engineViews) StepLimits(workflow, defHash string) (map[string]int, error) {
return v.e.StepLimits(workflow, defHash)
}
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, groups []string) (ci.WorkerToken, error) {
return v.e.MintWorkerToken(label, prefixes, groups)
}
func (v engineViews) RevokeWorkerToken(id string) error { return v.e.RevokeWorkerToken(id) }
// ciVerdictAttempts bounds the passes a finished run waits for its
// verdict tag — about five minutes at the default poll. Past it the
// failure is not transient (the check-in is gone, the store refuses),
// and a run that never completes keeps its landing waiting for ever.
const ciVerdictAttempts = 30
// verdictLedger is what this process knows about verdicts between
// passes. written covers the gap before the index shows a tag this
// node wrote: the tag log is append-only, so writing it again is a
// second artifact, not a no-op. failed counts passes toward the bound.
// Both forget a run once it is complete. Being in memory, it has one
// known gap, accepted: a restart after a write whose index update
// failed, before the index catches up, can write the tag twice.
type verdictLedger struct {
mu sync.Mutex
written map[string]bool
failed map[string]int
}
var ciLedger = &verdictLedger{written: map[string]bool{}, failed: map[string]int{}}
func (l *verdictLedger) wrote(run string) {
l.mu.Lock()
defer l.mu.Unlock()
l.written[run] = true
}
func (l *verdictLedger) hasWritten(run string) bool {
l.mu.Lock()
defer l.mu.Unlock()
return l.written[run]
}
func (l *verdictLedger) fail(run string) int {
l.mu.Lock()
defer l.mu.Unlock()
l.failed[run]++
return l.failed[run]
}
func (l *verdictLedger) forget(run string) {
l.mu.Lock()
defer l.mu.Unlock()
delete(l.written, run)
delete(l.failed, run)
}
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", "", "",
fault.Wrap(err, "CI could not be polled for verdicts").
Cost("landings wait on checks whose results are not being collected"))
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
}
// Retention (RFC 0009 §logs): the hot lane expires; the record
// beside the verdict does not. Drained before the verdict so a
// verdict retried over several passes cannot outlive its logs.
if err := drainLogs(engine, dataDir, r.Repo, r.RunID, run.Steps); err != nil {
log.Printf("ci: drain %s: %v", r.RunID, err)
errlog.Default.ReportIn("ci", "drain", r.Repo, "",
fault.Wrap(err, "a CI run's output could not be collected").
Cost("the run's log is missing from the record kept beside its verdict").
With("repo", r.Repo, "run", r.RunID))
}
// The verdict is written BEFORE the run is marked complete. A
// completed row leaves the poller's work list for good, so a
// verdict that failed after it would never be retried and the
// landing would wait on a result that never arrives. Written
// first, a failure leaves the row unfinished and the next pass
// tries again — up to a bound, because a failure that outlasts
// it is not going to clear, and a run that never completes
// strands its landing just as surely as a lost verdict.
if err := writeVerdict(db, repos, verdicts, r, run); err != nil {
// A lease refusal is a wait, not a fault, and never counts
// toward the bound: another node holding the lease writes
// the verdict itself (this node completes the run once it
// sees it), and a coordinator that is down comes back.
// Counting either would abandon a verdict that was only
// delayed — the lost result this ordering exists to prevent.
if errors.Is(err, lease.ErrNotHolder) || errors.Is(err, lease.ErrUnavailable) {
continue
}
n := ciLedger.fail(r.RunID)
if n < ciVerdictAttempts {
errlog.Default.ReportIn("ci", "verdict", r.Repo, "",
fault.Wrap(err, "a CI verdict could not be recorded").
Cost("the run finished but its result is not on the check-in yet; the next poll retries it").
With("repo", r.Repo, "run", r.RunID))
continue
}
log.Printf("ci: %s %.12s verdict abandoned after %d passes (run %s): %v",
r.Repo, r.Checkin, n, r.RunID, err)
errlog.Default.ReportIn("ci", "verdict-abandoned", r.Repo, "",
fault.Wrap(err, "a CI verdict was given up on").
Cost("the run is complete but its check-in carries no verdict, so a landing gated on it will not pass on this run").
With("repo", r.Repo, "run", r.RunID, "passes", fmt.Sprint(n)))
}
if err := db.CompleteCIRun(r.RunID, run.Status, run.CompletedAt); err != nil {
log.Printf("ci: status %s: %v", r.RunID, err)
continue
}
ciLedger.forget(r.RunID)
raise(evt.CICompleted)
}
}
// writeVerdict records a terminal run's result as a tag on its
// check-in, or decides no tag is owed. It errors only when a tag was
// owed and could not be written — the one case the caller must retry.
func writeVerdict(db *oltp.DB, repos *api.Repos, verdicts bool, r oltp.CIRun, run ci.Run) error {
// RFC 0009 §verdict-tags: the repository chooses; '' follows
// the fleet default the flag sets.
switch db.CITagsMode(r.Repo) {
case "off":
return nil
case "on":
default:
if !verdicts {
return nil
}
}
repo, ok := repos.Get(r.Repo)
if !ok {
return nil
}
// A cancelled run earns no verdict. quarry supersedes the runs
// still going against an older check-in whenever a newer one
// lands on the same branch, so "cancelled" is usually quarry's
// own doing — and recording it as a failure marks a check-in
// that was never finished being tested as one that failed, in
// a tag that clones with the repository.
//
// Found by counting: of the runs still on record here, 36 were
// cancelled against 10 that actually failed.
if run.Cancelled() {
log.Printf("ci: %s %.12s cancelled, no verdict (run %s)", r.Repo, r.Checkin, r.RunID)
return nil
}
// A landing's merge is not a check-in, so there is nothing to tag:
// its verdict is the run itself, which the landing reads, and the
// carry-forward tags it onto the merge check-in once an accept writes
// that merge from the same pair (quarry#290).
if r.Pair != "" {
return nil
}
verdict := "ci-fail"
if run.Passed() {
verdict = "ci-pass"
}
// The tag log is append-only: a verdict already on the check-in —
// written by this node on an earlier pass, or by the lease holder —
// must not be written again.
if ciLedger.hasWritten(r.RunID) || verdictRecorded(repo, r) {
return nil
}
// Written through the lease holder, like every other artifact
// quarry mints: a verdict is a write, and two nodes must not
// both record one.
err := repo.WriteTag(verdict, r.Checkin, r.RunID)
if errors.Is(err, api.ErrTagNotIndexed) {
log.Printf("ci: verdict %s on %.12s written, index lags: %v", verdict, r.Checkin, err)
err = nil
}
if err != nil {
log.Printf("ci: verdict %s on %.12s: %v", verdict, r.Checkin, err)
return err
}
ciLedger.wrote(r.RunID)
log.Printf("ci: %s %.12s %s (run %s)", r.Repo, r.Checkin, verdict, r.RunID)
return nil
}
// verdictRecorded reports whether the check-in's tag log already holds a
// verdict for this run, standing or cancelled: a verdict someone
// cancelled was cancelled on purpose, and writing it again undoes that.
// An index that cannot answer says no, and the write goes ahead.
func verdictRecorded(repo *api.Repo, r oltp.CIRun) bool {
if repo.Index == nil {
return false
}
events, err := repo.Index.TagHistory(r.Checkin)
if err != nil {
return false
}
for _, e := range events {
if (e.Name == "ci-pass" || e.Name == "ci-fail") && e.Value == r.RunID {
return true
}
}
return false
}
// 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, TxnID: e.TxnID,
})
}
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,
TxnID: e.TxnID, 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,
TxnID: e.TxnID, 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 "I did not answer that" onto the ops layer's own
// word for it, so internal/api can degrade around an unanswered
// preferences call 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
}
// notify names each message's workflow itself; the map is reached
// for only by rows that carry none, so a current companion's page
// costs no extra call at all.
reasons := i.names.Identify(ctx, msgs)
out := make([]api.InboxMessage, 0, len(msgs))
for _, m := range msgs {
out = append(out, api.InboxMessage{
ID: m.ID, Workflow: reasons[m.ID], Subject: m.Subject, Body: m.Content,
Link: m.Link(), At: m.CreatedAt, Read: m.Read, Seen: m.Seen,
TxnID: m.TransactionID,
})
}
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) {
prefs, err := i.client.Preferences(ctx, account)
if err != nil {
// Unreadable is UNKNOWN, not an error and emphatically not a
// default: a settings page that fails outright shows nobody
// their mutes, and one that renders quarry's defaults tells
// them a channel is on when notify may hold it off.
log.Printf("notify: preferences for %s unreadable: %v", account, err)
return nil, false, nil
}
// Only the per-workflow section 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.
return prefs.Workflows, true, nil
}
// DigestWindows reports the window every batching workflow is actually
// using for this person, whether they chose it or inherited it.
func (i inboxCounter) DigestWindows(ctx context.Context, account string) (map[string]api.DigestWindow, error) {
prefs, err := i.client.Preferences(ctx, account)
if err != nil {
return nil, err
}
if len(prefs.Digests) == 0 {
return nil, nil
}
out := make(map[string]api.DigestWindow, len(prefs.Digests))
for wf, w := range prefs.Digests {
out[wf] = api.DigestWindow{Amount: w.Amount, Unit: w.Unit, Chosen: w.Chosen()}
}
return out, nil
}
func (i inboxCounter) SetPreferences(ctx context.Context, account string, prefs map[string]map[string]bool,
digests map[string]*api.DigestWindow) error {
var windows map[string]*notify.DigestWindow
if len(digests) > 0 {
windows = make(map[string]*notify.DigestWindow, len(digests))
for wf, w := range digests {
if w == nil {
// An explicit null is how notify is told to clear one;
// omitting the key would leave it exactly as it was.
windows[wf] = nil
continue
}
windows[wf] = ¬ify.DigestWindow{Amount: w.Amount, Unit: w.Unit}
}
}
return remote(i.client.SetPreferences(ctx, account, prefs, windows))
}
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)
}
// chatBridgeStore adapts the OLTP bridge rows to the ops layer's view.
// The credential crosses in only: there is no direction on this seam
// that could carry it back out.
type chatBridgeStore struct{ db *oltp.DB }
func (c chatBridgeStore) SetChatBridge(repo, url, user, secret string) error {
return c.db.SetChatBridge(repo, url, user, secret)
}
func (c chatBridgeStore) DetachChatBridge(repo string) error {
return c.db.DetachChatBridge(repo)
}
func (c chatBridgeStore) ChatBridgeFor(repo string) (api.ChatBridge, bool, error) {
row, ok, err := c.db.ChatBridgeFor(repo)
if err != nil || !ok {
return api.ChatBridge{}, ok, err
}
return toAPIChatBridge(row), true, nil
}
func (c chatBridgeStore) ChatBridges() ([]api.ChatBridge, error) {
rows, err := c.db.ChatBridges()
if err != nil {
return nil, err
}
out := make([]api.ChatBridge, 0, len(rows))
for _, r := range rows {
out = append(out, toAPIChatBridge(r))
}
return out, nil
}
func toAPIChatBridge(r oltp.ChatBridge) api.ChatBridge {
return api.ChatBridge{Repo: r.Repo, URL: r.URL, User: r.User,
LastMsgID: r.LastMsgID, At: r.At, LastErr: r.LastErr}
}
// bridgeRunner keeps the running bridges in step with the configured
// ones. It reconciles on the mirror tick rather than starting them
// once at boot: a bridge added, re-pointed or detached through the ops
// surface has to take effect without a restart, and "reconcile a small
// list" is a cheaper way to say that than a notification channel from
// the op back into the composition root.
type bridgeRunner struct {
room *chat.Room
odb *oltp.DB
tick time.Duration
// code maps a repository name to its channel key; a repository the
// server does not serve has none, and no bridge is started for it.
code func(repo string) (string, bool)
// holds says whether this node holds a repository's lease; a bridge
// runs only on the holder. Nil means every repository.
holds func(repo string) bool
// allowPrivate carries --allow-private-bridges to every bridge:
// the guard is off by default, so an operator bridging a fossil on
// the same host opts in once for the whole node.
allowPrivate bool
// running is the live set, keyed by repository, remembering the
// URL and login each was started with so a reconfiguration
// restarts rather than being ignored.
running map[string]*runningBridge
}
type runningBridge struct {
url, user string
stop context.CancelFunc
}
func (br *bridgeRunner) run(ctx context.Context) {
br.running = map[string]*runningBridge{}
defer func() {
for _, r := range br.running {
r.stop()
}
}()
tick := br.tick
if tick <= 0 {
tick = time.Minute
}
t := time.NewTicker(tick)
defer t.Stop()
br.reconcile(ctx)
for {
select {
case <-ctx.Done():
return
case <-t.C:
br.reconcile(ctx)
}
}
}
func (br *bridgeRunner) reconcile(ctx context.Context) {
rows, err := br.odb.ChatBridges()
if err != nil {
log.Printf("chat bridge: listing: %v", err)
return
}
want := make(map[string]bool, len(rows))
elsewhere := map[string]bool{}
for _, row := range rows {
if br.holds != nil && !br.holds(row.Repo) {
elsewhere[row.Repo] = true // stopped below if it ran here
continue
}
want[row.Repo] = true
if cur, ok := br.running[row.Repo]; ok {
if cur.url == row.URL && cur.user == row.User {
continue
}
// Re-pointed or re-credentialed: the old poller is reading
// somewhere the operator no longer means.
cur.stop()
delete(br.running, row.Repo)
}
code, ok := br.code(row.Repo)
if !ok {
continue // not a repository this node serves
}
secret, ok := br.odb.ChatBridgeSecret(row.Repo)
if !ok {
continue
}
b := chat.NewBridge(br.room, br.odb, chat.BridgeConfig{
Repo: row.Repo, Code: code, URL: row.URL, User: row.User,
Secret: secret, LastMsgID: row.LastMsgID, AllowPrivate: br.allowPrivate,
})
// RFC 0011 §9: with quarry's own robot announcing check-ins,
// the upstream's timeline user would announce them again.
b.RobotOn = func(repo string) bool { return br.odb.ChatRobot(repo) != "" }
bctx, stop := context.WithCancel(ctx)
br.running[row.Repo] = &runningBridge{url: row.URL, user: row.User, stop: stop}
go b.Run(bctx)
log.Printf("chat bridge: %s <-> %s as %s", row.Repo, row.URL, row.User)
}
for repo, r := range br.running {
if !want[repo] {
r.stop()
delete(br.running, repo)
if elsewhere[repo] {
log.Printf("chat bridge: %s moved to the node holding its lease", repo)
} else {
log.Printf("chat bridge: %s detached", repo)
}
}
}
}
// cloneBaseFor is the base URL a CI worker clones from: the override
// when one is given, else this server's own address.
//
// A listen address may name no host — ":8950" means every interface —
// and pasting that into a URL produces "http://:8950/r/x", which zit
// rejects as invalid, failing every pipeline's clone in under a
// second. A server listening everywhere is reachable at loopback from
// the box it runs on, which is where the worker lives unless someone
// says otherwise, so that is the host we name.
func cloneBaseFor(addr, override string) string {
if override != "" {
return strings.TrimSuffix(override, "/") + "/r/"
}
if h, port, err := net.SplitHostPort(addr); err == nil && unspecifiedHost(h) {
addr = net.JoinHostPort("127.0.0.1", port)
}
return "http://" + addr + "/r/"
}
// unspecifiedHost reports whether a listen address's host half names
// every interface rather than one. All three spellings are legitimate
// and none of them is dialable as written: "" and "0.0.0.0" and "::"
// mean "listen everywhere", not "reach me here".
func unspecifiedHost(h string) bool {
if h == "" {
return true
}
ip := net.ParseIP(h)
return ip != nil && ip.IsUnspecified()
}
// depFlaps remembers how a watched dependency has behaved, so the
// outage row can say "this is the thirteenth time, it was last up at
// 12:04" instead of that history being a second row in the error list.
//
// Process-local by design: it describes what THIS server has watched,
// and a restart honestly knows nothing about the outage before it.
type depFlapper struct {
mu sync.Mutex
at map[string]*depFlap
}
type depFlap struct {
outages int
since string // when the current outage began
recovered string // when it was last seen up, empty if never
down bool
}
var depFlaps = &depFlapper{at: map[string]*depFlap{}}
// seen records one health beat and returns the dependency's history.
// A beat that repeats a state it already knows changes nothing: an
// outage that lasts an hour is one outage, not one per beat.
func (f *depFlapper) seen(name string, up bool) depFlap {
f.mu.Lock()
defer f.mu.Unlock()
d, ok := f.at[name]
if !ok {
d = &depFlap{}
f.at[name] = d
}
now := time.Now().UTC().Format(time.RFC3339)
switch {
case up && d.down:
d.down, d.recovered = false, now
case up:
d.recovered = now
case !up && !d.down:
d.down, d.outages, d.since = true, d.outages+1, now
}
return *d
}
// registerWorkers is how many repositories register at once at
// startup. Four: enough that one slow store (pkgsrc's submanifest
// probe alone cost ~33s serially) does not hold the others, few enough
// that two dozen index opens and zit probes do not contend for the
// disk they all sit on.
const registerWorkers = 4
// eachConcurrently runs fn over items with at most n in flight and
// returns when every one has finished. Order of completion is not the
// order of items, which is the point.
func eachConcurrently[T any](items []T, n int, fn func(T)) {
if n < 1 {
n = 1
}
var wg sync.WaitGroup
slots := make(chan struct{}, n)
for _, it := range items {
wg.Add(1)
slots <- struct{}{}
go func(it T) {
defer wg.Done()
defer func() { <-slots }()
fn(it)
}(it)
}
wg.Wait()
}