benzene

README

benzene-go

A Go port of Benzene, a middleware-based library for hexagonal (ports-and-adapters) architecture: a pipeline of middleware wraps calls to "ports" (interfaces representing external boundaries - DB, HTTP, queues, etc), dispatched by topic to a registered handler.

This repo is conformant with the main repo's language-neutral specification - see conformance/ for the fixtures this port runs against. The spec, not this README, is the source of truth for cross-language behavior; when the two disagree, the spec wins and this repo has a bug.

Quickstart

package main

import (
	"context"
	"net/http"

	benzene "github.com/daniellepelley/benzene-go"
	"github.com/daniellepelley/benzene-go/httpbinding"
)

type greetRequest struct {
	Name string `json:"name"`
}

type greetResponse struct {
	Greeting string `json:"greeting"`
}

func greetHandler(_ context.Context, req greetRequest) benzene.Result[greetResponse] {
	return benzene.Ok(greetResponse{Greeting: "Hello, " + req.Name + "!"})
}

func main() {
	registry := benzene.NewRegistry()
	benzene.Register(registry, benzene.NewTopic("greet"), benzene.Handler[greetRequest, greetResponse](greetHandler))

	builder := &benzene.ApplicationBuilder{
		Registry:  registry,
		Container: benzene.NewContainer(),
		Pipeline:  benzene.NewPipeline(benzene.RouterMiddleware(registry)),
	}

	routes := []httpbinding.Route{{Method: http.MethodPost, Path: "/greet", Topic: benzene.NewTopic("greet")}}
	http.ListenAndServe(":8080", httpbinding.Handler(builder, routes))
}

This Quickstart wires the ApplicationBuilder directly, which is the shortest thing that runs. A real service usually wraps that wiring in the three-phase App[TConfig] lifecycle (GetConfiguration → ConfigureServices → Configure) instead — App.Run() produces the very same ApplicationBuilder, just with a place for configuration and dependency registration to live. See examples/helloworld/ for the complete version with dependency injection, a health check, and both HTTP entry points wired through that lifecycle.

Scaffold a new service

The templates/ directory holds starter projects - the dotnet new equivalent for the Go port, driven by gonew (golang.org/x/tools/cmd/gonew). gonew instantiates a project by copying a template module and rewriting its module path to the one you choose - no template engine, no placeholders to fill in beyond the module path.

# API Gateway-fronted Lambda
go run golang.org/x/tools/cmd/gonew@latest \
  github.com/daniellepelley/benzene-go/templates/aws-apigateway example.com/myservice

# SQS-triggered Lambda
go run golang.org/x/tools/cmd/gonew@latest \
  github.com/daniellepelley/benzene-go/templates/aws-sqs example.com/myservice

cd myservice && go test ./...

Replace example.com/myservice with your own module path (its last segment becomes the new directory name).

Starter	Hosts
aws-apigateway	AWS Lambda fronted by API Gateway HTTP requests (plus direct wire-envelope invokes)
aws-sqs	AWS Lambda triggered by an SQS event source mapping

Each starter generates a complete, buildable module: a composition root (newApp in main.go) + a demo greet handler behind a Greeter port + the transport host + a benzenetest component test that drives a real message through the whole pipeline + an AWS SAM template.yaml + a Dockerfile.

Module resolution caveat: the templates require github.com/daniellepelley/benzene-go (and, for aws-sqs, the awssqs module) at their published version with no replace directive - a shipped replace would break the moment gonew copies the module out of this repo. Until those modules are tagged and published, a freshly generated project needs a replace you add pointing at a local checkout:

# in the generated project, while benzene-go is not yet published:
go mod edit -replace github.com/daniellepelley/benzene-go=/path/to/benzene-go
go mod tidy

See templates/README.md for the full model, per-template detail, and the maintainer verification steps.

Design notes for Go developers

Two things surprise Go developers on first read. Both are deliberate, and knowing why makes the rest of the API predictable. (For the fuller idiom rationale and where the design bends toward Go vs. stays consistent across language ports, see docs/go-idioms-review.md.)

Handlers return Result[T], not (T, error)

A handler is func(context.Context, TReq) benzene.Result[TRes] — there is no error return. That is not an oversight: Result[T] carries a Status from a fixed, wire-level status vocabulary (ok, bad-request, not-found, unexpected-error, …) that every Benzene language port and every transport shares, so a handler's outcome maps identically onto an HTTP status, a gRPC code, or a queue ack/nack. Returning a value instead of an error is also what lets a batch consumer turn one bad message into a bad-request result rather than crashing the whole batch.

func createOrder(_ context.Context, req orderRequest) benzene.Result[orderResponse] {
    if req.ID == "" {
        return benzene.BadRequest[orderResponse]("id is required") // not: return nil, errors.New(...)
    }
    return benzene.Ok(orderResponse{...})
}

Go error is still used everywhere it belongs — infrastructure speaks error (Register, Consumer.Run, request decoding all return error); only the handler boundary speaks Result. To map a Go error from a dependency at the handler edge, translate it to the status you want: if err != nil { return benzene.UnexpectedError[Res](err.Error()) }.

Container/Scope is DI-lite — prefer closures; use a typed key when you need the container

The Container/Scope is a small first-party DI helper (a named cross-language concept), not a reflection framework. For most dependencies you don't need it at all: capture a singleton in the handler's closure at registration time — plain Go, no lookup, no key.

func newApp(orders OrderStore) benzene.App[Config] { /* orders is captured by the handler closure */ }

Reach for the container only for a scoped (per-invocation) or transient dependency, resolved via benzene.ScopeFromContext(ctx) + benzene.GetService[T]. When you do, prefer a typed key over a bare string, so keys can't collide and the compiler helps you:

type orderStoreKey struct{}
benzene.AddScoped(container, orderStoreKey{}, func(*benzene.Scope) *OrderStore { return &OrderStore{} })
// in the handler:
scope, _ := benzene.ScopeFromContext(ctx)
store := benzene.GetService[*OrderStore](scope, orderStoreKey{})

GetService panics if a required service is missing (use TryGetService for the optional case) — the "required dependency" contract, surfaced loudly at startup rather than as a nil later.

Packages

Package	Coverage	What it is
benzene (root)	100%	Topic, Status, Result[T], Registry, Middleware/Pipeline, RouterMiddleware, the DI-lite Container/Scope, the three-phase App lifecycle
wire	100%	The transport-neutral message envelope (Request/Response/ErrorPayload) - no dependency on the rest of this module
httpstatus	100%	The Benzene<->HTTP status mapping tables
grpcstatus	100%	The Benzene<->gRPC status mapping tables (wire-contracts §4.2) - raw numeric gRPC status codes, so this stays zero-dependency like httpstatus; a gRPC binding wraps the result as codes.Code(...)
envelope	96%+	Dispatches a wire.Request through a Pipeline and produces a wire.Response (merging any invocation-set response headers - see benzene.SetResponseHeader) - shared by httpbinding, httpclient, and conformance
httpbinding	97%+	The HTTP transport binding: a native REST-style Handler (real HTTP status codes, explicit route table with {param} path templating - captured segments arrive as route-<name> wire headers) and an EnvelopeHandler (the wire envelope over HTTP); handler-set response headers come back as real HTTP headers
httpclient	97%+	The HTTP outbound client - one Send(topic, headers, message) method, mapping transport failures to ServiceUnavailable
healthcheck	100%	Middleware that intercepts the reserved healthcheck topic and responds with the standard aggregate health response, plus ready-made Checks: TCPCheck (opens a connection, Benzene.HealthChecks.Tcp), HTTPPingCheck (GET, healthy only on 200, URL credentials stripped, Benzene.HealthChecks.Http), and DiskSpaceCheck (host free-space self-check, Benzene.HealthChecks.Disk: WithMinimumFreeBytes/WithWarningFreeBytes gate health, else pure telemetry). All zero-dep and report a coarse error category, never the raw message; DiskSpaceCheck's one platform call sits behind build tags (syscall.Statfs on unix, GetDiskFreeSpaceExW on windows, no x/sys)
validation	100%	Request-validation building block (zero deps): Validated(validator, handler) wraps a handler so an invalid request short-circuits to a validation-error result before the handler runs (Validator[T]/ValidatorFunc[T] + a Combine composer). The Go-idiomatic form of Benzene.DataAnnotations/Benzene.FluentValidation's ValidationMiddleware - a typed handler wrapper, since this port's pipeline is type-erased until dispatch
idempotency	100%	De-duplicates redelivered messages on an at-least-once transport (zero deps), matching Benzene.Idempotency: a pipeline Middleware(store, key) that atomically claims a header-derived key in a pluggable Store and runs the handler only the first time - a completed duplicate is ignored (ack), an in-progress one is conflict (retry), the winning attempt records completion on success / releases on failure. InMemoryStore (separate short in-progress-lease and long completed-dedup TTLs, so a crashed worker's key frees fast; + clock) is the default; a store outage fails open
ratelimiting	100%	Best-effort per-instance rate-limiting middleware (zero deps), matching Benzene.RateLimiting: Middleware(limiter, cost) acquires each message's permit cost from a Limiter and short-circuits a rejected message to too-many-requests. A Limiter interface + a standard-library thread-safe TokenBucket default (plug a different algorithm - e.g. a golang.org/x/time/rate adapter - behind the interface), so the root module stays dependency-free. Per instance, not a fleet-wide limit
resilience	100%	Retry + timeout + bulkhead + fallback middleware (zero deps), matching most of Benzene.Resilience(.Polly) - only the circuit breaker (own module) and hedging (still to do) live elsewhere. Middleware(opts...) re-invokes the downstream with exponential backoff; two retry triggers since the router funnels failures onto ic.Result not a Go error - WithRetryOnError (default: any non-cancellation error) and WithRetryOnResult (default: never; pass RetryUnsuccessful/RetryOnStatus(...)). Backoff caps/jitters the sleep while growing the curve uncapped (AWS "full jitter", FullJitter helper); context-cancellable sleep. Timeout(d) bounds the downstream to a deadline (a cooperative context.WithTimeout, presented as a StatusTimeout result). Bulkhead(maxConcurrency, opts...) caps concurrent invocations (Polly's two-permit semaphore), shedding load fast to too-many-requests or, with WithMaxQueue(n), letting callers wait (context-bounded). Fallback(fn, opts...) substitutes a degraded ic.Result when an attempt fails (same *Unsuccessful/*OnStatus triggers), e.g. degrading an open circuit breaker to a cached response. Place above idempotent outbound calls
circuitbreaker (own module)	100%	Circuit-breaker middleware, the library-backed slice of Benzene.Resilience.Polly (needs sony/gobreaker/v2, hence its own module): Middleware[T](cb, opts...) runs the downstream inside a gobreaker CircuitBreaker - a next() error or a matching ic.Result (per WithTripOnResult, default TripOnServerError: only dependency-health statuses, so client errors never open the breaker; TripUnsuccessful/TripOnStatus(...) to broaden) counts as a failure; once open it short-circuits without invoking the downstream to a fail-fast status (WithOpenStatus, default service-unavailable). Open-state detected via a called flag (robust vs a downstream returning gobreaker's own sentinels); fail-fast result built at wiring time. Complements the zero-dep resilience (retry + timeout + bulkhead + fallback)
auth	99.5%	Authentication/authorization building block (zero deps), matching Benzene.Auth.Core+.Basic+.OAuth2: a Principal (name/roles/claims) threaded on the context; BasicAuth(validate, realm) RFC 7617 middleware; BearerAuth(validator, opts...) OAuth2/JWT bearer middleware (the Go form of OAuth2BearerMiddleware) - validates a JWT and sets the principal, or short-circuits with a generic unauthorized (real reason only via WithOnError, never an oracle). JWT validation is pure stdlib (so zero-dep where .NET uses Microsoft.IdentityModel): explicit algorithm allowlist (RFC 8725 - none/off-list rejected up front), HS/RS/ES 256/384/512 with per-family typed keys (no cross-family confusion), iss/aud/exp/nbf/iat with clock skew; keys from StaticKeys or a caching JWKSResolver (+ OIDC discovery via NewJWKSFromAuthority). Authorize/RequireRole/RequireScope authorization middleware (forbidden when not permitted, unauthorized when absent)
cache	100%	Caching building block (zero deps), matching the essence of Benzene.Cache.Core: a pluggable Store (Get/Set/Delete with per-entry TTL) + a generic read-through helper GetOrLoad[T](ctx, store, key, ttl, load) (the Go form of CacheEntry.LazyLoad). InMemoryStore (thread-safe, TTL + clock) is the default; a shared store (Redis) is its own module. Degrades safely - a store read error is a miss, a write error is ignored, a load error is returned and not cached
saga	100%	In-code saga orchestrator (zero deps, in-process), matching Benzene.Saga: New(stages...) runs NewStage(steps...) in order, steps within a stage concurrently; each NewStep[T](forward, compensate) is a forward action + optional compensation. On the first stage failure it compensates every effect in reverse (LIFO) order and returns a Result (Succeeded/RolledBack/PartiallyRolledBack). A SagaContext threads results between stages (Set/Get[T]). RunWith adds an observability StateStore and a RetryPolicy (retries only a clean rollback). In-process only - no durable crash-resume (use Step Functions/Durable Functions/Temporal for that)
responseevents	100%	Response-as-event middleware (zero deps), matching Benzene.ResponseEvents: Middleware(publisher, mappings, opts...) republishes a handler's response payload as a follow-up event on a fire-and-forget transport (an order:create handler's payload published as order:created). Map (source->event, When/Project options) and CrudConvention are the ready-made mappings (+ custom Mapping); every match publishes (fan-out). NewSenderPublisher(client.Sender) is the default outbound port. FailMessage (default) nacks/redelivers on a publish failure, LogAndContinue keeps the response (optional OnPublishError hook). The AsyncAPI spec-catalog half is not ported (Go has no spec generator; mesh descriptors are the introspection path)
clienthealthcheck	100%	Consumer-side dependency health check (zero deps), matching Benzene.Clients.HealthChecks: a ServiceCheck probes a downstream Benzene provider's reserved benzene:mesh descriptor via an outbound client.Sender and reports the contract relationship - unreachable/no-descriptor → failed, reachable+matching hash → ok, reachable+drifted hash → warning (degraded, doesn't flip health). Reachability comes from the descriptor (served health-independently), never benzene:healthcheck, so it doesn't couple to the provider's transient health. WithExpectedContractHash compares the provider's live descriptorHash against the hash the consumer was built against. For a contracts diagnostic surface, not a liveness probe
cloudserviceprobe	100%	External, black-box conformance checker for the Cloud Service Profile (zero deps, net/http only), matching Benzene.CloudService.Probe: Run(ctx, client, baseURL, opts...) hits a running service over HTTP and returns a tri-state Report (Satisfied/NotSatisfied/Inconclusive) for R1-R8 - never a bool, panic, or error. R8 and half of R6 are structurally unobservable from one service and stay Inconclusive by design. Independent of httpbinding/mesh (own path constants, own JSON parsing) so it can audit any Benzene Cloud Service, including a non-Go port
cloudservice	100%	One-call Cloud Service Profile builder (zero deps), the assembly counterpart of cloudserviceprobe and the Go form of Benzene.CloudService: New(name, registry, opts...) wires the profile's synchronous HTTP surface - R1 hosted pipeline, R2 registry handlers, R3 health + /benzene/health, R4 envelope-invoke /benzene/invoke, R5 derived spec /benzene/spec, R7 default paths, plus the benzene:mesh descriptor - over one ApplicationBuilder, and returns the http.Handler, Descriptor, Builder, and a wiring ProfileReport - a full R1-R8 checklist. It's honest: New doesn't wire R6's outbound feeds (register/heartbeat/traces) or R8 (trace propagation), so Satisfied() is false for a New-only build and Unsatisfied() is the exact to-do list. WithoutDescriptor() drops R5/R6 per §4 exposure control
logging	100%	Basic request logging/timing middleware, log/slog only (zero deps): one structured line per invocation - topic, status, duration - Info/Warn/Error by outcome. The dependency-free alternative to diagnostics
mesh	100%	Phases 1-2 of Benzene Mesh: the service Descriptor derived from the live Registry - topics, per-topic request/response JSON Schemas derived at startup from the registered handler types, and the contract descriptorHash - plus reserved-mesh-topic descriptor middleware and TraceMiddleware + LogExporter emitting semantic per-invocation trace events. Every feed is optional - a service with only some feeds provisioned runs a reduced mesh, never a broken one
meshd	100%	Phases 3-4 of Benzene Mesh: the collector - itself an ordinary Benzene service serving benzene:mesh:register/benzene:mesh:heartbeat/benzene:mesh:traces and the benzene:mesh:query:* read models over an in-memory store, plus the Mesh View (one embedded self-contained page, no JS framework). Accepts partial fleets: a service missing a feed renders as reduced, never breaks ingestion or queries
openapi	98%+	OpenAPI 3.0 document generation (zero deps), the Go form of Benzene.Schema.OpenApi: Generate(desc, opts...) turns a mesh.Descriptor (mesh.Describe(registry, info)) into an OpenAPI doc - each registered topic a POST operation whose request body is the topic's request schema and whose responses carry the response schema (200) and the Benzene failure vocabulary mapped to HTTP codes (httpstatus). Reuses mesh's derived schemas (no new reflection) and converts JSON Schema's nullable type-array to OpenAPI 3.0's nullable: true. Handler(desc, opts...) serves it over GET, the OpenAPI sibling of mesh.SpecHandler
asyncapi	98%+	AsyncAPI 3.0 document generation (zero deps), the event-driven sibling of openapi and the other half of Benzene.Schema.OpenApi: Generate(desc, opts...) maps every handled topic to a receive operation on a channel carrying the request, with the native reply object pointing at a <topic>:<suffix> reply channel (default response) - derived from the descriptor. Published events are a caller-declared input via WithSentEvent(topic, payload) (a send operation), the send side the descriptor can't provide (so no sync-vs-event classification is fabricated). Reuses mesh's derived schemas with no reshaping (AsyncAPI 3.0 is JSON Schema Draft 7, so the nullable ["T","null"] form is already valid). Handler serves it over GET
awslambda	93%+	AWS Lambda binding: a hand-rolled Lambda Runtime API bootstrap loop (Start), plus HTTPHandler (Function URL / API Gateway v2.0, API Gateway REST/v1.0, and ALB target-group events, detected per invocation) and EnvelopeHandler (direct invoke)
azurefunctions	93%+	Azure Functions custom-handler binding: Handler adapts the Data/Metadata JSON contract for HTTP-triggered functions; QueueHandler adapts queue-shaped triggers (Storage Queue, Service Bus), reporting a failed message via a non-2xx outer status so the platform's own redelivery/poison-queue machinery takes over; CosmosHandler adapts the Cosmos DB Change Feed trigger - fan-in, not topic-routed (the whole batch of changed documents is one invocation dispatched to a developer-named topic, handler takes the batch as a slice), checkpointing on success only (outer 500 redelivers the whole batch); TimerHandler adapts the Timer trigger (fan-in like Cosmos - a scheduled tick has no message, so the topic is named in code and the body is the tick's schedule info; outer 200/500, no redelivery)
client	100%	Outbound-client decorators (CorrelationDecorator, RetryDecorator) over a transport-agnostic Sender interface. The spec's third client behavior, trace-context propagation, is mesh.TraceContextDecorator (in mesh, which owns the Span it forwards) - it composes over the same Sender
inprocess	96%+	An in-process client.Sender: dispatches straight to a handler pipeline built in the same runtime, without going over any wire. PipelineSet accumulates one named *benzene.ApplicationBuilder per module (each its own independent Registry/Container/Pipeline); Sender binds to one, FanOutSender binds to several and dispatches to all of them concurrently, isolating each target's failure. No shared/process-wide handler registry to collide over (unlike the .NET and TypeScript ports), so fan-out targets may share a literal topic
cors	100%	Portable CORS middleware for HTTP-fronted services (origin/scheme/port matching, header wildcard, preflight)
benzenetest	100%	In-process test host for your application's tests - Invoke[TReq, TRes] runs one pipeline invocation without real HTTP/Lambda/etc.
cloudevents	99%+	CloudEvents 1.0 mapping (zero deps): type ↔ topic, data ↔ body, other attributes ↔ ce--prefixed wire headers; Handler accepts CloudEvents over HTTP in both content modes (binary ce-* headers and structured application/cloudevents+json) from Event Grid subscriptions, Knative triggers, EventBridge API destinations, etc.; FromRequest/MarshalJSON emit events for the outbound direction
gcppubsub	100%	Google Cloud Pub/Sub inbound binding (zero deps): an http.Handler for a push subscription's endpoint - decodes the push envelope, resolves the topic per wire-contracts §2 (topic attribute or envelope-in-body), acks with 204 / nacks with 500 so Pub/Sub's own redelivery/dead-letter machinery handles failures. Outbound publishing needs the Pub/Sub SDK - a pending dependency decision (see ROADMAP.md)
awssqs (own module)	100%	AWS SQS binding: inbound Handler for a Lambda triggered by an SQS event source mapping (zero deps), a self-hosted Consumer poller (Run(ctx) long-polls + deletes only successfully-dispatched messages, matching Benzene.Aws.Sqs), and an outbound Client publishing via SendMessage (needs aws-sdk-go-v2/service/sqs)
diagnostics (own module)	100%	OpenTelemetry diagnostics middleware - one server span per invocation (named by topic, joined to the caller's W3C traceparent, benzene.topic/benzene.version/benzene.status attributes) plus benzene.messages.processed/benzene.message.duration metrics (topic/transport/result-attributed, the cross-port observability conventions), and TraceContextDecorator (the OTel-path outbound client decorator that injects the active span context as a traceparent, sibling of mesh.TraceContextDecorator). Depends on the OTel API only; your app owns the SDK/exporter, and with no SDK installed the no-op defaults make it free
kafka (own module)	100%	Kafka binding, matching the main repo's spec exactly (one Kafka topic = one Benzene topic, headers pass through verbatim, no envelope wrapping): a Consumer loop over a consumer group (one pipeline invocation + DI scope per record, explicit commits, an OnFailure hook since Kafka has no broker-side redelivery/DLQ to hand a failed message to) and an outbound Client satisfying client.Sender (needs segmentio/kafka-go - a broker wire protocol isn't hand-rollable)
grpcbinding (own module)	100%	gRPC binding, unary RPCs only: UnaryServerInterceptor claims specific registered Routes (full method path → topic, case-insensitive) on an ordinary *grpc.Server - unclaimed methods fall through to the native generated service untouched, per spec - with proto3-JSON body bridging, incoming/outgoing metadata as wire headers, and the mandatory benzene-status trailer; an outbound Client satisfying client.Sender recovers the precise status from that trailer. Needs google.golang.org/grpc + google.golang.org/protobuf
awseventbridge (own module)	96%+	AWS EventBridge binding, matching the main repo's spec exactly: inbound Handler for a Lambda invoked by a rule (zero deps; topic is detail-type verbatim, body is the raw detail JSON, headers are eventbridge--prefixed envelope metadata plus any _benzeneHeaders object embedded inside detail; a failed event returns a Go error, triggering AWS's async-invoke retry) and an outbound Client publishing via PutEvents (embeds headers under _benzeneHeaders when the message is a JSON object; needs aws-sdk-go-v2/service/eventbridge)
awssns (own module)	100%	AWS SNS binding: inbound Handler for a Lambda subscribed directly to an SNS topic (zero deps; a failed notification returns a Go error, triggering AWS's own async-invoke retry, since SNS has no batch/partial-failure mechanism), outbound Client publishing via Publish (needs aws-sdk-go-v2/service/sns)
awslambdaclient (own module)	~96%	Outbound Lambda-invoke Client (satisfies client.Sender), matching Benzene.Clients.Aws.Lambda: invokes a target Lambda with a wire envelope payload. RequestResponse parses the target's envelope response back into a Result; Event is fire-and-forget → accepted; a FunctionError → unexpected-error; transport failure → service-unavailable (needs aws-sdk-go-v2/service/lambda)
awsstepfunctions (own module)	~96%	Outbound Step Functions Client (satisfies client.Sender), matching Benzene.Clients.Aws.StepFunctions: starts a state-machine execution with the wire envelope as Input → accepted (fire-and-forget). Optional idempotent ExecutionName (sanitized, 80-rune cap); ExecutionAlreadyExists on a same-name retry is an idempotent accepted (needs aws-sdk-go-v2/service/sfn)
azureservicebus (own module)	100%	Azure Service Bus binding: outbound Client (topic as the reserved application property, headers as the others, body verbatim → accepted) + self-hosted Worker owning its own receive loop (Run(ctx), the pull-loop counterpart of .NET's push ServiceBusProcessor and the sibling of awssqs.Consumer): completes only successfully-dispatched messages, settles a failure per AckMode (abandon→redeliver, default / dead-letter→quarantine); settlement on a cancellation-detached context (needs azure-sdk-for-go/.../azservicebus)
azureeventhub (own module)	78%+	Azure Event Hubs binding: outbound Client publishing one event as a batch-of-one → accepted, and a Consumer reading over a narrow Receiver with checkpointing handed back to a caller-owned Checkpoint hook (Event Hubs checkpointing needs a blob-store checkpoint store the app owns - a documented divergence). Coverage gap is the thin SDK-adapter constructors, uncoverable without live Event Hubs (needs azure-sdk-for-go/.../azeventhubs/v2)
azureeventgrid (own module)	100%	Azure Event Grid binding: outbound CloudEvents Client (topic → CloudEvent Type, body → Data as json.RawMessage so a JSON payload rides as JSON not base64, headers → lowercased extension attributes → accepted) (needs azure-sdk-for-go/.../eventgrid/azeventgrid)
azurequeuestorage (own module)	86%+	Azure Queue Storage binding: outbound Client enqueuing the whole wire.Request envelope as the message text (verbatim, not base64) → accepted. Coverage gap is the defensive marshal-error branch (needs azure-sdk-for-go/.../storage/azqueue)
azurecosmos (own module)	76%+ (core 100%)	Self-hosted Azure Cosmos DB Change Feed Worker (Benzene.Azure.CosmosDb), the standalone counterpart of the zero-dep azurefunctions.CosmosHandler: reads the change feed over a narrow ChangeFeedReader and dispatches each page fan-in (whole batch → one invocation to a code-named Topic, like CosmosHandler); stop-at-batch-failure (an unsuccessful dispatch/checkpoint doesn't advance the continuation token, so the batch redelivers); caller-owned Checkpoint hook (Cosmos needs an app-owned lease container) on a detached context; a PollInterval paces empty polls. Struct-fields + Validate(). Coverage gap is the live-only SDK adapter (needs azure-sdk-for-go/.../data/azcosmos)
gcpfunctions (own module)	100%	Google Cloud Functions Gen2 inbound binding (GoogleCloud.Functions.Http + .PubSub): RegisterHTTP(name, builder, routes) registers a Gen2 HTTP function serving httpbinding.Handler (thin), and RegisterCloudEvent(name, builder, opts...) registers a CloudEvent-triggered (Pub/Sub/Eventarc) function that maps the event onto a wire.Request by reusing cloudevents.ToRequest (identical to this port's other CloudEvents surface), dispatches, and returns nil on success / an error on failure so the platform retries - never a silent drop. WithReservedNames/WithOnFailure; framework signatures pinned by compile-time assertions (needs functions-framework-go + cloudevents/sdk-go/v2)
gcppubsubclient (own module)	100%	Google Cloud Pub/Sub outbound client (the invoking counterpart of the inbound gcppubsub push handler): interface-driven Publisher + NewTopicPublisher adapter; Send publishes with topic + headers as Pub/Sub attributes (empty headers dropped), body as Data → accepted. Requires go 1.25 (the one module forcing the workspace go directive + CI toolchain to 1.25; every other module stays 1.24.7) (needs cloud.google.com/go/pubsub)
rabbitmq (own module)	100%	RabbitMQ binding: outbound Client publishing with the topic as both the routing key and a "topic" header, Persistent delivery, and a self-hosted Consumer (the AMQP sibling of awssqs.Consumer) that Acks a successful delivery, Nacks a failure and requeues it exactly once (poison-message bounded to one retry) (needs rabbitmq/amqp091-go)
awsdynamodb	100%	AWS DynamoDB Streams inbound binding (zero deps, root module): a Lambda Handler for a stream event source mapping. Topic is {tableName}:{eventName} (table parsed from the stream ARN + INSERT/MODIFY/REMOVE), body is the record's image unmarshalled from DynamoDB AttributeValue format into plain JSON (NewImage, else OldImage, else Keys). Records are ordered CDC, so processing is sequential and stops at the first failure, reporting that record's SequenceNumber for Lambda to checkpoint and redeliver - no outbound side (writing the table is the publish)
awskinesis	100%	AWS Kinesis Data Streams inbound binding (zero deps, root module), the sibling of awsdynamodb: a Lambda Handler for a stream event source mapping. Topic is the stream name (parsed from the record's stream ARN - a Kinesis record has no per-record event type, so the stream is the routing key), body is the record's data base64-decoded into the producer's bytes (typically JSON), headers are kinesis--prefixed metadata. Same ordered stop-at-first-failure + SequenceNumber checkpointing; no outbound side (writing the stream is the publish)
awskafka	100%	AWS Lambda MSK/self-managed-Kafka inbound binding (zero deps, root module), DISTINCT from the self-hosted kafka module (that runs its own broker consumer loop; this is the zero-dep adapter for AWS's managed event source mapping, which delivers records as plain JSON). Topic is the Kafka topic verbatim (one Kafka topic = one Benzene topic, like the kafka module - unlike Kinesis's stream routing), body is the record's value base64-decoded, headers pass through verbatim. Records grouped by {topic}-{partition}; each partition processed sequentially, stopping at its first failure and reporting an object-shaped {partition, offset} (unlike the string identifier of SQS/Kinesis/DynamoDB), so the mapping needs FunctionResponseTypes: [ReportBatchItemFailures]; partitions are independent. No outbound side (producing to Kafka is the publish)
awss3	100%	AWS S3 event-notification inbound binding (zero deps, root module): a Lambda Handler invoked by S3 on object create/remove. Topic is {bucket}:{eventName} (bucket-qualified, vs .NET's bare event name - a local routing concern), body is the object metadata (bucket/key/size/etag, not contents), headers are s3--prefixed. An S3 notification is an async invocation, so a failed record returns a Go error (async-invoke retry, like awssns), never a silent drop; handlers must be idempotent
conformance	n/a (test-only)	Runs this port against the fixtures vendored from the main repo's docs/specification/conformance/
codegen (own module, not tied into go.work)	90%+	benzene-codegen (codegen/cmd/benzene-codegen): generates a typed, topic-scoped Go client from a service's committed Contract Document ({Service}.spec.json) - the Go port of the .NET repo's Benzene.CodeGen.Client. contractdoc parses the document and implements its topic-scoping/schema-closure/contractHash rules (generic-JSON only, no schema library needed); gengo emits the Go source (structs + json tags, a client type/constructor, RequiredTopics), depending only on client.Sender/httpclient.Unmarshal/benzene.Result[T]. Its own module because contractHash needs an RFC 8785 canonicalizer (github.com/gowebpki/jcs) - a dependency that must never reach the dependency-free root/client/httpclient. See docs/codegen-client.md
examples/helloworld	-	A runnable example service - DI, health check, both HTTP entry points
examples/aws-lambda-helloworld	-	The same service, deployable to AWS Lambda (Dockerfile + SAM template)
examples/azure-functions-helloworld	-	The same service, deployable to Azure Functions (host.json/function.json)
examples/gcp-cloudrun-helloworld	-	The same service, deployable to Google Cloud Run (Dockerfile, no new package needed)
examples/gcp-pubsub-helloworld	-	A Cloud Run service consuming a Pub/Sub push subscription via gcppubsub.Handler - publish with gcloud pubsub topics publish, no publisher code needed
examples/aws-sqs-helloworld (own module)	-	A publisher Lambda (Function URL) forwarding to SQS + a consumer Lambda triggered by that queue
examples/aws-sns-helloworld (own module)	-	A publisher Lambda (Function URL) forwarding to SNS + a consumer Lambda subscribed to that topic
examples/aws-dynamodb-helloworld	-	A consumer Lambda triggered by a DynamoDB table's stream via awsdynamodb.Handler - write to the table to drive it, no publisher code
examples/aws-kinesis-helloworld	-	A consumer Lambda triggered by a Kinesis data stream via awskinesis.Handler - PutRecord onto the stream to drive it, no publisher code
examples/aws-kafka-helloworld	-	A consumer Lambda triggered by an MSK orders topic via awskafka.Handler - produce a record to drive it, no publisher code (the MSK cluster is a prerequisite, passed by ARN)
examples/aws-s3-helloworld	-	A consumer Lambda invoked by an S3 bucket's ObjectCreated notifications via awss3.Handler - upload an object to drive it, no publisher code
examples/mesh-helloworld	-	The whole mesh story in one process: a meshd collector + two meshed services with a cross-service traced call - open the Mesh View and watch the derived fleet
examples/http-helloworld	-	The greet handler on a standalone net/http server via httpbinding - a net/http middleware wrapping the binding, plus graceful shutdown (the analog of the .NET Asp example)
examples/grpc-helloworld (own module)	-	The greet handler over a gRPC unary RPC via grpcbinding (protoc-free structpb stand-in messages) + an outbound grpcbinding.Client round trip
examples/kafka-helloworld (own module)	-	A Kafka consumer group running the greet handler via the kafka module + an outbound kafka.Client publish path
examples/opentelemetry-helloworld (own module)	-	The greet handler wrapped in diagnostics tracing middleware - one OTel span per invocation plus a nested adapter span, exported to stdout
examples/codegen-helloworld	-	Dogfoods benzene-codegen: a committed Contract Document (contracts/payments.spec.json), a //go:generate-regenerated + committed topic-scoped client for payments:capture (paymentscapture/), and a test proving the generated method sends the right topic/payload against a fake client.Sender

Every non-test-only package sits at 100% coverage, or just under it where the gap is a defensively-unreachable branch (documented at the call site - e.g. a json.Marshal failure on a type that can't actually fail to marshal). Run go test ./... -cover to see current numbers.

Deploying to a cloud provider

Provider	Path	New package needed?
AWS	Lambda (container image) + a Function URL	awslambda - Lambda has no HTTP-server contract, only the Runtime API
AWS	Lambda triggered by SQS + publish-to-SQS	awssqs - its own module (needs the AWS SDK)
AWS	Lambda subscribed to SNS + publish-to-SNS	awssns - its own module (needs the AWS SDK)
AWS	Lambda triggered by a DynamoDB table's stream	awsdynamodb (inbound only, zero deps) - the stream delivers change records as plain JSON, no SDK needed
AWS	Lambda triggered by a Kinesis data stream	awskinesis (inbound only, zero deps) - the stream delivers records as plain JSON (data base64-encoded), no SDK needed
AWS	Lambda triggered by an MSK / self-managed Kafka topic	awskafka (inbound only, zero deps) - the managed event source mapping delivers records as plain JSON (value base64-encoded), no SDK needed; distinct from the self-hosted kafka module
AWS	Lambda invoked by S3 event notifications	awss3 (inbound only, zero deps) - S3 delivers the notification (object metadata) as plain JSON, no SDK needed
Azure	Azure Functions custom handler (HTTP, queue, Cosmos DB Change Feed, Timer, or Event Grid trigger)	azurefunctions - Azure has no native Go worker
Google Cloud	Cloud Run	None - Cloud Run's contract is "listen on $PORT", which httpbinding + net/http already satisfies
Google Cloud	Cloud Run consuming a Pub/Sub push subscription	gcppubsub (inbound only, zero deps) - the push envelope's base64/attributes/ack contract is the one GCP shape httpbinding can't cover

Each examples/*-helloworld directory's README documents the concrete deploy steps and states what was and wasn't verified in this repo's own CI sandbox. Each also has a matching GitHub Actions workflow (deploy-*.yml, one per example) that runs that same deploy on every push to main touching it - each is gated on its provider's credential secret being set, so the job shows as skipped (not failed) until you add the secrets/variables listed in that example's own README. None have been run for real from this repo (no live cloud credentials in this sandbox) - only the code, cross-compilation, and unit tests have been verified here.

Modules

This is a multi-module repo - see RELEASING.md for the full explanation (and for how Go's decentralized module distribution works at all, if you're coming from an ecosystem with a central package registry like NuGet). Short version: everything is one module except awssqs, awssns, awseventbridge, kafka, diagnostics, grpcbinding, examples/aws-sqs-helloworld, examples/aws-sns-helloworld, examples/grpc-helloworld, examples/kafka-helloworld, and examples/opentelemetry-helloworld, which have their own go.mod because they need real third-party dependencies the rest of the repo shouldn't carry. go.work ties them together for local development.

codegen is also its own module (same reason - it needs github.com/gowebpki/jcs) but is deliberately not listed in go.work: nothing it generates imports it, so nothing in the workspace needs it resolvable as a local replace; run its build/tests from inside codegen/ directly (cd codegen && go build ./... && go test ./...), same as any other standalone module.

Scope

This port covers core-concepts.md and wire-contracts.md end to end (pipeline, DI, health checks, HTTP binding + client, conformance, AWS/Azure/GCP deployment, SQS/SNS/Pub-Sub/Kafka/ gRPC bindings, CloudEvents) but does not yet have: gRPC's client-streaming/server-streaming/ duplex-streaming shapes (grpcbinding covers unary RPCs only - a documented scope decision, not an oversight; see its package doc), or a source-generator/codegen equivalent to the C# attribute-scanning sugar (per porting-guide.md, explicit registration is the framework contract in every language; attribute scanning is .NET-specific idiom, not something every port needs).

See ROADMAP.md for the fuller picture: what's next with zero new dependencies, what's next pending a dependency decision, and what's deliberately not being ported at all (and why). For how this project compares to the other ways of building cloud-portable services in Go (Dapr, Go CDK, Watermill, Encore) and when you'd pick each, see docs/comparison.md.

Benzene Mesh

Benzene Mesh - a fleet-wide, multi-cloud view of every service, its topics/schemas, health, and live traffic stats, derived from running services rather than declared in a catalog - is designed in docs/design/ (mesh.md, with a static mockup of the Fleet Overview screen and the research and positioning behind it). All phases of its delivery plan are complete: the mesh and meshd packages and the examples/mesh-helloworld demo above implement it, and the wire contracts are promoted and merged as the main repo's docs/specification/mesh.md - now the normative text. The main repo's .NET implementation (Benzene.Mesh.Wire + Benzene.Mesh.Collector) is the primary implementation of that contract; this port is a fully conforming implementation - the contract was originally extracted from it, the vendored mesh-*.json fixtures in conformance/ pin it and pass, and the two implementations have hosted each other's services in live cross-language fleets.

Both implementations also follow the spec's default service standard (the main repo's docs/specification/design-principles.md): framework-provided HTTP surfaces mount under a well-known /benzene/ prefix - here httpbinding.EnvelopePath (/benzene/invoke), httpbinding.HealthPath (/benzene/health), and meshd.ViewPath (/benzene/fleet-ui) - so they read as infrastructure rather than domain endpoints, with every path overridable per service. The same document records the wider "opinionated but optional" strategy the port already embodies: message handlers, like everything else, are the steer, never a requirement.

Developing

go build ./...
go vet ./...
gofmt -l .              # should print nothing
go test ./... -race -cover

CI (.github/workflows/ci.yml) runs all of the above on every push/PR to main.

License

MIT - see LICENSE.

Documentation

Index

• func AddScoped[T any](c *Container, key serviceKey, factory func(s *Scope) T)
• func AddSingleton[T any](c *Container, key serviceKey, factory func(s *Scope) T)
• func AddTransient[T any](c *Container, key serviceKey, factory func(s *Scope) T)
• func ContextWithScope(ctx context.Context, scope *Scope) context.Context
• func GetService[T any](s *Scope, key serviceKey) T
• func Register[TReq, TRes any](r *Registry, topic Topic, handler Handler[TReq, TRes]) error
• func SetResponseHeader(ctx context.Context, name, value string) (ok bool)
• func TryAddScoped[T any](c *Container, key serviceKey, factory func(s *Scope) T)
• func TryAddSingleton[T any](c *Container, key serviceKey, factory func(s *Scope) T)
• func TryAddTransient[T any](c *Container, key serviceKey, factory func(s *Scope) T)
• func TryGetService[T any](s *Scope, key serviceKey) (T, bool)
• type App
  • func (a App[TConfig]) Run() *ApplicationBuilder
• type ApplicationBuilder
  • func (b *ApplicationBuilder) UsePipeline(pipeline *Pipeline) *ApplicationBuilder
  • func (b *ApplicationBuilder) UseReservedNames(names wire.ReservedNames) *ApplicationBuilder
• type Container
  • func NewContainer() *Container
  • func (c *Container) NewScope() *Scope
• type Handler
• type InvocationContext
  • func NewInvocationContext(topic Topic, headers map[string]string, request any, scope *Scope) *InvocationContext
  • func (ic *InvocationContext) SetResponseHeader(name, value string)
• type Middleware
  • func RouterMiddleware(registry *Registry, opts ...RouterOption) Middleware
• type Pipeline
  • func NewPipeline(middlewares ...Middleware) *Pipeline
  • func (p *Pipeline) Run(ctx context.Context, ic *InvocationContext) error
• type Registry
  • func NewRegistry() *Registry
  • func (r *Registry) Has(topic Topic) bool
  • func (r *Registry) TopicTypes(topic Topic) (request, response reflect.Type, ok bool)
  • func (r *Registry) Topics() []Topic
• type Result
  • func Accepted[T any](payload T) Result[T]
  • func BadRequest[T any](errors ...string) Result[T]
  • func Conflict[T any](errors ...string) Result[T]
  • func Created[T any](payload T) Result[T]
  • func Deleted[T any](payload T) Result[T]
  • func Fail[T any](status Status, errors ...string) Result[T]
  • func Forbidden[T any](errors ...string) Result[T]
  • func Ignored[T any](payload T) Result[T]
  • func NotFound[T any](errors ...string) Result[T]
  • func NotImplemented[T any](errors ...string) Result[T]
  • func Ok[T any](payload T) Result[T]
  • func ServiceUnavailable[T any](errors ...string) Result[T]
  • func SetResult[T any](status Status, payload T, successful bool) Result[T]
  • func Timeout[T any](errors ...string) Result[T]
  • func TooManyRequests[T any](errors ...string) Result[T]
  • func Unauthorized[T any](errors ...string) Result[T]
  • func UnexpectedError[T any](errors ...string) Result[T]
  • func Updated[T any](payload T) Result[T]
  • func ValidationError[T any](errors ...string) Result[T]
  • func (r Result[T]) IsSuccessful() bool
  • func (r Result[T]) ResultErrors() []string
  • func (r Result[T]) ResultIsSuccessful() bool
  • func (r Result[T]) ResultPayload() any
  • func (r Result[T]) ResultStatus() Status
• type ResultInfo
• type RouterOption
  • func WithVersionKeys(keys ...string) RouterOption
• type Scope
  • func ScopeFromContext(ctx context.Context) (*Scope, bool)
• type Status
  • func (s Status) IsFailure() bool
  • func (s Status) IsKnown() bool
  • func (s Status) IsSuccess() bool
• type Topic
  • func NewTopic(id string) Topic
  • func (t Topic) String() string
  • func (t Topic) WithVersion(version string) Topic

Examples

• GetService
• Register
• Result
• RouterMiddleware (Versioned)

Functions

func AddScoped

func AddScoped[T any](c *Container, key serviceKey, factory func(s *Scope) T)

AddScoped registers factory to be called once per invocation scope; the same instance is reused for the lifetime of that scope, then discarded.

func AddSingleton

func AddSingleton[T any](c *Container, key serviceKey, factory func(s *Scope) T)

AddSingleton registers factory to be called at most once; the same instance is reused for every scope thereafter.

func AddTransient

func AddTransient[T any](c *Container, key serviceKey, factory func(s *Scope) T)

AddTransient registers factory to be called every time the service is resolved.

func ContextWithScope

func ContextWithScope(ctx context.Context, scope *Scope) context.Context

ContextWithScope returns a copy of ctx carrying scope, retrievable with ScopeFromContext. core-concepts.md §4 says invocation-scoped facts ride on the context "(or an accessor resolved from the invocation's scope)" - this is that accessor. RouterMiddleware calls this before invoking a handler, so a handler that needs a scoped or transient dependency (a singleton can simply be captured in the handler's closure at registration time) resolves it via ScopeFromContext(ctx) rather than needing Scope added to the Handler signature itself.

func GetService

func GetService[T any](s *Scope, key serviceKey) T

GetService resolves key, panicking if it has no registration - mirroring the spec's "required" resolution operation, which throws/panics rather than returning a zero value on a missing registration (a missing required dependency is a programming error, not a recoverable runtime condition).

Example

ExampleGetService shows the DI-lite Container/Scope. Register a per-invocation (scoped) dependency under a typed key; a handler then resolves it from the context via ScopeFromContext + GetService. Resolving twice inside one scope returns the same instance. (For a singleton you don't need the container at all - capture it in the handler's closure at registration time.)

package main

import (
	"fmt"

	benzene "github.com/daniellepelley/benzene-go"
)

type greetingCount struct{ n int }

// countKey is a typed DI key (a struct key can't collide with another package's key the way a bare
// string could).
type countKey struct{}

func main() {
	container := benzene.NewContainer()
	benzene.AddScoped(container, countKey{}, func(*benzene.Scope) *greetingCount {
		return &greetingCount{}
	})

	scope := container.NewScope() // one scope per invocation; a transport binding creates it for you
	first := benzene.GetService[*greetingCount](scope, countKey{})
	first.n++
	second := benzene.GetService[*greetingCount](scope, countKey{})

	fmt.Println(first == second, second.n)
}

Output:
true 1

func Register

func Register[TReq, TRes any](r *Registry, topic Topic, handler Handler[TReq, TRes]) error

Register adds handler for topic. Returns an error if topic is already registered - registering two handlers for the same (id, version) pair is a startup error, not a runtime dispatch ambiguity (core-concepts.md §2).

Example

ExampleRegister shows the core loop: a handler is a plain func(context, TReq) Result[TRes]; Register binds it to a topic on a Registry; RouterMiddleware turns the Registry into a Pipeline that dispatches an incoming message to the matching handler. (In a real service a transport binding - httpbinding, awslambda, a queue Consumer - feeds the pipeline; here we drive it directly to show the moving parts.)

package main

import (
	"context"
	"fmt"

	benzene "github.com/daniellepelley/benzene-go"
)

type greetReq struct {
	Name string `json:"name"`
}

type greetResp struct {
	Greeting string `json:"greeting"`
}

func main() {
	registry := benzene.NewRegistry()
	if err := benzene.Register(registry, benzene.NewTopic("greet"),
		benzene.Handler[greetReq, greetResp](func(_ context.Context, req greetReq) benzene.Result[greetResp] {
			return benzene.Ok(greetResp{Greeting: "Hello, " + req.Name + "!"})
		})); err != nil {
		panic(err)
	}

	pipeline := benzene.NewPipeline(benzene.RouterMiddleware(registry))
	scope := benzene.NewContainer().NewScope()
	ic := benzene.NewInvocationContext(benzene.NewTopic("greet"), nil, greetReq{Name: "World"}, scope)
	if err := pipeline.Run(context.Background(), ic); err != nil {
		panic(err)
	}

	fmt.Println(ic.Result.ResultStatus())
	fmt.Println(ic.Result.ResultPayload().(greetResp).Greeting)
}

Output:
ok
Hello, World!

func SetResponseHeader

func SetResponseHeader(ctx context.Context, name, value string) (ok bool)

SetResponseHeader records an outbound transport header for the invocation ctx belongs to - the handler-side counterpart of InvocationContext.SetResponseHeader, for handlers (whose signature carries no *InvocationContext). ok = false if ctx carries no invocation (e.g. in a unit test that calls a handler directly) - the header is then dropped, matching how a handler must keep working when a transport has nowhere to put response headers anyway.

func TryAddScoped

func TryAddScoped[T any](c *Container, key serviceKey, factory func(s *Scope) T)

TryAddScoped is TryAddSingleton's scoped-lifetime counterpart.

func TryAddSingleton

func TryAddSingleton[T any](c *Container, key serviceKey, factory func(s *Scope) T)

TryAddSingleton registers factory as a singleton only if key has no registration yet - this is how framework defaults are made overridable (core-concepts.md §8): the framework tryAdds its defaults, and the application's own explicit Add* registration (applied first) wins.

func TryAddTransient

func TryAddTransient[T any](c *Container, key serviceKey, factory func(s *Scope) T)

TryAddTransient is TryAddSingleton's transient-lifetime counterpart.

func TryGetService

func TryGetService[T any](s *Scope, key serviceKey) (T, bool)

TryGetService resolves key, returning ok = false if it has no registration instead of panicking.

Types

type App

type App[TConfig any] struct {
	GetConfiguration  func() TConfig
	ConfigureServices func(registry *Registry, container *Container, config TConfig)
	Configure         func(builder *ApplicationBuilder, config TConfig)
}

App is a Benzene application definition: the three-phase lifecycle of core-concepts.md §7, run once, in order, at startup:

1. GetConfiguration produces the configuration object. No service resolution is available yet.
2. ConfigureServices registers handlers, middleware dependencies, and adapters with the registry/container.
3. Configure builds the pipeline(s) against a platform-neutral ApplicationBuilder. Transport-specific entry points are attached by calling a transport binding's own constructor against the returned ApplicationBuilder.

TConfig is application-defined; Benzene itself doesn't prescribe its shape.

func (App[TConfig]) Run

func (a App[TConfig]) Run() *ApplicationBuilder

Run executes the three-phase lifecycle once and returns the built ApplicationBuilder, ready for a transport binding to attach entry points to (e.g. an http.Handler for the HTTP binding). All three phases are optional: GetConfiguration, ConfigureServices, and Configure may each be left nil - an application with no configuration yields the zero value of TConfig, and one with no dependencies to register (or nothing further to configure beyond the defaults) simply skips that phase.

type ApplicationBuilder

type ApplicationBuilder struct {
	Registry  *Registry
	Container *Container
	Pipeline  *Pipeline
	// ReservedNames overrides the reserved metadata/header names (wire-contracts.md §2). It is
	// the single injectable value the spec calls for: set it once here and every inbound binding
	// built off this builder reads it, so a service renames a colliding key in one place. Its
	// zero value means the standard defaults. The same value MUST also be given to the service's
	// outbound clients (the queue Client structs' ReservedNames field), since an override applies
	// to both directions.
	ReservedNames wire.ReservedNames
}

ApplicationBuilder is the platform-neutral application builder handed to App.Configure. A transport binding's `Use<Transport>(builder, ...)`-shaped constructor reads Registry/ Container/Pipeline off it to build that transport's native entry point (an http.Handler, a Lambda handler function, ...) - core-concepts.md §7's "one application definition can target several platforms" rule. Go typically compiles one binary per deployment target rather than runtime-detecting the host, so the "no-op on other platforms" half of that rule mostly falls out for free here; a future binding that DOES need runtime platform detection (e.g. a single binary that can run as either an HTTP server or a Lambda function depending on environment) can still check for its own platform indicators before activating, exactly as any other Go code would.

func (*ApplicationBuilder) UsePipeline

func (b *ApplicationBuilder) UsePipeline(pipeline *Pipeline) *ApplicationBuilder

UsePipeline sets the middleware pipeline transport bindings will run invocations through. Call this from Configure before any binding constructor that needs it. Returns the builder so calls can be chained.

func (*ApplicationBuilder) UseReservedNames

func (b *ApplicationBuilder) UseReservedNames(names wire.ReservedNames) *ApplicationBuilder

UseReservedNames overrides the reserved metadata/header names (wire-contracts.md §2) for every inbound binding built off this builder. Call it from Configure before the binding constructors. Returns the builder so calls can be chained. Remember to pass the same names to the service's outbound clients - an override applies to both directions.

type Container

type Container struct {
	// contains filtered or unexported fields
}

Container is the shared registration set an application configures once at startup (core-concepts.md §8): singleton/scoped/transient registrations, by factory. Languages without a DI culture (Go included) MAY implement the container-abstraction concept as an explicit registry/context object rather than a full framework - this Container is that explicit object, not a general-purpose reflection-based DI container.

func NewContainer

func NewContainer() *Container

NewContainer returns an empty Container.

func (*Container) NewScope

func (c *Container) NewScope() *Scope

NewScope creates a new per-invocation scope over c. A scope is created per pipeline invocation; scoped services live and die with it (core-concepts.md §8).

type Handler

type Handler[TReq, TRes any] func(ctx context.Context, req TReq) Result[TRes]

Handler is a function from a request to a result (core-concepts.md §3). The handler never sees the transport - a transport binding maps its native payload to TReq and the Result[TRes] back to a native response.

type InvocationContext

type InvocationContext struct {
	Topic   Topic
	Headers map[string]string
	// Request is the native/raw request payload for this invocation (e.g. a JSON body as
	// []byte, or an already-typed value for zero-copy passthrough). The router middleware
	// converts it into the resolved handler's declared TReq.
	Request any
	// Result is populated by the router middleware once the handler (or the NotFound /
	// error fallback) has run. Middleware registered after the router can inspect or
	// replace it; middleware registered before the router runs before it exists.
	Result ResultInfo
	// Scope is this invocation's per-invocation DI scope (scope.go).
	Scope *Scope
	// ResponseHeaders holds outbound transport headers set during this invocation - by
	// middleware directly, or by a handler via SetResponseHeader(ctx, ...) (the router puts
	// this invocation context on the handler's ctx, the same accessor pattern as
	// ScopeFromContext). A binding merges these onto its response after dispatch: the wire
	// envelope's headers for envelope-shaped transports, real response headers for the native
	// HTTP binding. Nil until the first set - fire-and-forget transports never allocate it.
	ResponseHeaders map[string]string
}

InvocationContext carries the state of a single pipeline invocation (core-concepts.md §6): the resolved topic, headers, the native/raw request payload (converted to a handler's declared TReq by the terminal router middleware - see convertRequest), and a slot for the result once a handler has run.

Cancellation, deadlines, and other invocation-scoped facts ride on the ctx.Context parameter threaded through Pipeline.Run and every Middleware call, per core-concepts.md §4 ("the pipeline carries no cancellation parameter... rides on the context") - Go's context.Context already carries that here, so nothing extra is needed on InvocationContext itself for that concern.

func NewInvocationContext

func NewInvocationContext(topic Topic, headers map[string]string, request any, scope *Scope) *InvocationContext

NewInvocationContext builds an InvocationContext for one pipeline invocation. headers may be nil, in which case an empty map is used.

func (*InvocationContext) SetResponseHeader

func (ic *InvocationContext) SetResponseHeader(name, value string)

SetResponseHeader records an outbound header on this invocation, to be merged onto the transport response by the binding. Names are lower-cased, matching wire-contracts.md §2's "SHOULD be written lower-case" and the inbound flattening every binding already does; a repeated name overwrites (last write wins).

type Middleware

type Middleware func(ctx context.Context, ic *InvocationContext, next func(context.Context) error) error

Middleware wraps invocation handling in an ordered onion pipeline (core-concepts.md §4). A middleware that does not call next terminates the pipeline; everything after it (including the handler dispatch, if registered later) does not run - this is the mechanism behind features like health-check interception.

Cancellation/deadlines ride on ctx, not on this signature, so the shape is identical across transports that have no cancellation concept at all.

func RouterMiddleware

func RouterMiddleware(registry *Registry, opts ...RouterOption) Middleware

RouterMiddleware returns the terminal middleware that resolves ic.Topic against registry and dispatches to the matching handler, storing the outcome on ic.Result. Conventionally registered last in a Pipeline (core-concepts.md §4).

It reads the message's payload schema version off the wire (wire-contracts.md §2 tier C, versioning.md §2.1) when a binding has not already resolved one: the version travels as a header on every transport (queues, the envelope, and the native HTTP binding's request headers all land it on ic.Headers), so reading it here covers them uniformly. A binding that resolves a version another way - e.g. an HTTP /v{version} route segment setting ic.Topic.Version before the pipeline runs - wins, since a version already on the topic is left untouched.

Handler selection stays exact-match (core-concepts.md §2), with one fallback: a signalled version that has no exact (id, version) handler routes to the unversioned (default-version) handler if one exists. That is core-concepts.md §2's absent-means-default applied to an unmatched version, and the guarantee that turning on the read path stays non-regressive - a stray version header on a service that registered only unversioned handlers still routes to them rather than falling to not-found. versioning.md §3's richer exact-else-highest-supported selection is a deliberate future addition (see the port ROADMAP), not implemented here.

Per core-concepts.md §2/§5, this middleware never returns a Go error for an application- level outcome - a missing topic, a missing handler, a request-conversion failure, or a handler panic all become a Result on ic.Result (ValidationError, NotFound, BadRequest, and ServiceUnavailable respectively), so every caller uniformly reads ic.Result rather than distinguishing "no handler" from "handler ran" via the Go error return. A handler panic specifically MUST NOT crash the transport adapter (§5) - recovered here and mapped to ServiceUnavailable, which wire-contracts.md §3 defines as "also the mapping for uncaught handler exceptions."

Example (Versioned)

ExampleRouterMiddleware_versioned shows inbound handler-version dispatch. Two handlers register for the same topic id under different versions; the router reads the message's benzene-version header off the wire and dispatches to the exact match. A message with no version header routes to the unversioned handler (the default version), and so does one whose version has no exact handler - so turning versioning on for a topic never breaks a producer that doesn't send one.

package main

import (
	"context"
	"fmt"

	benzene "github.com/daniellepelley/benzene-go"
)

type greetReq struct {
	Name string `json:"name"`
}

type greetResp struct {
	Greeting string `json:"greeting"`
}

func main() {
	registry := benzene.NewRegistry()
	mustRegister := func(topic benzene.Topic, greeting string) {
		if err := benzene.Register(registry, topic,
			benzene.Handler[greetReq, greetResp](func(_ context.Context, req greetReq) benzene.Result[greetResp] {
				return benzene.Ok(greetResp{Greeting: greeting + req.Name})
			})); err != nil {
			panic(err)
		}
	}
	mustRegister(benzene.NewTopic("greet"), "Hello ")               // the default (unversioned) handler
	mustRegister(benzene.NewTopic("greet").WithVersion("2"), "Hi ") // the v2 handler

	pipeline := benzene.NewPipeline(benzene.RouterMiddleware(registry))
	greet := func(headers map[string]string) string {
		ic := benzene.NewInvocationContext(benzene.NewTopic("greet"), headers, greetReq{Name: "World"}, nil)
		if err := pipeline.Run(context.Background(), ic); err != nil {
			panic(err)
		}
		return ic.Result.ResultPayload().(greetResp).Greeting
	}

	fmt.Println(greet(map[string]string{"benzene-version": "2"})) // exact match -> v2
	fmt.Println(greet(nil))                                       // no version -> default
	fmt.Println(greet(map[string]string{"benzene-version": "9"})) // unknown version -> default (non-regressive)
}

Output:
Hi World
Hello World
Hello World

type Pipeline

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline is an ordered list of middleware. The first registered is outermost.

func NewPipeline

func NewPipeline(middlewares ...Middleware) *Pipeline

NewPipeline builds a Pipeline from middlewares in registration order. The terminal message router (see RouterMiddleware) is an ordinary middleware and, per core-concepts.md §4, is conventionally registered last.

func (*Pipeline) Run

func (p *Pipeline) Run(ctx context.Context, ic *InvocationContext) error

Run executes the pipeline exactly once for ic. One transport event (one HTTP request, one queue message, ...) is exactly one Run call, per core-concepts.md §4 - a batch delivery is one Run per message, each with its own InvocationContext/Scope; arranging that is the transport binding's responsibility, not Pipeline's.

type Registry

type Registry struct {
	// contains filtered or unexported fields
}

Registry holds (topic -> handler) registrations.

The concept behind handler discovery is explicit registration (core-concepts.md §9); Register is that explicit path, and is the ONLY mechanism this Go port provides. Go has no reflection-based assembly-scanning culture equivalent to C#'s [Message("topic")] attribute scanning, and core-concepts §9 already requires explicit registration to be a first-class path in every language regardless - so there is nothing to defer to later here, this is simply the Go idiom.

func NewRegistry

func NewRegistry() *Registry

NewRegistry returns an empty Registry.

func (*Registry) Has

func (r *Registry) Has(topic Topic) bool

Has reports whether a handler is registered for topic.

func (*Registry) TopicTypes

func (r *Registry) TopicTypes(topic Topic) (request, response reflect.Type, ok bool)

TopicTypes returns the request and response types captured when topic's handler was registered (reflect.TypeOf TReq and TRes), or ok = false when topic isn't registered. Startup-time introspection for service self-description - not a dispatch mechanism.

func (*Registry) Topics

func (r *Registry) Topics() []Topic

Topics returns every registered topic, sorted by ID then Version. This is the enumeration behind service self-description (the mesh package's Descriptor): explicit registration means the Registry is the complete, authoritative list of what this service serves, so a catalog derived from it cannot drift from the running code.

type Result

type Result[T any] struct {
	Status Status
	// Payload is present on success (and optionally on failure). It's a pointer so
	// "absent" is representable without colliding with T's own zero value.
	Payload *T
	// Errors holds zero or more human-readable error messages, populated on failure.
	Errors []string
	// contains filtered or unexported fields
}

Result is the outcome of a single handler invocation (docs/specification/core-concepts.md §5 in the main Benzene repo). Results are values, not exceptions - a transport binding translates a non-success Status into that transport's native failure signal.

Example

ExampleResult shows how a handler signals its outcome with the shared, wire-level status vocabulary rather than a Go error: Ok for success and BadRequest/NotFound/... for the failure modes. Every transport maps the same status the same way (an HTTP code, a gRPC code, a queue ack/nack), so the handler names the outcome once and stays transport-agnostic.

package main

import (
	"context"
	"fmt"

	benzene "github.com/daniellepelley/benzene-go"
)

type greetReq struct {
	Name string `json:"name"`
}

type greetResp struct {
	Greeting string `json:"greeting"`
}

func main() {
	lookup := func(_ context.Context, req greetReq) benzene.Result[greetResp] {
		switch req.Name {
		case "":
			return benzene.BadRequest[greetResp]("name is required")
		case "nobody":
			return benzene.NotFound[greetResp]("no such user")
		default:
			return benzene.Ok(greetResp{Greeting: "Hello, " + req.Name + "!"})
		}
	}

	for _, name := range []string{"World", "", "nobody"} {
		result := lookup(context.Background(), greetReq{Name: name})
		fmt.Printf("%-8q -> %s\n", name, result.ResultStatus())
	}
}

Output:
"World"  -> ok
""       -> bad-request
"nobody" -> not-found

func Accepted

func Accepted[T any](payload T) Result[T]

Accepted returns a successful Result with StatusAccepted.

func BadRequest

func BadRequest[T any](errors ...string) Result[T]

BadRequest returns a failed Result with StatusBadRequest.

func Conflict

func Conflict[T any](errors ...string) Result[T]

Conflict returns a failed Result with StatusConflict.

func Created

func Created[T any](payload T) Result[T]

Created returns a successful Result with StatusCreated.

func Deleted

func Deleted[T any](payload T) Result[T]

Deleted returns a successful Result with StatusDeleted.

func Fail

func Fail[T any](status Status, errors ...string) Result[T]

Fail returns a failed Result with the given status and error messages. The result is always unsuccessful - even for an application-defined status that IsFailure does not recognise - which is what makes a custom failure status nack/redeliver on a queue and render its errors rather than being mistaken for a success. Panics if status is in the framework success class, since that would produce a self-contradictory Result.

func Forbidden

func Forbidden[T any](errors ...string) Result[T]

Forbidden returns a failed Result with StatusForbidden.

func Ignored

func Ignored[T any](payload T) Result[T]

Ignored returns a successful Result with StatusIgnored - handled deliberately, not an error.

func NotFound

func NotFound[T any](errors ...string) Result[T]

NotFound returns a failed Result with StatusNotFound.

func NotImplemented

func NotImplemented[T any](errors ...string) Result[T]

NotImplemented returns a failed Result with StatusNotImplemented.

func Ok

func Ok[T any](payload T) Result[T]

Ok returns a successful Result with StatusOk.

func ServiceUnavailable

func ServiceUnavailable[T any](errors ...string) Result[T]

ServiceUnavailable returns a failed Result with StatusServiceUnavailable - also the mapping used for uncaught handler panics and client-side send failures.

func SetResult

func SetResult[T any](status Status, payload T, successful bool) Result[T]

SetResult builds a Result whose success classification is set explicitly, decoupled from the status class. The intended use is the reserved health check returning StatusServiceUnavailable - so an HTTP probe sees 503 and a load balancer drains the instance - while still rendering its report body (successful=true) rather than an error payload. For ordinary results prefer Ok/Fail and the status-derived default; reach for this only when the transport outcome and the body's meaning genuinely diverge.

func Timeout

func Timeout[T any](errors ...string) Result[T]

Timeout returns a failed Result with StatusTimeout - a downstream deadline elapsed; transient, but whether the operation was applied is unknown, so blind retries are only safe for idempotent operations (unlike StatusServiceUnavailable, WithRetry does not retry this status by default).

func TooManyRequests

func TooManyRequests[T any](errors ...string) Result[T]

TooManyRequests returns a failed Result with StatusTooManyRequests - throttled/rate limited; transient, safe to retry after backing off.

func Unauthorized

func Unauthorized[T any](errors ...string) Result[T]

Unauthorized returns a failed Result with StatusUnauthorized.

func UnexpectedError

func UnexpectedError[T any](errors ...string) Result[T]

UnexpectedError returns a failed Result with StatusUnexpectedError.

func Updated

func Updated[T any](payload T) Result[T]

Updated returns a successful Result with StatusUpdated.

func ValidationError

func ValidationError[T any](errors ...string) Result[T]

ValidationError returns a failed Result with StatusValidationError.

func (Result[T]) IsSuccessful

func (r Result[T]) IsSuccessful() bool

IsSuccessful reports whether this result should be treated as a success. Unless an explicit flag was set via SetResult, it is derived from the status class as "not a failure" (core-concepts.md §5), so a framework success status and an application-defined status both count as successful and carry their payload, while only a framework failure status does not - the extensibility promise that custom statuses flow through untouched (design-principles.md).

func (Result[T]) ResultErrors

func (r Result[T]) ResultErrors() []string

func (Result[T]) ResultIsSuccessful

func (r Result[T]) ResultIsSuccessful() bool

ResultIsSuccessful exposes IsSuccessful on the type-erased ResultInfo path. A transport binding renders the payload vs an error body from the ResultInfo it holds, and this lets an explicit success flag (SetResult) survive type erasure; a binding checks for it via the optional interface { ResultIsSuccessful() bool } and falls back to the status otherwise.

func (Result[T]) ResultPayload

func (r Result[T]) ResultPayload() any

func (Result[T]) ResultStatus

func (r Result[T]) ResultStatus() Status

type ResultInfo

type ResultInfo interface {
	ResultStatus() Status
	ResultErrors() []string
	// ResultPayload returns the payload as `any` (nil if absent) for generic serialization.
	ResultPayload() any
}

ResultInfo is the type-erased view of a Result[T], implemented by every instantiation. The registry stores handlers behind a non-generic dispatch signature (Go generics can't hold heterogeneous Result[T] instantiations in one collection), so transport bindings and the pipeline recover status/errors/payload through this interface instead of the concrete generic type, which they can't name without knowing T.

type RouterOption

type RouterOption func(*routerConfig)

RouterOption configures RouterMiddleware.

func WithVersionKeys

func WithVersionKeys(keys ...string) RouterOption

WithVersionKeys sets the ordered fallback list of header names the inbound payload schema version is read from (versioning.md §2.1) - first present wins, matched case-insensitively. Pass a service's ApplicationBuilder.ReservedNames.Version() so a reserved-name override made once via UseReservedNames drives routing here too, or a literal list to narrow or replace it (e.g. WithVersionKeys("benzene-version") when a producer already emits a "version" header meaning something unrelated). Unset, RouterMiddleware uses wire.DefaultVersionKeys.

type Scope

type Scope struct {
	// contains filtered or unexported fields
}

Scope resolves services for a single invocation. GetService/TryGetService are the only resolution operations (core-concepts.md §8).

func ScopeFromContext

func ScopeFromContext(ctx context.Context) (*Scope, bool)

ScopeFromContext retrieves the Scope previously attached with ContextWithScope, ok = false if ctx carries none (e.g. in a unit test that calls a handler directly).

type Status

type Status string

Status is a Benzene result status: a wire-level string, not a closed enum, so applications can extend it (docs/specification/wire-contracts.md §3). The values below are the framework-defined vocabulary. Their wire strings are lowercase-kebab-case and case-sensitive (e.g. "not-found", "validation-error") - that casing is the wire contract shared by every Benzene port, so the Go identifier names are PascalCase per Go convention while the string values are held verbatim to the spec (never emit the identifier name as the wire value).

const (
	StatusOk                 Status = "ok"
	StatusCreated            Status = "created"
	StatusAccepted           Status = "accepted"
	StatusUpdated            Status = "updated"
	StatusDeleted            Status = "deleted"
	StatusIgnored            Status = "ignored"
	StatusBadRequest         Status = "bad-request"
	StatusValidationError    Status = "validation-error"
	StatusUnauthorized       Status = "unauthorized"
	StatusForbidden          Status = "forbidden"
	StatusNotFound           Status = "not-found"
	StatusConflict           Status = "conflict"
	StatusTooManyRequests    Status = "too-many-requests"
	StatusTimeout            Status = "timeout"
	StatusNotImplemented     Status = "not-implemented"
	StatusServiceUnavailable Status = "service-unavailable"
	StatusUnexpectedError    Status = "unexpected-error"
)

The framework-defined status vocabulary (wire-contracts.md §3). The string values are the case-sensitive lowercase-kebab-case wire contract - do not translate them to the Go identifier casing.

func (Status) IsFailure

func (s Status) IsFailure() bool

IsFailure reports whether status is one of the framework-defined failure statuses. It is false for success, unknown (application-defined), and empty statuses - an application-defined status is not assumed to be a failure, which is what keeps custom statuses flowing through the pipeline, envelope, and mesh untouched (design-principles.md).

func (Status) IsKnown

func (s Status) IsKnown() bool

IsKnown reports whether status is part of the framework-defined vocabulary (success or failure).

func (Status) IsSuccess

func (s Status) IsSuccess() bool

IsSuccess reports whether status is one of the framework-defined success statuses (StatusOk, StatusCreated, StatusAccepted, StatusUpdated, StatusDeleted, StatusIgnored). It is false for failure, unknown (application-defined), and empty statuses. This is the narrow classifier the per-protocol mapping tables use for their generic-success row; for deciding whether an invocation succeeded, prefer IsFailure/Result.IsSuccessful, which do not treat an application-defined status as a failure (design-principles.md §"custom statuses").

type Topic

type Topic struct {
	ID      string
	Version string
}

Topic identifies a message type and routes it to a handler, per docs/specification/core-concepts.md §2 in the main Benzene repo (the spec this package implements).

A (ID, Version) pair maps to at most one handler. When a message arrives without a version, the unversioned handler (Version == "") handles it; versioned handlers are selected only by an exact match. RouterMiddleware reads an inbound message's version off the wire (wire-contracts.md §2 tier C) and, when a signalled version has no exact handler, falls back to the unversioned one - see its doc.

func NewTopic

func NewTopic(id string) Topic

NewTopic returns an unversioned Topic with the given id.

func (Topic) String

func (t Topic) String() string

func (Topic) WithVersion

func (t Topic) WithVersion(version string) Topic

WithVersion returns a copy of the topic with the given version.