Realtime

Three optional features push changes to clients. All of them ride the runtime's StreamBus (NATS, Redis, or Postgres LISTEN/NOTIFY), so an update made on one replica reaches clients connected to any other.

cargo add riverbase_http --git https://github.com/riverbase-io/riverbase --features websocket,sse,rxdb

WebSocket bridge (websocket)

RtcBridge connects browser clients to bus channels for bidirectional messages:

use std::sync::Arc;
use riverbase_http::websocket::{RtcBridge, StreamRtcTransport};

let bus = runtime.stream_bus.clone();
let rtc = RtcBridge::new("default", "/rtc", bus.clone())
    .with_transport(Arc::new(StreamRtcTransport::new(bus)))
    .register_builtins()?
    .into_router();

The bridge serves:

Path	Purpose
{prefix}/{name}~websocket/{*channel}	The WebSocket endpoint for a channel
{prefix}/{name}~ssestream/{*channel}	The same channel as server-sent events
{prefix}/{name}.meta, {prefix}/{name}.info	Bridge metadata

Channel access is authorized per principal (authorize_channel, default_channel_permissions). Messages are framed with a WireCodec: JSON by default (rtc_wire_encoding), or CBOR when the client negotiates it through Sec-WebSocket-Protocol. Custom message types register on a MessageRegistry.

Server-sent events (sse)

bus_sse_stream turns any StreamBus subscription into an SSE response. Use it in a custom route for one-way feeds such as progress or notifications.

RxDB replication (rxdb)

RxDB clients keep a local database in sync with a resource. Implement RxdbCollection for each collection:

Method	Role
collection_name()	The collection's public name
resource_name()	The resource it maps to
policy_required()	Whether scope authorization applies
pull(ctx, checkpoint, limit, scope)	Return up to limit documents changed since checkpoint, plus the next checkpoint
push(ctx, rows, scope)	Apply client writes, usually by dispatching commands

Then register and mount:

use riverbase_http::rxdb::RxdbRegistry;

let rxdb = RxdbRegistry::new()
    .with_stream_bus(runtime.stream_bus.clone())
    .register(Arc::new(TodoRxdbCollection::new(dbpool)))?
    .into_router();
app.merge(rxdb);                         // pull/push appear in OpenAPI

Routes live under /rxdb within api_base. To tell clients to re-pull after a change, publish on rxdb_notify_channel(collection), for example from a custom DomainCommandEngine::execute after each successful command (see Commands).

Choosing one

Need	Use
Offline-first UI with a local database	RxDB
One-way live feed	SSE
Bidirectional messages, presence, custom protocols	WebSocket bridge