Write an Aggregate

An aggregate is the unit that decides: it takes one command at a time, looks at its own state, and decides the events. This page writes one, with the eShop's order as the example. Deciding and Evolving explains the model behind it.

Where the code goes

An aggregate spans two modules of its context:

Module	Holds	Read by
*-api	The commands and events, plain data classes	The service, and every peer that sends its commands or reads its events
*-domain	The root class and its state class	The service only

The compiler reads the domain module at build time and generates the aggregate's model, its binding and its part of the Manifest. Nothing is scanned at run time.

Write the commands

A command is a data class of the api module. It holds what the caller asks, never who the caller is: the tenant, owner and space come from the caller, and the expected version from If-Match. The aggregate's id comes from the route or from the saga that sends the command, unless the body already names it (Name the aggregate in the body).

/**
 * Submits an order of [items]. The basket context sends it when its buyer checks out, with the order's id fixed by the
 * basket. The order belongs to the buyer: a saga's command carries the scope of the basket it is derived from, whose
 * owner is its buyer, and the order takes its owner from the command that creates it. An order has
 * items, each valid ([OrderItem]'s constraints): the gateway checks that before the order decides.
 */
@CreateAggregate
@Summary("Submit an order")
data class SubmitOrder(
    @field:NotEmpty @field:Valid val items: List<OrderItem>,
)

@Summary gives the command a one-line summary for the Manifest and the OpenAPI document. A command that takes no data is a data object, such as ShipOrder.

Check its shape

A rule about the command alone, such as a positive quantity, a name that is not blank or an order that has items, is a Jakarta Bean Validation constraint on the command class, with Kotlin's @field: target:

/**
 * Adds [quantity] of [productId] at [unitPrice] to the basket, creating the basket on its first item. Adding a product
 * already in the basket adds to its quantity, at the new price. The constraints are the command's shape: the gateway
 * checks them before the basket decides, and answers a violation `wow.command.invalid` (400).
 */
@AllowCreate
@Summary("Add an item to the basket")
data class AddBasketItem(
    val productId: String,
    @field:PositiveOrZero val unitPrice: BigDecimal,
    @field:Positive val quantity: Int,
)

The gateway checks the constraints of every command it takes before the aggregate decides: a command from a route, from an in-process CommandClient, from a saga, from a timer or from another context. A violation is answered wow.command.invalid (400), and the problem lists each one with its property path (items[0].quantity), its constraint (positive, the annotation's name on the wire) and its message. A timer's command it refuses is that timer's rejected command: logged with its violations, never decided, and not retried. A nested value is checked when its property is @field:Valid, as SubmitOrder's items are. The api module depends on jakarta.validation:jakarta.validation-api, and the service needs a Bean Validation provider (Host on Spring Boot).

The constraints are part of the command's contract. They are the validation keywords of its schema in the Manifest and the OpenAPI document (quantity is {"type": "integer", "format": "int32", "exclusiveMinimum": 0}), so a generated client or Swagger UI sees them before it sends. Tightening one, such as raising a minimum, is a breaking change (The Manifest and Compatibility).

A rule that needs the aggregate's state or the caller stays in the command function, as a rejection with an error code (Rejections and Error Codes): the order's function below has nothing left to check.

Whether a command may create its aggregate is part of its class:

The command class	Is accepted when
@CreateAggregate	The aggregate does not exist yet; otherwise wow.aggregate.exists
@AllowCreate	Whether or not the aggregate exists
Neither	The aggregate exists; otherwise wow.aggregate.not-found

Each command is served at POST /<context>/<aggregate>/{id}/<command>, the command's name in kebab case, as the Manifest lists its routes: POST /eshop-ordering/order/{id}/submit-order. A creating command is also served without {id}, and the server gives the aggregate its id. Admission first looks for the aggregate the same request id already created and reuses its id, so a creation sent twice creates one aggregate. An aggregate whose id is its owner's, one per owner, serves every command without {id} too, as the caller's own, and generates no id (One aggregate per owner).

Name the aggregate in the body

Some commands already hold their aggregate's id in their business data: a restock names the product whose stock it adds to. @AggregateId on that property makes the body name the aggregate, so a sender gives no id beside it:

@AllowCreate
@Summary("Restock a product")
data class RestockProduct(
    @AggregateId val productId: String,
    @field:Positive val quantity: Int,
)

Such a command is served at POST /<context>/<aggregate>/<command> alone, without {id}, and it acts on the aggregate its body names. A saga sends it with CommandRequest.of(body) and a client without an id. A command for another context goes to that context's command topic keyed by its aggregate, so the commands of one aggregate keep their order. Admission reads the body before it authorizes the command, so a policy still sees the aggregate, <aggregate>/<id>. The aggregate's lifecycle commands keep their routes with {id}.

The property	Holds
A String with a field or a constructor parameter	The id, as the body holds it
A computed String, val stockId: String get() = "$warehouseId-$sku"	The id, computed from the other fields: a sender that sends only those leaves it out, and the service computes it
A String? of a command that may create its aggregate	The id, or null to leave it to the server, as for any creation that names none; a retry keeps the first one

Admission checks the id: a blank one is wow.command.malformed (400), and so is a request that names another aggregate beside the body, such as a Kotlin client's CommandRequest.to(ref, body). A request that names the same one is accepted.

The property is declared by the command class or by an interface it implements. It is public or internal, and Wow's JSON writes it (it is not @JsonIgnored), so every sender's body holds it. One property of a command names its aggregate. The compiler refuses it on an @Unaddressed or mounted command, whose transformer finds the aggregate, and on an aggregate whose id is its owner's. A CommandTransformer that takes such a command stops the service at start.

Wow 10 takes a command's aggregate from its body only by @AggregateId, never by a property's name. So the compiler warns of a command that has a String property named id and no @AggregateId, where it could name its aggregate: the id may be stored or computed, the command's own or an interface's, such as a 9.x Identifier.id. If that id names the aggregate, mark it @AggregateId, as a 9.x command whose body named its aggregate needs, a creation a saga sends without an id among them. If 9.x bound it from the path, it only repeated the route: remove it and read command.ref.id (What a command body holds). An id that names something else stays, and @Suppress("UnmarkedAggregateId") on the command class silences the warning.

Keep a command internal

Some commands record what the service's own processes found, never what a client asks: the payment gateway's answer, a timer's verdict, a step only a saga takes. @InternalCommand on the command class keeps it off the public port:

/**
 * Records that the gateway approved the payment, under its [transactionId]. The gateway saga sends it, and only it:
 * the command is internal, so the public port serves no route of it, and a client cannot confirm a payment the gateway
 * never charged.
 */
@InternalCommand
@Summary("Confirm a payment")
data class ConfirmPayment(
    val transactionId: String,
)

Only the platform sends an internal command: the service's sagas and timers, its in-process CommandClient, and another service's call on the internal port, which wow.callers.<service>.commands permits as it permits any command (Call Other Services). The Manifest lists it with "internal": true and the routes any command of its kind has, and the internal port serves them on the same paths. The public port does not: a request to POST /eshop-payment/payment/{id}/confirm-payment there is answered 404, the problem of any unknown path, which does not say the command exists. Its OpenAPI document and the clients wow-generator generates leave the command out, with the types only it uses. The command authorizer still decides every call the internal port takes.

The 404 keeps a probe from telling an internal command from an unknown path; the contract itself stays public. GET /wow/manifest on the public port lists the command, marked "internal": true, as it lists every command: tools that call the service by name read the contract there, and the sagas' derives and the wait chains name the command. A deployment that does not publish its contract turns that endpoint off with wow.http.manifest.enabled: false.

The eShop's ConfirmPayment and FailPayment are internal: the gateway saga records the gateway's answer with them, and a buyer who could send ConfirmPayment would pay with no charge. Making a public command internal removes its public routes, a breaking change the compatibility check reports; making an internal command public adds them (The Manifest and Compatibility). The system commands, the lifecycle commands among them, are never internal.

Write the root

The root is a class annotated @AggregateRoot, built from its state. Its command functions decide: each takes a command and returns what it decided.

fun onCommand(command: SubmitOrder): OrderSubmitted = OrderSubmitted(command.items)

The compiler finds every non-private function annotated @OnCommand, or named onCommand. The first parameter is the command; the return value is the decision:

Returns	Means
Decision.events(e1, e2, …)	These events, appended together as one event stream
Decision.NoOp	The command is accepted and changes nothing; nothing is written
Decision.reject(code, message)	The command is refused; nothing is written, and the caller gets the code (Rejections and Error Codes)
An event, or a list of events	These events
Unit, null or an empty list	Nothing changes

A function that returns Decision does not tell the compiler which events it can produce, so it lists them in @OnCommand(produces = [...]). The Manifest publishes that list. A function whose return type already names its events, such as an event class or a sealed type of them, does not repeat them.

A function that returns Decision (or Any, or a type parameter) and lists no produces compiles, with a warning: its events are not in the Manifest, so the kernel cannot refuse an event it was not meant to produce, and the clients and peers do not know what it commits. List them in produces. A function that only rejects or changes nothing has no event to list; @Suppress("UnlistedEvents") on it says so and silences the warning. A command function inherited from a library is not warned of in the module that subclasses it: its author lists what it produces.

A return type that is an interface or abstract class, and not sealed, is a compile error. Such a function returns one of several implementations, and the type does not say which ones. List them instead: make the type a sealed interface whose subclasses are the events, return the event classes themselves, or return a Decision with produces. The same holds for an element type (List<OrderEvent>) and for a subclass of a sealed type.

The aggregate is named after the root class in snake case (Order is order, ViewPreferences is view_preferences), or by @Name.

Read the admitted command

A command function that needs what admission learned about the request takes Command<X> instead of X:

Member	Holds
body	The command X
ref	The aggregate: its context, name, tenant and id
scope	The caller's owner and space, as the aggregate's ownership keeps them: a tenant's aggregate (Ownership.TENANT) sees no owner, and one whose id is its owner's takes its id for the owner of a derived command or of one with none
headers	The request facts the host keeps (wow.http.request-facts), such as the calling application
causation, operator	The chain the command belongs to, and whose name it is recorded in

A feedback form that records the application it was sent from reads the request fact:

fun onCommand(command: Command<SubmitFeedback>): FeedbackSubmitted =
    FeedbackSubmitted(command.body.text, app = command.headers["app_id"])

A function need not check that the caller may reach the aggregate: the kernel does it from the scope, or from the aggregate's access policy (What the kernel checks). An aggregate that exists once per owner, such as a basket, takes its owner's id (One aggregate per owner).

Take domain services

The parameters after the command are domain services, or the DecideScope. A service becomes a parameter of the generated model, and the host binds it once, when it builds the model: it takes the application's only bean of that type, or its primary one. @Name("…") on the parameter takes the bean of that name instead, so two parameters of one type can take two beans.

The order takes its payment policy as a service, and schedules its own cancellation with the DecideScope once its stock is confirmed:

/**
 * Confirms one product. A confirmation the order does not wait for changes nothing: one read again from the
 * stock stream, or one that arrives after another product's shortage cancelled the order (that reservation is
 * released when the inventory hears of the cancellation).
 */
// Decision: the last confirmation produces two events, the item's and the order's.
@OnCommand(
    produces = [OrderStockItemConfirmed::class, OrderStockConfirmed::class],
    schedules = [CancelOrder::class],
)
fun onCommand(command: ConfirmOrderStock, paymentPolicy: PaymentPolicy, scope: DecideScope): Decision {
    val waiting = state.status == OrderStatus.AWAITING_VALIDATION &&
        command.productId in state.products() &&
        command.productId !in state.confirmedProducts
    if (!waiting) {
        return Decision.NoOp
    }
    val confirmed = OrderStockItemConfirmed(command.productId)
    if (state.confirmedProducts + command.productId != state.products()) {
        return Decision.events(confirmed)
    }
    scope.schedule(
        PAYMENT_GRACE_PERIOD,
        ScheduleRule.after(paymentPolicy.gracePeriod),
        CancelOrder(PAYMENT_TIMEOUT)
    )
    return Decision.events(confirmed, OrderStockConfirmed)
}

A service may do I/O. A function that blocks its thread, such as one that calls JDBC, is annotated @Blocking and runs on a virtual thread. A function may also be suspend, or return a Reactor Mono, which needs wow-reactor on the class path.

Query the read side

A command function that needs more than its aggregate's state, such as whether an account it refers to exists, takes the node's read API as a service: a QueryApi, or a NodeQueryApi for trusted reads. Both are in wow-api (me.ahoo.wow.api.read), so the domain module needs nothing more, and the host gives the model the node's own. When the caller already knows the fact, put it in the command instead (a CommandTransformer, or the saga that sends it): a read runs again on every retry after a version conflict, and what it read may change before the events are appended.

Read as the command's caller, rebuilt from the command: its tenant, its owner and space, and its operator. The read sees what that caller sees:

@OnCommand
suspend fun onCommand(command: LendBook, queries: QueryApi, scope: DecideScope): BookLent {
    val admitted = scope.command
    val caller = Caller.of(admitted.ref.tenantId, admitted.scope, admitted.operator?.id)
    queries.of(caller).snapshots<AccountState>().state(command.accountId)
        ?: reject("library.book.no-account", "No account ${command.accountId}.")
    return BookLent(command.accountId)
}

The rebuilt caller has no roles, policies or claims. They live only with the request, and a command may be decided on another node, or taken from its command topic. A read authorizer that reads them sees none, and one that requires a role refuses the read.

For a read that no policy should restrict, take a NodeQueryApi and read with trusted().

Trusted reads cross tenants

trusted() reads every tenant. In a service with several tenants, the function filters by the command's tenant itself, as the sample does.

@OnCommand
suspend fun onCommand(command: LendBook, queries: NodeQueryApi, scope: DecideScope): BookLent {
    val lent = queries.trusted().snapshots<BookState>().count(
        countQuery {
            filter {
                tenantId(scope.command.ref.tenantId)
                "state.account" eq command.accountId
            }
        },
    )
    if (lent.count >= MAX_BOOKS) {
        reject("library.book.too-many", "Account ${command.accountId} has $MAX_BOOKS books.")
    }
    return BookLent(command.accountId)
}

The read side follows the log. Snapshots are written after each commit, not with it, so a query in a command function, or in a saga, may not see a write made a moment before, such as an aggregate a saga changed just now. Design for it: where a missing or stale row would refuse the command, let the caller retry that refusal, or have the sender carry the fact the check needs in the command. The aggregate's own state, which the function receives, is always current.

A read that fails fails the command with wow.command.failed (500), as any unexpected exception does. That includes a read the read side refuses, and one the code gets wrong, such as a state class no aggregate has: the fault is the code's, not the caller's. In an aggregate spec, give the model factory memoryQueries { snapshots(...) } as its QueryApi (Testing).

Ask the kernel for more

The DecideScope is what a decision can ask of the kernel besides its events:

Member	Does
command	The command being decided, as Command<*>
now	The time of the decision on the node's clock, in epoch milliseconds: the commit time its events are recorded at. Compute a deadline or an expiry from it, not from the system clock, so that a test sets it (Testing)
ids	The node's id generators, for an id the decision needs
putResult(key, value)	Adds a value to what the caller is answered once the events are appended. It is not stored: a repeated request is not answered it again
reject(code, message)	Rejects the command with a code, optionally with extensions for the problem (Rejections)
reserve(namespace, key)	Reserves a unique key, such as a user name, for this aggregate; false when another aggregate holds it. A ReservationTenancy argument widens its scope (Unique Keys)
release(namespace, key)	Releases a key this aggregate holds, once the events are appended; with the ReservationTenancy it was reserved with
onAppendFailure { … }	Undoes an outside act when the events are surely not appended (below)
afterAppend { … }	Runs once the events are appended; a failure is logged, not raised (below)
schedule(key, rule, command, target)	Schedules a command, to this aggregate unless target names another. It is written with the events, or not at all
cancelSchedule(key)	Cancels the schedule this aggregate keeps under key

A decision answers the caller with a value it worked out, a quote say, through putResult; the caller reads it from the result's results, here {"cost": 4.50}:

fun onCommand(command: QuoteShipping, scope: DecideScope): ShippingQuoted {
    val cost = rates.costOf(state.weight, command.destination)
    scope.putResult("cost", cost)
    return ShippingQuoted(command.destination, cost)
}

What a command creates is better found by its aggregate id, which a repeated request is answered again. The eShop's checkout answers no order id: the ordering numbers the order it creates (SO-261031-42), and a checkout that waits for the order's commit, committed(eshop-ordering), reads the number as that hop's aggregate id (Sagas).

A function that schedules a command declares the command's class in @OnCommand(schedules = [...]), as the order does with CancelOrder::class: the compiler cannot read the calls in a function's body, and the Manifest lists them by full type name.

Undo an outside act

A decision may act outside the aggregate before its events are appended: authorize a card, hold a seat in another system. onAppendFailure registers how to undo the act, and the kernel runs it when it knows the events were not appended:

@OnCommand
suspend fun onCommand(command: PayOrder, scope: DecideScope): Decision {
    val authorization = payments.authorize(state.totalAmount, key = scope.command.requestId)
    scope.onAppendFailure { payments.void(authorization) }
    return Decision.events(OrderPaid(authorization.id))
}

The compensation runs when:

• the decision refuses the command, in any of the three ways, or appends nothing (Decision.NoOp);
• the decide function throws;
• the append meets a version conflict. Each attempt undoes its own act before the command is decided again on the newer state;
• the store refuses the append, or fails in a way that surely wrote nothing;
• the append finds the request already committed by another attempt.

It does not run when the append's outcome is unknown. The kernel first reads the version the append would have taken. Only when that still cannot tell does it answer wow.command.unknown, and then the events may be committed. Nor can it run when the process dies. An act that must not outlive a lost command needs a clean-up of its own besides: an expiry in the outside system, or a reconciliation that reads the log.

• Order. Compensations run latest first, one after another, and each is awaited, even when the caller has gone away. There is no timeout: a compensation that hangs holds the aggregate's commands, so bound its call.
• Failure. A failing compensation does not stop the others. It is logged, or added to the failure it undoes, and the command's answer does not change.
• Twice. A command refused, or left unchanged, on a cached state that the log has moved past is decided again, its first attempt compensated first. The outside act may then be done twice and undone once, so key it on the command, as requestId does above.
• Reserved keys. reserve registers its own compensation: the hold this attempt took is given up when the events are not appended, while a key the aggregate already held stays held. The keys an attempt reserves and releases travel with its appended events, so when the outcome is unknown, or the process dies after the append, the built-in wow.reservation subscription confirms and releases them from the log. Only a hold that no appended entry carries, because the process died before appending, expires, after 30 seconds, for the next reservation of the key to take.

afterAppend is the other side: it runs once the events are appended, before the caller is answered, in the order the actions were registered. It does not run on a duplicate, nor on any outcome that appended nothing, nor on a rejection, even one that commits events: the compensations run there instead. A failure is logged, and the events stay committed. Subscriptions may see the commit before the action ends, and a process that dies in between skips it. So it suits work that may be lost, such as warming a cache. Work that must follow every commit belongs in a reactor or a saga, which reads the committed events.

Write the state

The state class holds a sourcing function for each event, annotated @OnSourcing or named onSourcing. A sourcing function applies one event: it takes no service and does no I/O.

fun onSourcing(event: OrderSubmitted) {
    items = event.items
    totalAmount = event.items.fold(BigDecimal.ZERO) { total, item ->
        total + item.unitPrice.multiply(BigDecimal(item.quantity))
    }
    status = OrderStatus.SUBMITTED
}

• Every event the aggregate can produce needs a sourcing function, even one that changes nothing.
• A sourcing function that changes the state in place returns nothing. One that returns the state type returns the new state instead, for an immutable state.
• Give every property a value before the first event. The kernel serializes the state as it is, for snapshots and for the published stream.
• A property in a val with no setter reads back into its field from a snapshot, or from a state a query returns: a collection or map the state fills as it goes (val items = sortedSetOf<String>(), a MutableList or a MutableMap) and a value set at construction alike. A computed property, a getter with no field behind it, is written and not read. So hold a collection the state fills in such a val, not behind a getter of its own over a private field (val items: List<Item> get() = lines): that one is not read back, and the kernel replays the aggregate's log instead of loading its snapshot.
• A state and its events hold no type of Wow's query protocol (me.ahoo.wow.api.query: a filter, a sort, a projection, a field). Those are wire types that evolve with Wow, and the build refuses a state or an event that holds one, naming its path: hold a type of your own, such as a list of field names, and build the query from it when it runs. A command or a query holds them freely.
• A command function never changes the state, it only reads it. The test DSL compares the state before and after each decision and fails the test if it changed.
• A member marked @Access is one of the aggregate's access attributes, which its access policy compares with the caller's claims (Declare access attributes).

Construct the initial state

The generated model builds an aggregate's state, before its first event, with the state class's constructor: its primary one, a Java class's first. The compiler takes four shapes and refuses any other:

The constructor	The state is built
()	With no argument. Recommended: the aggregate's id and tenant are in its ref and its event streams, not in its state
Parameters that all have default values	With no argument: each parameter keeps its default
(id: String), with no default	With the aggregate's id
(id: String, tenantId: String), with no default	With the aggregate's id and its tenant

• The id and the tenant are passed by position, whatever the parameters are named.
• A parameter with a default does not receive the id: (id: String = "") is a constructor whose parameters all have defaults, so the state starts with "". Give id no default for the state to receive it. Take the id only when the domain cannot work without it.
• A root that is its own state, with no state class beside it, is built the same way.

Read the commit

A sourcing function may take, after its event, the commit the event belongs to: an EventStream, with the commit time, the operator, the version and the command's ids, as the log records them. The kernel gives the same commit when it decides and when it replays, from the log or from a snapshot and the streams after it, so a state rebuilt later is the same state. A state that keeps a time reads it from the commit, never from a clock:

fun onSourcing(event: OrderPaid, commit: EventStream) {
    paidAt = commit.commitTime
    paidBy = commit.operator?.id
}

• An immutable state takes it the same way: fun onSourcing(event: OrderPaid, commit: EventStream): OrderState.
• Written as code, the model takes it as evolve<OrderPaid> { event, commit -> … }, or evolveTo<OrderPaid> { state, event, commit -> … }.
• The commit time is the decision's scope.now: a deadline the decision computes from it and the time the state reads agree.
• History imported from 9.x keeps each stream's create time and operator as its commit's (Migrate).

Run code after every command

A function annotated @AfterCommand, or named afterCommand, runs after the command functions it applies to: those in include, all of them when include is empty, less those in exclude. It sees the state the command function saw, and returns events, which are appended after the command's own in the same stream. It does not run when the command is rejected, and it cannot reject. When it fails, the whole command fails and nothing is written.

Lifecycle commands

Every aggregate also handles four system commands of me.ahoo.wow.api.lifecycle: DeleteAggregate, RestoreAggregate, TransferOwner and TransferSpace. They change the aggregate's metadata, not its state, and a deleted aggregate refuses every command but RestoreAggregate and those whose function is marked @OnCommand(onDeleted = true). @AggregateRoot(lifecycle = [...]) narrows the list; an empty list enables none.

Test it

Test each command function with aggregateSpec: given these events, when this command, expect these events and this state. Test Domain Code shows how.

Next

• Rejections and Error Codes: declare the codes a command is refused with.
• Write a Saga: derive commands from what an aggregate decided.