Putting It Together

It uses the pieces as the other pages of this documentation define them: BookEvent from Defining State, acquire_book, borrow_book, and return_book from Making Decisions, new_catalog and CatalogProjection from Defining Views and Defining Projections, ListBooks and list_books from Defining Queries, BOOKS from Composing Subjects, App and User from Setting Up an HTTP API, handle_borrow_book from Handling Commands over HTTP, ListBooksRequest and BookBody from Handling Queries over HTTP, and the handle_list_books that lets a caller read its own writes from Reading Your Own Writes over HTTP. handle_borrow_book builds the BorrowBook with the field expected_event_id, which checks the revision of the caller (see Checking the Revision of the Caller).

Besides the crates of the other pages, main needs the feature signal of Tokio, and a logger for the log crate, which the kit logs with (see Setting Up an HTTP API), here env_logger:

cargo add tokio --features macros,rt-multi-thread,signal
cargo add env_logger log

use std::{error::Error, sync::Arc, time::Duration};

use architecturekit::{
  CancellationToken, Events, ProjectionOptions, Store, Subject, Subjects,
  eventsourcingdb::Client,
  httpapi::{Health, answer_panic, query},
  wait_caught_up,
};
use axum::{
  Router,
  routing::{get, post},
};
use tokio::{
  net::TcpListener,
  signal::{
    self,
    unix::{SignalKind, signal},
  },
};
use tower_http::catch_panic::CatchPanicLayer;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
  env_logger::init();

  let shutdown = CancellationToken::new();
  tokio::spawn(cancel_on_signal(shutdown.clone()));

  let client = Client::new("http://localhost:3000".parse()?, "secret");
  let store = Store::builder(client, "https://library.eventsourcingdb.io").build();

  store.register_schemas(BookEvent::schemas()).await?;

  // The projection stops only after the server, since the requests that the
  // server finishes may still wait for the view.
  let stop_projection = CancellationToken::new();

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

  let waiting =
    async { tokio::time::timeout(Duration::from_secs(60), wait_caught_up(&[&run])).await };
  let caught_up = tokio::select! {
    // The process was asked to stop while waiting.
    () = shutdown.cancelled() => return Ok(()),
    caught_up = waiting => caught_up,
  };
  match caught_up {
    Ok(Ok(())) => {}
    Ok(Err(error)) => return Err(format!("the catalog has not caught up: {error}").into()),
    Err(_) => return Err("the catalog has not caught up within a minute".into()),
  }

  let app = App {
    store: Arc::new(store),
    acquire_book: Arc::new(acquire_book()),
    borrow_book: Arc::new(borrow_book()),
    return_book: Arc::new(return_book()),
    catalog_projection,
  };

  let router = Router::new()
    .route("/api/books/{id}/borrow", post(handle_borrow_book))
    .route("/api/books", query(handle_list_books))
    .route("/ready", get(Health::readiness([run.clone()])))
    .route("/live", get(Health::liveness([run.clone()])))
    .layer(CatchPanicLayer::custom(answer_panic))
    .with_state(app);

  let listener = TcpListener::bind("0.0.0.0:8080").await?;
  axum::serve(listener, router).with_graceful_shutdown(shutdown.cancelled_owned()).await?;

  stop_projection.cancel();
  run.done().await;

  if let Some(error) = run.error() {
    log::error!("{error}");
  }

  Ok(())
}

/// Cancels the token once the process is asked to stop, with Ctrl+C, or with
/// SIGTERM, as an orchestrator does.
async fn cancel_on_signal(shutdown: CancellationToken) {
  let mut terminate = signal(SignalKind::terminate()).expect("SIGTERM can be handled");

  tokio::select! {
    _ = signal::ctrl_c() => {}
    _ = terminate.recv() => {}
  }

  shutdown.cancel();
}

cancel_on_signal cancels shutdown once the process is asked to stop, as an orchestrator does with SIGTERM. with_graceful_shutdown then lets the server finish the requests it has begun, which may still wait for the view (see Reading Your Own Writes over HTTP). That is why the projection runs with a token of its own, which main cancels only after the server has stopped. Then it waits for done, and logs the failure of the run, which is None unless the run had ended on a failure before. SignalKind::terminate comes from the module tokio::signal::unix, which exists on Unix only, where orchestrators such as Kubernetes run their containers.

A database that can not be reached at the start makes register_schemas fail right away, and main returns the error. The server starts only once the view has caught up, so that no query sees a half-built view. main waits for that for at most a minute, or until the process is asked to stop, which ends the waiting, but not the projection, since that runs with a token of its own. If catching up takes longer than a minute, for example because the database has stopped answering in the meantime, main gives up rather than wait without end, so that the orchestrator restarts the application. Choose the limit with room for the history to grow, since a view that never catches up within it keeps the application from ever starting. Until the server starts, the health checks do not answer either, so give the application that long to start, for example with a startup probe in Kubernetes.

Set up this way, everything lives as long as main does. If you move the setup of the projection into a function of its own, the run goes on after that function has returned, since it runs on a task of its own, and it stops only once its token is canceled, or on a failure that trying again will not fix. So have the function return the token along with the run, and cancel it on shutdown, as main cancels stop_projection.