pub use super::traits::{
    AccountOps, ApiKeyOps, InvitationOps, MemberOps, OrganizationOps, PasskeyOps, SessionOps,
    TwoFactorOps, UserOps, VerificationOps,
};

/// Database adapter trait for persistence.
///
/// Combines all entity-specific operation traits. Any type that implements
/// all sub-traits (`UserOps`, `SessionOps`, etc.) automatically implements
/// `DatabaseAdapter` via the blanket impl.
///
/// Use the sub-traits directly when you only need a subset of operations
/// (e.g., a plugin that only accesses users and sessions).
pub trait DatabaseAdapter:
    UserOps
    + SessionOps
    + AccountOps
    + VerificationOps
    + OrganizationOps
    + MemberOps
    + InvitationOps
    + TwoFactorOps
    + ApiKeyOps
    + PasskeyOps
{
}

impl<T> DatabaseAdapter for T where
    T: UserOps
        + SessionOps
        + AccountOps
        + VerificationOps
        + OrganizationOps
        + MemberOps
        + InvitationOps
        + TwoFactorOps
        + ApiKeyOps
        + PasskeyOps
{
}

#[cfg(feature = "sqlx-postgres")]
pub mod sqlx_adapter {
    use super::*;
    use async_trait::async_trait;
    use chrono::{DateTime, Utc};

    use crate::email::default_event_type_for_template_key;
    use crate::entity::{
        AuthAccount, AuthAccountMeta, AuthApiKey, AuthApiKeyMeta, AuthInvitation,
        AuthInvitationMeta, AuthMember, AuthMemberMeta, AuthOrganization, AuthOrganizationMeta,
        AuthPasskey, AuthPasskeyMeta, AuthSession, AuthSessionMeta, AuthTwoFactor,
        AuthTwoFactorMeta, AuthUser, AuthUserMeta, AuthVerification, AuthVerificationMeta,
    };
    use crate::error::{AuthError, AuthResult, DatabaseError};
    use crate::types::{
        Account, AdminAuditLog, ApiKey, AthenaClient, CreateAccount, CreateAdminAuditLog,
        CreateApiKey, CreateAthenaClient, CreateEmailSendFailure, CreateEmailTemplate,
        CreateInvitation, CreateMember, CreateOAuthClient, CreateOrganization, CreatePasskey,
        CreateSentEmail, CreateSession, CreateSessionIpProfileRecord, CreateTwoFactor, CreateUser,
        CreateVerification, EmailEventType, EmailSendFailure, EmailTemplate, Invitation,
        InvitationStatus, IpProfileRecord, ListAdminAuditLogsParams, ListEmailEventTypesParams,
        ListEmailSendFailuresParams, ListEmailTemplatesParams, ListSentEmailsParams,
        ListUsersParams, Member, OAuthClient, Organization, Passkey, SentEmail, Session,
        SessionIpProfileRecord, TwoFactor, UpdateAccount, UpdateApiKey, UpdateEmailSendFailure,
        UpdateEmailTemplate, UpdateOrganization, UpdateSentEmail, UpdateUser,
        UpsertIpProfileRecord, User, Verification,
    };
    use sqlx::PgPool;
    use sqlx::postgres::PgRow;
    use std::marker::PhantomData;
    use uuid::Uuid;

    const DEFAULT_AUTH_SCHEMA: &str = "athena";

    /// Quote a SQL identifier with double quotes for PostgreSQL.
    ///
    /// This prevents issues with reserved words (e.g. `user`, `key`, `order`)
    /// and ensures correct identifier handling regardless of the names returned
    /// by `Auth*Meta` traits.
    #[inline]
    fn qi(ident: &str) -> String {
        format!("\"{}\"", ident.replace('"', "\"\""))
    }

    /// Quote a table reference, adding the adapter schema when unqualified.
    #[inline]
    fn qtable(schema: &str, table: &str) -> String {
        let normalized_table = table.trim();
        if normalized_table.contains('.') {
            return normalized_table
                .split('.')
                .map(str::trim)
                .filter(|segment| !segment.is_empty())
                .map(qi)
                .collect::<Vec<String>>()
                .join(".");
        }

        format!("{}.{}", qi(schema), qi(normalized_table))
    }

    /// Convert a SQLx error into an [`AuthError`] with operation/table context.
    #[inline]
    fn db_error_with_context(operation: &str, table: &str, err: sqlx::Error) -> AuthError {
        let message = format!("operation={operation}, table={table}, error={err}");
        match err {
            sqlx::Error::Database(db_err) => {
                if db_err.is_unique_violation() {
                    AuthError::Database(DatabaseError::Constraint(message))
                } else {
                    AuthError::Database(DatabaseError::Query(message))
                }
            }
            sqlx::Error::PoolClosed | sqlx::Error::PoolTimedOut => {
                AuthError::Database(DatabaseError::Connection(message))
            }
            _ => AuthError::Database(DatabaseError::Query(message)),
        }
    }

    /// Blanket trait combining all bounds needed for SQLx-based entity types.
    ///
    /// Any type that implements `sqlx::FromRow` plus the standard marker traits
    /// automatically satisfies this bound. Custom entity types just need
    /// `#[derive(sqlx::FromRow)]` (or a manual `FromRow` impl) alongside
    /// their `Auth*` derive.
    pub trait SqlxEntity:
        for<'r> sqlx::FromRow<'r, PgRow> + Send + Sync + Unpin + Clone + 'static
    {
    }

    impl<T> SqlxEntity for T where
        T: for<'r> sqlx::FromRow<'r, PgRow> + Send + Sync + Unpin + Clone + 'static
    {
    }

    type SqlxAdapterEntities<U, S, A, O, M, I, V, TF, AK, PK> = (U, S, A, O, M, I, V, TF, AK, PK);

    /// PostgreSQL database adapter via SQLx.
    ///
    /// Generic over entity types — use default type parameters for the built-in
    /// types, or supply your own custom structs that implement `Auth*` + `sqlx::FromRow`.
    #[derive(Clone)]
    pub struct SqlxAdapter<
        U = User,
        S = Session,
        A = Account,
        O = Organization,
        M = Member,
        I = Invitation,
        V = Verification,
        TF = TwoFactor,
        AK = ApiKey,
        PK = Passkey,
    > {
        pool: PgPool,
        auth_schema: String,
        #[allow(clippy::type_complexity)]
        _phantom: PhantomData<SqlxAdapterEntities<U, S, A, O, M, I, V, TF, AK, PK>>,
    }

    /// Constructors for the default (built-in) entity types.
    impl SqlxAdapter {
        pub async fn new(database_url: &str) -> Result<Self, sqlx::Error> {
            let pool: sqlx::Pool<sqlx::Postgres> = PgPool::connect(database_url).await?;
            Ok(Self::from_pool(pool))
        }

        pub async fn with_config(
            database_url: &str,
            config: PoolConfig,
        ) -> Result<Self, sqlx::Error> {
            let pool: sqlx::Pool<sqlx::Postgres> = sqlx::postgres::PgPoolOptions::new()
                .max_connections(config.max_connections)
                .min_connections(config.min_connections)
                .acquire_timeout(config.acquire_timeout)
                .idle_timeout(config.idle_timeout)
                .max_lifetime(config.max_lifetime)
                .connect(database_url)
                .await?;
            Ok(Self::from_pool(pool))
        }
    }

    /// Methods available for all type parameterizations.
    impl<U, S, A, O, M, I, V, TF, AK, PK> SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK> {
        pub fn from_pool(pool: PgPool) -> Self {
            Self::from_pool_with_schema(pool, DEFAULT_AUTH_SCHEMA)
        }

        pub fn from_pool_with_schema(pool: PgPool, auth_schema: &str) -> Self {
            let normalized_schema = auth_schema.trim();
            Self {
                pool,
                auth_schema: if normalized_schema.is_empty() {
                    DEFAULT_AUTH_SCHEMA.to_string()
                } else {
                    normalized_schema.to_string()
                },
                _phantom: PhantomData,
            }
        }

        #[inline]
        fn table_ref(&self, table: &str) -> String {
            qtable(self.auth_schema.as_str(), table)
        }

        pub fn auth_schema(&self) -> &str {
            self.auth_schema.as_str()
        }

        pub async fn test_connection(&self) -> Result<(), sqlx::Error> {
            sqlx::query("SELECT 1").execute(&self.pool).await?;
            Ok(())
        }

        pub fn pool_stats(&self) -> PoolStats {
            PoolStats {
                size: self.pool.size(),
                idle: self.pool.num_idle(),
            }
        }

        pub async fn close(&self) {
            self.pool.close().await;
        }
    }

    #[derive(Debug, Clone)]
    pub struct PoolConfig {
        pub max_connections: u32,
        pub min_connections: u32,
        pub acquire_timeout: std::time::Duration,
        pub idle_timeout: Option<std::time::Duration>,
        pub max_lifetime: Option<std::time::Duration>,
    }

    impl Default for PoolConfig {
        fn default() -> Self {
            Self {
                max_connections: 10,
                min_connections: 0,
                acquire_timeout: std::time::Duration::from_secs(30),
                idle_timeout: Some(std::time::Duration::from_secs(600)),
                max_lifetime: Some(std::time::Duration::from_secs(1800)),
            }
        }
    }

    #[derive(Debug, Clone)]
    pub struct PoolStats {
        pub size: u32,
        pub idle: usize,
    }

    // -- UserOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> UserOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type User = U;

        async fn create_user(&self, create_user: CreateUser) -> AuthResult<U> {
            let id: String = create_user.id.unwrap_or_else(|| Uuid::new_v4().to_string());
            let now: DateTime<Utc> = Utc::now();

            let sql: String = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) RETURNING *",
                self.table_ref(U::table()),
                qi(U::col_id()),
                qi(U::col_email()),
                qi(U::col_name()),
                qi(U::col_image()),
                qi(U::col_email_verified()),
                qi(U::col_username()),
                qi(U::col_display_username()),
                qi(U::col_role()),
                qi(U::col_created_at()),
                qi(U::col_updated_at()),
                qi(U::col_metadata()),
            );
            let user: U = sqlx::query_as::<_, U>(&sql)
                .bind(&id)
                .bind(&create_user.email)
                .bind(&create_user.name)
                .bind(&create_user.image)
                .bind(create_user.email_verified.unwrap_or(false))
                .bind(&create_user.username)
                .bind(&create_user.display_username)
                .bind(&create_user.role)
                .bind(now)
                .bind(now)
                .bind(sqlx::types::Json(
                    create_user.metadata.unwrap_or(serde_json::json!({})),
                ))
                .fetch_one(&self.pool)
                .await?;

            Ok(user)
        }

        async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<U>> {
            let sql: String = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(U::table()),
                qi(U::col_id())
            );
            let user: Option<U> = sqlx::query_as::<_, U>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(user)
        }

        async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<U>> {
            let sql: String = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(U::table()),
                qi(U::col_email())
            );
            let user = sqlx::query_as::<_, U>(&sql)
                .bind(email)
                .fetch_optional(&self.pool)
                .await?;
            Ok(user)
        }

        async fn get_user_by_username(&self, username: &str) -> AuthResult<Option<U>> {
            let sql: String = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(U::table()),
                qi(U::col_username())
            );
            let user = sqlx::query_as::<_, U>(&sql)
                .bind(username)
                .fetch_optional(&self.pool)
                .await?;
            Ok(user)
        }

        async fn update_user(&self, id: &str, update: UpdateUser) -> AuthResult<U> {
            let mut query: sqlx::QueryBuilder<'_, sqlx::Postgres> =
                sqlx::QueryBuilder::new(format!(
                    "UPDATE {} SET {} = NOW()",
                    self.table_ref(U::table()),
                    qi(U::col_updated_at())
                ));
            let mut has_updates: bool = false;

            if let Some(email) = &update.email {
                query.push(format!(", {} = ", qi(U::col_email())));
                query.push_bind(email);
                has_updates = true;
            }
            if let Some(name) = &update.name {
                query.push(format!(", {} = ", qi(U::col_name())));
                query.push_bind(name);
                has_updates = true;
            }
            if let Some(image) = &update.image {
                query.push(format!(", {} = ", qi(U::col_image())));
                query.push_bind(image);
                has_updates = true;
            }
            if let Some(email_verified) = update.email_verified {
                query.push(format!(", {} = ", qi(U::col_email_verified())));
                query.push_bind(email_verified);
                has_updates = true;
            }
            if let Some(username) = &update.username {
                query.push(format!(", {} = ", qi(U::col_username())));
                query.push_bind(username);
                has_updates = true;
            }
            if let Some(display_username) = &update.display_username {
                query.push(format!(", {} = ", qi(U::col_display_username())));
                query.push_bind(display_username);
                has_updates = true;
            }
            if let Some(role) = &update.role {
                query.push(format!(", {} = ", qi(U::col_role())));
                query.push_bind(role);
                has_updates = true;
            }
            if let Some(banned) = update.banned {
                query.push(format!(", {} = ", qi(U::col_banned())));
                query.push_bind(banned);
                has_updates = true;
                // When explicitly unbanning, clear ban_reason and ban_expires
                if !banned {
                    query.push(format!(
                        ", {} = NULL, {} = NULL",
                        qi(U::col_ban_reason()),
                        qi(U::col_ban_expires())
                    ));
                }
            }
            // Only process ban_reason and ban_expires when we are NOT
            // explicitly unbanning.  When banned == Some(false) the block
            // above already emits `ban_reason = NULL, ban_expires = NULL`,
            // so applying these fields again would overwrite the NULLs.
            if update.banned != Some(false) {
                if let Some(ban_reason) = &update.ban_reason {
                    query.push(format!(", {} = ", qi(U::col_ban_reason())));
                    query.push_bind(ban_reason);
                    has_updates = true;
                }
                if let Some(ban_expires) = update.ban_expires {
                    query.push(format!(", {} = ", qi(U::col_ban_expires())));
                    query.push_bind(ban_expires);
                    has_updates = true;
                }
            }
            if let Some(last_sign_in_at) = update.last_sign_in_at {
                query.push(", last_sign_in_at = ");
                query.push_bind(last_sign_in_at);
                has_updates = true;
            }
            if let Some(two_factor_enabled) = update.two_factor_enabled {
                query.push(format!(", {} = ", qi(U::col_two_factor_enabled())));
                query.push_bind(two_factor_enabled);
                has_updates = true;
            }
            if let Some(metadata) = &update.metadata {
                query.push(format!(", {} = ", qi(U::col_metadata())));
                query.push_bind(sqlx::types::Json(metadata.clone()));
                has_updates = true;
            }

            if !has_updates {
                return self
                    .get_user_by_id(id)
                    .await?
                    .ok_or(AuthError::UserNotFound);
            }

            query.push(format!(" WHERE {} = ", qi(U::col_id())));
            query.push_bind(id);
            query.push(" RETURNING *");

            let user = query.build_query_as::<U>().fetch_one(&self.pool).await?;
            Ok(user)
        }

        async fn delete_user(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(U::table()),
                qi(U::col_id())
            );
            sqlx::query(&sql).bind(id).execute(&self.pool).await?;
            Ok(())
        }

        async fn list_users(&self, params: ListUsersParams) -> AuthResult<(Vec<U>, usize)> {
            let limit = params.limit.unwrap_or(100) as i64;
            let offset = params.offset.unwrap_or(0) as i64;

            // Build WHERE clause
            let mut conditions: Vec<String> = Vec::new();
            let mut bind_values: Vec<String> = Vec::new();

            if let Some(search_value) = &params.search_value {
                let field = params.search_field.as_deref().unwrap_or("email");
                let col = qi(match field {
                    "name" => U::col_name(),
                    _ => U::col_email(),
                });
                let op = params.search_operator.as_deref().unwrap_or("contains");
                let escaped = search_value.replace('%', "\\%").replace('_', "\\_");
                let pattern = match op {
                    "starts_with" => format!("{}%", escaped),
                    "ends_with" => format!("%{}", escaped),
                    _ => format!("%{}%", escaped),
                };
                let idx = bind_values.len() + 1;
                conditions.push(format!("{} ILIKE ${}", col, idx));
                bind_values.push(pattern);
            }

            if let Some(filter_value) = &params.filter_value {
                let field = params.filter_field.as_deref().unwrap_or("email");
                let col = qi(match field {
                    "name" => U::col_name(),
                    "role" => U::col_role(),
                    _ => U::col_email(),
                });
                let op = params.filter_operator.as_deref().unwrap_or("eq");
                let idx = bind_values.len() + 1;
                match op {
                    "contains" => {
                        let escaped = filter_value.replace('%', "\\%").replace('_', "\\_");
                        conditions.push(format!("{} ILIKE ${}", col, idx));
                        bind_values.push(format!("%{}%", escaped));
                    }
                    "ne" => {
                        conditions.push(format!("{} != ${}", col, idx));
                        bind_values.push(filter_value.clone());
                    }
                    _ => {
                        conditions.push(format!("{} = ${}", col, idx));
                        bind_values.push(filter_value.clone());
                    }
                }
            }

            let where_clause = if conditions.is_empty() {
                String::new()
            } else {
                format!(" WHERE {}", conditions.join(" AND "))
            };

            // Sort
            let order_clause = if let Some(sort_by) = &params.sort_by {
                let col = qi(match sort_by.as_str() {
                    "name" => U::col_name(),
                    "createdAt" | "created_at" => U::col_created_at(),
                    _ => U::col_email(),
                });
                let dir = if params.sort_direction.as_deref() == Some("desc") {
                    "DESC"
                } else {
                    "ASC"
                };
                format!(" ORDER BY {} {}", col, dir)
            } else {
                format!(" ORDER BY {} DESC", qi(U::col_created_at()))
            };

            // Count query
            let count_sql = format!(
                "SELECT COUNT(*) as count FROM {}{}",
                self.table_ref(U::table()),
                where_clause
            );
            let mut count_query = sqlx::query_scalar::<_, i64>(&count_sql);
            for v in &bind_values {
                count_query = count_query.bind(v);
            }
            let total = count_query.fetch_one(&self.pool).await? as usize;

            // Data query
            let limit_idx = bind_values.len() + 1;
            let offset_idx = bind_values.len() + 2;
            let data_sql = format!(
                "SELECT * FROM {}{}{} LIMIT ${} OFFSET ${}",
                self.table_ref(U::table()),
                where_clause,
                order_clause,
                limit_idx,
                offset_idx
            );
            let mut data_query = sqlx::query_as::<_, U>(&data_sql);
            for v in &bind_values {
                data_query = data_query.bind(v);
            }
            data_query = data_query.bind(limit).bind(offset);
            let users = data_query.fetch_all(&self.pool).await?;

            Ok((users, total))
        }
    }

    // -- SessionOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> SessionOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type Session = S;

        async fn create_session(&self, create_session: CreateSession) -> AuthResult<S> {
            let id = Uuid::new_v4().to_string();
            let token = format!("session_{}", Uuid::new_v4());
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}, {}, {}, {}, {}) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING *",
                self.table_ref(S::table()),
                qi(S::col_id()),
                qi(S::col_user_id()),
                qi(S::col_token()),
                qi(S::col_expires_at()),
                qi(S::col_created_at()),
                qi(S::col_ip_address()),
                qi(S::col_user_agent()),
                qi(S::col_impersonated_by()),
                qi(S::col_active_organization_id()),
                qi(S::col_active()),
            );
            let session = sqlx::query_as::<_, S>(&sql)
                .bind(&id)
                .bind(&create_session.user_id)
                .bind(&token)
                .bind(create_session.expires_at)
                .bind(now)
                .bind(&create_session.ip_address)
                .bind(&create_session.user_agent)
                .bind(&create_session.impersonated_by)
                .bind(&create_session.active_organization_id)
                .bind(true)
                .fetch_one(&self.pool)
                .await?;

            Ok(session)
        }

        async fn get_session(&self, token: &str) -> AuthResult<Option<S>> {
            let deactivate_sql = format!(
                "UPDATE {} SET {} = false, {} = NOW() \
                 WHERE {} = $1 AND {} = true AND {} <= NOW()",
                self.table_ref(S::table()),
                qi(S::col_active()),
                qi(S::col_updated_at()),
                qi(S::col_token()),
                qi(S::col_active()),
                qi(S::col_expires_at())
            );
            sqlx::query(&deactivate_sql)
                .bind(token)
                .execute(&self.pool)
                .await?;

            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND {} = true AND {} > NOW()",
                self.table_ref(S::table()),
                qi(S::col_token()),
                qi(S::col_active()),
                qi(S::col_expires_at())
            );
            let session = sqlx::query_as::<_, S>(&sql)
                .bind(token)
                .fetch_optional(&self.pool)
                .await?;
            Ok(session)
        }

        async fn get_user_sessions(&self, user_id: &str) -> AuthResult<Vec<S>> {
            let deactivate_sql = format!(
                "UPDATE {} SET {} = false, {} = NOW() \
                 WHERE {} = $1 AND {} = true AND {} <= NOW()",
                self.table_ref(S::table()),
                qi(S::col_active()),
                qi(S::col_updated_at()),
                qi(S::col_user_id()),
                qi(S::col_active()),
                qi(S::col_expires_at())
            );
            sqlx::query(&deactivate_sql)
                .bind(user_id)
                .execute(&self.pool)
                .await?;

            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND {} = true AND {} > NOW() ORDER BY {} DESC",
                self.table_ref(S::table()),
                qi(S::col_user_id()),
                qi(S::col_active()),
                qi(S::col_expires_at()),
                qi(S::col_created_at())
            );
            let sessions = sqlx::query_as::<_, S>(&sql)
                .bind(user_id)
                .fetch_all(&self.pool)
                .await?;
            Ok(sessions)
        }

        async fn update_session_expiry(
            &self,
            token: &str,
            expires_at: DateTime<Utc>,
        ) -> AuthResult<()> {
            let deactivate_sql = format!(
                "UPDATE {} SET {} = false, {} = NOW() \
                 WHERE {} = $1 AND {} = true AND {} <= NOW()",
                self.table_ref(S::table()),
                qi(S::col_active()),
                qi(S::col_updated_at()),
                qi(S::col_token()),
                qi(S::col_active()),
                qi(S::col_expires_at())
            );
            sqlx::query(&deactivate_sql)
                .bind(token)
                .execute(&self.pool)
                .await?;

            let sql = format!(
                "UPDATE {} SET {} = $1, {} = $2 WHERE {} = $3 AND {} = true AND {} > NOW()",
                self.table_ref(S::table()),
                qi(S::col_expires_at()),
                qi(S::col_updated_at()),
                qi(S::col_token()),
                qi(S::col_active()),
                qi(S::col_expires_at())
            );
            sqlx::query(&sql)
                .bind(expires_at)
                .bind(Utc::now())
                .bind(token)
                .execute(&self.pool)
                .await?;
            Ok(())
        }

        async fn delete_session(&self, token: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(S::table()),
                qi(S::col_token())
            );
            sqlx::query(&sql).bind(token).execute(&self.pool).await?;
            Ok(())
        }

        async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(S::table()),
                qi(S::col_user_id())
            );
            sqlx::query(&sql).bind(user_id).execute(&self.pool).await?;
            Ok(())
        }

        async fn delete_expired_sessions(&self) -> AuthResult<usize> {
            let sql = format!(
                "DELETE FROM {} WHERE {} < NOW() OR {} = false",
                self.table_ref(S::table()),
                qi(S::col_expires_at()),
                qi(S::col_active())
            );
            let result = sqlx::query(&sql).execute(&self.pool).await?;
            Ok(result.rows_affected() as usize)
        }

        async fn update_session_active_organization(
            &self,
            token: &str,
            organization_id: Option<&str>,
        ) -> AuthResult<S> {
            let deactivate_sql = format!(
                "UPDATE {} SET {} = false, {} = NOW() \
                 WHERE {} = $1 AND {} = true AND {} <= NOW()",
                self.table_ref(S::table()),
                qi(S::col_active()),
                qi(S::col_updated_at()),
                qi(S::col_token()),
                qi(S::col_active()),
                qi(S::col_expires_at())
            );
            sqlx::query(&deactivate_sql)
                .bind(token)
                .execute(&self.pool)
                .await?;

            let sql = format!(
                "UPDATE {} SET {} = $1, {} = NOW() WHERE {} = $2 AND {} = true AND {} > NOW() RETURNING *",
                self.table_ref(S::table()),
                qi(S::col_active_organization_id()),
                qi(S::col_updated_at()),
                qi(S::col_token()),
                qi(S::col_active()),
                qi(S::col_expires_at())
            );
            let session = sqlx::query_as::<_, S>(&sql)
                .bind(organization_id)
                .bind(token)
                .fetch_one(&self.pool)
                .await?;
            Ok(session)
        }

        async fn get_ip_profile_by_ip(
            &self,
            ip_address: &str,
        ) -> AuthResult<Option<IpProfileRecord>> {
            let sql = format!(
                "SELECT id, ip_address, provider, country_code, country_name, region_name, city, latitude, longitude, timezone, isp, organization, asn, is_vpn, is_proxy, is_tor, is_hosting, raw_payload, lookup_expires_at, first_seen_at, last_seen_at, created_at, updated_at \
                FROM {} WHERE ip_address = $1",
                self.table_ref("ip_profiles")
            );
            let row = sqlx::query_as::<_, IpProfileRecord>(&sql)
                .bind(ip_address)
                .fetch_optional(&self.pool)
                .await?;
            Ok(row)
        }

        async fn upsert_ip_profile(
            &self,
            input: UpsertIpProfileRecord,
        ) -> AuthResult<Option<IpProfileRecord>> {
            let now = Utc::now();
            let sql = format!(
                "INSERT INTO {} \
                (id, ip_address, provider, country_code, country_name, region_name, city, latitude, longitude, timezone, isp, organization, asn, is_vpn, is_proxy, is_tor, is_hosting, raw_payload, lookup_expires_at, first_seen_at, last_seen_at, created_at, updated_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23) \
                ON CONFLICT (ip_address) DO UPDATE SET \
                    provider = EXCLUDED.provider, \
                    country_code = EXCLUDED.country_code, \
                    country_name = EXCLUDED.country_name, \
                    region_name = EXCLUDED.region_name, \
                    city = EXCLUDED.city, \
                    latitude = EXCLUDED.latitude, \
                    longitude = EXCLUDED.longitude, \
                    timezone = EXCLUDED.timezone, \
                    isp = EXCLUDED.isp, \
                    organization = EXCLUDED.organization, \
                    asn = EXCLUDED.asn, \
                    is_vpn = EXCLUDED.is_vpn, \
                    is_proxy = EXCLUDED.is_proxy, \
                    is_tor = EXCLUDED.is_tor, \
                    is_hosting = EXCLUDED.is_hosting, \
                    raw_payload = EXCLUDED.raw_payload, \
                    lookup_expires_at = EXCLUDED.lookup_expires_at, \
                    last_seen_at = EXCLUDED.last_seen_at, \
                    updated_at = EXCLUDED.updated_at \
                RETURNING id, ip_address, provider, country_code, country_name, region_name, city, latitude, longitude, timezone, isp, organization, asn, is_vpn, is_proxy, is_tor, is_hosting, raw_payload, lookup_expires_at, first_seen_at, last_seen_at, created_at, updated_at",
                self.table_ref("ip_profiles")
            );
            let row = sqlx::query_as::<_, IpProfileRecord>(&sql)
                .bind(Uuid::new_v4().to_string())
                .bind(input.ip_address)
                .bind(input.provider)
                .bind(input.country_code)
                .bind(input.country_name)
                .bind(input.region_name)
                .bind(input.city)
                .bind(input.latitude)
                .bind(input.longitude)
                .bind(input.timezone)
                .bind(input.isp)
                .bind(input.organization)
                .bind(input.asn)
                .bind(input.is_vpn)
                .bind(input.is_proxy)
                .bind(input.is_tor)
                .bind(input.is_hosting)
                .bind(sqlx::types::Json(input.raw_payload))
                .bind(input.lookup_expires_at)
                .bind(now)
                .bind(now)
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;
            Ok(Some(row))
        }

        async fn create_session_ip_profile(
            &self,
            input: CreateSessionIpProfileRecord,
        ) -> AuthResult<Option<SessionIpProfileRecord>> {
            let sql = format!(
                "INSERT INTO {} \
                (id, session_id, session_token, user_id, ip_address, ip_profile_id, country_code, country_name, is_vpn, is_proxy, is_tor, is_hosting, country_changed, previous_country_code, provider, created_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) \
                ON CONFLICT (session_id) DO UPDATE SET \
                    session_token = EXCLUDED.session_token, \
                    user_id = EXCLUDED.user_id, \
                    ip_address = EXCLUDED.ip_address, \
                    ip_profile_id = EXCLUDED.ip_profile_id, \
                    country_code = EXCLUDED.country_code, \
                    country_name = EXCLUDED.country_name, \
                    is_vpn = EXCLUDED.is_vpn, \
                    is_proxy = EXCLUDED.is_proxy, \
                    is_tor = EXCLUDED.is_tor, \
                    is_hosting = EXCLUDED.is_hosting, \
                    country_changed = EXCLUDED.country_changed, \
                    previous_country_code = EXCLUDED.previous_country_code, \
                    provider = EXCLUDED.provider \
                RETURNING id, session_id, session_token, user_id, ip_address, ip_profile_id, country_code, country_name, is_vpn, is_proxy, is_tor, is_hosting, country_changed, previous_country_code, provider, created_at",
                self.table_ref("session_ip_profiles")
            );
            let row = sqlx::query_as::<_, SessionIpProfileRecord>(&sql)
                .bind(Uuid::new_v4().to_string())
                .bind(input.session_id)
                .bind(input.session_token)
                .bind(input.user_id)
                .bind(input.ip_address)
                .bind(input.ip_profile_id)
                .bind(input.country_code)
                .bind(input.country_name)
                .bind(input.is_vpn)
                .bind(input.is_proxy)
                .bind(input.is_tor)
                .bind(input.is_hosting)
                .bind(input.country_changed)
                .bind(input.previous_country_code)
                .bind(input.provider)
                .bind(Utc::now())
                .fetch_one(&self.pool)
                .await?;
            Ok(Some(row))
        }

        async fn get_latest_session_country_code(
            &self,
            user_id: &str,
            exclude_session_id: Option<&str>,
        ) -> AuthResult<Option<String>> {
            let session_ip_profiles_table = self.table_ref("session_ip_profiles");
            let maybe_row = if let Some(exclude_session_id) = exclude_session_id {
                sqlx::query_scalar::<_, Option<String>>(&format!(
                    "SELECT country_code FROM {} \
                        WHERE user_id = $1 AND session_id <> $2 AND country_code IS NOT NULL \
                        ORDER BY created_at DESC LIMIT 1",
                    session_ip_profiles_table
                ))
                .bind(user_id)
                .bind(exclude_session_id)
                .fetch_optional(&self.pool)
                .await?
            } else {
                sqlx::query_scalar::<_, Option<String>>(&format!(
                    "SELECT country_code FROM {} \
                        WHERE user_id = $1 AND country_code IS NOT NULL \
                        ORDER BY created_at DESC LIMIT 1",
                    session_ip_profiles_table
                ))
                .bind(user_id)
                .fetch_optional(&self.pool)
                .await?
            };

            Ok(maybe_row.flatten())
        }

        async fn get_session_ip_profile_by_session_id(
            &self,
            session_id: &str,
        ) -> AuthResult<Option<SessionIpProfileRecord>> {
            let sql = format!(
                "SELECT id, session_id, session_token, user_id, ip_address, ip_profile_id, country_code, country_name, is_vpn, is_proxy, is_tor, is_hosting, country_changed, previous_country_code, provider, created_at \
                FROM {} WHERE session_id = $1",
                self.table_ref("session_ip_profiles")
            );
            let row = sqlx::query_as::<_, SessionIpProfileRecord>(&sql)
                .bind(session_id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(row)
        }
    }

    // -- AccountOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> AccountOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type Account = A;

        async fn create_account(&self, create_account: CreateAccount) -> AuthResult<A> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}) \
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) RETURNING *",
                self.table_ref(A::table()),
                qi(A::col_id()),
                qi(A::col_account_id()),
                qi(A::col_provider_id()),
                qi(A::col_user_id()),
                qi(A::col_access_token()),
                qi(A::col_refresh_token()),
                qi(A::col_id_token()),
                qi(A::col_access_token_expires_at()),
                qi(A::col_refresh_token_expires_at()),
                qi(A::col_scope()),
                qi(A::col_password()),
                qi(A::col_created_at()),
                qi(A::col_updated_at()),
            );
            let account = sqlx::query_as::<_, A>(&sql)
                .bind(&id)
                .bind(&create_account.account_id)
                .bind(&create_account.provider_id)
                .bind(&create_account.user_id)
                .bind(&create_account.access_token)
                .bind(&create_account.refresh_token)
                .bind(&create_account.id_token)
                .bind(create_account.access_token_expires_at)
                .bind(create_account.refresh_token_expires_at)
                .bind(&create_account.scope)
                .bind(&create_account.password)
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await
                .map_err(|e| db_error_with_context("insert", A::table(), e))?;

            Ok(account)
        }

        async fn get_account(
            &self,
            provider: &str,
            provider_account_id: &str,
        ) -> AuthResult<Option<A>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND {} = $2",
                self.table_ref(A::table()),
                qi(A::col_provider_id()),
                qi(A::col_account_id())
            );
            let account = sqlx::query_as::<_, A>(&sql)
                .bind(provider)
                .bind(provider_account_id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(account)
        }

        async fn get_user_accounts(&self, user_id: &str) -> AuthResult<Vec<A>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 ORDER BY {} DESC",
                self.table_ref(A::table()),
                qi(A::col_user_id()),
                qi(A::col_created_at())
            );
            let accounts = sqlx::query_as::<_, A>(&sql)
                .bind(user_id)
                .fetch_all(&self.pool)
                .await?;
            Ok(accounts)
        }

        async fn update_account(&self, id: &str, update: UpdateAccount) -> AuthResult<A> {
            let mut query = sqlx::QueryBuilder::new(format!(
                "UPDATE {} SET {} = NOW()",
                self.table_ref(A::table()),
                qi(A::col_updated_at())
            ));

            if let Some(access_token) = &update.access_token {
                query.push(format!(", {} = ", qi(A::col_access_token())));
                query.push_bind(access_token);
            }
            if let Some(refresh_token) = &update.refresh_token {
                query.push(format!(", {} = ", qi(A::col_refresh_token())));
                query.push_bind(refresh_token);
            }
            if let Some(id_token) = &update.id_token {
                query.push(format!(", {} = ", qi(A::col_id_token())));
                query.push_bind(id_token);
            }
            if let Some(access_token_expires_at) = &update.access_token_expires_at {
                query.push(format!(", {} = ", qi(A::col_access_token_expires_at())));
                query.push_bind(access_token_expires_at);
            }
            if let Some(refresh_token_expires_at) = &update.refresh_token_expires_at {
                query.push(format!(", {} = ", qi(A::col_refresh_token_expires_at())));
                query.push_bind(refresh_token_expires_at);
            }
            if let Some(scope) = &update.scope {
                query.push(format!(", {} = ", qi(A::col_scope())));
                query.push_bind(scope);
            }
            if let Some(password) = &update.password {
                query.push(format!(", {} = ", qi(A::col_password())));
                query.push_bind(password);
            }

            query.push(format!(" WHERE {} = ", qi(A::col_id())));
            query.push_bind(id);
            query.push(" RETURNING *");

            let account = query.build_query_as::<A>().fetch_one(&self.pool).await?;
            Ok(account)
        }

        async fn delete_account(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(A::table()),
                qi(A::col_id())
            );
            sqlx::query(&sql)
                .bind(id)
                .execute(&self.pool)
                .await
                .map_err(|e| db_error_with_context("delete", A::table(), e))?;
            Ok(())
        }
    }

    // -- VerificationOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> VerificationOps
        for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type Verification = V;

        async fn create_verification(
            &self,
            create_verification: CreateVerification,
        ) -> AuthResult<V> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *",
                self.table_ref(V::table()),
                qi(V::col_id()),
                qi(V::col_identifier()),
                qi(V::col_value()),
                qi(V::col_expires_at()),
                qi(V::col_created_at()),
                qi(V::col_updated_at()),
            );
            let verification = sqlx::query_as::<_, V>(&sql)
                .bind(&id)
                .bind(&create_verification.identifier)
                .bind(&create_verification.value)
                .bind(create_verification.expires_at)
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await
                .map_err(|e| db_error_with_context("insert", V::table(), e))?;

            Ok(verification)
        }

        async fn get_verification(&self, identifier: &str, value: &str) -> AuthResult<Option<V>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND {} = $2 AND {} > NOW()",
                self.table_ref(V::table()),
                qi(V::col_identifier()),
                qi(V::col_value()),
                qi(V::col_expires_at())
            );
            let verification = sqlx::query_as::<_, V>(&sql)
                .bind(identifier)
                .bind(value)
                .fetch_optional(&self.pool)
                .await
                .map_err(|e| db_error_with_context("select", V::table(), e))?;
            Ok(verification)
        }

        async fn get_verification_by_value(&self, value: &str) -> AuthResult<Option<V>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND {} > NOW()",
                self.table_ref(V::table()),
                qi(V::col_value()),
                qi(V::col_expires_at())
            );
            let verification = sqlx::query_as::<_, V>(&sql)
                .bind(value)
                .fetch_optional(&self.pool)
                .await
                .map_err(|e| db_error_with_context("select", V::table(), e))?;
            Ok(verification)
        }

        async fn get_verification_by_identifier(&self, identifier: &str) -> AuthResult<Option<V>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND {} > NOW()",
                self.table_ref(V::table()),
                qi(V::col_identifier()),
                qi(V::col_expires_at())
            );
            let verification = sqlx::query_as::<_, V>(&sql)
                .bind(identifier)
                .fetch_optional(&self.pool)
                .await
                .map_err(|e| db_error_with_context("select", V::table(), e))?;
            Ok(verification)
        }

        async fn consume_verification(
            &self,
            identifier: &str,
            value: &str,
        ) -> AuthResult<Option<V>> {
            let sql = format!(
                "DELETE FROM {tbl} WHERE {id} IN (\
                    SELECT {id} FROM {tbl} \
                    WHERE {ident} = $1 AND {val} = $2 AND {exp} > NOW() \
                    ORDER BY {ca} DESC \
                    LIMIT 1\
                ) RETURNING *",
                tbl = self.table_ref(V::table()),
                id = qi(V::col_id()),
                ident = qi(V::col_identifier()),
                val = qi(V::col_value()),
                exp = qi(V::col_expires_at()),
                ca = qi(V::col_created_at()),
            );
            let verification = sqlx::query_as::<_, V>(&sql)
                .bind(identifier)
                .bind(value)
                .fetch_optional(&self.pool)
                .await
                .map_err(|e| db_error_with_context("delete-returning", V::table(), e))?;
            Ok(verification)
        }

        async fn delete_verification(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(V::table()),
                qi(V::col_id())
            );
            sqlx::query(&sql)
                .bind(id)
                .execute(&self.pool)
                .await
                .map_err(|e| db_error_with_context("delete", V::table(), e))?;
            Ok(())
        }

        async fn create_email_send_failure(
            &self,
            failure: CreateEmailSendFailure,
        ) -> AuthResult<()> {
            self.create_email_send_failure_record(failure).await?;
            Ok(())
        }

        async fn create_email_send_failure_record(
            &self,
            failure: CreateEmailSendFailure,
        ) -> AuthResult<EmailSendFailure> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();
            let metadata = failure.metadata.unwrap_or_else(|| serde_json::json!({}));
            let sql = format!(
                "INSERT INTO {} (id, user_id, recipient_email, flow, provider, error_message, metadata, resolved, resolution_note, created_at, updated_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \
                RETURNING id, user_id, recipient_email, flow, provider, error_message, metadata, resolved, resolution_note, created_at, updated_at",
                self.table_ref("email_send_failures")
            );

            let row = sqlx::query_as::<_, EmailSendFailure>(&sql)
                .bind(id)
                .bind(failure.user_id)
                .bind(failure.recipient_email)
                .bind(failure.flow)
                .bind(failure.provider)
                .bind(failure.error_message)
                .bind(sqlx::types::Json(metadata))
                .bind(false)
                .bind(Option::<String>::None)
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;
            Ok(row)
        }

        async fn create_sent_email(&self, sent: CreateSentEmail) -> AuthResult<()> {
            self.create_sent_email_record(sent).await?;
            Ok(())
        }

        async fn create_sent_email_record(&self, sent: CreateSentEmail) -> AuthResult<SentEmail> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();
            let metadata = sent.metadata.unwrap_or_else(|| serde_json::json!({}));
            let sql = format!(
                "INSERT INTO {} (id, recipient_email, subject, from_address, from_name, text_body, html_body, provider, flow, metadata, created_at, updated_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \
                RETURNING id, recipient_email, subject, from_address, from_name, text_body, html_body, provider, flow, metadata, created_at, updated_at",
                self.table_ref("emails")
            );

            let row = sqlx::query_as::<_, SentEmail>(&sql)
                .bind(id)
                .bind(sent.recipient_email)
                .bind(sent.subject)
                .bind(sent.from_address)
                .bind(sent.from_name)
                .bind(sent.text_body)
                .bind(sent.html_body)
                .bind(sent.provider)
                .bind(sent.flow)
                .bind(sqlx::types::Json(metadata))
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;
            Ok(row)
        }

        async fn list_sent_emails(
            &self,
            limit: usize,
            offset: usize,
        ) -> AuthResult<Vec<SentEmail>> {
            self.list_sent_emails_with_params(ListSentEmailsParams {
                limit: Some(limit),
                offset: Some(offset),
                ..Default::default()
            })
            .await
        }

        async fn list_sent_emails_for_recipient(
            &self,
            recipient_email: &str,
            limit: usize,
            offset: usize,
        ) -> AuthResult<Vec<SentEmail>> {
            self.list_sent_emails_with_params(ListSentEmailsParams {
                limit: Some(limit),
                offset: Some(offset),
                recipient_email: Some(recipient_email.to_string()),
                ..Default::default()
            })
            .await
        }

        async fn list_sent_emails_with_params(
            &self,
            params: ListSentEmailsParams,
        ) -> AuthResult<Vec<SentEmail>> {
            let limit = params.limit.unwrap_or(100).min(1000);
            let offset = params.offset.unwrap_or(0);
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT id, recipient_email, subject, from_address, from_name, text_body, html_body, provider, flow, metadata, created_at, updated_at \
                FROM {}",
                self.table_ref("emails")
            ));
            let mut has_where = false;

            if let Some(recipient_email) = params
                .recipient_email
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(recipient_email) = LOWER(");
                query_builder.push_bind(recipient_email);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(provider) = params
                .provider
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(provider) = LOWER(");
                query_builder.push_bind(provider);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(flow) = params
                .flow
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(COALESCE(flow, '')) = LOWER(");
                query_builder.push_bind(flow);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(subject_contains) = params
                .subject_contains
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(subject) LIKE LOWER(");
                query_builder.push_bind(format!("%{subject_contains}%"));
                query_builder.push(")");
                has_where = true;
            }

            if let Some(created_after) = params.created_after {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at >= ");
                query_builder.push_bind(created_after);
                has_where = true;
            }

            if let Some(created_before) = params.created_before {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at <= ");
                query_builder.push_bind(created_before);
            }

            query_builder.push(" ORDER BY created_at DESC LIMIT ");
            query_builder.push_bind(limit as i64);
            query_builder.push(" OFFSET ");
            query_builder.push_bind(offset as i64);

            let rows = query_builder
                .build_query_as::<SentEmail>()
                .fetch_all(&self.pool)
                .await?;

            Ok(rows)
        }

        async fn count_sent_emails(&self, params: ListSentEmailsParams) -> AuthResult<usize> {
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT COUNT(*)::BIGINT FROM {}",
                self.table_ref("emails")
            ));
            let mut has_where = false;

            if let Some(recipient_email) = params
                .recipient_email
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(recipient_email) = LOWER(");
                query_builder.push_bind(recipient_email);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(provider) = params
                .provider
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(provider) = LOWER(");
                query_builder.push_bind(provider);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(flow) = params
                .flow
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(COALESCE(flow, '')) = LOWER(");
                query_builder.push_bind(flow);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(subject_contains) = params
                .subject_contains
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(subject) LIKE LOWER(");
                query_builder.push_bind(format!("%{subject_contains}%"));
                query_builder.push(")");
                has_where = true;
            }

            if let Some(created_after) = params.created_after {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at >= ");
                query_builder.push_bind(created_after);
                has_where = true;
            }

            if let Some(created_before) = params.created_before {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at <= ");
                query_builder.push_bind(created_before);
            }

            let (total,): (i64,) = query_builder.build_query_as().fetch_one(&self.pool).await?;
            Ok(total as usize)
        }

        async fn get_sent_email_by_id(&self, id: &str) -> AuthResult<Option<SentEmail>> {
            let sql = format!(
                "SELECT id, recipient_email, subject, from_address, from_name, text_body, html_body, provider, flow, metadata, created_at, updated_at \
                FROM {} WHERE id = $1",
                self.table_ref("emails")
            );
            let row = sqlx::query_as::<_, SentEmail>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(row)
        }

        async fn update_sent_email(
            &self,
            id: &str,
            update: UpdateSentEmail,
        ) -> AuthResult<SentEmail> {
            let current = self
                .get_sent_email_by_id(id)
                .await?
                .ok_or_else(|| AuthError::not_found("Sent email not found"))?;

            let recipient_email = update.recipient_email.unwrap_or(current.recipient_email);
            let subject = update.subject.unwrap_or(current.subject);
            let from_address = update.from_address.unwrap_or(current.from_address);
            let from_name = match update.from_name {
                Some(value) => value,
                None => current.from_name,
            };
            let text_body = match update.text_body {
                Some(value) => value,
                None => current.text_body,
            };
            let html_body = match update.html_body {
                Some(value) => value,
                None => current.html_body,
            };
            let provider = update.provider.unwrap_or(current.provider);
            let flow = match update.flow {
                Some(value) => value,
                None => current.flow,
            };
            let metadata = update.metadata.unwrap_or(current.metadata);

            let sql = format!(
                "UPDATE {} SET recipient_email = $2, subject = $3, from_address = $4, from_name = $5, text_body = $6, html_body = $7, provider = $8, flow = $9, metadata = $10, updated_at = $11 \
                WHERE id = $1 \
                RETURNING id, recipient_email, subject, from_address, from_name, text_body, html_body, provider, flow, metadata, created_at, updated_at",
                self.table_ref("emails")
            );
            let row = sqlx::query_as::<_, SentEmail>(&sql)
                .bind(id)
                .bind(recipient_email)
                .bind(subject)
                .bind(from_address)
                .bind(from_name)
                .bind(text_body)
                .bind(html_body)
                .bind(provider)
                .bind(flow)
                .bind(sqlx::types::Json(metadata))
                .bind(Utc::now())
                .fetch_one(&self.pool)
                .await?;
            Ok(row)
        }

        async fn delete_sent_email(&self, id: &str) -> AuthResult<()> {
            let sql = format!("DELETE FROM {} WHERE id = $1", self.table_ref("emails"));
            let result = sqlx::query(&sql).bind(id).execute(&self.pool).await?;
            if result.rows_affected() == 0 {
                return Err(AuthError::not_found("Sent email not found"));
            }
            Ok(())
        }

        async fn list_email_send_failures(
            &self,
            params: ListEmailSendFailuresParams,
        ) -> AuthResult<Vec<EmailSendFailure>> {
            let limit = params.limit.unwrap_or(100).min(1000);
            let offset = params.offset.unwrap_or(0);
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT id, user_id, recipient_email, flow, provider, error_message, metadata, resolved, resolution_note, created_at, updated_at \
                FROM {}",
                self.table_ref("email_send_failures")
            ));
            let mut has_where = false;

            if let Some(recipient_email) = params
                .recipient_email
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(recipient_email) = LOWER(");
                query_builder.push_bind(recipient_email);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(provider) = params
                .provider
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(COALESCE(provider, '')) = LOWER(");
                query_builder.push_bind(provider);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(flow) = params
                .flow
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(flow) = LOWER(");
                query_builder.push_bind(flow);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(resolved) = params.resolved {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("resolved = ");
                query_builder.push_bind(resolved);
                has_where = true;
            }

            if let Some(created_after) = params.created_after {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at >= ");
                query_builder.push_bind(created_after);
                has_where = true;
            }

            if let Some(created_before) = params.created_before {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at <= ");
                query_builder.push_bind(created_before);
            }

            query_builder.push(" ORDER BY created_at DESC LIMIT ");
            query_builder.push_bind(limit as i64);
            query_builder.push(" OFFSET ");
            query_builder.push_bind(offset as i64);

            let rows = query_builder
                .build_query_as::<EmailSendFailure>()
                .fetch_all(&self.pool)
                .await?;
            Ok(rows)
        }

        async fn count_email_send_failures(
            &self,
            params: ListEmailSendFailuresParams,
        ) -> AuthResult<usize> {
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT COUNT(*)::BIGINT FROM {}",
                self.table_ref("email_send_failures")
            ));
            let mut has_where = false;

            if let Some(recipient_email) = params
                .recipient_email
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(recipient_email) = LOWER(");
                query_builder.push_bind(recipient_email);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(provider) = params
                .provider
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(COALESCE(provider, '')) = LOWER(");
                query_builder.push_bind(provider);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(flow) = params
                .flow
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(flow) = LOWER(");
                query_builder.push_bind(flow);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(resolved) = params.resolved {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("resolved = ");
                query_builder.push_bind(resolved);
                has_where = true;
            }

            if let Some(created_after) = params.created_after {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at >= ");
                query_builder.push_bind(created_after);
                has_where = true;
            }

            if let Some(created_before) = params.created_before {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at <= ");
                query_builder.push_bind(created_before);
            }

            let (total,): (i64,) = query_builder.build_query_as().fetch_one(&self.pool).await?;
            Ok(total as usize)
        }

        async fn get_email_send_failure_by_id(
            &self,
            id: &str,
        ) -> AuthResult<Option<EmailSendFailure>> {
            let sql = format!(
                "SELECT id, user_id, recipient_email, flow, provider, error_message, metadata, resolved, resolution_note, created_at, updated_at \
                FROM {} WHERE id = $1",
                self.table_ref("email_send_failures")
            );
            let row = sqlx::query_as::<_, EmailSendFailure>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(row)
        }

        async fn update_email_send_failure(
            &self,
            id: &str,
            update: UpdateEmailSendFailure,
        ) -> AuthResult<EmailSendFailure> {
            let current = self
                .get_email_send_failure_by_id(id)
                .await?
                .ok_or_else(|| AuthError::not_found("Email send failure not found"))?;

            let resolved = update.resolved.unwrap_or(current.resolved);
            let resolution_note = match update.resolution_note {
                Some(value) => value,
                None => current.resolution_note,
            };
            let metadata = update.metadata.unwrap_or(current.metadata);

            let sql = format!(
                "UPDATE {} SET resolved = $2, resolution_note = $3, metadata = $4, updated_at = $5 \
                WHERE id = $1 \
                RETURNING id, user_id, recipient_email, flow, provider, error_message, metadata, resolved, resolution_note, created_at, updated_at",
                self.table_ref("email_send_failures")
            );
            let row = sqlx::query_as::<_, EmailSendFailure>(&sql)
                .bind(id)
                .bind(resolved)
                .bind(resolution_note)
                .bind(sqlx::types::Json(metadata))
                .bind(Utc::now())
                .fetch_one(&self.pool)
                .await?;
            Ok(row)
        }

        async fn delete_email_send_failure(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE id = $1",
                self.table_ref("email_send_failures")
            );
            let result = sqlx::query(&sql).bind(id).execute(&self.pool).await?;
            if result.rows_affected() == 0 {
                return Err(AuthError::not_found("Email send failure not found"));
            }
            Ok(())
        }

        async fn get_active_email_template(
            &self,
            template_key: &str,
            locale: Option<&str>,
        ) -> AuthResult<Option<EmailTemplate>> {
            let preferred_locale = locale
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or("en")
                .to_ascii_lowercase();
            let fallback_locale = preferred_locale
                .split('-')
                .next()
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or("en")
                .to_string();

            let sql = format!(
                "SELECT id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at \
                FROM {} \
                WHERE template_key = $1 \
                AND is_active = TRUE \
                  AND LOWER(locale) IN (LOWER($2), LOWER($3), 'en') \
                ORDER BY \
                  CASE \
                    WHEN LOWER(locale) = LOWER($2) THEN 0 \
                    WHEN LOWER(locale) = LOWER($3) THEN 1 \
                    WHEN LOWER(locale) = 'en' THEN 2 \
                    ELSE 3 \
                  END, \
                  updated_at DESC \
                LIMIT 1",
                self.table_ref("email_templates")
            );

            let row = sqlx::query_as::<_, EmailTemplate>(&sql)
                .bind(template_key)
                .bind(preferred_locale)
                .bind(fallback_locale)
                .fetch_optional(&self.pool)
                .await?;

            Ok(row)
        }

        async fn get_active_email_template_for_event(
            &self,
            event_type: &str,
            template_key: &str,
            locale: Option<&str>,
        ) -> AuthResult<Option<EmailTemplate>> {
            let preferred_locale = locale
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or("en")
                .to_ascii_lowercase();
            let fallback_locale = preferred_locale
                .split('-')
                .next()
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or("en")
                .to_string();

            let sql = format!(
                "SELECT id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at \
                FROM {} \
                  WHERE is_active = TRUE \
                  AND (event_type = $1 OR template_key = $2) \
                  AND LOWER(locale) IN (LOWER($3), LOWER($4), 'en') \
                ORDER BY \
                  CASE \
                    WHEN event_type = $1 THEN 0 \
                    WHEN template_key = $2 THEN 1 \
                    ELSE 2 \
                  END, \
                  CASE \
                    WHEN LOWER(locale) = LOWER($3) THEN 0 \
                    WHEN LOWER(locale) = LOWER($4) THEN 1 \
                    WHEN LOWER(locale) = 'en' THEN 2 \
                    ELSE 3 \
                  END, \
                  updated_at DESC \
                LIMIT 1",
                self.table_ref("email_templates")
            );

            let row = sqlx::query_as::<_, EmailTemplate>(&sql)
                .bind(event_type)
                .bind(template_key)
                .bind(preferred_locale)
                .bind(fallback_locale)
                .fetch_optional(&self.pool)
                .await?;

            Ok(row)
        }

        async fn create_email_template(
            &self,
            input: CreateEmailTemplate,
        ) -> AuthResult<EmailTemplate> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();
            let template_key = input.template_key.trim().to_string();
            if template_key.is_empty() {
                return Err(AuthError::bad_request("templateKey is required"));
            }
            let event_type = input
                .event_type
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(str::to_string)
                .or_else(|| default_event_type_for_template_key(&template_key).map(str::to_string));
            if let Some(ref value) = event_type {
                let exists_sql = format!(
                    "SELECT 1::INT FROM {} WHERE event_type = $1 LIMIT 1",
                    self.table_ref("email_event_types")
                );
                let exists = sqlx::query_scalar::<_, i32>(&exists_sql)
                    .bind(value)
                    .fetch_optional(&self.pool)
                    .await?
                    .is_some();
                if !exists {
                    return Err(AuthError::bad_request(format!(
                        "Unknown eventType '{value}'"
                    )));
                }
            }
            let locale = input
                .locale
                .map(|value| value.trim().to_ascii_lowercase())
                .filter(|value| !value.is_empty())
                .unwrap_or_else(|| "en".to_string());
            let variables = input.variables.unwrap_or_else(|| serde_json::json!([]));
            let metadata = input.metadata.unwrap_or_else(|| serde_json::json!({}));
            let sql = format!(
                "INSERT INTO {} (id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) \
                RETURNING id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at",
                self.table_ref("email_templates")
            );

            let row = sqlx::query_as::<_, EmailTemplate>(&sql)
                .bind(id)
                .bind(template_key)
                .bind(event_type)
                .bind(locale)
                .bind(input.subject_template)
                .bind(input.text_template)
                .bind(input.html_template)
                .bind(sqlx::types::Json(variables))
                .bind(sqlx::types::Json(
                    input
                        .variable_bindings
                        .unwrap_or_else(|| serde_json::json!([])),
                ))
                .bind(input.is_active.unwrap_or(true))
                .bind(sqlx::types::Json(metadata))
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;

            Ok(row)
        }

        async fn list_email_templates(
            &self,
            params: ListEmailTemplatesParams,
        ) -> AuthResult<Vec<EmailTemplate>> {
            let limit = params.limit.unwrap_or(100).min(1000);
            let offset = params.offset.unwrap_or(0);
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at \
                FROM {}",
                self.table_ref("email_templates")
            ));
            let mut has_where = false;

            if let Some(template_key) = params
                .template_key
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("template_key = ");
                query_builder.push_bind(template_key);
                has_where = true;
            }

            if let Some(event_type) = params
                .event_type
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("event_type = ");
                query_builder.push_bind(event_type);
                has_where = true;
            }

            if let Some(locale) = params
                .locale
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(locale) = LOWER(");
                query_builder.push_bind(locale);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(is_active) = params.is_active {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("is_active = ");
                query_builder.push_bind(is_active);
            }

            query_builder.push(" ORDER BY updated_at DESC LIMIT ");
            query_builder.push_bind(limit as i64);
            query_builder.push(" OFFSET ");
            query_builder.push_bind(offset as i64);

            let rows = query_builder
                .build_query_as::<EmailTemplate>()
                .fetch_all(&self.pool)
                .await?;

            Ok(rows)
        }

        async fn get_email_template_by_id(&self, id: &str) -> AuthResult<Option<EmailTemplate>> {
            let sql = format!(
                "SELECT id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at \
                FROM {} WHERE id = $1",
                self.table_ref("email_templates")
            );
            let row = sqlx::query_as::<_, EmailTemplate>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(row)
        }

        async fn count_email_templates(
            &self,
            params: ListEmailTemplatesParams,
        ) -> AuthResult<usize> {
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT COUNT(*)::BIGINT FROM {}",
                self.table_ref("email_templates")
            ));
            let mut has_where = false;

            if let Some(template_key) = params
                .template_key
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("template_key = ");
                query_builder.push_bind(template_key);
                has_where = true;
            }

            if let Some(event_type) = params
                .event_type
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("event_type = ");
                query_builder.push_bind(event_type);
                has_where = true;
            }

            if let Some(locale) = params
                .locale
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(locale) = LOWER(");
                query_builder.push_bind(locale);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(is_active) = params.is_active {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("is_active = ");
                query_builder.push_bind(is_active);
            }

            let (total,): (i64,) = query_builder.build_query_as().fetch_one(&self.pool).await?;
            Ok(total as usize)
        }

        async fn update_email_template(
            &self,
            id: &str,
            update: UpdateEmailTemplate,
        ) -> AuthResult<EmailTemplate> {
            let select_sql = format!(
                "SELECT id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at \
                FROM {} WHERE id = $1",
                self.table_ref("email_templates")
            );
            let current = sqlx::query_as::<_, EmailTemplate>(&select_sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?
                .ok_or_else(|| AuthError::not_found("Email template not found"))?;

            let template_key = update.template_key.unwrap_or(current.template_key);
            let event_type = match update.event_type {
                Some(value) => value,
                None => current.event_type,
            };
            if let Some(ref value) = event_type {
                let exists_sql = format!(
                    "SELECT 1::INT FROM {} WHERE event_type = $1 LIMIT 1",
                    self.table_ref("email_event_types")
                );
                let exists = sqlx::query_scalar::<_, i32>(&exists_sql)
                    .bind(value)
                    .fetch_optional(&self.pool)
                    .await?
                    .is_some();
                if !exists {
                    return Err(AuthError::bad_request(format!(
                        "Unknown eventType '{value}'"
                    )));
                }
            }
            let locale = update
                .locale
                .map(|value| value.trim().to_ascii_lowercase())
                .filter(|value| !value.is_empty())
                .unwrap_or(current.locale);
            let subject_template = update.subject_template.unwrap_or(current.subject_template);
            let text_template = match update.text_template {
                Some(value) => value,
                None => current.text_template,
            };
            let html_template = match update.html_template {
                Some(value) => value,
                None => current.html_template,
            };
            let variables = update.variables.unwrap_or(current.variables);
            let variable_bindings = update
                .variable_bindings
                .unwrap_or(current.variable_bindings);
            let is_active = update.is_active.unwrap_or(current.is_active);
            let metadata = update.metadata.unwrap_or(current.metadata);
            let sql = format!(
                "UPDATE {} SET template_key = $2, event_type = $3, locale = $4, subject_template = $5, text_template = $6, html_template = $7, variables = $8, variable_bindings = $9, is_active = $10, metadata = $11, updated_at = $12 \
                WHERE id = $1 \
                RETURNING id, template_key, event_type, locale, subject_template, text_template, html_template, variables, variable_bindings, is_active, metadata, created_at, updated_at",
                self.table_ref("email_templates")
            );

            let row = sqlx::query_as::<_, EmailTemplate>(&sql)
                .bind(id)
                .bind(template_key)
                .bind(event_type)
                .bind(locale)
                .bind(subject_template)
                .bind(text_template)
                .bind(html_template)
                .bind(sqlx::types::Json(variables))
                .bind(sqlx::types::Json(variable_bindings))
                .bind(is_active)
                .bind(sqlx::types::Json(metadata))
                .bind(Utc::now())
                .fetch_one(&self.pool)
                .await?;

            Ok(row)
        }

        async fn delete_email_template(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE id = $1",
                self.table_ref("email_templates")
            );
            let result = sqlx::query(&sql).bind(id).execute(&self.pool).await?;
            if result.rows_affected() == 0 {
                return Err(AuthError::not_found("Email template not found"));
            }
            Ok(())
        }

        async fn list_email_event_types(
            &self,
            params: ListEmailEventTypesParams,
        ) -> AuthResult<Vec<EmailEventType>> {
            let limit = params.limit.unwrap_or(100).min(1000);
            let offset = params.offset.unwrap_or(0);
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT event_type, category, description, default_template_key, required_variables, optional_variables, is_active, is_system, metadata, created_at, updated_at \
                FROM {}",
                self.table_ref("email_event_types")
            ));
            let mut has_where = false;

            if let Some(event_type) = params
                .event_type
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("event_type = ");
                query_builder.push_bind(event_type);
                has_where = true;
            }

            if let Some(category) = params
                .category
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("category = ");
                query_builder.push_bind(category);
                has_where = true;
            }

            if let Some(is_active) = params.is_active {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("is_active = ");
                query_builder.push_bind(is_active);
                has_where = true;
            }

            if let Some(is_system) = params.is_system {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("is_system = ");
                query_builder.push_bind(is_system);
            }

            query_builder.push(" ORDER BY event_type ASC LIMIT ");
            query_builder.push_bind(limit as i64);
            query_builder.push(" OFFSET ");
            query_builder.push_bind(offset as i64);

            let rows = query_builder
                .build_query_as::<EmailEventType>()
                .fetch_all(&self.pool)
                .await?;

            Ok(rows)
        }

        async fn count_email_event_types(
            &self,
            params: ListEmailEventTypesParams,
        ) -> AuthResult<usize> {
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT COUNT(*)::BIGINT FROM {}",
                self.table_ref("email_event_types")
            ));
            let mut has_where = false;

            if let Some(event_type) = params
                .event_type
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("event_type = ");
                query_builder.push_bind(event_type);
                has_where = true;
            }

            if let Some(category) = params
                .category
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("category = ");
                query_builder.push_bind(category);
                has_where = true;
            }

            if let Some(is_active) = params.is_active {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("is_active = ");
                query_builder.push_bind(is_active);
                has_where = true;
            }

            if let Some(is_system) = params.is_system {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("is_system = ");
                query_builder.push_bind(is_system);
            }

            let (total,): (i64,) = query_builder.build_query_as().fetch_one(&self.pool).await?;
            Ok(total as usize)
        }

        async fn get_email_event_type(
            &self,
            event_type: &str,
        ) -> AuthResult<Option<EmailEventType>> {
            let sql = format!(
                "SELECT event_type, category, description, default_template_key, required_variables, optional_variables, is_active, is_system, metadata, created_at, updated_at \
                FROM {} WHERE event_type = $1",
                self.table_ref("email_event_types")
            );
            let row = sqlx::query_as::<_, EmailEventType>(&sql)
                .bind(event_type)
                .fetch_optional(&self.pool)
                .await?;
            Ok(row)
        }

        async fn create_admin_audit_log(&self, log: CreateAdminAuditLog) -> AuthResult<()> {
            let id = Uuid::new_v4().to_string();
            let metadata = log.metadata.unwrap_or_else(|| serde_json::json!({}));
            let sql = format!(
                "INSERT INTO {} (id, actor_user_id, actor_session_token, action, target_type, target_id, success, error_message, metadata, created_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
                self.table_ref("admin_audit_logs")
            );

            sqlx::query(&sql)
                .bind(id)
                .bind(log.actor_user_id)
                .bind(log.actor_session_token)
                .bind(log.action)
                .bind(log.target_type)
                .bind(log.target_id)
                .bind(log.success)
                .bind(log.error_message)
                .bind(sqlx::types::Json(metadata))
                .bind(Utc::now())
                .execute(&self.pool)
                .await?;
            Ok(())
        }

        async fn list_admin_audit_logs(
            &self,
            limit: usize,
            offset: usize,
        ) -> AuthResult<Vec<AdminAuditLog>> {
            self.list_admin_audit_logs_with_params(ListAdminAuditLogsParams {
                limit: Some(limit),
                offset: Some(offset),
                ..Default::default()
            })
            .await
        }

        async fn list_admin_audit_logs_with_params(
            &self,
            params: ListAdminAuditLogsParams,
        ) -> AuthResult<Vec<AdminAuditLog>> {
            let limit = params.limit.unwrap_or(100).min(1000);
            let offset = params.offset.unwrap_or(0);
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT id, actor_user_id, actor_session_token, action, target_type, target_id, success, error_message, metadata, created_at \
                FROM {}",
                self.table_ref("admin_audit_logs")
            ));
            let mut has_where = false;

            if let Some(actor_user_id) = params
                .actor_user_id
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("actor_user_id = ");
                query_builder.push_bind(actor_user_id);
                has_where = true;
            }

            if let Some(action) = params
                .action
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(action) = LOWER(");
                query_builder.push_bind(action);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(target_type) = params
                .target_type
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(COALESCE(target_type, '')) = LOWER(");
                query_builder.push_bind(target_type);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(target_id) = params
                .target_id
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("target_id = ");
                query_builder.push_bind(target_id);
                has_where = true;
            }

            if let Some(success) = params.success {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("success = ");
                query_builder.push_bind(success);
                has_where = true;
            }

            if let Some(created_after) = params.created_after {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at >= ");
                query_builder.push_bind(created_after);
                has_where = true;
            }

            if let Some(created_before) = params.created_before {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at <= ");
                query_builder.push_bind(created_before);
            }

            query_builder.push(" ORDER BY created_at DESC LIMIT ");
            query_builder.push_bind(limit as i64);
            query_builder.push(" OFFSET ");
            query_builder.push_bind(offset as i64);

            let rows = query_builder
                .build_query_as::<AdminAuditLog>()
                .fetch_all(&self.pool)
                .await?;
            Ok(rows)
        }

        async fn count_admin_audit_logs(
            &self,
            params: ListAdminAuditLogsParams,
        ) -> AuthResult<usize> {
            let mut query_builder = sqlx::QueryBuilder::new(format!(
                "SELECT COUNT(*)::BIGINT FROM {}",
                self.table_ref("admin_audit_logs")
            ));
            let mut has_where = false;

            if let Some(actor_user_id) = params
                .actor_user_id
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("actor_user_id = ");
                query_builder.push_bind(actor_user_id);
                has_where = true;
            }

            if let Some(action) = params
                .action
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(action) = LOWER(");
                query_builder.push_bind(action);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(target_type) = params
                .target_type
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("LOWER(COALESCE(target_type, '')) = LOWER(");
                query_builder.push_bind(target_type);
                query_builder.push(")");
                has_where = true;
            }

            if let Some(target_id) = params
                .target_id
                .as_deref()
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("target_id = ");
                query_builder.push_bind(target_id);
                has_where = true;
            }

            if let Some(success) = params.success {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("success = ");
                query_builder.push_bind(success);
                has_where = true;
            }

            if let Some(created_after) = params.created_after {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at >= ");
                query_builder.push_bind(created_after);
                has_where = true;
            }

            if let Some(created_before) = params.created_before {
                query_builder.push(if has_where { " AND " } else { " WHERE " });
                query_builder.push("created_at <= ");
                query_builder.push_bind(created_before);
            }

            let (total,): (i64,) = query_builder.build_query_as().fetch_one(&self.pool).await?;
            Ok(total as usize)
        }

        async fn create_athena_client(
            &self,
            input: CreateAthenaClient,
        ) -> AuthResult<AthenaClient> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();
            let metadata = input.metadata.unwrap_or_else(|| serde_json::json!({}));
            let sql = format!(
                "INSERT INTO {} (id, client_name, description, is_active, pg_uri_env_var, metadata, created_at, updated_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8) \
                RETURNING id, client_name, description, is_active, pg_uri_env_var, metadata, created_at, updated_at",
                self.table_ref("athena_clients")
            );

            let row = sqlx::query_as::<_, AthenaClient>(&sql)
                .bind(id)
                .bind(input.client_name)
                .bind(input.description)
                .bind(input.is_active.unwrap_or(true))
                .bind(input.pg_uri_env_var)
                .bind(sqlx::types::Json(metadata))
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;
            Ok(row)
        }

        async fn list_athena_clients(&self) -> AuthResult<Vec<AthenaClient>> {
            let sql = format!(
                "SELECT id, client_name, description, is_active, pg_uri_env_var, metadata, created_at, updated_at \
                FROM {} ORDER BY created_at DESC",
                self.table_ref("athena_clients")
            );
            let rows = sqlx::query_as::<_, AthenaClient>(&sql)
                .fetch_all(&self.pool)
                .await?;
            Ok(rows)
        }

        async fn create_oauth_client(&self, input: CreateOAuthClient) -> AuthResult<OAuthClient> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();
            let metadata = input.metadata.unwrap_or_else(|| serde_json::json!({}));
            let sql = format!(
                "INSERT INTO {} (id, client_name, provider, description, response_type, grant_type, token_authentication_method, redirect_uris, client_url, scopes, is_active, metadata, created_at, updated_at) \
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) \
                RETURNING id, client_name, provider, description, response_type, grant_type, token_authentication_method, redirect_uris, client_url, scopes, is_active, metadata, created_at, updated_at",
                self.table_ref("oauth_clients")
            );

            let row = sqlx::query_as::<_, OAuthClient>(&sql)
                .bind(id)
                .bind(input.client_name)
                .bind(input.provider)
                .bind(input.description)
                .bind(input.response_type.unwrap_or_else(|| "code".to_string()))
                .bind(
                    input
                        .grant_type
                        .unwrap_or_else(|| "authorization_code".to_string()),
                )
                .bind(
                    input
                        .token_authentication_method
                        .unwrap_or_else(|| "none".to_string()),
                )
                .bind(input.redirect_uris)
                .bind(input.client_url)
                .bind(input.scopes.unwrap_or_default())
                .bind(input.is_active.unwrap_or(true))
                .bind(sqlx::types::Json(metadata))
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;
            Ok(row)
        }

        async fn list_oauth_clients(&self) -> AuthResult<Vec<OAuthClient>> {
            let sql = format!(
                "SELECT id, client_name, provider, description, response_type, grant_type, token_authentication_method, redirect_uris, client_url, scopes, is_active, metadata, created_at, updated_at \
                FROM {} ORDER BY created_at DESC",
                self.table_ref("oauth_clients")
            );
            let rows = sqlx::query_as::<_, OAuthClient>(&sql)
                .fetch_all(&self.pool)
                .await?;
            Ok(rows)
        }

        async fn find_active_oauth_client_by_provider_and_redirect_uri(
            &self,
            provider: &str,
            redirect_uri: &str,
        ) -> AuthResult<Option<OAuthClient>> {
            let sql = format!(
                "SELECT id, client_name, provider, description, response_type, grant_type, token_authentication_method, redirect_uris, client_url, scopes, is_active, metadata, created_at, updated_at \
                FROM {} \
                WHERE is_active = TRUE AND lower(provider) = lower($1) AND $2 = ANY(redirect_uris) \
                ORDER BY created_at DESC LIMIT 1",
                self.table_ref("oauth_clients")
            );
            let row = sqlx::query_as::<_, OAuthClient>(&sql)
                .bind(provider)
                .bind(redirect_uri)
                .fetch_optional(&self.pool)
                .await?;
            Ok(row)
        }

        async fn delete_expired_verifications(&self) -> AuthResult<usize> {
            let sql = format!(
                "DELETE FROM {} WHERE {} < NOW()",
                self.table_ref(V::table()),
                qi(V::col_expires_at())
            );
            let result = sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(|e| db_error_with_context("delete-expired", V::table(), e))?;
            Ok(result.rows_affected() as usize)
        }
    }

    // -- OrganizationOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> OrganizationOps
        for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type Organization = O;

        async fn create_organization(&self, create_org: CreateOrganization) -> AuthResult<O> {
            let id = create_org.id.unwrap_or_else(|| Uuid::new_v4().to_string());
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}, {}) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING *",
                self.table_ref(O::table()),
                qi(O::col_id()),
                qi(O::col_name()),
                qi(O::col_slug()),
                qi(O::col_logo()),
                qi(O::col_metadata()),
                qi(O::col_created_at()),
                qi(O::col_updated_at()),
            );
            let organization = sqlx::query_as::<_, O>(&sql)
                .bind(&id)
                .bind(&create_org.name)
                .bind(&create_org.slug)
                .bind(&create_org.logo)
                .bind(sqlx::types::Json(
                    create_org.metadata.unwrap_or(serde_json::json!({})),
                ))
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;

            Ok(organization)
        }

        async fn get_organization_by_id(&self, id: &str) -> AuthResult<Option<O>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(O::table()),
                qi(O::col_id())
            );
            let organization = sqlx::query_as::<_, O>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(organization)
        }

        async fn get_organization_by_slug(&self, slug: &str) -> AuthResult<Option<O>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(O::table()),
                qi(O::col_slug())
            );
            let organization = sqlx::query_as::<_, O>(&sql)
                .bind(slug)
                .fetch_optional(&self.pool)
                .await?;
            Ok(organization)
        }

        async fn update_organization(&self, id: &str, update: UpdateOrganization) -> AuthResult<O> {
            let mut query = sqlx::QueryBuilder::new(format!(
                "UPDATE {} SET {} = NOW()",
                self.table_ref(O::table()),
                qi(O::col_updated_at())
            ));

            if let Some(name) = &update.name {
                query.push(format!(", {} = ", qi(O::col_name())));
                query.push_bind(name);
            }
            if let Some(slug) = &update.slug {
                query.push(format!(", {} = ", qi(O::col_slug())));
                query.push_bind(slug);
            }
            if let Some(logo) = &update.logo {
                query.push(format!(", {} = ", qi(O::col_logo())));
                query.push_bind(logo);
            }
            if let Some(metadata) = &update.metadata {
                query.push(format!(", {} = ", qi(O::col_metadata())));
                query.push_bind(sqlx::types::Json(metadata.clone()));
            }

            query.push(format!(" WHERE {} = ", qi(O::col_id())));
            query.push_bind(id);
            query.push(" RETURNING *");

            let organization = query.build_query_as::<O>().fetch_one(&self.pool).await?;
            Ok(organization)
        }

        async fn delete_organization(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(O::table()),
                qi(O::col_id())
            );
            sqlx::query(&sql).bind(id).execute(&self.pool).await?;
            Ok(())
        }

        async fn list_user_organizations(&self, user_id: &str) -> AuthResult<Vec<O>> {
            let sql = format!(
                "SELECT o.* FROM {} o INNER JOIN {} m ON o.{} = m.{} WHERE m.{} = $1 ORDER BY o.{} DESC",
                self.table_ref(O::table()),
                self.table_ref(M::table()),
                qi(O::col_id()),
                qi(M::col_organization_id()),
                qi(M::col_user_id()),
                qi(O::col_created_at()),
            );
            let organizations = sqlx::query_as::<_, O>(&sql)
                // Avoid stale prepared-plan failures when organization row shape changes.
                .persistent(false)
                .bind(user_id)
                .fetch_all(&self.pool)
                .await?;
            Ok(organizations)
        }
    }

    // -- MemberOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> MemberOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type Member = M;

        async fn create_member(&self, create_member: CreateMember) -> AuthResult<M> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}) VALUES ($1, $2, $3, $4, $5) RETURNING *",
                self.table_ref(M::table()),
                qi(M::col_id()),
                qi(M::col_organization_id()),
                qi(M::col_user_id()),
                qi(M::col_role()),
                qi(M::col_created_at()),
            );
            let member = sqlx::query_as::<_, M>(&sql)
                .bind(&id)
                .bind(&create_member.organization_id)
                .bind(&create_member.user_id)
                .bind(&create_member.role)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;

            Ok(member)
        }

        async fn get_member(&self, organization_id: &str, user_id: &str) -> AuthResult<Option<M>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND {} = $2",
                self.table_ref(M::table()),
                qi(M::col_organization_id()),
                qi(M::col_user_id())
            );
            let member = sqlx::query_as::<_, M>(&sql)
                .bind(organization_id)
                .bind(user_id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(member)
        }

        async fn get_member_by_id(&self, id: &str) -> AuthResult<Option<M>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(M::table()),
                qi(M::col_id())
            );
            let member = sqlx::query_as::<_, M>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(member)
        }

        async fn update_member_role(&self, member_id: &str, role: &str) -> AuthResult<M> {
            let sql = format!(
                "UPDATE {} SET {} = $1 WHERE {} = $2 RETURNING *",
                self.table_ref(M::table()),
                qi(M::col_role()),
                qi(M::col_id())
            );
            let member = sqlx::query_as::<_, M>(&sql)
                .bind(role)
                .bind(member_id)
                .fetch_one(&self.pool)
                .await?;
            Ok(member)
        }

        async fn delete_member(&self, member_id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(M::table()),
                qi(M::col_id())
            );
            sqlx::query(&sql)
                .bind(member_id)
                .execute(&self.pool)
                .await?;
            Ok(())
        }

        async fn list_organization_members(&self, organization_id: &str) -> AuthResult<Vec<M>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 ORDER BY {} ASC",
                self.table_ref(M::table()),
                qi(M::col_organization_id()),
                qi(M::col_created_at())
            );
            let members = sqlx::query_as::<_, M>(&sql)
                .bind(organization_id)
                .fetch_all(&self.pool)
                .await?;
            Ok(members)
        }

        async fn count_organization_members(&self, organization_id: &str) -> AuthResult<usize> {
            let sql = format!(
                "SELECT COUNT(*) FROM {} WHERE {} = $1",
                self.table_ref(M::table()),
                qi(M::col_organization_id())
            );
            let count: (i64,) = sqlx::query_as(&sql)
                .bind(organization_id)
                .fetch_one(&self.pool)
                .await?;
            Ok(count.0 as usize)
        }

        async fn count_organization_owners(&self, organization_id: &str) -> AuthResult<usize> {
            let sql = format!(
                "SELECT COUNT(*) FROM {} WHERE {} = $1 AND {} = 'owner'",
                self.table_ref(M::table()),
                qi(M::col_organization_id()),
                qi(M::col_role())
            );
            let count: (i64,) = sqlx::query_as(&sql)
                .bind(organization_id)
                .fetch_one(&self.pool)
                .await?;
            Ok(count.0 as usize)
        }
    }

    // -- InvitationOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> InvitationOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type Invitation = I;

        async fn create_invitation(&self, create_inv: CreateInvitation) -> AuthResult<I> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}, {}, {}) \
                 VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7) RETURNING *",
                self.table_ref(I::table()),
                qi(I::col_id()),
                qi(I::col_organization_id()),
                qi(I::col_email()),
                qi(I::col_role()),
                qi(I::col_status()),
                qi(I::col_inviter_id()),
                qi(I::col_expires_at()),
                qi(I::col_created_at()),
            );
            let invitation = sqlx::query_as::<_, I>(&sql)
                .bind(&id)
                .bind(&create_inv.organization_id)
                .bind(&create_inv.email)
                .bind(&create_inv.role)
                .bind(&create_inv.inviter_id)
                .bind(create_inv.expires_at)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;

            Ok(invitation)
        }

        async fn get_invitation_by_id(&self, id: &str) -> AuthResult<Option<I>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(I::table()),
                qi(I::col_id())
            );
            let invitation = sqlx::query_as::<_, I>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(invitation)
        }

        async fn get_pending_invitation(
            &self,
            organization_id: &str,
            email: &str,
        ) -> AuthResult<Option<I>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 AND LOWER({}) = LOWER($2) AND {} = 'pending'",
                self.table_ref(I::table()),
                qi(I::col_organization_id()),
                qi(I::col_email()),
                qi(I::col_status())
            );
            let invitation = sqlx::query_as::<_, I>(&sql)
                .bind(organization_id)
                .bind(email)
                .fetch_optional(&self.pool)
                .await?;
            Ok(invitation)
        }

        async fn update_invitation_status(
            &self,
            id: &str,
            status: InvitationStatus,
        ) -> AuthResult<I> {
            let sql = format!(
                "UPDATE {} SET {} = $1 WHERE {} = $2 RETURNING *",
                self.table_ref(I::table()),
                qi(I::col_status()),
                qi(I::col_id())
            );
            let invitation = sqlx::query_as::<_, I>(&sql)
                .bind(status.to_string())
                .bind(id)
                .fetch_one(&self.pool)
                .await?;
            Ok(invitation)
        }

        async fn list_organization_invitations(&self, organization_id: &str) -> AuthResult<Vec<I>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 ORDER BY {} DESC",
                self.table_ref(I::table()),
                qi(I::col_organization_id()),
                qi(I::col_created_at())
            );
            let invitations = sqlx::query_as::<_, I>(&sql)
                .bind(organization_id)
                .fetch_all(&self.pool)
                .await?;
            Ok(invitations)
        }

        async fn list_user_invitations(&self, email: &str) -> AuthResult<Vec<I>> {
            let sql = format!(
                "SELECT * FROM {} WHERE LOWER({}) = LOWER($1) AND {} = 'pending' AND {} > NOW() ORDER BY {} DESC",
                self.table_ref(I::table()),
                qi(I::col_email()),
                qi(I::col_status()),
                qi(I::col_expires_at()),
                qi(I::col_created_at())
            );
            let invitations = sqlx::query_as::<_, I>(&sql)
                .bind(email)
                .fetch_all(&self.pool)
                .await?;
            Ok(invitations)
        }
    }

    // -- TwoFactorOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> TwoFactorOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type TwoFactor = TF;

        async fn create_two_factor(&self, create: CreateTwoFactor) -> AuthResult<TF> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *",
                self.table_ref(TF::table()),
                qi(TF::col_id()),
                qi(TF::col_secret()),
                qi(TF::col_backup_codes()),
                qi(TF::col_user_id()),
                qi(TF::col_created_at()),
                qi(TF::col_updated_at()),
            );
            let two_factor = sqlx::query_as::<_, TF>(&sql)
                .bind(&id)
                .bind(&create.secret)
                .bind(&create.backup_codes)
                .bind(&create.user_id)
                .bind(now)
                .bind(now)
                .fetch_one(&self.pool)
                .await?;

            Ok(two_factor)
        }

        async fn get_two_factor_by_user_id(&self, user_id: &str) -> AuthResult<Option<TF>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(TF::table()),
                qi(TF::col_user_id())
            );
            let two_factor = sqlx::query_as::<_, TF>(&sql)
                .bind(user_id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(two_factor)
        }

        async fn update_two_factor_backup_codes(
            &self,
            user_id: &str,
            backup_codes: &str,
        ) -> AuthResult<TF> {
            let sql = format!(
                "UPDATE {} SET {} = $1, {} = NOW() WHERE {} = $2 RETURNING *",
                self.table_ref(TF::table()),
                qi(TF::col_backup_codes()),
                qi(TF::col_updated_at()),
                qi(TF::col_user_id())
            );
            let two_factor = sqlx::query_as::<_, TF>(&sql)
                .bind(backup_codes)
                .bind(user_id)
                .fetch_one(&self.pool)
                .await?;
            Ok(two_factor)
        }

        async fn delete_two_factor(&self, user_id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(TF::table()),
                qi(TF::col_user_id())
            );
            sqlx::query(&sql).bind(user_id).execute(&self.pool).await?;
            Ok(())
        }
    }

    // -- ApiKeyOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> ApiKeyOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type ApiKey = AK;

        async fn create_api_key(&self, input: CreateApiKey) -> AuthResult<AK> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}) \
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14::timestamptz, $15, $16, $17, $18) RETURNING *",
                self.table_ref(AK::table()),
                qi(AK::col_id()),
                qi(AK::col_name()),
                qi(AK::col_start()),
                qi(AK::col_prefix()),
                qi(AK::col_key_hash()),
                qi(AK::col_user_id()),
                qi(AK::col_refill_interval()),
                qi(AK::col_refill_amount()),
                qi(AK::col_enabled()),
                qi(AK::col_rate_limit_enabled()),
                qi(AK::col_rate_limit_time_window()),
                qi(AK::col_rate_limit_max()),
                qi(AK::col_remaining()),
                qi(AK::col_expires_at()),
                qi(AK::col_created_at()),
                qi(AK::col_updated_at()),
                qi(AK::col_permissions()),
                qi(AK::col_metadata()),
            );
            let api_key = sqlx::query_as::<_, AK>(&sql)
                .bind(&id)
                .bind(&input.name)
                .bind(&input.start)
                .bind(&input.prefix)
                .bind(&input.key_hash)
                .bind(&input.user_id)
                .bind(input.refill_interval)
                .bind(input.refill_amount)
                .bind(input.enabled)
                .bind(input.rate_limit_enabled)
                .bind(input.rate_limit_time_window)
                .bind(input.rate_limit_max)
                .bind(input.remaining)
                .bind(&input.expires_at)
                .bind(now)
                .bind(now)
                .bind(&input.permissions)
                .bind(&input.metadata)
                .fetch_one(&self.pool)
                .await?;

            Ok(api_key)
        }

        async fn get_api_key_by_id(&self, id: &str) -> AuthResult<Option<AK>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(AK::table()),
                qi(AK::col_id())
            );
            let api_key = sqlx::query_as::<_, AK>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(api_key)
        }

        async fn get_api_key_by_hash(&self, hash: &str) -> AuthResult<Option<AK>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(AK::table()),
                qi(AK::col_key_hash())
            );
            let api_key = sqlx::query_as::<_, AK>(&sql)
                .bind(hash)
                .fetch_optional(&self.pool)
                .await?;
            Ok(api_key)
        }

        async fn list_api_keys_by_user(&self, user_id: &str) -> AuthResult<Vec<AK>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 ORDER BY {} DESC",
                self.table_ref(AK::table()),
                qi(AK::col_user_id()),
                qi(AK::col_created_at())
            );
            let keys = sqlx::query_as::<_, AK>(&sql)
                .bind(user_id)
                .fetch_all(&self.pool)
                .await?;
            Ok(keys)
        }

        async fn update_api_key(&self, id: &str, update: UpdateApiKey) -> AuthResult<AK> {
            let mut query = sqlx::QueryBuilder::new(format!(
                "UPDATE {} SET {} = NOW()",
                self.table_ref(AK::table()),
                qi(AK::col_updated_at())
            ));

            if let Some(name) = &update.name {
                query.push(format!(", {} = ", qi(AK::col_name())));
                query.push_bind(name);
            }
            if let Some(enabled) = update.enabled {
                query.push(format!(", {} = ", qi(AK::col_enabled())));
                query.push_bind(enabled);
            }
            if let Some(remaining) = update.remaining {
                query.push(format!(", {} = ", qi(AK::col_remaining())));
                query.push_bind(remaining);
            }
            if let Some(rate_limit_enabled) = update.rate_limit_enabled {
                query.push(format!(", {} = ", qi(AK::col_rate_limit_enabled())));
                query.push_bind(rate_limit_enabled);
            }
            if let Some(rate_limit_time_window) = update.rate_limit_time_window {
                query.push(format!(", {} = ", qi(AK::col_rate_limit_time_window())));
                query.push_bind(rate_limit_time_window);
            }
            if let Some(rate_limit_max) = update.rate_limit_max {
                query.push(format!(", {} = ", qi(AK::col_rate_limit_max())));
                query.push_bind(rate_limit_max);
            }
            if let Some(refill_interval) = update.refill_interval {
                query.push(format!(", {} = ", qi(AK::col_refill_interval())));
                query.push_bind(refill_interval);
            }
            if let Some(refill_amount) = update.refill_amount {
                query.push(format!(", {} = ", qi(AK::col_refill_amount())));
                query.push_bind(refill_amount);
            }
            if let Some(permissions) = &update.permissions {
                query.push(format!(", {} = ", qi(AK::col_permissions())));
                query.push_bind(permissions);
            }
            if let Some(metadata) = &update.metadata {
                query.push(format!(", {} = ", qi(AK::col_metadata())));
                query.push_bind(metadata);
            }
            if let Some(expires_at) = &update.expires_at {
                query.push(format!(", {} = ", qi(AK::col_expires_at())));
                query.push_bind(expires_at.as_deref().map(|s| s.to_string()));
            }
            if let Some(last_request) = &update.last_request {
                query.push(format!(", {} = ", qi(AK::col_last_request())));
                query.push_bind(last_request.as_deref().map(|s| s.to_string()));
            }
            if let Some(request_count) = update.request_count {
                query.push(format!(", {} = ", qi(AK::col_request_count())));
                query.push_bind(request_count);
            }
            if let Some(last_refill_at) = &update.last_refill_at {
                query.push(format!(", {} = ", qi(AK::col_last_refill_at())));
                query.push_bind(last_refill_at.as_deref().map(|s| s.to_string()));
            }

            query.push(format!(" WHERE {} = ", qi(AK::col_id())));
            query.push_bind(id);
            query.push(" RETURNING *");

            let api_key = query
                .build_query_as::<AK>()
                .fetch_one(&self.pool)
                .await
                .map_err(|err| match err {
                    sqlx::Error::RowNotFound => AuthError::not_found("API key not found"),
                    other => AuthError::from(other),
                })?;
            Ok(api_key)
        }

        async fn delete_api_key(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(AK::table()),
                qi(AK::col_id())
            );
            sqlx::query(&sql).bind(id).execute(&self.pool).await?;
            Ok(())
        }

        async fn delete_expired_api_keys(&self) -> AuthResult<usize> {
            // Use a bind-parameter with an RFC 3339 timestamp instead of
            // NOW() / ::timestamptz so the comparison is straightforward.
            // (This adapter is PostgreSQL-only; $1 placeholders are correct.)
            let now = Utc::now().to_rfc3339();
            let sql = format!(
                "DELETE FROM {} WHERE {} IS NOT NULL AND {} < $1",
                self.table_ref(AK::table()),
                qi(AK::col_expires_at()),
                qi(AK::col_expires_at()),
            );
            let result = sqlx::query(&sql).bind(&now).execute(&self.pool).await?;
            Ok(result.rows_affected() as usize)
        }
    }

    // -- PasskeyOps --

    #[async_trait]
    impl<U, S, A, O, M, I, V, TF, AK, PK> PasskeyOps for SqlxAdapter<U, S, A, O, M, I, V, TF, AK, PK>
    where
        U: AuthUser + AuthUserMeta + SqlxEntity,
        S: AuthSession + AuthSessionMeta + SqlxEntity,
        A: AuthAccount + AuthAccountMeta + SqlxEntity,
        O: AuthOrganization + AuthOrganizationMeta + SqlxEntity,
        M: AuthMember + AuthMemberMeta + SqlxEntity,
        I: AuthInvitation + AuthInvitationMeta + SqlxEntity,
        V: AuthVerification + AuthVerificationMeta + SqlxEntity,
        TF: AuthTwoFactor + AuthTwoFactorMeta + SqlxEntity,
        AK: AuthApiKey + AuthApiKeyMeta + SqlxEntity,
        PK: AuthPasskey + AuthPasskeyMeta + SqlxEntity,
    {
        type Passkey = PK;

        async fn create_passkey(&self, input: CreatePasskey) -> AuthResult<PK> {
            let id = Uuid::new_v4().to_string();
            let now = Utc::now();
            let counter = i64::try_from(input.counter)
                .map_err(|_| AuthError::bad_request("Passkey counter exceeds i64 range"))?;

            let sql = format!(
                "INSERT INTO {} ({}, {}, {}, {}, {}, {}, {}, {}, {}, {}) \
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING *",
                self.table_ref(PK::table()),
                qi(PK::col_id()),
                qi(PK::col_name()),
                qi(PK::col_public_key()),
                qi(PK::col_user_id()),
                qi(PK::col_credential_id()),
                qi(PK::col_counter()),
                qi(PK::col_device_type()),
                qi(PK::col_backed_up()),
                qi(PK::col_transports()),
                qi(PK::col_created_at()),
            );
            let passkey = sqlx::query_as::<_, PK>(&sql)
                .bind(&id)
                .bind(&input.name)
                .bind(&input.public_key)
                .bind(&input.user_id)
                .bind(&input.credential_id)
                .bind(counter)
                .bind(&input.device_type)
                .bind(input.backed_up)
                .bind(&input.transports)
                .bind(now)
                .fetch_one(&self.pool)
                .await
                .map_err(|e| match e {
                    sqlx::Error::Database(ref db_err) if db_err.is_unique_violation() => {
                        AuthError::conflict("A passkey with this credential ID already exists")
                    }
                    other => AuthError::from(other),
                })?;

            Ok(passkey)
        }

        async fn get_passkey_by_id(&self, id: &str) -> AuthResult<Option<PK>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(PK::table()),
                qi(PK::col_id())
            );
            let passkey = sqlx::query_as::<_, PK>(&sql)
                .bind(id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(passkey)
        }

        async fn get_passkey_by_credential_id(
            &self,
            credential_id: &str,
        ) -> AuthResult<Option<PK>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1",
                self.table_ref(PK::table()),
                qi(PK::col_credential_id())
            );
            let passkey = sqlx::query_as::<_, PK>(&sql)
                .bind(credential_id)
                .fetch_optional(&self.pool)
                .await?;
            Ok(passkey)
        }

        async fn list_passkeys_by_user(&self, user_id: &str) -> AuthResult<Vec<PK>> {
            let sql = format!(
                "SELECT * FROM {} WHERE {} = $1 ORDER BY {} DESC",
                self.table_ref(PK::table()),
                qi(PK::col_user_id()),
                qi(PK::col_created_at())
            );
            let passkeys = sqlx::query_as::<_, PK>(&sql)
                .bind(user_id)
                .fetch_all(&self.pool)
                .await?;
            Ok(passkeys)
        }

        async fn update_passkey_counter(&self, id: &str, counter: u64) -> AuthResult<PK> {
            let counter = i64::try_from(counter)
                .map_err(|_| AuthError::bad_request("Passkey counter exceeds i64 range"))?;
            let sql = format!(
                "UPDATE {} SET {} = $2 WHERE {} = $1 RETURNING *",
                self.table_ref(PK::table()),
                qi(PK::col_counter()),
                qi(PK::col_id())
            );
            let passkey = sqlx::query_as::<_, PK>(&sql)
                .bind(id)
                .bind(counter)
                .fetch_one(&self.pool)
                .await
                .map_err(|err| match err {
                    sqlx::Error::RowNotFound => AuthError::not_found("Passkey not found"),
                    other => AuthError::from(other),
                })?;
            Ok(passkey)
        }

        async fn update_passkey_name(&self, id: &str, name: &str) -> AuthResult<PK> {
            let sql = format!(
                "UPDATE {} SET {} = $2 WHERE {} = $1 RETURNING *",
                self.table_ref(PK::table()),
                qi(PK::col_name()),
                qi(PK::col_id())
            );
            let passkey = sqlx::query_as::<_, PK>(&sql)
                .bind(id)
                .bind(name)
                .fetch_one(&self.pool)
                .await
                .map_err(|err| match err {
                    sqlx::Error::RowNotFound => AuthError::not_found("Passkey not found"),
                    other => AuthError::from(other),
                })?;
            Ok(passkey)
        }

        async fn delete_passkey(&self, id: &str) -> AuthResult<()> {
            let sql = format!(
                "DELETE FROM {} WHERE {} = $1",
                self.table_ref(PK::table()),
                qi(PK::col_id())
            );
            sqlx::query(&sql).bind(id).execute(&self.pool).await?;
            Ok(())
        }
    }

    #[cfg(test)]
    mod tests {
        use super::{SqlxAdapter, qtable};
        use crate::adapters::VerificationOps;
        use crate::config::AuthConfig;
        use crate::email::verification_email_for_context;
        use crate::plugin::AuthContext;
        use crate::types::{
            CreateAdminAuditLog, CreateEmailSendFailure, CreateEmailTemplate, CreateSentEmail,
            ListAdminAuditLogsParams, ListEmailSendFailuresParams, ListEmailTemplatesParams,
            ListSentEmailsParams, UpdateEmailSendFailure, UpdateEmailTemplate, UpdateSentEmail,
        };
        use std::sync::Arc;

        fn sanitized_test_nonce() -> String {
            let nanos = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .expect("system time should be after unix epoch")
                .as_nanos();
            format!("{}_{}", std::process::id(), nanos)
        }

        async fn create_email_tables_for_schema(
            pool: &sqlx::PgPool,
            schema: &str,
        ) -> Result<(), sqlx::Error> {
            let create_schema_sql = format!("CREATE SCHEMA IF NOT EXISTS \"{schema}\"");
            sqlx::query(create_schema_sql.as_str())
                .execute(pool)
                .await?;

            let create_emails_sql = format!(
                "CREATE TABLE IF NOT EXISTS \"{schema}\".emails (
                    id TEXT PRIMARY KEY,
                    recipient_email TEXT NOT NULL,
                    subject TEXT NOT NULL,
                    from_address TEXT NOT NULL,
                    from_name TEXT,
                    text_body TEXT,
                    html_body TEXT,
                    provider TEXT NOT NULL,
                    flow TEXT,
                    metadata JSONB NOT NULL DEFAULT '{{}}',
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )"
            );
            sqlx::query(create_emails_sql.as_str())
                .execute(pool)
                .await?;

            let create_failures_sql = format!(
                "CREATE TABLE IF NOT EXISTS \"{schema}\".email_send_failures (
                    id TEXT PRIMARY KEY,
                    user_id TEXT,
                    recipient_email TEXT NOT NULL,
                    flow TEXT NOT NULL,
                    provider TEXT,
                    error_message TEXT NOT NULL,
                    metadata JSONB NOT NULL DEFAULT '{{}}',
                    resolved BOOLEAN NOT NULL DEFAULT FALSE,
                    resolution_note TEXT,
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )"
            );
            sqlx::query(create_failures_sql.as_str())
                .execute(pool)
                .await?;

            let create_event_types_sql = format!(
                "CREATE TABLE IF NOT EXISTS \"{schema}\".email_event_types (
                    event_type TEXT PRIMARY KEY,
                    category TEXT NOT NULL,
                    description TEXT NOT NULL,
                    default_template_key TEXT,
                    required_variables JSONB NOT NULL DEFAULT '[]',
                    optional_variables JSONB NOT NULL DEFAULT '[]',
                    is_active BOOLEAN NOT NULL DEFAULT TRUE,
                    is_system BOOLEAN NOT NULL DEFAULT TRUE,
                    metadata JSONB NOT NULL DEFAULT '{{}}',
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )"
            );
            sqlx::query(create_event_types_sql.as_str())
                .execute(pool)
                .await?;

            let create_templates_sql = format!(
                "CREATE TABLE IF NOT EXISTS \"{schema}\".email_templates (
                    id TEXT PRIMARY KEY,
                    template_key TEXT NOT NULL,
                    event_type TEXT REFERENCES \"{schema}\".email_event_types(event_type) ON UPDATE CASCADE ON DELETE SET NULL,
                    locale TEXT NOT NULL DEFAULT 'en',
                    subject_template TEXT NOT NULL,
                    text_template TEXT,
                    html_template TEXT,
                    variables JSONB NOT NULL DEFAULT '[]',
                    variable_bindings JSONB NOT NULL DEFAULT '[]',
                    is_active BOOLEAN NOT NULL DEFAULT TRUE,
                    metadata JSONB NOT NULL DEFAULT '{{}}',
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    UNIQUE (template_key, locale)
                )"
            );
            sqlx::query(create_templates_sql.as_str())
                .execute(pool)
                .await?;

            let add_variable_bindings_sql = format!(
                "ALTER TABLE \"{schema}\".email_templates \
                 ADD COLUMN IF NOT EXISTS variable_bindings JSONB NOT NULL DEFAULT '[]'"
            );
            sqlx::query(add_variable_bindings_sql.as_str())
                .execute(pool)
                .await?;

            let create_admin_audit_logs_sql = format!(
                "CREATE TABLE IF NOT EXISTS \"{schema}\".admin_audit_logs (
                    id TEXT PRIMARY KEY,
                    actor_user_id TEXT,
                    actor_session_token TEXT,
                    action TEXT NOT NULL,
                    target_type TEXT,
                    target_id TEXT,
                    success BOOLEAN NOT NULL DEFAULT FALSE,
                    error_message TEXT,
                    metadata JSONB NOT NULL DEFAULT '{{}}',
                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
                )"
            );
            sqlx::query(create_admin_audit_logs_sql.as_str())
                .execute(pool)
                .await?;

            Ok(())
        }

        async fn insert_event_type_for_schema(
            pool: &sqlx::PgPool,
            schema: &str,
            event_type: &str,
            default_template_key: Option<&str>,
        ) -> Result<(), sqlx::Error> {
            let sql = format!(
                "INSERT INTO \"{schema}\".email_event_types \
                (event_type, category, description, default_template_key, required_variables, optional_variables, is_active, is_system, metadata) \
                VALUES ($1, 'user_lifecycle', 'test event type', $2, '[]'::jsonb, '[]'::jsonb, TRUE, TRUE, '{{}}'::jsonb) \
                ON CONFLICT (event_type) DO NOTHING"
            );

            sqlx::query(sql.as_str())
                .bind(event_type)
                .bind(default_template_key)
                .execute(pool)
                .await?;

            Ok(())
        }

        #[test]
        fn qtable_qualifies_unqualified_table_with_schema() {
            assert_eq!(
                qtable("auth_runtime", "users"),
                "\"auth_runtime\".\"users\""
            );
        }

        #[test]
        fn qtable_quotes_each_segment_for_prequalified_table() {
            assert_eq!(
                qtable("public", "auth_runtime.users"),
                "\"auth_runtime\".\"users\""
            );
        }

        #[tokio::test]
        async fn sqlx_adapter_email_entities_are_isolated_per_schema() {
            let Some(database_url) = std::env::var("TEST_DATABASE_URL")
                .ok()
                .map(|value| value.trim().to_string())
                .filter(|value| !value.is_empty())
            else {
                println!("Skipping schema isolation test - TEST_DATABASE_URL not set");
                return;
            };

            let pool = sqlx::PgPool::connect(database_url.as_str())
                .await
                .expect("should connect to test postgres");

            let nonce = sanitized_test_nonce();
            let schema_a = format!("tenant_email_a_{nonce}");
            let schema_b = format!("tenant_email_b_{nonce}");

            create_email_tables_for_schema(&pool, schema_a.as_str())
                .await
                .expect("should create schema A email tables");
            create_email_tables_for_schema(&pool, schema_b.as_str())
                .await
                .expect("should create schema B email tables");

            let adapter_a: SqlxAdapter =
                SqlxAdapter::from_pool_with_schema(pool.clone(), schema_a.as_str());
            let adapter_b: SqlxAdapter =
                SqlxAdapter::from_pool_with_schema(pool.clone(), schema_b.as_str());

            adapter_a
                .create_sent_email(CreateSentEmail {
                    recipient_email: "tenant-a@example.com".to_string(),
                    subject: "Tenant A Subject".to_string(),
                    from_address: "noreply@example.com".to_string(),
                    from_name: Some("Tenant A".to_string()),
                    text_body: Some("Tenant A body".to_string()),
                    html_body: Some("<p>Tenant A</p>".to_string()),
                    provider: "tenant-a-provider".to_string(),
                    flow: Some("tenant_a_flow".to_string()),
                    metadata: Some(serde_json::json!({ "tenant": "a" })),
                })
                .await
                .expect("tenant A should create sent email");

            adapter_a
                .create_email_send_failure(CreateEmailSendFailure {
                    user_id: None,
                    recipient_email: "tenant-a@example.com".to_string(),
                    flow: "tenant_a_flow".to_string(),
                    provider: Some("tenant-a-provider".to_string()),
                    error_message: "smtp timeout".to_string(),
                    metadata: Some(serde_json::json!({ "tenant": "a" })),
                })
                .await
                .expect("tenant A should create failure");

            let template = adapter_a
                .create_email_template(CreateEmailTemplate {
                    template_key: "tenant_a_template".to_string(),
                    event_type: None,
                    locale: Some("en".to_string()),
                    subject_template: "Hello {{name}}".to_string(),
                    text_template: Some("Hi {{name}}".to_string()),
                    html_template: Some("<p>Hi {{name}}</p>".to_string()),
                    variables: Some(serde_json::json!(["name"])),
                    variable_bindings: None,
                    is_active: Some(true),
                    metadata: Some(serde_json::json!({ "tenant": "a" })),
                })
                .await
                .expect("tenant A should create template");

            let sent_a = adapter_a
                .list_sent_emails_with_params(ListSentEmailsParams {
                    flow: Some("tenant_a_flow".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant A should list sent emails");
            let sent_b = adapter_b
                .list_sent_emails_with_params(ListSentEmailsParams {
                    flow: Some("tenant_a_flow".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant B should list sent emails");
            assert_eq!(sent_a.len(), 1, "tenant A should see its sent email");
            assert_eq!(sent_b.len(), 0, "tenant B must not see tenant A sent email");

            let failures_a = adapter_a
                .list_email_send_failures(ListEmailSendFailuresParams {
                    flow: Some("tenant_a_flow".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant A should list failures");
            let failures_b = adapter_b
                .list_email_send_failures(ListEmailSendFailuresParams {
                    flow: Some("tenant_a_flow".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant B should list failures");
            assert_eq!(failures_a.len(), 1, "tenant A should see its failure");
            assert_eq!(
                failures_b.len(),
                0,
                "tenant B must not see tenant A failure"
            );

            let templates_a = adapter_a
                .list_email_templates(ListEmailTemplatesParams {
                    template_key: Some("tenant_a_template".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant A should list templates");
            let templates_b = adapter_b
                .list_email_templates(ListEmailTemplatesParams {
                    template_key: Some("tenant_a_template".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant B should list templates");
            assert_eq!(templates_a.len(), 1, "tenant A should see its template");
            assert_eq!(
                templates_b.len(),
                0,
                "tenant B must not see tenant A template"
            );

            let sent_id = sent_a[0].id.clone();
            let updated_sent = adapter_a
                .update_sent_email(
                    sent_id.as_str(),
                    UpdateSentEmail {
                        subject: Some("Tenant A Subject Updated".to_string()),
                        ..Default::default()
                    },
                )
                .await
                .expect("tenant A should update sent email");
            assert_eq!(updated_sent.subject, "Tenant A Subject Updated");
            let sent_in_b = adapter_b
                .get_sent_email_by_id(sent_id.as_str())
                .await
                .expect("tenant B lookup should succeed");
            assert!(
                sent_in_b.is_none(),
                "tenant B must not resolve tenant A sent email ID"
            );

            let failure_id = failures_a[0].id.clone();
            let updated_failure = adapter_a
                .update_email_send_failure(
                    failure_id.as_str(),
                    UpdateEmailSendFailure {
                        resolved: Some(true),
                        resolution_note: Some(Some("resolved".to_string())),
                        metadata: None,
                    },
                )
                .await
                .expect("tenant A should update failure");
            assert!(
                updated_failure.resolved,
                "failure should be marked resolved"
            );
            let failure_in_b = adapter_b
                .get_email_send_failure_by_id(failure_id.as_str())
                .await
                .expect("tenant B lookup should succeed");
            assert!(
                failure_in_b.is_none(),
                "tenant B must not resolve tenant A failure ID"
            );

            let updated_template = adapter_a
                .update_email_template(
                    template.id.as_str(),
                    UpdateEmailTemplate {
                        subject_template: Some("Hello Updated {{name}}".to_string()),
                        ..Default::default()
                    },
                )
                .await
                .expect("tenant A should update template");
            assert_eq!(
                updated_template.subject_template, "Hello Updated {{name}}",
                "tenant A template update should persist"
            );
            let templates_b_after = adapter_b
                .list_email_templates(ListEmailTemplatesParams {
                    template_key: Some("tenant_a_template".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant B template list should succeed");
            assert_eq!(
                templates_b_after.len(),
                0,
                "tenant B must still have no tenant A templates"
            );

            adapter_a
                .create_admin_audit_log(CreateAdminAuditLog {
                    actor_user_id: Some("tenant-a-admin".to_string()),
                    actor_session_token: Some("session-a".to_string()),
                    action: "admin.email.create".to_string(),
                    target_type: Some("email".to_string()),
                    target_id: Some(sent_id.clone()),
                    success: true,
                    error_message: None,
                    metadata: Some(serde_json::json!({ "tenant": "a", "op": "create" })),
                })
                .await
                .expect("tenant A should create admin audit log");

            let audit_a = adapter_a
                .list_admin_audit_logs_with_params(ListAdminAuditLogsParams {
                    action: Some("admin.email.create".to_string()),
                    target_type: Some("email".to_string()),
                    target_id: Some(sent_id.clone()),
                    success: Some(true),
                    ..Default::default()
                })
                .await
                .expect("tenant A audit list should succeed");
            let audit_b = adapter_b
                .list_admin_audit_logs_with_params(ListAdminAuditLogsParams {
                    action: Some("admin.email.create".to_string()),
                    target_type: Some("email".to_string()),
                    target_id: Some(sent_id.clone()),
                    success: Some(true),
                    ..Default::default()
                })
                .await
                .expect("tenant B audit list should succeed");
            assert_eq!(audit_a.len(), 1, "tenant A should see its admin audit log");
            assert_eq!(
                audit_b.len(),
                0,
                "tenant B must not see tenant A admin audit log"
            );

            let audit_a_total = adapter_a
                .count_admin_audit_logs(ListAdminAuditLogsParams {
                    action: Some("admin.email.create".to_string()),
                    target_type: Some("email".to_string()),
                    target_id: Some(sent_id.clone()),
                    success: Some(true),
                    ..Default::default()
                })
                .await
                .expect("tenant A audit count should succeed");
            let audit_b_total = adapter_b
                .count_admin_audit_logs(ListAdminAuditLogsParams {
                    action: Some("admin.email.create".to_string()),
                    target_type: Some("email".to_string()),
                    target_id: Some(sent_id.clone()),
                    success: Some(true),
                    ..Default::default()
                })
                .await
                .expect("tenant B audit count should succeed");
            assert_eq!(audit_a_total, 1, "tenant A audit count should match");
            assert_eq!(
                audit_b_total, 0,
                "tenant B audit count should remain isolated"
            );

            adapter_a
                .delete_sent_email(sent_id.as_str())
                .await
                .expect("tenant A should delete sent email");
            adapter_a
                .delete_email_send_failure(failure_id.as_str())
                .await
                .expect("tenant A should delete failure");
            adapter_a
                .delete_email_template(template.id.as_str())
                .await
                .expect("tenant A should delete template");

            let sent_a_after = adapter_a
                .list_sent_emails_with_params(ListSentEmailsParams {
                    flow: Some("tenant_a_flow".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant A final sent list should succeed");
            let failures_a_after = adapter_a
                .list_email_send_failures(ListEmailSendFailuresParams {
                    flow: Some("tenant_a_flow".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant A final failure list should succeed");
            let templates_a_after = adapter_a
                .list_email_templates(ListEmailTemplatesParams {
                    template_key: Some("tenant_a_template".to_string()),
                    ..Default::default()
                })
                .await
                .expect("tenant A final template list should succeed");

            assert!(
                sent_a_after.is_empty(),
                "tenant A sent emails should be empty"
            );
            assert!(
                failures_a_after.is_empty(),
                "tenant A failure records should be empty"
            );
            assert!(
                templates_a_after.is_empty(),
                "tenant A templates should be empty"
            );

            let drop_a_sql = format!("DROP SCHEMA IF EXISTS \"{schema_a}\" CASCADE");
            let drop_b_sql = format!("DROP SCHEMA IF EXISTS \"{schema_b}\" CASCADE");
            sqlx::query(drop_a_sql.as_str())
                .execute(&pool)
                .await
                .expect("should drop schema A");
            sqlx::query(drop_b_sql.as_str())
                .execute(&pool)
                .await
                .expect("should drop schema B");

            pool.close().await;
        }

        #[tokio::test]
        async fn sqlx_adapter_verification_templates_resolve_by_event_type_per_schema() {
            let Some(database_url) = std::env::var("TEST_DATABASE_URL")
                .ok()
                .map(|value| value.trim().to_string())
                .filter(|value| !value.is_empty())
            else {
                println!(
                    "Skipping verification template event-type schema test - TEST_DATABASE_URL not set"
                );
                return;
            };

            let pool = sqlx::PgPool::connect(database_url.as_str())
                .await
                .expect("should connect to test postgres");

            let nonce = sanitized_test_nonce();
            let schema_a = format!("tenant_event_a_{nonce}");
            let schema_b = format!("tenant_event_b_{nonce}");

            create_email_tables_for_schema(&pool, schema_a.as_str())
                .await
                .expect("should create schema A email tables");
            create_email_tables_for_schema(&pool, schema_b.as_str())
                .await
                .expect("should create schema B email tables");

            insert_event_type_for_schema(
                &pool,
                schema_a.as_str(),
                "user.email.verify",
                Some("verification_email"),
            )
            .await
            .expect("schema A should seed user.email.verify event type");
            insert_event_type_for_schema(
                &pool,
                schema_a.as_str(),
                "user.password.reset",
                Some("password_reset_email"),
            )
            .await
            .expect("schema A should seed user.password.reset event type");
            insert_event_type_for_schema(
                &pool,
                schema_b.as_str(),
                "user.email.verify",
                Some("verification_email"),
            )
            .await
            .expect("schema B should seed user.email.verify event type");

            let adapter_a: SqlxAdapter =
                SqlxAdapter::from_pool_with_schema(pool.clone(), schema_a.as_str());
            let adapter_b: SqlxAdapter =
                SqlxAdapter::from_pool_with_schema(pool.clone(), schema_b.as_str());

            adapter_a
                .create_email_template(CreateEmailTemplate {
                    template_key: "tenant_a_event_verification_template".to_string(),
                    event_type: Some("user.email.verify".to_string()),
                    locale: Some("en".to_string()),
                    subject_template: "Tenant A Event Subject".to_string(),
                    text_template: Some("Tenant A Event Text {{verification_url}}".to_string()),
                    html_template: Some(
                        "<p>Tenant A Event Html {{verification_url}}</p>".to_string(),
                    ),
                    variables: Some(serde_json::json!(["verification_url"])),
                    variable_bindings: None,
                    is_active: Some(true),
                    metadata: None,
                })
                .await
                .expect("schema A should create event-type template");

            adapter_a
                .create_email_template(CreateEmailTemplate {
                    template_key: "verification_email".to_string(),
                    event_type: Some("user.password.reset".to_string()),
                    locale: Some("en".to_string()),
                    subject_template: "Tenant A Key Fallback Subject".to_string(),
                    text_template: Some("fallback {{verification_url}}".to_string()),
                    html_template: Some("<p>fallback {{verification_url}}</p>".to_string()),
                    variables: Some(serde_json::json!(["verification_url"])),
                    variable_bindings: None,
                    is_active: Some(true),
                    metadata: None,
                })
                .await
                .expect("schema A should create template-key fallback template");

            adapter_b
                .create_email_template(CreateEmailTemplate {
                    template_key: "tenant_b_event_verification_template".to_string(),
                    event_type: Some("user.email.verify".to_string()),
                    locale: Some("en".to_string()),
                    subject_template: "Tenant B Event Subject".to_string(),
                    text_template: Some("Tenant B Event Text {{verification_url}}".to_string()),
                    html_template: Some(
                        "<p>Tenant B Event Html {{verification_url}}</p>".to_string(),
                    ),
                    variables: Some(serde_json::json!(["verification_url"])),
                    variable_bindings: None,
                    is_active: Some(true),
                    metadata: None,
                })
                .await
                .expect("schema B should create event-type template");

            let config = Arc::new(
                AuthConfig::new("sqlx-event-template-test-secret-key-at-least-32")
                    .app_name("Athena Auth"),
            );
            let ctx_a = AuthContext::new(config.clone(), Arc::new(adapter_a));
            let ctx_b = AuthContext::new(config, Arc::new(adapter_b));

            let rendered_a = verification_email_for_context(
                &ctx_a,
                "https://tenant-a.example/verify?token=a",
                crate::email::AuthEmailRenderContext::new(),
            )
            .await;
            assert_eq!(
                rendered_a.subject, "Tenant A Event Subject",
                "schema A should resolve its event-type template"
            );
            assert!(
                rendered_a.text.contains("Tenant A Event Text"),
                "schema A should render text from event-type template"
            );
            assert!(
                rendered_a.text.contains("tenant-a.example/verify?token=a"),
                "schema A custom template should receive verification_url variable"
            );
            assert!(
                !rendered_a.subject.contains("Fallback"),
                "event-type match should take precedence over template-key fallback"
            );

            let rendered_b = verification_email_for_context(
                &ctx_b,
                "https://tenant-b.example/verify?token=b",
                crate::email::AuthEmailRenderContext::new(),
            )
            .await;
            assert_eq!(
                rendered_b.subject, "Tenant B Event Subject",
                "schema B should resolve its own event-type template"
            );
            assert!(
                rendered_b.text.contains("Tenant B Event Text"),
                "schema B should render text from its own event-type template"
            );
            assert!(
                rendered_b.text.contains("tenant-b.example/verify?token=b"),
                "schema B custom template should receive verification_url variable"
            );

            let drop_a_sql = format!("DROP SCHEMA IF EXISTS \"{schema_a}\" CASCADE");
            let drop_b_sql = format!("DROP SCHEMA IF EXISTS \"{schema_b}\" CASCADE");
            sqlx::query(drop_a_sql.as_str())
                .execute(&pool)
                .await
                .expect("should drop schema A");
            sqlx::query(drop_b_sql.as_str())
                .execute(&pool)
                .await
                .expect("should drop schema B");

            pool.close().await;
        }
    }
}

#[cfg(feature = "sqlx-postgres")]
pub use sqlx_adapter::{SqlxAdapter, SqlxEntity};
