leptos-wasi-cqrs

# Leptos WASI + CQRS/ES Integration Skill This skill provides step-by-step instructions and reference patterns for an AI agent to build, modify, debug, and expand full-stack Event Sourced (CQRS) applications integrating the `ddd_cqrs_es` framework with the `leptos_wasi` component model server on both **Fermyon Spin** and **generic Wasmtime** runtimes. The current **counter app** (`examples/counter-app`) supports `sqlite`, `postgres`, `neon`, `supabase`, `turso`, `mysql`, and `redis` through the Makefile. **Scope note:** This skill is oriented at the **counter** multi-backend example and pure CQRS wiring. For the production **SaaS product shell** (`examples/fullstack-app` / `ddd init --preset fullstack`), use: - `.agents/skills/ddd-cli/SKILL.md` — scaffold, serve (`make dev`), dual-sync, what CLI ops are allowed - `docs/tutorial/leptos-islands-persistent-chrome.md` — soft-nav, persistent sidebar chrome, skeleton rules - `docs/production/wasi-auth-fullstack.md` — auth/outbox/operator runbook - `examples/fullstack-app/README.md` — day-to-day product commands On fullstack, `ddd add aggregate|event|command` bootstraps optional `src/domain` beside wasi-auth; route/UI wiring stays manual. Unwired stubs (`add projection`, etc.) are refused. --- ## 🗺️ High-Level Architectural Flow In this architecture, incoming write operations (Commands) can enter through **Leptos Server Functions**, explicit JSON REST endpoints, or Spin gRPC. All command surfaces delegate to the same application service, validate through the pure business domain (Aggregate), and persist immutable history (Events) in the Event Store. The command response returns a unified read view immediately, then projections and realtime notifications catch up from durable event history: ```mermaid sequenceDiagram autonumber actor User as 🌐 Browser Client participant Client as 🖥️ Leptos Client (WASM) participant Server as ⚙️ Leptos Server (WASI) participant Domain as 🧠 Aggregate (Domain) participant EventDB as 🗄️ Event Store participant ReadDB as 📊 Read Model User->>Client: Clicks command button or calls REST/gRPC Client->>Server: Server Function, REST, or gRPC command Server->>EventDB: Fetch historical Event Stream for Aggregate ID EventDB-->>Server: [Historical Events...] Server->>Domain: Replay events to reconstruct current aggregate state Server->>Domain: Validate Command against current state invariants Domain-->>Server: Ok([New Committed Events]) Server->>EventDB: Append new events (Optimistic Concurrency Control) Server->>Server: Build CounterViewDto from authoritative state Server-->>Client: HTTP 200 OK with updated view Server->>ReadDB: Catch up read model/checkpoint from durable events Server-->>Client: SSE wake/update for other sessions when realtime is enabled ``` --- ## Concept Map for Agents When working in this repo, keep these boundaries separate: - `domain.rs`: pure commands, events, IDs, errors, and `Aggregate` implementations. No database, HTTP, Redis, Leptos, logging side effects, or runtime-specific APIs belong here. - `error.rs`: typed counter-app application errors plus REST, server-function, and gRPC mapping helpers. Do not erase shared-service failures with `to_string()` before this boundary. - `app.rs`: Leptos UI, server functions, DTOs, and optimistic UI reconciliation. Server functions delegate command execution to `application.rs`. - `application.rs`: shared command/read application service used by Leptos server functions, REST, and gRPC. Keep repository execution, realtime publish, and projection catch-up here. - `rest.rs`: explicit JSON REST endpoints for curlable integration checks: `/api/counter/view`, `/api/counter/increment`, `/api/counter/decrement`, and `/api/counter/reset`. - `grpc.rs`: Spin-only gRPC service generated from `proto/counter.proto` and served through the Spin HTTP trigger with `spin-sdk/grpc`. - `store.rs`: backend selection, schema initialization, event-store/checkpoint/read-model adapters, projection catch-up, SSE stream response, and Redis wake publishing. - `server.rs`: WASI HTTP routing, static file serving, one-time schema initialization before dynamic requests, gRPC detection, transport guards, REST routing, manual `/api/counter/stream` routing, and Leptos server-function registration. - `Makefile`: canonical build/run/reset entrypoint. Do not invent alternate public commands; validate with `make help`, `make help-db`, `make help-realtime`, and `make help-matrix`. Current API names to prefer: - Aggregate replay from envelopes: `Aggregate::replay`. - Raw aggregate unit-test replay: `Aggregate::replay_raw_events_from_zero`. - Command execution that returns updated aggregate state: `execute_returning_state`. - Production SQL request idempotency: `execute_idempotent_atomic` / async `execute_idempotent_atomic` with SQLite, PostgreSQL, or MySQL event stores. - Unified counter read call: `get_counter_view` / `CounterViewDto`. - Shared command application service: `execute_counter_command`. - Realtime transport: SSE/EventSource in the browser; Redis is a wake/notification transport unless `db=redis` is selected. Avoid stale APIs and command names: - Do not implement `Aggregate::id()`; it was removed. - Do not call `Aggregate::replay_events`; use `replay_raw_events_from_zero` for raw event tests. - Do not use `NewEvent::from_domain_event`; use `NewEvent::new`. - Do not treat `EventType` as a `String`; use `as_str()`, `into_string()`, `Display`, or `From<&str>` / `From<String>`. - Do not document `db=libsql`, `db=psql`, `NEON_URL`, or `realime=redis`. ## 🧠 1. Pure Domain Design Pattern (`domain.rs`) The business domain must remain completely decoupled from databases, networking, or frameworks. Define pure state-machine commands, events, and validations using the framework's `Aggregate` trait: ```rust use serde::{Deserialize, Serialize}; use ddd_cqrs_es::{Aggregate, DomainEvent}; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct EntityId(pub String); #[derive(Clone, Debug, Serialize, Deserialize)] pub enum EntityCommand { DoAction { arg: i32 }, } #[derive(Clone, Debug, Serialize, Deserialize)] pub enum EntityEvent { ActionDone { arg: i32 }, } impl DomainEvent for EntityEvent { fn event_type(&self) -> &'static str { match self { EntityEvent::ActionDone { .. } => "action_done", } } } pub struct EntityAggregate { pub id: EntityId, pub state_val: i32, pub revision: u64, } impl Aggregate for EntityAggregate { type Id = EntityId; type Command = EntityCommand; type Event = EntityEvent; type Error = String; fn aggregate_type() -> &'static str { "entity" } fn revision(&self) -> u64 { self.revision } fn new() -> Self { Self { id: EntityId(String::new()), state_val: 0, revision: 0, } } fn apply(&mut self, event: &Self::Event) { match event { EntityEvent::ActionDone { arg } => { self.state_val += arg; } } self.revision += 1; } fn handle(&self, command: Self::Command) -> Result<Vec<Self::Event>, Self::Error> { match command { EntityCommand::DoAction { arg } => { if arg <= 0 { return Err("Argument must be positive".to_string()); } Ok(vec![EntityEvent::ActionDone { arg }]) } } } } ``` Aggregate rules: - `handle` validates current state and returns events; it must not mutate state or call infrastructure. - `apply` mutates state from already-decided facts; it must be deterministic and side-effect free. - `revision()` must reflect replayed stream revision. Increment it in `apply` when the aggregate stores a revision field. - Aggregate identity is passed to the repository. The trait no longer asks the aggregate for an ID. - `DomainEvent::event_type()` should return a stable wire/storage name. Persisted envelopes wrap it as the `EventType` newtype. --- ## 💾 2. Multi-Runtime WASI Persistence (`store.rs`) The counter app uses `MultiBackendEventStore`, checkpoint helpers, and read-model helpers to select storage from runtime environment variables set by the Makefile. Backend-specific code must stay feature/cfg gated so both Spin and Wasmtime compile: - `db=sqlite`: Spin SQLite host store on Spin; JSON files mounted at `/data` on Wasmtime. - `db=postgres`: PostgreSQL URL from `POSTGRES_URL`, passed internally as `DATABASE_URL`. - `db=neon`: Neon HTTP SQL URL from `NEON_DB_URL`, passed internally as `DATABASE_URL`. - `db=supabase`: Supabase REST URL/key from `SUPABASE_URL` and `SUPABASE_SECRET_KEY`, passed internally as `DATABASE_URL` and `DATABASE_AUTH_TOKEN`. - `db=turso`: Turso/LibSQL Hrana URL/token from `TURSO_URL` and `TURSO_AUTH_TOKEN`, passed internally as `DATABASE_URL` and `DATABASE_AUTH_TOKEN`; public command name remains `db=turso`. - `db=mysql`: MySQL URL from `MYSQL_URL`, passed internally as `DATABASE_URL`; raw TCP MySQL on Wasmtime with `wasi-mysql`; Spin host MySQL with `spin-mysql`. - `db=redis`: Redis is the durable event/checkpoint/read-model store and uses `REDIS_URL` directly. Do not document `DATABASE_URL` or `DATABASE_AUTH_TOKEN` as public `.env` configuration for the counter app. They are internal runtime env values derived by the Makefile. MySQL intentionally does not fall back to `DATABASE_URL`; use `MYSQL_URL` for `db=mysql`. `realtime=redis` is independent of the durable backend. It uses Redis as notification/wake transport, then the SSE route replays durable events after the client's `last_sequence`. For production examples that use the native SQL adapters directly, prefer `Repository::execute_idempotent_atomic` or `AsyncRepository::execute_idempotent_atomic` over the portable `execute_idempotent` path. The portable API is still valid for demos and non-strict workloads, but it coordinates separate stores and should not be described as crash-atomic. ### 2.1 Database Query and Consistency Rules Before adding or changing SQL, match each query to the access pattern documented in `docs/production/db-query-patterns.md`: - Aggregate stream loads use `WHERE aggregate_type = ? AND aggregate_id = ? ORDER BY revision ASC`; the `UNIQUE (aggregate_type, aggregate_id, revision)` constraint is the required access path. - Aggregate-scoped global replay uses `WHERE aggregate_type = ? AND sequence > ? ORDER BY sequence ASC`; schemas need an `(aggregate_type, sequence)` index. - Point stores use primary keys: checkpoints by `projection_name`, idempotency by `idempotency_key`, snapshots by `(aggregate_type, aggregate_id)`, and read models by their application key. - Do not query event `payload` JSON for product screens. Project business fields into a read model and index that read model for the UI query. - Do not add duplicate indexes that mirror a unique constraint unless an actual query planner check proves a need. - Checkpoint saves must be monotonic. A stale checkpoint write must not move `sequence` backward. - Projection catch-up from durable events can be expensive. `load_global_after_limited(...)` is the required store primitive and the only global read a runner performs; `run(...)` loops `run_batch(...)` until `caught_up`. Production examples should still drive catch-up with explicit `run_batch(...)` / `ProjectionBatchConfig` so one poll's duration is bounded, and reserve `load_global_after(...)` — which returns the whole backlog in one allocation — for small tests or explicit maintenance jobs. When explaining eventual consistency, state both sides plainly: it keeps command writes small and read models scalable, but read models and realtime notifications can lag, duplicate, or arrive out of order. Redis/SSE/WebSocket are wake transports; durable replay by sequence remains the truth. For optimistic UI, never let an older server-action or SSE snapshot rewind a newer visible sequence. ### 2.2 Event Store Implementation Pattern ```rust pub struct MultiBackendEventStore<A> { _phantom: std::marker::PhantomData<fn() -> A>, } impl<A> MultiBackendEventStore<A> where A: ddd_cqrs_es::Aggregate + 'static, { pub fn new() -> Self { Self { _phantom: std::marker::PhantomData } } } // Dispatch in EventStore/AsyncEventStore impls by get_backend(): // sqlite, postgres, neon, supabase, turso, mysql, redis. // Keep runtime-specific imports behind feature/cfg gates. ``` Schema initialization belongs in `initialize_schema_async()` and must be guarded by an async lock/once guard. Do not run migrations from every command handler. `make db=<backend> fresh` is reset-only and never starts the server. --- ## ⚡ 3. Server-Side Integration & Command Runner (`application.rs`) Handle write commands through a unified application service. Use `AsyncRepository::execute_returning_state` so the command path loads the aggregate once, appends events, and returns the authoritative aggregate state for the response. The read model can catch up afterward from durable events. Leptos server functions, explicit REST routes, and Spin gRPC must all call this same service instead of duplicating command execution. In the shared server-side application layer, protect server-only code behind `#[cfg(feature = "ssr")]` at module boundaries: ```rust #[cfg(feature = "ssr")] pub async fn execute_counter_command( command: crate::domain::CounterCommand, ) -> crate::error::CounterAppResult<CounterViewDto> { use crate::domain::{Counter, CounterId}; use crate::error::CounterAppError; use crate::store::MultiBackendEventStore; use ddd_cqrs_es::AsyncRepository; let event_store = MultiBackendEventStore::<Counter>::new(); let repository = AsyncRepository::new(event_store); let aggregate_id = CounterId("global".to_string()); const COMMAND_CONCURRENCY_RETRIES: usize = 6; let mut attempts = 0; let (loaded, committed_events) = loop { match repository .execute_returning_state( &aggregate_id, command.clone(), ddd_cqrs_es::Metadata::default(), ) .await { Ok(outcome) => break outcome, Err(error) if attempts < COMMAND_CONCURRENCY_RETRIES && is_retryable_counter_write_conflict(&error) => { attempts += 1; wasip3::clocks::monotonic_clock::wait_for(attempts as u64 * 5_000_000).await; } Err(error) => return Err(CounterAppError::from_repository_error(error)), } } let mut view = get_counter_view().await?; view.count = loaded.state.value; if let Some(last_sequence) = committed_events.last().and_then(|event| event.sequence) { view.last_sequence = last_sequence; } if let Err(error) = crate::store::publish_counter_realtime(&view).await { tracing::error!(error = %error, error_code = error.public_code()); } if let Err(error) = crate::store::catch_up_counter_projection().await { tracing::error!(error = %error, error_code = error.public_code()); } Ok(view) } ``` Error handling rules: - Preserve `RepositoryError` and `EventStoreError` classifications in the shared application service. - Map typed errors to JSON REST status/body, `ServerFnError`, or `tonic::Status` only at the transport edge. - Use `tracing` for internal error details; public transport messages should be stable and safe. Leptos server functions should remain thin adapters around the shared service: ```rust #[server(prefix = "/api")] pub async fn increment_count(amount: i32) -> Result<CounterViewDto, ServerFnError> { #[cfg(feature = "ssr")] { if amount <= 0 { return Err(crate::error::CounterAppError::validation("amount must be positive") .server_fn_error()); } crate::application::execute_counter_command(