// quarry is a multi-repository forge for zit. See docs/rfcs/.
package main
import (
"context"
"craftdesign.group/quarry/internal/errlog"
"encoding/json"
"errors"
"flag"
"fmt"
"log"
"net/http"
"os"
"path/filepath"
"regexp"
"sort"
"strings"
"sync"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/archive"
"craftdesign.group/quarry/internal/chat"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/notify"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/web"
"craftdesign.group/quarry/internal/webhook"
"craftdesign.group/quarry/internal/work"
"craftdesign.group/quarry/internal/xfer"
"craftdesign.group/quarry/internal/zitcli"
)
// version is stamped at build time; every page footer and the API report it
// (0001 Principle 1).
var version = "0.0.0-dev"
func main() {
if len(os.Args) > 1 {
switch os.Args[1] {
case "version":
fmt.Println("quarry", version)
return
case "spec":
if err := printSpec(os.Stdout); err != nil {
fmt.Fprintln(os.Stderr, "quarry:", err)
os.Exit(1)
}
return
case "serve":
if err := serveCmd(os.Args[2:]); err != nil {
fmt.Fprintln(os.Stderr, "quarry:", err)
os.Exit(1)
}
return
case "reindex":
if err := reindexCmd(os.Args[2:]); err != nil {
fmt.Fprintln(os.Stderr, "quarry:", err)
os.Exit(1)
}
return
case "docs":
r := ops.NewRegistry()
if err := api.RegisterAll(r, api.Deps{Version: version}); err != nil {
fmt.Fprintln(os.Stderr, "quarry:", err)
os.Exit(1)
}
os.Stdout.Write(ops.Docs(r, "quarry"))
return
case "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|account|webhook|docs")
os.Exit(2)
}
// userCmd manages sync users (0001 §7 v1 auth): `quarry user add`
// derives the RFC §6.3 shared secret from the store's project code and
// stores it in forge OLTP — plaintext is never persisted. A repo with
// at least one sync user requires authenticated pushes.
func userCmd(args []string) error {
if len(args) < 1 || args[0] != "add" {
return fmt.Errorf("usage: quarry user add --repo name=path --user U [--password P] [--data DIR]")
}
fs := flag.NewFlagSet("user add", flag.ContinueOnError)
repos := repoFlags{}
fs.Var(repos, "repo", "repository as name=path")
user := fs.String("user", "", "sync login name")
password := fs.String("password", "", "password (omit to read from QUARRY_PASSWORD)")
data := fs.String("data", "", "derived-data directory (default: .quarry beside the repo)")
if err := fs.Parse(args[1:]); err != nil {
return err
}
if len(repos) != 1 || *user == "" {
return fmt.Errorf("user add wants exactly one --repo and a --user")
}
pw := *password
if pw == "" {
pw = os.Getenv("QUARRY_PASSWORD")
}
if pw == "" {
return fmt.Errorf("no password: use --password or QUARRY_PASSWORD")
}
var name, path string
for n, p := range repos {
name, path = n, p
}
dir := *data
if dir == "" {
dir = filepath.Join(path, ".quarry")
}
if err := os.MkdirAll(dir, 0o755); err != nil {
return err
}
code, err := xfer.ProjectCode(path)
if err != nil {
return fmt.Errorf("project code of %s: %w", name, err)
}
db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
if err != nil {
return err
}
defer db.Close()
if err := db.SetSyncUser(name, *user, xfer.SharedSecret(code, *user, pw)); err != nil {
return err
}
fmt.Printf("sync user %q added to %s; pushes now require authentication\n", *user, name)
return nil
}
// reindexCmd is 0001 §5's full rebuild as a verb: for each --repo, drop
// the derived tables and cursor, then replay the store. Run it with the
// same --repo/--data flags as serve, while serve is stopped (the
// embedded NATS store directory admits one process).
func reindexCmd(args []string) error {
fs := flag.NewFlagSet("reindex", flag.ContinueOnError)
repos := repoFlags{}
fs.Var(repos, "repo", "repository as name=path (repeatable; bare path uses its basename)")
data := fs.String("data", "", "derived-data directory (default: .quarry beside a SINGLE repo; required for a fleet)")
natsURL := fs.String("nats", "", "external NATS URL (multi-node); reindex then publishes a cluster-wide reset that every node's fold absorbs")
if err := fs.Parse(args); err != nil {
return err
}
if len(repos) == 0 {
return fmt.Errorf("reindex wants at least one --repo")
}
dir, err := dataDir(*data, repos)
if err != nil {
return err
}
// On external NATS the reset republishes the census to the shared
// stream, so every node re-folds (idempotent) — reindex is a
// cluster-wide operation there, not a local-only one. Embedded is the
// offline single-node rebuild (serve must be stopped: one process per
// store dir).
ev, err := startEvents(*natsURL, dir)
if err != nil {
return fmt.Errorf("%w (is serve still running against this data dir?)", err)
}
defer ev.Close()
for name, path := range repos {
if err := os.MkdirAll(filepath.Join(dir, name), 0o755); err != nil {
return err
}
ix, err := index.Open(filepath.Join(dir, name, "index.db"))
if err != nil {
return err
}
n, err := ingest.NewPipeline(zitcli.New(path), ev, ix, name, "reindex").Reindex()
ix.Close()
if err != nil {
return fmt.Errorf("reindex %s: %w", name, err)
}
fmt.Printf("reindexed %s: %d artifacts replayed\n", name, n)
}
return nil
}
// repoFlags collects repeatable --repo name=path values (bare paths take
// their directory basename as the repository name).
type repoFlags map[string]string
func (r repoFlags) String() string { return fmt.Sprint(map[string]string(r)) }
func (r repoFlags) Set(v string) error {
name, path, found := strings.Cut(v, "=")
if !found {
path = v
name = filepath.Base(strings.TrimSuffix(v, "/"))
}
if name == "" || path == "" {
return fmt.Errorf("--repo wants name=path or path, got %q", v)
}
if !repoName.MatchString(name) {
return fmt.Errorf("repository name %q: use letters, digits, dot, dash, underscore", name)
}
if _, dup := r[name]; dup {
return fmt.Errorf("repository %q registered twice", name)
}
r[name] = path
return nil
}
var repoName = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$`)
// serveCmd runs the HTTP surface: the op-generated API plus the HTML
// pages, both consumers of one registry. Each --repo gets its own index
// and watcher pipeline; one embedded NATS carries every store's events
// under zit.evt.<name>.* (0001 §4: a watcher per registered store).
// dataDir resolves the derived-data directory. With one repository the
// default is .quarry beside it — deterministic and obviously derived.
// With several there IS no principled default: the old rule took
// "the first --repo", which for a Go map was a coin flip per boot, and
// on 2026-08-16 it planted indexes, NATS state and the OLTP database
// inside a corpus meant to hold only source repositories. 0001 §11
// says derived data is disposable; that only helps if you can tell
// which directory is the derived one — so a fleet must say.
func dataDir(explicit string, repos map[string]string) (string, error) {
if explicit != "" {
return explicit, nil
}
if len(repos) == 1 {
for _, p := range repos {
return filepath.Join(p, ".quarry"), nil
}
}
names := make([]string, 0, len(repos))
for n := range repos {
names = append(names, n)
}
sort.Strings(names)
return "", fmt.Errorf("serving %d repositories needs an explicit --data: "+
"there is no principled default, and the old rule silently wrote "+
"indexes, event-stream state and the account database into an "+
"arbitrary one of %v", len(repos), names)
}
func serveCmd(args []string) error {
fs := flag.NewFlagSet("serve", flag.ContinueOnError)
addr := fs.String("addr", "127.0.0.1:8080", "listen address")
repos := repoFlags{}
fs.Var(repos, "repo", "repository to serve as name=path (repeatable; bare path uses its basename)")
data := fs.String("data", "", "derived-data directory (default: .quarry beside a SINGLE repo; required for a fleet)")
interval := fs.Duration("sync-interval", 2*time.Second, "store poll interval")
leaseTTL := fs.Duration("lease-ttl", lease.DefaultTTL, "single-writer lease lifetime")
natsURL := fs.String("nats", "", "external NATS URL for multi-node (default: embedded, single node); the server must allow max_payload >= 8MiB")
notifyURL := fs.String("notify", "", "notify companion base URL for mentions and activity (default: notifications disabled)")
dagnatsURL := fs.String("dagnats", "", "dagnats control-plane URL for CI (0001 §6); empty disables CI")
ciVerdicts := fs.Bool("ci-verdicts", true, "record ci-pass/ci-fail as tag artifacts on the check-in, so CI history clones with the repository")
ciPoll := fs.Duration("ci-poll", 10*time.Second, "how often to refresh unfinished CI runs")
// How often DUE-ness is evaluated, not how often anything is pulled:
// each mirror carries its own interval. A minute is fine because the
// loop does nothing when no mirror is due.
mirrorTick := fs.Duration("mirror-tick", time.Minute, "how often to check for due mirrors")
// Bulk work is bounded by memory, not cores: each archive buffers its
// whole output. Two at a time with a short queue keeps the worst case
// near half a gigabyte instead of unbounded.
bulkLimit := fs.Int("bulk-limit", 2, "concurrent bulk operations (archive)")
// Where repositories created or cloned through the API are placed.
// Without it an API caller must supply a filesystem path, which only
// the machine's operator can meaningfully choose.
reposRoot := fs.String("repos-root", "", "directory for repositories created through the API")
bulkQueue := fs.Int("bulk-queue", 8, "bulk operations allowed to wait before refusing")
archiveCache := fs.Int64("archive-cache", 2<<30, "bytes of built archives to keep on disk (0 disables)")
// Ingest is bounded because it is the one path that streams a WHOLE
// store: `zit frames` on sqlite peaks at 507 MB and tcl is larger.
// One goroutine per repository with nothing between them means a
// cold start, or a mirror round that lands artifacts in several big
// stores at once, runs that many censuses simultaneously and the
// memory adds. Observed: zit spiking to gigabytes after pulling
// sqlite and tcl together.
ingestLimit := fs.Int("ingest-limit", 3, "stores that may ingest at once")
errorlogPath := fs.String("errorlog", "", "append server errors to this file (the in-process journal restarts with the server; this survives)")
tracker := fs.String("tracker", "", "repository whose tickets are the forge's own; enables /fleet's file-ticket buttons")
notifyKey := fs.String("notify-key", "", "notify console API key (or QUARRY_NOTIFY_KEY); notify's /v1 routes answer 401 without one")
if err := fs.Parse(args); err != nil {
return err
}
// Refuse leftovers. Go's flag package stops at the first non-flag
// argument and hands the rest back, so a malformed invocation — one
// unquoted shell variable is enough — silently drops every flag past
// that point. Doing this cost twelve --repo flags once, and quarry
// started and served the remaining eleven without a word.
if rest := fs.Args(); len(rest) > 0 {
return fmt.Errorf("unexpected argument %q: serve takes only flags "+
"(a shell variable holding flags must be word-split, e.g. \"${args[@]}\")", rest[0])
}
deps := api.Deps{Version: version, Repos: api.NewRepos(nil), Errors: errlog.Default}
var notifier web.Notifier
var chatLive web.ChatLive
var verify api.Verify
var sessions web.Sessions
var inbox web.Inbox
var engine *ci.Engine
var ciCloneBase string
var fronter *xfer.Fronter
if len(repos) > 0 {
fronter = xfer.New()
defer fronter.Close()
// Registration happens in the registrar below, which owns the
// whole lifetime; registering here too would mean two places
// that must agree about which repositories exist.
}
if len(repos) > 0 {
dir, err := dataDir(*data, repos)
if err != nil {
return err
}
if err := os.MkdirAll(dir, 0o755); err != nil {
return err
}
odb, err := oltp.Open(filepath.Join(dir, "oltp.db"))
if err != nil {
return err
}
defer odb.Close()
fronter.SetAuth(odb)
// Two credentials, on purpose (0001 §8). Accounts are the forge
// identity and carry per-repo roles; the RFC §6.3 shared secret
// stays the SYNC credential, because the protocol fixes its
// derivation and the server must be able to recompute it. Once
// any account exists it governs web/API writes; before that,
// v1's per-repo secret still does, so an existing deployment
// keeps working and a fresh local one stays open.
// Cookie sessions for the browser; Basic stays for the API.
sessions = sessionPlane{db: odb}
var codeMu sync.Mutex
codes := map[string]string{}
verify = func(repo, user, pass string, need ops.Auth) api.Decision {
if odb.HasAccounts() {
acct, ok := odb.Authenticate(user, pass)
if !ok {
return api.Denied
}
// Authenticated. From here a refusal is about capability,
// not identity, and saying so is the difference between a
// client that stops and one that retries forever.
if odb.RoleOf(acct, repo) >= requiredRole(need) {
return api.Allowed
}
return api.Insufficient
}
if !odb.Gated(repo) {
return api.Allowed
}
secret, ok := odb.Secret(repo, user)
if !ok || pass == "" {
return api.Denied
}
codeMu.Lock()
code, cached := codes[repo]
codeMu.Unlock()
if !cached {
path, ok := repos[repo]
if !ok {
return api.Denied
}
c, err := xfer.ProjectCode(path)
if err != nil {
log.Printf("auth: project code of %s: %v", repo, err)
return api.Denied
}
codeMu.Lock()
codes[repo], code = c, c
codeMu.Unlock()
}
if secret == xfer.SharedSecret(code, user, pass) {
return api.Allowed
}
return api.Denied
}
ev, err := startEvents(*natsURL, dir)
if err != nil {
return err
}
defer ev.Close()
notifier = ev
// Single-writer coordination (0001 §3). node identifies this
// process; one embedded NATS means one node, but the machinery
// and its failure posture (coordinator down → writes refuse) run
// the same on one node as on N.
leases, err := lease.New(ev.JS(), *addr, *leaseTTL)
if err != nil {
return err
}
defer leases.Close()
// Per-repo chat plane (0001 §5): messages fan out over the same
// NATS the events ride, with a 7-day JetStream window.
room, err := chat.New(ev.Conn(), ev.JS())
if err != nil {
return err
}
deps.Chat = room
chatLive = &chatAdapter{room: room, codes: codes, mu: &codeMu}
// Notification triggers (0001 §5): fired best-effort to the notify
// companion; disabled when --notify is unset.
if *notifyURL != "" {
key := *notifyKey
if key == "" {
key = os.Getenv("QUARRY_NOTIFY_KEY")
}
if key == "" {
// Better to say so at startup than to discover it as 401s
// in the log after a write nobody was told about.
log.Printf("notify: no API key (--notify-key or QUARRY_NOTIFY_KEY); triggers will be refused")
}
nclient := notify.New(*notifyURL, key)
deps.Notify = ¬ifyAdapter{client: nclient}
inbox = inboxPlane{client: nclient}
fmt.Println("quarry: notifications ->", *notifyURL)
// A server error's FIRST sighting notifies; repeats only
// count in the journal — an alert channel that repeats
// itself gets muted. Signature dedup makes this cheap.
np := deps.Notify
errlog.Default.OnFirst = func(e errlog.Entry) {
_ = np.Trigger(context.Background(), api.NotifyEvent{
Workflow: "quarry-error", To: "admins",
Payload: map[string]any{
"surface": e.Surface, "path": e.Path,
"summary": e.Message,
},
TxnID: "err:" + e.First.Format(time.RFC3339Nano),
})
}
}
// CI (0001 §6): dagnats is the engine, quarry the trigger and
// display. Disabled unless --dagnats names one, so a forge with
// no engine simply has no CI rather than a broken one.
engine = ci.New(*dagnatsURL)
if engine.Enabled() {
// The clone URL handed to a pipeline: quarry's own sync
// fronting, on the address quarry was told to listen on.
ciCloneBase = "http://" + *addr + "/r/"
deps.CI = ciRuns{db: odb}
fmt.Println("quarry: CI ->", *dagnatsURL)
go func() {
for range time.Tick(*ciPoll) {
pollCI(engine, odb, deps.Repos, *ciVerdicts)
}
}()
}
// Mirroring: always available, because it costs nothing until an
// admin configures an upstream — the scheduler wakes, finds no
// rows, and sleeps. The tick is how often DUE-ness is checked,
// not how often anything is pulled; each mirror's own interval
// decides that.
// One bulk lane for the whole process, not per repository: the
// resource being protected is this machine's memory, and eleven
// per-repo lanes of two would be a bound of twenty-two.
deps.Bulk = work.NewLane("bulk", *bulkLimit, *bulkQueue)
// Archives are reproducible, so a cached one is never stale and
// the budget is purely a disk-space choice. Without it every
// reader of a popular tag pays the same seconds of compression.
if *archiveCache > 0 {
ac, err := archive.New(filepath.Join(dir, "archives"), *archiveCache)
if err != nil {
return err
}
deps.Archives = ac
}
// The queue is per-repository and generous: ingest must WAIT,
// never be shed. Dropping a round would leave an index behind
// its store until the next poll, which is a correctness surprise
// in exchange for a memory saving the limit already delivers.
ingestLane := work.NewLane("ingest", *ingestLimit, len(repos)+1)
deps.Mirrors = mirrorStore{db: odb}
go mirrorScheduler(&deps, odb, *mirrorTick)
// Webhook dispatch (Principle 6): always on, because it costs
// nothing until somebody subscribes — Raise finds no endpoints and
// returns. Delivery is durable, so an endpoint that is down during
// a restart still gets its events.
dispatcher, err := webhook.New(ev.JS(), hookSubs{db: odb})
if err != nil {
return err
}
hookCtx, stopHooks := context.WithCancel(context.Background())
defer stopHooks()
if err := dispatcher.Start(hookCtx); err != nil {
return err
}
deps.Hooks = hookPlane{d: dispatcher, at: func() string {
return time.Now().UTC().Format(time.RFC3339)
}}
// The project codes learned below (shared with the auth verifier's
// cache) key both the lease and the push guard.
fronter.SetGuard(func(repo string, proxy func() error) error {
codeMu.Lock()
code, ok := codes[repo]
codeMu.Unlock()
if !ok {
return fmt.Errorf("unknown repo %q", repo)
}
return leases.Guard(code, proxy)
})
rg := ®istrar{
dir: dir, reposRoot: *reposRoot, node: *addr, interval: *interval, ciCloneBase: ciCloneBase,
ev: ev, leases: leases, deps: &deps, fronter: fronter,
ingestLane: ingestLane, bulk: deps.Bulk, engine: engine, odb: odb,
codes: codes, codeMu: &codeMu,
}
defer rg.CloseAll()
deps.Registry = rg
deps.Accounts = accountStore{db: odb}
deps.CanRead = rg.CanRead
deps.Deletable = rg.Deletable
deps.MirrorFreshness = func() map[string]string {
rows, err := odb.Mirrors()
if err != nil {
log.Printf("mirror freshness: %v", err)
return nil
}
m := make(map[string]string, len(rows))
for _, r := range rows {
if r.Enabled && r.LastOK != "" {
m[r.Repo] = r.LastOK
}
}
return m
}
rg.refreshPrivate()
// The sync path takes the SAME visibility answer as the pages.
// A clone reaching artifacts the web refuses would not be a
// smaller hole than an open web page — it would be a bigger one,
// because it hands over everything at once.
fronter.SetReadGate(func(req *http.Request, repo string) bool {
return rg.CanRead(identityCtx(req, odb), repo)
})
deps.Private = func() map[string]bool {
set, err := odb.PrivateRepos()
if err != nil {
log.Printf("visibility: %v", err)
return nil
}
return set
}
deps.Descriptions = func() map[string]string {
rows, err := odb.Repos()
if err != nil {
log.Printf("repo descriptions: %v", err)
return nil
}
m := make(map[string]string, len(rows))
for _, r := range rows {
m[r.Name] = r.Description
}
return m
}
// Persisted registrations first, then --repo. The flag stays as
// bootstrap and override — it is how the first repository gets
// in, and how an operator recovers a registry they cannot read —
// but it is no longer the only source of truth.
if odb != nil {
stored, rerr := odb.Repos()
if rerr != nil {
return rerr
}
for _, r := range stored {
if _, dup := repos[r.Name]; dup {
continue // --repo wins: an explicit flag is the override
}
if err := rg.Register(r.Name, r.Path); err != nil {
// One unreadable stored repository must not stop the
// server: the rest are fine and the operator needs a
// running forge to fix this one.
log.Printf("registered repository %s (%s): %v", r.Name, r.Path, err)
}
}
}
for name, path := range repos {
if err := rg.Register(name, path); err != nil {
return err
}
// A flag-named repository still belongs in the registry, or
// it cannot be described, taken private, or managed at all —
// it would be served and simultaneously invisible to every
// admin surface.
if odb != nil {
if err := odb.EnsureRepo(name, path); err != nil {
log.Printf("registry: recording %s: %v", name, err)
}
}
}
rg.refreshPrivate()
}
if *errorlogPath != "" {
f, ferr := os.OpenFile(*errorlogPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if ferr != nil {
return fmt.Errorf("errorlog: %w", ferr)
}
errlog.Default.SetOutput(f)
}
mux, err := buildHandler(deps, notifier, chatLive, verify, sessions, inbox, *tracker)
if err != nil {
return err
}
if fronter != nil {
mount := http.NewServeMux()
// Sync fronting (0001 §7): zit clients POST the repo URL
// verbatim; /xfer is the alias zit serve itself answers on.
mount.Handle("POST /r/{repo}", fronter)
mount.Handle("POST /r/{repo}/xfer", fronter)
mount.Handle("/", mux)
mux = mount
}
fmt.Println("quarry", version, "listening on", *addr)
return http.ListenAndServe(*addr, mux)
}
// chatAdapter bridges the chat data plane to web.ChatLive: it maps a
// repo name to its project code and converts messages to the view type,
// keeping web free of the data-plane package.
type chatAdapter struct {
room *chat.Room
codes map[string]string
mu *sync.Mutex
}
func (a *chatAdapter) SubscribeChat(repo string) (<-chan web.ChatMessage, func()) {
a.mu.Lock()
code, ok := a.codes[repo]
a.mu.Unlock()
if !ok {
ch := make(chan web.ChatMessage)
close(ch)
return ch, func() {}
}
src, cancel := a.room.Subscribe(code)
out := make(chan web.ChatMessage, 32)
go func() {
defer close(out)
for m := range src {
select {
case out <- web.ChatMessage{User: m.User, Text: m.Text, At: m.At}:
default:
}
}
}()
return out, cancel
}
// notifyAdapter bridges api.NotifyPlane to the notify client, mapping a
// quarry event to a Novu-shaped trigger. The event's To is the
// subscriber id (a user or a "repo:<name>" topic key).
type notifyAdapter struct{ client *notify.Client }
func (a *notifyAdapter) Trigger(ctx context.Context, e api.NotifyEvent) error {
// NotifyEvent.To is documented as "a user, or a repo:<name> topic".
// notify addresses those through different fields — subscriberId
// names one person, topicKey fans out — so the convention has to be
// honoured here or an activity event reaches nobody: it lands on an
// invented subscriber literally called "repo:synthetic" while every
// watcher of that repository hears nothing.
to := notify.Recipient{SubscriberID: e.To}
if repo, ok := strings.CutPrefix(e.To, "repo:"); ok && repo != "" {
to = notify.Recipient{TopicKey: repoTopicKey(repo)}
}
return a.client.Trigger(ctx, e.Workflow, to, e.Payload, e.TxnID)
}
// startEvents opens the event plane: an external NATS when a URL is
// given (multi-node), else an embedded server under dir (single node).
func startEvents(url, dir string) (*events.Server, error) {
if url != "" {
fmt.Println("quarry: connecting to external NATS", url)
return events.Connect(url)
}
return events.Start(filepath.Join(dir, "nats"))
}
// buildHandler wires the two registry consumers onto one mux. notifier
// may be nil (no live pages).
func buildHandler(deps api.Deps, notifier web.Notifier, chatLive web.ChatLive, verify api.Verify, sessions web.Sessions, inbox web.Inbox, tracker string) (http.Handler, error) {
r := ops.NewRegistry()
if err := api.RegisterAll(r, deps); err != nil {
return nil, err
}
apiRouter, err := api.Router(r, deps.Version, verify)
if err != nil {
return nil, err
}
mux := http.NewServeMux()
mux.Handle("/api/", apiRouter)
var opts []web.Option
if sessions != nil {
opts = append(opts, web.WithSessions(sessions))
}
if tracker != "" {
opts = append(opts, web.WithTracker(tracker))
}
if inbox != nil {
opts = append(opts, web.WithInbox(inbox))
}
mux.Handle("/", web.HandlerWith(r, notifier, chatLive, opts, web.Verify(verify)))
return mux, nil
}
// printSpec emits the OpenAPI document generated from the full registry —
// the same registry serve will run, so the spec cannot drift (0001 §9).
func printSpec(w *os.File) error {
r := ops.NewRegistry()
if err := api.RegisterAll(r, api.Deps{Version: version}); err != nil {
return err
}
spec, err := ops.Spec(r, "quarry", version)
if err != nil {
return err
}
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
return enc.Encode(spec)
}
// requiredRole maps an op's declared capability onto the role an account
// must hold on that repo. This is the only place the ops vocabulary and
// the ACL vocabulary meet, so adding a level is a change here and
// nowhere else.
func requiredRole(need ops.Auth) oltp.Role {
switch need {
case ops.Admin:
return oltp.RepoAdmin
case ops.Token:
return oltp.Write
}
return oltp.NoRole
}
// accountCmd manages forge accounts and per-repo roles (0001 §8's v2
// tables). Distinct from `quarry user add`, which writes the SYNC
// credential: that one is the RFC §6.3 shared secret a push presents,
// this one is the login a person uses. --repo-name names a REGISTERED
// repository, because roles are held against served names, not paths.
func accountCmd(args []string) error {
const usage = `usage:
quarry account add --user U [--password P] [--admin]
quarry account passwd --user U [--password P]
quarry account grant --user U --repo-name N --role read|write|admin
quarry account revoke --user U --repo-name N
quarry account delete --user U
quarry account list
every form also takes --data DIR, or one --repo name=path to locate it`
if len(args) < 1 {
return errors.New(usage)
}
sub := args[0]
fs := flag.NewFlagSet("account "+sub, flag.ContinueOnError)
repos := repoFlags{}
fs.Var(repos, "repo", "a registered repository as name=path, used only to locate the data directory")
data := fs.String("data", "", "derived-data directory (default: .quarry beside the repo)")
user := fs.String("user", "", "account name")
password := fs.String("password", "", "password (omit to read from QUARRY_PASSWORD)")
repoTarget := fs.String("repo-name", "", "repository the role applies to")
role := fs.String("role", "", "read, write or admin")
admin := fs.Bool("admin", false, "site admin: every repository, every capability")
if err := fs.Parse(args[1:]); err != nil {
return err
}
dir := *data
if dir == "" {
if len(repos) != 1 {
return errors.New("need --data DIR, or one --repo name=path to locate it\n" + usage)
}
for _, p := range repos {
dir = filepath.Join(p, ".quarry")
}
}
db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
if err != nil {
return err
}
defer db.Close()
// Password sources in order: the flag, then the environment, so a
// password need not reach the process table or shell history.
readPassword := func() (string, error) {
if *password != "" {
return *password, nil
}
if pw := os.Getenv("QUARRY_PASSWORD"); pw != "" {
return pw, nil
}
return "", errors.New("no password: use --password or QUARRY_PASSWORD")
}
switch sub {
case "add":
if *user == "" {
return errors.New("account add wants --user")
}
pw, err := readPassword()
if err != nil {
return err
}
first := !db.HasAccounts()
if err := db.AddAccount(*user, pw, *admin); err != nil {
return err
}
if *admin {
fmt.Printf("account %q created (site admin)\n", *user)
} else {
fmt.Printf("account %q created\n", *user)
}
if first {
fmt.Println("this is the first account: web and API writes now require one")
}
return nil
case "passwd":
if *user == "" {
return errors.New("account passwd wants --user")
}
pw, err := readPassword()
if err != nil {
return err
}
if err := db.SetPassword(*user, pw); err != nil {
return err
}
fmt.Printf("password changed for %q\n", *user)
return nil
case "grant":
if *user == "" || *repoTarget == "" || *role == "" {
return errors.New("account grant wants --user, --repo-name and --role")
}
r, err := oltp.ParseRole(*role)
if err != nil {
return err
}
if err := db.Grant(*user, *repoTarget, r); err != nil {
return err
}
fmt.Printf("%q now has %s on %s\n", *user, r, *repoTarget)
return nil
case "revoke":
if *user == "" || *repoTarget == "" {
return errors.New("account revoke wants --user and --repo-name")
}
if err := db.Revoke(*user, *repoTarget); err != nil {
return err
}
fmt.Printf("%q no longer has a role on %s\n", *user, *repoTarget)
return nil
case "delete":
if *user == "" {
return errors.New("account delete wants --user")
}
if err := db.DeleteAccount(*user); err != nil {
return err
}
fmt.Printf("account %q deleted, with every role it held\n", *user)
return nil
case "list":
accts, err := db.Accounts()
if err != nil {
return err
}
if len(accts) == 0 {
fmt.Println("no accounts: web and API writes are open (single-user local serving)")
return nil
}
for _, a := range accts {
line := a.Name
if a.Admin {
line += " (site admin)"
}
grants, err := db.GrantsFor(a.Name)
if err != nil {
return err
}
for _, repo := range oltp.SortedRepos(grants) {
line += fmt.Sprintf(" %s=%s", repo, grants[repo])
}
fmt.Println(line)
}
return nil
}
return errors.New(usage)
}
// hookSubs adapts the OLTP database to the dispatcher's Subscriptions
// view. The two packages keep their own types on purpose: oltp owns the
// stored row (secrets, timestamps, paused flag), webhook owns what a
// delivery needs, and this is the only place they meet.
type hookSubs struct{ db *oltp.DB }
func (h hookSubs) ActiveWebhooks(repo string) ([]webhook.Sub, error) {
rows, err := h.db.ActiveWebhooks(repo)
if err != nil {
return nil, err
}
out := make([]webhook.Sub, 0, len(rows))
for _, w := range rows {
out = append(out, webhook.Sub{ID: w.ID, URL: w.URL, Secret: w.Secret, Active: true})
}
return out, nil
}
func (h hookSubs) SubByID(id int64) (webhook.Sub, bool) {
w, ok := h.db.WebhookByID(id)
if !ok {
return webhook.Sub{}, false
}
return webhook.Sub{ID: w.ID, URL: w.URL, Secret: w.Secret, Active: w.Active}, true
}
// hookPlane adapts the dispatcher to the ops layer's WebhookPlane, which
// speaks primitives so the ops never import the delivery machinery.
type hookPlane struct {
d *webhook.Dispatcher
at func() string
}
func (h hookPlane) Raise(id, repo, kind, actor, summary, link string) error {
return h.d.Raise(webhook.Event{
ID: id, Repo: repo, Kind: kind, Actor: actor,
Summary: summary, Link: link, At: h.at(),
})
}
// webhookCmd manages subscriptions to quarry's outbound event surface
// (Principle 6). The signing secret is printed once at creation and
// never again — a key a listing can read back is a key that leaks
// through logs, screenshots and shoulders.
func webhookCmd(args []string) error {
const usage = `usage:
quarry webhook add --repo-name N --url https://...
quarry webhook list --repo-name N
quarry webhook pause --id N
quarry webhook resume --id N
quarry webhook delete --id N
every form also takes --data DIR, or one --repo name=path to locate it`
if len(args) < 1 {
return errors.New(usage)
}
sub := args[0]
fs := flag.NewFlagSet("webhook "+sub, flag.ContinueOnError)
repos := repoFlags{}
fs.Var(repos, "repo", "a registered repository as name=path, used only to locate the data directory")
data := fs.String("data", "", "derived-data directory")
repoTarget := fs.String("repo-name", "", "repository whose events to deliver")
endpoint := fs.String("url", "", "endpoint to POST events to")
id := fs.Int64("id", 0, "subscription id")
if err := fs.Parse(args[1:]); err != nil {
return err
}
dir := *data
if dir == "" {
if len(repos) != 1 {
return errors.New("need --data DIR, or one --repo name=path to locate it\n" + usage)
}
for _, p := range repos {
dir = filepath.Join(p, ".quarry")
}
}
db, err := oltp.Open(filepath.Join(dir, "oltp.db"))
if err != nil {
return err
}
defer db.Close()
switch sub {
case "add":
if *repoTarget == "" || *endpoint == "" {
return errors.New("webhook add wants --repo-name and --url")
}
w, err := db.AddWebhook(*repoTarget, *endpoint)
if err != nil {
return err
}
fmt.Printf("webhook %d: %s -> %s\n", w.ID, w.Repo, w.URL)
fmt.Printf("signing secret (shown once): %s\n", w.Secret)
fmt.Println("verify X-Quarry-Signature as sha256=HMAC-SHA256(secret, body)")
return nil
case "list":
if *repoTarget == "" {
return errors.New("webhook list wants --repo-name")
}
hooks, err := db.Webhooks(*repoTarget)
if err != nil {
return err
}
if len(hooks) == 0 {
fmt.Printf("no webhooks on %s\n", *repoTarget)
return nil
}
for _, w := range hooks {
state := "active"
if !w.Active {
state = "paused"
}
fmt.Printf("%d %-7s %s (added %s)\n", w.ID, state, w.URL, w.Created)
}
return nil
case "pause", "resume":
if *id == 0 {
return fmt.Errorf("webhook %s wants --id", sub)
}
if err := db.SetWebhookActive(*id, sub == "resume"); err != nil {
return err
}
fmt.Printf("webhook %d %sd\n", *id, sub)
return nil
case "delete":
if *id == 0 {
return errors.New("webhook delete wants --id")
}
if err := db.DeleteWebhook(*id); err != nil {
return err
}
fmt.Printf("webhook %d deleted\n", *id)
return nil
}
return errors.New(usage)
}
// sessionPlane adapts the OLTP database to the web's Sessions seam. The
// authorization half deliberately reuses the same role lookup the Basic
// path uses, so a cookie can never be worth more than a password.
type sessionPlane struct{ db *oltp.DB }
func (s sessionPlane) Login(user, password string) (string, time.Time, bool) {
return s.db.NewSession(user, password)
}
func (s sessionPlane) Resolve(token string) (string, bool) {
acct, ok := s.db.SessionAccount(token)
return acct.Name, ok
}
func (s sessionPlane) Logout(token string) { s.db.EndSession(token) }
func (s sessionPlane) Allow(user, repo string, need ops.Auth) bool {
if !s.db.HasAccounts() {
return true
}
// The caller has already proven the cookie, so this is authorization,
// not authentication — but it runs the SAME role lookup the password
// path does, so a cookie is never worth more than a password.
a, found := s.db.AccountByName(user)
if !found {
return false
}
return s.db.RoleOf(a, repo) >= requiredRole(need)
}
// 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) DeleteMirror(repo string) error { return m.db.DeleteMirror(repo) }
func (m mirrorStore) RecordMirrorRun(repo string, err error) error {
return m.db.RecordMirrorRun(repo, time.Now(), err)
}
func toAPIMirror(r oltp.Mirror) api.Mirror {
return api.Mirror{
Repo: r.Repo, URL: r.URL, Every: int(r.Every.Seconds()), Enabled: r.Enabled,
LastAt: r.LastAt, LastOK: r.LastOK, LastErr: r.LastErr, Failures: r.Failures,
}
}
// mirrorScheduler pulls every due mirror on a tick. It is one goroutine
// for the whole fleet rather than one per store on purpose: pulls are
// network-bound against OTHER PEOPLE'S servers, and eleven goroutines
// each free to hit fossil-scm.org whenever their own timer fires is a
// worse neighbour than a single loop that walks the list.
func mirrorScheduler(deps *api.Deps, odb *oltp.DB, tick time.Duration) {
for range time.Tick(tick) {
rows, err := odb.Mirrors()
if err != nil {
log.Printf("mirror: listing: %v", err)
continue
}
now := time.Now()
for _, row := range rows {
if !row.Due(now) {
continue
}
changed, err := deps.PullMirror(toAPIMirror(row))
switch {
case err != nil:
log.Printf("mirror: %s from %s: %v", row.Repo, row.URL, err)
case changed:
log.Printf("mirror: %s pulled new artifacts from %s", row.Repo, row.URL)
}
// A quiet pull is not logged. Almost every one is quiet, and
// a log line per store per interval buries the ones that matter.
}
}
}
type ciStore struct{ db *oltp.DB }
// ciRuns adapts the OLTP rows to the ops layer's view.
type ciRuns struct{ db *oltp.DB }
func (c ciRuns) CIRuns(repo string, limit int) ([]api.CIRun, error) {
rows, err := c.db.CIRuns(repo, limit)
return toAPIRuns(rows), err
}
func (c ciRuns) CIRunsFor(repo, checkin string) ([]api.CIRun, error) {
rows, err := c.db.CIRunsFor(repo, checkin)
return toAPIRuns(rows), err
}
func toAPIRuns(rows []oltp.CIRun) []api.CIRun {
out := make([]api.CIRun, 0, len(rows))
for _, r := range rows {
out = append(out, api.CIRun{RunID: r.RunID, Checkin: r.Checkin,
Workflow: r.Workflow, Pipeline: r.Pipeline, Status: r.Status, Started: r.Started})
}
return out
}
func (s ciStore) CIWatermark(repo string) (string, bool) { return s.db.CIWatermark(repo) }
func (s ciStore) SetCIWatermark(repo, c string) error { return s.db.SetCIWatermark(repo, c) }
func (s ciStore) AddRun(runID, repo, checkin, workflow, pipeline, status string) error {
return s.db.AddCIRun(oltp.CIRun{RunID: runID, Repo: repo, Checkin: checkin,
Workflow: workflow, Pipeline: pipeline, Status: status})
}
// ciRepo adapts one repository's index and store to the trigger's read
// side: the index answers "what is new", zit answers "what is in it".
type ciRepo struct {
ix *index.Index
zit *zitcli.Client
}
func (r ciRepo) NewCheckins(after string, limit int) ([]ci.Checkin, error) {
entries, err := r.ix.CheckinsAfter(after, limit)
if err != nil {
return nil, err
}
out := make([]ci.Checkin, 0, len(entries))
for _, e := range entries {
out = append(out, ci.Checkin{Hash: e.Hash, Branch: e.Branch,
Date: e.Date, User: e.User, Comment: e.Comment})
}
return out, nil
}
func (r ciRepo) Files(version string) ([]string, error) {
entries, err := r.zit.Ls(context.Background(), version)
if err != nil {
return nil, err
}
out := make([]string, 0, len(entries))
for _, e := range entries {
out = append(out, e.Name)
}
return out, nil
}
func (r ciRepo) Read(path, version string) ([]byte, error) {
return r.zit.Cat(context.Background(), path, version)
}
// pollCI refreshes the status of runs quarry still thinks are going and
// writes the durable verdict when one finishes.
//
// dagnats owns the run; quarry caches only the last status it saw, so a
// missed poll costs nothing but freshness. The verdict is the part that
// must not be lost: it becomes a tag artifact on the check-in, which
// clones with the repository (0001 §6's no-hostage rule).
func pollCI(engine *ci.Engine, db *oltp.DB, repos *api.Repos, verdicts bool) {
runs, err := db.UnfinishedCIRuns()
if err != nil {
log.Printf("ci: poll: %v", err)
return
}
for _, r := range runs {
run, err := engine.Status(r.RunID)
if err != nil {
continue // the engine may be restarting; try again next tick
}
if run.Status == r.Status {
continue
}
if err := db.SetCIStatus(r.RunID, run.Status); err != nil {
log.Printf("ci: status %s: %v", r.RunID, err)
continue
}
if !run.Terminal() || !verdicts {
continue
}
repo, ok := repos.Get(r.Repo)
if !ok {
continue
}
verdict := "ci-fail"
if run.Passed() {
verdict = "ci-pass"
}
// Written through the lease holder, like every other artifact
// quarry mints: a verdict is a write, and two nodes must not
// both record one.
if err := repo.WriteTag(verdict, r.Checkin, r.RunID); err != nil {
log.Printf("ci: verdict %s on %.12s: %v", verdict, r.Checkin, err)
continue
}
log.Printf("ci: %s %.12s %s (run %s)", r.Repo, r.Checkin, verdict, r.RunID)
}
}