shared_framework/data/

repository.rs

//! Generic repository for SeaORM entities.
//!
//! [`PersistentRepository`] centralizes CRUD, counts, cursor-paginated reads,
//! updates, and deletes for one entity type whose model implements
//! [`BaseEntity`]. Queries are built with [`QueryData`] and parameterized with
//! [`RepositoryOptions`].
//!
//! Cursor filtering compares `id` against the decoded cursor: ascending order
//! uses `id > cursor`, descending uses `id < cursor`, and deletes always use
//! `id > cursor`. On creation, `uid`/`createdAt`/`updatedAt` (plus `createdById`/
//! `updatedById` when a user id is present) is filled in; on update,
//! `updatedAt` (plus `updatedById` when a user id is present) is refreshed.
//! Database columns for these fields use camelCase names.
//!
//! When a [`RepositoryOptions`] carries `Some(txn)` (see
//! [`RepositoryOptions::with_transaction`](RepositoryOptions::with_transaction)),
//! the operation runs against that transaction; otherwise it uses the
//! repository's own connection. The [`QueryData`] transaction (set via
//! `QueryData::with_options`) is honored as a fallback when the method-level
//! options carry none.
//!
//! Repositories can be shared through a process-wide registry keyed by entity
//! type (see [`PersistentRepository::initialize`] and
//! [`PersistentRepository::find`]), or used directly via
//! [`PersistentRepository::new`] without registration.
//!
//! ```ignore
//! use shared_framework::data::{PersistentRepository, QueryData, RepositoryOptions};
//!
//! let repo = PersistentRepository::<my_entity::Entity>::new(db);
//! let page = repo.get_paginated_view(None, RepositoryOptions::from_ctx(ctx).join(RepositoryOptions::new().with_limit(50))).await?;
//! let all = repo.get_all::<shared_framework::data::NoUser>(Some(QueryData::new(my_entity::Entity::find())), None).await?;
//! ```

use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::fmt::Display;
use std::str::FromStr;
use std::sync::{Arc, Mutex, OnceLock};

use crate::data::query::{
    CursorData, DeleteQueryData, NoUser, PageResult, QueryData, RepositoryOptions,
};
use crate::data::types_extra::ChangeResultModel;
use crate::data::{BaseAuditableEntity, BaseEntity};
use chrono::Utc;
use sea_orm::entity::prelude::DateTimeWithTimeZone;
use sea_orm::sea_query::{Alias, Expr, ExprTrait};
use sea_orm::{
    ActiveModelTrait, ColumnTrait, DatabaseConnection, DatabaseTransaction, DbErr, EntityTrait,
    IntoActiveModel, Iterable, ModelTrait, PaginatorTrait, PrimaryKeyToColumn, PrimaryKeyTrait,
    QueryFilter, QuerySelect, TransactionTrait, Value, entity::EntityLoaderTrait,
};
use serde::Serialize;
use uuid::Uuid;

// ── Singleton registry ───────────────────────────────────────────────────────

static REGISTRY: OnceLock<Mutex<HashMap<TypeId, Arc<dyn Any + Send + Sync>>>> = OnceLock::new();

fn registry() -> &'static Mutex<HashMap<TypeId, Arc<dyn Any + Send + Sync>>> {
    REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}

// ── cursor helpers (asc → gt, desc → lt; deletes always gt) ────────────────
// Select: respects order (asc → gt, desc → lt). Delete: always gt.

fn with_cursor<E>(
    select: sea_orm::Select<E>,
    cursor: &Option<String>,
    order_asc: Option<bool>,
) -> sea_orm::Select<E>
where
    E: EntityTrait,
{
    if let Some(s) = cursor {
        if let Some(data) = CursorData::decode(s) {
            let is_asc = order_asc.unwrap_or(true);
            return if is_asc {
                select.filter(Expr::col(Alias::new("id")).gt(data.cursor))
            } else {
                select.filter(Expr::col(Alias::new("id")).lt(data.cursor))
            };
        }
    }
    select
}

fn with_cursor_delete<E>(
    delete: sea_orm::DeleteMany<E>,
    cursor: &Option<String>,
) -> sea_orm::DeleteMany<E>
where
    E: EntityTrait,
{
    if let Some(s) = cursor {
        if let Some(data) = CursorData::decode(s) {
            return delete.filter(Expr::col(Alias::new("id")).gt(data.cursor));
        }
    }
    delete
}

fn with_cursor_loader<E, L>(loader: L, cursor: &Option<String>, order_asc: Option<bool>) -> L
where
    E: EntityTrait,
    L: EntityLoaderTrait<E>,
{
    if let Some(s) = cursor {
        if let Some(data) = CursorData::decode(s) {
            let is_asc = order_asc.unwrap_or(true);
            return if is_asc {
                loader.filter(Expr::col(Alias::new("id")).gt(data.cursor))
            } else {
                loader.filter(Expr::col(Alias::new("id")).lt(data.cursor))
            };
        }
    }
    loader
}

fn model_id<E, M>(model: &M) -> i64
where
    E: EntityTrait,
    M: ModelTrait<Entity = E>,
{
    match model.get(E::PrimaryKey::iter().next().unwrap().into_column()) {
        Value::BigInt(Some(id)) => id,
        Value::Int(Some(id)) => i64::from(id),
        value => panic!("repository cursor requires an integer primary key, got {value:?}"),
    }
}

// ── transaction-executor helpers ───────────────────────────────────────────
// When a `RepositoryOptions` carries `Some(txn)`, operations run against that
// transaction; otherwise they fall back to the repository connection.

fn opts_txn<U>(opts: Option<&RepositoryOptions<U>>) -> Option<Arc<DatabaseTransaction>>
where
    U: BaseEntity + Clone + Send + Sync + 'static,
{
    opts.and_then(|o| o.txn.clone())
}

/// Prefers the explicit method-level `opts` transaction, falling back to the
/// transaction attached to the [`QueryData`] options (populated via
/// `QueryData::with_options`).
fn pick_txn<U>(
    opts: Option<&RepositoryOptions<U>>,
    qd_opts_txn: Option<Arc<DatabaseTransaction>>,
) -> Option<Arc<DatabaseTransaction>>
where
    U: BaseEntity + Clone + Send + Sync + 'static,
{
    opts_txn(opts).or(qd_opts_txn)
}

// ── base-field helpers (uid, timestamps, actor ids) ────────────────────────

fn try_set<A>(active: &mut A, col_name: &str, value: Value)
where
    A: ActiveModelTrait,
    A::Entity: EntityTrait,
    <A::Entity as EntityTrait>::Column: ColumnTrait + FromStr,
    <<A::Entity as EntityTrait>::Column as FromStr>::Err: std::fmt::Debug,
{
    if let Ok(col) = <A::Entity as EntityTrait>::Column::from_str(col_name) {
        active.set(col, value);
    }
}

fn apply_base_create<A, U>(active: &mut A, opts: &RepositoryOptions<U>)
where
    A: ActiveModelTrait,
    A::Entity: EntityTrait,
    <A::Entity as EntityTrait>::Column: ColumnTrait + FromStr,
    <<A::Entity as EntityTrait>::Column as FromStr>::Err: std::fmt::Debug,
    U: BaseEntity + Clone + Send + Sync + 'static,
{
    let now: DateTimeWithTimeZone = Utc::now().into();
    let uid = Uuid::new_v4();
    // All database columns are camelCase — single canonical name, no fallbacks
    try_set(active, "uid", Value::Uuid(Some(uid)));
    try_set(active, "createdAt", Value::ChronoDateTimeWithTimeZone(Some(now)));
    try_set(active, "updatedAt", Value::ChronoDateTimeWithTimeZone(Some(now)));
    let user_id_opt = opts.user_id.or_else(|| opts.user.as_ref().map(|u| u.id()));
    if let Some(uid) = user_id_opt {
        try_set(active, "createdById", Value::BigInt(Some(uid)));
        try_set(active, "updatedById", Value::BigInt(Some(uid)));
    }
}

fn apply_base_update<A, U>(active: &mut A, opts: &RepositoryOptions<U>)
where
    A: ActiveModelTrait,
    A::Entity: EntityTrait,
    <A::Entity as EntityTrait>::Column: ColumnTrait + FromStr,
    <<A::Entity as EntityTrait>::Column as FromStr>::Err: std::fmt::Debug,
    U: BaseEntity + Clone + Send + Sync + 'static,
{
    let now: DateTimeWithTimeZone = Utc::now().into();
    try_set(active, "updatedAt", Value::ChronoDateTimeWithTimeZone(Some(now)));
    let user_id_opt = opts.user_id.or_else(|| opts.user.as_ref().map(|u| u.id()));
    if let Some(uid) = user_id_opt {
        try_set(active, "updatedById", Value::BigInt(Some(uid)));
    }
}

// ── Repository ───────────────────────────────────────────────────────────────

/// Generic database access for one SeaORM entity.
///
/// `E` is the SeaORM entity; its model must implement [`BaseEntity`].
/// Holds a cloned [`DatabaseConnection`] used by every operation.
#[derive(Clone)]
pub struct PersistentRepository<E>
where
    E: EntityTrait,
    E::ModelEx: BaseEntity + Send + Sync,
    <E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
{
    /// Connection used by every operation.
    pub db: DatabaseConnection,
    _marker: std::marker::PhantomData<E>,
}

impl<E> PersistentRepository<E>
where
    E: EntityTrait,
    E::ModelEx: BaseEntity + Send + Sync + Clone + Serialize,
    E::Model: Into<E::ModelEx> + Send + Sync,
    <E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
    E::Column: ColumnTrait + FromStr,
    <E::Column as FromStr>::Err: std::fmt::Debug,
{
    /// Creates a repository holding the given connection.
    pub fn new(db: DatabaseConnection) -> Self {
        Self {
            db,
            _marker: std::marker::PhantomData,
        }
    }

    /// Returns the underlying connection.
    pub fn db(&self) -> &DatabaseConnection {
        &self.db
    }

    // ── Singleton registry ───────────────────────────────────────────────

    /// Registers a repository for `E` in the process-wide registry and returns it.
    ///
    /// Panics if a repository for `E` was already initialized.
    pub fn initialize(db: &DatabaseConnection) -> Arc<Self> {
        let type_id = TypeId::of::<E>();
        let mut reg = registry().lock().unwrap();
        if reg.contains_key(&type_id) {
            panic!(
                "Repository for {} has already been initialized",
                std::any::type_name::<E>()
            );
        }
        let repo = Arc::new(Self::new(db.clone()));
        reg.insert(type_id, repo.clone() as Arc<dyn Any + Send + Sync>);
        repo
    }

    /// Returns the registered repository for `E`.
    ///
    /// Panics if [`PersistentRepository::initialize`] was not called first.
    pub fn find() -> Arc<Self> {
        Self::try_find().unwrap_or_else(|| panic!("Repository for {} has not been initialized — call PersistentRepository::<{}>::initialize(db) first", std::any::type_name::<E>(), std::any::type_name::<E>()))
    }

    /// Returns the registered repository for `E`, or `None` when uninitialized.
    pub fn try_find() -> Option<Arc<Self>> {
        let reg = registry().lock().unwrap();
        reg.get(&TypeId::of::<E>())
            .and_then(|arc| Arc::downcast::<Self>(arc.clone()).ok())
    }

    /// Creates a repository without touching the shared registry.
    pub fn new_ephemeral(db: DatabaseConnection) -> Self {
        Self::new(db)
    }

    /// Starts a fresh [`QueryData`] over `E::find()`.
    pub fn start(&self) -> QueryData<E>
    where
        E::ModelEx: BaseEntity,
    {
        QueryData::new(<E::ModelEx as BaseEntity>::load())
    }

    /// Starts a fresh [`DeleteQueryData`] over `E::delete_many()`.
    pub fn start_delete(&self) -> DeleteQueryData<E> {
        DeleteQueryData::new(E::delete_many())
    }

    /// Returns true when the model and user share the same `id`.
    pub fn is_owner<U>(&self, model: &E::ModelEx, user: &U) -> bool
    where
        U: BaseEntity,
    {
        model.id() == user.id()
    }
    /// Returns true when the model's creator id matches the user id.
    pub fn is_owner_auditable<U>(&self, model: &E::ModelEx, user: &U) -> bool
    where
        U: BaseEntity + Clone + Send + Sync,
        E::ModelEx: BaseAuditableEntity<User = U>,
    {
        model.created_by().id() == user.id()
    }
    /// Returns true when the user may access the model; currently same as [`PersistentRepository::is_owner`].
    pub fn is_accessible<U>(&self, model: &E::ModelEx, user: &U) -> bool
    where
        U: BaseEntity,
    {
        self.is_owner(model, user)
    }
    /// Returns true when the user created or last updated the model.
    pub fn is_accessible_auditable<U>(&self, model: &E::ModelEx, user: &U) -> bool
    where
        U: BaseEntity + Clone + Send + Sync,
        E::ModelEx: BaseAuditableEntity<User = U>,
    {
        model.created_by().id() == user.id() || model.updated_by().id() == user.id()
    }

    /// Runs the closure inside a transaction, committing on success.
    ///
    /// The closure receives an `Arc<DatabaseTransaction>` so the handle can be
    /// freely shared — e.g., moved into
    /// [`RepositoryOptions::with_transaction`](RepositoryOptions::with_transaction)
    /// for nested repository calls. Do not retain clones beyond the closure:
    /// committing requires sole ownership of the handle, and a closure error
    /// drops it uncommitted, rolling the transaction back.
    ///
    /// Returns the closure value or the [`DbErr`] from beginning, the closure, or commit.
    pub async fn transaction<F, T, EType>(&self, f: F) -> Result<T, EType>
    where
        EType: From<DbErr> + Display + Send,
        F: FnOnce(
                Arc<DatabaseTransaction>,
            ) -> futures::future::BoxFuture<'static, Result<T, EType>>
            + Send,
        T: Send,
    {
        let txn = Arc::new(self.db.begin().await?);
        let res = f(txn.clone()).await?;
        let owned = Arc::try_unwrap(txn).map_err(|_| {
            DbErr::Custom(
                "transaction handle still shared after closure; refusing to commit".to_string(),
            )
        })?;
        owned.commit().await?;
        Ok(res)
    }

    /// Runs the closure inside `opts.txn` when one is attached, else in a fresh transaction.
    ///
    /// When `opts` carries a transaction, the closure receives that shared
    /// handle directly and no commit/rollback is performed here — the owner of
    /// that transaction controls its lifecycle. Otherwise a new transaction is
    /// begun from the repository connection and committed on success.
    pub async fn transaction_with_opts<U, F, T, EType>(
        &self,
        opts: RepositoryOptions<U>,
        f: F,
    ) -> Result<T, EType>
    where
        EType: From<DbErr> + Display + Send,
        U: BaseEntity + Clone + Send + Sync + 'static,
        F: FnOnce(
                Arc<DatabaseTransaction>,
            ) -> futures::future::BoxFuture<'static, Result<T, EType>>
            + Send,
        T: Send,
    {
        if let Some(txn) = opts.txn {
            return f(txn).await;
        }
        self.transaction(f).await
    }

    // ── COUNT ───────────────────────────────────────────────────────────────

    /// Counts rows matching the filter, applying the cursor when one is set.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. The cursor comes from
    /// `opts` when present, else from the [`QueryData`] options. Runs against
    /// `opts.txn` when attached, else against the repository connection.
    pub async fn get_count<U>(
        &self,
        filter: Option<QueryData<E>>,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<u64, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: BaseEntity,
    {
        let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
        let txn = pick_txn(
            opts.as_ref(),
            filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
        );
        let (loader, query_cursor, order) = match filter {
            Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
            None => (<E::ModelEx as BaseEntity>::load(), None, None),
        };
        let loader = with_cursor_loader(loader, &cursor.or(query_cursor), order);
        if let Some(txn) = txn.as_ref() {
            loader.num_items(txn.as_ref(), 0).await
        } else {
            loader.num_items(&self.db, 0).await
        }
    }

    /// Counts rows matching the filter inside the given transaction.
    pub async fn get_count_txn(
        &self,
        filter: Option<QueryData<E>>,
        txn: &DatabaseTransaction,
    ) -> Result<u64, DbErr>
    where
        E::ModelEx: BaseEntity,
    {
        let (loader, cursor, order) = match filter {
            Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
            None => (<E::ModelEx as BaseEntity>::load(), None, None),
        };
        with_cursor_loader(loader, &cursor, order)
            .num_items(txn, 0)
            .await
    }

    /// Counts rows matching the filter with default options.
    pub async fn count(&self, filter: Option<QueryData<E>>) -> Result<u64, DbErr>
    where
        E::ModelEx: BaseEntity,
    {
        self.get_count::<NoUser>(filter, None).await
    }

    // ── PAGINATED ───────────────────────────────────────────────────────────

    /// Returns one cursor page: totals ignore cursor/limit, rows fetch `limit + 1` to detect `has_next`.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. The next-page cursor encodes
    /// `limit` and the last row `id`; it is `None` when there is no next page.
    /// Runs against `opts.txn` when attached, else against the repository connection.
    pub async fn get_paginated_view<U>(
        &self,
        filter: Option<QueryData<E>>,
        opts: RepositoryOptions<U>,
    ) -> Result<PageResult<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: BaseEntity,
    {
        let order_asc = filter.as_ref().and_then(|qd| qd.order_asc);
        let txn = opts
            .txn
            .clone()
            .or_else(|| filter.as_ref().and_then(|qd| qd.opts.txn.clone()));
        let loader = filter
            .map(|qd| qd.loader)
            .unwrap_or_else(<E::ModelEx as BaseEntity>::load);
        let total_loader = loader.clone();
        let total = if let Some(txn) = txn.as_ref() {
            total_loader.num_items(txn.as_ref(), 0).await? as i64
        } else {
            total_loader.num_items(&self.db, 0).await? as i64
        };

        let sel = with_cursor_loader(loader, &opts.cursor, order_asc);
        let limit = opts.limit;
        let rows: Vec<E::ModelEx> = if let Some(txn) = txn.as_ref() {
            sel.fetch(txn.as_ref(), 0, (limit + 1) as u64).await?
        } else {
            sel.fetch(&self.db, 0, (limit + 1) as u64).await?
        };
        let has_next = rows.len() > limit;
        let mut data = rows;
        if has_next {
            data.truncate(limit);
        }
        let next_cursor = if has_next {
            Some(
                CursorData {
                    limit,
                    cursor: model_id::<E, _>(data.last().unwrap()),
                }
                .encode(),
            )
        } else {
            None
        };
        Ok(PageResult {
            data,
            has_next,
            cursor: opts.cursor.clone(),
            next_cursor,
            total,
            limit,
        })
    }

    /// Paginates a raw `Select`: totals ignore cursor/limit and cursors use `id > cursor`.
    ///
    /// Unlike [`PersistentRepository::get_paginated_view`], no recorded ordering
    /// direction is consulted because the select carries none.
    /// Runs against `opts.txn` when attached, else against the repository connection.
    pub async fn paginated(
        &self,
        select: sea_orm::Select<E>,
        opts: RepositoryOptions<NoUser>,
    ) -> Result<PageResult<E::ModelEx>, DbErr>
    where
        E::ModelEx: BaseEntity,
        E::Model: Into<E::ModelEx> + Send + Sync,
    {
        let limit = opts.limit;
        let cursor = opts.cursor.clone();
        let txn = opts.txn.clone();
        // total without cursor/limit.
        let mut count_sel = select.clone();
        if opts.distinct {
            count_sel = count_sel.distinct();
        }
        let total = if let Some(txn) = txn.as_ref() {
            count_sel.paginate(txn.as_ref(), 1).num_items().await? as i64
        } else {
            count_sel.paginate(&self.db, 1).num_items().await? as i64
        };

        let mut sel = with_cursor(select, &cursor, None);
        if opts.distinct {
            sel = sel.distinct();
        }
        sel = sel.limit((limit + 1) as u64);
        let rows: Vec<E::ModelEx> = if let Some(txn) = txn.as_ref() {
            sel.all(txn.as_ref())
                .await?
                .into_iter()
                .map(Into::into)
                .collect()
        } else {
            sel.all(&self.db)
                .await?
                .into_iter()
                .map(Into::into)
                .collect()
        };
        let has_next = rows.len() > limit;
        let mut data = rows;
        if has_next {
            data.truncate(limit);
        }
        let next_cursor = if has_next {
            Some(
                CursorData {
                    limit,
                    cursor: model_id::<E, _>(data.last().unwrap()),
                }
                .encode(),
            )
        } else {
            None
        };
        Ok(PageResult {
            data,
            has_next,
            cursor: opts.cursor,
            next_cursor,
            total,
            limit: opts.limit,
        })
    }

    // ── GET MANY / ALL / DISTINCT / ONE / BY_ID ────────────────────────────

    /// Returns all matching rows, applying cursor (`opts` cursor wins over [`QueryData`] cursor) and distinct.
    ///
    /// `U` is the [`RepositoryOptions`] identity type.
    /// Runs against `opts.txn` (or the [`QueryData`] transaction) when attached,
    /// else against the repository connection.
    pub async fn get_all<U>(
        &self,
        filter: Option<QueryData<E>>,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<Vec<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: BaseEntity,
    {
        let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
        let txn = pick_txn(
            opts.as_ref(),
            filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
        );
        let (loader, qd_cursor, order) = match filter {
            Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
            None => (<E::ModelEx as BaseEntity>::load(), None, None),
        };
        let cursor = cursor.or(qd_cursor);
        let loader = with_cursor_loader(loader, &cursor, order);
        if let Some(txn) = txn.as_ref() {
            loader.fetch(txn.as_ref(), 0, 0).await
        } else {
            loader.fetch(&self.db, 0, 0).await
        }
    }

    /// Returns up to `opts.limit` matching rows, applying cursor and distinct.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. The `opts` cursor is
    /// preferred when set, else the [`QueryData`] cursor is used.
    /// Runs against `opts.txn` (or the [`QueryData`] transaction) when attached.
    pub async fn get_many<U>(
        &self,
        filter: Option<QueryData<E>>,
        opts: RepositoryOptions<U>,
    ) -> Result<Vec<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: BaseEntity,
    {
        let cursor = opts.cursor.clone();
        let txn = opts
            .txn
            .clone()
            .or_else(|| filter.as_ref().and_then(|qd| qd.opts.txn.clone()));
        let (loader, qd_cursor, order) = match filter {
            Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
            None => (<E::ModelEx as BaseEntity>::load(), None, None),
        };
        let cursor = if cursor.is_some() { cursor } else { qd_cursor };
        let loader = with_cursor_loader(loader, &cursor, order);
        if let Some(txn) = txn.as_ref() {
            loader.fetch(txn.as_ref(), 0, opts.limit as u64).await
        } else {
            loader.fetch(&self.db, 0, opts.limit as u64).await
        }
    }

    /// Alias for [`PersistentRepository::get_many`].
    pub async fn find_many<U>(
        &self,
        filter: Option<QueryData<E>>,
        opts: RepositoryOptions<U>,
    ) -> Result<Vec<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: BaseEntity,
    {
        self.get_many(filter, opts).await
    }

    /// Returns all matching rows with default options.
    pub async fn find_all(&self, filter: Option<QueryData<E>>) -> Result<Vec<E::ModelEx>, DbErr>
    where
        E::ModelEx: BaseEntity,
    {
        self.get_all::<NoUser>(filter, None).await
    }

    /// Returns up to `opts.limit` distinct rows matching the filter, applying cursor.
    ///
    /// `U` is the [`RepositoryOptions`] identity type.
    /// Runs against `opts.txn` (or the [`QueryData`] transaction) when attached.
    pub async fn get_distinct_rows<U>(
        &self,
        filter: Option<QueryData<E>>,
        opts: RepositoryOptions<U>,
    ) -> Result<Vec<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: BaseEntity,
    {
        let cursor = opts.cursor.clone();
        let txn = opts
            .txn
            .clone()
            .or_else(|| filter.as_ref().and_then(|qd| qd.opts.txn.clone()));
        let (loader, qd_cursor, order) = match filter {
            Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
            None => (<E::ModelEx as BaseEntity>::load(), None, None),
        };
        let loader = with_cursor_loader(loader, &cursor.or(qd_cursor), order);
        if let Some(txn) = txn.as_ref() {
            loader.fetch(txn.as_ref(), 0, opts.limit as u64).await
        } else {
            loader.fetch(&self.db, 0, opts.limit as u64).await
        }
    }

    /// Returns the first matching row, if any, applying cursor when one is set.
    ///
    /// `U` is the [`RepositoryOptions`] identity type.
    /// Runs against `opts.txn` (or the [`QueryData`] transaction) when attached.
    pub async fn get_one<U>(
        &self,
        filter: Option<QueryData<E>>,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: BaseEntity,
    {
        let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
        let txn = pick_txn(
            opts.as_ref(),
            filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
        );
        let (loader, qd_cursor, order) = match filter {
            Some(qd) => (qd.loader, qd.opts.cursor, qd.order_asc),
            None => (<E::ModelEx as BaseEntity>::load(), None, None),
        };
        let loader = with_cursor_loader(loader, &cursor.or(qd_cursor), order);
        if let Some(txn) = txn.as_ref() {
            Ok(loader.fetch(txn.as_ref(), 0, 1).await?.into_iter().next())
        } else {
            Ok(loader.fetch(&self.db, 0, 1).await?.into_iter().next())
        }
    }

    /// Returns the first matching row with default options.
    pub async fn find_one(&self, filter: Option<QueryData<E>>) -> Result<Option<E::ModelEx>, DbErr>
    where
        E::ModelEx: BaseEntity,
    {
        self.get_one::<NoUser>(filter, None).await
    }

    /// Finds a row by primary key converted from `id`; returns `None` when absent.
    pub async fn get_by_id(&self, id: i64) -> Result<Option<E::ModelEx>, DbErr>
    where
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::ModelEx: BaseEntity,
        <E::ModelEx as BaseEntity>::LoaderType: EntityLoaderTrait<E, ModelEx = E::ModelEx>,
    {
        <E::ModelEx as BaseEntity>::load()
            .filter_by_id(id)
            .fetch(&self.db, 0, 1)
            .await
            .map(|mut rows| rows.pop())
    }

    /// Alias for [`PersistentRepository::get_by_id`].
    pub async fn find_by_id(&self, id: i64) -> Result<Option<E::ModelEx>, DbErr>
    where
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::ModelEx: BaseEntity,
    {
        self.get_by_id(id).await
    }

    /// Finds a row by id; runs inside `opts.txn` when one is attached.
    ///
    /// `U` is the [`RepositoryOptions`] identity type.
    pub async fn get_by_id_with_opts<U>(
        &self,
        id: i64,
        opts: RepositoryOptions<U>,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::ModelEx: BaseEntity,
    {
        let loader = <E::ModelEx as BaseEntity>::load().filter_by_id(id);
        if let Some(txn) = opts.txn.as_ref() {
            Ok(loader.fetch(txn.as_ref(), 0, 1).await?.into_iter().next())
        } else {
            Ok(loader.fetch(&self.db, 0, 1).await?.into_iter().next())
        }
    }

    // ── CREATE ──────────────────────────────────────────────────────────────

    /// Inserts one row, filling `uid`/`createdAt`/`updatedAt` (and audit ids when a user id is present).
    ///
    /// `A` is the SeaORM active model for `E`; `U` is the [`RepositoryOptions`] identity type.
    /// Inserts inside `opts.txn` when attached, else on the repository connection.
    pub async fn create_one<A, U>(
        &self,
        mut active: A,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<E::ModelEx, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::Model: IntoActiveModel<A>,
        E::ModelEx: BaseEntity,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        if let Some(o) = opts.as_ref() {
            apply_base_create(&mut active, o);
        } else {
            let empty: RepositoryOptions<U> = RepositoryOptions::default();
            apply_base_create(&mut active, &empty);
        }
        if let Some(txn) = opts_txn(opts.as_ref()).as_ref() {
            active.insert(txn.as_ref()).await.map(Into::into)
        } else {
            active.insert(&self.db).await.map(Into::into)
        }
    }

    // keep legacy no-opts overload
    /// Inserts one row with default options.
    pub async fn create_one_simple<A>(&self, active: A) -> Result<E::ModelEx, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::Model: IntoActiveModel<A>,
        E::ModelEx: BaseEntity,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        self.create_one::<A, NoUser>(active, None).await
    }

    /// Inserts one row with the given options.
    ///
    /// `A` is the SeaORM active model for `E`; `U` is the [`RepositoryOptions`] identity type.
    pub async fn create_one_with_opts<A, U>(
        &self,
        active: A,
        opts: RepositoryOptions<U>,
    ) -> Result<E::ModelEx, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::Model: IntoActiveModel<A>,
        E::ModelEx: BaseEntity,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        self.create_one(active, Some(opts)).await
    }

    /// Inserts each row one at a time, applying base/audit fields to every row.
    ///
    /// `A` is the SeaORM active model for `E`; `U` is the [`RepositoryOptions`] identity type.
    /// Inserts inside `opts.txn` when attached, else on the repository connection.
    pub async fn create_many<A, U>(
        &self,
        mut actives: Vec<A>,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<Vec<E::ModelEx>, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::ModelEx: BaseEntity,
        E::Model: IntoActiveModel<A>,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let mut out = Vec::with_capacity(actives.len());
        for active in &mut actives {
            if let Some(o) = opts.as_ref() {
                apply_base_create(active, o);
            } else {
                let empty: RepositoryOptions<U> = RepositoryOptions::default();
                apply_base_create(active, &empty);
            }
        }
        let txn = opts_txn(opts.as_ref());
        for a in actives {
            if let Some(txn) = txn.as_ref() {
                out.push(a.insert(txn.as_ref()).await?.into());
            } else {
                out.push(a.insert(&self.db).await?.into());
            }
        }
        Ok(out)
    }

    /// Inserts many rows with default options.
    pub async fn create_many_simple<A>(&self, actives: Vec<A>) -> Result<Vec<E::ModelEx>, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::ModelEx: BaseEntity,
        E::Model: IntoActiveModel<A>,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        self.create_many::<A, NoUser>(actives, None).await
    }

    /// Inserts many rows with the given options.
    pub async fn create_many_with_opts<A, U>(
        &self,
        actives: Vec<A>,
        opts: RepositoryOptions<U>,
    ) -> Result<Vec<E::ModelEx>, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::ModelEx: BaseEntity,
        E::Model: IntoActiveModel<A>,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        self.create_many(actives, Some(opts)).await
    }

    // ── UPSERT ──────────────────────────────────────────────────────────────

    /// Inserts one row with fresh base fields; does not check for an existing id.
    pub async fn upsert_one<A>(&self, mut active: A) -> Result<E::ModelEx, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::Model: IntoActiveModel<A> + Clone,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
        apply_base_create(&mut active, &empty);
        active.insert(&self.db).await.map(Into::into)
    }

    /// Inserts each row via [`PersistentRepository::upsert_one`], one at a time.
    pub async fn upsert_many<A>(&self, actives: Vec<A>) -> Result<Vec<E::ModelEx>, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::ModelEx: BaseEntity,
        E::Model: IntoActiveModel<A> + Clone,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let mut out = Vec::with_capacity(actives.len());
        for a in actives {
            out.push(self.upsert_one(a).await?);
        }
        Ok(out)
    }

    /// Updates the row when `id_fn` yields an existing id, else inserts it.
    ///
    /// Updates refresh `updatedAt`; inserts fill the full base field set.
    pub async fn upsert_one_with_id<A, F>(
        &self,
        mut active: A,
        id_fn: F,
    ) -> Result<E::ModelEx, DbErr>
    where
        A: ActiveModelTrait<Entity = E> + Send,
        A: sea_orm::ActiveModelBehavior + Send,
        E::Model: IntoActiveModel<A> + Clone + BaseEntity,
        F: Fn(&A) -> Option<i64> + Send,
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        if let Some(id) = id_fn(&active) {
            if self.get_by_id(id).await?.is_some() {
                // for update, set updated_at/updated_by
                let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
                apply_base_update(&mut active, &empty);
                return active.update(&self.db).await.map(Into::into);
            }
        }
        let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
        apply_base_create(&mut active, &empty);
        active.insert(&self.db).await.map(Into::into)
    }

    // ── UPDATE ──────────────────────────────────────────────────────────────

    /// Loads rows via [`PersistentRepository::get_many`], applies `effector`, and saves changed rows.
    ///
    /// The effector mutates each model and returns true to persist it; untouched
    /// rows are skipped. Returns the pre-save models that were persisted.
    /// `U` is the [`RepositoryOptions`] identity type used for audit fields.
    /// Reads and writes run inside `opts.txn` when attached.
    pub async fn update_many<F, U>(
        &self,
        filter: Option<QueryData<E>>,
        mut effector: F,
        opts: RepositoryOptions<U>,
    ) -> Result<ChangeResultModel<E::ModelEx>, DbErr>
    where
        F: FnMut(&mut E::Model) -> bool + Send,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: Into<E::Model> + BaseEntity,
        E::Model: IntoActiveModel<E::ActiveModel> + Clone,
        E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let txn = opts.txn.clone();
        let items = self.get_many(filter, opts.clone()).await?;
        let mut affected = Vec::new();
        for model in items {
            let mut model: E::Model = model.into();
            if effector(&mut model) {
                let mut am: E::ActiveModel = model.clone().into_active_model();
                apply_base_update(&mut am, &opts);
                if let Some(txn) = txn.as_ref() {
                    am.update(txn.as_ref()).await?;
                } else {
                    am.update(&self.db).await?;
                }
                affected.push(model.into());
            }
        }
        Ok(ChangeResultModel::new(affected))
    }

    /// Alias for [`PersistentRepository::update_many`].
    pub async fn find_update_many<F, U>(
        &self,
        filter: Option<QueryData<E>>,
        effector: F,
        opts: RepositoryOptions<U>,
    ) -> Result<ChangeResultModel<E::ModelEx>, DbErr>
    where
        F: FnMut(&mut E::Model) -> bool + Send,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: Into<E::Model> + BaseEntity,
        E::Model: IntoActiveModel<E::ActiveModel> + Clone,
        E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        self.update_many(filter, effector, opts).await
    }

    /// Loads the first matching row, applies `effector`, and saves it when changed.
    ///
    /// Returns `None` when no row matches; returns the row unchanged when the
    /// effector returns false.
    pub async fn update_one<F>(
        &self,
        filter: Option<QueryData<E>>,
        mut effector: F,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        F: FnMut(&mut E::Model) -> bool + Send,
        E::ModelEx: Into<E::Model> + BaseEntity,
        E::Model: IntoActiveModel<E::ActiveModel> + Clone ,
        E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let Some(model) = self.get_one::<NoUser>(filter, None).await? else {
            return Ok(None);
        };
        let mut model: E::Model = model.into();
        if !effector(&mut model) {
            return Ok(Some(model.into()));
        }
        let mut am: E::ActiveModel = model.clone().into_active_model();
        let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
        apply_base_update(&mut am, &empty);
        let updated = am.update(&self.db).await?;
        Ok(Some(updated.into()))
    }

    /// Loads a row by id, applies `effector` to the model, and saves it when changed.
    ///
    /// Returns `None` when the id does not exist.
    pub async fn update_by_id<F>(
        &self,
        id: i64,
        mut effector: F,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        F: FnMut(&mut E::Model) -> bool + Send,
        E::ModelEx: Into<E::Model> + BaseEntity,
        E::Model: IntoActiveModel<E::ActiveModel> + Clone,
        E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let Some(model) = self.get_by_id(id).await? else {
            return Ok(None);
        };
        let mut model: E::Model = model.into();
        if !effector(&mut model) {
            return Ok(Some(model.into()));
        }
        let mut am: E::ActiveModel = model.clone().into_active_model();
        let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
        apply_base_update(&mut am, &empty);
        let updated = am.update(&self.db).await?;
        Ok(Some(updated.into()))
    }

    /// Loads a row by id, applies `effector` to the active model, and saves it when changed.
    ///
    /// Returns the refetched row unchanged when the effector returns false,
    /// or `None` when the id does not exist.
    pub async fn update_by_id_with_active<F>(
        &self,
        id: i64,
        mut effector: F,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        F: FnMut(&mut E::ActiveModel) -> bool + Send,
        E::ModelEx: Into<E::Model> + BaseEntity,
        E::Model: IntoActiveModel<E::ActiveModel> + Clone,
        E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let Some(model) = self.get_by_id(id).await? else {
            return Ok(None);
        };
        let model: E::Model = model.into();
        let mut am: E::ActiveModel = model.into_active_model();
        if !effector(&mut am) {
            return self.get_by_id(id).await;
        }
        let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
        apply_base_update(&mut am, &empty);
        let updated = am.update(&self.db).await?;
        Ok(Some(updated.into()))
    }

    /// Loads the first matching row and updates it, recording audit fields from `opts` when present.
    ///
    /// `U` is the [`RepositoryOptions`] identity type.
    /// Reads and writes run inside `opts.txn` (or the filter transaction) when attached.
    pub async fn update_one_with_opts<F, U>(
        &self,
        filter: Option<QueryData<E>>,
        mut effector: F,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        F: FnMut(&mut E::Model) -> bool + Send,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: Into<E::Model>,
        E::Model: IntoActiveModel<E::ActiveModel> + Clone,
        E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let txn = pick_txn(
            opts.as_ref(),
            filter.as_ref().and_then(|qd| qd.opts.txn.clone()),
        );
        let Some(model) = self.get_one(filter, opts.clone()).await? else {
            return Ok(None);
        };
        let mut model: E::Model = model.into();
        if !effector(&mut model) {
            return Ok(Some(model.into()));
        }
        let mut am: E::ActiveModel = model.clone().into_active_model();
        if let Some(o) = opts.as_ref() {
            apply_base_update(&mut am, o);
        } else {
            let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
            apply_base_update(&mut am, &empty);
        }
        let updated = if let Some(txn) = txn.as_ref() {
            am.update(txn.as_ref()).await?
        } else {
            am.update(&self.db).await?
        };
        Ok(Some(updated.into()))
    }

    /// Loads a row by id and updates it, recording audit fields from `opts` when present.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. Returns `None` when the id does not exist.
    /// Reads and writes run inside `opts.txn` when attached.
    pub async fn update_by_id_with_opts<F, U>(
        &self,
        id: i64,
        mut effector: F,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        F: FnMut(&mut E::Model) -> bool + Send,
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::ModelEx: Into<E::Model> + BaseEntity,
        E::Model: IntoActiveModel<E::ActiveModel> + Clone,
        E::ActiveModel: ActiveModelTrait<Entity = E> + Send,
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::Column: ColumnTrait + FromStr,
        <E::Column as FromStr>::Err: std::fmt::Debug,
    {
        let txn = opts_txn(opts.as_ref());
        let Some(model) = (if let Some(o) = opts.clone() {
            // `get_by_id_with_opts` is txn-aware, so the read stays in-transaction.
            self.get_by_id_with_opts(id, o).await?
        } else {
            self.get_by_id(id).await?
        }) else {
            return Ok(None);
        };
        let mut model: E::Model = model.into();
        if !effector(&mut model) {
            return Ok(Some(model.into()));
        }
        let mut am: E::ActiveModel = model.clone().into_active_model();
        if let Some(o) = opts.as_ref() {
            apply_base_update(&mut am, o);
        } else {
            let empty: RepositoryOptions<NoUser> = RepositoryOptions::default();
            apply_base_update(&mut am, &empty);
        }
        let updated = if let Some(txn) = txn.as_ref() {
            am.update(txn.as_ref()).await?
        } else {
            am.update(&self.db).await?
        };
        Ok(Some(updated.into()))
    }

    // ── DELETE ──────────────────────────────────────────────────────────────

    /// Deletes rows matching the filter and returns the affected row count.
    pub async fn delete_many(&self, filter: Option<DeleteQueryData<E>>) -> Result<u64, DbErr> {
        self.delete_many_with_opts::<NoUser>(filter, None).await
    }
    /// Deletes rows matching the filter with cursor support and returns the affected row count.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. Deletes run inside
    /// `opts.txn` when attached, else on the repository connection.
    pub async fn delete_many_with_opts<U>(
        &self,
        filter: Option<DeleteQueryData<E>>,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<u64, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
    {
        let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
        let txn = opts_txn(opts.as_ref());
        let del = filter
            .map(|qd| qd.delete)
            .unwrap_or_else(|| E::delete_many());
        let del = with_cursor_delete(del, &cursor);
        let res = if let Some(txn) = txn.as_ref() {
            del.exec(txn.as_ref()).await?
        } else {
            del.exec(&self.db).await?
        };
        Ok(res.rows_affected)
    }
    /// Alias for [`PersistentRepository::delete_many`].
    pub async fn remove_many(&self, filter: Option<DeleteQueryData<E>>) -> Result<u64, DbErr> {
        self.delete_many(filter).await
    }
    /// Alias for [`PersistentRepository::delete_many_with_opts`].
    pub async fn remove_many_with_opts<U>(
        &self,
        filter: Option<DeleteQueryData<E>>,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<u64, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
    {
        self.delete_many_with_opts(filter, opts).await
    }
    /// Deletes matching rows and returns true when at least one row was deleted.
    pub async fn delete_one(&self, filter: Option<DeleteQueryData<E>>) -> Result<bool, DbErr> {
        self.delete_one_with_opts::<NoUser>(filter, None).await
    }
    /// Deletes matching rows with cursor support; returns true when at least one row was deleted.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. Deletes run inside
    /// `opts.txn` when attached.
    pub async fn delete_one_with_opts<U>(
        &self,
        filter: Option<DeleteQueryData<E>>,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<bool, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
    {
        Ok(self.delete_many_with_opts(filter, opts).await? > 0)
    }

    /// Deletes the row with the given id and returns the deleted model, or `None` when absent.
    pub async fn delete_by_id(&self, id: i64) -> Result<Option<E::ModelEx>, DbErr>
    where
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::ModelEx: BaseEntity,
    {
        let Some(model) = self.get_by_id(id).await? else {
            return Ok(None);
        };
        let pk = <<E::PrimaryKey as PrimaryKeyTrait>::ValueType as From<i64>>::from(id);
        E::delete_by_id(pk).exec(&self.db).await?;
        Ok(Some(model))
    }

    /// Deletes the row with the given id using `opts.txn` when attached.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. Reads and deletes run
    /// inside the attached transaction; otherwise on the repository connection.
    pub async fn delete_by_id_with_opts<U>(
        &self,
        id: i64,
        opts: Option<RepositoryOptions<U>>,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::ModelEx: BaseEntity,
    {
        let txn = opts_txn(opts.as_ref());
        let Some(model) = (if let Some(txn) = txn.as_ref() {
            <E::ModelEx as BaseEntity>::load()
                .filter_by_id(id)
                .fetch(txn.as_ref(), 0, 1)
                .await?
                .into_iter()
                .next()
        } else {
            self.get_by_id(id).await?
        }) else {
            return Ok(None);
        };
        let pk = <<E::PrimaryKey as PrimaryKeyTrait>::ValueType as From<i64>>::from(id);
        if let Some(txn) = txn.as_ref() {
            E::delete_by_id(pk).exec(txn.as_ref()).await?;
        } else {
            E::delete_by_id(pk).exec(&self.db).await?;
        }
        Ok(Some(model))
    }

    /// Deletes the row with the given id inside the transaction, returning the deleted model if found.
    pub async fn delete_by_id_txn(
        &self,
        id: i64,
        txn: &DatabaseTransaction,
    ) -> Result<Option<E::ModelEx>, DbErr>
    where
        E::PrimaryKey: PrimaryKeyTrait,
        <E::PrimaryKey as PrimaryKeyTrait>::ValueType: From<i64>,
        E::ModelEx: BaseEntity,
    {
        let pk = <<E::PrimaryKey as PrimaryKeyTrait>::ValueType as From<i64>>::from(id);
        let Some(model) = <E::ModelEx as BaseEntity>::load()
            .filter_by_id(id)
            .fetch(txn, 0, 1)
            .await?
            .into_iter()
            .next()
        else {
            return Ok(None);
        };
        E::delete_by_id(pk).exec(txn).await?;
        Ok(Some(model))
    }

    /// Deletes rows matching the filter inside the transaction, returning the affected row count.
    pub async fn delete_many_txn(
        &self,
        filter: Option<DeleteQueryData<E>>,
        txn: &DatabaseTransaction,
    ) -> Result<u64, DbErr> {
        self.delete_many_txn_with_opts::<NoUser>(filter, None, txn)
            .await
    }
    /// Deletes rows matching the filter inside the transaction with cursor support.
    ///
    /// `U` is the [`RepositoryOptions`] identity type. When `opts` carries its
    /// own transaction it takes precedence; otherwise the explicit `txn` is used.
    pub async fn delete_many_txn_with_opts<U>(
        &self,
        filter: Option<DeleteQueryData<E>>,
        opts: Option<RepositoryOptions<U>>,
        txn: &DatabaseTransaction,
    ) -> Result<u64, DbErr>
    where
        U: BaseEntity + Clone + Send + Sync + 'static,
    {
        let cursor = opts.as_ref().and_then(|o| o.cursor.clone());
        let del = filter
            .map(|qd| qd.delete)
            .unwrap_or_else(|| E::delete_many());
        let del = with_cursor_delete(del, &cursor);
        if let Some(opts_txn) = opts_txn(opts.as_ref()) {
            // `RepositoryOptions` stays the single source of truth when both
            // handles are supplied.
            let res = del.exec(opts_txn.as_ref()).await?;
            return Ok(res.rows_affected);
        }
        let res = del.exec(txn).await?;
        Ok(res.rows_affected)
    }
}

// Helper to generate uid for new entities
/// Generates a new UUID v4 for `uid` columns on new rows.
pub fn new_uid() -> Uuid {
    Uuid::new_v4()
}