# Projections

This page shows how to define projections, which turn events into views, how to run them in the background and wait until they have caught up, how to resume them where they stopped, and how to apply events in batches. A projection can also forward events to other systems (see **[Publishing Events](/docs/architecturekit/rust/publishing-events)**).

## Defining Projections

A projection turns events into a view. Define it as a struct that holds the view, and implement the `Projection` trait. Its `Event` names the enum of the events it depends on, with `#[derive(Events)]`, and its `apply` method receives every event as an `Envelope`, which holds the metadata of the event, such as its `id`, `time`, and `subject`, and its data, decoded into the enum.

Which parts of the subject a view needs is up to the application. In this example, the `SubjectScheme` called `BOOKS` takes the ID of the book out of the subject (see **[Composing Subjects](/docs/architecturekit/rust/composing-subjects)**):

```rust
static BOOKS: LazyLock<SubjectScheme> = LazyLock::new(|| SubjectScheme::new("/books/{book_id}"));

struct CatalogProjection {
  catalog: InMemoryView<String, BookItem>,
}

impl Projection for CatalogProjection {
  type Event = BookEvent;
  type Error = ViewError;

  async fn apply(&self, event: Envelope<BookEvent>) -> Result<(), ViewError> {
    let Some(book_id) = BOOKS.value(&event.subject, "book_id") else {
      return Ok(());
    };

    match &event.data {
      BookEvent::Acquired(acquired) => {
        self.catalog.insert(
          &event,
          BookItem {
            id: book_id.to_owned(),
            title: acquired.title.clone(),
            author: acquired.author.clone(),
            ..BookItem::default()
          },
        )?;
      }
      BookEvent::Borrowed(borrowed) => {
        self.catalog.update(book_id, &event, |item| {
          item.is_borrowed = true;
          item.borrowed_until = borrowed.borrowed_until.clone();
        })?;
      }
      BookEvent::Returned(_) => {
        self.catalog.update(book_id, &event, |item| {
          item.is_borrowed = false;
          item.borrowed_until = String::new();
        })?;
      }
    }

    Ok(())
  }

  fn seen(&self, event_id: Revision) {
    self.catalog.seen(event_id);
  }
}
```

The event type of every variant is taken from the event it holds, so it does not have to be repeated, and **the type of an event and the type its data is decoded into can not drift apart**. An event whose subject does not follow the scheme is skipped. The `seen` method records every event in the view, so that a query can wait until the view has seen a write (see **[Tracking Revisions](/docs/architecturekit/rust/reading-your-own-writes#tracking-revisions)**).

The projection shares its view with the queries that read it, so a run takes it in an `Arc`, and the queries read the view through the same `Arc`:

```rust
let catalog_projection = Arc::new(CatalogProjection { catalog: new_catalog() });
```

`apply` takes `&self`, since the projection shares what it builds with those who read it. **Events of types the enum has no variant for are skipped**, since a projection usually reads more events than it depends on, so a projection may have an enum of its own, with only the event types it needs, or share the enum of the state, as here. If the data of an event can not be decoded, the run fails with `ProjectionError::Permanent`. An error returned by `apply` is passed on together with the category its type names (see **[Running Projections](/docs/architecturekit/rust/projections#running-projections)**).

If the view keeps the revisions of its items for a precondition, a skipped event makes the revision of its item fall behind the subject (see **[Defining Views](/docs/architecturekit/rust/defining-views)**). In that case, give every event type of the subject a variant, also one that does not change the item, such as `BookInspected`, whose arm in `evolve` returns the state unchanged. Its arm calls `update` with a change that does nothing, which only moves the revision of the item on. A projection that shares the enum of the state gets the variant along, and the compiler insists on an arm for it:

```rust
impl Projection for CatalogProjection {
  // ...

  async fn apply(&self, event: Envelope<BookEvent>) -> Result<(), ViewError> {
    // ...

    match &event.data {
      // ...
      BookEvent::Inspected(_) => {
        self.catalog.update(book_id, &event, |_| {})?;
      }
    }

    Ok(())
  }

  // ...
}
```

To have the projection see the same events as the state, return the same upcasters from its `upcasters` method as the state does (see **[Versioning Events](/docs/architecturekit/rust/versioning-events)**):

```rust
impl Projection for CatalogProjection {
  // ...

  fn upcasters() -> Option<&'static Upcasters> {
    Some(&LIBRARY_UPCASTERS)
  }
}
```

### Handling Every Event

A projection that has to see every event, for example to log it, names `eventsourcingdb::Event`, the event of the client SDK, as its `Event`. Its `apply` method then receives every event as it is stored, with its hash, without running any upcasters:

```rust
struct LogProjection;

impl Projection for LogProjection {
  type Event = eventsourcingdb::Event;
  type Error = Infallible;

  async fn apply(&self, event: Envelope<eventsourcingdb::Event>) -> Result<(), Infallible> {
    println!("{} {}", event.subject, event.event_type);

    Ok(())
  }
}
```

For a projection that needs no type of its own, use the `projection_fn` function, which turns a closure into a projection. The closure gets every event as an `Envelope`, typed by an enum or as it is stored, and returns a future of the result:

```rust
let log_projection = projection_fn(|event: Envelope<eventsourcingdb::Event>| async move {
  println!("{} {}", event.subject, event.event_type);

  Ok::<(), Infallible>(())
});
```

*Note that a closure brings along neither upcasters nor a `seen` method, so a view that it fills reaches no revision by itself, and waiting for one runs into the timeout. A projection that needs upcasters or a reliable revision has a type of its own.*

## Running Projections

To run a projection, call the `start_projection` method of the store with the subjects to read, the projection in an `Arc`, a `CancellationToken`, and the options. Say which subjects with `Subjects::tree`, which stands for the given subject together with every subject below it, or with `Subjects::exact`, which stands for the given subject alone. A projection usually reads a tree, such as every book below `/books`. **There is no default**, so that every projection says which one it means, since the client SDK reads a single subject unless told otherwise.

The method runs the projection on a task of its own and returns a `ProjectionRun` at once, which can be cloned, every clone watching the same run. The run first applies all events that are already stored, then follows new events until the token is canceled. **An application usually answers queries only once its views have caught up**, since a half-built view answers wrongly rather than slowly, so wait for that with the `wait_caught_up` function, which takes the runs:

```rust
let token = CancellationToken::new();

let run = store.start_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::clone(&catalog_projection),
  token.clone(),
  ProjectionOptions::named("catalog"),
);

wait_caught_up(&[&run]).await?;

// Start to answer queries.
```

`ProjectionOptions::named` gives the projection a name, by which its status, its failures, the observer of reconnects, and the health checks report it (see **[Checking Health over HTTP](/docs/architecturekit/rust/checking-health-over-http)**). `ProjectionOptions::default()` leaves it without a name, and the subject it reads tells it apart instead. The `name` method of the run returns it.

*Note that an empty name makes `start_projection` panic, and so does `Subject::new` for a subject that is not one, such as one that does not start with a slash. `start_projection` needs a Tokio runtime with the time driver, as the client SDK does.*

`wait_caught_up` waits until every run has applied the events that were stored when it started, and returns `Ok(())` then, and only then, so that `Ok(())` always means that the views are complete. To wait for several projections, hand over all of their runs, as in `wait_caught_up(&[&catalog_run, &readers_run])`. It waits for all of them at once, not one after the other, and fails with a `CaughtUpError` as soon as one of them ends before it has caught up:

- If a run ends with a failure before it catches up, it fails with `CaughtUpError::Failed`, which holds the failure of the run. Its text names the run by its name, as in `projection "catalog": …`, or, for a run without a name, by the subject it reads, as in `projection of "/books": …`, and it keeps the category of the failure.
- If a run ends without a failure before it catches up, because its token was canceled, it fails with `CaughtUpError::Stopped`, which belongs to no category, and whose text names the run the same way, as in `the projection "catalog" stopped before it caught up` or `the projection of "/books" stopped before it caught up`.

A run that has caught up counts as caught up, even if it has ended since. If several runs have ended before they caught up by the time `wait_caught_up` looks, the failure comes first, since it tells why, that of the first such run among the ones handed over, and a run that stopped comes next. **Dropping the future stops waiting**, so to wait for a limited time only, wrap it in `tokio::time::timeout`.

*Note that `wait_caught_up` returns `Ok(())` at once without any run, since none of them is behind.*

For finer control, the run offers what `wait_caught_up` waits for. `caught_up` waits until the run has applied the events that were stored when it started, and counts once only, also while the run reconnects later on. If the run ends before it catches up, `caught_up` never returns, so wait for `done` as well. `done` waits until the run has ended, which happens when the token is canceled, or on a failure that trying again will not fix. `error` returns the failure the run ended with, as a `ProjectionError` in an `Arc`. It returns `None` as long as the run has not ended, and if it ended because its token was canceled, since canceling the token is how a projection is stopped. If `apply` fails with an error that trying again will not fix, the run ends, and `error` returns it.

The failure names the event that `apply` failed at by its ID, as `ProjectionError::Projection`, whose text reads `applying event 42: …`, so that the log tells which event to look at. It keeps the error of the projection, which a caller can downcast to its type, together with its category. A failure outside of `apply`, such as one of `checkpoint`, or of beginning or committing a transaction, names no event (see **[Resuming Projections](/docs/architecturekit/rust/projections#resuming-projections)**).

If the projection panics, for example because `apply` unwraps a value that is not there, the run ends as well, rather than the task. `error` then returns `ProjectionError::Panicked`, which belongs to `Category::Permanent`, since a panic is a mistake in the code that trying again will not fix. It holds the value of the panic, if it is text, and for a panic in `apply`, it names the event as well, as in `permanent failure: applying event 42: the projection panicked: …`, while the panic hook of Rust reports where the panic happened, as for any other panic. This holds for every method of the projection that the run calls, and for the observer of reconnects (see below). Once the run has ended, `Health::liveness` answers `503`, so that the orchestrator restarts the application (see **[Checking Health over HTTP](/docs/architecturekit/rust/checking-health-over-http)**).

*Note that only canceling the token stops the projection. Dropping the run, or the token, does not, so keep a clone of the token for as long as the application runs, and cancel it when it shuts down (see **[Putting It Together](/docs/architecturekit/rust/putting-it-together)**). Canceling it stops waiting for the next event at once, while an event that is being applied is applied to the end.*

To wait until the run ends, as a process does that runs nothing else, wait for `done`:

```rust
run.done().await;

match run.error() {
  Some(error) => Err(error.into()),
  None => Ok(()),
}
```

*Note that if the database can not be reached at the start, the run keeps trying, and `caught_up` keeps waiting. To wait for a limited time only, wrap `wait_caught_up` in `tokio::time::timeout`. Its end stops the waiting, but not the run (see **[Putting It Together](/docs/architecturekit/rust/putting-it-together)**).*

If reading fails with a transient failure, the run waits and continues after the last event it has applied, until the token is canceled. That includes a stream that ends, for example because the database restarts, or because a load balancer ends long-lived connections: the database never ends a stream on its own, so the run takes it for a stream that broke off. It also includes a failure of the projection whose error belongs to `Category::Transient`. **The delay starts at one second, doubles with every attempt in a row, and never exceeds one minute.** It starts over once the projection has applied an event again, or has caught up and followed the stream for longer than the initial delay, even if no event arrived. That way, a load balancer that ends long-lived connections regularly does not hold back a quiet projection, while a database that fails before the projection has caught up, or within the initial delay after, is given ever more time.

*Note that how long the projection followed the stream is compared with the initial delay, not with the delay it has grown to. Otherwise, once an outage had let the delay grow to one minute, a load balancer that ends connections every 30 seconds would keep a quiet projection waiting for a minute after each of them, and a caller who wants to read their own write in the meantime would not see it (see **[Reading Your Own Writes](/docs/architecturekit/rust/reading-your-own-writes)**). In return, a database that ends every stream a few seconds after the projection has caught up is tried again every few seconds rather than once a minute. That is not a loop without a pause, and catching up, the expensive part, succeeds each time.*

A connection can also stall without being closed, for example behind a proxy that keeps it open but no longer passes anything on. While there is nothing else to send, the database sends a heartbeat every second, so the client ends a stream on which neither an event nor a heartbeat has arrived for 30 seconds, its stall timeout, and the run reconnects as after any other transient failure. Without that, the run would wait for the next event forever, while its status kept reporting `Phase::Live`.

To use other delays, or to learn about every attempt, for example to log it, call the `reconnect_delays` and `reconnect_observer` methods of the builder when creating the store:

```rust
let store = Store::builder(client, "https://library.eventsourcingdb.io")
  .reconnect_delays(Duration::from_millis(500), Duration::from_secs(30))
  .reconnect_observer(|reconnect| {
    println!(
      "observing again {:?} {} {} {:?}",
      reconnect.projection, reconnect.attempt, reconnect.error, reconnect.delay
    );
  })
  .build();
```

*Note that an initial delay of zero panics, since the projections would then read again without any pause, and so does a maximum delay below the initial one.*

The observer receives a `Reconnect` with these fields:

- `projection` is the name the projection was given with `ProjectionOptions::named`, or `None` if it has none.
- `subject` is the subject the projection reads.
- `error` is the reason, such as a stream that broke off, as a `ProjectionError` in an `Arc`.
- `delay` is how long the projection waits before the next attempt.
- `attempt` counts the attempts in a row, starting at one. It starts over together with the delay once the projection has applied an event, or has followed the stream for longer than the initial delay. So it stays at `1` while a load balancer ends long-lived connections regularly.

*Note that a database that can not be reached is retried as well, since that is usually transient. The observer is how to notice a database that stays unreachable. A failure that trying again will not fix, for example a rejected API token, ends the run (see **[Handling Errors](/docs/architecturekit/rust/handling-errors)**).*

To find out where a run stands, call the `status` method. It returns a `ProjectionStatus` with these fields:

- `phase` is `Phase::CatchingUp`, `Phase::Live`, `Phase::Reconnecting`, or `Phase::Stopped`.
- `since` is when the phase began. For `Phase::Reconnecting`, that is when the disruption began, not when the latest attempt did.
- `error` is why the run is reconnecting or has stopped, such as a stream that broke off. It is `None` if the run stopped because its token was canceled.
- `attempts` counts the attempts to read again within the current disruption. Unlike `attempt` of `Reconnect`, it starts over whenever the run has caught up, even if the stream ends again within the initial delay.
- `revision` is the ID of the last event the run has applied and committed, or `None` if there is none.
- `has_caught_up` tells whether the run has caught up at least once. Like `caught_up`, it stays `true` while the run reconnects later on.

```rust
let status = run.status();

let is_stale = SystemTime::from(status.since)
  .elapsed()
  .is_ok_and(|elapsed| elapsed > Duration::from_secs(5 * 60));
if status.phase == Phase::Reconnecting && is_stale {
  // The view has not been up to date for more than five minutes.
}
```

To only apply the events that are already stored, for example for a batch job or in a test, call the `catch_up_projection` method instead. It takes the subjects and a reference to the projection, and returns once all stored events have been applied. Dropping its future stops it, so that a read model that is only partly built never looks complete. If the projection panics, it fails with the same `ProjectionError::Panicked` a run ends with, rather than panicking:

```rust
store
  .catch_up_projection(&Subjects::tree(Subject::new("/books")), catalog_projection.as_ref())
  .await?;
```

For a resumable or a transactional projection, call the methods for its kind instead (see **[Resuming Projections](/docs/architecturekit/rust/projections#resuming-projections)**).

## Resuming Projections

By default, a projection starts from the first event every time it runs, which fits a view held in memory. For a view that keeps its data, the projection can resume where it stopped instead.

If the view can store a checkpoint, but not together with the data, additionally implement the `ResumableProjection` trait. `checkpoint` returns the ID of the last event saved, or `None` if there is none, and `save_checkpoint` saves it:

```rust
struct BookTableProjection {
  // ...
}

impl Projection for BookTableProjection {
  type Event = BookEvent;
  type Error = BookTableError;

  async fn apply(&self, event: Envelope<BookEvent>) -> Result<(), BookTableError> {
    todo!()
  }
}

impl ResumableProjection for BookTableProjection {
  async fn checkpoint(&self) -> Result<Option<Revision>, BookTableError> {
    todo!()
  }

  async fn save_checkpoint(&self, event_id: Revision) -> Result<(), BookTableError> {
    todo!()
  }
}
```

`BookTableError` is the error of the view in a database (see **[Indexing Items](/docs/architecturekit/rust/defining-views#indexing-items)**).

*Note that the checkpoint is saved after the events have been applied. After a crash, events may therefore be applied a second time, so `apply` must be idempotent.*

Which way a projection is run is decided by the method that starts it, not by the traits it implements. **`start_projection` and `catch_up_projection` run every projection from the first event**, a resumable one as well, without reading or saving its checkpoint. To resume a projection, call `start_resumable_projection` or `catch_up_resumable_projection` instead, which take the same arguments, and which the compiler accepts only for a projection that implements `ResumableProjection`:

```rust
let run = store.start_resumable_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::new(BookTableProjection {}),
  token.clone(),
  ProjectionOptions::named("book-table"),
);
```

If the view can store the data and the checkpoint together, as a relational database can, implement the `TransactionalProjection` trait instead of `Projection`. A transactional projection applies events only within a transaction, so it has no `apply` method of its own. `begin` starts a transaction and returns a `Transaction`, which applies the events, and commits them together with the ID of the last event, or rolls them back:

```rust
struct TransactionalBookTableProjection {
  // ...
}

impl TransactionalProjection for TransactionalBookTableProjection {
  type Event = BookEvent;
  type Error = BookTableError;
  type Transaction = BookTableTransaction;

  async fn checkpoint(&self) -> Result<Option<Revision>, BookTableError> {
    todo!()
  }

  async fn begin(&self) -> Result<BookTableTransaction, BookTableError> {
    todo!()
  }
}

struct BookTableTransaction {
  // ...
}

impl Transaction for BookTableTransaction {
  type Event = BookEvent;
  type Error = BookTableError;

  async fn apply(&mut self, event: Envelope<BookEvent>) -> Result<(), BookTableError> {
    todo!()
  }

  async fn commit(self, last_event_id: Revision) -> Result<(), BookTableError> {
    todo!()
  }

  async fn rollback(self) -> Result<(), BookTableError> {
    todo!()
  }
}
```

With sqlx, `begin` starts a transaction of the database, which the `Transaction` holds, and `apply` writes the events into it, decoded into the enum of the projection, as `apply` of a `Projection` gets them:

```rust
struct TransactionalBookTableProjection {
  pool: PgPool,
}

impl TransactionalProjection for TransactionalBookTableProjection {
  // ...

  async fn begin(&self) -> Result<BookTableTransaction, BookTableError> {
    let transaction = self.pool.begin().await.map_err(BookTableError)?;

    Ok(BookTableTransaction { transaction })
  }
}

struct BookTableTransaction {
  transaction: sqlx::Transaction<'static, Postgres>,
}

impl Transaction for BookTableTransaction {
  type Event = BookEvent;
  type Error = BookTableError;

  async fn apply(&mut self, event: Envelope<BookEvent>) -> Result<(), BookTableError> {
    let Some(book_id) = BOOKS.value(&event.subject, "book_id") else {
      return Ok(());
    };

    if let BookEvent::Acquired(acquired) = &event.data {
      sqlx::query("INSERT INTO books (id, title, author) VALUES ($1, $2, $3)")
        .bind(book_id)
        .bind(&acquired.title)
        .bind(&acquired.author)
        .execute(&mut *self.transaction)
        .await
        .map_err(BookTableError)?;
    }

    Ok(())
  }

  async fn commit(mut self, last_event_id: Revision) -> Result<(), BookTableError> {
    let saved = sqlx::query("UPDATE checkpoint SET event_id = $1")
      .bind(last_event_id.to_string())
      .execute(&mut *self.transaction)
      .await;
    if let Err(error) = saved {
      let _ = self.transaction.rollback().await;
      return Err(BookTableError(error));
    }

    self.transaction.commit().await.map_err(BookTableError)
  }

  async fn rollback(self) -> Result<(), BookTableError> {
    self.transaction.rollback().await.map_err(BookTableError)
  }
}
```

*Note that `rollback` is not called after `commit` fails, so `commit` has to roll back itself if it can not finish, as above.*

To run a transactional projection, call the `start_transactional_projection` or the `catch_up_transactional_projection` method. They take the same arguments as those for the other kinds:

```rust
let run = store.start_transactional_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::new(TransactionalBookTableProjection { pool }),
  token.clone(),
  ProjectionOptions::named("book-table"),
);
```

*Note that a transactional projection has no `seen` method, since where data and revision have to become durable together, the revision belongs inside the transaction, and the projection writes it itself (see **[Tracking Revisions](/docs/architecturekit/rust/reading-your-own-writes#tracking-revisions)**).*

The place where a view keeps its data can fail as well, and such a failure counts like an error of `apply`, whether it comes from `checkpoint`, `save_checkpoint`, `begin`, `commit`, or the `apply` method of a `Transaction`. **An error that does not belong to `Category::Transient` ends the run**, so `Health::liveness` answers `503`, and the orchestrator restarts the application (see **[Checking Health over HTTP](/docs/architecturekit/rust/checking-health-over-http)**). That fits a mistake, such as a statement that the database refuses, but not a failure that may pass, such as a lost connection or a deadlock. Report such a failure with an error whose category is `Category::Transient`, as `BookTableError` does for a lost connection. The run then waits and tries again from where it stopped, with the same growing delay as when reading from EventSourcingDB fails (see **[Running Projections](/docs/architecturekit/rust/projections#running-projections)**). A panic in any of these methods, on the other hand, always ends the run, since it is a mistake in the code. A panic in the `apply` method of a `Transaction` rolls the transaction back first, as an error does, and if rolling back fails as well, the run reports that failure as `ProjectionError::RollbackFailed`, alongside the one that made it roll back, never instead of it.

*Note that which failures may pass depends on the database and its driver, so mark only those. A failure that is marked as transient but never passes keeps the run trying forever, while `Health::liveness` keeps answering `200`.*

## Batching Events

By default, the checkpoint is saved, or the transaction is committed, after every event. While a projection catches up, for example when it starts for the first time, that takes a lot longer than needed. To apply several events at once while catching up, implement the `catch_up_batch_size` method of `ResumableProjection` or `TransactionalProjection`, and return how many:

```rust
impl ResumableProjection for BookTableProjection {
  // ...

  fn catch_up_batch_size(&self) -> usize {
    1000
  }
}
```

The default is `1`, and `0` counts as `1`. For a transactional projection, a batch is one transaction, so a failure rolls back all of its events, and the next attempt starts after the last commit.

Once the projection has caught up, it saves the checkpoint, or commits, after every event again. Events then arrive one at a time, and a batch that waited to fill up would hold back the ones that have already arrived.

*Note that a resumable projection may apply up to that many events a second time after a crash.*