// Package ops is quarry's seam: every capability is an op — one named,
// typed unit in a single registry. The registry has exactly three consumers
// (the API router, HTML handlers, the test harnesses), and the API spec is
// generated FROM the registry, never maintained beside it. Only this package
// may import the data planes. See docs/rfcs/0001-quarry-forge.md §9.
package ops
import (
"context"
"errors"
"fmt"
"io"
"log"
"net/http"
"reflect"
"sort"
"strings"
"time"
"craftdesign.group/quarry/internal/obs"
"craftdesign.group/quarry/internal/timing"
)
// ErrNotFound marks an op error as "the referenced entity does not
// exist" — consumers map it to 404-shaped responses instead of treating
// it as a server fault. Wrap with fmt.Errorf("...: %w", ErrNotFound).
var ErrNotFound = errors.New("not found")
// ErrInvalid marks an op error as a client input fault the validator
// couldn't catch (a value only the Run closure can reject) — consumers
// map it to 400, not 500. Wrap with fmt.Errorf("...: %w", ErrInvalid).
var ErrInvalid = errors.New("invalid request")
// ErrUnavailable marks a write that this node cannot serve right now —
// it does not hold the store's write lease, or the coordinator is
// unreachable (0001 §3). Consumers map it to 503; the client may retry
// (another node holds the lease, or the coordinator will return).
var ErrUnavailable = errors.New("temporarily unavailable")
// ErrStarting is ErrUnavailable with a cause a reader can act on: the
// repository is registering after the listener bound and will exist in
// a moment. errors.Is(err, ErrUnavailable) stays true, so every
// existing 503 path applies; the web can name the wait.
var ErrStarting = fmt.Errorf("still starting: %w", ErrUnavailable)
// ErrStopping is the other end of ErrStarting: the forge was SHUTTING
// DOWN while this request was in flight. systemd SIGTERMs the whole
// cgroup on a restart, so a deploy landing mid-operation kills the
// subprocess doing the work.
//
// Its own sentinel rather than bare ErrUnavailable, for ErrStarting's
// reason: "at capacity" is a story about load, and a deploy is not
// load, so the generic 503 wording would send a reader looking for a
// problem that is not there. errors.Is(err, ErrUnavailable) stays
// true, so every existing 503 path still applies.
var ErrStopping = fmt.Errorf("the forge was restarting: %w", ErrUnavailable)
// ErrForbidden marks a write the policy refuses for this store however
// well-formed and well-authenticated it is — a mirror that does not
// accept local writes. Consumers map it to 403; retrying will not
// help, the message says where the write belongs.
var ErrForbidden = errors.New("forbidden")
// Refusal is a refusal a client is expected to act on rather than
// read: a stable Code and the Fields that go with it travel beside the
// sentence. The JSON surface writes them as top-level keys of the error
// body, so a field is never named "error" or "code". Class is the
// sentinel it answers to and so its status; nil means ErrInvalid.
type Refusal struct {
Code string
Message string
Fields map[string]string
Class error
}
func (r *Refusal) Error() string { return r.Message }
func (r *Refusal) Unwrap() error {
if r.Class == nil {
return ErrInvalid
}
return r.Class
}
// Kind separates reads from writes. Queries are in-process reads over the
// local index and must be safe to retry and cache; Commands mutate and are
// routed (lease, durability) inside their Run.
type Kind int
const (
Query Kind = iota
Command
// Stream is a Query that never returns one JSON body: it serves
// Server-Sent Events instead. Registered like a Query (GET, no
// surprises for Auth/Caps) so it enumerates and documents the same
// way; the router recognizes Kind == Stream and switches transports
// rather than calling Run. RFC 0012 §1: the registry's first
// streaming kind, added as the smallest seam that lets chat.stream
// (and later me.attention) be an op like any other rather than a
// bespoke web-only route.
Stream
)
func (k Kind) String() string {
switch k {
case Query:
return "query"
case Stream:
return "stream"
default:
return "command"
}
}
// Auth is the whole v1 auth model (0001 §8).
//
// Admin here means "site admin", and nothing narrower — a site admin
// administers the FORGE. A repository, mirrored or not, belongs to
// whoever created it, and every decision about THAT repository —
// visibility, roles, protected branches, webhooks, its bucket, its
// mirror, deleting it — is theirs; those ops are Token, refined by the
// `a` letter on the repository named in the request (docs/hosting.md).
// A fleet belongs to its own owner the same way. Admin is reserved for
// what stays with the forge regardless of who owns what: accounts and
// invites, the ops registry's own introspection, and the machine's
// shared resources — pull-interval floors, quotas, storage and archive
// caches, CI worker capacity, rate limits, the exception panel. When a
// gate here reads Admin, that is the claim being made about it.
type Auth int
const (
Anonymous Auth = iota
Token
Admin
)
// Op is the registration shape. In and Out must be structs whose fields are
// expressible as both form/URL values and JSON — flat, no exotic nesting.
type Op[In, Out any] struct {
Name string // "timeline.list", "chat.post"
Doc string // one-line summary; becomes the spec's operation summary
Kind Kind
Auth Auth
// Caps is the op's fossil capability set (RFC 0004 §4): possession
// of ANY listed letter passes. Auth remains the coarse router
// floor; Caps is the per-surface refinement. Empty = ungated
// beyond Auth (meta ops, echo).
Caps string
// Scope says WHAT the Caps letters are asked about. A gate that
// resolves the wrong subject answers correctly about the wrong
// thing, which is indistinguishable from a policy decision and has
// cost this forge four separate incidents: a repository resolving
// to "" so caps were read at the fleet scope and every repository
// admin was refused (b683c54b); an empty pusher owning nothing, so
// branch protection refused everybody for eleven hours; the
// settings cog asking the fleet about a repository; and a project
// code that came back empty, reported as "no credential".
//
// An op whose Route names {repo} is ScopeRepo without saying so —
// the router always has a repository there. Every other op with
// Caps must declare, and Register refuses one that does not, so the
// question is answered when an op is written rather than discovered
// when somebody is refused.
Scope Scope
Route string // "GET /api/v1/r/{repo}/timeline"
// MaxBody bounds the request body for an op whose input can be
// large, given the repository the route names. The router refuses
// a longer body with 413 before reading it whole. nil, or an answer
// of zero, leaves the body unbounded here, as every other op is.
MaxBody func(repo string) int64
Run func(context.Context, In) (Out, error)
// StreamRun replaces Run when Kind == Stream. It blocks, emitting
// events through Streamer until ctx is done or it returns an error;
// the router owns the transport (headers, flushing), the op owns
// the subject and the pacing. Out documents the shape each Emit
// call's data argument carries — it is never itself marshaled.
StreamRun func(context.Context, In, Streamer) error
}
// Streamer is what a Stream op writes Server-Sent Events through. The
// router supplies the transport-bound implementation; a Stream op never
// touches http directly, matching every other op's separation of the
// subject from the wire (0001 Principle 6).
type Streamer interface {
// Emit writes one event: id becomes the `id:` line (0 omits it, for
// an op's first, unresumable event), data is JSON-marshaled as the
// `data:` line.
Emit(id uint64, data any) error
// Event writes one NAMED event: an `event: <name>` line ahead of
// the data, for a frame that is not another item of the stream's
// ordinary payload (chat.stream's `gap`, say). It carries no `id:`,
// because it is not a resume point.
Event(name string, data any) error
// Ping writes a comment line — keeps idle connections alive through
// proxies that time out a quiet stream; never seen by the client as
// an event.
Ping()
}
// Descriptor is the type-erased registry entry the consumers work from.
// Scope is the subject a Caps check is about.
type Scope uint8
const (
// ScopeUnset is the zero value. It is legal only for an op whose
// route names {repo}, where the subject cannot be missing.
ScopeUnset Scope = iota
// ScopeRepo asks about one repository, named in the body rather
// than the path. The router refuses to evaluate these against an
// empty repository: a missing subject is a fault in the forge, not
// a decision about the caller, and saying so out loud is the whole
// point of this type.
ScopeRepo
// ScopeFleet asks about the forge as a whole. There is no
// repository and none is expected.
ScopeFleet
// ScopeCaller asks about the signed-in account across every
// repository — "my feed", "my tickets", "my saved views".
//
// These are evaluated at the fleet scope today, which is what makes
// a role-less account fail them, and whether that is right is the
// open question in 7ddd1768. Declaring it changes no behaviour; it
// records which ops the question is about, so the answer can be
// applied in one place instead of found one refusal at a time.
ScopeCaller
)
func (s Scope) String() string {
switch s {
case ScopeRepo:
return "repo"
case ScopeFleet:
return "fleet"
case ScopeCaller:
return "caller"
}
return "unset"
}
// File is what an op returns when its result is BYTES rather than a
// value: an archive, a raw blob. JSON has no binary type, so an op that
// answered bytes used to read the file into memory, base64 it into a
// JSON result, and have the web route decode it back — three copies of
// a 13 MB archive and a 17 MB string per request, and an API client got
// an envelope where it wanted a download (6dbaeb21, a0770478).
//
// A File is served as itself: the router writes the bytes with the
// content type and name here, the web route hands the same value to
// the same writer, and RFC 0001 §9 still holds — one op, one rule,
// both surfaces. Open is called at most once and only when the body
// will be sent: ETag is answered first, so a reader who already has the
// bytes costs nothing, and an archive that has to be built is built
// after the 304 check, not before.
type File interface {
// Name is the filename the download is offered under.
Name() string
// ContentType is the media type of the bytes.
ContentType() string
// ETag names the bytes when they are addressable — for an archive,
// the resolved check-in — and "" when they are not. Quoted by the
// writer, and honoured for a 304 whenever present.
ETag() string
// Immutable says the bytes at THIS URL can never change, so a
// cache may keep them for a year without asking. Distinct from
// having an ETag: /archive/trunk resolves to a hash and names its
// bytes, but trunk moves, so the URL is not immutable and must
// revalidate — no-cache plus the ETag. Only a URL that names a full
// hash is.
Immutable() bool
// Open produces the bytes and their length. It is where any work
// happens, so an error here is an op error and is mapped like one.
Open() (io.ReadCloser, int64, error)
}
type Descriptor struct {
Name string
Doc string
Kind Kind
Auth Auth
Caps string
Scope Scope
Method string // "GET", "POST"
Pattern string // ServeMux 1.22 pattern: "/api/v1/r/{repo}/timeline"
In reflect.Type
Out reflect.Type
MaxBody func(repo string) int64
run func(context.Context, any) (any, error)
streamRun func(context.Context, any, Streamer) error
}
// Registry holds every op. One instance per process, filled by an explicit
// RegisterAll at startup — no init() magic, so the wiring stays greppable.
type Registry struct {
observe Observed
byName map[string]*Descriptor
byRoute map[string]*Descriptor
}
func NewRegistry() *Registry {
return &Registry{
byName: make(map[string]*Descriptor),
byRoute: make(map[string]*Descriptor),
}
}
// Register validates and adds one op. It rejects, rather than papers over:
// duplicate names, duplicate routes, kind/method mismatches (queries are
// GETs, commands are POSTs — Principle 1), and non-struct In/Out.
func Register[In, Out any](r *Registry, op Op[In, Out]) error {
if op.Name == "" {
return fmt.Errorf("ops: op with route %q has no name", op.Route)
}
if op.Kind == Stream {
if op.StreamRun == nil {
return fmt.Errorf("ops: %s: nil StreamRun", op.Name)
}
} else if op.Run == nil {
return fmt.Errorf("ops: %s: nil Run", op.Name)
}
if op.Doc == "" {
return fmt.Errorf("ops: %s: Doc is required (it becomes the spec summary)", op.Name)
}
if _, dup := r.byName[op.Name]; dup {
return fmt.Errorf("ops: duplicate op name %q", op.Name)
}
method, pattern, err := splitRoute(op.Route)
if err != nil {
return fmt.Errorf("ops: %s: %w", op.Name, err)
}
if (op.Kind == Query || op.Kind == Stream) && method != http.MethodGet {
return fmt.Errorf("ops: %s: query/stream ops must route as GET, got %s", op.Name, method)
}
if op.Kind == Command && method != http.MethodPost {
return fmt.Errorf("ops: %s: command ops must route as POST, got %s", op.Name, method)
}
// A capability check needs a subject. An op whose route names
// {repo} always has one; every other gated op must say what its
// letters are about, because a gate that resolves nothing and asks
// anyway gets a confident answer to the wrong question. Refused at
// REGISTRATION so a new op cannot reach production without the
// question being answered — the alternative is finding out when
// somebody is wrongly refused, which is how we found the last four.
scope := op.Scope
if scope == ScopeUnset && strings.Contains(pattern, "{repo}") {
scope = ScopeRepo
}
if op.Caps != "" && scope == ScopeUnset {
return fmt.Errorf("ops: %s: has Caps %q and no Scope, and its route %q does not name {repo}: "+
"say whether those letters are about a repository (ScopeRepo), the forge (ScopeFleet), "+
"or the signed-in account (ScopeCaller)", op.Name, op.Caps, pattern)
}
// The other half of the same rule: a caller-scoped op cannot carry
// letters at all, because there is no repository to resolve them
// against.
//
// ScopeCaller means "about the signed-in account across every
// repository". The router evaluates a letter against the op's
// repository, and a caller-scoped op has none — so it resolves to ""
// and answers with the FLEET, where an ordinary account holds nothing
// a repository granted it. Every op that declared one was therefore
// refused for every ordinary account: me.feed, the saved views, the
// cross-repository ticket listing, mirrors. The letters read like
// gates and were in effect a site-admin-only filter nobody had
// written down, and two more (people.list, search.all) passed only
// because the letters they named happen to be in the fleet default.
//
// The gate for these is Auth — signed in — with the op's own body
// filtering by reach (myRepos, CanRead). Refused at REGISTRATION for
// the reason the rule above gives: the alternative is discovering it
// one refusal at a time, which is 7ddd1768.
if op.Caps != "" && scope == ScopeCaller {
return fmt.Errorf("ops: %s: is ScopeCaller and declares Caps %q, which nothing can evaluate: "+
"a caller-scoped letter resolves against no repository, so the router asks the fleet and "+
"refuses every account a repository granted it (7ddd1768). Gate it with Auth, and filter "+
"by reach in Run — myRepos, CanRead", op.Name, op.Caps)
}
routeKey := method + " " + pattern
if other, dup := r.byRoute[routeKey]; dup {
return fmt.Errorf("ops: %s: route %q already registered by %s", op.Name, routeKey, other.Name)
}
inT := reflect.TypeFor[In]()
outT := reflect.TypeFor[Out]()
if inT.Kind() != reflect.Struct {
return fmt.Errorf("ops: %s: In must be a struct, got %s", op.Name, inT.Kind())
}
if outT.Kind() != reflect.Struct {
return fmt.Errorf("ops: %s: Out must be a struct, got %s", op.Name, outT.Kind())
}
if err := checkValidateTags(inT); err != nil {
return fmt.Errorf("ops: %s: In: %w", op.Name, err)
}
d := &Descriptor{
Name: op.Name,
Doc: op.Doc,
Kind: op.Kind,
Auth: op.Auth,
Caps: op.Caps,
Scope: scope,
Method: method,
Pattern: pattern,
In: inT,
Out: outT,
MaxBody: op.MaxBody,
}
if op.Kind == Stream {
d.streamRun = func(ctx context.Context, in any, s Streamer) error {
typed, ok := in.(In)
if !ok {
return fmt.Errorf("ops: %s: input is %T, want %s", op.Name, in, inT)
}
return op.StreamRun(ctx, typed, s)
}
} else {
d.run = func(ctx context.Context, in any) (any, error) {
typed, ok := in.(In)
if !ok {
return nil, fmt.Errorf("ops: %s: input is %T, want %s", op.Name, in, inT)
}
return op.Run(ctx, typed)
}
}
r.byName[op.Name] = d
r.byRoute[routeKey] = d
return nil
}
// NewIn returns a pointer to a zero In struct for this op — the decode
// target the codecs fill before Invoke.
func (d *Descriptor) NewIn() any {
return reflect.New(d.In).Interface()
}
// slowOp is when an op is worth naming in the log. Low enough to catch
// the ones that make a page feel broken, high enough that an ordinary
// page logs nothing.
const slowOp = 250 * time.Millisecond
// Invoke runs a registered op by name. in must be the op's In struct value.
func (r *Registry) Invoke(ctx context.Context, name string, in any) (any, error) {
d, ok := r.byName[name]
if !ok {
return nil, fmt.Errorf("ops: unknown op %q", name)
}
// Timed here because this is the one dispatch every surface shares:
// timed in the web's page helper instead, an op called over /api/v1
// was never measured, and a slow API ticket.set could not be seen.
// The span CONTAINS the zit and sql time beneath it; the footer and
// the ops panels report it as a container, never a sum.
start := time.Now()
defer timing.RecordNamed(ctx, timing.Ops, name, start)
// A page that takes ten seconds says "ops 11.09s×8" and stops there:
// the container is timed but not its contents, so the slow one
// cannot be named. Past this threshold the log says which op it was,
// and which surface asked for it.
defer func() {
if el := time.Since(start); el > slowOp {
surface := obs.Surface(ctx)
if surface == "" {
surface = "none"
}
log.Printf("slow op: %s took %s surface=%s", name, el.Round(time.Millisecond), surface)
}
}()
out, err := d.run(ctx, in)
// One seam for "something was done", because there is one Invoke.
// Only commands: a query changes nothing and a log of reads is a
// different thing with different consent questions attached.
// Only successes: a refusal is not an act.
if r.observe != nil && d.Kind == Command && err == nil {
r.observe(ctx, name, in)
}
return out, err
}
// Observed is told that a command succeeded. It is given the op's name
// and its input, and decides for itself whether the act is worth
// recording -- the registry knows nothing about owners or operators.
type Observed func(ctx context.Context, op string, in any)
// Observe installs the witness. One per registry, set before serving.
func (r *Registry) Observe(fn Observed) { r.observe = fn }
// InvokeStream runs a registered Stream op by name, blocking until s's
// caller cancels the context or the op's StreamRun returns.
func (r *Registry) InvokeStream(ctx context.Context, name string, in any, s Streamer) error {
d, ok := r.byName[name]
if !ok {
return fmt.Errorf("ops: unknown op %q", name)
}
if d.streamRun == nil {
return fmt.Errorf("ops: %s: not a stream op", name)
}
return d.streamRun(ctx, in, s)
}
// Lookup returns the descriptor for one op.
func (r *Registry) Lookup(name string) (*Descriptor, bool) {
d, ok := r.byName[name]
return d, ok
}
// All returns every descriptor sorted by name — the enumeration order the
// spec generator and the golden-test harness both rely on.
func (r *Registry) All() []*Descriptor {
out := make([]*Descriptor, 0, len(r.byName))
for _, d := range r.byName {
out = append(out, d)
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out
}
func splitRoute(route string) (method, pattern string, err error) {
method, pattern, found := strings.Cut(route, " ")
if !found || pattern == "" {
return "", "", fmt.Errorf("route %q must be \"METHOD /path\"", route)
}
if method != http.MethodGet && method != http.MethodPost {
return "", "", fmt.Errorf("route %q: method must be GET or POST", route)
}
if !strings.HasPrefix(pattern, "/") {
return "", "", fmt.Errorf("route %q: pattern must start with /", route)
}
return method, pattern, nil
}
// ctxUserKey carries the authenticated principal through Invoke; the
// HTTP consumers set it after verifying credentials (0001 §8 — the
// registry stays transport-free, enforcement lives at the edges).
type ctxUserKey struct{}
// WithUser returns ctx carrying the authenticated user name.
func WithUser(ctx context.Context, user string) context.Context {
return context.WithValue(ctx, ctxUserKey{}, user)
}
// User returns the authenticated user, or "" when anonymous.
func User(ctx context.Context) string {
u, _ := ctx.Value(ctxUserKey{}).(string)
return u
}
// ctxCapsKey carries the caller's resolved capabilities. Op.Caps is the
// coarse question the edge already answers — "may this caller reach
// this surface at all" — and an op that never sees the answer can only
// return the same thing to everyone it lets in. Fossil's model is finer
// than that: e decides which FIELDS a reader sees, and whether a
// private ticket exists for them. So the edge passes the set it
// resolved rather than discarding it after the check.
type ctxCapsKey struct{}
// Caps is what an op needs of a capability set: does the caller hold
// any of these letters. Kept as an interface so ops stays free of the
// OLTP types that implement it.
type Caps interface{ HasAny(letters string) bool }
// WithCaps returns ctx carrying the caller's resolved capabilities.
func WithCaps(ctx context.Context, c Caps) context.Context {
return context.WithValue(ctx, ctxCapsKey{}, c)
}
// CapsOf returns the caller's capabilities, or nil when the edge
// resolved none. A nil result answers false to everything, so an op
// asking "may they see private data" gets the safe answer on a forge
// with no capability plane at all — the default is deny, and reaching
// for a capability that was never resolved cannot grant one.
// HasCaps says a capability plane answered for this call. Ops that
// hide detail from callers lacking a letter must not hide it from a
// forge with no plane at all — there, every caller is the operator.
func HasCaps(ctx context.Context) bool {
_, ok := ctx.Value(ctxCapsKey{}).(Caps)
return ok
}
func CapsOf(ctx context.Context) Caps {
c, _ := ctx.Value(ctxCapsKey{}).(Caps)
if c == nil {
return noCaps{}
}
return c
}
type noCaps struct{}
func (noCaps) HasAny(string) bool { return false }