3. Commands

Commands are the write path. You declare them with command_engine!, implement the behavior on an aggregate, and call them over HTTP or from code through the engine's typed helpers.

Command kinds

Kind	Target	HTTP suffix	Use for
`collection`	`CommandTarget::Collection(resource)`	`:post`	Creating, or acting on many rows (`create-todo`, `clear-done`)
`object`	`CommandTarget::Object(AggregateRoot)`	`:exec`	Acting on one row (`update-todo`)

Declare the command engine

domain/src/command.rs
use std::sync::Arc;

use riverbase_core::base::AggregateRoot;
use riverbase_core::command::{CommandTarget, MessageBus};

use crate::aggregate::TodoAggregate;

type TodoMessageBus = Arc<dyn MessageBus>;

pub const TODO_RESOURCE: &str = "todo";

riverbase_core::command_engine! {
    engine TodoCommandEngine {
        meta {
            actor_pool_size: 4,
        }
        aggregate: TodoAggregate,
        component: {
            msgbus: TodoMessageBus,
        },
    }
    commands {
        command CreateTodo {
            meta {
                key: "create-todo",
                title: "Create Todo Item",
                kind: collection,
                resources: [TODO_RESOURCE],
            }
            payload {
                title: String,
            }
            target collection(TODO_RESOURCE);
            handle (payload, aggregate) {
                aggregate.create(&payload.title).await
            }
        }

        command UpdateTodo {
            meta {
                key: "update-todo",
                title: "Update Todo Item",
                kind: object,
                resources: [TODO_RESOURCE],
            }
            payload {
                #[serde(default)]
                done: Option<bool>,
                #[serde(default)]
                notes: Option<String>,
            }
            handle (payload, aggregate) {
                if payload.done.is_none() && payload.notes.is_none() {
                    return Err(riverbase_core::base::InvalidArgumentError::with(
                        "TDO-001",
                        "Todo update requires at least one of 'done' or 'notes'.",
                        "update requires at least one of done or notes",
                    ));
                }
                aggregate.update(payload.done, payload.notes).await
            }
        }

        command ClearDone {
            meta {
                key: "clear-done",
                title: "Clear Completed Todos",
                kind: collection,
                resources: [TODO_RESOURCE],
            }
            payload {}
            target collection(TODO_RESOURCE);
            handle (payload, aggregate) {
                let _ = payload;
                aggregate.clear_done().await
            }
        }
    }
    methods (
        create(title: [impl Into<String>]) for CreateTodo {
            payload CreateTodoPayload { title: title.into() },
            target none
        }
        set_done(id: [impl Into<String>]; done: [bool]) for UpdateTodo {
            payload UpdateTodoPayload { done: Some(done), notes: None },
            target some(CommandTarget::Object(AggregateRoot::new(TODO_RESOURCE, id.into())))
        }
        clear_done() for ClearDone {
            payload ClearDonePayload {},
            target none
        }
    )
}

From one block the macro generates:

a payload struct per command (CreateTodoPayload, …) deriving serde and JSON Schema;
handler registration with the metadata (key, title, kind, resources);
the HTTP routes and OpenAPI operations;
typed helper methods from methods (…), such as engine.create("Buy milk") and engine.set_done(id, true).

meta { actor_pool_size: 4 } sizes the ractor actor pool that schedules in-flight commands. command_pool_size is the preferred spelling. The pool controls scheduling only; correctness comes from database transactions.

Labels in the payload schema

GET …/{cmdkey}.meta returns the payload's JSON Schema. Labels use standard schemars attributes, which command_engine! forwards onto the generated fields:

Intent	Annotation	Schema key
Field label	`#[schemars(title = "Code")]`	`properties.code.title`
Help text	`#[schemars(description = "…")]` or a `///` doc comment	`properties.code.description`
Widget hints	`#[schemars(extend("x-widget" = "select"))]`	vendor keys on the property

Implement the aggregate

An aggregate wraps AggregateCore and exposes domain methods:

domain/src/aggregate.rs
use std::sync::Arc;

use async_trait::async_trait;
use riverbase_core::base::{AggregateContext, RiverbaseResult};
use riverbase_core::datastore::DataStore;
use riverbase_core::{aggregate_core_deref, Aggregate as RiverbaseAggregate, AggregateCore};
use riverbase_proc::domain_action;
use serde_json::{json, Value};
use uuid::Uuid;

pub struct TodoAggregate<S: DataStore + 'static> {
    pub(crate) core: AggregateCore<S>,
}

// Forward AggregateCore methods (upsert, emit_activity, …) onto TodoAggregate.
aggregate_core_deref!(TodoAggregate);

#[async_trait]
impl<S: DataStore + 'static> RiverbaseAggregate<S> for TodoAggregate<S> {
    async fn open(
        statemgr: Arc<S>,
        logstore: riverbase_core::logstore::DomainLogStore,
        message: &riverbase_core::command::CommandMessage,
        cmdmeta: &riverbase_core::command::CommandMeta,
        context: &AggregateContext,
    ) -> RiverbaseResult<Self> {
        Ok(Self {
            core: AggregateCore::open(statemgr, logstore, message, cmdmeta, context.clone()).await?,
        })
    }

    fn core(&self) -> &AggregateCore<S> { &self.core }
    fn core_mut(&mut self) -> &mut AggregateCore<S> { &mut self.core }

    // Every event key a #[domain_action] in this aggregate may record.
    fn registered_event_keys() -> &'static [&'static str] {
        &["todo.created", "todo.updated", "todo.cleared"]
    }
}

impl<S: DataStore + 'static> TodoAggregate<S> {
    #[domain_action(event = "todo.created")]
    pub async fn create(&mut self, title: &str) -> RiverbaseResult<Value> {
        let id = Uuid::new_v4();
        let id_str = id.to_string();

        self.upsert("todo", id, json!({ "title": title, "done": false, "notes": "" }))
            .await?;
        self.bind_aggroot("todo", id);
        self.emit_message("riverbase.todo.created", json!({ "id": id_str })).await?;
        self.emit_activity("todo.create", json!({ "title": title })).await?;

        Ok(json!({ "id": id_str, "title": title, "notes": "" }))
    }
}

A domain method should:

Be a #[domain_action(event = "…")]. On Ok, the wrapper queues one event_log row keyed by the action, not the command. When the key is only known at runtime, call begin_domain_action, enqueue_action_event, and end_domain_action yourself.
Write inside the action. Call upsert, update, or remove there. Writes outside an action fail with DOM-016.
Bind the root on create. Call bind_aggroot after a collection create, so the event names the new row.
Emit explicitly. Integration messages (emit_message) and audit activities (emit_activity) are separate from the action event; emit them yourself where they apply.
Return the HTTP response. Return JSON for the caller.

Object commands such as update-todo arrive with the row already loaded. Read it with self.rootobj() and the locator with self.aggroot().

Keep registered_event_keys() complete. The framework ships missing_action_event_keys so a unit test can check it:

#[test]
fn domain_action_keys_are_registered() {
    let missing = riverbase_core::missing_action_event_keys(
        include_str!("aggregate.rs"),
        TodoAggregate::<riverbase_core::datastore::PgDataStore>::registered_event_keys(),
    );
    assert!(missing.is_empty(), "unregistered #[domain_action] event keys: {missing:?}");
}

Call it over HTTP

Collection command, resource in the path:

curl -s -X POST "http://127.0.0.1:8080/api/riverbase.todo/create-todo:post/todo" \
  -H "content-type: application/json" \
  -d '{"payload":{"title":"Buy milk"}}'

Object command, identifier in the path:

curl -s -X POST "http://127.0.0.1:8080/api/riverbase.todo/update-todo:exec/todo/{uuid}" \
  -H "content-type: application/json" \
  -d '{"payload":{"done":true}}'

The body always wraps the fields under "payload". Set deny_unknown_fields = true in config to reject properties the payload type does not declare.

Call it from code

Every invocation carries an EngineContext: actor, roles, claims, correlation id, idempotency key. HTTP middleware fills it in. In tests and workers, use the typed helpers:

let created = engine.create("Buy milk").await?;
engine.set_done(created["id"].as_str().unwrap(), true).await?;

Errors

Handlers return RiverbaseResult<T>. Raise structured errors with a stable XXX-NNN code at the point of failure:

return Err(riverbase_core::base::InvalidArgumentError::with(
    "TDO-001",
    "Todo update requires at least one of 'done' or 'notes'.",
    "update requires at least one of done or notes",
));

For codes reused across a crate, declare them once with declare_errors!. See Errors.

Debugging the macro

When command_engine! produces an opaque type error, expand it:

cargo install cargo-expand
cargo expand -p todo-domain command

Common mistakes are reported as compile_error!:

an unknown meta option in a command block;
an unknown domain_trait value (use generate or manual);
engine Foo { … } without a following commands { … } or methods ( … ) block;
malformed methods entries (missing for CommandName, payload, or target).

Use domain_trait: manual when you need a custom DomainCommandEngine::execute, for example to publish a notification after a command, such as telling RxDB clients to re-pull.

Checklist

command_engine! with kind, resources, a payload, and handle for each command
Aggregate implements Aggregate<S> and lists every event in registered_event_keys()
Every state change happens inside a #[domain_action]
The store spec registers the todo resource
:post and :exec routes return the expected JSON

Next: 4. Queries.

Command kinds​

Declare the command engine​

Labels in the payload schema​

Implement the aggregate​

Call it over HTTP​

Call it from code​

Errors​

Debugging the macro​

Checklist​