Skip to main content
eventdbx-client is an async crate that speaks the EventDBX control socket (Cap’n Proto over TCP with a Noise XX handshake derived from your token). It covers list/get/select/events reads plus create/apply/patch/archive mutations.

​
Feature highlights

  • Tokio-native async client; connect once with EventDbxClient::connect and reuse the socket.
  • Noise XX (+ PSK from your token) on by default; opt out with .with_noise(false) when the server permits plaintext.
  • Serde-friendly JSON payloads/metadata for create, append, patch, and archive/restore helpers.
  • Pagination, filters, and sort directives (created_at, aggregate_id, etc.) on list_aggregates and list_events.
  • Per-call overrides for publish targets, notes, metadata, and tokens to scope mutations.
  • Configurable connect/request timeouts (defaults: 5s connect, 10s request).

​
Install

cargo add eventdbx-client

​
Connect and configure

use eventdbx_client::{
    AppendEventRequest, ClientConfig, EventDbxClient, ListAggregatesOptions, PublishTarget,
    SetAggregateArchiveRequest,
};
use serde_json::json;

#[tokio::main]
async fn main() -> eventdbx_client::Result<()> {
    let config = ClientConfig::new(
        std::env::var("EVENTDBX_HOST").unwrap_or_else(|_| "127.0.0.1".into()),
        std::env::var("EVENTDBX_TOKEN")?,
    )
    .with_tenant(std::env::var("EVENTDBX_TENANT").unwrap_or_else(|_| "default".into()))
    .with_port(
        std::env::var("EVENTDBX_PORT")
            .ok()
            .and_then(|p| p.parse().ok())
            .unwrap_or(6363),
    )
    .with_noise(true); // set false to request plaintext when the server allows it

    let client = EventDbxClient::connect(config).await?;

    // append an event with metadata and publish targets
    let mut append = AppendEventRequest::new(
        "person",
        "p-110",
        "person_registered",
        json!({ "first_name": "Nia", "last_name": "Patel", "email": "[email protected]" }),
    );
    append.metadata = Some(json!({ "@actor": "svc-directory" }));
    append.note = Some("seed data".into());
    append
        .publish_targets
        .push(PublishTarget::new("search-indexer").with_mode("event-only"));
    client.append_event(append).await?;

    // list aggregates with sort + pagination
    let mut list_opts = ListAggregatesOptions::default();
    list_opts.take = Some(10);
    list_opts.sort_text = Some("created_at:desc, aggregate_id:asc".into());
    let aggregates = client.list_aggregates(list_opts).await?;
    println!("aggregates: {}", aggregates.aggregates);

    // archive an aggregate with a note
    let mut archive = SetAggregateArchiveRequest::new("person", "p-110", true);
    archive.note = Some("cleanup".into());
    client.set_aggregate_archive(archive).await?;

    Ok(())
}
ClientConfig::new(host, token) defaults to port = 6363, tenant = "default", Noise enabled, connect_timeout = 5s, and request_timeout = 10s per RPC. Build one config per tenant/token and reuse the client for the process lifetime. Call .with_request_timeout(None) to disable per-call timeouts.

​
Runtime configuration

VariableDefaultDescription
EVENTDBX_HOST127.0.0.1Hostname or IP address of the control socket.
EVENTDBX_PORT6363TCP port for the control plane.
EVENTDBX_TOKENemptyControl token forwarded during the handshake.
EVENTDBX_TENANTdefaultTenant identifier included in the initial hello.
EVENTDBX_NO_NOISEfalseSet 1/true on the server side to permit plaintext; pair with .with_noise(false).
The crate does not read these automatically; pass them into ClientConfig as shown above.

​
Manage authentication and tenancy

Pass the control token and tenant into ClientConfig up front. Per-call token overrides are available on request structs/options (ListAggregatesOptions.token, AppendEventRequest.token, etc.) so you can scope permissions to a user session without rebuilding the client.

​
Publish targets

Direct mutations to specific plugins and override payload mode/priority per call: 
use eventdbx_client::{AppendEventRequest, PublishTarget};
use serde_json::json;

let mut request =
    AppendEventRequest::new("invoice", "inv-1", "invoice_created", json!({}));
request.publish_targets = vec![
    PublishTarget::new("analytics-engine").with_mode("event-only"),
    PublishTarget::new("fraud-worker").with_mode("all").with_priority("high"),
];
client.append_event(request).await?;
Skip publish_targets to fan out to all enabled plugins with their default payload mode. Publish targets are only effective when the named plugins are installed, configured, and running.

​
Write aggregates and events

use eventdbx_client::{
    AppendEventRequest, CreateAggregateRequest, EventDbxClient, PatchEventRequest, PublishTarget,
    SetAggregateArchiveRequest,
};
use serde_json::json;

let mut create = CreateAggregateRequest::new(
    "person",
    "p-120",
    "person_registered",
    json!({ "first_name": "Ari", "email": "[email protected]" }),
);
create.publish_targets.push(PublishTarget::new("search"));
let created = client.create_aggregate(create).await?;

let mut append = AppendEventRequest::new(
    "person",
    "p-120",
    "person_email_updated",
    json!({ "email": "[email protected]" }),
);
append.note = Some("user changed email".into());
client.append_event(append).await?;

let mut patch = PatchEventRequest::new(
    "person",
    "p-120",
    "person_registered",
    json!([{ "op": "replace", "path": "/email", "value": "[email protected]" }]),
);
client.patch_event(patch).await?;

let mut archive = SetAggregateArchiveRequest::new("person", "p-120", true);
archive.note = Some("cleanup".into());
client.set_aggregate_archive(archive).await?;

let restore = SetAggregateArchiveRequest::new("person", "p-120", false);
client.set_aggregate_archive(restore).await?;
create_aggregate seeds a snapshot and first event atomically, append_event appends events, patch_event issues RFC 6902 operations against historical payloads, and set_aggregate_archive toggles write access while preserving history.

​
Read aggregates and events

use eventdbx_client::{
    EventDbxClient, ListAggregatesOptions, ListEventsOptions, SelectAggregateRequest,
};

let got = client.get_aggregate("person", "p-110").await?;

let select = client
    .select_aggregate(SelectAggregateRequest::new(
        "person",
        "p-110",
        vec!["payload.email".into(), "metadata.@actor".into()],
    ))
    .await?;

let mut list = ListAggregatesOptions::default();
list.take = Some(20);
list.sort_text = Some("created_at:desc".into());
let aggregates = client.list_aggregates(list).await?;
if let Some(cursor) = aggregates.next_cursor.as_deref() {
    let mut next = ListAggregatesOptions::default();
    next.cursor = Some(cursor.to_string());
    let more = client.list_aggregates(next).await?;
}

let events = client
    .list_events("person", "p-120", ListEventsOptions::default())
    .await?;

let merkle = client.verify_aggregate("person", "p-120").await?;
Use next_cursor to resume pagination. verify_aggregate returns the Merkle root for integrity checks.

​
Filters, sorting, and pagination

Filters use the same SQL-like grammar as the server (field = value AND other_field > 10). Sort directives accept fields like aggregate_id, aggregate_type, created_at, updated_at, and archived. Pass a string ("created_at:desc, aggregate_id:asc") via ListAggregatesOptions.sort_text or build AggregateSort entries. Use include_archived / archived_only when traversing soft-deleted aggregates. Feed next_cursor back into subsequent list_aggregates calls to continue paging.

​
Noise transport

Noise XX is enabled by default using a PSK derived from your control token. Disable it only for controlled testing with .with_noise(false) and a server configured to honor EVENTDBX_NO_NOISE; production deployments should keep Noise on.

​
Development & testing

cargo test
Use cargo check for quick compilation cycles and cargo fmt --check in CI to keep formatting consistent.