Datastore

riverbase_core::datastore is the persistence layer that command and query engines share. It is built on Diesel with diesel-async, and Postgres is the only backend.

Defining a table

Every framework-backed table carries the shared domain columns (_id, _created, _updated, _creator, _updater, _tenant, _etag, _deleted). Declare the Diesel schema once with them:

domain/src/postgres/schema.rs

use riverbase_core::diesel_table_with_domain_fields;

diesel_table_with_domain_fields! {
    todo_examples.todo_items (_id) {
        title -> Text,
        done -> Bool,
        notes -> Text,
        version -> Int4,
    }
}

The migration creates the same columns. riverbase_core::DOMAIN_FIELDS_DDL holds the canonical DDL fragment for them.

domain/migrations/…/up.sql

CREATE SCHEMA IF NOT EXISTS todo_examples;

CREATE TABLE IF NOT EXISTS todo_examples.todo_items (
    _id UUID PRIMARY KEY,
    _created TIMESTAMPTZ NOT NULL DEFAULT now(),
    _updated TIMESTAMPTZ,
    _creator UUID,
    _updater UUID,
    _tenant UUID,
    _etag UUID,
    _deleted TIMESTAMPTZ,
    title TEXT NOT NULL,
    done BOOLEAN NOT NULL DEFAULT FALSE,
    notes TEXT NOT NULL DEFAULT '',
    version INTEGER NOT NULL DEFAULT 1
);

Rows and entities

domain_row! declares the row struct with the domain fields prepended. pg_domain_entity! tells PgDataStore how to read, write, and sort it:

domain/src/postgres/entity.rs

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

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

Key	Meaning
schema, row	The Diesel table and row struct
resources	Resource keys this entity serves. Typed …ResourceKeys come from query_resource!.
source	The physical source name used in DataQuery
row_to_json / upsert_from_json	Your conversions between rows and the JSON that aggregates and queries use
order	Sortable columns. Startup checks that every sortable query field is covered.

Inside upsert_from_json, domain_fields_from_payload and domain_insert_values! fill the domain columns from the aggregate's audit patch. For read-only views, use pg_readonly_entity!.

Migrations

domain/src/postgres/migrate.rs

use diesel_migrations::{embed_migrations, EmbeddedMigrations};
use riverbase_core::datastore::postgres::{establish_dbpool, run_embedded_migrations, PgPool};

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_examples").map_err(|e| {
        riverbase_core::base::StorageError::with("TDO-024", "Failed to apply todo migrations.", e.to_string())
    })
}

establish_dbpool creates the pool and runs the framework migrations (audit, outbox, process manager). Each crate's migration bookkeeping lives in its own schema (todo_examples.__diesel_schema_migrations), so framework, extension, and domain migrations never collide.

The store

let store = PgDataStore::with_registry(pool, ResourceRegistry::new(["todo", "todo_item"]))
    .register_entity(Arc::new(TodoEntity));

Most code never builds a store by hand: DomainStoreSpec plus S::init_migrated does it in domain! spawn.

DataStore method	Use
registry().parse(resource)	Resolve a public resource name
find_all(resource, filter, order, projection)	Read many rows
fetch, find_one	Read one row
remove(uow, resource, id)	Delete inside a command unit of work
begin_command_transaction	Used by the command engine
state_fetch, state_upsert, state_create	Low-level JSON state helpers

On the command path, pass the command's CommandUnitOfWork (self.core.ctx.unit_of_work) to every write.

Query DSL

DataQuery is the lowered form of every query:

use riverbase_core::datastore::dsl::{DataQuery, Expr};

let q = DataQuery::list("todo_items")
    .with_page(0, 50);
let only_open = Expr::eq("done", false);
let mine = Expr::in_list("_tenant", vec![tenant.to_string()]);
let both = Expr::and_opt(Some(only_open), Some(mine));

Expr is a tree of Field { path, op, value }, And, Or, and Not. Server policy goes in policy_filter and the client's statements in the filter; they are ANDed at execution.

Errors

Diesel errors become RiverbaseErrors with DAT-* codes at the detection site: not-found is DAT-001, and database error kinds are DAT-090 to DAT-098. Propagate them with ?. See Errors.