Runtime API

The generated code runs on github.com/craftgodotdev/craftgo/pkg/server - a thin wrapper over net/http. This page is the API reference for that package. You rarely call most of it directly: the generated main.go wires server.New, wiring.Register (which calls each service's RegisterRoutes), and Start. You reach for this when adding middleware, swapping the JSON codec, customizing health checks, or shaping error responses.

Everything here is plain standard-library shape - http.Handler, http.HandlerFunc, func(http.Handler) http.Handler. There is no custom router: routes go on http.ServeMux, request binding is generated code, and JSON goes through encoding/json unless you install another codec.

Server

srv := server.New(svcCtx, opts...)

New takes your ServiceContext (accepted for the documented constructor shape; the runtime does not introspect it) and any number of Option values.

Option	Effect
WithHealthPaths(HealthPaths{Liveness, Readiness})	Override the default /healthz + /readyz paths.
WithoutDefaultHealth()	Disable the auto-registered health endpoints entirely.
WithTelemetry(mw Middleware)	Install mw - tel.HTTPMiddleware() - outside Recovery and every Use middleware, so their log lines carry the span it opens. The health probes bypass it, a panic in mw itself is not recovered, and WithTelemetry(nil) installs nothing and keeps an earlier one.

Lifecycle

Method	Description
Use(mw Middleware) *Server	Append a global middleware. Outermost-added wraps first.
Handle(pattern, h http.Handler, mws ...Middleware) *Server	Register a route. Optional per-route middlewares wrap the handler outermost-first (first arg = outermost frame); the route's limits - a WithLimits h's own, else the server's defaults - wrap them all.
HandleFunc(pattern, fn http.HandlerFunc) *Server	Register a route from a bare function.
Handler() http.Handler	Build the fully-wrapped handler (the WithTelemetry middleware → Recovery → CORS → global chain → mux, so a preflight never reaches a Use middleware; the health probes are answered ahead of it, wrapped in Recovery only). Use it with httptest.NewServer(srv.Handler()) to exercise the full stack without binding a port.
Start(addr string) error	Bind and serve until Stop; returns nil after a graceful Stop. Request headers must arrive within 10s, and idle connections close after 120s.
Serve(lis net.Listener) error	Start on a listener you bound, which it closes when it returns; binding first lets the caller log the address it holds, the real port of a :0 included.
Stop(ctx context.Context) error	Graceful shutdown. A Stop that comes before Start or Serve makes them return nil at once, as http.Server.Shutdown does.
Mux() *http.ServeMux	The underlying mux, if you need raw access. Routes registered on it directly skip the default body cap and handler timeout that Handle applies.
RegisterDocs(opts DocsOptions) error	Serve opts.Spec at GET SpecPath (/openapi.yaml) and a docs page loading the UI (redoc, swagger or scalar) at GET Path (/docs); nothing when Spec is empty. Fails, registering neither, on a path that does not start with / or that http.ServeMux refuses, on the two being equal, and on one a health probe or a registered route already serves.
ServeDocs(opts DocsOptions) *Server	RegisterDocs for a chain: panics with its error.

pattern is Go 1.22+ syntax: "GET /users/{id}".

Configuration setters

Each setter returns *Server for chaining, except the deprecated SetJSONCodec and SetStrictJSON, which return an error.

Method	Description
SetLogger(l log.Logger) / Logger() log.Logger	SetLogger installs l as log.Default() (nil is ignored), the logger the server's Recovery and the generated logic write to; the server keeps none of its own. Logger returns log.Default() as it is at the call; AccessLog(log.Follow()) follows a later SetLogger.
SetJSONCodec(c JSONCodec) error / Codec() JSONCodec	Deprecated: call server.SetGlobalJSONCodec(c) / server.JSON(), which they call. The codec is the process's, not one server's: setting it on one Server sets it for every server in the process. See JSON codec.
SetStrictJSON(strict bool) error	Deprecated: call server.SetStrictJSON(strict), which it calls. Strict decoding is the process's, not one server's: a second server switching it off switches it off for the first. See JSON codec.
SetCORS(opts CORSOptions)	Install CORS. Calling twice replaces the previous config. Panics on credentials allowed to the * origin.
SetHandleNotFound(h http.Handler)	Answer the requests the mux would answer 404, which otherwise get 404 {"message":"not found"}; nil restores that default. A method mismatch keeps its 405 {"message":"method not allowed"} with Allow.
SetDefaultReadTimeout(d) / SetDefaultWriteTimeout(d)	The *http.Server's deadlines for reading a whole request, 30s by default (server.readTimeout in config.yaml), and for writing a whole response, none (0) by default so streaming and long downloads are not cut. 0 sets none.
SetDefaultMaxBodySize(bytes)	Body cap for each route Handle registers afterwards, unless its WithLimits sets one (a method's @maxBodySize). DefaultMaxBodySize, 10 MiB, by default (server.maxBodySize in config.yaml); 0 sets none.
SetDefaultHandlerTimeout(d)	Request-context deadline for each route Handle registers afterwards, unless its WithLimits sets one (a method's @timeout). 0, the default, sets none.
SetDefaultMaxHeaderSize(kb)	The *http.Server's cap on request headers, in kilobytes; 32 by default.

Health checks

srv.RegisterHealthCheck("db", 2*time.Second, func(ctx context.Context) error {
    return db.PingContext(ctx)
})

RegisterHealthCheck(name, timeout, fn) adds, or replaces, the probe name on /readyz. Each run calls fn under context.WithTimeout: a check that returns the context's error on expiry fails (context deadline exceeded under checks), while one that ignores its context is waited for and judged by what it returns. A zero or negative timeout sets no deadline: fn runs under the probe request's context alone. A check that panics fails too, reported as panic: <value> under checks, and the panic is logged with its stack. A failure answers 503 {"checks":{…},"status":"not_ready"}. /healthz (liveness) always returns 200 once the process is up.

Both probes are answered ahead of the middleware chain (only Recovery wraps them): they are never access-logged, traced, counted in the HTTP metrics or CORS-processed, and no srv.Use middleware runs for them. WithoutDefaultHealth() removes them; register your own route for observed probes.

Middleware

Middleware is a defined type over the standard shape, so a func(http.Handler) http.Handler value is assignable to it as is:

type Middleware func(http.Handler) http.Handler

Built-in middleware

Constructor	Purpose
Recovery(logger)	Converts a panic into 500 {"message":"internal server error"} and logs it (panic recovered) with its stack. Once the response has started it logs the panic and aborts the connection instead, so the client sees the response cut off. A panic with http.ErrAbortHandler goes on to net/http, which aborts the connection without logging it. Handler() installs it itself, logging to log.Default(): outside every Use middleware and inside the WithTelemetry middleware, so the panic line carries the request's trace ids.
AccessLog(logger, opts...)	One http access line at Info per request: method, path, status (499 for a client that left before anything was written), latency, plus the trace_id / span_id an outer tracing middleware (WithTelemetry) put on the context. A request whose handler panics gets no line; Recovery logs the panic instead. AccessLogSkipPaths(paths...) keeps chosen routes out; AccessLogFields(fn) appends fields fn derives from the request (client address, user agent, the matched r.Pattern).
BodyLimit(maxBytes)	Caps request bodies at maxBytes: a declared Content-Length above it is answered 413 {"message":"request entity too large"} before the handler runs, and a read past it fails with an *http.MaxBytesError, which WriteValidationError answers 413 the same way without calling the validation hook.
Timeout(d)	Deprecated: use srv.SetDefaultHandlerTimeout(d), or WithLimits for one route. Runs the handler under http.TimeoutHandler: 503 on deadline, and a buffered response that cannot flush.

WithLimits(h, Limits{...}) applies timeout + body limits to a single handler - this is what @timeout / @maxBodySize compile to. Handle moves them, and the defaults standing in for those a route leaves unset, outside the route's own middlewares, so those are held to them too; srv.Use middlewares stay outside.

Chain

Chain composes middlewares outermost-first without nesting calls:

type Chain []Middleware

base := server.NewChain(server.BodyLimit(1 << 20), server.AccessLog(logger))
authed := base.Append(authMiddleware)          // returns a NEW chain (value semantics)

srv.Handle("GET /me", authed.Then(meHandler))  // Then folds the chain over the handler
srv.Handle("GET /ping", base.ThenFunc(pingFn)) // ThenFunc for bare functions

NewChain(A, B, C).Then(h) yields A(B(C(h))) - a request flows A → B → C → h, the response leaves in reverse. Nil entries are skipped, so an optional middleware can sit in the slice without an if != nil guard.

DSL-driven middleware

@middlewares(Name) in the DSL resolves at compile time, not by name at runtime. Codegen adds a Name field of type server.Middleware to the Middlewares struct embedded in ServiceContext and a gen-once constructor in internal/middleware/<name>_middleware.go; main.go assigns svc.Name = middleware.NewNameMiddleware(), and the generated routes.go passes the fields to srv.Handle as per-route middlewares.

RegisterMiddleware(name, mw) and With(names, h), a registry keyed by name, are deprecated: nothing generated calls them. Pass the middleware to Handle or Use, or build a Chain.

JSON codec

Swap the JSON implementation process-wide (e.g. for sonic or goccy/go-json):

type JSONCodec interface {
    Encode(w io.Writer, v any) error
    Decode(r io.Reader, v any) error
}

// Optional: needed only while server.strictJSON is on.
type StrictDecoder interface {
    DecodeStrict(r io.Reader, v any) error // rejects unknown fields and data after the value
}

if err := server.SetGlobalJSONCodec(myCodec{}); err != nil { /* strict JSON is on and myCodec has no DecodeStrict */ }
server.JSON().Encode(w, payload) // every generated handler reads through this accessor

server.SetStrictJSON(strict bool) error switches strict decoding: while on, JSON() decodes through the codec's DecodeStrict, which rejects a JSON body with an unknown field (400 {"message":"<field>: unknown field"}) or data after the JSON value (400 {"message":"body: unexpected data after the JSON value"}). It is off by default; server.strictJSON in config.yaml drives it, and the generated config.yaml sets it true.

The codec and the strict switch are the process's, not a Server's: generated handlers, error rendering and the health probes all go through server.JSON(), which has no server to ask, so every server in the process shares them. The Server methods SetJSONCodec, SetStrictJSON and Codec are deprecated wrappers of these functions.

A codec must honour the omitzero option of a json tag: generated types put it on each optional list, map and bytes member, which stays out of the body while nil. A codec that ignores the option writes such a member null, a value the OpenAPI document does not allow it. encoding/json, sonic from v1.15.0 (the first release that builds with Go 1.26) and goccy/go-json from v0.11.1 honour it; goccy/go-json v0.10.x and v0.11.0, jsoniter v1.1.12 (its last release) and segmentio/encoding v0.5.4 do not. A codec must also pass a null to a type's json.Unmarshaler, as each of these does, for the required-member check.

SetGlobalJSONCodec fails, keeping the previous codec, when strict JSON is on and c has no DecodeStrict; SetStrictJSON(true) fails the same way when the installed codec has none, so config.yaml can never claim a strictness the server does not enforce. The swap is atomic, so it is safe while requests are served.

The built-in codec names a body value of the wrong JSON type by its JSON path, with no array index or map key: name: expected string, got number, home.name: expected string, got number, age: expected integer, got number 1.5, age: 300 is out of range, m: "x" is not an integer (a map key), and body: expected object, got array at the root. A value a type decodes itself is named the same way: at: not a valid RFC 3339 datetime for a time.Time, data: not a valid base64 for a []byte, and another type's own error text after its path. An empty body is body: required. Each answers 400 {"message": …} through the validation hook, and reads the same whichever encoding/json engine the Go toolchain builds; the decoder's error stays in the chain for errors.As. A codec installed with SetGlobalJSONCodec reports its own errors.

A wrapper for another JSON library adds DecodeStrict with that library's own unknown-field switch and finishes with server.TrailingData, the shared check every codec uses to report leftover data the same way. It reads the rest of the body, so it also returns a failed read, such as the *http.MaxBytesError of a body past its cap, which answers 413. With sonic:

type Sonic struct{}

func (Sonic) Encode(w io.Writer, v any) error { return sonic.ConfigDefault.NewEncoder(w).Encode(v) }
func (Sonic) Decode(r io.Reader, v any) error { return sonic.ConfigDefault.NewDecoder(r).Decode(v) }

var strictAPI = sonic.Config{DisallowUnknownFields: true}.Froze()

func (Sonic) DecodeStrict(r io.Reader, v any) error {
    dec := strictAPI.NewDecoder(r)
    if err := dec.Decode(v); err != nil {
        return err
    }
    return server.TrailingData(dec.Buffered(), r)
}

goccy/go-json (NewDecoder(r).DisallowUnknownFields(), the stdlib API) wraps the same way.

Cookies

A response's @cookie field is set by the generated handler with http.SetCookie, an error's by WriteError from its ResponseCookies(); each cookie carries the attributes its @cookie declares (Cookie attributes), and a cookie field's is the one the logic set.

Type	Description
ResponseCookieSetter	ResponseCookies() []*http.Cookie: what a generated error with a @cookie field implements; WriteError sets each cookie as it returns it, after WriteResponseHeaders.

Multipart forms

A generated multipart handler parses the form with r.ParseMultipartForm(server.MultipartMemory): up to 32 MiB of its files in memory, whatever the route's @maxBodySize, the rest in temporary files removed when the handler returns. Before that it calls server.LimitParts(r, limits) with each file part's @maxSize by form name: a part read past its limit fails the read there, and WriteValidationError answers 400 {"message":"<name>: file size exceeds <n> bytes"}. A body of another media type is answered 415.

Required members

A generated handler checks that a JSON body carries every member declared without ?. It decodes the body with the installed codec into an unexported struct beside the handler, a pointer per member, and copies it into the request type; a member left out, or sent as null without @nullable, answers 400 {"message":"<json path>: required"} through the validation hook, and the @default of a member left out or sent as null is filled at any depth. A @nullable member decodes through server.Nullable[T], whose UnmarshalJSON tells an absent member from a null one, so a codec installed with SetGlobalJSONCodec must call a type's json.Unmarshaler, as sonic, jsoniter, goccy/go-json and segmentio/encoding do. It must also decode [], {} and "" to a non-nil list, map and bytes, since nil stands for an absent one.

Function	Description
MissingMember(path) error	The error "<path>: required" a handler reports for a member the body lacks.
Nullable[T]	A member that may be null: Present() reports that it came, Value() returns it, nil when absent or null.

Validation error hook

Generated handlers route req.Validate() failures through one global, swappable hook:

type ValidationFailedHandler func(w http.ResponseWriter, r *http.Request, err error)

server.SetDefaultValidationFailed(func(w http.ResponseWriter, r *http.Request, err error) {
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusBadRequest)
    json.NewEncoder(w).Encode(map[string]string{"error": err.Error()})
})

The handler calls server.WriteValidationError(w, r, err) on a decode, binding or validation failure; it dispatches to your installed hook, or to the default: 400 {"message":"<err text>"}. A body read past its cap is answered 413 {"message":"request entity too large"}, http.ErrNotMultipart, a form route's body of another media type, 415 {"message":"unsupported media type"}, and a body the connection failed to deliver 408 {"message":"request timeout"} for the read timeout or 400 {"message":"bad request"} otherwise, the failure's detail logged at Warn, all without the hook. WriteError answers the same when a raw request's logic returns such a read's error. The default is post-commit safe - if the response already started, it logs the dropped validation rather than smearing a 400 into a half-sent body; a hook you install is called even after the response started.

Before it reads a JSON body, the handler calls server.RequireJSONBody(w, r): a Content-Type other than application/json or an application/*+json type, parameters and letter case aside, is answered 415 {"message":"unsupported media type"}, and so is a body without a Content-Type, so the bodies a cross-site HTML form, a text/plain fetch or a fetch/sendBeacon of an untyped Blob or an ArrayBuffer can send without a CORS preflight never reach logic. A request without a Content-Type passes only when it has no body, which the decoder then reports missing. A @rawRequest or @passthrough handler reads its own body and checks nothing.

Raw responses

Helpers for @rawResponse / @passthrough handlers that already hold the bytes they want on the wire:

Function	Description
WriteResponse(w, r, status, v, setHeaders...)	Encodes v as it is with the installed codec before writing - a nil list or map is null, and v is only read, so a value shared between requests is safe - then sets Content-Type: application/json; charset=utf-8, writes the status and the body; a value the codec cannot encode (a NaN float) goes to WriteError as an unhandled error, a 500. Each setHeaders runs once v has encoded, before the status, to set the response's @header and @cookie values, so a failed encode never carries them. Generated handlers answer through it.
WriteBytes(w, status, contentType, body) error	Sets Content-Type (when non-empty) and Content-Length, writes the status, writes body.
WritePrecompressed(w, r, status, contentType, coding, body, decode) error	Serves a body stored already compressed ("gzip", "zstd", "br", ...). When the client accepts coding the bytes go out verbatim with Content-Encoding; otherwise decode produces the identity form first. Always adds Vary: Accept-Encoding. A nil decode with a client that does not accept the coding returns ErrNoDecoder before anything is written.
AcceptsEncoding(r, coding) bool	Whether the request's Accept-Encoding lists coding with a non-zero quality (gzip;q=0 is a refusal). Shares its parser with the Compress middleware.

func (l *SnapshotService) Snapshot(w http.ResponseWriter, r *http.Request, req *types.SnapshotReq) error {
	region := "global"
	if req.Region != nil {
		region = *req.Region
	}
	blob, ok := snapshotCache[region] // gzip bytes, exactly as stored
	if !ok {
		http.NotFound(w, r)
		return nil
	}
	return server.WritePrecompressed(w, r, http.StatusOK, "application/json; charset=utf-8", "gzip", blob, gunzip)
}

The framework pulls in no compression library: you supply decode, using the library that filled the cache. Compress leaves a response that already carries Content-Encoding untouched, so a verbatim body is never re-encoded.

server.WriteError, and server.WriteValidationError with the default hook, are post-commit safe: when a raw handler has already written a status or body, or flushed, and then returns an error, the error is logged with the request's trace context and the wire is left alone rather than splicing an envelope into the body. A context error is dropped without a line, except a dependency's deadline on a live request, logged at Warn.

CORS

srv.SetCORS(server.CORSPermissive())          // dev: any origin (Access-Control-Allow-Origin: *)
srv.SetCORS(server.CORSStrict("https://app")) // prod: one allowed origin

Or build a CORSOptions value directly for fine control over methods, headers, credentials, and max-age. Every response carries Vary: Origin, since whether the CORS headers are there depends on the request's Origin. SetCORS panics on AllowCredentials with the * origin, a pair browsers refuse: list the origins, or use a pattern such as https://*.example.com, which echoes the request's origin.

Event runtime

Generated event code runs on github.com/craftgodotdev/craftgo/pkg/events, the transport- and codec-neutral counterpart of pkg/server. It knows nothing about any broker and nothing about any serialisation format.

type Message struct {
	Event          string            // the contract name the design declared
	Key            string            // the WithKey value, "" for a keyless message
	DedupID        string            // the WithDedupID value; a transport without the notion ignores it
	Payload        []byte            // already encoded
	Metadata       map[string]string // side-band values a publisher sets; MetaCodec always present
	AdapterOptions map[string]map[string]any // what WithAdapterOption filled in, per adapter
}

func (m *Message) AdapterOption(adapter, key string) (any, bool)

type Codec interface {
	Name() string
	Marshal(v any) ([]byte, error)
	Unmarshal(data []byte, v any) error
}

type Publisher interface {
	Publish(ctx context.Context, msg *Message) error
}

type Subscriber interface {
	Subscribe(ctx context.Context, subs []Subscription) error
}

type Handler func(ctx context.Context, msg *Message) error

type Group string // the broker identity: Kafka group, NATS queue group, JetStream durable

type Subscription struct {
	Event    string // the contract, matching Message.Event
	Consumer string // names the handler in diagnostics and Plan; Event[T].Subscription sets it to the contract
	Group    Group  // the broker identity; Register refuses an empty one
	Chain    Chain  // this subscription's own middleware, applied inside the bus chain
	Handle   Handler
	Batch    *Batch // in place of Handle and Chain: consume in batches, see Batches
}

Subscribe takes the whole batch, and there is no one-at-a-time path beside it. A broker that binds one identity to several contracts - a JetStream durable filtering every subject its group consumes - cannot register a group one contract at a time, so every adapter is handed the set. It registers and returns; it must not block. A push transport hands the handler its callback, a pull transport starts its own loop; delivery runs until ctx is cancelled. A handler error does not decide what becomes of the message: the chain does, through the disposition it asks for. Once the handler returns, the transport answers the delivery with that disposition, and settles it when nothing was asked or the transport cannot honour what was.

The exception is the shutdown. Once ctx has ended, a transport hands no message to a handler, and a delivery whose handler failed after ctx ended, with the chain asking for nothing or for redelivery, was most likely cut short by it: a transport that can hand that message back to its broker does so rather than settle it. What each shipped transport does is in Shutting down.

Group is a named type so an application declares its groups once and passes them around as values rather than as loose strings. It has no fallback: a subscription without one is refused at registration, because the name is where a consumer resumes on a broker that keeps a position, and that is the deployable's to choose rather than something to default into.

events.New(opts...) builds a *Bus binding one transport to one codec; WithPublisher / WithSubscriber / WithTransport install the transport, WithCodec the codec, WithCodecFor(contract, c) overrides it for a single contract, WithMiddleware(mws...) installs the consumer chain, WithPublishDefaults(opts...) sets the publish options every message through this bus starts from, and WithDispositionRequired(d) refuses a transport that cannot honour a disposition. There is no default codec - a bus built without one fails rather than picking an encoding.

Register and start

func (b *Bus) Register(sub Subscription) error
func (b *Bus) Start(ctx context.Context) error

type RegisterError struct {
	Event    string
	Consumer string
	Group    Group
	Err      error // one of the sentinels; errors.Is reaches it
}

Register records one subscription, to be handed over by Start. It makes the checks the bus can make on its own: the bus has not started, Handle is non-nil, Group is non-empty, a codec resolves for the contract, the transport can honour every disposition WithDispositionRequired named, and no earlier subscription holds the same contract and group. A refusal is a *RegisterError carrying the offending subscription, so errors.Is(err, events.ErrNoGroup) reaches the reason while the message still names which consumer. Whether the broker accepts the set is the transport's answer, and it comes from Start.

A module states its whole consumption as errors.Join over one Subscribe line per contract. Every line is offered to the bus that way - a refusal in the middle stops neither the registrations after it nor the refusals beside it - and the joined error carries each *RegisterError, so a deployable wired wrongly in two places hears about both at once. Nothing has started either way, so a caller returning the error abandons the bus.

Start hands every registered subscription to the transport in one call, each handler already wrapped: a recover outermost, then the bus-wide chain, then the subscription's own Chain innermost. The batch is sorted by group, contract then consumer. Group leads because it is the identity a transport can refuse a second claim on, so the order decides which of two colliding subscriptions registers first; contract and consumer complete it, since two services may name a consumer alike and an order that is not total would move the refusal between runs.

A second Start, or a Register after one, is ErrStarted - including after a Start that failed, since the transport may have taken part of the batch before it did. A bus with nothing registered starts successfully and hands the transport nothing.

Event descriptors

A generated contract package declares one descriptor per event, and everything typed about that event goes through it:

type Event[T any] struct{ ... }

func NewEvent[T any](contract string, validate func(*T) error) Event[T]

func (e Event[T]) Contract() string
func (e Event[T]) Publish(ctx context.Context, bus *Bus, payload *T, opts ...PublishOption) error
func (e Event[T]) Handler(bus *Bus, fn func(ctx context.Context, payload *T) error) Handler
func (e Event[T]) Subscribe(bus *Bus, group Group,
	fn func(ctx context.Context, payload *T) error) error
func (e Event[T]) Subscription(bus *Bus, group Group,
	fn func(ctx context.Context, payload *T) error) Subscription

validate may be nil, for a payload type that carries none. Publish validates first and sends nothing when that fails - the contract is refused where it is broken rather than at every consumer. Handler adapts a typed function to the untyped one a transport delivers to, decoding and validating before it runs.

Subscribe is one line of an application's consumption: it builds the subscription and registers it, so a module is errors.Join over its lines. It takes neither a consumer nor a chain - the middleware every handler runs behind belongs on the bus through Use, and Consumer defaults to the contract.

Subscription returns the value instead of registering it, for the line that needs one of those two fields set - two subscriptions of one contract to tell apart, one handler to wrap alone. Set the field, then hand the value to Register.

The bus is a parameter at every call and never a field: a descriptor is a value in a contract package and knows nothing about how any deployable is wired, so one contract catalogue serves every binary that imports it.

Batches

type BatchHandler func(ctx context.Context, batch []*Message) error
type BatchMiddleware func(sub Subscription, next BatchHandler) BatchHandler
type BatchChain []BatchMiddleware

func NewBatchChain(mws ...BatchMiddleware) BatchChain
func (c BatchChain) Append(mws ...BatchMiddleware) BatchChain

type BatchSize struct {
	Max  int           // at most this many messages; Register refuses below 1
	Wait time.Duration // no message waits longer for its batch to fill; Register refuses <= 0
}

type Batch struct {
	BatchSize
	Chain  BatchChain   // this subscription's own batch middleware
	Handle BatchHandler // wrapped in a recover by Start
}

type BatchSubscriber interface {
	SubscribesBatches() bool
}

type ItemErrors map[int]error // index into the batch -> that message's failure

func WithBatchMiddleware(mws ...BatchMiddleware) Option
func (b *Bus) UseBatch(mws ...BatchMiddleware)

type Item[T any] struct {
	Payload *T
	Msg     *Message
}

func (it Item[T]) Fail(err error)
func (e Event[T]) BatchHandler(bus *Bus, fn func(ctx context.Context, batch []Item[T]) error) BatchHandler
func (e Event[T]) SubscribeBatch(bus *Bus, group Group, size BatchSize,
	fn func(ctx context.Context, batch []Item[T]) error) error
func (e Event[T]) BatchSubscription(bus *Bus, group Group, size BatchSize,
	fn func(ctx context.Context, batch []Item[T]) error) Subscription

A subscription with Batch set, in place of Handle and Chain, consumes in batches. Register refuses it with ErrInvalidBatch for a Max below 1, a Wait that is not positive, or a Handle or Chain set beside it; with ErrBatchUnsupported on a transport that is not a BatchSubscriber answering true (a bus with no subscribe half leaves the refusal to Start); and with ErrBatchGroupShared when a subscription of another contract holds its group, as it refuses one of another contract in a batch subscription's group. The same contract twice in a group is ErrDuplicateSubscription, as ever.

Start wraps Batch.Handle as it wraps a Handle: a recover outermost, the bus batch chain (WithBatchMiddleware, UseBatch), Batch.Chain, and a recover innermost. The per-message chain does not run for a batch. A panic in the batch chain asks for every message of the batch to be redelivered where the transport can honour it; one in the handler leaves each message undecided.

A BatchSubscriber gathers a subscription's messages into batches of its BatchSize, hands each to Batch.Handle, and answers every message of it per its disposition once Handle returns - the Subscriber contract, per message. It reports an error Handle returns once, with no message.

BatchHandler decodes and validates each message as Handler does. A message that fails is left out of fn's batch, and fn does not run when none is left. fn's error goes to every item not failed with Item.Fail, unless it is an ItemErrors keyed by index into the items fn was handed: that fails only the items it names, and is taken for the whole batch's error when it names an index outside them. The failures come back as ItemErrors, keyed by index into the whole batch; its Unwrap lists them in index order, so errors.Is and errors.As reach each.

The plan

func (b *Bus) Plan() Plan

type Plan struct {
	Groups []PlanGroup `json:"groups"`
}

type PlanGroup struct {
	Name      Group          `json:"name"`
	Consumers []PlanConsumer `json:"consumers"`
}

type PlanConsumer struct {
	Event    string     `json:"event"`
	Consumer string     `json:"consumer"`
	Batch    *BatchSize `json:"batch,omitempty"` // a batch consumer's bounds: {"max":100,"wait":"1s"}
}

func (p Plan) MarshalJSON() ([]byte, error)

Plan reports what is registered, before or after Start: groups ordered by name, consumers within a group by contract then consumer, so two runs of the same wiring produce the same plan. MarshalJSON renders that order rather than the one the plan was built in, so a golden file compares a plan and not a map iteration. No generated file states a deployable's consumption - the groups are the application's - so a project that wants it stated pins this in a test.

Publishing

type PublishOption func(*Envelope)

func WithKey(key string) PublishOption
func WithDedupID(id string) PublishOption
func WithHeader(key, value string) PublishOption
func WithAdapterOption(adapter, key string, value any) PublishOption

func JoinOptions(defaults, opts []PublishOption) []PublishOption
func (env *Envelope) Apply(opts ...PublishOption)

type Envelope struct {
	Event          string            // the contract
	Key            string            // the WithKey value; "" keeps a default key, else is keyless
	DedupID        string            // the WithDedupID value
	Payload        any               // the value to encode
	Metadata       map[string]string // optional side-band values; nil and empty behave alike
	AdapterOptions map[string]map[string]any
}

func (b *Bus) Publish(ctx context.Context, event string, payload any, opts ...PublishOption) error
func (b *Bus) PublishAll(ctx context.Context, envs []Envelope) error

type EnvelopeError struct {
	Index int    // the envelope's position in the batch
	Event string // its contract, empty when it has none
	Err   error  // why: the codec's error, ErrNoCodec, an *UnknownOptionError, ...
}

Options apply in order, so the last one setting a given value wins - which is what makes a defaults list defaults. WithPublishDefaults is that list for a whole bus. JoinOptions does the same join for a hand-written publisher carrying its own - defaults first, the per-call options after, without writing into either. An empty value clears a default only as a per-call option: WithKey("") on Publish clears a default key, while an Envelope whose Key is empty keeps it.

PublishAll encodes every envelope up front, so one it cannot encode - no contract, no codec, an option its adapter does not read, a payload the codec refuses - refuses the whole batch, before any of it is sent, with an *EnvelopeError whose Index names it: an outbox drainer sets that row aside and sends the rest. A transport that cannot carry an envelope refuses the batch the same way. Then PublishAll hands the batch to the transport in one call when it implements BatchPublisher and one message at a time otherwise; a failure partway through returns a *PartialPublishError whose Unsent holds the indices, ascending, of the envelopes that did not go out, so retrying exactly those sends nothing twice. It is a set, not a count, because a transport publishing to several partitions at once does not fail in batch order - the ones that landed need not be the first. Sent is the length of the leading published run, which is Unsent[0].

An adapter builds its report through one of two constructors, whose shapes say which kind of failure it had - and whose signatures make the wrong one a compile error rather than a wrong report:

func UnsentFrom(i int, msgs []*Message, err error) *PartialPublishError
func UnsentAt(indices []int, msgs []*Message, err error) *PartialPublishError

UnsentFrom is for a transport that STOPPED at i, so everything from i onward is unsent - the shape a one-at-a-time loop produces. UnsentAt is for one whose failures are SCATTERED, which is what publishing to several partitions at once produces. Both sort the indices and derive Sent and Event from the first, so the invariants hold by construction.

PublishAll checks the adapter's report against the batch before returning it. One that names an index outside the batch, or out of order, is replaced with "none of it was sent" and the error names the adapter: an understated Unsent loses the messages it calls delivered, and nothing downstream can tell.

Dispositions

type Disposition uint8

const (
	DispositionUnset Disposition = iota // nothing decided; settles
	DispositionSettle
	DispositionRedeliver
	DispositionReject
)

func (m *Message) Settle()
func (m *Message) Redeliver()
func (m *Message) Reject()
func (m *Message) Disposition() Disposition
func (m *Message) Deliveries() int   // the broker's count; 0 where it keeps none
func (m *Message) SetDeliveries(n int) // transport adapters only

type Dispositioner interface {
	CanDisposition(d Disposition) bool
}

func WithDispositionRequired(d Disposition) Option
var ErrDispositionUnsupported = errors.New(...)

A middleware asks for something other than "done" through the message. Options apply in chain order and the last writer wins: the chain returns innermost first, so the outermost middleware decides last. A frame that panicked did not finish deciding, so the recover clears what it asked for; Recovery says what is asked in its place.

Dispositioner is asked per INSTANCE, not per type: one adapter may be built in a mode that can redeliver and in a mode that cannot. A transport that does not implement it honours settle alone. WithDispositionRequired refuses at Register rather than at the first message, because a chain calling Redeliver() on a transport that settles instead loses every message it meant to retry with nothing to report it.

Per-adapter options

type OptionAware interface {
	AdapterName() string   // the namespace WithAdapterOption addresses
	KnownOptions() []string // every option key this adapter reads
}

type UnknownOptionError struct {
	Adapter, Key, Event string
	Known               []string
}

A transport implementing OptionAware gets both halves of the rule, enforced by the bus before anything is encoded: an option under another adapter's name is ignored, and one under its own name that KnownOptions does not list fails the publish with an *UnknownOptionError. A transport that does not implement it gets neither - nothing can tell an option meant for it from one meant for somebody else. Every adapter craftgo ships implements it; kafka.OptionTimestamp (a time.Time) is the only option any of them reads.

const MetaCodec  = "content-codec" // the codec that encoded the payload
const MetaPrefix = "craftgo-"      // reserved for transport adapters

func IsReservedMeta(key string) bool

IsReservedMeta names every key the caller does not own, case-insensitively: MetaCodec, and every key under MetaPrefix, which holds the adapter headers - Kafka's craftgo-event, craftgo-key and craftgo-dedup-id, NATS's Craftgo-Key. An Envelope.Metadata entry under one of those is dropped silently and the runtime's or the adapter's own value takes its place. Every other key is carried untouched, except that the NATS adapters never let one override Nats-Msg-Id, the header that carries the dedup ID. NATS headers cannot hold everything, so the NATS adapters refuse, with nats.ErrNotCarried and before anything is sent, a metadata key that is no NATS header name (printable ASCII without spaces or any of "(),/:;<=>?@[\]{}) and a key, dedup ID or metadata value that NATS would change on the way - one padded with spaces or tabs, or holding a line break - rather than deliver it altered; in a batch the refusal is an *EnvelopeError naming that message. A generated consumer is handed the decoded payload, so metadata is read in a middleware or a hand-written Subscription, both of which are handed the Message.

Recovery

type PanicError struct {
	Event    string // the contract being delivered
	Consumer string // the consumer whose handler or chain panicked
	Group    Group  // its broker identity
	Value    any    // what was passed to panic
	Stack    []byte // the trace where the panic fired; not part of Error()
}

Bus.Start wraps every handler it hands over in a recover before the batch reaches the transport, so a panicking consumer cannot end the process - the same rule Recovery is for the HTTP chain, and every transport inherits it, including adapters written outside craftgo. With a chain installed the wrap goes on both sides of it, so the chain observes the panic and a panic in the chain is caught too. The recovered panic is returned as a *PanicError, which the transport sees as an ordinary handler error: it reaches the error handler installed on the transport, and delivery continues with the next message. A panicking handler leaves the disposition unset, not settled, so the chain above it decides as it does for any error; a panic in the chain itself, with nothing above it to decide, asks for redelivery where the transport can honour one.

*PanicError is a concrete type, so errors.As picks one out of a chain, and its Unwrap reaches the panic value when that value is an error.

type PayloadError struct {
	Event string // the contract the payload arrived on
	Err   error
}

var ErrCodecMismatch = errors.New(...)

A payload Event[T].Handler could not decode or that failed its Validate() comes back as a *PayloadError - the same bytes fail the same way on every delivery. A message stamped with a codec the consumer is not configured for fails with ErrCodecMismatch instead, a configuration error rather than a poison payload. Nothing else is classified: what to do with a failure is a middleware's decision.

Consumer middleware

type Middleware func(sub Subscription, next Handler) Handler
type Chain []Middleware

func WithMiddleware(mws ...Middleware) Option // install on the bus at construction
func (b *Bus) Use(mws ...Middleware)          // append to the same chain afterwards

func NewChain(mws ...Middleware) Chain
func (c Chain) Append(mws ...Middleware) Chain
func (c Chain) Apply(subs []Subscription) []Subscription
func Recover() Middleware

WithMiddleware installs the chain every subscription registered through the bus is wrapped in, outermost first. The bus is the seam that covers all of them because every subscription passes through Bus.Register - a descriptor's Subscription, a hand-built one, and one built against another design's contracts alike. A subscription's own Chain is applied inside this one, so a bus-wide concern - logging, tracing - still sees what a per-consumer chain did. Repeating the option appends. Nothing is generated for any of it.

Bus.Use(mws...) appends to that same chain after construction, for a deployable whose delivery chain is assembled out of a service context or out of configuration rather than held back until the New call. Use after Start panics: the batch has gone to the transport with its handlers already wrapped, so a middleware arriving then would cover nothing at all and say nothing about it - a wiring mistake to fix in the code, like net/http's ServeMux on a duplicate pattern, rather than an ErrStarted for the caller to handle.

Each middleware is handed the Subscription it wraps, so one chain can read the contract, the consumer and the group it is running for.

Chain is server.Chain for the consumer side and folds the same way: NewChain(A, B, C) makes A outermost, Append returns a new chain without mutating the receiver, and nil entries are skipped. The verb differs - Apply over a slice of subscriptions rather than Then over one handler - because each wrap is handed the subscription it wraps. Most projects never call Apply; it is for decorating a slice the bus will not see, or one slice differently from the rest.

Bus.Start recovers on both sides of the chain: the inner recover turns a panicking handler into a *PanicError your middleware observes as an ordinary error, the outer one catches a panic in the chain itself. Exactly one PanicError is built per panic. A bus with no middleware installs the inner one alone.

Recover() is for a chain folded by Apply instead of installed on the bus - the bus wraps that from outside as one opaque handler, so place Recover() at its innermost end for the same visibility. A bus chain needs it nowhere. See Events.

Shipped implementations

• pkg/events/memory - an in-process transport for tests, local development and single-binary deployments. Subscriptions sharing a group for one contract form one competing-consumer group, and each group gets its own copy of a message, payload included. A handler runs on the context its subscription was given, as on a broker, so it carries that context's values and sees it end. Drain() waits for in-flight deliveries, and is safe to call while another goroutine publishes - which is what a shutdown racing a request still in flight looks like.
• pkg/events/nats - core NATS, where a contract is a subject and a group is a queue group, plus a separate JetStream transport binding one durable per group, filtered to every subject that group consumes. See NATS JetStream.
• pkg/events/kafka - one contract per topic by default, the publish key as the record key. A classic consumer group by default; WithShareGroup asks for a KIP-932 share group, which is what makes Redeliver and Reject mean anything there.
• pkg/events/codecjson - a JSON codec.
• pkg/events/logging - AccessLog(l *slog.Logger, opts ...AccessLogOption), one line per delivery, and BatchAccessLog with the same options, one line per batch. It is a sub-package so log/slog stays out of the exported surface of pkg/events, which every generated contract package imports.

Every transport above is a BatchSubscriber; how each gathers a batch is in Kafka, core NATS and memory and NATS JetStream.

Any other broker - RabbitMQ, SQS, Pub/Sub, Redis Streams - is an outside package implementing Publisher and/or Subscriber, and those two interfaces are all it needs: none of the above is privileged. See Events.

gRPC runtime

The generated gRPC layer runs on github.com/craftgodotdev/craftgo/pkg/rpc - a thin wrapper over *grpc.Server that installs the guards the HTTP server installs. The generated main.go wires it; see gRPC.

grpcSrv := rpc.New(svcCtx, opts...)

Option	Effect
WithStatsHandler(h stats.Handler)	Install a stats handler - tel.GRPCServerHandler() - which runs in the transport, ahead of every interceptor. A nil handler is ignored.
WithReflection(on bool)	Serve the gRPC reflection service.
WithoutDefaultHealth()	Disable the auto-registered grpc.health.v1 service.
WithServerOptions(opts ...grpc.ServerOption)	Pass options straight to grpc.NewServer - message size limits, keepalive, credentials.

Method	Description
Use(i Interceptor) *Server	Append an interceptor to the chain, outermost first. Recovery is always ahead of the chain, and health and reflection calls bypass it. A Use after the server is built has no effect.
SetLogger(l log.Logger) / Logger() log.Logger	SetLogger installs l as log.Default() (nil is ignored), the logger the server's Recovery and Error write to; Logger returns log.Default() as it is at the call; rpc.AccessLog(log.Follow()) follows a later SetLogger. The server keeps none of its own.
RegisterService(desc *grpc.ServiceDesc, impl any)	grpc.ServiceRegistrar, so pb.RegisterXServer(srv, impl) takes the server directly.
GRPCServer() *grpc.Server	Build the underlying server once and return it.
Start(addr string) error	Listen and serve until Stop; returns nil once stopped.
Serve(lis net.Listener) error	Serve on a listener you own - a bufconn in tests.
Stop(ctx context.Context) error	Health flips to NOT_SERVING, in-flight RPCs finish, and when ctx expires first the rest are cut off and ctx.Err() is returned. A Stop that comes before Start or Serve makes them return nil at once, as http.Server.Shutdown does.

Interceptor pairs the two shapes gRPC needs, Unary grpc.UnaryServerInterceptor and Stream grpc.StreamServerInterceptor; rpc.Unary(f) and rpc.Stream(f) wrap one side.

Function	Description
Recovery(logger)	Panic → Internal with an opaque message, logged with the stack and the call's trace ids. Installed outermost by the server itself.
AccessLog(logger, opts...)	One grpc access line per call with method, code, latency and the trace ids; a stream logs once when it ends, and a call that panics logs the Internal that Recovery answers. AccessLogSkipMethods(...) and AccessLogFields(fn) shape it.
Timeout(d)	Bound unary calls to d; a late response is dropped for DeadlineExceeded. Streams are not bounded; d <= 0 installs nothing.
Error(ctx, err) error	What the generated server layer returns on failure: status errors pass through, a craftgo typed error (HTTPStatus() int) becomes the matching code with an ErrorInfo{Reason: ErrCode()} detail, a context error becomes Canceled / DeadlineExceeded, anything else goes to the unknown-error handler.
SetHandleUnknownError(h)	Swap the process-wide handler for errors that carry neither a status nor an HTTP status; the default logs and answers Internal.
Validate(msg any) error	Run the message's own Validate() error (protoc-gen-validate) and answer InvalidArgument.
ValidateClientStream(stream) / ValidateBidiStream(stream)	Wrap a client-streaming or bidi server stream so that Recv runs Validate on every message it receives; a message that fails is consumed and Recv returns the InvalidArgument error.
IsInfrastructureMethod(fullMethod string) bool	Whether a full method belongs to the health or reflection services.

tel.GRPCServerHandler(), a method of the *telemetry.Telemetry that telemetry.Init returns, is the stats handler the generated main.go installs: one otelgrpc handler emitting the span and the rpc.server.call.duration histogram against the stack's providers, adopting the caller's W3C trace context from the request metadata, and leaving the health and reflection calls out.

Calling another service

conn, err := rpc.Dial(addr, rpc.WithClientStatsHandler(tel.GRPCClientHandler()))

Dial returns a *grpc.ClientConn with the guards a craftgo service makes calls under. It connects lazily, so an unreachable target is a failed call rather than a failed startup, and the caller closes it.

Option	Effect
WithClientStatsHandler(h stats.Handler)	The client span, the rpc.client.call.duration histogram, and - the reason a caller needs one - the W3C trace context written into the request metadata. A nil handler is ignored.
WithClientTimeout(d)	A default deadline for a unary call whose context carries none; a caller with its own keeps it. Streams are not bounded.
WithClientAccessLog(l log.Logger)	One grpc client line per unary call with method, code, latency and the trace ids.
WithClientTransportCredentials(c)	Transport security; the default is insecure.
WithDialOptions(opts ...grpc.DialOption)	Straight to grpc.NewClient.

tel.GRPCClientHandler() is the caller-side twin of GRPCServerHandler(). Without it a gRPC client sends no traceparent, so the service it calls starts a trace of its own.

Related packages

• pkg/log - the structured Logger interface and default zap-backed implementation. log.SetLevel(level) / log.GetLevel() retune the process-wide level (shared by the server and generated logic); log.SetDefault / log.Default swap or read the package-level logger. log.Follow() returns a logger that writes each line - and each line of the loggers its With and WithContext return - through the log.Default() of that moment. log.SetDefault given a Follow logger installs the logger it writes through at that call, its With fields and context included; a logger that writes through one, as a wrapper around it does, cannot be the default, since its lines would come back to it.
• pkg/telemetry - traces and metrics as one stack. telemetry.Init(ctx, cfg) builds the providers the otel: / metrics: blocks of config.yaml select (spans: none / stdout / otlp_grpc / otlp_http; metrics: prometheus / otlp_grpc / otlp_http / none; Init fails, naming the key, on an enabled signal's other exporter and on a metrics.path that does not start with /) and returns a *Telemetry, whose HTTPMiddleware() instruments every HTTP request, GRPCServerHandler() every RPC served and GRPCClientHandler() every RPC made, ScrapeURL() names the Prometheus listener and ScrapeHandler() serves the same scrape on a route of your own, and Shutdown flushes both signals. Generated main.go wires Init, HTTPMiddleware through server.WithTelemetry (or GRPCServerHandler through rpc.WithStatsHandler), ScrapeURL and Shutdown; GRPCClientHandler goes to the rpc.Dial calls you write, and ScrapeHandler to a route of yours.