es_query

The es_query! macro is a helper that allows you only to query the index table without needing to join with the events table.

The expansion of es_query! results in a call to the sqlx::query_as! macro - which means that you still get typesafety and compile time column validation.

Given the query we arrived at in the previous section - this is what a find_by_name fn could look like:

extern crate es_entity;
extern crate sqlx;
extern crate serde;
fn main () {}
use serde::{Deserialize, Serialize};
use es_entity::*;
es_entity::entity_id! { UserId }
#[derive(EsEvent, Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
#[es_event(id = "UserId")]
pub enum UserEvent {
    Initialized { id: UserId, name: String },
}
pub struct NewUser { name: String }
impl IntoEvents<UserEvent> for NewUser {
    type IdSource = FromRepo;
    fn into_events(self, _builder: EntityEventsBuilder<UserEvent, FromRepo>) -> EntityEvents<UserEvent> {
        unimplemented!()
    }
}
#[derive(EsEntity)]
pub struct User {
    pub id: UserId,
    events: EntityEvents<UserEvent>,
}
impl TryFromEvents<UserEvent> for User {
    fn try_from_events(events: EntityEvents<UserEvent>) -> Result<Self, EntityHydrationError> {
        unimplemented!()
    }
}
use sqlx::PgPool;
use es_entity::*;

#[derive(Debug)]
pub enum UserError {
    Sqlx(sqlx::Error),
    HydrationError(EntityHydrationError),
    NotFound,
}
impl From<sqlx::Error> for UserError {
    fn from(e: sqlx::Error) -> Self { Self::Sqlx(e) }
}
impl From<EntityHydrationError> for UserError {
    fn from(e: EntityHydrationError) -> Self { Self::HydrationError(e) }
}

pub struct Users {
    pool: PgPool
}
impl Users {
    pub async fn find_by_name(&self, name: String) -> Result<User, UserError> {
        let rows = sqlx::query_as!(
            GenericEvent::<UserId>,
            r#"
            WITH target_entity AS (
              SELECT id
              FROM users
              WHERE name = $1
            )
            SELECT e.id as entity_id, e.sequence, e.event, e.context as "context: ContextData", e.recorded_at, NULL::jsonb as "forgettable_payload?"
            FROM user_events e
            JOIN target_entity te ON e.id = te.id
            ORDER BY e.sequence;
        "#,
            name,
        )
        .fetch_all(&self.pool)
        .await?;
        EntityEvents::load_first(rows.into_iter())?
            .ok_or(UserError::NotFound)
    }
}

The es_query! macro removes the boilerplate of fetching the events and lets you just write the part that queries the index table:

SELECT id FROM users WHERE name = $1

On expansion it constructs the complete query (adding the JOIN with the events table) and hydrates the entities from the events. This simplifies the above implementation into:

extern crate es_entity;
extern crate sqlx;
extern crate serde;
fn main () {}
use serde::{Deserialize, Serialize};
use es_entity::*;
es_entity::entity_id! { UserId }
#[derive(EsEvent, Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
#[es_event(id = "UserId")]
pub enum UserEvent {
    Initialized { id: UserId, name: String },
}
pub struct NewUser { name: String }
impl IntoEvents<UserEvent> for NewUser {
    type IdSource = FromRepo;
    fn into_events(self, _builder: EntityEventsBuilder<UserEvent, FromRepo>) -> EntityEvents<UserEvent> {
        unimplemented!()
    }
}
#[derive(EsEntity)]
pub struct User {
    pub id: UserId,
    events: EntityEvents<UserEvent>,
}
impl TryFromEvents<UserEvent> for User {
    fn try_from_events(events: EntityEvents<UserEvent>) -> Result<Self, EntityHydrationError> {
        unimplemented!()
    }
}
use sqlx::PgPool;
use es_entity::*;

#[derive(EsRepo)]
#[es_repo(entity = "User")]
pub struct Users {
    pool: PgPool
}
impl Users {
    pub async fn find_by_name(&self, name: String) -> Result<User, es_entity::RepoFault> {
        Ok(es_query!(
            "SELECT id FROM users WHERE name = $1",
            name
        )
        .fetch_optional(&self.pool)
        .await?
        .ok_or(NotFound { entity: "User", column: Some("name"), value: name })?)
    }
}

The es_query! macro only works within fns defined on structs with EsRepo derived.

es_query! provides fetch_optional, returning Result<Option<Entity>, es_entity::RepoFault>. A repository read can fail transiently or fatally. To require an entity, construct a NotFound and propagate it with ?: it converts into a carrier accepting Fatal as Fatal(Invariant). A caller that tolerates absence uses maybe_find_by_*, returning Option:

async fn fetch_optional(<executor>) -> Result<Option<Entity>, es_entity::RepoFault>

// The `(_, bool)` signifies whether or not the query could have fetched more or the list is exhausted:
async fn fetch_n(<executor>, n) -> Result<(Vec<Entity>, bool), es_entity::RepoFault>