es-entity-macros 0.19.0

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,
    /// `RepositoryOptions::create_can_reject`: `false` narrows every
    /// generated signature from `RepoWriteError<CV>` to `RepoFault`.
    can_reject: bool,
    /// The generated signatures' error type, see
    /// `RepositoryOptions::create_error_ty`.
    write_error_ty: TokenStream,
    /// Whether `create_or_find_by_id_in_op` is generated, and whether it can
    /// reject, see `RepositoryOptions::create_or_find_enabled`.
    create_or_find: bool,
    create_or_find_can_reject: bool,
    #[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(),
            create_or_find: opts.create_or_find_enabled(),
            create_or_find_can_reject: opts.create_or_find_can_reject(),
            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! {
                    // A single create already knows its id from its own
                    // input, so a duplicate-id conflict always attributes
                    // to exactly this id -- captured before `new_entity` is
                    // consumed into `events` below.
                    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 = self.nested_fn_names.iter().map(|f| {
            quote! {
                self.#f(op, &mut [&mut entity]).await.map_err(es_entity::LiftNestedWrite::<#constraint_violation>::lift_nested)?;
            }
        });
        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();

        // Index insert and events insert combined into one statement. The
        // outer insert reads from the CTE, which provides the id without a
        // second bind. Postgres interleaves the CTE and main inserts with no
        // guaranteed ordering, so a duplicate id may surface as either
        // table's constraint — the classifier maps both to the same
        // `ConstraintViolation`.
        //
        // A brand-new entity has no persisted events, so sequences start at 1
        // and no offset parameter is needed. A new entity is never
        // snapshotted here, even when its repo enables `snapshot` — it gets
        // its first snapshot on its first `update`.
        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 = format!(
            "WITH new_row AS (INSERT INTO {} ({}, created_at) VALUES ({}, COALESCE(${}, NOW())) RETURNING id) {}",
            table_name,
            column_names.join(", "),
            placeholders,
            now_p,
            events_insert.sql(&source, now_p, now_p + 1),
        );

        // The event arrays are encoded after the index columns, whose borrows
        // on the new entity must end before it is consumed into events.
        let event_arg_adds = events_insert
            .arg_exprs(&source)
            .into_iter()
            .skip(1)
            .map(|expr| quote! { __query_args.add(#expr).map_err(sqlx::Error::Encode)?; });
        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)
                }
            }
        });

        // The conflict on the id's primary key becomes `Found`; every other
        // outcome of the create is the create's own. The create runs on a
        // savepoint so the conflict does not poison the caller's operation.
        let create_or_find = if self.create_or_find {
            let (ensure_error_ty, other_error) = if self.create_or_find_can_reject {
                (
                    quote! { es_entity::RepoWriteError<#constraint_violation> },
                    quote! { other },
                )
            } else {
                (
                    quote! { es_entity::RepoFault },
                    quote! { other.narrow_rejected() },
                )
            };
            quote! {
                /// Creates the entity, or — when an entity with this id
                /// already exists — returns the stored one. The conflict on the
                /// primary key is the only outcome that is not an error; every
                /// other constraint still rejects.
                pub async fn create_or_find_by_id_in_op<OP>(
                    &self,
                    op: &mut OP,
                    new_entity: <#entity as es_entity::EsEntity>::New
                ) -> Result<es_entity::Ensured<#entity>, #ensure_error_ty>
                where
                    OP: es_entity::AtomicOperation + ?Sized
                {
                    use es_entity::SavepointOperation as _;

                    let mut __savepoint = op.begin_savepoint().await?;
                    match self.create_in_op(&mut __savepoint, new_entity).await {
                        Ok(entity) => {
                            __savepoint.release().await?;
                            Ok(es_entity::Ensured::Created(entity))
                        }
                        Err(es_entity::Fail::Rejected(#constraint_violation::Pkey(conflict))) => {
                            __savepoint.rollback().await?;
                            let entity = self.find_by_id_in_op(&mut *op, conflict.attempted).await?;
                            Ok(es_entity::Ensured::Found(entity))
                        }
                        Err(other) => {
                            __savepoint.rollback().await?;
                            Err(#other_error)
                        }
                    }
                }
            }
        } else {
            quote! {}
        };

        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(#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)?;

                    #(#nested)*

                    #post_hydrate_check
                    #post_persist_check
                    Ok(entity)
                }.await;

                #error_recording
                __result
            }

            #create_or_find
        });
    }
}

#[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> },
            create_or_find: false,
            create_or_find_can_reject: false,
            #[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 },
            create_or_find: false,
            create_or_find_can_reject: false,
            #[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> },
            create_or_find: false,
            create_or_find_can_reject: false,
            #[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());
    }
}