Defining Views

Define the shape of an item as a struct that derives Clone, Debug, Default, and PartialEq, and create an InMemoryView with a closure that returns the key of an item, to get a view that holds such items in memory:

#[derive(Clone, Debug, Default, PartialEq)]
struct BookItem {
  id: String,
  title: String,
  author: String,
  is_borrowed: bool,
  borrowed_until: String,
  revision: Option<Revision>,
}

fn new_catalog() -> InMemoryView<String, BookItem> {
  InMemoryView::with_revision_in(|item: &BookItem| item.id.clone(), |item| &mut item.revision)
}

let catalog = new_catalog();

An item derives neither Serialize nor Deserialize, since what a caller sees is decided by a query and its answer, not by the view (see Handling Queries over HTTP). It derives Clone, since the view hands out clones of its items (see Sharing Items with Readers), Default, which upsert starts from (see Adding Items), and Debug and PartialEq, so that a test can compare the items of the view with the ones it expects (see Testing Projections).

Every item has a revision of its own, which is the ID of the last event that changed it. with_revision_in takes a second closure, which returns a field of the item of the type Option<Revision>, and makes the view keep the revision there, so that a caller can hand it over to a command that uses Precondition::on_event_id (see Checking the Revision of the Caller). The view sets the field whenever it changes an item, so you never set it yourself. A view created with InMemoryView::new, which takes the first closure only, keeps the revisions to itself.

The revision of an item fits such a precondition only if the item stands for exactly one subject, and the projection applies every event type of that subject to the item. The precondition checks the last event of the subject, so as soon as an event lands in the subject that the view does not apply to the item, the two drift apart, and every command with the revision of the item fails with a conflict, until an event changes the item again. Apply an event type that does not change the item, such as one whose arm in evolve returns the state unchanged, with a change that does nothing (see Defining Projections). For an item that gathers several subjects, such as all books a reader has borrowed, there is no single subject its revision could stand for, so use Precondition::on_state_read for the commands instead (see Guarding Against Concurrent Changes).

Every method that changes the view takes the event it applies, as the Envelope that a projection gets (see Defining Projections). The ID of an event is a Revision, so the view compares IDs as numbers: event 10 is newer than event 9. An event that is not newer than the item it is about is skipped, so applying the same event twice changes nothing, as long as the item is still there. The view forgets the revision of an item it removes, so an event that adds the item, applied again after a later event has removed it, adds it again. A projection that is rebuilt applies every event once, in order, so with a view in memory, that does not happen.

Reading a view is asynchronous and may fail, as it does for a view in a database: it goes through the traits View and KeyedView, whose methods hand out a Result, so that code that reads a view works the same, whatever is behind it. The view in memory never fails, so its error is Infallible, and a reader needs no handling for it. Changing a view, on the other hand, goes through the methods of InMemoryView itself, which take effect at once. The kit has no trait for the methods that change a view, though: a projection holds its view by its type, such as InMemoryView, so moving it to a view in a database changes that type.

If an upcaster splits a stored event into several events, they all carry the ID of the stored event (see Versioning Events), and each its position among them, as the part field of the envelope. For the same ID, the view counts a later part as newer than an earlier one, so every one of them is applied, in order, also if several of them change the same item. Applying the stored event again still changes nothing, and the revision of the item stays the ID of the stored event. That is why a projection hands the envelope it gets on to the view, rather than one it builds itself.

Note that the view as a whole has a revision as well, which is the last event it has seen at all, rather than the last one that changed a particular item (see Reading Your Own Writes).

Adding Items#

To add an item, call the insert method with the event and the item:

let outcome = catalog.insert(
  &event,
  BookItem {
    id: "42".to_owned(),
    title: "2001 – A Space Odyssey".to_owned(),
    author: "Arthur C. Clarke".to_owned(),
    ..BookItem::default()
  },
)?;

insert returns an Outcome, which tells what it did: Outcome::Added if it added the item, and Outcome::AlreadyApplied if an item with the same key has seen the event, or a newer one, so the event was skipped. The latter is not an error, since it happens when events are applied a second time. So a projection that does more than store the item, for example one that also counts the books, does so only for Outcome::Added (see Changing and Removing Items).

If the key of the item is already taken, and the event is newer than the item with that key, insert fails with ViewError, which belongs to Category::Permanent, since two items with the same key point to a mistake in the events or in the key. To add an item or change the existing one, call the upsert method instead, with the key, the event, and a closure that changes the item. It starts from the existing item, or from the Default of the item if there is none, so it has to set the fields that make up the key:

let outcome = catalog.upsert("42", &event, |item| {
  item.id = "42".to_owned();
  item.is_borrowed = true;
})?;

upsert returns an Outcome as well: Outcome::Added if it added the item, Outcome::Applied if it changed the existing one, and Outcome::AlreadyApplied if the existing item has seen the event, or a newer one, so the event was skipped.

Note that an item whose key, after the change, differs from the given one fails with ViewError, which also catches a closure that forgets to set the key.

Reading Items#

To read the item with a given key, call the get method of the KeyedView trait. It returns None if there is none:

let Ok(book) = catalog.get("42").await;

To read all items, call the all method of the View trait. It returns a stream of clones of the items, taken when all is called, in the order in which they were added, which hands out each item as a Result. To go through it, pin it with std::pin::pin!, and call the next method of StreamExt in a loop, or hand it to the steps of a query (see Defining Queries). StreamExt comes from the futures crate, which you add with cargo add futures:

let mut items = pin!(catalog.all());

while let Some(item) = items.next().await {
  let Ok(item) = item;

  // ...
}

The view in memory never fails while it hands out its items, so its error is Infallible, and let Ok(item) = item; needs no other arm. A view in a database, however, may fail halfway, which is why every view hands out a Result, and its stream ends after an error.

Note that the items you get are clones, which a later change of the view does not reach, and that taking them when all is called lets a query take its time without holding up the projection that feeds the view (see Sharing Items with Readers).

Changing and Removing Items#

To change the item with a given key, call the update method with the event and a closure that changes it. To remove it, call the delete method. Like insert and upsert, both return an Outcome, which tells what they did. Removing never fails, so delete returns the Outcome alone:

let updated = catalog.update("42", &event, |item| {
  item.is_borrowed = true;
})?;

let deleted = catalog.delete("42", &event);

Outcome	Meaning
`Outcome::Added`	A new item was added. Only `insert` and `upsert` report it.
`Outcome::Applied`	An existing item was changed or removed.
`Outcome::Missing`	There is no item with the key. Only `update` and `delete` report it.
`Outcome::AlreadyApplied`	The item has seen the event, or a newer one, so nothing was changed.

Neither Outcome::Missing nor Outcome::AlreadyApplied is an error, since both happen when events are applied a second time, as a later event may have removed the item already. If an event about an item that does not exist means that the view and the database disagree, say so in the projection, with an error type of its own that belongs to Category::Permanent:

#[derive(Debug)]
enum CatalogError {
  View(ViewError),
  NeverAcquired(String),
}

impl From<ViewError> for CatalogError {
  fn from(error: ViewError) -> Self {
    Self::View(error)
  }
}

impl fmt::Display for CatalogError {
  fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
    match self {
      Self::View(error) => error.fmt(formatter),
      Self::NeverAcquired(book_id) => write!(formatter, "book {book_id} was never acquired"),
    }
  }
}

impl std::error::Error for CatalogError {}

impl Categorized for CatalogError {
  fn category(&self) -> Option<Category> {
    Some(Category::Permanent)
  }
}

The projection then refuses the event, and the run names the event in the text of its failure, as in applying event 7: book 42 was never acquired (see Running Projections):

let outcome = catalog.update(book_id, &event, |item| {
  item.is_borrowed = true;
})?;
if outcome == Outcome::Missing {
  return Err(CatalogError::NeverAcquired(book_id.to_owned()));
}

Note that changing the key of an item fails with ViewError, and leaves the item as it was.

To change or remove several items at once, call the update_where or the delete_where method with a closure that selects them. Both return the number of items they changed or removed:

let is_by_clarke = |item: &BookItem| item.author == "Arthur C. Clarke";

let changed = catalog.update_where(is_by_clarke, &event, |item| {
  item.author = "Sir Arthur C. Clarke".to_owned();
})?;

let removed = catalog.delete_where(is_by_clarke, &event);

Note that the view stays locked while it runs a closure you hand over, such as a change, or the closure that selects the items for update_where and delete_where. Such a closure must not use the same view, not even to read an item with get or all, since that blocks forever. The same holds for upsert, and for the update method of an index. To copy data from another item of the view into the one you change, read it before, and use it inside the change. A panic in such a closure leaves the view usable for those who read it.

get and all hand out clones of the items, so every reader holds a copy of its own, which a later change does not reach, and which it may change as it likes without changing the view. A change, in turn, works on a clone of the item, which replaces the item only once the change is accepted. So a change may write into a field that holds a Vec or a HashMap, rather than building it anew, as in a view of the readers of the library and the books they have borrowed:

#[derive(Clone, Default)]
struct ReaderItem {
  id: String,
  book_ids: Vec<String>,
  due_dates: HashMap<String, String>,
}

fn new_readers() -> InMemoryView<String, ReaderItem> {
  InMemoryView::new(|item: &ReaderItem| item.id.clone())
}

For a BookBorrowed event, whose data is borrowed, the projection adds the book to its reader:

let outcome = readers.update(&borrowed.borrowed_by, &event, |item| {
  item.book_ids.push(book_id.to_owned());
  item.due_dates.insert(book_id.to_owned(), borrowed.borrowed_until.clone());
})?;

The view clones an item whenever it hands it out, with get, all, and the lookup method of an index, and whenever it changes one, with update, upsert, update_where, and the update method of an index. That is why an item derives Clone, which InMemoryView requires.

Note that a change that panics leaves the item as it was, since its clone never replaces the item.

Indexing Items#

Selecting items with a closure reads every item. To find items by a value directly, for example all books by an author, add a secondary index with the index method. It takes a closure that returns the value of an item, and returns an InMemoryIndex, which offers the lookup, update, and delete methods:

let by_author = catalog.index(|item| item.author.clone());

let books = by_author.lookup("Arthur C. Clarke");

let changed = by_author.update("Arthur C. Clarke", &event, |item| {
  item.is_borrowed = false;
})?;

let removed = by_author.delete("Arthur C. Clarke", &event);

Several items may share a value, so update and delete return the number of items they changed or removed, as update_where and delete_where do. The index follows every change to the view, also when the value of an item changes, and lookup hands out the items in the order in which they were added, as a stream of Results, as all does.

Note that adding an index reads every item, so add indexes before the view is used, and that its closure runs while the view is locked, so it must not use the view either.

To keep items somewhere else, for example in a database, implement the View trait, whose all method returns a stream of items and errors. Its error type implements Categorized, like every error the kit hands on, so that its category decides what happens with it, for example a failure that may pass, such as a lost connection, as Category::Transient. With the sqlx crate and PostgreSQL, which you add with cargo add sqlx --features postgres,runtime-tokio, the view and its error look like this:

struct BookTable {
  pool: PgPool,
}

#[derive(Debug)]
struct BookTableError(sqlx::Error);

impl fmt::Display for BookTableError {
  fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
    write!(formatter, "book table: {}", self.0)
  }
}

impl std::error::Error for BookTableError {
  fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
    Some(&self.0)
  }
}

impl Categorized for BookTableError {
  fn category(&self) -> Option<Category> {
    match self.0 {
      // The connection broke, or no connection was free in time, which may pass.
      sqlx::Error::Io(_) | sqlx::Error::PoolTimedOut => Some(Category::Transient),
      _ => Some(Category::Permanent),
    }
  }
}

A database fails not only before it reads the rows, but also while it reads them, for example when the connection breaks halfway, or when a row can not be read. So hand out such an error as an element of the stream, and end the stream there. Hand on the rows as they come, rather than reading all of them first, since a caller that has seen enough, for example with take or first, drops the stream (see Defining Queries). The stream of sqlx goes on after a row that can not be read, so scan ends it after the first error. try_get reads a column of a row, and is a method of the trait Row of sqlx, which use sqlx::Row; imports:

impl FromRow<'_, PgRow> for BookItem {
  fn from_row(row: &PgRow) -> Result<Self, sqlx::Error> {
    let revision: Option<String> = row.try_get("revision")?;

    Ok(BookItem {
      id: row.try_get("id")?,
      title: row.try_get("title")?,
      author: row.try_get("author")?,
      is_borrowed: row.try_get("is_borrowed")?,
      borrowed_until: row.try_get("borrowed_until")?,
      revision: revision
        .map(|revision| revision.parse())
        .transpose()
        .map_err(|error| sqlx::Error::Decode(Box::new(error)))?,
    })
  }
}

impl View<BookItem> for BookTable {
  type Error = BookTableError;

  fn all(&self) -> impl Stream<Item = Result<BookItem, BookTableError>> + Send {
    sqlx::query_as("SELECT id, title, author, is_borrowed, borrowed_until, revision FROM books")
      .fetch(&self.pool)
      .map_err(BookTableError)
      // A view ends after its first error.
      .scan(false, |has_failed, item| {
        if *has_failed {
          return future::ready(None);
        }
        *has_failed = item.is_err();

        future::ready(Some(item))
      })
  }
}

The error reaches the caller unchanged, through every step of a query, so its category, such as Category::Transient, still decides which status code an HTTP route answers with (see Mapping Errors to Status Codes).

If the view can find a single item by its key, also implement the KeyedView trait, whose get method returns the item, or None:

impl KeyedView<str, BookItem> for BookTable {
  async fn get(&self, id: &str) -> Result<Option<BookItem>, BookTableError> {
    sqlx::query_as(
      "SELECT id, title, author, is_borrowed, borrowed_until, revision FROM books WHERE id = $1",
    )
    .bind(id)
    .fetch_optional(&self.pool)
    .await
    .map_err(BookTableError)
  }
}

A query that reads a single item can then take any view that implements KeyedView<str, BookItem> instead of running over all items, and work with the view in memory and the one in a database alike. InMemoryView implements KeyedView too, for the type of its keys, and for every type a key can be borrowed as, such as str for a String.

Defining Views

Adding Items#

Reading Items#

Changing and Removing Items#

Sharing Items with Readers#

Indexing Items#