fletch-orm 1.0.0

//! Dialect-generic CRUD operations.
//!
//! Core functions accept any [`Executor`] so the same CRUD runs against a pool
//! or an in-flight transaction. [`Pool`] and [`Transaction`] delegate to these
//! functions.

use sqlx::{Database, Executor, FromRow};

use crate::bind::DatabaseBindable;
use crate::built_query::BuiltQuery;
use crate::column::Column;
use crate::dialect::Dialect;
use crate::entity::Entity;
use crate::error::FletchError;
use crate::filter::{Filter, Op};
#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mysql"))]
use crate::pool::{Pool, Transaction};
use crate::query_builder::QueryBuilder;
use crate::value::Value;

/// Find an entity by its primary key.
pub async fn find_by_id<'e, E, DB, D, Ex>(
    dialect: &D,
    executor: Ex,
    id: E::Id,
) -> Result<E, FletchError>
where
    E: Entity,
    D: Dialect,
    DB: Database + DatabaseBindable,
    Ex: Executor<'e, Database = DB> + Send,
    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
{
    let query = QueryBuilder::select(dialect, E::table_name())
        .columns(&column_names::<E>())
        .filter(Filter::new(E::id_column(), Op::Eq, id))
        .build();

    query.fetch_one::<DB, E, _>(executor).await
}

/// Fetch all entities of this type.
pub async fn find_all<'e, E, DB, D, Ex>(dialect: &D, executor: Ex) -> Result<Vec<E>, FletchError>
where
    E: Entity,
    D: Dialect,
    DB: Database + DatabaseBindable,
    Ex: Executor<'e, Database = DB> + Send,
    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
{
    let query = QueryBuilder::select(dialect, E::table_name())
        .columns(&column_names::<E>())
        .build();

    query.fetch_all::<DB, E, _>(executor).await
}

/// Build an INSERT statement for `entity`, excluding the primary key column.
pub(crate) fn build_insert_query<E, D>(dialect: &D, entity: &E) -> BuiltQuery
where
    E: Entity,
    D: Dialect,
{
    let cols = E::columns();
    let id_col = E::id_column();
    let values = entity.values();

    let mut builder = QueryBuilder::insert(dialect, E::table_name());
    for (col, value) in cols.iter().zip(values.iter()) {
        if col.name() != id_col {
            builder = builder.set(col.name(), value.clone());
        }
    }

    if dialect.supports_returning() {
        builder.returning(&column_names::<E>()).build()
    } else {
        builder.build()
    }
}

/// Build an UPDATE statement for `entity`, excluding the primary key from SET.
pub(crate) fn build_update_query<E, D>(dialect: &D, entity: &E) -> BuiltQuery
where
    E: Entity,
    D: Dialect,
{
    let cols = E::columns();
    let id_col = E::id_column();
    let values = entity.values();
    let id = entity.id();

    let mut builder = QueryBuilder::update(dialect, E::table_name());
    for (col, value) in cols.iter().zip(values.iter()) {
        if col.name() != id_col {
            builder = builder.set(col.name(), value.clone());
        }
    }
    builder = builder.filter(Filter::new(id_col, Op::Eq, id));

    if dialect.supports_returning() {
        builder.returning(&column_names::<E>()).build()
    } else {
        builder.build()
    }
}

/// Build a SELECT for the row most recently inserted on engines without RETURNING.
#[cfg(test)]
pub(crate) fn build_select_last_inserted_query<E, D>(dialect: &D) -> BuiltQuery
where
    E: Entity,
    D: Dialect,
{
    let table = dialect.quote_identifier(E::table_name());
    let id_col = dialect.quote_identifier(E::id_column());
    let columns = E::columns()
        .iter()
        .map(|c| dialect.quote_identifier(c.name()))
        .collect::<Vec<_>>()
        .join(", ");

    BuiltQuery {
        sql: format!("SELECT {columns} FROM {table} WHERE {id_col} = LAST_INSERT_ID()"),
        values: Vec::new(),
    }
}

/// Insert a new entity and return it.
///
/// The id column is excluded from the INSERT statement so that the database
/// can assign an autoincrement primary key. On engines without `RETURNING`,
/// the inserted row is re-selected via `LAST_INSERT_ID()`.
pub async fn insert<'e, E, DB, D, Ex>(
    dialect: &D,
    executor: Ex,
    entity: &E,
) -> Result<E, FletchError>
where
    E: Entity,
    D: Dialect,
    DB: Database + DatabaseBindable,
    Ex: Executor<'e, Database = DB> + Send,
    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
{
    let query = build_insert_query::<E, D>(dialect, entity);

    if dialect.supports_returning() {
        return query.fetch_one::<DB, E, _>(executor).await;
    }

    query.execute::<DB, _>(executor).await?;
    Err(FletchError::Configuration(
        "insert on engines without RETURNING requires a re-select executor pass; use Pool::insert or Transaction::insert".into(),
    ))
}

/// Update an existing entity (matched by id) and return the updated row.
pub async fn update<'e, E, DB, D, Ex>(
    dialect: &D,
    executor: Ex,
    entity: &E,
) -> Result<E, FletchError>
where
    E: Entity,
    D: Dialect,
    DB: Database + DatabaseBindable,
    Ex: Executor<'e, Database = DB> + Send,
    E: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
{
    let query = build_update_query::<E, D>(dialect, entity);

    if dialect.supports_returning() {
        return query.fetch_one::<DB, E, _>(executor).await;
    }

    let id = entity.id();
    let affected = query.execute::<DB, _>(executor).await?;
    if affected == 0 {
        return Err(FletchError::NotFound);
    }

    let _ = id;
    Err(FletchError::Configuration(
        "update on engines without RETURNING requires a re-select executor pass; use Pool::update or Transaction::update".into(),
    ))
}

/// Delete an entity by its primary key.
pub async fn delete<'e, E, DB, D, Ex>(
    dialect: &D,
    executor: Ex,
    id: E::Id,
) -> Result<(), FletchError>
where
    E: Entity,
    D: Dialect,
    DB: Database + DatabaseBindable,
    Ex: Executor<'e, Database = DB> + Send,
{
    let query = QueryBuilder::delete(dialect, E::table_name())
        .filter(Filter::new(E::id_column(), Op::Eq, id))
        .build();

    let affected = query.execute::<DB, _>(executor).await?;
    if affected == 0 {
        return Err(FletchError::NotFound);
    }
    Ok(())
}

/// Execute a raw SQL query and return all matching rows.
pub async fn fetch_all<'e, T, DB, Ex>(
    sql: &str,
    params: &[Value],
    executor: Ex,
) -> Result<Vec<T>, FletchError>
where
    DB: Database + DatabaseBindable,
    Ex: Executor<'e, Database = DB> + Send,
    T: for<'r> FromRow<'r, <DB as Database>::Row> + Send + Unpin,
{
    BuiltQuery {
        sql: sql.to_owned(),
        values: params.to_vec(),
    }
    .fetch_all::<DB, T, _>(executor)
    .await
}

fn column_names<E: Entity>() -> Vec<&'static str> {
    E::columns().iter().map(Column::name).collect()
}

// ---------------------------------------------------------------------------
// Pool and Transaction passthrough
// ---------------------------------------------------------------------------

#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mysql"))]
macro_rules! impl_crud {
    ($db:ty, $dialect:expr, $fetch_one:path, $fetch_all:path, $execute:path) => {
        impl Pool<$db> {
            /// Find an entity by its primary key.
            pub async fn find_by_id<E>(&self, id: E::Id) -> Result<E, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = QueryBuilder::select($dialect, E::table_name())
                    .columns(&column_names::<E>())
                    .filter(Filter::new(E::id_column(), Op::Eq, id))
                    .build();
                ($fetch_one)(query, self.inner()).await
            }

            /// Fetch all entities of this type.
            pub async fn find_all<E>(&self) -> Result<Vec<E>, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = QueryBuilder::select($dialect, E::table_name())
                    .columns(&column_names::<E>())
                    .build();
                ($fetch_all)(query, self.inner()).await
            }

            /// Insert a new entity and return it.
            pub async fn insert<E>(&self, entity: &E) -> Result<E, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = build_insert_query::<E, _>($dialect, entity);
                if $dialect.supports_returning() {
                    ($fetch_one)(query, self.inner()).await
                } else {
                    let outcome = ($execute)(query, self.inner()).await?;
                    let insert_id = outcome.last_insert_id.ok_or_else(|| {
                        FletchError::Configuration("insert did not return a last insert id".into())
                    })?;
                    let insert_id = i64::try_from(insert_id).map_err(|_| {
                        FletchError::Configuration("insert last insert id exceeds i64 range".into())
                    })?;
                    let reselect = QueryBuilder::select($dialect, E::table_name())
                        .columns(&column_names::<E>())
                        .filter(Filter::new(E::id_column(), Op::Eq, Value::I64(insert_id)))
                        .build();
                    ($fetch_one)(reselect, self.inner()).await
                }
            }

            /// Update an existing entity and return the updated row.
            pub async fn update<E>(&self, entity: &E) -> Result<E, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = build_update_query::<E, _>($dialect, entity);
                if $dialect.supports_returning() {
                    ($fetch_one)(query, self.inner()).await
                } else {
                    let id = entity.id();
                    let affected = ($execute)(query, self.inner()).await?.rows_affected;
                    if affected == 0 {
                        return Err(FletchError::NotFound);
                    }
                    self.find_by_id::<E>(id).await
                }
            }

            /// Delete an entity by its primary key.
            pub async fn delete<E>(&self, id: E::Id) -> Result<(), FletchError>
            where
                E: Entity,
            {
                let query = QueryBuilder::delete($dialect, E::table_name())
                    .filter(Filter::new(E::id_column(), Op::Eq, id))
                    .build();
                let affected = ($execute)(query, self.inner()).await?.rows_affected;
                if affected == 0 {
                    return Err(FletchError::NotFound);
                }
                Ok(())
            }

            /// Execute a raw SQL query and return all matching rows.
            pub async fn fetch_all<T>(
                &self,
                sql: &str,
                params: &[Value],
            ) -> Result<Vec<T>, FletchError>
            where
                T: for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                ($fetch_all)(
                    BuiltQuery {
                        sql: sql.to_owned(),
                        values: params.to_vec(),
                    },
                    self.inner(),
                )
                .await
            }
        }

        impl<'a> Transaction<'a, $db> {
            /// Find an entity by its primary key.
            pub async fn find_by_id<E>(&mut self, id: E::Id) -> Result<E, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = QueryBuilder::select($dialect, E::table_name())
                    .columns(&column_names::<E>())
                    .filter(Filter::new(E::id_column(), Op::Eq, id))
                    .build();
                ($fetch_one)(query, &mut **self.inner_mut()).await
            }

            /// Fetch all entities of this type.
            pub async fn find_all<E>(&mut self) -> Result<Vec<E>, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = QueryBuilder::select($dialect, E::table_name())
                    .columns(&column_names::<E>())
                    .build();
                ($fetch_all)(query, &mut **self.inner_mut()).await
            }

            /// Insert a new entity and return it.
            pub async fn insert<E>(&mut self, entity: &E) -> Result<E, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = build_insert_query::<E, _>($dialect, entity);
                if $dialect.supports_returning() {
                    ($fetch_one)(query, &mut **self.inner_mut()).await
                } else {
                    let outcome = ($execute)(query, &mut **self.inner_mut()).await?;
                    let insert_id = outcome.last_insert_id.ok_or_else(|| {
                        FletchError::Configuration("insert did not return a last insert id".into())
                    })?;
                    let insert_id = i64::try_from(insert_id).map_err(|_| {
                        FletchError::Configuration("insert last insert id exceeds i64 range".into())
                    })?;
                    let reselect = QueryBuilder::select($dialect, E::table_name())
                        .columns(&column_names::<E>())
                        .filter(Filter::new(E::id_column(), Op::Eq, Value::I64(insert_id)))
                        .build();
                    ($fetch_one)(reselect, &mut **self.inner_mut()).await
                }
            }

            /// Update an existing entity and return the updated row.
            pub async fn update<E>(&mut self, entity: &E) -> Result<E, FletchError>
            where
                E: Entity + for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                let query = build_update_query::<E, _>($dialect, entity);
                if $dialect.supports_returning() {
                    ($fetch_one)(query, &mut **self.inner_mut()).await
                } else {
                    let id = entity.id();
                    let affected = ($execute)(query, &mut **self.inner_mut())
                        .await?
                        .rows_affected;
                    if affected == 0 {
                        return Err(FletchError::NotFound);
                    }
                    self.find_by_id::<E>(id).await
                }
            }

            /// Delete an entity by its primary key.
            pub async fn delete<E>(&mut self, id: E::Id) -> Result<(), FletchError>
            where
                E: Entity,
            {
                let query = QueryBuilder::delete($dialect, E::table_name())
                    .filter(Filter::new(E::id_column(), Op::Eq, id))
                    .build();
                let affected = ($execute)(query, &mut **self.inner_mut())
                    .await?
                    .rows_affected;
                if affected == 0 {
                    return Err(FletchError::NotFound);
                }
                Ok(())
            }

            /// Execute a raw SQL query and return all matching rows.
            pub async fn fetch_all<T>(
                &mut self,
                sql: &str,
                params: &[Value],
            ) -> Result<Vec<T>, FletchError>
            where
                T: for<'r> FromRow<'r, <$db as Database>::Row> + Send + Unpin,
            {
                ($fetch_all)(
                    BuiltQuery {
                        sql: sql.to_owned(),
                        values: params.to_vec(),
                    },
                    &mut **self.inner_mut(),
                )
                .await
            }
        }
    };
}

#[cfg(feature = "sqlite")]
impl_crud!(
    sqlx::Sqlite,
    &crate::dialect::SqliteDialect,
    crate::bind::sqlite_bind::fetch_one_row,
    crate::bind::sqlite_bind::fetch_all_rows,
    crate::bind::sqlite_bind::execute_query
);

#[cfg(feature = "postgres")]
impl_crud!(
    sqlx::Postgres,
    &crate::dialect::PostgresDialect,
    crate::bind::postgres_bind::fetch_one_row,
    crate::bind::postgres_bind::fetch_all_rows,
    crate::bind::postgres_bind::execute_query
);

#[cfg(feature = "mysql")]
impl_crud!(
    sqlx::MySql,
    &crate::dialect::MySqlDialect,
    crate::bind::mysql_bind::fetch_one_row,
    crate::bind::mysql_bind::fetch_all_rows,
    crate::bind::mysql_bind::execute_query
);

#[cfg(test)]
mod tests {
    use super::*;
    use crate::column::{Column, ColumnType};

    #[derive(Debug, Clone, PartialEq, sqlx::FromRow)]
    struct User {
        id: i64,
        name: String,
    }

    impl Entity for User {
        type Id = i64;

        fn table_name() -> &'static str {
            "users"
        }

        fn id_column() -> &'static str {
            "id"
        }

        fn columns() -> &'static [Column] {
            static COLS: [Column; 2] = [
                Column::new("id", ColumnType::Integer),
                Column::new("name", ColumnType::Text),
            ];
            &COLS
        }

        fn id(&self) -> Self::Id {
            self.id
        }

        fn values(&self) -> Vec<Value> {
            vec![Value::I64(self.id), Value::Text(self.name.clone())]
        }
    }

    #[cfg(feature = "sqlite")]
    async fn setup_pool() -> Pool<sqlx::Sqlite> {
        let pool = Pool::<sqlx::Sqlite>::connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::query("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
            .execute(pool.inner())
            .await
            .unwrap();
        pool
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn insert_and_find_by_id() {
        let pool = setup_pool().await;
        let user = User {
            id: 0,
            name: "Alice".into(),
        };
        let inserted = pool.insert(&user).await.unwrap();
        assert_ne!(inserted.id, 0);
        assert_eq!(inserted.name, "Alice");

        let found = pool.find_by_id::<User>(inserted.id).await.unwrap();
        assert_eq!(found, inserted);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn find_by_id_not_found() {
        let pool = setup_pool().await;
        let err = pool.find_by_id::<User>(999).await.unwrap_err();
        assert!(matches!(err, FletchError::NotFound));
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn find_all_empty() {
        let pool = setup_pool().await;
        let users = pool.find_all::<User>().await.unwrap();
        assert!(users.is_empty());
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn find_all_with_data() {
        let pool = setup_pool().await;
        pool.insert(&User {
            id: 0,
            name: "Alice".into(),
        })
        .await
        .unwrap();
        pool.insert(&User {
            id: 0,
            name: "Bob".into(),
        })
        .await
        .unwrap();

        let users = pool.find_all::<User>().await.unwrap();
        assert_eq!(users.len(), 2);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn update_entity() {
        let pool = setup_pool().await;
        let inserted = pool
            .insert(&User {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();

        let updated = pool
            .update(&User {
                id: inserted.id,
                name: "Alicia".into(),
            })
            .await
            .unwrap();
        assert_eq!(updated.name, "Alicia");

        let found = pool.find_by_id::<User>(inserted.id).await.unwrap();
        assert_eq!(found.name, "Alicia");
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn delete_entity() {
        let pool = setup_pool().await;
        let inserted = pool
            .insert(&User {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();

        pool.delete::<User>(inserted.id).await.unwrap();

        let err = pool.find_by_id::<User>(inserted.id).await.unwrap_err();
        assert!(matches!(err, FletchError::NotFound));
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn delete_not_found() {
        let pool = setup_pool().await;
        let err = pool.delete::<User>(999).await.unwrap_err();
        assert!(matches!(err, FletchError::NotFound));
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn fetch_all_raw_sql() {
        let pool = setup_pool().await;
        pool.insert(&User {
            id: 0,
            name: "Alice".into(),
        })
        .await
        .unwrap();
        pool.insert(&User {
            id: 0,
            name: "Bob".into(),
        })
        .await
        .unwrap();

        let users: Vec<User> = pool
            .fetch_all(
                "SELECT id, name FROM users WHERE name LIKE ?",
                &[Value::Text("%li%".into())],
            )
            .await
            .unwrap();
        assert_eq!(users.len(), 1);
        assert_eq!(users[0].name, "Alice");
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn insert_multiple_gets_unique_ids() {
        let pool = setup_pool().await;
        let a = pool
            .insert(&User {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();
        let b = pool
            .insert(&User {
                id: 0,
                name: "Bob".into(),
            })
            .await
            .unwrap();
        let c = pool
            .insert(&User {
                id: 0,
                name: "Carol".into(),
            })
            .await
            .unwrap();

        assert_ne!(a.id, 0);
        assert_ne!(b.id, 0);
        assert_ne!(c.id, 0);
        assert_ne!(a.id, b.id);
        assert_ne!(b.id, c.id);
        assert_ne!(a.id, c.id);

        let all = pool.find_all::<User>().await.unwrap();
        assert_eq!(all.len(), 3);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn crud_in_transaction() {
        let pool = setup_pool().await;

        let mut tx = pool.begin().await.unwrap();
        tx.insert(&User {
            id: 0,
            name: "Alice".into(),
        })
        .await
        .unwrap();
        tx.commit().await.unwrap();

        let users = pool.find_all::<User>().await.unwrap();
        assert_eq!(users.len(), 1);
        assert_eq!(users[0].name, "Alice");
    }

    #[cfg(feature = "postgres")]
    mod postgres {
        use super::*;

        static SCHEMA_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

        #[derive(Debug, Clone, PartialEq, sqlx::FromRow)]
        struct PgUser {
            id: i64,
            name: String,
        }

        impl Entity for PgUser {
            type Id = i64;

            fn table_name() -> &'static str {
                "fletch_users"
            }

            fn id_column() -> &'static str {
                "id"
            }

            fn columns() -> &'static [Column] {
                static COLS: [Column; 2] = [
                    Column::new("id", ColumnType::Integer),
                    Column::new("name", ColumnType::Text),
                ];
                &COLS
            }

            fn id(&self) -> Self::Id {
                self.id
            }

            fn values(&self) -> Vec<Value> {
                vec![Value::I64(self.id), Value::Text(self.name.clone())]
            }
        }

        fn database_url() -> String {
            std::env::var("DATABASE_URL")
                .unwrap_or_else(|_| "postgres://postgres:postgres@127.0.0.1:5432/aro_test".into())
        }

        async fn setup_pool() -> (Pool<sqlx::Postgres>, tokio::sync::MutexGuard<'static, ()>) {
            let lock = SCHEMA_LOCK.lock().await;
            let pool = Pool::<sqlx::Postgres>::connect(&database_url())
                .await
                .unwrap();
            sqlx::query("DROP TABLE IF EXISTS fletch_users")
                .execute(pool.inner())
                .await
                .unwrap();
            sqlx::query("CREATE TABLE fletch_users (id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL)")
                .execute(pool.inner())
                .await
                .unwrap();
            (pool, lock)
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running Postgres"]
        async fn insert_and_find_by_id() {
            let (pool, _lock) = setup_pool().await;
            let user = PgUser {
                id: 0,
                name: "Alice".into(),
            };
            let inserted = pool.insert(&user).await.unwrap();
            assert_ne!(inserted.id, 0);
            assert_eq!(inserted.name, "Alice");

            let found = pool.find_by_id::<PgUser>(inserted.id).await.unwrap();
            assert_eq!(found, inserted);
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running Postgres"]
        async fn find_all_with_data() {
            let (pool, _lock) = setup_pool().await;
            pool.insert(&PgUser {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();
            pool.insert(&PgUser {
                id: 0,
                name: "Bob".into(),
            })
            .await
            .unwrap();

            let users = pool.find_all::<PgUser>().await.unwrap();
            assert_eq!(users.len(), 2);
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running Postgres"]
        async fn update_entity() {
            let (pool, _lock) = setup_pool().await;
            let inserted = pool
                .insert(&PgUser {
                    id: 0,
                    name: "Alice".into(),
                })
                .await
                .unwrap();

            let updated = pool
                .update(&PgUser {
                    id: inserted.id,
                    name: "Alicia".into(),
                })
                .await
                .unwrap();
            assert_eq!(updated.name, "Alicia");

            let found = pool.find_by_id::<PgUser>(inserted.id).await.unwrap();
            assert_eq!(found.name, "Alicia");
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running Postgres"]
        async fn delete_entity() {
            let (pool, _lock) = setup_pool().await;
            let inserted = pool
                .insert(&PgUser {
                    id: 0,
                    name: "Alice".into(),
                })
                .await
                .unwrap();

            pool.delete::<PgUser>(inserted.id).await.unwrap();

            let err = pool.find_by_id::<PgUser>(inserted.id).await.unwrap_err();
            assert!(matches!(err, FletchError::NotFound));
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running Postgres"]
        async fn delete_not_found() {
            let (pool, _lock) = setup_pool().await;
            let err = pool.delete::<PgUser>(999).await.unwrap_err();
            assert!(matches!(err, FletchError::NotFound));
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running Postgres"]
        async fn fetch_all_raw_sql() {
            let (pool, _lock) = setup_pool().await;
            pool.insert(&PgUser {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();
            pool.insert(&PgUser {
                id: 0,
                name: "Bob".into(),
            })
            .await
            .unwrap();

            let users: Vec<PgUser> = pool
                .fetch_all(
                    "SELECT id, name FROM fletch_users WHERE name LIKE $1",
                    &[Value::Text("%li%".into())],
                )
                .await
                .unwrap();
            assert_eq!(users.len(), 1);
            assert_eq!(users[0].name, "Alice");
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running Postgres"]
        async fn crud_in_transaction() {
            let (pool, _lock) = setup_pool().await;

            let mut tx = pool.begin().await.unwrap();
            tx.insert(&PgUser {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();
            tx.commit().await.unwrap();

            let users = pool.find_all::<PgUser>().await.unwrap();
            assert_eq!(users.len(), 1);
            assert_eq!(users[0].name, "Alice");
        }
    }

    #[cfg(feature = "mysql")]
    mod mysql {
        use super::*;

        static SCHEMA_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

        #[derive(Debug, Clone, PartialEq, sqlx::FromRow)]
        struct MyUser {
            id: i64,
            name: String,
        }

        impl Entity for MyUser {
            type Id = i64;

            fn table_name() -> &'static str {
                "fletch_users"
            }

            fn id_column() -> &'static str {
                "id"
            }

            fn columns() -> &'static [Column] {
                static COLS: [Column; 2] = [
                    Column::new("id", ColumnType::Integer),
                    Column::new("name", ColumnType::Text),
                ];
                &COLS
            }

            fn id(&self) -> Self::Id {
                self.id
            }

            fn values(&self) -> Vec<Value> {
                vec![Value::I64(self.id), Value::Text(self.name.clone())]
            }
        }

        fn database_url() -> String {
            std::env::var("DATABASE_URL")
                .unwrap_or_else(|_| "mysql://root:mysql@127.0.0.1:3306/aro_test".into())
        }

        async fn setup_pool() -> (Pool<sqlx::MySql>, tokio::sync::MutexGuard<'static, ()>) {
            let lock = SCHEMA_LOCK.lock().await;
            let pool = Pool::<sqlx::MySql>::connect(&database_url()).await.unwrap();
            sqlx::query("DROP TABLE IF EXISTS fletch_users")
                .execute(pool.inner())
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE fletch_users (id BIGINT AUTO_INCREMENT PRIMARY KEY, name TEXT NOT NULL)",
            )
            .execute(pool.inner())
            .await
            .unwrap();
            (pool, lock)
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn mysql_insert_without_returning() {
            let (pool, _lock) = setup_pool().await;
            let inserted = pool
                .insert(&MyUser {
                    id: 0,
                    name: "Alice".into(),
                })
                .await
                .unwrap();
            assert!(inserted.id > 0);
            assert_eq!(inserted.name, "Alice");

            let updated = pool
                .update(&MyUser {
                    id: inserted.id,
                    name: "Alicia".into(),
                })
                .await
                .unwrap();
            assert_eq!(updated.name, "Alicia");

            pool.delete::<MyUser>(inserted.id).await.unwrap();
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn find_all_with_data() {
            let (pool, _lock) = setup_pool().await;
            pool.insert(&MyUser {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();
            pool.insert(&MyUser {
                id: 0,
                name: "Bob".into(),
            })
            .await
            .unwrap();

            let users = pool.find_all::<MyUser>().await.unwrap();
            assert_eq!(users.len(), 2);
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn find_by_id_not_found() {
            let (pool, _lock) = setup_pool().await;
            let err = pool.find_by_id::<MyUser>(999).await.unwrap_err();
            assert!(matches!(err, FletchError::NotFound));
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn delete_not_found() {
            let (pool, _lock) = setup_pool().await;
            let err = pool.delete::<MyUser>(999).await.unwrap_err();
            assert!(matches!(err, FletchError::NotFound));
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn fetch_all_raw_sql() {
            let (pool, _lock) = setup_pool().await;
            pool.insert(&MyUser {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();
            pool.insert(&MyUser {
                id: 0,
                name: "Bob".into(),
            })
            .await
            .unwrap();

            let users: Vec<MyUser> = pool
                .fetch_all(
                    "SELECT id, name FROM fletch_users WHERE name LIKE ?",
                    &[Value::Text("%li%".into())],
                )
                .await
                .unwrap();
            assert_eq!(users.len(), 1);
            assert_eq!(users[0].name, "Alice");
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn crud_in_transaction() {
            let (pool, _lock) = setup_pool().await;

            let mut tx = pool.begin().await.unwrap();
            tx.insert(&MyUser {
                id: 0,
                name: "Alice".into(),
            })
            .await
            .unwrap();
            tx.commit().await.unwrap();

            let users = pool.find_all::<MyUser>().await.unwrap();
            assert_eq!(users.len(), 1);
            assert_eq!(users[0].name, "Alice");
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn update_in_transaction() {
            let (pool, _lock) = setup_pool().await;
            let inserted = pool
                .insert(&MyUser {
                    id: 0,
                    name: "Alice".into(),
                })
                .await
                .unwrap();

            let mut tx = pool.begin().await.unwrap();
            let updated = tx
                .update(&MyUser {
                    id: inserted.id,
                    name: "Alicia".into(),
                })
                .await
                .unwrap();
            assert_eq!(updated.name, "Alicia");
            tx.commit().await.unwrap();

            let found = pool.find_by_id::<MyUser>(inserted.id).await.unwrap();
            assert_eq!(found.name, "Alicia");
        }

        #[tokio::test]
        #[ignore = "requires DATABASE_URL and running MySQL"]
        async fn insert_multiple_gets_unique_ids() {
            let (pool, _lock) = setup_pool().await;
            let a = pool
                .insert(&MyUser {
                    id: 0,
                    name: "Alice".into(),
                })
                .await
                .unwrap();
            let b = pool
                .insert(&MyUser {
                    id: 0,
                    name: "Bob".into(),
                })
                .await
                .unwrap();
            let c = pool
                .insert(&MyUser {
                    id: 0,
                    name: "Carol".into(),
                })
                .await
                .unwrap();

            assert_ne!(a.id, 0);
            assert_ne!(b.id, 0);
            assert_ne!(c.id, 0);
            assert_ne!(a.id, b.id);
            assert_ne!(b.id, c.id);
            assert_ne!(a.id, c.id);

            let all = pool.find_all::<MyUser>().await.unwrap();
            assert_eq!(all.len(), 3);
        }
    }

    #[test]
    fn build_insert_query_sqlite_uses_returning() {
        let dialect = crate::dialect::SqliteDialect;
        let user = User {
            id: 0,
            name: "Alice".into(),
        };
        let query = build_insert_query::<User, _>(&dialect, &user);
        assert!(query.sql.contains("RETURNING"));
        assert!(query.sql.contains("(\"name\")"));
        assert!(!query.sql.contains("(\"id\""));
    }

    #[test]
    fn build_insert_query_mysql_omits_returning() {
        let dialect = crate::dialect::MySqlDialect;
        let user = User {
            id: 0,
            name: "Alice".into(),
        };
        let query = build_insert_query::<User, _>(&dialect, &user);
        assert!(!query.sql.contains("RETURNING"));
        assert!(query.sql.contains("`name`"));
    }

    #[test]
    fn build_select_last_inserted_mysql() {
        let dialect = crate::dialect::MySqlDialect;
        let query = build_select_last_inserted_query::<User, _>(&dialect);
        assert!(query.sql.contains("LAST_INSERT_ID()"));
        assert!(query.values.is_empty());
    }
}