Chuyển tới nội dung chính

Quick start

In this page you build a small todo service from an empty directory: one domain with two commands and a query, served over HTTP and backed by Postgres. Everything fits in one crate. The Tutorial then explains each piece and grows it into a full service.

You need Rust 1.78+ and a Postgres 14+ database. See Installation.

1. Create the project​

cargo new todo-service
cd todo-service
mkdir -p migrations/2026-10-01-000000_todo_items
Cargo.toml
[package]
name = "todo-service"
version = "0.1.0"
edition = "2021"

[dependencies]
riverbase_core = { git = "https://github.com/riverbase-io/riverbase" }
riverbase_proc = { git = "https://github.com/riverbase-io/riverbase" }
riverbase_http = { git = "https://github.com/riverbase-io/riverbase" }

# The Riverbase macros expand to code that uses these crates directly.
async-trait = "0.1"
chrono = { version = "0.4", features = ["serde"] }
diesel = { version = "2.2", features = ["postgres", "chrono", "uuid", "serde_json", "r2d2"] }
diesel-async = { version = "0.5", features = ["postgres", "deadpool"] }
diesel_migrations = "2.2"
garde = { version = "0.22", default-features = false, features = ["derive", "serde"] }
schemars = { version = "0.9", features = ["preserve_order", "chrono04", "uuid1"] }
serde = { version = "1", features = ["derive"] }
serde_json = { version = "1", features = ["preserve_order"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time"] }
uuid = { version = "1", features = ["v4", "serde"] }

The finished project has six source files and one migration:

todo-service/
├── Cargo.toml
├── migrations/2026-10-01-000000_todo_items/{up.sql, down.sql}
└── src/
    ├── main.rs        # RiverbaseApp: load config, build the runtime, mount, serve
    ├── domain.rs      # domain!: namespace and engine slots
    ├── command.rs     # command_engine!: create-todo, set-notes
    ├── aggregate.rs   # TodoAggregate: the #[domain_action] methods
    ├── query.rs       # query_resource! + query_engine!: todo.list / .item / .meta
    └── store.rs       # Diesel table, entity, migrations, store spec

2. The table​

Framework rows carry shared columns (_id, _created, _tenant, _etag, …) next to your own.

migrations/2026-10-01-000000_todo_items/up.sql
CREATE SCHEMA IF NOT EXISTS todo_app;

CREATE TABLE IF NOT EXISTS todo_app.todo_items (
    _id UUID PRIMARY KEY,
    _created TIMESTAMPTZ NOT NULL DEFAULT now(),
    _updated TIMESTAMPTZ NOT NULL DEFAULT now(),
    _creator UUID,
    _updater UUID,
    _tenant UUID,
    _etag UUID NOT NULL DEFAULT gen_random_uuid(),
    _deleted TIMESTAMPTZ,
    title TEXT NOT NULL,
    notes TEXT NOT NULL DEFAULT ''
);
migrations/2026-10-01-000000_todo_items/down.sql
DROP TABLE IF EXISTS todo_app.todo_items;

3. The query​

A query resource declares the public fields and maps them onto the table. The macro also generates TodoQueryResourceKey, which the other files use to name the resource.

src/query.rs
riverbase_core::query_resource! {
    TodoQueryResource name "todo" {
        meta {
            title: "Todo",
            default_order: [created.desc, id.desc],
            scope: none,
        }
        fields {
            field id      { preset: Uuid,     label: "ID", identifier }
            field title   { preset: String,   label: "Title" }
            field notes   { preset: String,   label: "Notes" }
            field created { preset: Datetime, label: "Created" }
        }
        binding source "todo_items" {
            id      => "_id",
            created => "_created",
        }
        policy: public("quick start: no tenants yet", "QUICKSTART"),
    }
}

riverbase_core::query_engine! {
    engine TodoQueryEngine {
        meta { query_pool_size: 4 }
        component: {},
    }
    resources { TodoQueryResource, }
}

4. The persistence glue​

The Diesel table, a row type, JSON converters, the entity that PgDataStore dispatches to, and the domain's migrations, bundled into a store spec:

src/store.rs
use std::sync::Arc;

use diesel::prelude::*;
use diesel_migrations::{embed_migrations, EmbeddedMigrations};
use riverbase_core::base::{RiverbaseResult, StorageError};
use riverbase_core::datastore::postgres::run_embedded_migrations;
use riverbase_core::datastore::{ErasedEntity, ResourceRegistry};
use riverbase_core::domain::DomainStoreSpec;

use crate::query::TodoQueryResourceKey;

riverbase_core::diesel_table_with_domain_fields! {
    todo_app.todo_items (_id) {
        title -> Text,
        notes -> Text,
    }
}

riverbase_core::domain_row! {
    #[derive(Debug, Clone, Queryable, Selectable)]
    #[diesel(table_name = todo_items)]
    pub struct TodoRow {
        pub title: String,
        pub notes: String,
    }
}

riverbase_core::domain_json_converters! {
    TodoRow, todo_items {
        to_json: todo_row_to_json,
        upsert: todo_upsert_from_json,
        title: text,
        notes: text_or(""),
    }
}

riverbase_core::pg_domain_entity! {
    TodoEntity {
        schema: todo_items,
        row: TodoRow,
        resources: [TodoQueryResourceKey],
        source: "todo_items",
        row_to_json: todo_row_to_json,
        upsert_from_json: todo_upsert_from_json,
        order: [
            "_id" => _id,
            "_created" => _created,
            "title" => title,
            "notes" => notes,
        ],
    }
}

pub const TODO_MIGRATIONS: EmbeddedMigrations = embed_migrations!("migrations");

pub fn run_todo_migrations(db_url: &str) -> RiverbaseResult<()> {
    run_embedded_migrations(db_url, TODO_MIGRATIONS, "todo_app").map_err(|e| {
        StorageError::from_driver("TDO-002", "Failed to apply todo migrations.", e)
    })
}

/// What this domain owns in the shared database: resources, entities, migrations.
pub fn todo_store_spec() -> DomainStoreSpec {
    DomainStoreSpec::new(ResourceRegistry::of::<TodoQueryResourceKey>())
        .with_entities(vec![Arc::new(TodoEntity) as Arc<dyn ErasedEntity>])
        .with_migrate(run_todo_migrations)
        .with_migration_name("todo_app")
}

5. The aggregate​

State changes only inside #[domain_action] methods. Each successful action records one event.

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

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

use crate::query::TodoQueryResourceKey;

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: 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
    }

    fn registered_event_keys() -> &'static [&'static str] {
        &["todo.created", "todo.noted"]
    }
}

impl<S: DataStore + 'static> TodoAggregate<S> {
    /// Collection command: create a row and name it as this command's aggregate root.
    #[domain_action(event = "todo.created")]
    pub async fn create(&mut self, title: &str) -> RiverbaseResult<Value> {
        let id = Uuid::new_v4();
        self.upsert(TodoQueryResourceKey::NAME, id, json!({ "title": title, "notes": "" }))
            .await?;
        self.bind_aggroot(TodoQueryResourceKey::NAME, id);
        Ok(json!({ "id": id.to_string(), "title": title }))
    }

    /// Object command: the engine has already loaded the row named in the URL.
    #[domain_action(event = "todo.noted")]
    pub async fn set_notes(&mut self, notes: &str) -> RiverbaseResult<Value> {
        let root = self.rootobj().cloned().unwrap_or_default();
        let id = self
            .aggroot()
            .and_then(|r| Uuid::parse_str(&r.identifier).ok())
            .ok_or_else(|| {
                InvalidArgumentError::with("TDO-001", "A todo id is required.", "missing aggroot")
            })?;
        let title = root.get("title").cloned().unwrap_or(Value::Null);
        self.upsert(TodoQueryResourceKey::NAME, id, json!({ "title": title, "notes": notes }))
            .await?;
        Ok(json!({ "id": id.to_string(), "title": title, "notes": notes }))
    }
}

6. The commands​

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

use riverbase_core::command::MessageBus;
use riverbase_core::ResourceKey;

use crate::aggregate::TodoAggregate;
use crate::query::TodoQueryResourceKey;

riverbase_core::command_engine! {
    engine TodoCommandEngine {
        meta { command_pool_size: 4 }
        aggregate: TodoAggregate,
        component: { msgbus: Arc<dyn MessageBus> },
    }
    commands {
        command CreateTodo {
            meta {
                key: "create-todo",
                title: "Create a todo",
                kind: collection,
                resources: [TodoQueryResourceKey::NAME],
            }
            payload { title: String, }
            target collection(TodoQueryResourceKey::NAME);
            handle (payload, aggregate) {
                aggregate.create(&payload.title).await
            }
        }

        command SetNotes {
            meta {
                key: "set-notes",
                title: "Set a todo's notes",
                kind: object,
                resources: [TodoQueryResourceKey::NAME],
            }
            payload { notes: String, }
            handle (payload, aggregate) {
                aggregate.set_notes(&payload.notes).await
            }
        }
    }
    methods ()
}

7. The domain and the service​

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

use riverbase_core::domain::DomainRuntime;

use crate::command::TodoCommandEngine;
use crate::query::TodoQueryEngine;

riverbase_core::domain! {
    domain TodoDomain {
        meta { namespace: "riverbase.todo", title: "Todo" }
        command: TodoCommandEngine,
        query: TodoQueryEngine,
        spawn(runtime: &DomainRuntime) {
            let ctx = Self::engine_context(runtime);
            let store = S::init_migrated(&runtime.dbconn, crate::store::todo_store_spec())?;
            let command = Arc::new(
                TodoCommandEngine::spawn(
                    ctx.clone(),
                    runtime.logstore.clone(),
                    runtime.msgbus.clone(),
                    store.clone(),
                )
                .await?,
            );
            let query = Arc::new(TodoQueryEngine::spawn(ctx, store, runtime.logstore.clone()).await?);
            Ok(Self::from_engines(command, query))
        }
    }
}
src/main.rs
mod aggregate;
mod command;
mod domain;
mod query;
mod store;

use riverbase_core::datastore::PgDataStore;
use riverbase_core::domain::DomainRuntime;
use riverbase_http::RiverbaseApp;

use crate::domain::TodoDomain;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Loads riverbase.toml and RIVERBASE_* variables, and sets up logging.
    let mut app = RiverbaseApp::init("Todo", "Riverbase quick start")?;

    // Connects Postgres (framework migrations run here), the audit log, and the message bus.
    let runtime = DomainRuntime::from_config(app.config()).await?;

    // Runs the todo migrations, starts the engines, and mounts their routes.
    let todo = TodoDomain::<PgDataStore>::spawn(&runtime).await?.into_arc();
    app.mount_domain(todo).await?;

    app.serve().await?;
    Ok(())
}

8. Run it​

Point the service at your database. For a single local process, the same database can carry the message bus through LISTEN/NOTIFY, so you don't need NATS yet:

export RIVERBASE_DB_URL=postgres://riverbase:change-me@127.0.0.1:5432/riverbase
export RIVERBASE_BUS_URL=$RIVERBASE_DB_URL
cargo run

The first start applies the framework migrations (the riverbase_audit tables and friends), then todo_app.todo_items. The service listens on 0.0.0.0:8080.

9. Call it​

A collection command uses :post and targets a resource:

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"}}'

The response contains the new id. An object command uses :exec and names the row:

curl -s -X POST "http://127.0.0.1:8080/api/riverbase.todo/set-notes:exec/todo/<id>" \
  -H 'content-type: application/json' \
  -d '{"payload":{"notes":"2%, organic"}}'

The body always wraps fields in "payload". Queries read without changing state:

curl -s http://127.0.0.1:8080/api/riverbase.todo/todo.list
curl -s http://127.0.0.1:8080/api/riverbase.todo/todo.item/<id>
curl -s http://127.0.0.1:8080/api/riverbase.todo/todo.meta     # fields, labels, sortable columns

Filter with a URL-encoded query object, and sort and page with plain parameters:

# titles containing "milk", oldest first, 10 per page: query={"title.has":"milk"}
curl -s "http://127.0.0.1:8080/api/riverbase.todo/todo.list?query=%7B%22title.has%22%3A%22milk%22%7D&sort=created.asc&limit=10"

Every command and query is also described in the generated OpenAPI document:

curl -s http://127.0.0.1:8080/api/openapi.json | head -c 600

What just happened​

  • The command engine opened one transaction, ran TodoAggregate::create, and committed the row together with the command, event and response records in the audit schema.
  • The query engine validated your query against the declared fields, lowered it to SQL, and returned the rows.

Next: the Tutorial adds updates, reports, custom routes, authentication and permissions. Project layout shows how to split the service into a domain crate and a service crate as it grows.