use darling::ToTokens;
use proc_macro2::TokenStream;
use quote::{TokenStreamExt, quote};
use super::{
events_write::{EventSource, EventsInsert, ForgettablePayloads},
options::*,
};
pub struct CreateFn<'a> {
in_op_only: bool,
entity: &'a syn::Ident,
id: &'a syn::Ident,
event: &'a syn::Ident,
table_name: &'a str,
events_table_name: &'a str,
event_ctx: bool,
forgettable_table_name: Option<&'a str>,
columns: &'a Columns,
constraint_violation: syn::Ident,
nested_fn_names: Vec<syn::Ident>,
post_hydrate_hook: bool,
post_persist_hook: bool,
id_source: IdSourceOpt,
can_reject: bool,
write_error_ty: TokenStream,
ensure_keys: Vec<EnsureKey>,
ensure_rejection: Option<syn::Ident>,
scope_ty: Option<syn::Ident>,
#[cfg(feature = "instrument")]
repo_name_snake: String,
}
impl<'a> From<&'a RepositoryOptions> for CreateFn<'a> {
fn from(opts: &'a RepositoryOptions) -> Self {
Self {
id_source: opts.id_source(),
can_reject: opts.create_can_reject(),
write_error_ty: opts.create_error_ty(),
ensure_keys: opts.ensure_keys(),
ensure_rejection: opts.ensure_can_reject().then(|| opts.ensure_rejection()),
scope_ty: (!opts.columns.scope_columns().is_empty()).then(|| opts.scope_type_ident()),
in_op_only: opts.in_op_only(),
table_name: opts.table_name(),
entity: opts.entity(),
id: opts.id(),
event: opts.event(),
events_table_name: opts.events_table_name(),
event_ctx: opts.event_context_enabled(),
forgettable_table_name: opts.forgettable_table_name(),
constraint_violation: opts.constraint_violation(),
nested_fn_names: opts
.all_nested()
.map(|f| f.create_nested_fn_name())
.collect(),
columns: &opts.columns,
post_hydrate_hook: opts.post_hydrate_hook.is_some(),
post_persist_hook: opts.post_persist_hook.is_some(),
#[cfg(feature = "instrument")]
repo_name_snake: opts.repo_name_snake_case(),
}
}
}
impl ToTokens for CreateFn<'_> {
fn to_tokens(&self, tokens: &mut TokenStream) {
let entity = self.entity;
let constraint_values = if self.can_reject {
self.columns.constraint_values(entity, true)
} else {
quote! {}
};
let constraint_violation = &self.constraint_violation;
let write_error_ty = &self.write_error_ty;
let (rejection_prelude, rejection_tail) = if self.can_reject {
(
quote! { use es_entity::errlanes::ResultExt as _; },
quote! { .map_rejected(|r| r.with_attempted(__constraint_values))? },
)
} else {
(
quote! { use es_entity::errlanes::ResultExt as _; },
quote! { .narrow_rejected()? },
)
};
let generated = self.id_source.is_repo();
let id_source = self.id_source.marker();
let resolve_id = if generated {
let id_ty = self.id;
let resolved = self.id_source.resolve(id_ty);
quote! { let __id: #id_ty = #resolved; }
} else {
quote! {}
};
let builder = self.id_source.builder(quote! { __id.clone() });
let (attempted_id, caller_id_check, classify_create) = if generated {
(
quote! {},
quote! {},
quote! { Self::classify_create_write(e) },
)
} else {
(
quote! {
let __attempted_id = (*id).clone();
},
quote! {
if events.id() != &__attempted_id {
return Err(errlanes::Fatal::invariant("caller's id differs from the id the entity was built under").into());
}
},
quote! { Self::classify_create_write(e, move |_| Some(__attempted_id)) },
)
};
let nested = |target: &TokenStream| -> Vec<TokenStream> {
self.nested_fn_names
.iter()
.map(|f| {
quote! {
self.#f(op, &mut [&mut entity]).await.map_err(es_entity::LiftNestedWrite::<#target>::lift_nested)?;
}
})
.collect()
};
let maybe_mut_entity = if self.nested_fn_names.is_empty() {
quote! { entity }
} else {
quote! { mut entity }
};
let assignments = self
.columns
.variable_assignments_for_create(syn::parse_quote! { new_entity }, generated);
let table_name = self.table_name;
let column_names = self.columns.insert_column_names();
let placeholders = self.columns.insert_placeholders(0);
let arg_adds = self.columns.create_query_arg_adds();
let events_insert = EventsInsert::new(self.events_table_name, self.event_ctx);
let now_p = column_names.len() + 1;
let source = EventSource::PerEntityCte {
cte: "new_row",
offset_param: None,
};
let query = |on_conflict: &str| {
format!(
"WITH new_row AS (INSERT INTO {} ({}, created_at) VALUES ({}, COALESCE(${}, NOW())){} RETURNING id) {}",
table_name,
column_names.join(", "),
placeholders,
now_p,
on_conflict,
events_insert.sql(&source, now_p, now_p + 1),
)
};
let event_arg_adds: Vec<_> = events_insert
.arg_exprs(&source)
.into_iter()
.skip(1)
.map(|expr| quote! { __query_args.add(#expr).map_err(sqlx::Error::Encode)?; })
.collect();
let ctx_var = if self.event_ctx {
quote! { let contexts = events.serialize_new_event_contexts(); }
} else {
quote! {}
};
let forgettable_code = match self.forgettable_table_name {
Some(table) => {
let payloads = ForgettablePayloads {
table,
id_type: self.id,
event_type: self.event,
}
.insert_per_entity(quote! { events }, self.events_table_name, None);
quote! {
let offset = events.len_persisted();
let id = events.id();
#payloads
}
}
None => quote! {},
};
#[cfg(feature = "instrument")]
let (instrument_attr, record_id, error_recording) = {
let entity_name = entity.to_string();
let repo_name = &self.repo_name_snake;
let span_name = format!("{}.create", repo_name);
(
quote! {
#[tracing::instrument(name = #span_name, skip_all, fields(entity = #entity_name, id = tracing::field::Empty, error = tracing::field::Empty, exception.message = tracing::field::Empty, exception.type = tracing::field::Empty))]
},
quote! {
tracing::Span::current().record("id", tracing::field::debug(&id));
},
quote! {
if let Err(ref e) = __result {
tracing::Span::current().record("error", true);
tracing::Span::current().record("exception.message", tracing::field::display(e));
tracing::Span::current().record("exception.type", std::any::type_name_of_val(e));
}
},
)
};
#[cfg(not(feature = "instrument"))]
let (instrument_attr, record_id, error_recording) = (quote! {}, quote! {}, quote! {});
let post_hydrate_check = if self.post_hydrate_hook {
quote! {
self.execute_post_hydrate_hook(&entity)?;
}
} else {
quote! {}
};
let post_persist_check = if self.post_persist_hook {
quote! {
self.execute_post_persist_hook(op, &entity, entity.events().last_persisted(n_events)).await?;
}
} else {
quote! {}
};
let standalone = (!self.in_op_only).then(|| {
quote! {
pub async fn create(
&self,
new_entity: <#entity as es_entity::EsEntity>::New
) -> Result<#entity, #write_error_ty> {
let mut op = self.begin_op().await?;
let res = self.create_in_op(&mut op, new_entity).await?;
op.commit().await?;
Ok(res)
}
}
});
let id_ty = self.id;
let create_query = query("");
let create_nested = nested("e! { #constraint_violation });
let ensure_fn = (!self.ensure_keys.is_empty()).then(|| {
let (error_ty, target, ensure_constraint_values, map_error) = match &self.ensure_rejection {
Some(rejection) => (
quote! { es_entity::RepoWriteError<#rejection> },
quote! { #rejection },
self.columns.constraint_values(entity, true),
quote! {
.map_err(|e| {
let __fail = #classify_create;
#rejection::narrow_write(
__fail.map_rejected(|r| r.with_attempted(__constraint_values)),
)
})?
},
),
None => (
quote! { es_entity::RepoFault },
quote! { #constraint_violation },
quote! {},
quote! { .map_err(|e| #classify_create.narrow_rejected())? },
),
};
let ensure_query = query(" ON CONFLICT DO NOTHING");
let mut next_param = 0;
let conditions: Vec<String> = self
.ensure_keys
.iter()
.map(|key| {
let parts: Vec<String> = key
.columns
.iter()
.map(|c| {
next_param += 1;
format!("{c} = ${next_param}")
})
.collect();
if parts.len() == 1 {
parts.join("")
} else {
format!("({})", parts.join(" AND "))
}
})
.collect();
let scope_columns = self.columns.scope_columns();
let lookup_query = format!(
"SELECT id, {} FROM {} WHERE {}",
conditions
.iter()
.map(|c| format!("COALESCE({c}, FALSE)"))
.chain(scope_columns.iter().map(|c| c.name().to_string()))
.collect::<Vec<_>>()
.join(", "),
table_name,
conditions.join(" OR ")
);
let missing_context = format!(
"{table_name}: ON CONFLICT DO NOTHING inserted no row and no unique key matches a stored row"
);
let spanning_context = format!("{table_name}: the unique keys match more than one row");
let key_columns = self.ensure_keys.iter().flat_map(|k| k.columns.iter());
let key_types = self.ensure_keys.iter().flat_map(|k| k.column_types.iter());
let key_names = self.ensure_keys.iter().map(|k| k.name.as_str());
let n_keys = self.ensure_keys.len();
let ensure_nested = nested(&target);
let docs = [
"Creates the entity, or returns the stored one when the `new` entity's unique keys (a caller-chosen id included) already name it.",
"",
"- **`Created`**: the insert hit no unique constraint. Events are persisted and the post-persist hook runs, as in `create_in_op`.",
"- **`Found`**: the insert hit a unique constraint, and the `new` entity's unique keys (a caller-chosen id included) match exactly one stored entity. That entity is returned unchanged: `new`'s events are dropped, nothing is written, the hook does not run. The stored entity's id may differ from `new`'s id (it was found through another key). Callers take the id from the returned entity, never from the `new` they passed in.",
"- **`Rejected(KeysSpanEntities)`**: the `new` entity's keys match more than one stored entity (one holds its code, another its email). No single stored entity is the one the caller described. This rejection does not abort the caller's transaction: the insert was `DO NOTHING` and the lookup is a `SELECT`.",
"- **`Transient(OptimisticConflict)`**: the insert conflicted and the lookup found nothing. Not reachable under READ COMMITTED (the lookup runs as a second statement and sees the committed row); kept for REPEATABLE READ / SERIALIZABLE, where the insert itself usually raises 40001 first (`Transient(SerializationFailure)`).",
"",
"`Found` compares nothing but keys: a `new` entity whose non-key fields differ from the stored entity still gets the stored entity. Ensure never updates.",
]
.map(|line| format!(" {line}"));
let on_span = match &self.ensure_rejection {
Some(rejection) => quote! {
let __keys: Vec<&'static str> = [#(#key_names),*]
.into_iter()
.zip(__matched)
.filter(|(_, matched)| *matched)
.map(|(key, _)| key)
.collect();
return Err(es_entity::errlanes::Fail::Rejected(
#rejection::KeysSpanEntities { ids: __found_ids, keys: __keys },
));
},
None => quote! {
return Err(es_entity::errlanes::Fatal::invariant(
"the unique keys match more than one row",
)
.with_context(#spanning_context)
.into());
},
};
let (matched_decl, track_matches) = if self.ensure_rejection.is_some() {
(
quote! { let mut __matched = [false; #n_keys]; },
quote! {
for (i, matched) in __matched.iter_mut().enumerate() {
*matched |= row.try_get::<bool, _>(i + 1)?;
}
},
)
} else {
(quote! {}, quote! {})
};
let push_found = quote! {
__found_ids.push(row.try_get(0)?);
#track_matches
};
let (scope_capture, scope_decl, collect_row, on_outside, find_scope) =
match (&self.scope_ty, &self.ensure_rejection) {
(Some(scope_ty), Some(rejection)) => {
let names: Vec<_> = scope_columns.iter().map(|c| c.name()).collect();
let tys: Vec<_> = scope_columns.iter().map(|c| c.ty()).collect();
let vars: Vec<_> = names
.iter()
.map(|n| quote::format_ident!("__scope_{}", n))
.collect();
let idxs = (0..scope_columns.len()).map(|i| i + n_keys + 1);
let key_names = self.ensure_keys.iter().map(|k| k.name.as_str());
(
quote! { #(let #vars: #tys = <#tys as Clone>::clone(#names);)* },
quote! { let mut __outside = [false; #n_keys]; },
quote! {
if #(row.try_get::<#tys, _>(#idxs)? == #vars)&&* {
#push_found
} else {
for (i, outside) in __outside.iter_mut().enumerate() {
*outside |= row.try_get::<bool, _>(i + 1)?;
}
}
},
quote! {
if __outside.iter().any(|outside| *outside) {
let __keys: Vec<&'static str> = [#(#key_names),*]
.into_iter()
.zip(__outside)
.filter(|(_, outside)| *outside)
.map(|(key, _)| key)
.collect();
return Err(es_entity::errlanes::Fail::Rejected(
#rejection::KeysHeldOutsideScope { keys: __keys },
));
}
},
quote! { #scope_ty::All, },
)
}
_ => (quote! {}, quote! {}, push_found, quote! {}, quote! {}),
};
quote! {
#(#[doc = #docs])*
pub async fn create_or_find_in_op<OP>(
&self,
op: &mut OP,
new_entity: <#entity as es_entity::EsEntity>::New
) -> Result<es_entity::Ensured<#entity>, #error_ty>
where
OP: es_entity::AtomicOperation + ?Sized
{
use es_entity::prelude::sqlx::{Arguments, Row};
#resolve_id
#assignments
#attempted_id
#ensure_constraint_values
let mut __query_args = sqlx::postgres::PgArguments::default();
#(#arg_adds)*
__query_args.add(op.maybe_now()).map_err(sqlx::Error::Encode)?;
let mut __key_args = sqlx::postgres::PgArguments::default();
#(__key_args.add(#key_columns as &#key_types).map_err(sqlx::Error::Encode)?;)*
#scope_capture
let mut events = Self::convert_new(new_entity, #builder);
#caller_id_check
let events_types = events.new_event_types();
let serialized_events = events.serialize_new_events();
#ctx_var
#(#event_arg_adds)*
let rows = sqlx::query_with(#ensure_query, __query_args)
.fetch_all(op.as_executor())
.await
#map_error;
if rows.is_empty() {
let __lookup = sqlx::query_with(#lookup_query, __key_args)
.fetch_all(op.as_executor())
.await?;
let mut __found_ids: Vec<#id_ty> = Vec::new();
#matched_decl
#scope_decl
for row in &__lookup {
#collect_row
}
#on_outside
match __found_ids.len() {
0 => {
return Err(es_entity::errlanes::Transient::new(
es_entity::errlanes::TransientKind::OptimisticConflict,
)
.with_context(#missing_context)
.into());
}
1 => {
let entity = self
.find_by_id_in_op(&mut *op, #find_scope __found_ids.remove(0))
.await?;
return Ok(es_entity::Ensured::Found(entity));
}
_ => {
#on_span
}
}
}
#forgettable_code
let recorded_at = rows
.first()
.ok_or(sqlx::Error::RowNotFound)
.and_then(|row| row.try_get("recorded_at"))?;
let n_events = events.mark_new_events_persisted_at(recorded_at);
let #maybe_mut_entity = Self::hydrate_entity(events)?;
#(#ensure_nested)*
#post_hydrate_check
#post_persist_check
Ok(es_entity::Ensured::Created(entity))
}
}
});
tokens.append_all(quote! {
#[inline(always)]
fn convert_new<Entity, Event, Snapshot>(
item: Entity,
builder: es_entity::EntityEventsBuilder<Event, #id_source>,
) -> es_entity::EntityEvents<Event, Snapshot>
where
Entity: es_entity::IntoEvents<Event, IdSource = #id_source>,
Event: es_entity::EsEvent,
Snapshot: es_entity::EsSnapshot,
{
item.into_events(builder).widen_snapshot()
}
#[inline(always)]
fn hydrate_entity<Entity>(events: es_entity::EntityEvents<<Entity as es_entity::EsEntity>::Event, <Entity as es_entity::EsEntity>::Snapshot>) -> Result<Entity, es_entity::EntityHydrationError>
where
Entity: es_entity::EsEntity,
{
es_entity::hydrate_checked(events)
}
#standalone
#instrument_attr
pub async fn create_in_op<OP>(
&self,
op: &mut OP,
new_entity: <#entity as es_entity::EsEntity>::New
) -> Result<#entity, #write_error_ty>
where
OP: es_entity::AtomicOperation + ?Sized
{
let __result: Result<#entity, #write_error_ty> = async {
#rejection_prelude
use es_entity::prelude::sqlx::{Arguments, Row};
#resolve_id
#assignments
#attempted_id
#constraint_values
#record_id
let mut __query_args = sqlx::postgres::PgArguments::default();
#(#arg_adds)*
__query_args.add(op.maybe_now()).map_err(sqlx::Error::Encode)?;
let mut events = Self::convert_new(new_entity, #builder);
#caller_id_check
let events_types = events.new_event_types();
let serialized_events = events.serialize_new_events();
#ctx_var
#(#event_arg_adds)*
let rows = sqlx::query_with(#create_query, __query_args)
.fetch_all(op.as_executor())
.await
.map_err(|e| #classify_create)
#rejection_tail;
#forgettable_code
let recorded_at = rows
.first()
.ok_or(sqlx::Error::RowNotFound)
.and_then(|row| row.try_get("recorded_at"))?;
let n_events = events.mark_new_events_persisted_at(recorded_at);
let #maybe_mut_entity = Self::hydrate_entity(events)?;
#(#create_nested)*
#post_hydrate_check
#post_persist_check
Ok(entity)
}.await;
#error_recording
__result
}
#ensure_fn
});
}
}
#[cfg(test)]
mod tests {
use super::*;
use proc_macro2::Span;
use syn::Ident;
#[test]
fn create_fn() {
let entity = Ident::new("Entity", Span::call_site());
let constraint_violation = syn::Ident::new("EntityConstraintViolation", Span::call_site());
let id = Ident::new("EntityId", Span::call_site());
let event = Ident::new("EntityEvent", Span::call_site());
let mut columns = Columns::default();
columns.set_id_column(&id);
let create_fn = CreateFn {
in_op_only: false,
table_name: "entities",
entity: &entity,
id: &id,
event: &event,
events_table_name: "entity_events",
event_ctx: false,
forgettable_table_name: None,
constraint_violation: constraint_violation.clone(),
columns: &columns,
nested_fn_names: Vec::new(),
post_hydrate_hook: false,
post_persist_hook: false,
id_source: IdSourceOpt::Caller,
can_reject: true,
write_error_ty: quote! { es_entity::RepoWriteError<EntityConstraintViolation> },
ensure_keys: Vec::new(),
ensure_rejection: None,
scope_ty: None,
#[cfg(feature = "instrument")]
repo_name_snake: "test_repo".to_string(),
};
let mut tokens = TokenStream::new();
create_fn.to_tokens(&mut tokens);
let expected = quote! {
#[inline(always)]
fn convert_new<Entity, Event, Snapshot>(
item: Entity,
builder: es_entity::EntityEventsBuilder<Event, es_entity::FromCaller>,
) -> es_entity::EntityEvents<Event, Snapshot>
where
Entity: es_entity::IntoEvents<Event, IdSource = es_entity::FromCaller>,
Event: es_entity::EsEvent,
Snapshot: es_entity::EsSnapshot,
{
item.into_events(builder).widen_snapshot()
}
#[inline(always)]
fn hydrate_entity<Entity>(events: es_entity::EntityEvents<<Entity as es_entity::EsEntity>::Event, <Entity as es_entity::EsEntity>::Snapshot>) -> Result<Entity, es_entity::EntityHydrationError>
where
Entity: es_entity::EsEntity,
{
es_entity::hydrate_checked(events)
}
pub async fn create(
&self,
new_entity: <Entity as es_entity::EsEntity>::New
) -> Result<Entity, es_entity::RepoWriteError<EntityConstraintViolation> > {
let mut op = self.begin_op().await?;
let res = self.create_in_op(&mut op, new_entity).await?;
op.commit().await?;
Ok(res)
}
pub async fn create_in_op<OP>(
&self,
op: &mut OP,
new_entity: <Entity as es_entity::EsEntity>::New
) -> Result<Entity, es_entity::RepoWriteError<EntityConstraintViolation> >
where
OP: es_entity::AtomicOperation + ?Sized
{
let __result: Result<Entity, es_entity::RepoWriteError<EntityConstraintViolation> > = async {
use es_entity::errlanes::ResultExt as _;
use es_entity::prelude::sqlx::{Arguments, Row};
let id = &new_entity.id;
let __attempted_id = (*id).clone();
let __constraint_values = EntityConstraintValues { id: Some((*id).clone()), };
let mut __query_args = sqlx::postgres::PgArguments::default();
__query_args.add(id as &EntityId).map_err(sqlx::Error::Encode)?;
__query_args.add(op.maybe_now()).map_err(sqlx::Error::Encode)?;
let mut events = Self::convert_new(new_entity, es_entity::EntityEventsBuilder::__from_caller());
if events.id() != &__attempted_id {
return Err(errlanes::Fatal::invariant("caller's id differs from the id the entity was built under").into());
}
let events_types = events.new_event_types();
let serialized_events = events.serialize_new_events();
__query_args.add(&events_types).map_err(sqlx::Error::Encode)?;
__query_args.add(&serialized_events).map_err(sqlx::Error::Encode)?;
let rows = sqlx::query_with(
"WITH new_row AS (INSERT INTO entities (id, created_at) VALUES ($1, COALESCE($2, NOW())) RETURNING id) INSERT INTO entity_events (id, recorded_at, sequence, event_type, event) SELECT new_row.id, COALESCE($2, NOW()), ROW_NUMBER() OVER (), unnested.event_type, unnested.event FROM new_row CROSS JOIN UNNEST($3::TEXT[], $4::JSONB[]) AS unnested(event_type, event) RETURNING recorded_at",
__query_args
)
.fetch_all(op.as_executor())
.await
.map_err(|e| Self::classify_create_write(e, move |_| Some(__attempted_id)))
.map_rejected(|r| r.with_attempted(__constraint_values))?;
let recorded_at = rows
.first()
.ok_or(sqlx::Error::RowNotFound)
.and_then(|row| row.try_get("recorded_at"))?;
let n_events = events.mark_new_events_persisted_at(recorded_at);
let entity = Self::hydrate_entity(events)?;
Ok(entity)
}.await;
__result
}
};
assert_eq!(tokens.to_string(), expected.to_string());
}
#[test]
fn create_fn_generated_id_has_no_route_for_a_caller_id() {
let entity = Ident::new("Entity", Span::call_site());
let id = Ident::new("EntityId", Span::call_site());
let event = Ident::new("EntityEvent", Span::call_site());
let mut columns = Columns::default();
columns.set_id_column(&id);
let create_fn = CreateFn {
in_op_only: false,
table_name: "entities",
entity: &entity,
id: &id,
event: &event,
events_table_name: "entity_events",
event_ctx: false,
forgettable_table_name: None,
constraint_violation: syn::Ident::new("EntityConstraintViolation", Span::call_site()),
columns: &columns,
nested_fn_names: Vec::new(),
post_hydrate_hook: false,
post_persist_hook: false,
id_source: IdSourceOpt::default(),
can_reject: false,
write_error_ty: quote! { es_entity::RepoFault },
ensure_keys: Vec::new(),
ensure_rejection: None,
scope_ty: None,
#[cfg(feature = "instrument")]
repo_name_snake: "test_repo".to_string(),
};
let mut tokens = TokenStream::new();
create_fn.to_tokens(&mut tokens);
let out = tokens.to_string();
assert!(
out.contains("let __id : EntityId = EntityId :: new ()"),
"{out}"
);
assert!(out.contains("let id = & __id"), "{out}");
assert!(
!out.contains("new_entity . id"),
"a generated id must never be read off the New* type: {out}"
);
assert!(
out.contains("EntityEventsBuilder :: __from_repo (__id . clone ())"),
"{out}"
);
assert!(
out.contains("IntoEvents < Event , IdSource = es_entity :: FromRepo >"),
"{out}"
);
assert!(
out.contains("Result < Entity , es_entity :: RepoFault >"),
"a repo that cannot reject returns RepoFault: {out}"
);
assert!(!out.contains("RepoWriteError"), "{out}");
assert!(!out.contains("map_rejected"), "{out}");
}
#[test]
fn create_fn_with_columns() {
let entity = Ident::new("Entity", Span::call_site());
let constraint_violation = syn::Ident::new("EntityConstraintViolation", Span::call_site());
let id = Ident::new("EntityId", Span::call_site());
let event = Ident::new("EntityEvent", Span::call_site());
use darling::FromMeta;
let input: syn::Meta =
syn::parse_quote!(columns(name(ty = "String", create(accessor = "name()"))));
let mut columns = Columns::from_meta(&input).expect("Failed to parse Fields");
columns.set_id_column(&id);
let create_fn = CreateFn {
in_op_only: false,
table_name: "entities",
entity: &entity,
id: &id,
event: &event,
events_table_name: "entity_events",
event_ctx: false,
forgettable_table_name: None,
constraint_violation: constraint_violation.clone(),
columns: &columns,
nested_fn_names: Vec::new(),
post_hydrate_hook: false,
post_persist_hook: false,
id_source: IdSourceOpt::Caller,
can_reject: true,
write_error_ty: quote! { es_entity::RepoWriteError<EntityConstraintViolation> },
ensure_keys: Vec::new(),
ensure_rejection: None,
scope_ty: None,
#[cfg(feature = "instrument")]
repo_name_snake: "test_repo".to_string(),
};
let mut tokens = TokenStream::new();
create_fn.to_tokens(&mut tokens);
let expected = quote! {
#[inline(always)]
fn convert_new<Entity, Event, Snapshot>(
item: Entity,
builder: es_entity::EntityEventsBuilder<Event, es_entity::FromCaller>,
) -> es_entity::EntityEvents<Event, Snapshot>
where
Entity: es_entity::IntoEvents<Event, IdSource = es_entity::FromCaller>,
Event: es_entity::EsEvent,
Snapshot: es_entity::EsSnapshot,
{
item.into_events(builder).widen_snapshot()
}
#[inline(always)]
fn hydrate_entity<Entity>(events: es_entity::EntityEvents<<Entity as es_entity::EsEntity>::Event, <Entity as es_entity::EsEntity>::Snapshot>) -> Result<Entity, es_entity::EntityHydrationError>
where
Entity: es_entity::EsEntity,
{
es_entity::hydrate_checked(events)
}
pub async fn create(
&self,
new_entity: <Entity as es_entity::EsEntity>::New
) -> Result<Entity, es_entity::RepoWriteError<EntityConstraintViolation> > {
let mut op = self.begin_op().await?;
let res = self.create_in_op(&mut op, new_entity).await?;
op.commit().await?;
Ok(res)
}
pub async fn create_in_op<OP>(
&self,
op: &mut OP,
new_entity: <Entity as es_entity::EsEntity>::New
) -> Result<Entity, es_entity::RepoWriteError<EntityConstraintViolation> >
where
OP: es_entity::AtomicOperation + ?Sized
{
let __result: Result<Entity, es_entity::RepoWriteError<EntityConstraintViolation> > = async {
use es_entity::errlanes::ResultExt as _;
use es_entity::prelude::sqlx::{Arguments, Row};
let id = &new_entity.id;
let name = &new_entity.name();
let __attempted_id = (*id).clone();
let __constraint_values = EntityConstraintValues { id: Some((*id).clone()), name: Some((*name).clone()), };
let mut __query_args = sqlx::postgres::PgArguments::default();
__query_args.add(id as &EntityId).map_err(sqlx::Error::Encode)?;
__query_args.add(name as &String).map_err(sqlx::Error::Encode)?;
__query_args.add(op.maybe_now()).map_err(sqlx::Error::Encode)?;
let mut events = Self::convert_new(new_entity, es_entity::EntityEventsBuilder::__from_caller());
if events.id() != &__attempted_id {
return Err(errlanes::Fatal::invariant("caller's id differs from the id the entity was built under").into());
}
let events_types = events.new_event_types();
let serialized_events = events.serialize_new_events();
__query_args.add(&events_types).map_err(sqlx::Error::Encode)?;
__query_args.add(&serialized_events).map_err(sqlx::Error::Encode)?;
let rows = sqlx::query_with(
"WITH new_row AS (INSERT INTO entities (id, name, created_at) VALUES ($1, $2, COALESCE($3, NOW())) RETURNING id) INSERT INTO entity_events (id, recorded_at, sequence, event_type, event) SELECT new_row.id, COALESCE($3, NOW()), ROW_NUMBER() OVER (), unnested.event_type, unnested.event FROM new_row CROSS JOIN UNNEST($4::TEXT[], $5::JSONB[]) AS unnested(event_type, event) RETURNING recorded_at",
__query_args
)
.fetch_all(op.as_executor())
.await
.map_err(|e| Self::classify_create_write(e, move |_| Some(__attempted_id)))
.map_rejected(|r| r.with_attempted(__constraint_values))?;
let recorded_at = rows
.first()
.ok_or(sqlx::Error::RowNotFound)
.and_then(|row| row.try_get("recorded_at"))?;
let n_events = events.mark_new_events_persisted_at(recorded_at);
let entity = Self::hydrate_entity(events)?;
Ok(entity)
}.await;
__result
}
};
assert_eq!(tokens.to_string(), expected.to_string());
}
}