// 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"
"net/http"
"reflect"
"sort"
"strings"
)
// 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")
// 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")
// 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).
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
Route string // "GET /api/v1/r/{repo}/timeline"
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.
type Descriptor struct {
Name string
Doc string
Kind Kind
Auth Auth
Caps string
Method string // "GET", "POST"
Pattern string // ServeMux 1.22 pattern: "/api/v1/r/{repo}/timeline"
In reflect.Type
Out reflect.Type
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 {
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)
}
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,
Method: method,
Pattern: pattern,
In: inT,
Out: outT,
}
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()
}
// 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)
}
return d.run(ctx, in)
}
// 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.
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 }