# Rust SDK

The **Rust SDK** provides programmatic access to EventSourcingDB **from Rust applications**. It offers a convenient interface for **writing**, **reading**, and **observing events** without having to interact with the HTTP API directly.

> **Official SDK**
>
> This SDK is **officially developed** and maintained by **[the native web GmbH](https://www.thenativeweb.io)**, the **creators of EventSourcingDB**. It is **aligned with the latest platform features** and **receives regular updates**.

## Installation

To **install the SDK**, run:

```shell
cargo add eventsourcingdb
```

You can also find this SDK on **[crates.io](https://crates.io/crates/eventsourcingdb)**.

## Capabilities

This SDK intentionally focuses on **application-level use cases**. It provides **core functionality** for **writing**, **reading**, and **observing events**, and deliberately **excludes operational concerns** such as backup, restore, or health monitoring.

For a detailed explanation of the specific requirements behind each capability, see the **[Compliance Criteria](/docs/eventsourcingdb/sdks/compliance-criteria)**.

> **Compatibility**
>
> Rust SDK 2.3 runs with **EventSourcingDB 1.1** and later.

> **Essentials**
>
> <span style="color:lime;">✓</span> Initialize the Client<br />
> <span style="color:lime;">✓</span> Ping the Server<br />
> <span style="color:lime;">✓</span> Verify API Token<br />
> <span style="color:lime;">✓</span> Abort a Request<br />

> **Writing Events**
>
> <span style="color:lime;">✓</span> Write Single Events<br />
> <span style="color:lime;">✓</span> Write Multiple Events<br />
> <span style="color:lime;">✓</span> Use `isSubjectPristine`<br />
> *(Requires EventSourcingDB 1.2)* <span style="color:lime;">✓</span> Use `isSubjectPopulated`<br />
> <span style="color:lime;">✓</span> Use `isSubjectOnEventId`<br />
> <span style="color:lime;">✓</span> Use `isEventQlQueryTrue`<br />
> <span style="color:lime;">✓</span> Include OpenTelemetry Trace Context<br />

> **Reading Events**
>
> <span style="color:lime;">✓</span> Read Events by Subject<br />
> <span style="color:lime;">✓</span> Read Recursively<br />
> <span style="color:lime;">✓</span> Read Chronological and Anti-Chronological<br />
> <span style="color:lime;">✓</span> Use Boundaries<br />
> <span style="color:lime;">✓</span> Use `fromLatestEvent`

> **Using EventQL**
>
> <span style="color:lime;">✓</span> Execute EventQL Queries

> **Observing Events**
>
> <span style="color:lime;">✓</span> Observe Events By Subject<br />
> <span style="color:lime;">✓</span> Observe Recursively<br />
> <span style="color:lime;">✓</span> Use Boundaries<br />
> <span style="color:lime;">✓</span> Use `fromLatestEvent`

> **Metadata and Discovery**
>
> <span style="color:lime;">✓</span> Register Event Schemas<br />
> <span style="color:lime;">✓</span> List Subjects<br />
> <span style="color:lime;">✓</span> List Event Types<br />
> <span style="color:lime;">✓</span> List a Specific Event Type<br />
> <span style="color:lime;">✓</span> Verify Event Hashes<br />
> <span style="color:lime;">✓</span> Verify Event Signatures

> **Testcontainers Support**
>
> <span style="color:lime;">✓</span> Start and Stop EventSourcingDB in Tests<br />
> <span style="color:lime;">✓</span> Use Custom Configuration (e.g., API Token, Port, …)

## Getting Started

Import the package and create an instance by providing the URL of your EventSourcingDB instance and the API token to use:

```rust
use eventsourcingdb::client::Client;
use url::Url;

// ...

let base_url: Url = "http://localhost:3000".parse().unwrap();
let api_token = "secret";
let client = Client::new(base_url, api_token);
```

Creating the client does not send any request.

Then call the `ping` function to check whether the instance is reachable. If it is not, the function will return an error:

```rust
let result = client.ping().await;
if let Err(err) = result {
  // handle error...
}
```

*Note that `ping` does not require authentication, so the call may succeed even if the API token is invalid.*

If you want to verify the API token, call `verify_api_token`. If the token is invalid, the function will return an error:

```rust
let result = client.verify_api_token().await;
if let Err(err) = result {
  // handle error...
}
```

## Writing Events

Call the `write_events` function and hand over a vector with one or more events. You do not have to provide all event fields – some are automatically added by the server.

Specify `source`, `subject`, `type` (using `ty`), and `data` according to the [CloudEvents](/docs/eventsourcingdb/cloud-events) format.

For `data` provide a JSON object using a `serde_json::Value`.

Optionally, pass the **trace context** of OpenTelemetry with `traceinfo`, either as `TraceInfo::Traceparent { traceparent }` or, together with a `tracestate`, as `TraceInfo::WithState { traceparent, tracestate }`.

The function returns the written events, including the fields added by the server:

```rust
let event = EventCandidate::builder()
  .source("https://library.eventsourcingdb.io".to_string())
  .subject("/books/42".to_string())
  .ty("io.eventsourcingdb.library.book-acquired")
  .data(json!({
    "title": "2001 - A Space Odyssey",
    "author": "Arthur C. Clarke",
    "isbn": "978-0756906788",
  }))
  .build();

let result = client.write_events(vec![event.clone()], vec![]).await;
match result {
  Ok(written_events) => // ...
  Err(err) => // ...
}
```

### Using the `IsSubjectPristine` precondition *(Changed in 4.0)*

If you only want to write events in case a subject (such as `/books/42`) does not yet have any events, use the `IsSubjectPristine` precondition to create a precondition and pass it in a vector as the second argument:

```rust
let result = client.write_events(
  vec![event.clone()],
  vec![Precondition::IsSubjectPristine {
    subject: "/books/42".to_string(),
  }],
).await;
match result {
  Ok(written_events) => // ...
  Err(err) => // ...
}
```

### Using the `IsSubjectPopulated` precondition *(Changed in 4.0; Requires EventSourcingDB 1.2)*

If you only want to write events in case a subject (such as `/books/42`) already has at least one event, use the `IsSubjectPopulated` precondition to create a precondition and pass it in a vector as the second argument:

```rust
let result = client.write_events(
  vec![event.clone()],
  vec![Precondition::IsSubjectPopulated {
    subject: "/books/42".to_string(),
  }],
).await;
match result {
  Ok(written_events) => // ...
  Err(err) => // ...
}
```

### Using the `IsSubjectOnEventId` precondition *(Changed in 4.0)*

If you only want to write events in case the last event of a subject (such as `/books/42`) has a specific ID (e.g., `0`), use the `IsSubjectOnEventId` precondition to create a precondition and pass it in a vector as the second argument:

```rust
let result = client.write_events(
  vec![event.clone()],
  vec![Precondition::IsSubjectOnEventId {
    subject: "/books/42".to_string(),
    event_id: "0".to_string(),
  }],
).await;
match result {
  Ok(written_events) => // ...
  Err(err) => // ...
}
```

*Note that according to the CloudEvents standard, event IDs must be of type string.*

### Using the `IsEventQLQueryTrue` precondition *(Changed in 4.0)*

If you want to write events depending on an EventQL query, use the `IsEventQLQueryTrue` precondition to create a precondition and pass it in a vector as the second argument:

```rust
let result = client.write_events(
  vec![event.clone()],
  vec![Precondition::IsEventQLQueryTrue {
    query: "FROM e IN events WHERE e.type == 'io.eventsourcingdb.library.book-borrowed' PROJECT INTO COUNT() < 10".to_string(),
  }],
).await;
match result {
  Ok(written_events) => // ...
  Err(err) => // ...
}
```

## Reading Events

To read all events of a subject, call the `read_events` function with the subject and an options object. Set the `recursive` option to `false`. This ensures that only events of the given subject are returned, not events of nested subjects.

The function returns a stream from which you can retrieve one event at a time:

```rust
let result = client
  .read_events("/books/42", Some(
    ReadEventsOptions {
      recursive: false,
      ..Default::default()
    }
  ))
  .await;

match result {
  Err(err) => // ...
  Ok(mut stream) => {
    while let Some(event) = stream.next().await {
      // ...
    }
  }
}
```

### Reading From Subjects Recursively

If you want to read not only all the events of a subject, but also the events of all nested subjects, set the `recursive` option to `true`:

```rust
let result = client
  .read_events("/books/42", Some(
    ReadEventsOptions {
      recursive: true,
      ..Default::default()
    }
  ))
  .await;
```

This also allows you to read *all* events ever written. To do so, provide `/` as the subject and set `recursive` to `true`, since all subjects are nested under the root subject.

### Reading in Anti-Chronological Order

By default, events are read in chronological order. To read in anti-chronological order, provide the `order` option and set it using the `Antichronological` ordering:

```rust
let result = client
  .read_events("/books/42", Some(
    ReadEventsOptions {
      recursive: false,
      order: Some(Ordering::Antichronological),
      ..Default::default()
    }
  ))
  .await;
```

*Note that you can also use the `Chronological` ordering to explicitly enforce the default order.*

### Specifying Bounds

Sometimes you do not want to read all events, but only a range of events. For that, you can specify the `lower_bound` and `upper_bound` options – either one of them or even both at the same time.

Specify the ID and whether to include or exclude it, for both the lower and upper bound:

```rust
let result = client
  .read_events("/books/42", Some(
    ReadEventsOptions {
      recursive: false,
      lower_bound: Some(Bound {
        bound_type: BoundType::Inclusive,
        id: "100",
      }),
      upper_bound: Some(Bound {
        bound_type: BoundType::Exclusive,
        id: "200",
      }),
      ..Default::default()
    }
  ))
  .await;
```

### Starting From the Latest Event of a Given Type

To read starting from the latest event of a given type, provide the `from_latest_event` option and specify the subject, the type, and how to proceed if no such event exists.

Possible options are `ReadNothing`, which skips reading entirely, or `ReadEverything`, which effectively behaves as if `from_latest_event` was not specified:

```rust
let result = client
  .read_events("/books/42", Some(
    ReadEventsOptions {
      recursive: false,
      from_latest_event: Some(
        ReadFromLatestEventOptions {
          subject: "/books/42",
          ty: "io.eventsourcingdb.library.book-borrowed",
          if_event_is_missing: ReadEventMissingStrategy::ReadEverything,
        }
      ),
      ..Default::default()
    }
  ))
  .await;
```

*Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*

## Running EventQL Queries

To run an EventQL query, call the `run_eventql_query` function and provide the query as argument. The function returns a stream.

```rust
let result = client
  .run_eventql_query("FROM e IN events PROJECT INTO e")
  .await;

match result {
  Err(err) => // ...
  Ok(mut stream) => {
    while let Some(row) = stream.next().await {
      // ...
    }
  }
}
```

*Note that each row returned by the stream is of type `serde_json::Value` and matches the projection specified in your query.*

*(Added in 2.3; Changed in 2.4)* *Note that EventSourcingDB sends a heartbeat every second while there are no rows to send. If neither a row nor a heartbeat arrives for 30 seconds, for example because the connection stalled, the stream returns a `ClientError::HeartbeatTimeout` error, closes the connection, and ends.*

## Converting Events to Polars DataFrame *(Changed in 2.0)*

For data analysis and exploration, you can convert events to Polars DataFrames. To use this feature, add the SDK with the `polars` feature:

```shell
cargo add eventsourcingdb --features polars
```

Import the `ToDataFrame` trait and collect events into a `Vec<Event>`, then call `to_dataframe` on it:

```rust
use eventsourcingdb::{Event, event::ToDataFrame};
use futures::StreamExt;

let event_stream = client
  .read_events("/books", Some(ReadEventsOptions {
    recursive: true,
    ..Default::default()
  }))
  .await?;

let events = event_stream
  .filter_map(async |e| e.ok())
  .collect::<Vec<Event>>()
  .await;

let df = events.as_slice().to_dataframe();
println!("{}", df);
```

The resulting DataFrame includes all event fields as columns: `event_id`, `time`, `source`, `subject`, `type`, `data`, `spec_version`, `data_content_type`, `predecessor_hash`, `hash`, `trace_parent`, `trace_state`, and `signature`.

The `data` field is stored as a JSON string. Use Polars' JSON functions to extract values. They need Polars itself as a dependency, in the version the SDK uses, with the features `lazy`, `strings`, and `extract_jsonpath`:

```shell
cargo add polars@0.55 --features lazy,strings,extract_jsonpath
```

```rust
use polars::prelude::*;

// Filter for a specific event type and extract data fields
let result = df.lazy()
  .filter(col("type").eq(lit("io.eventsourcingdb.library.book-acquired")))
  .with_columns([
    col("data").str().json_path_match(lit("$.title")).alias("title"),
    col("data").str().json_path_match(lit("$.author")).alias("author"),
  ])
  .collect()?;

println!("{}", result);
```

## Observing Events

To observe all events of a subject, call the `observe_events` function with the subject and an options object. Set the `recursive` option to `false`. This ensures that only events of the given subject are returned, not events of nested subjects.

The function returns a stream from which you can retrieve one event at a time:

```rust
let result = client
  .observe_events("/books/42", Some(
    ObserveEventsOptions {
      recursive: false,
      from_latest_event: None,
      lower_bound: None,
    }
  ))
  .await;

match result {
  Err(err) => // ...
  Ok(mut stream) => {
    while let Some(event) = stream.next().await {
      // ...
    }
  }
}
```

*(Added in 2.3; Changed in 2.4)* *Note that EventSourcingDB sends a heartbeat every second while there are no events to send. If neither an event nor a heartbeat arrives for 30 seconds, for example because the connection stalled, the stream returns a `ClientError::HeartbeatTimeout` error, closes the connection, and ends.*

### Observing From Subjects Recursively

If you want to observe not only all the events of a subject, but also the events of all nested subjects, set the `recursive` option to `true`:

```rust
let result = client
  .observe_events("/books/42", Some(
    ObserveEventsOptions {
      recursive: true,
      ..Default::default()
    }
  ))
  .await;
```

This also allows you to observe *all* events ever written. To do so, provide `/` as the subject and set `recursive` to `true`, since all subjects are nested under the root subject.

### Specifying Bounds

Sometimes you do not want to observe all events, but only a range of events. For that, you can specify the `lower_bound` option.

Specify the ID and whether to include or exclude it:

```rust
let result = client
  .observe_events("/books/42", Some(
    ObserveEventsOptions {
      recursive: false,
      lower_bound: Some(Bound {
        bound_type: BoundType::Inclusive,
        id: "100",
      }),
      ..Default::default()
    }
  ))
  .await;
```

### Starting From the Latest Event of a Given Type

To observe starting from the latest event of a given type, provide the `from_latest_event` option and specify the subject, the type, and how to proceed if no such event exists.

*(Changed in 2.2)* Possible options are `WaitForEvent`, which waits for an event of the given type to happen, or `ReadEverything`, which effectively behaves as if `from_latest_event` was not specified:

```rust
let result = client
  .observe_events("/books/42", Some(
    ObserveEventsOptions {
      recursive: false,
      from_latest_event: Some(
        ObserveFromLatestEventOptions {
          subject: "/books/42",
          ty: "io.eventsourcingdb.library.book-borrowed",
          if_event_is_missing: ObserveEventMissingStrategy::ReadEverything,
        }
      ),
      ..Default::default()
    }
  ))
  .await;
```

*Note that `from_latest_event` and `lower_bound` can not be provided at the same time.*

### Aborting Observing

To stop observing, drop the stream, which closes the connection.

## Registering an Event Schema

To register an event schema, call the `register_event_schema` function and hand over an event type and the desired schema:

```rust
client.register_event_schema(
  "io.eventsourcingdb.library.book-acquired",
  &json!({
    "type": "object",
    "properties": {
      "title":  { "type": "string" },
      "author": { "type": "string" },
      "isbn":   { "type": "string" },
    },
    "required": [
      "title",
      "author",
      "isbn",
    ],
    "additionalProperties": false,
  }),
).await;
```

## Listing Subjects *(Changed in 3.0)*

To list all subjects, call the `list_subjects` function with `/` as the base subject. The function returns a stream from which you can retrieve one subject at a time:

```rust
let result = client.list_subjects(Some("/")).await;
match result {
  Ok(subjects) => // ...
  Err(err) => // ...
}
```

If you only want to list subjects within a specific branch, provide the desired base subject instead:

```rust
let result = client.list_subjects(Some("/books")).await;
```

## Listing Event Types *(Changed in 3.0)*

To list all event types, call the `list_event_types` function. The function returns a stream from which you can retrieve one event type at a time:

```rust
let result = client.list_event_types().await;
match result {
  Ok(event_types) => // ...
  Err(err) => // ...
}
```

## Listing a Specific Event Type *(Changed in 3.0)*

To list a specific event type, call the `read_event_type` function. The function returns the detailed event type, which includes the schema:

```rust
let event_type_name = "io.eventsourcingdb.library.book-acquired";
let result = client.read_event_type(event_type_name).await;
match result {
  Ok(event_type) => // ...
  Err(err) => // ...
}
```

## Verifying an Event's Hash

To verify the integrity of an event, call the `verify_hash` function on the event instance. This recomputes the event's hash locally and compares it to the hash stored in the event. If the hashes differ, the function returns an `EventError::HashVerificationFailed` error:

```rust
let result = event.verify_hash();
match result {
  Ok(()) => // ...
  Err(err) => // ...
}
```

*Note that this only verifies the hash. If you also want to verify the signature, you can skip this step and call `verify_signature` directly, which performs a hash verification internally.*

## Verifying an Event's Signature *(Changed in 3.0)*

To verify the authenticity of an event, call the `verify_signature` function on the event instance. This requires the public key that matches the private key used for signing on the server.

The function first verifies the event's hash, and then checks the signature. If any verification step fails, it returns an error: the errors of `verify_hash` for the hash, `EventError::MissingSignature` if the event has no signature, `EventError::MalformedSignature` if the signature can not be decoded, and `EventError::SignatureError` if the signature is invalid:

```rust
use ed25519_dalek::VerifyingKey;

// ...

let verification_key = // public key as VerifyingKey

let result = event.verify_signature(&verification_key);
match result {
  Ok(()) => // ...
  Err(err) => // ...
}
```

## Using Testcontainers

The support for Testcontainers is part of the **`testcontainer` feature** of the crate, so enable it for your tests:

```shell
cargo add eventsourcingdb --dev --features testcontainer
```

Then call the `Container::start_default()` function, get a client, and run your test code:

```rust
let container  = Container::start_default().await.unwrap();
let client = container.get_client().await.unwrap();
```

### Configuring the Container Instance

By default, `Container` uses the `latest` tag of the official EventSourcingDB Docker image. To change that use the provided builder and call the `with_image_tag` function.

```rust
let container = Container::builder()
  .with_image_tag("1.0.0")
  .start()
  .await.unwrap();
```

Similarly, you can configure the port to use and the API token. Call the `with_port` or the `with_api_token` function respectively:

```rust
let container = Container::builder()
  .with_port(4000)
  .with_api_token("secret")
  .start()
  .await.unwrap();
```

If you want to sign events, call the `with_signing_key` function. This generates a new signing and verification key pair inside the container:

```rust
let container = Container::builder()
  .with_signing_key()
  .start()
  .await.unwrap();
```

*(Changed in 2.2)* You can retrieve the private key (for signing) and the public key (for verifying signatures) once the container has been started. If the container was started without a signing key, there are none:

```rust
let signing_key = container.get_signing_key().unwrap();
let verification_key = container.get_verification_key().unwrap();
```

The `signing_key` is the private key EventSourcingDB signs events with.

The `verification_key` can be passed to `verify_signature` when verifying events read from the database.

### Configuring the Client Manually

In case you need to set up the client yourself, use the following functions to get details on the container:

- `get_host()` returns the host name
- `get_mapped_port()` returns the port
- `get_base_url()` returns the full URL of the container
- `get_api_token()` returns the API token

## Further Resources

- **Repository**: **[thenativeweb/eventsourcingdb-client-rust](https://github.com/thenativeweb/eventsourcingdb-client-rust)**
- **Registry**: **[crates.io](https://crates.io/crates/eventsourcingdb)**
- **License (MIT)**: **[License file](https://github.com/thenativeweb/eventsourcingdb-client-rust/blob/main/LICENSE.md)**
- **Issue tracker**: **[GitHub Issues](https://github.com/thenativeweb/eventsourcingdb-client-rust/issues)**