# Publishing Events

Other systems often need to learn about events, for example to send an email once a book has been borrowed, to call a webhook, to fill a search index, or to tell another team. How they learn about them depends on whether they can read the database themselves.

## Observing the Database

If the other system can read EventSourcingDB, let it observe the events itself. It then gets every event in the order it was stored, and after a restart, it resumes from the last event it has seen. Nothing needs to be forwarded (see **[Observing Events](/docs/eventsourcingdb/sdks/rust#observing-events)**).

## Publishing with a Projection

If it can not, because it is a mail server, a webhook, or a message broker, or if it should not depend on how the events of your application look, **forward the events with a projection that publishes them instead of writing a view**. Do not publish from where a command is executed: if the application stops right after `execute`, the event is stored, but never published, and nothing tells.

A projection that publishes needs to be resumable (see **[Resuming Projections](/docs/architecturekit/rust/projections#resuming-projections)**), and it reports a failure of the other system as an error of the category `Category::Transient`. Its events are an enum of their own, which lists only the event type it publishes, so that it is handed nothing else:

```rust
use std::{error::Error, fmt};

use architecturekit::{
  Categorized, Category, Envelope, Events, Projection, ResumableProjection, Revision,
};

#[derive(Events)]
pub enum LoanEvent {
  Borrowed(BookBorrowed),
}

pub struct LoanMailer {
  mailer: Mailer,
  checkpoints: CheckpointStore,
}

impl Projection for LoanMailer {
  type Event = LoanEvent;
  type Error = LoanMailerError;

  async fn apply(&self, event: Envelope<LoanEvent>) -> Result<(), LoanMailerError> {
    let LoanEvent::Borrowed(borrowed) = &event.data;

    // The ID of the event lets the receiver recognize a mail it got before.
    self
      .mailer
      .send(event.id, &borrowed.borrowed_by, "You have borrowed a book.")
      .await
      .map_err(LoanMailerError::Mail)
  }
}

impl ResumableProjection for LoanMailer {
  async fn checkpoint(&self) -> Result<Option<Revision>, LoanMailerError> {
    self.checkpoints.load("loan-mailer").await.map_err(LoanMailerError::Checkpoint)
  }

  async fn save_checkpoint(&self, event_id: Revision) -> Result<(), LoanMailerError> {
    self.checkpoints.save("loan-mailer", event_id).await.map_err(LoanMailerError::Checkpoint)
  }
}

#[derive(Debug)]
pub enum LoanMailerError {
  Mail(MailError),
  Checkpoint(CheckpointError),
}

impl fmt::Display for LoanMailerError {
  fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
    match self {
      Self::Mail(error) => write!(formatter, "sending a mail: {error}"),
      Self::Checkpoint(error) => write!(formatter, "keeping the checkpoint: {error}"),
    }
  }
}

impl Error for LoanMailerError {}

impl Categorized for LoanMailerError {
  fn category(&self) -> Option<Category> {
    match self {
      // The mail server may answer again later.
      Self::Mail(_) => Some(Category::Transient),
      Self::Checkpoint(error) => error.category(),
    }
  }
}
```

Here, `Mailer` and `CheckpointStore` stand for whatever your application uses to send mails and to keep a value, and `MailError` and `CheckpointError` for the errors they fail with. A projection has one error type for `apply`, `checkpoint`, and `save_checkpoint`, so `LoanMailerError` covers both. It hands on the category of the errors of the checkpoint store as it is, so the store reports a failure that may pass, such as a lost connection, as an error of the category `Category::Transient` itself. Otherwise, such a failure ends the run (see **[Resuming Projections](/docs/architecturekit/rust/projections#resuming-projections)**). Start the projection with `start_resumable_projection`, which goes on from its checkpoint (see **[Running Projections](/docs/architecturekit/rust/projections#running-projections)**):

```rust
use std::sync::Arc;

use architecturekit::{ProjectionOptions, Subject, Subjects};

let run = store.start_resumable_projection(
  Subjects::tree(Subject::new("/books")),
  Arc::new(LoanMailer { mailer, checkpoints }),
  token.clone(),
  ProjectionOptions::named("loan-mailer"),
);
```

## How a Publisher Behaves

Since publishing rides on a projection, it behaves like one:

- **It resumes only with a checkpoint.** A projection started with `start_projection` starts from the first event whenever the application starts, which suits a view held in memory. A publisher started that way would send every event again on every start. Within a run, the position is kept either way, so reconnecting repeats nothing.
- **An event may be published twice.** The checkpoint is saved after an event has been applied, so an event published right before the application stopped is published again after the restart. Hand over the ID of the event, so that the receiver can recognize an event it got before.
- **A transient failure is tried again.** For an error of the category `Category::Transient`, the run tries again from its checkpoint, with a growing delay. Any other error ends the run, and so does a panic.
- **Events arrive in order,** one at a time, as they were stored.

On its first start, the checkpoint is empty, so the projection reads every event that has ever been stored, and publishes all of them, for example by sending a mail for every book that has ever been borrowed. To publish only the events from now on, save the ID of the latest event as the checkpoint before the first start. To find it, read the subjects of the projection newest first, and stop after the first event:

```rust
use std::pin::pin;

use architecturekit::{Order, ReadOptions, Subject, Subjects};
use futures::StreamExt;

if checkpoints.load("loan-mailer").await?.is_none() {
  let subjects = Subjects::tree(Subject::new("/books"));
  let options = ReadOptions { order: Order::NewestFirst, ..ReadOptions::default() };
  let mut events = pin!(store.read(&subjects, &options).await?);

  if let Some(event) = events.next().await {
    checkpoints.save("loan-mailer", event?.id().parse()?).await?;
  }
}
```

If there are no events yet, the checkpoint stays empty, since there is nothing to skip.

A transient failure is tried again without a limit. So if the mail server stays down, the run keeps trying, and since it does not end, the liveness probe keeps answering `200`, and so does the readiness probe once the run has caught up (see **[Checking Health over HTTP](/docs/architecturekit/rust/checking-health-over-http)**). To notice it, check the `status` of the run, whose `phase` stays `Phase::Reconnecting`, with `since` telling since when and `error` telling why, or log every attempt with the observer of reconnects, whose `attempt` keeps growing (see **[Running Projections](/docs/architecturekit/rust/projections#running-projections)**).

*Note that how a projection is run depends on the method that starts it, not on the traits it implements. `start_resumable_projection` accepts only a projection that implements `ResumableProjection`, so the compiler makes sure that a publisher started with it keeps a checkpoint.*