Aggregates

An aggregate is where a domain's rules live. In Riverbase it is a session: one aggregate value exists per command dispatch, wrapping an AggregateCore that holds the context, the loaded root, the store, and buffered effects.

Implementing one

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

aggregate_core_deref!(TodoAggregate);

#[async_trait]
impl<S: DataStore + 'static> Aggregate<S> for TodoAggregate<S> {
    async fn open(
        statemgr: Arc<S>,
        logstore: DomainLogStore,
        message: &CommandMessage,
        cmdmeta: &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 }
    fn registered_event_keys() -> &'static [&'static str] {
        &["todo.created", "todo.updated", "todo.cleared"]
    }
}

The field must be named core, because #[domain_action] expands to code that uses self.core. aggregate_core_deref! lets you call self.upsert(…) instead of self.core.upsert(…).

#[domain_action]

From riverbase_proc:

#[domain_action(event = "todo.created")]
pub async fn create(&mut self, title: &str) -> RiverbaseResult<Value> { … }

Attribute	Required	Meaning
event = "…"	yes	The event key queued when the method returns Ok
resources = ["…"]	no	Allow-list of resources the action may write

The wrapper opens an action scope, runs your body, and on Ok queues one event_log row whose payload is {input, result}. On Err it queues nothing. When the key is only known at runtime, use begin_domain_action, enqueue_action_event, and end_domain_action yourself.

Writes outside an action fail with DOM-016. Every key you use must appear in registered_event_keys(); test that with missing_action_event_keys (see the tutorial).

Session helpers

Group	Methods
Context	rootobj() (loaded root JSON), aggroot() (locator), resource(), statemgr(), ctx
Read (no optimistic check)	fetch, find_one, find_all
Write (audited, optimistic via _etag)	create, update, compare_and_swap
Write (raw)	upsert, remove
Identity	bind_aggroot(resource, id), which names the root after a collection create
Effects	emit_activity(kind, data), emit_message(topic, data)

All writes take the command's unit of work, so they commit or roll back with the audit records.

Rules of thumb

• Validate input and load state first; mutate last.
• One aggregate root per command. Coordinating several roots or domains is a job for a process manager.
• Never open your own database connection inside an action; it escapes the transaction.
• Keep side effects (HTTP calls, emails) out of the aggregate. Emit a message and handle it after commit.