Skip to content
quarry Artifact [5a3c937fc9]
Login

Artifact [5a3c937fc9]

Check-ins Using Download Hex

Artifact 5a3c937fc92e761b8fcb3cf498f9e52a9a8dc14983aff4f6a26a6705806c7539:

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

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"

	"craftdesign.group/quarry/internal/api"
	"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 the first repo)")
	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 := *data
	if dir == "" {
		for _, p := range repos {
			dir = filepath.Join(p, ".quarry")
			break
		}
	}
	// 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).
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 the first repo)")
	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")
	// 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")
	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)}
	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 := *data
		if dir == "" {
			for _, p := range repos {
				dir = filepath.Join(p, ".quarry")
				break
			}
		}
		if err := os.MkdirAll(dir, 0o755); err != nil {
			return err
		}
		odb, err := oltp.Open(filepath.Join(dir, "oltp.db"))
		if err != nil {
			return err
		}
		defer odb.Close()
		fronter.SetAuth(odb)
		// Two credentials, on purpose (0001 §8). Accounts are the forge
		// identity and carry per-repo roles; the RFC §6.3 shared secret
		// stays the SYNC credential, because the protocol fixes its
		// derivation and the server must be able to recompute it. Once
		// any account exists it governs web/API writes; before that,
		// v1's per-repo secret still does, so an existing deployment
		// keeps working and a fresh local one stays open.
		// Cookie sessions for the browser; Basic stays for the API.
		sessions = sessionPlane{db: odb}
		var codeMu sync.Mutex
		codes := map[string]string{}
		verify = func(repo, user, pass string, need ops.Auth) api.Decision {
			if odb.HasAccounts() {
				acct, ok := odb.Authenticate(user, pass)
				if !ok {
					return api.Denied
				}
				// Authenticated. From here a refusal is about capability,
				// not identity, and saying so is the difference between a
				// client that stops and one that retries forever.
				if odb.RoleOf(acct, repo) >= requiredRole(need) {
					return api.Allowed
				}
				return api.Insufficient
			}
			if !odb.Gated(repo) {
				return api.Allowed
			}
			secret, ok := odb.Secret(repo, user)
			if !ok || pass == "" {
				return api.Denied
			}
			codeMu.Lock()
			code, cached := codes[repo]
			codeMu.Unlock()
			if !cached {
				path, ok := repos[repo]
				if !ok {
					return api.Denied
				}
				c, err := xfer.ProjectCode(path)
				if err != nil {
					log.Printf("auth: project code of %s: %v", repo, err)
					return api.Denied
				}
				codeMu.Lock()
				codes[repo], code = c, c
				codeMu.Unlock()
			}
			if secret == xfer.SharedSecret(code, user, pass) {
				return api.Allowed
			}
			return api.Denied
		}

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

		// Single-writer coordination (0001 §3). node identifies this
		// process; one embedded NATS means one node, but the machinery
		// and its failure posture (coordinator down → writes refuse) run
		// the same on one node as on N.
		leases, err := lease.New(ev.JS(), *addr, *leaseTTL)
		if err != nil {
			return err
		}
		defer leases.Close()
		// Per-repo chat plane (0001 §5): messages fan out over the same
		// NATS the events ride, with a 7-day JetStream window.
		room, err := chat.New(ev.Conn(), ev.JS())
		if err != nil {
			return err
		}
		deps.Chat = room
		chatLive = &chatAdapter{room: room, codes: codes, mu: &codeMu}
		// Notification triggers (0001 §5): fired best-effort to the notify
		// companion; disabled when --notify is unset.
		if *notifyURL != "" {
			key := *notifyKey
			if key == "" {
				key = os.Getenv("QUARRY_NOTIFY_KEY")
			}
			if key == "" {
				// Better to say so at startup than to discover it as 401s
				// in the log after a write nobody was told about.
				log.Printf("notify: no API key (--notify-key or QUARRY_NOTIFY_KEY); triggers will be refused")
			}
			nclient := notify.New(*notifyURL, key)
			deps.Notify = &notifyAdapter{client: nclient}
			inbox = inboxPlane{client: nclient}
			fmt.Println("quarry: notifications ->", *notifyURL)
		}
		// 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)
		// 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 := &registrar{
			dir: dir, reposRoot: *reposRoot, node: *addr, interval: *interval, ciCloneBase: ciCloneBase,
			ev: ev, leases: leases, deps: &deps, fronter: fronter,
			ingestLane: ingestLane, bulk: deps.Bulk, engine: engine, odb: odb,
			codes: codes, codeMu: &codeMu,
		}
		defer rg.CloseAll()
		deps.Registry = rg
		deps.Accounts = accountStore{db: odb}
		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
			}
		}
	}

	mux, err := buildHandler(deps, notifier, chatLive, verify, sessions, inbox)
	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) (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 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)
	}
}