Events

craftgo generates event code the way protoc generates message code: payload types, their validation, one descriptor per contract, and nothing else. Which events a deployable listens to, on which group, behind which middleware, is that deployable's own Go.

The model

The contract sits in the design, between one producer and any number of listeners.

One producer publishes an event, and the design does not know how many listeners exist or where they run. Each listener joins under a group; every group receives every message once, and replicas inside a group divide that work. The runtime half is the bus - events.Bus is server.Server for the listener side, the one thing every subscription passes through.

Declaring events

An event is a file-level declaration, like type; a service holds HTTP methods only. Files group into packages by their package declaration, so the usual layout is one folder per version and one event per file, a new version published beside the old one until every listener has moved.

design/orders/
├── v1/
│   ├── placed.craftgo         package ordersv1
│   └── shipped.craftgo        package ordersv1
└── v2/
    └── placed.craftgo         package ordersv2

package orders

type OrderPlaced {
    orderId string @minLength(1)
    total   int64  @gte(0)
}

@doc("An order was accepted.")
event Placed {
    payload OrderPlaced
}

payload must name a type, so every contract has named fields and its own Validate(). The payload is one JSON message, so a field it reaches may not carry @path, @query, @header, @cookie or @form, nor hold a file. Events have their own namespace, so type OrderPlaced and event OrderPlaced coexist. A payload may also be an array of a declared type - a JSON array on the wire, typed on the slice and validated element by element:

event BatchPlaced {
    payload OrderPlaced[]
}

The wire identity defaults to <package>.<Event> - orders.Placed above - and @contract("payments.settled.v1") overrides it to match a name another system already publishes. Two events resolving to one identity are rejected at design time. The only event decorators are @contract, @doc and @deprecated.

What is generated

One file per design package that declares an event, and nothing beside it: <events out>/<pkg>/events.go.

// Code generated by craftgo. DO NOT EDIT.

package orders

import (
	craftevents "github.com/craftgodotdev/craftgo/pkg/events"

	types "example.com/app/internal/types/orders"
)

// PlacedContract is the wire identity of Placed.
const PlacedContract = "orders.Placed"

// An order was accepted.
//
// Placed is the orders.Placed event contract.
var Placed = craftevents.NewEvent[types.OrderPlaced](PlacedContract, (*types.OrderPlaced).Validate)

@doc, or the comment above the event when it has none, heads the descriptor's comment, an empty // line above the line craftgo writes. The validator is the payload's Validate, or for an array payload a generated function that validates each element. A descriptor holds no bus - the bus is a parameter at the call - so one contract package serves every deployable that imports it.

There is no handler interface, no registration function, no publisher type and no transport adapter. Where the file lands is events.targets[].out; see Configuration.

Publishing

The descriptor validates and encodes; the transport under the bus decides where the bytes go.

err := orders.Placed.Publish(ctx, bus, &types.OrderPlaced{OrderID: order.ID, Total: order.Total},
	craftevents.WithKey(string(order.ID)))

Publish validates first: a payload that does not validate is a *PayloadError and nothing goes on the wire. The options fill in everything beside the payload - WithKey (the entity the message is about, used by a transport that orders per entity), WithDedupID (JetStream drops a repeat within its stream's duplicate window; Kafka, core NATS and memory carry the ID to the consumer and act on nothing), WithHeader, WithAdapterOption. bus.PublishAll(ctx, envs) publishes a batch and reports a partial failure as a *PartialPublishError naming the indices that did not go out.

A transport adapter is anything implementing one method:

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

An outbox is that method writing the encoded message to a table in the same transaction as the business write. A drainer then reads the table and calls bus.PublishAll against a bus wired to the real broker, each stored payload a wire.Raw so the codec carries its bytes through without encoding them again. Nothing generated changes, because nothing generated names a broker.

Consuming

Everything between the broker and your method is the same for every subscription on the bus.

A listener is one line: the descriptor, the group it joins, and the method that handles it.

// internal/handler/orders/handler.go
var (
	NotificationGroup = craftevents.Group("order-notifications")
	LedgerGroup       = craftevents.Group("order-ledger")
)

func Register(bus *craftevents.Bus) error {
	notifier, ledger := Notifier{}, Ledger{}
	return errors.Join(
		orders.Placed.Subscribe(bus, NotificationGroup, notifier.SendReceipt),
		orders.Shipped.Subscribe(bus, NotificationGroup, notifier.SendDispatchNote),
		orders.Placed.Subscribe(bus, LedgerGroup, ledger.BookOrder),
	)
}

Subscribe(bus, group, fn) is typed against the contract: fn takes (ctx, *types.OrderPlaced) error with the payload already decoded and validated, so a method with the wrong signature does not compile at the call. errors.Join offers every line to the bus, so a refusal in the middle neither hides the refusals beside it nor cancels the registrations after it; each is a *RegisterError naming its contract and group.

main.go is where the bus, its chain and the modules meet:

bus := craftevents.New(craftevents.WithTransport(tr), craftevents.WithCodec(codecjson.Codec{}))
bus.Use(logging.AccessLog(craftlog.Slog()))

if err := handler.Register(bus); err != nil {
	return err
}
return bus.Start(ctx)

There is no default codec, and WithPublisher or WithSubscriber alone is enough for a binary that only publishes or only listens. Start hands the whole batch to the transport in one call, every handler wrapped; a second Start, or a Register after one, is ErrStarted.

Groups

A group is the broker identity a subscription joins: the Kafka consumer group, the NATS queue group, the JetStream durable. Subscriptions sharing one divide the stream between them, so a group is the unit of scaling and of failure isolation - not of ordering. events.Group is a named type so an application declares its groups once, as values beside the registrations that use them, rather than as loose strings. Register refuses an empty group. On a transport that remembers a position per group - Kafka, JetStream - the name is where those consumers resume: if it has an offset, write the name down.

Middleware

A consumer middleware is ordinary Go, never a declaration:

type Middleware func(sub craftevents.Subscription, next craftevents.Handler) craftevents.Handler

sub carries the contract, the consumer and the group being wrapped, so one chain can behave differently per group. Chains compose outermost first: Use(A, B, C) wraps a handler as A(B(C(h))). The chain belongs to the bus, the only thing every subscription passes through - WithMiddleware installs it at construction, bus.Use appends afterwards, and Use after Start panics. One subscription that needs its own wrap is built with the descriptor's Subscription(bus, group, fn), given a Chain, applied inside the bus chain, and handed to bus.Register. A failed decode or validation reaches the chain as *PayloadError, a message stamped with another codec as ErrCodecMismatch, a panicking handler as *PanicError.

Dispositions

A middleware asks for something other than "done" through the message: msg.Settle() takes the delivery, msg.Redeliver() hands it back, msg.Reject() gives it up. Asking for nothing settles.

A frame that panicked did not finish deciding, so what it asked for is dropped. A panicking handler leaves the message undecided and the chain above it decides, as it does for any other error. A panic in a middleware unwinds past the whole chain, leaving nothing above to decide - so the bus asks for redelivery on a transport that can honour one, and the delivery is retried rather than acked away.

Only some transports can honour this

JetStream and a Kafka share group track each record; a classic Kafka consumer group, core NATS and the in-process transport do not, and there Redeliver settles instead. Name what you need with WithDispositionRequired and Register refuses a transport that cannot honour it.

The plan

No generated file states what a deployable listens to, so bus.Plan() reports it instead, before or after Start. It sorts groups by name and consumers by contract, and renders that order however the plan was built, which makes it a golden file:

bus := craftevents.New(craftevents.WithTransport(memory.New()), craftevents.WithCodec(codecjson.Codec{}))
if err := handler.Register(bus); err != nil {
	t.Fatal(err)
}
got, err := json.MarshalIndent(bus.Plan(), "", "  ")
// compare got against testdata/plan.json

A rename, a lost listener or a group that drifted between two deployables then fails a test rather than a deploy.

NATS JetStream

A group is a durable name, and the durable's filter set is the plan this process registered.

nats.NewJetStream(conn, opts...) reads from a stream, and is what makes Redeliver and Reject mean something on NATS. It creates no stream: Start refuses a subject no stream carries. A durable reads one stream, so a group whose subjects span two is refused at start-up. At Start an existing durable is verified, not reshaped:

carried filter vs plan	what happens
equal	adopted as is
a strict subset of the plan	widened and logged - this version added a listener
anything else	refused, naming both sets

"Anything else" is a plan that drops subjects, one that only partly overlaps, or a durable with no filter at all. Re-pointing any of those stops those subjects reaching anyone, so it is refused unless the group carries nats.AllowNarrow() - a deliberate removal.

Per-group settings

Delivery settings belong to the group, because the group is what the server keeps them on:

js, err := nats.NewJetStream(conn,
	nats.WithGroupConfig(LedgerGroup, nats.DeliverPolicy(jetstream.DeliverNewPolicy),
		nats.MaxInFlight(8), nats.AckWait(2*time.Minute)))

MaxInFlight, AckWait, DeliverPolicy, ConsumerConfig and AllowNarrow are the group options, and repeated calls for one group accumulate. AckWait, DeliverPolicy and ConsumerConfig apply when the durable is created, and AllowNarrow when Start checks an existing one; WithMaxInFlight and WithAckWait are the transport-wide defaults.

MaxInFlight is how many messages one durable's pull keeps buffered here, and the default is 1 deliberately: a buffered message waits for every handler ahead of it with the server's AckWait clock already running. Raise it only where n × the slowest handler stays under AckWait, or a message is redelivered while it still sits in the buffer. WithMaxDeliveries (default 5) caps a redelivery loop and WithRedeliverBackoff(fn) delays each redelivery by fn(deliveries).

Rolling deploys

During a rolling deploy two versions of a deployable share a durable. A subject no listener in this process handles is handed back - NAK'd with the group's AckWait as the delay - so the replica that does handle it gets it, and never terminated. Every hand-back is reported through WithJetStreamErrorHandler with only sub.Group set.

Kafka, core NATS and memory

• pkg/events/kafka - one contract per topic, ordering key as the record key; WithShareGroup asks for a KIP-932 share group instead of a classic consumer group.
• pkg/events/nats (core) - contract to subject, group to queue group; at most once and no nack, so install nats.WithErrorHandler or a failed message is observed by nothing.
• pkg/events/memory - in-process, for tests, single-binary deployments and the plan golden test; Drain() waits for in-flight deliveries.

Every option and error type is listed in Runtime API.

Who owns what

The design is the contract two deployables share; everything operational is the application's.

The design holds the event declaration and its @contract name, the payload types and their validators, one folder per version, and the HTTP services beside them. The application holds which events this deployable listens to, the group per subscription, the middleware on bus.Use, and its own folder layout, retries and transport settings.

Nothing in between is declarable: no listener declaration, no group decorator, no middleware declaration for listeners, no projection. A group or a chain in the design would tie a shared contract to one deployable's operations.