autumn_web/

repository.rs

//! Repository support types for framework-generated CRUD operations.
//!
//! Generated `#[repository]` read-only methods (`find_by_id`, `find_all`,
//! `count`, `paginate`, `cursor_page`, derived `find_by_*`, full-text-search
//! reads) route to the configured read replica automatically: set
//! `database.replica_url` and the extractor snapshots a [`ReadRoute`] per
//! request via [`crate::AppState::read_pool`]. Mutating methods (`save`,
//! `update`, `delete_by_id`, bulk writes) always run on the primary pool.
//! Pin a read-after-write-sensitive repository to the primary with
//! `#[repository(Model, primary_reads)]`, or pin a single call chain with
//! the generated `on_primary()` method. When no replica is configured, all
//! methods use the primary — nothing changes for single-pool apps.
//!
//! Sharded repositories built from a
//! [`ShardedDb`](crate::sharding::ShardedDb) via the generated `from_shard`
//! constructor get the same treatment **per shard**: reads route to the
//! shard's replica when one is configured and healthy (honoring the shard's
//! `replica_fallback` policy via
//! [`Shard::read_route`](crate::sharding::Shard::read_route)), writes stay on
//! the shard primary, and `primary_reads` / `on_primary()` pin reads back to
//! the shard primary.
//!
//! [`RepositoryError`] surfaces typed errors that arise during repository
//! operations — most notably optimistic-lock conflicts when two replicas
//! write the same row concurrently.

use sha2::{Digest, Sha256};
use thiserror::Error;

// Declarative counter caches (#1325) live in their own module but are part of
// the repository surface as far as callers are concerned — `#[model]`-emitted
// code, generated repositories and application escape hatches all name them
// through `autumn_web::repository::…`, next to `AutumnDependents`.
pub use crate::counter_cache::{
    AutumnCounterCaches, ChildState, CounterCacheSpec, TenantScope, counter_cache_after_insert,
    counter_cache_after_insert_by_id, counter_cache_after_insert_many, counter_cache_after_update,
    counter_cache_after_update_many, counter_cache_after_upsert_many, counter_cache_apply_delta,
    counter_cache_apply_delta_by_child_id, counter_cache_before_delete_by_id,
    counter_cache_before_delete_many, counter_cache_before_detach_many,
    counter_cache_before_restore_by_id, counter_cache_capture_fks, counter_cache_capture_fks_many,
    counter_cache_recompute, counter_cache_serialize_self_referential,
};
#[doc(inline)]
pub use crate::counter_cache::{Captured, CapturedContribution, CapturedContributions};

// Derivations (#1769) are maintained through the counter-cache specs above, so
// the hooks a generated repository calls are already in scope from that
// re-export. What `#[model]` needs additionally is the registry it submits each
// declared derivation into.
#[doc(inline)]
pub use crate::derivation::{DerivationDef, DerivationDescriptor};

// ── Dependent-cascade primitives (issue #2309 follow-up) ─────────────
//
// `dependent(...)` (#1369) generates an `__autumn_apply_dependent_on_conn`
// method on every repository. Three of its four `DependentAction` arms —
// restrict, nullify and delete_all — are pure dynamic SQL over a table name and
// a foreign-key column, both of which are already runtime `&str` values in the
// generated code. Emitting them per repository (254 lines, four
// `QueryableByName` derives) bought nothing: nothing in them is checked against
// the model's `schema.rs`.
//
// They live here instead, following the same shape `counter_cache_after_insert`
// and friends already use — `&'static` specs plus `fn` pointers erase the
// per-column typing, so `M` is the only generic parameter and the SQL compiles
// once. The `destroy` arm stays generated: it loads and deletes through diesel's
// typed DSL, recurses through the repository's own pool, and is genuinely
// model-specific.

/// Refuse the parent delete when any live child still references it.
///
/// `live_filter` is the soft-delete-aware `AND` fragment the generated code
/// baked in (empty for a hard-delete model), which makes `restrict`
/// soft-delete-aware: an already-soft-deleted child must not block the parent
/// with a 409, since it is no longer a live orphan.
///
/// This is a runtime support function for code generated by Autumn proc macros.
/// It is semver-exempt; do not call it directly.
///
/// # Errors
///
/// Returns a 409 [`crate::AutumnError::conflict_msg`] when a dependent row
/// exists, or the database error if the probe fails.
#[cfg(feature = "db")]
#[doc(hidden)]
pub async fn dependent_restrict(
    conn: &mut crate::db::RuntimeConnection,
    table: &str,
    model_name: &str,
    live_filter: &str,
    fk_column: &str,
    parent_id: i64,
) -> crate::AutumnResult<()> {
    use diesel_async::RunQueryDsl;

    // `AS dep_present` (not `AS exists`): `exists` is a Postgres reserved word
    // and would need quoting.
    #[derive(diesel::QueryableByName)]
    struct DepExists {
        #[diesel(sql_type = diesel::sql_types::Bool)]
        dep_present: bool,
    }

    let row: DepExists = diesel::sql_query(format!(
        "SELECT EXISTS(SELECT 1 FROM \"{table}\" WHERE \"{fk_column}\" = $1{live_filter}) \
         AS dep_present"
    ))
    .bind::<diesel::sql_types::BigInt, _>(parent_id)
    .get_result::<DepExists>(conn)
    .await
    .map_err(crate::AutumnError::from)?;

    if row.dep_present {
        return Err(crate::AutumnError::conflict_msg(format!(
            "cannot delete: dependent {model_name} row(s) referencing this record via \
             \"{table}\".\"{fk_column}\" still exist (dependent = restrict)"
        )));
    }
    Ok(())
}

/// Resolve the ids of every child row pointing at `parent_id`.
///
/// Ordered by `id` so the counter-cache decrements below apply in a stable
/// order, matching the generated code this replaces.
#[cfg(feature = "db")]
async fn dependent_child_ids(
    conn: &mut crate::db::RuntimeConnection,
    table: &str,
    fk_column: &str,
    parent_id: i64,
) -> crate::AutumnResult<Vec<i64>> {
    use diesel_async::RunQueryDsl;

    #[derive(diesel::QueryableByName)]
    struct ChildId {
        #[diesel(sql_type = diesel::sql_types::BigInt)]
        id: i64,
    }

    Ok(diesel::sql_query(format!(
        "SELECT id FROM \"{table}\" WHERE \"{fk_column}\" = $1 ORDER BY id"
    ))
    .bind::<diesel::sql_types::BigInt, _>(parent_id)
    .load::<ChildId>(conn)
    .await
    .map_err(crate::AutumnError::from)?
    .into_iter()
    .map(|row| row.id)
    .collect())
}

/// Detach every child row from `parent_id` by nulling its foreign key.
///
/// #1325: the children survive but detach, so the leg being cleared loses them
/// — and ONLY that leg. The ids are resolved first (the FK is about to be NULL,
/// so they are unrecoverable afterwards) and the decrement runs before the
/// UPDATE. Decrementing every spec — as [`dependent_delete_all`] correctly does,
/// since there the rows go away entirely — would permanently undercount the
/// children's other counter-cached legs.
///
/// `has_counter_caches` is the model's `const HAS_COUNTER_CACHES`; when it is
/// false no id-resolution query is issued at all.
///
/// This is a runtime support function for code generated by Autumn proc macros.
/// It is semver-exempt; do not call it directly.
///
/// # Errors
///
/// Returns the database error if the id resolution, the counter-cache
/// decrement, or the UPDATE fails.
#[cfg(feature = "db")]
#[doc(hidden)]
pub async fn dependent_nullify<M: 'static>(
    conn: &mut crate::db::RuntimeConnection,
    table: &str,
    specs: &[CounterCacheSpec<M>],
    has_counter_caches: bool,
    fk_column: &str,
    parent_id: i64,
) -> crate::AutumnResult<()> {
    use diesel_async::RunQueryDsl;

    if has_counter_caches {
        let ids = dependent_child_ids(conn, table, fk_column, parent_id).await?;
        counter_cache_before_detach_many(conn, specs, fk_column, &ids).await?;
    }

    diesel::sql_query(format!(
        "UPDATE \"{table}\" SET \"{fk_column}\" = NULL WHERE \"{fk_column}\" = $1"
    ))
    .bind::<diesel::sql_types::BigInt, _>(parent_id)
    .execute(conn)
    .await
    .map_err(crate::AutumnError::from)?;
    Ok(())
}

/// [`dependent_child_ids`], but takes `FOR UPDATE` on the selected rows, in
/// ascending id order (Postgres only — a no-op clause on `SQLite`, whose
/// single-writer lock already serializes this).
///
/// Codex review, PR #2647: the snapshot this feeds a subsequent bulk DELETE,
/// so it must be safe to act on. Under READ COMMITTED, `SELECT ... FOR UPDATE`
/// blocks on a row a concurrent transaction is mid-write to, and once that
/// transaction commits, re-checks `fk_column = parent_id` against the *new*
/// row version before including it — so a row reparented away in that window
/// is correctly excluded, matching the locked `SELECT ... FOR UPDATE` the
/// per-row Destroy loop has always taken over the same predicate. A plain
/// unlocked read has no such re-check: it can return a row that is reparented
/// and committed away before the caller's later DELETE/decrement run,
/// decrementing the wrong parent's counter cache for a row that was never
/// actually removed. Used only when a counter-cache decrement needs the ids
/// back afterwards — see [`lock_dependent_children_for_update`] for the
/// id-free variant, and [`dependent_delete_all`]'s doc comment for why the
/// `ORDER BY id` acquisition order matters even with no counter caches.
#[cfg(feature = "db")]
async fn dependent_child_ids_for_update(
    conn: &mut crate::db::RuntimeConnection,
    table: &str,
    fk_column: &str,
    parent_id: i64,
) -> crate::AutumnResult<Vec<i64>> {
    use diesel_async::RunQueryDsl;

    #[derive(diesel::QueryableByName)]
    struct ChildId {
        #[diesel(sql_type = diesel::sql_types::BigInt)]
        id: i64,
    }

    let for_update: &str = crate::backend_select! {
        pg => { " FOR UPDATE" },
        sqlite => { "" },
    };

    Ok(diesel::sql_query(format!(
        "SELECT id FROM \"{table}\" WHERE \"{fk_column}\" = $1 ORDER BY id{for_update}"
    ))
    .bind::<diesel::sql_types::BigInt, _>(parent_id)
    .load::<ChildId>(conn)
    .await
    .map_err(crate::AutumnError::from)?
    .into_iter()
    .map(|row| row.id)
    .collect())
}

/// Take the same `FOR UPDATE`, ascending-id locks [`dependent_child_ids_for_update`]
/// does, without transmitting the ids back to the client at all.
///
/// Codex review, PR #2647: [`dependent_delete_all`]'s deterministic pre-lock
/// only needs ids at all when there are counter caches to decrement; a
/// cache-free cascade needs the LOCKS (for deadlock-free ordering against a
/// concurrent cascade through a different foreign key) but never reads an id
/// back. An earlier revision selected `id` and called `execute()` instead of
/// `load()`, which does stop the driver from materializing a `Vec` — but
/// `execute()` still receives every row Postgres sends: the server has no
/// notion of a client that discards results, so a naked `SELECT id ... FOR
/// UPDATE` still puts one `DataRow` per matching child on the wire, an O(rows)
/// *network* cost even with O(1) heap use. Wrapping the same locking `SELECT`
/// in an outer `count(*)` keeps the fix: `FOR UPDATE` is legal on the
/// unaggregated inner query (locking every row it touches, in the same
/// `ORDER BY id` acquisition order as before), and only the outer query's
/// single count row ever reaches the client — O(1) statements, O(1) memory,
/// O(1) network, for a `delete_all`/leaf-`destroy` cascade of any fan-out.
#[cfg(feature = "db")]
async fn lock_dependent_children_for_update(
    conn: &mut crate::db::RuntimeConnection,
    table: &str,
    fk_column: &str,
    parent_id: i64,
) -> crate::AutumnResult<()> {
    use diesel_async::RunQueryDsl;

    let for_update: &str = crate::backend_select! {
        pg => { " FOR UPDATE" },
        sqlite => { "" },
    };

    diesel::sql_query(format!(
        "SELECT count(*) FROM (SELECT id FROM \"{table}\" WHERE \"{fk_column}\" = $1 \
         ORDER BY id{for_update}) AS __autumn_locked"
    ))
    .bind::<diesel::sql_types::BigInt, _>(parent_id)
    .execute(conn)
    .await
    .map_err(crate::AutumnError::from)?;
    Ok(())
}

/// Bulk-delete every child row pointing at `parent_id`.
///
/// #1325: same shape as [`dependent_nullify`] — the rows are about to go, so
/// their counter-cached parents are decremented first, across every spec.
///
/// Codex review, PR #2647: an ordered pre-lock now runs unconditionally, not
/// only when `has_counter_caches` — without it, two concurrent cascades
/// hitting an overlapping child set through *different* foreign keys (a
/// diamond: the same child table `dependent(..., on_delete = destroy)` of two
/// different parents) would each fall straight through to a plan-dependent-
/// order bulk `DELETE`, and two deletes locking the same overlapping rows in
/// opposite orders can deadlock, aborting one parent's cascade. The pre-lock,
/// taken in the same `ORDER BY id` every concurrent caller uses, makes the
/// acquisition order deterministic regardless of which foreign key drove the
/// scan — matching the per-row Destroy loop's existing guarantee. Which
/// pre-lock variant runs depends on whether the ids are needed afterwards:
/// [`dependent_child_ids_for_update`] when a counter-cache decrement will read
/// them back, [`lock_dependent_children_for_update`] otherwise, so a cache-free
/// cascade over a huge fan-out never materializes a child id it doesn't need.
/// Either way this stays O(1) statements: one extra locked `SELECT`, not one
/// per row.
///
/// This is a runtime support function for code generated by Autumn proc macros.
/// It is semver-exempt; do not call it directly.
///
/// # Errors
///
/// Returns the database error if the id resolution, the counter-cache
/// decrement, or the DELETE fails.
#[cfg(feature = "db")]
#[doc(hidden)]
pub async fn dependent_delete_all<M: 'static>(
    conn: &mut crate::db::RuntimeConnection,
    table: &str,
    specs: &[CounterCacheSpec<M>],
    has_counter_caches: bool,
    fk_column: &str,
    parent_id: i64,
) -> crate::AutumnResult<()> {
    use diesel_async::RunQueryDsl;

    if has_counter_caches {
        // The id selection below row-locks the children before the
        // counter-cache hook can take the lock that serializes mutations on
        // a table with a leg onto itself, so take it here first (see
        // `counter_cache_serialize_self_referential`).
        counter_cache_serialize_self_referential(conn, specs).await?;
        let ids = dependent_child_ids_for_update(conn, table, fk_column, parent_id).await?;
        counter_cache_before_delete_many(conn, specs, &ids).await?;
    } else {
        lock_dependent_children_for_update(conn, table, fk_column, parent_id).await?;
    }

    diesel::sql_query(format!(
        "DELETE FROM \"{table}\" WHERE \"{fk_column}\" = $1"
    ))
    .bind::<diesel::sql_types::BigInt, _>(parent_id)
    .execute(conn)
    .await
    .map_err(crate::AutumnError::from)?;
    Ok(())
}

/// Backend-portable `FOR UPDATE` seam for generated `#[repository]` CRUD.
///
/// Postgres pessimistic-lock reads chain `.for_update()` onto a select query to
/// take a row lock, but diesel's `LockingClause` only implements
/// `QueryFragment<Pg>` — the same query is not constructible against the
/// `SQLite` backend at all, so even a *simple* repository's always-emitted lock
/// reads would fail to compile under `--features sqlite`. Worse, diesel's
/// `FOR UPDATE` lock marker is `pub(crate)`, so no generic extension trait in a
/// downstream crate can name the bound `for_update()` needs — a portable seam
/// cannot be a plain trait.
///
/// This macro is that seam: the `#[repository]` macro emits
/// `::autumn_web::maybe_for_update!(<query>)` where it used to chain
/// `<query>.for_update()`, and the returned query is still loadable
/// (`.first(...)`/`.load(...)`), so the generated call chains compose
/// identically on either backend. Because the `#[cfg]` that selects the
/// expansion lives on the macro **definition**, it is evaluated in
/// **autumn-web's** compilation (where the `sqlite` feature actually lives) —
/// not the consumer crate's — so a downstream app needs no `sqlite` feature of
/// its own for the right arm to be chosen.
///
/// - **Postgres** (default): expands to `QueryDsl::for_update(<query>)` — the
///   `FOR UPDATE` locking clause, unchanged. Marker resolution happens at the
///   concrete call site, so no unnameable bound is ever written.
/// - **`SQLite`**: expands to `<query>` verbatim — `SQLite` has no
///   `SELECT … FOR UPDATE` (it serializes writers with a database-level lock),
///   so the row-lock read degrades to a plain read. Write-write correctness
///   still rests on the optimistic `lock_version` check plus the pool's
///   `busy_timeout`; see [`crate::db::RuntimeConnection`] and issue #1996 for
///   the residual single-writer concurrency caveat.
#[cfg(all(feature = "db", not(feature = "sqlite")))]
#[macro_export]
macro_rules! maybe_for_update {
    ($query:expr $(,)?) => {
        $crate::reexports::diesel::query_dsl::QueryDsl::for_update($query)
    };
}

/// `SQLite` arm of [`maybe_for_update!`] — the identity.
///
/// See the Postgres
/// definition for the full contract; `SQLite` has no `SELECT … FOR UPDATE`, so a
/// pessimistic-lock read degrades to a plain read.
#[cfg(all(feature = "db", feature = "sqlite"))]
#[macro_export]
macro_rules! maybe_for_update {
    ($query:expr $(,)?) => {
        $query
    };
}

/// Compile-time backend block selector for generated `#[repository]`/`#[model]` CRUD.
///
/// Used where the two backends need *structurally different* code (not just a
/// swapped receiver), e.g. Postgres multi-row batch insert vs. the `SQLite`
/// per-row loop, or the batched `ON CONFLICT` upsert vs. `SQLite`'s per-row
/// upsert.
///
/// The `#[cfg]` selecting the arm lives on the macro **definition**, so it is
/// evaluated in **autumn-web's** compilation (where the `sqlite` feature lives)
/// — the un-selected arm's tokens are never even type-checked, which is what
/// lets each arm use backend-specific diesel fragments (Postgres `DEFAULT`-keyword
/// batch inserts, `SQLite` per-row `RETURNING`) that would not compile against
/// the other backend.
///
/// Expands to the chosen block as an expression:
/// `::autumn_web::backend_select! { pg => { … }, sqlite => { … } }`.
#[cfg(all(feature = "db", not(feature = "sqlite")))]
#[macro_export]
macro_rules! backend_select {
    (pg => $pg:block, sqlite => $sqlite:block $(,)?) => {
        $pg
    };
}

/// `SQLite` arm of [`backend_select!`]. See the Postgres definition for the
/// contract.
#[cfg(all(feature = "db", feature = "sqlite"))]
#[macro_export]
macro_rules! backend_select {
    (pg => $pg:block, sqlite => $sqlite:block $(,)?) => {
        $sqlite
    };
}

/// Take the same transaction-scoped advisory lock the `position` (issue
/// #1358) insert-assign/delete-compact triggers take, from generated
/// `move_to`.
///
/// Closes a deadlock where `move_to`'s `SELECT ... FOR UPDATE ORDER BY id`
/// row locks and a concurrent compaction trigger's `UPDATE ... WHERE
/// position > $1` (which Postgres may satisfy via the `(scope, position)`
/// index, i.e. **not** in `id` order) can lock the same rows in different
/// orders.
///
/// `lock_name` is the literal `"{table}_{position}_assign"` string baked
/// into the migration's trigger SQL (see
/// `autumn-cli`'s `position_triggers_up_sql_for`); `scope` is the row's
/// scope column value (`Some`) or `None` for an unscoped position column —
/// mirroring the trigger's own `hashtext(NEW."{scope}"::text)` / literal
/// `0` key2 exactly, so `move_to` and the triggers contend on the *same*
/// advisory lock key and fully serialize against each other.
///
/// **Postgres**: takes `pg_advisory_xact_lock(hashtext(lock_name),
/// hashtext(scope::text))` (or key2 `0` when unscoped) — released
/// automatically at transaction end.
/// **`SQLite`**: a no-op — `SQLite` has no advisory locks and no trigger
/// counterpart takes one either; write-write correctness there rests on
/// the database-level write lock `scoped_immediate_transaction` already
/// acquires via `BEGIN IMMEDIATE`.
///
/// # Errors
///
/// Returns the underlying [`diesel::result::Error`] if the
/// `pg_advisory_xact_lock` query fails (e.g. the connection was lost).
#[cfg(all(feature = "db", not(feature = "sqlite")))]
pub async fn position_advisory_lock(
    conn: &mut crate::db::RuntimeConnection,
    lock_name: &str,
    scope: ::core::option::Option<i64>,
) -> ::std::result::Result<(), diesel::result::Error> {
    use crate::reexports::diesel_async::RunQueryDsl as _;
    match scope {
        ::core::option::Option::Some(scope_value) => {
            diesel::sql_query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))")
                .bind::<diesel::sql_types::Text, _>(lock_name)
                .bind::<diesel::sql_types::Text, _>(scope_value.to_string())
                .execute(conn)
                .await?;
        }
        ::core::option::Option::None => {
            diesel::sql_query("SELECT pg_advisory_xact_lock(hashtext($1), 0)")
                .bind::<diesel::sql_types::Text, _>(lock_name)
                .execute(conn)
                .await?;
        }
    }
    Ok(())
}

/// `SQLite` arm of [`position_advisory_lock`] — a no-op. See the Postgres
/// definition for the full contract.
///
/// # Errors
///
/// Never returns `Err` — kept `Result`-returning to match the Postgres
/// arm's signature, since both are called from the same generated
/// `move_to` body regardless of backend.
#[cfg(all(feature = "db", feature = "sqlite"))]
pub async fn position_advisory_lock(
    _conn: &mut crate::db::RuntimeConnection,
    _lock_name: &str,
    _scope: ::core::option::Option<i64>,
) -> ::std::result::Result<(), diesel::result::Error> {
    Ok(())
}

/// Where a generated repository routes its read-only methods (`find_by_id`,
/// `find_all`, `count`, `paginate`, `cursor_page`, derived `find_by_*`,
/// full-text-search reads, …).
///
/// The route is snapshotted from [`crate::AppState`] when the repository is
/// extracted, so every read within a request sees one consistent decision.
/// Mutating methods (`save`, `update`, `delete_by_id`, bulk writes) always
/// use the primary pool regardless of this route, as do pessimistic-lock
/// reads (`with_lock`) and reads running on an explicit transaction
/// connection.
#[cfg(feature = "db")]
#[derive(Clone)]
pub enum ReadRoute {
    /// Reads use the primary/write pool: no replica is configured, the
    /// repository was declared with `#[repository(..., primary_reads)]`, or
    /// the caller pinned this instance with `on_primary()`.
    Primary,
    /// Reads use this read-role pool snapshot — the replica when healthy,
    /// or the primary when the replica is unready and the
    /// [`ReplicaFallback::Primary`](crate::config::ReplicaFallback) policy
    /// applies.
    ReadPool(diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>),
    /// A replica is configured but currently unready, and the
    /// [`ReplicaFallback::FailReadiness`](crate::config::ReplicaFallback)
    /// policy forbids falling back to the primary.
    ///
    /// Generated reads fail
    /// fast with `503 Service Unavailable` instead of silently serving
    /// from the wrong role.
    Unavailable,
}

#[cfg(feature = "db")]
impl ReadRoute {
    /// Snapshot the read-routing decision for one request from the app
    /// state, mirroring [`crate::AppState::read_pool`] semantics.
    #[must_use]
    pub fn from_state(state: &crate::AppState) -> Self {
        if state.replica_pool().is_some() {
            state
                .read_pool()
                .map_or(Self::Unavailable, |pool| Self::ReadPool(pool.clone()))
        } else {
            Self::Primary
        }
    }
}

#[cfg(feature = "db")]
impl std::fmt::Debug for ReadRoute {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Primary => f.write_str("ReadRoute::Primary"),
            Self::ReadPool(pool) => {
                write!(f, "ReadRoute::ReadPool(max={})", pool.status().max_size)
            }
            Self::Unavailable => f.write_str("ReadRoute::Unavailable"),
        }
    }
}

/// Upper bound on bound parameters in one prepared statement, per backend —
/// used by generated bulk-write CRUD to size insert/update chunks so a batch
/// never exceeds the driver's placeholder limit.
///
/// Postgres caps a statement at 65535 (`u16`) parameters; `SQLite`'s default
/// `SQLITE_MAX_VARIABLE_NUMBER` is 32766 (since 3.32.0). The generated code
/// divides this by the per-row column count to pick a chunk size, so the
/// value tracks the active backend via the same `sqlite` feature that flips
/// [`crate::db::RuntimeConnection`]. (The `SQLite` write path also inserts
/// row-by-row, so this bound is a conservative belt-and-suspenders there.)
#[cfg(not(feature = "sqlite"))]
pub const MAX_BIND_PARAMS: usize = 65535;
/// See [`MAX_BIND_PARAMS`] (Postgres). `SQLite`'s default
/// `SQLITE_MAX_VARIABLE_NUMBER` is 32766.
#[cfg(feature = "sqlite")]
pub const MAX_BIND_PARAMS: usize = 32766;

/// Typed errors returned by generated repository methods.
///
/// Distinct from [`crate::AutumnError`] so callers can match on the
/// variant without parsing an HTTP status code.
#[derive(Debug, Clone, Error)]
pub enum RepositoryError {
    /// Two writers raced on the same row.
    ///
    /// Returned by generated `update`/`save` methods when the
    /// `#[lock_version]` field no longer matches the value the client
    /// sent — meaning another replica committed a write in the meantime.
    ///
    /// Map this to `409 Conflict` via [`crate::AutumnError::conflict`].
    #[error(
        "optimistic lock conflict on record {id}: \
         client expected version {expected_version}, \
         row was already modified (actual: {actual_version:?})"
    )]
    Conflict {
        /// Primary key of the contested record.
        id: i64,
        /// The version the client read and expected to still be current.
        expected_version: i64,
        /// The version actually stored when the conflict was detected,
        /// or `None` if the row was deleted between the read and the write.
        actual_version: Option<i64>,
    },
}

/// How a parent's dependent children are handled when the parent is deleted.
///
/// Declared per association on the `#[repository]` macro via
/// `dependent(ChildRepository, fk = "parent_fk", on_delete = <action>)`, and
/// applied inside the parent's `delete_by_id` transaction so a parent delete
/// never leaves orphaned child rows (issue #1369).
///
/// The cascade runs app-side (not via a DB `ON DELETE` constraint) so that
/// soft-delete, lifecycle/commit hooks, and counter caches stay correct — a
/// guarantee raw `ON DELETE CASCADE` cannot provide.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DependentAction {
    /// Delete each child through the child repository's delete path, honoring
    /// the child's soft-delete mode and firing its lifecycle/commit hooks.
    Destroy,
    /// Bulk hard-delete every child in a single `DELETE`, skipping per-row hooks.
    DeleteAll,
    /// Set the child's foreign-key column to `NULL` (the child rows survive,
    /// detached from the deleted parent).
    Nullify,
    /// Refuse to delete the parent while any child exists, surfacing a typed
    /// `409 Conflict` error instead of orphaning or cascading.
    Restrict,
}

/// The boxed, `Send` future a [`RuntimeDependentCascadeFn`] returns.
///
/// It resolves to the deferred `(topic, dom_id)` OOB delete broadcasts a single
/// child association's cascade accumulated (empty for a non-broadcasting child).
pub type RuntimeDependentCascadeFuture<'a> = ::std::pin::Pin<
    ::std::boxed::Box<
        dyn ::std::future::Future<Output = crate::AutumnResult<Vec<(String, String)>>> + Send + 'a,
    >,
>;

/// A type-erased entry point into one child repository's
/// `__autumn_apply_dependent_on_conn` leaf executor (#1738).
///
/// `#[model]` emits one of these per model-declared
/// `#[has_many(Child, dependent = …)]` association, resolving the child
/// repository through the `Pg{Child}Repository` naming convention. Given the
/// parent repository's pool, its live connection/transaction, the parent id,
/// the parent's soft-delete kind, and the two shared cascade-guard sets, it
/// constructs the child repository and applies this one association's cascade,
/// returning any deferred OOB delete broadcasts to publish post-commit.
///
/// The three guard sets serve distinct roles (Codex round-5-B + #1800 case 1):
/// `__path` is the ACTIVE recursion stack (pushed before descending into a node's
/// children, popped once that subtree completes) used only to break
/// self-/mutual-referential cycles; `__deleted` is a monotonic set of every row
/// HANDLED by the cascade — soft OR physical — used by the bulk `delete_many` root
/// dedup / hook double-fire guard to skip a root already processed as another
/// root's descendant (so its `before_delete` never fires twice); `__physical` is
/// the subset of rows PHYSICALLY removed, used only by the diamond traversal
/// revisit-skip so a hard-delete path can still physically remove a row a
/// soft-delete path merely marked `deleted_at` on. Keeping `__deleted` and
/// `__physical` separate is what lets the bulk root skip stay complete for
/// soft-delete graphs while the diamond hard path is still free to run; keeping
/// `__path` separate lets a batch process a descendant root before its ancestor
/// without the ancestor's cascade skipping a still-referenced intermediate
/// (immediate-FK failure).
///
/// All references share one lifetime so the returned future can borrow the
/// connection and all three guard sets for exactly as long as it is awaited.
pub type RuntimeDependentCascadeFn = for<'a> fn(
    &'a ::diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>,
    &'a mut crate::db::RuntimeConnection,
    i64,
    bool,
    &'a mut ::std::collections::HashSet<(&'static str, i64)>,
    &'a mut ::std::collections::HashSet<(&'static str, i64)>,
    &'a mut ::std::collections::HashSet<(&'static str, i64)>,
) -> RuntimeDependentCascadeFuture<'a>;

/// One model-declared dependent-cascade association, produced at compile time
/// by `#[model]` and consulted at run time by the parent's `delete_by_id`
/// (#1738).
///
/// The `#[model]` and `#[repository]` derives are separate proc-macro
/// invocations, so the repository macro — which owns the `delete_by_id` cascade
/// codegen — never sees the model struct's `#[has_many]` attributes. This spec
/// bridges the two at run time: the parent iterates
/// [`AutumnDependents::dependents`] and drives each spec's [`cascade`] inside
/// its transaction, reproducing exactly the cascade the repository-attribute
/// `dependent(PgChildRepository, …)` form produces.
///
/// Framework plumbing; not constructed by hand.
///
/// [`cascade`]: RuntimeDependentSpec::cascade
pub struct RuntimeDependentSpec {
    /// The child foreign-key column referencing the parent id. Informational:
    /// the [`cascade`] thunk already binds it. Mirrors the repository-attribute
    /// `fk = "…"`.
    ///
    /// [`cascade`]: RuntimeDependentSpec::cascade
    pub fk: &'static str,
    /// What to do with the children when the parent is deleted.
    pub action: DependentAction,
    /// Type-erased entry into the child repository's cascade leaf executor.
    pub cascade: RuntimeDependentCascadeFn,
}

/// Exposes a model's runtime dependent-cascade specs to the parent
/// repository's generated `delete_by_id` (#1738).
///
/// A blanket impl returns an empty slice for every type; `#[model]` emits an
/// *inherent* `dependents()` associated function — which shadows this trait
/// method under Rust's inherent-before-trait resolution, mirroring the
/// [`AutumnLockVersionModelExt`] pattern — only when the model declares at
/// least one `#[has_many(…, dependent = …)]` / `#[has_one(…, dependent = …)]`
/// association. The generated repository calls `Model::dependents()`, so a
/// model with no dependents (or one defined without `#[model]`) resolves to the
/// empty blanket and keeps its exact prior delete codegen.
///
/// Precedence: the repository-attribute `dependent(…)` form is the explicit
/// escape hatch (needed for child repositories that do not follow the
/// `Pg{Child}Repository` convention, e.g. cross-crate children). When a
/// repository declares `dependent(…)`, its compile-time specs are authoritative
/// and the model-declared runtime specs are ignored; a repository with no
/// `dependent(…)` drives the cascade from `Model::dependents()`.
pub trait AutumnDependents {
    /// The model's dependent-cascade specs, in declaration order. Defaults to
    /// none; `#[model]` overrides via an inherent shadow when dependents exist.
    #[must_use]
    fn dependents() -> &'static [RuntimeDependentSpec] {
        &[]
    }
}

// Blanket fallback — any type without an inherent `dependents()` (i.e. a model
// with no dependent associations, or a type not built through `#[model]`)
// resolves to the empty slice.
impl<T: ?Sized> AutumnDependents for T {}

/// Publish a batch of deferred dependent-cascade OOB *delete* broadcasts.
///
/// Accumulated during a `dependent = destroy` cascade over `broadcasts = true`
/// children and published **after** the parent transaction commits (#1369).
///
/// Each entry is `(topic, dom_id)`; the fragment is empty and the swap is
/// [`OobSwap::Delete`](crate::htmx::OobSwap::Delete). Deferring to post-commit
/// means a rolled-back cascade publishes nothing, so live clients are never told
/// to remove a child that still exists. Commit-hook children defer via the
/// durable outbox instead and are not routed through here.
///
/// Framework plumbing; not a public API. No-op when the live-channel /
/// `maud` / `htmx` features are not built in (the buffer is always empty then).
#[cfg(all(feature = "ws", feature = "maud", feature = "htmx"))]
#[doc(hidden)]
pub fn publish_deferred_dependent_broadcasts(broadcasts: Vec<(String, String)>) {
    if broadcasts.is_empty() {
        return;
    }
    if let Some(channels) = crate::__private::get_global_channels() {
        let fragment = crate::html! {};
        // Consume the buffer so the owned (topic, dom_id) strings move straight
        // into the publish call (no needless clone, and the `Vec` is consumed —
        // avoiding `clippy::needless_pass_by_value`).
        for (topic, dom_id) in broadcasts {
            if let Err(err) = channels.broadcast().publish_oob(
                &topic,
                &dom_id,
                &crate::htmx::OobSwap::Delete,
                &fragment,
            ) {
                tracing::warn!(
                    error = %err,
                    "auto-broadcast dependent-destroy delete failed"
                );
            }
        }
    }
}

/// No-op fallback when live broadcasting is not compiled in.
///
/// The accumulation
/// side only runs for `broadcasts = true` children (which require these
/// features), so the buffer is always empty in this configuration.
#[cfg(not(all(feature = "ws", feature = "maud", feature = "htmx")))]
#[doc(hidden)]
pub fn publish_deferred_dependent_broadcasts(_broadcasts: Vec<(String, String)>) {}

/// Extension trait that provides a fallback `None` for model structs that do
/// not have a `#[lock_version]` field — or that are defined manually without
/// going through `#[model]`.
///
/// `#[model]` generates an *inherent* method with the same name on the model
/// and on `UpdateModel`; inherent methods take priority over trait methods in
/// Rust's method-resolution order.  For types without `#[lock_version]` (or
/// without `#[model]` altogether), the trait provides the `None` fallback so
/// the generated repository code can call these methods unconditionally.
#[doc(hidden)]
pub trait AutumnLockVersionModelExt {
    fn __autumn_lock_version_actual(&self) -> Option<i64> {
        None
    }
}

#[doc(hidden)]
pub trait AutumnLockVersionUpdateExt {
    fn __autumn_lock_version_expected(&self) -> Option<i64> {
        None
    }
}

// Blanket impls — any type that doesn't have an inherent implementation
// (generated by `#[model]`) falls through to these, returning `None`.
impl<T: ?Sized> AutumnLockVersionModelExt for T {}
impl<T: ?Sized> AutumnLockVersionUpdateExt for T {}

#[doc(hidden)]
pub trait AutumnColumnCountExt {
    fn __autumn_column_count(&self) -> usize;
}

#[doc(hidden)]
pub trait AutumnColumnCountSpecific {
    fn __autumn_column_count(self) -> usize;
}
impl<T: AutumnColumnCountExt> AutumnColumnCountSpecific for &T {
    fn __autumn_column_count(self) -> usize {
        self.__autumn_column_count()
    }
}

#[doc(hidden)]
pub trait AutumnColumnCountFallback {
    fn __autumn_column_count(self) -> usize;
}
impl<T: ?Sized> AutumnColumnCountFallback for &&T {
    fn __autumn_column_count(self) -> usize {
        30
    }
}

#[doc(hidden)]
pub trait AutumnUpsertSetExt {
    type UpsertSet;
    fn __autumn_upsert_set() -> Self::UpsertSet;
}

#[doc(hidden)]
pub trait AutumnUpsertExecutionExt {
    type Model;
    fn __autumn_execute_upsert<'a>(
        chunk: &'a [Self::Model],
        tenant_id: ::core::option::Option<&'a str>,
        conn: &'a mut crate::db::RuntimeConnection,
    ) -> impl ::std::future::Future<
        Output = ::core::result::Result<::std::vec::Vec<Self::Model>, ::diesel::result::Error>,
    > + Send
    + 'a;
}

#[doc(hidden)]
pub trait AutumnCorrelateExt: Sized {
    type NewModel: Sized;
    fn __autumn_correlate_new(
        inputs: &[Self::NewModel],
        record: &Self,
        matched: &mut [bool],
    ) -> ::core::option::Option<usize>;

    fn __autumn_correlate_model(
        inputs: &[Self],
        record: &Self,
        matched: &mut [bool],
    ) -> ::core::option::Option<usize>;
}

/// Extension trait to override `tenant_id` on changesets in tenant-scoped updates.
pub trait CanSetTenantId {
    fn set_tenant_id(&mut self, tenant_id: String);
}

/// Trait implemented by models to expose their primary key value.
pub trait ModelPrimaryKey {
    type IdType: ::std::fmt::Display + ::core::clone::Clone + Send + Sync + 'static;
    fn primary_key_value(&self) -> Self::IdType;
}

/// Framework plumbing bridging a `#[repository]`-generated struct to the
/// many-to-many (`#[has_many(Target, through = join_table)]`) mutation
/// helpers `#[model]` generates.
///
/// `#[repository(Model, ...)]` implements this
/// once, unconditionally, alongside the model's other generated impls; the
/// `Model` associated type is what keeps `add_*`/`remove_*`/`set_*` method
/// resolution unambiguous when more than one m2m trait is in scope (e.g. two
/// models both associate `through` the same join table, or a model has more
/// than one m2m association) — each mutation trait is blanket-implemented
/// only for `M2mConnSource<Model = TheAssociationsOwner>`.
///
/// It also backs the `#[votable(by = ..., aggregate = sum|count)]` reaction
/// helpers (#1362): the `{Model}Reactions` trait `#[model]` emits is
/// blanket-implemented over `M2mConnSource<Model = TheVotableModel>`, and
/// `react()` acquires its own connection through
/// [`M2mConnSource::__autumn_m2m_write_conn`] exactly like an `add_*` does,
/// while the read-only `reaction_of()` uses
/// [`M2mConnSource::__autumn_m2m_read_conn`]. Both scope their **target**
/// lookups with [`M2mConnSource::__autumn_m2m_tenant_scope`] when the target
/// model carries a `tenant_id` column.
///
/// Not part of the public API; not implemented by hand.
#[cfg(feature = "db")]
#[doc(hidden)]
pub trait M2mConnSource: Send + Sync {
    /// The model whose `#[has_many(..., through = ...)]` associations (or
    /// `#[votable]` reactions) this repository's mutation helpers operate on.
    type Model;

    /// Acquire a primary-pool connection for an `add_*`/`remove_*`/`set_*`
    /// many-to-many mutation, or for a `#[votable]` `react()`.
    ///
    /// Mirrors the write-connection acquisition every
    /// other mutating generated method uses (marks the read-your-writes pin
    /// on success).
    fn __autumn_m2m_write_conn(
        &self,
    ) -> impl ::std::future::Future<
        Output = crate::AutumnResult<
            diesel_async::pooled_connection::deadpool::Object<crate::db::RuntimeConnection>,
        >,
    > + Send;

    /// Acquire a connection for a *read-only* association/reaction accessor —
    /// currently `#[votable]`'s `reaction_of()`.
    ///
    /// Routes exactly like every other generated read: per the repository's
    /// [`ReadRoute`] snapshot, so a configured replica serves it (and an
    /// `Unavailable` route fails fast). Unlike
    /// [`__autumn_m2m_write_conn`](M2mConnSource::__autumn_m2m_write_conn) it
    /// does **not** mark the read-your-writes pin — it is a read, so it neither
    /// pins subsequent reads to the primary nor promises to observe a write
    /// this request has not yet committed. A caller that must see its own
    /// just-committed `react()` on a replica-backed repository should read
    /// through a `primary_reads` / `on_primary()` repository, or use the
    /// `Reaction` value `react()` already returned.
    fn __autumn_m2m_read_conn(
        &self,
    ) -> impl ::std::future::Future<
        Output = crate::AutumnResult<
            diesel_async::pooled_connection::deadpool::Object<crate::db::RuntimeConnection>,
        >,
    > + Send;

    /// The tenant this repository would scope its queries to, for the
    /// `#[votable]` reaction helpers to apply to their **target** lookups.
    ///
    /// Synchronous (it only reads the `CURRENT_TENANT` task-local) and
    /// deliberately three-valued, mirroring what a derived finder on the same
    /// repository would do:
    ///
    /// - `Ok(Some(tenant))` — a `#[repository(..., tenant_scoped)]` repository
    ///   used inside a tenant context: `react()` / `reaction_of()` must filter
    ///   the target on `tenant_id`, so a foreign-tenant target id is `NotFound`
    ///   (respectively `None`) before any write.
    /// - `Ok(None)` — a repository that is not `tenant_scoped`, or one used via
    ///   `across_tenants()`: no tenant predicate, exactly like its finders.
    /// - `Err(..)` — a `tenant_scoped` repository with no tenant context: the
    ///   same "no tenant context was established" failure the derived queries
    ///   raise, so the reaction surface fails closed rather than writing
    ///   unscoped.
    ///
    /// Only the codegen for a model that actually has a `tenant_id` column
    /// calls it; a model without one pays nothing.
    ///
    /// # Errors
    ///
    /// Returns an error when the repository is `tenant_scoped`, is not in
    /// `across_tenants()` mode, and no tenant context was established.
    fn __autumn_m2m_tenant_scope(&self) -> crate::AutumnResult<::core::option::Option<String>>;
}

/// Which of the three edge mutations a `#[votable]` `react()` call performed
/// (#1362).
///
/// A reaction edge is a single `(reactor, target)` row in the edge table, so
/// every `react()` resolves to exactly one of: create it, replace its value, or
/// delete it. The discrimination is returned rather than inferred from
/// [`Reaction::value`] so a route can tell "the user changed their mind"
/// (`Flipped`) apart from "the user reacted for the first time" (`Inserted`)
/// without a second query — useful for flash messages, analytics and
/// notification fan-out.
///
/// # Examples
///
/// ```
/// use autumn_web::repository::ReactionOutcome;
///
/// // Only a signed (`aggregate = sum`) reaction can be flipped.
/// assert_ne!(ReactionOutcome::Inserted, ReactionOutcome::Flipped);
/// ```
///
/// `#[non_exhaustive]`: a future aggregate mode may add an outcome, so match
/// arms in downstream crates must carry a `_ => …` fallback.
#[cfg(feature = "db")]
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReactionOutcome {
    /// The reactor had no reaction on this target; one was created.
    Inserted,
    /// The reactor had the *opposite* reaction; its value was replaced in
    /// place (no second edge row is ever created).
    ///
    /// Never produced by an `aggregate = count` reaction: a count edge carries
    /// no value, so a repeat click can only toggle it off.
    Flipped,
    /// The reactor repeated its existing reaction, so the edge was deleted
    /// (toggle-off).
    Removed,
}

/// The post-commit state a `#[votable]` `react()` call leaves behind (#1362).
///
/// `react()` toggles/flips/inserts the `(reactor, target)` edge **and**
/// recomputes the target's aggregate column from ground truth in the same
/// transaction. Everything a caller needs to re-render the reaction control is
/// therefore already known when the transaction commits, and is returned here —
/// a route never has to issue a follow-up `SELECT` for the new score or for the
/// viewer's own reaction.
///
/// `#[non_exhaustive]`: `react()` is the only constructor, so a later field
/// (say, the edge's `created_at`) is not a breaking change. Read the fields;
/// do not struct-literal one from outside this crate.
///
/// # Examples
///
/// ```
/// use autumn_web::repository::{Reaction, ReactionOutcome};
///
/// // `react(user_id, post_id, 1)` on a post whose score was 6 returns
/// // value `Some(1)`, aggregate `7`, outcome `Inserted`. Callers read those
/// // fields — `react()` is the only constructor, and the wildcard arm is what
/// // `#[non_exhaustive]` asks of a downstream match.
/// fn label(reaction: &Reaction) -> String {
///     match reaction.outcome {
///         ReactionOutcome::Removed => format!("score {}", reaction.aggregate),
///         _ => format!("score {} (yours: {:?})", reaction.aggregate, reaction.value),
///     }
/// }
/// ```
#[cfg(feature = "db")]
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Reaction {
    /// The reactor's reaction *after* the call: `Some(value)` after an insert
    /// or a flip, `None` after a toggle-off.
    ///
    /// An `aggregate = count` reaction reports `Some(1)` while the membership
    /// row exists, so view code is mode-independent.
    pub value: Option<i16>,
    /// The target's newly persisted aggregate (`score` for `aggregate = sum`,
    /// `{name}_count` for `aggregate = count`) — ground truth as of this
    /// transaction's commit, not an accumulated delta.
    pub aggregate: i64,
    /// Which edge mutation ran.
    pub outcome: ReactionOutcome,
}

#[cfg(feature = "db")]
impl Reaction {
    /// Framework plumbing: the constructor `#[votable]`'s generated `react()`
    /// uses.
    ///
    /// `Reaction` is `#[non_exhaustive]`, so the generated code — which expands
    /// in the *application's* crate — cannot write a struct literal. Not a
    /// public API; call `react()` instead.
    #[doc(hidden)]
    #[must_use]
    pub const fn __new(value: Option<i16>, aggregate: i64, outcome: ReactionOutcome) -> Self {
        Self {
            value,
            aggregate,
            outcome,
        }
    }
}

/// Metadata trait implemented for model structs to expose FTS configuration.
pub trait AutumnSearchableModel {
    const IS_SEARCHABLE: bool;
    const SEARCH_LANGUAGE: &'static str;
    const SEARCH_FIELDS: &'static [(&'static str, char)];
}

/// Build a fail-closed `SQLite` FTS5 `MATCH` query string from raw user input
/// (issue #1910).
///
/// `SQLite` FTS5's `MATCH` right-hand operand is a query *language*, not a plain
/// string: bare words are terms, but `AND` / `OR` / `NOT` / `NEAR`, a `col:`
/// prefix, `*` (prefix), `^` (initial-token), `(`/`)` grouping, `"` phrases and
/// a leading `-` are all operators. Passing raw user input straight to `MATCH`
/// would let a caller inject FTS5 query syntax — or, more commonly, trigger a
/// hard syntax error on innocuous punctuation. This helper neutralizes all of
/// it: the input is split on Unicode whitespace and each token is emitted as a
/// quoted FTS5 string literal (a `"..."` phrase), with any embedded `"` doubled
/// (`""`) per FTS5's own string escaping. The quoted tokens are joined with a
/// single space, which FTS5 reads as an implicit AND of literal phrases. Every
/// operator character therefore becomes part of a literal term and can never
/// change the query's structure (the analogue of the Postgres path's
/// `websearch_to_tsquery` and the old LIKE path's metacharacter escaping).
///
/// Returns `None` when the input yields no tokens (empty or whitespace-only).
/// The caller MUST treat `None` as an empty result set and run **no** query —
/// never an unfiltered scan.
#[must_use]
pub fn sqlite_fts5_match_query(input: &str) -> Option<String> {
    // Worst case (no embedded quotes to double): every byte kept plus the two
    // surrounding quotes of a single token — pre-size to avoid reallocation.
    let mut out = String::with_capacity(input.len() + 2);
    for token in input.split_whitespace() {
        if !out.is_empty() {
            out.push(' ');
        }
        out.push('"');
        for ch in token.chars() {
            if ch == '"' {
                // FTS5 escapes a literal double-quote inside a "..." string by
                // doubling it.
                out.push('"');
            }
            out.push(ch);
        }
        out.push('"');
    }
    if out.is_empty() { None } else { Some(out) }
}

/// Derive a stable signed 64-bit advisory lock key for repository upserts.
///
/// Generated versioned repositories use this before pre-reading rows for
/// `upsert_many`, so concurrent generated upserts for the same table/id cannot
/// classify audit history from a stale missing-row snapshot.
#[doc(hidden)]
#[must_use]
pub fn repository_upsert_advisory_lock_key(table_name: &str, record_id: i64) -> i64 {
    let mut hasher = Sha256::new();
    hasher.update(b"repository_upsert\0");
    hasher.update(table_name.as_bytes());
    hasher.update(b"\0");
    hasher.update(record_id.to_be_bytes());
    let digest = hasher.finalize();
    let mut bytes = [0_u8; 8];
    bytes.copy_from_slice(&digest[..8]);
    i64::from_be_bytes(bytes)
}

/// Trait for formatting repository model fields into SSE broadcast topic segments.
///
/// This provides a uniform display format for both primitive types and optional/nullable values.
pub trait DisplayTopicField {
    /// Formats the field into a string suitable for a topic name.
    fn to_topic_string(&self) -> String;
}

impl DisplayTopicField for String {
    fn to_topic_string(&self) -> String {
        self.clone()
    }
}

impl DisplayTopicField for &str {
    fn to_topic_string(&self) -> String {
        (*self).to_string()
    }
}

impl DisplayTopicField for bool {
    fn to_topic_string(&self) -> String {
        self.to_string()
    }
}

impl DisplayTopicField for char {
    fn to_topic_string(&self) -> String {
        self.to_string()
    }
}

macro_rules! impl_display_topic_field_num {
    ($($t:ty),*) => {
        $(
            impl DisplayTopicField for $t {
                fn to_topic_string(&self) -> String {
                    self.to_string()
                }
            }
        )*
    };
}

impl_display_topic_field_num!(i8, i16, i32, i64, isize, u8, u16, u32, u64, usize, f32, f64);

impl DisplayTopicField for ::uuid::Uuid {
    fn to_topic_string(&self) -> String {
        self.to_string()
    }
}

impl DisplayTopicField for ::chrono::NaiveDateTime {
    fn to_topic_string(&self) -> String {
        self.to_string()
    }
}

impl DisplayTopicField for ::chrono::NaiveDate {
    fn to_topic_string(&self) -> String {
        self.to_string()
    }
}

impl DisplayTopicField for ::chrono::NaiveTime {
    fn to_topic_string(&self) -> String {
        self.to_string()
    }
}

impl<T> DisplayTopicField for ::chrono::DateTime<T>
where
    T: ::chrono::TimeZone,
    T::Offset: std::fmt::Display,
{
    fn to_topic_string(&self) -> String {
        self.to_string()
    }
}

impl<T: DisplayTopicField> DisplayTopicField for Option<T> {
    fn to_topic_string(&self) -> String {
        self.as_ref()
            .map_or_else(|| "none".to_string(), DisplayTopicField::to_topic_string)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn fts5_match_quotes_plain_terms_and_ands_them() {
        // A plain multi-word query becomes an implicit-AND of quoted phrases.
        assert_eq!(
            sqlite_fts5_match_query("rust web").as_deref(),
            Some("\"rust\" \"web\"")
        );
        assert_eq!(
            sqlite_fts5_match_query("hello").as_deref(),
            Some("\"hello\"")
        );
    }

    #[test]
    fn fts5_match_neutralizes_operators_as_literals() {
        // FTS5 operators are quoted into literal terms — never parsed as syntax.
        assert_eq!(
            sqlite_fts5_match_query("foo OR bar").as_deref(),
            Some("\"foo\" \"OR\" \"bar\"")
        );
        assert_eq!(
            sqlite_fts5_match_query("title:secret").as_deref(),
            Some("\"title:secret\"")
        );
        assert_eq!(sqlite_fts5_match_query("pre*").as_deref(), Some("\"pre*\""));
        assert_eq!(
            sqlite_fts5_match_query("NEAR(a b)").as_deref(),
            Some("\"NEAR(a\" \"b)\"")
        );
    }

    #[test]
    fn fts5_match_doubles_embedded_quotes() {
        // An embedded double-quote is escaped by doubling, so it can never close
        // the phrase early and inject trailing syntax.
        assert_eq!(
            sqlite_fts5_match_query("a\"b").as_deref(),
            Some("\"a\"\"b\"")
        );
        assert_eq!(
            sqlite_fts5_match_query("\" OR x").as_deref(),
            Some("\"\"\"\" \"OR\" \"x\"")
        );
    }

    #[test]
    fn fts5_match_empty_input_is_none() {
        // Empty / whitespace-only / no-token input yields None so the caller
        // returns an empty page WITHOUT running any query (fail-closed).
        assert_eq!(sqlite_fts5_match_query(""), None);
        assert_eq!(sqlite_fts5_match_query("   "), None);
        assert_eq!(sqlite_fts5_match_query("\t\n  \r"), None);
    }

    #[test]
    fn conflict_variant_stores_all_fields() {
        let err = RepositoryError::Conflict {
            id: 42,
            expected_version: 3,
            actual_version: Some(4),
        };
        match err {
            RepositoryError::Conflict {
                id,
                expected_version,
                actual_version,
            } => {
                assert_eq!(id, 42);
                assert_eq!(expected_version, 3);
                assert_eq!(actual_version, Some(4));
            }
        }
    }

    #[test]
    fn conflict_with_no_actual_version() {
        let err = RepositoryError::Conflict {
            id: 1,
            expected_version: 0,
            actual_version: None,
        };
        assert!(matches!(
            err,
            RepositoryError::Conflict {
                actual_version: None,
                ..
            }
        ));
    }

    #[test]
    fn conflict_display_includes_id_and_expected_version() {
        let err = RepositoryError::Conflict {
            id: 99,
            expected_version: 7,
            actual_version: Some(8),
        };
        let s = err.to_string();
        assert!(s.contains("99"), "display should include id");
        assert!(s.contains('7'), "display should include expected_version");
    }

    #[test]
    fn conflict_is_clone() {
        let err = RepositoryError::Conflict {
            id: 1,
            expected_version: 0,
            actual_version: Some(1),
        };
        let cloned = err.clone();
        assert!(matches!(err, RepositoryError::Conflict { id: 1, .. }));
        assert!(matches!(cloned, RepositoryError::Conflict { id: 1, .. }));
    }

    #[test]
    fn conflict_implements_std_error() {
        let err = RepositoryError::Conflict {
            id: 1,
            expected_version: 0,
            actual_version: None,
        };
        let _: &dyn std::error::Error = &err;
    }

    #[test]
    fn repository_upsert_advisory_lock_key_is_stable_for_same_table_and_id() {
        let a = repository_upsert_advisory_lock_key("posts", 42);
        let b = repository_upsert_advisory_lock_key("posts", 42);

        assert_eq!(a, b);
        assert_ne!(a, 0);
    }

    #[test]
    fn repository_upsert_advisory_lock_key_separates_table_and_id() {
        let key = repository_upsert_advisory_lock_key("posts", 42);

        assert_ne!(key, repository_upsert_advisory_lock_key("comments", 42));
        assert_ne!(key, repository_upsert_advisory_lock_key("posts", 43));
    }
}