Defining a View and a Query

Separating Reading from Writing#

The state of a book from Defining Events and State holds only what decisions need, and only for a single book. A list of all books with their titles needs a different shape, built from the same events. Keeping the two apart is the idea behind CQRS: the writing side decides on commands, and the reading side answers queries from read models, which ArchitectureKit calls views.

Defining the View#

A view holds items, each under a key of its own. For the catalog, an item describes a book, and its key is the ID of the book. A projection turns events into items: it adds an item when a book is acquired, and changes it when the book is borrowed or returned.

Create a file src/catalog.rs:

use std::sync::Arc;

use architecturekit::{Envelope, InMemoryView, Projection, Revision, Subject, ViewError};

use crate::book::BookEvent;
use crate::commands::BOOK_SUBJECT;

#[derive(Clone, Debug)]
pub struct BookItem {
  pub id: String,
  pub title: String,
  pub author: String,
  pub is_borrowed: bool,
  pub borrowed_until: String,
}

pub fn new_catalog() -> Arc<InMemoryView<String, BookItem>> {
  Arc::new(InMemoryView::new(|item: &BookItem| item.id.clone()))
}

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

impl CatalogProjection {
  pub fn new(catalog: Arc<InMemoryView<String, BookItem>>) -> Self {
    CatalogProjection { catalog }
  }
}

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

  async fn apply(&self, event: Envelope<BookEvent>) -> Result<(), ViewError> {
    let book_id = book_id_of(&event.subject);

    match &event.data {
      BookEvent::Acquired(acquired) => {
        self.catalog.insert(
          &event,
          BookItem {
            id: book_id,
            title: acquired.title.clone(),
            author: acquired.author.clone(),
            is_borrowed: false,
            borrowed_until: String::new(),
          },
        )?;
      }
      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);
  }
}

fn book_id_of(subject: &Subject) -> String {
  BOOK_SUBJECT.value(subject, "book").unwrap_or_default().to_owned()
}

InMemoryView::new creates a view that holds its items in memory, and takes the key of an item from the given closure. new_catalog wraps it in an Arc, since the projection writes to the catalog while the queries read from it.

A projection implements Projection: Event names the enum of the events it applies, here the one of the state, and Error the error it fails with, here the ViewError of the view. apply receives every event in an Envelope, which holds the metadata of the event, such as its id and subject, and its data in data, already decoded into the enum, so that a match handles every event type.

apply calls insert to add an item, and update to change one. Both take the event they apply, so that the view can skip an event it has applied before. Both also return an outcome, which tells what they did, for example whether they skipped the event. The catalog does not need it, so ? hands on only the error. Since the events do not contain the ID of the book, book_id_of takes it out of the subject with value of the subject scheme from Making Decisions, which returns the value of the placeholder book.

seen tells the catalog about every event the projection has seen, including those it has nothing to do with. The last of them is the catalog's revision, which lets callers read their own writes in Serving Commands and Queries over HTTP.

For details, see Defining Views and Projections.

Defining the Queries#

A query describes what someone wants to know, and a function answers it by reading a view. A view hands out its items as a stream, and the steps that filter and collect a stream come from the futures crate. Add it:

cargo add futures

Then create a file src/queries.rs with a query that lists books, optionally only those that are available, and one that gets a single book by its ID:

use architecturekit::{KeyedView, View};
use futures::{TryStreamExt, future};
use serde::Serialize;

use crate::catalog::BookItem;

#[derive(Serialize)]
pub struct ListBooks {
  pub only_available: bool,
}

pub async fn list_books<V: View<BookItem>>(
  catalog: &V,
  query: &ListBooks,
) -> Result<Vec<BookItem>, V::Error> {
  catalog
    .all()
    .try_filter(|item| future::ready(!query.only_available || !item.is_borrowed))
    .try_collect()
    .await
}

#[derive(Serialize)]
pub struct GetBook {
  pub book_id: String,
}

pub async fn get_book<V: KeyedView<str, BookItem>>(
  catalog: &V,
  query: &GetBook,
) -> Result<Option<BookItem>, V::Error> {
  catalog.get(&query.book_id).await
}

all of the trait View returns a stream of the items of the view, which InMemoryView hands out in the order in which they were added, each as a Result, since a view that keeps its items in a database may fail while it reads them. try_filter of TryStreamExt keeps only the items that match, and try_collect collects them into a Vec, or returns the error if the view has failed. For everything else a query can do with the stream, such as ordering and paging, see Defining Queries.

get_book does not run over all items, since the catalog can find a book by its key. So it takes a KeyedView, a view that offers get, which returns the item with the given key, or None if there is none. That way, a missing book can be told apart from a failure, which is the error. InMemoryView is a KeyedView, also for a key of type str when its keys are Strings, so the catalog serves both queries (see Defining Views). You put get_book to use in Serving Commands and Queries over HTTP, which also needs the queries to derive Serialize.

Running the Projection#

Add the new modules to src/lib.rs, which now reads:

pub mod book;
pub mod catalog;
pub mod commands;
pub mod deciders;
pub mod events;
pub mod queries;
pub mod store;

Replace the content of src/main.rs with a program that builds the catalog from the stored events and lists all books:

use std::error::Error;
use std::sync::Arc;

use architecturekit::Subjects;
use library::catalog::{CatalogProjection, new_catalog};
use library::commands::BOOK_SUBJECT;
use library::queries::{ListBooks, list_books};
use library::store::new_store;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
  let store = new_store().await?;

  let catalog = new_catalog();
  let projection = CatalogProjection::new(Arc::clone(&catalog));

  store.catch_up_projection(&Subjects::tree(BOOK_SUBJECT.root()), &projection).await?;

  let query = ListBooks { only_available: false };
  let books = list_books(&catalog, &query).await?;

  for book in books {
    println!("{book:?}");
  }

  Ok(())
}

root of the subject scheme returns /books, the part of its pattern before the first placeholder, and Subjects::tree stands for that subject together with every subject below it. So catch_up_projection reads all events that are stored below /books, and applies them to the projection. It returns once all stored events have been applied.

Run the program:

cargo run

It prints the book from Making Decisions, as BookItem derives Debug for it, now with its title and author from the BookAcquired event, and the return date from the BookBorrowed event:

BookItem { id: "42", title: "2001 – A Space Odyssey", author: "Arthur C. Clarke", is_borrowed: true, borrowed_until: "2026-10-24" }

The view lives in memory, so it is gone once the program ends, and the projection builds it again from the events on the next start. That is fine, because the events are the source of truth, and a view can always be rebuilt from them. For views that keep their data and resume where they stopped, see Resuming Projections.