distributed 3.3.0

//! Backend-agnostic read-model logic shared by the Postgres and SQLite repositories.
//!
//! Two layers live here:
//!
//! 1. **Pure helpers** (no `sqlx` types): write-plan validation, key/patch
//!    reconciliation, row-version arithmetic, relationship-include resolution.
//! 2. **The generic relational write path**: free functions over
//!    `DB: SqlxReadModelBackend` that build and run the upsert/patch/delete SQL via
//!    `QueryBuilder<DB>`. Only the value-binding (`Bool`/typed-`NULL`), the
//!    `rows_affected` accessor, and two label strings differ per dialect; the
//!    [`SqlxReadModelBackend`] trait carries exactly those — one trait, two
//!    one-line-per-method impls — so the SQL-building logic exists once rather than
//!    being mirrored (and risking silent drift) in `postgres_repo`/`sqlite_repo`.

use std::collections::BTreeMap;
use std::sync::RwLock;

use sqlx::{Database, Encode, Executor, IntoArguments, QueryBuilder, Row, Transaction, Type};

use crate::read_model::{
    ReadModelIncludeRows, ReadModelLoadGraph, ReadModelLoadRequest, ReadModelQueryCapabilities,
    Versioned,
};
use crate::table::TableSchemaRegistry;
use crate::table::{
    key_fingerprint, validate_key, validate_row_values, DeleteTableRowMutation, ExpectedVersion,
    PatchMode, PatchTableRowMutation, RelationshipDef, RelationshipKind, RowKey, RowValue,
    RowValues, RowWriteMode, TableAdapterCapabilities, TableColumn, TableCommitOutcome,
    TableMutation, TableRowMutation, TableSchema, TableStoreError, TableWritePlan,
};

/// A resolved relationship include: the relationship metadata plus the registered
/// schema of the target model, ready for the relational load path to query.
#[derive(Clone)]
pub(crate) struct IncludeSpec {
    pub(crate) name: String,
    pub(crate) relationship: RelationshipDef,
    pub(crate) target_schema: TableSchema,
}

pub(crate) fn remember_read_model_schemas(
    stored: &RwLock<TableSchemaRegistry>,
    registry: &TableSchemaRegistry,
) -> Result<(), TableStoreError> {
    let mut stored = stored
        .write()
        .map_err(|_| TableStoreError::Storage("read-model schema registry lock poisoned".into()))?;

    for schema in registry.schemas() {
        if let Some(existing) = stored.schema_for_table(&schema.table_name) {
            if existing != schema {
                return Err(TableStoreError::Metadata(format!(
                    "read-model schema registry already contains table `{}` with different metadata",
                    schema.table_name
                )));
            }
            continue;
        }
        stored.register_schema(schema.clone())?;
    }

    Ok(())
}

pub(crate) fn resolve_registered_read_model_schemas(
    registry: &RwLock<TableSchemaRegistry>,
    request: &ReadModelLoadRequest,
) -> Result<(TableSchema, Vec<IncludeSpec>), TableStoreError> {
    if request.includes.is_empty() {
        return Ok((request.schema.clone(), Vec::new()));
    }

    let registry = registry
        .read()
        .map_err(|_| TableStoreError::Storage("read-model schema registry lock poisoned".into()))?;
    let root_schema = registry
        .schema_for_model(&request.schema.model_name)
        .cloned()
        .ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "read model `{}` is not registered for relationship includes",
                request.schema.model_name
            ))
        })?;
    if root_schema != request.schema {
        return Err(TableStoreError::Metadata(format!(
            "read model `{}` load request does not match registered schema",
            request.schema.model_name
        )));
    }

    let mut include_specs = Vec::with_capacity(request.includes.len());
    for include_name in &request.includes {
        let relationship = root_schema
            .relationships
            .iter()
            .find(|relationship| relationship.field_name == *include_name)
            .ok_or_else(|| {
                TableStoreError::Metadata(format!(
                    "read model `{}` has no relationship `{}`",
                    root_schema.model_name, include_name
                ))
            })?;
        if matches!(relationship.kind, RelationshipKind::ManyToMany) {
            return Err(TableStoreError::Metadata(format!(
                "many-to-many relationship `{}` includes are not supported until join metadata declares source and target keys",
                relationship.field_name
            )));
        }
        let target_schema = registry
            .schema_for_model(&relationship.target_model)
            .ok_or_else(|| {
                TableStoreError::Metadata(format!(
                    "read model `{}` relationship `{}` targets unregistered model `{}`",
                    root_schema.model_name, relationship.field_name, relationship.target_model
                ))
            })?;

        include_specs.push(IncludeSpec {
            name: include_name.clone(),
            relationship: relationship.clone(),
            target_schema: target_schema.clone(),
        });
    }

    Ok((root_schema, include_specs))
}

pub(crate) fn sql_read_model_capabilities() -> TableAdapterCapabilities {
    TableAdapterCapabilities {
        relational_rows: true,
        sparse_patches: true,
        deletes: true,
    }
}

pub(crate) fn validate_sql_write_plan(plan: &TableWritePlan) -> Result<(), TableStoreError> {
    plan.validate_for(&sql_read_model_capabilities())
}

pub(crate) fn initial_row_version() -> u64 {
    1
}

pub(crate) fn validate_row_expected_version(
    schema: &TableSchema,
    key: &RowKey,
    expected_version: &ExpectedVersion,
    current_version: Option<u64>,
) -> Result<(), TableStoreError> {
    match (expected_version, current_version) {
        (ExpectedVersion::Any, _) => Ok(()),
        (ExpectedVersion::Exact(expected), Some(actual)) if expected == &actual => Ok(()),
        (ExpectedVersion::Exact(expected), Some(actual)) => {
            Err(row_concurrency_conflict(schema, key, *expected, actual))
        }
        (ExpectedVersion::Exact(_), None) => Err(TableStoreError::NotFound {
            collection: schema.table_name.clone(),
            id: key_fingerprint(key),
        }),
        (ExpectedVersion::NotExists, None) => Ok(()),
        (ExpectedVersion::NotExists, Some(actual)) => {
            Err(row_concurrency_conflict(schema, key, 0, actual))
        }
    }
}

pub(crate) fn row_concurrency_conflict(
    schema: &TableSchema,
    key: &RowKey,
    expected: u64,
    actual: u64,
) -> TableStoreError {
    TableStoreError::ConcurrencyConflict {
        collection: schema.table_name.clone(),
        id: key_fingerprint(key),
        expected,
        actual,
    }
}

pub(crate) fn row_values_from_key_and_patch(
    schema: &TableSchema,
    key: &RowKey,
    patch: crate::table::RowPatch,
) -> Result<RowValues, TableStoreError> {
    let mut values = RowValues::new();
    for (column, value) in key.iter() {
        values.insert(column.to_string(), value.clone());
    }
    for (column, value) in patch.into_values() {
        if schema
            .primary_key
            .columns
            .iter()
            .any(|primary_key| primary_key == &column)
        {
            let key_value = key.get(&column).ok_or_else(|| {
                TableStoreError::Metadata(format!(
                    "read model `{}` row key is missing primary-key column `{}`",
                    schema.model_name, column
                ))
            })?;
            if key_value != &value {
                return Err(TableStoreError::Metadata(format!(
                    "read model `{}` patch cannot change primary-key column `{}`",
                    schema.model_name, column
                )));
            }
        }
        values.insert(column, value);
    }
    validate_row_values(schema, &values, true)?;
    validate_values_match_key(schema, key, &values)?;
    Ok(values)
}

pub(crate) fn patch_values_preserving_key<'schema>(
    schema: &'schema TableSchema,
    key: &RowKey,
    patch: &crate::table::RowPatch,
) -> Result<Vec<(&'schema TableColumn, RowValue)>, TableStoreError> {
    let mut values = Vec::new();
    for (column_name, value) in patch.iter() {
        let column = column_by_name(schema, column_name)?;
        if column.primary_key {
            let key_value = key.get(column_name).ok_or_else(|| {
                TableStoreError::Metadata(format!(
                    "read model `{}` row key is missing primary-key column `{}`",
                    schema.model_name, column_name
                ))
            })?;
            if key_value != value {
                return Err(TableStoreError::Metadata(format!(
                    "read model `{}` patch cannot change primary-key column `{}`",
                    schema.model_name, column_name
                )));
            }
            continue;
        }
        values.push((column, value.clone()));
    }
    Ok(values)
}

pub(crate) fn validate_values_match_key(
    schema: &TableSchema,
    key: &RowKey,
    values: &RowValues,
) -> Result<(), TableStoreError> {
    for column in &schema.primary_key.columns {
        let key_value = key.get(column).ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "read model `{}` row key is missing primary-key column `{}`",
                schema.model_name, column
            ))
        })?;
        let row_value = values.get(column).ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "read model `{}` row is missing primary-key column `{}`",
                schema.model_name, column
            ))
        })?;
        if row_value != key_value {
            return Err(TableStoreError::Metadata(format!(
                "read model `{}` row values cannot change primary-key column `{}`",
                schema.model_name, column
            )));
        }
    }
    Ok(())
}

pub(crate) fn belongs_to_target_column(
    target_schema: &TableSchema,
    source_column: &str,
) -> Result<String, TableStoreError> {
    if target_schema.primary_key.columns.len() != 1 {
        return Err(TableStoreError::Metadata(format!(
            "belongs_to target `{}` must have a single-column primary key to load from `{}`",
            target_schema.model_name, source_column
        )));
    }

    Ok(target_schema.primary_key.columns[0].clone())
}

pub(crate) fn empty_string_as_none(value: &str) -> Option<&str> {
    if value.is_empty() {
        None
    } else {
        Some(value)
    }
}

pub(crate) fn row_write_values<'schema>(
    schema: &'schema TableSchema,
    values: &RowValues,
) -> Result<Vec<(&'schema TableColumn, RowValue)>, TableStoreError> {
    values
        .iter()
        .map(|(column_name, value)| Ok((column_by_name(schema, column_name)?, value.clone())))
        .collect()
}

pub(crate) fn column_by_name<'schema>(
    schema: &'schema TableSchema,
    column_name: &str,
) -> Result<&'schema TableColumn, TableStoreError> {
    schema
        .columns
        .iter()
        .find(|column| column.column_name == column_name)
        .ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "read model `{}` write references missing column `{}`",
                schema.model_name, column_name
            ))
        })
}

pub(crate) fn version_column(schema: &TableSchema) -> Result<&str, TableStoreError> {
    schema.version_column.as_deref().ok_or_else(|| {
        TableStoreError::Metadata(format!(
            "read model `{}` requires a version column for SQL write-plan persistence",
            schema.model_name
        ))
    })
}

pub(crate) fn quote_identifier(value: &str) -> String {
    format!("\"{}\"", value.replace('"', "\"\""))
}

use super::{read_model_i64_from_u64, read_model_storage_error, read_model_u64_from_i64};

/// Dialect surface for the shared relational write path.
///
/// The upsert/patch/delete SQL is identical across Postgres and SQLite because
/// `QueryBuilder<DB>` renders the right placeholder dialect; only two things
/// genuinely differ: the value-binding for `Bool`/typed-`NULL` (Postgres binds
/// native `bool` and needs explicit per-type `NULL` casts for `$N` inference;
/// SQLite stores booleans as `i64` and collapses integer/bool `NULL`s) and the
/// backend/storage labels used in numeric-conversion error messages. Those —
/// and nothing else — live behind this trait, implemented once per backend.
pub trait SqlxReadModelBackend: Database {
    /// Backend name used in numeric-conversion error messages (`"postgres"`/`"sqlite"`).
    const BACKEND: &'static str;
    /// Human-readable storage label for the signed-64-bit version column.
    const INTEGER_STORAGE: &'static str;

    /// Bind one `RowValue` into the builder (dialect-specific encoding), then push
    /// any required type cast (Postgres `::jsonb`/`::timestamptz`; SQLite none).
    fn push_row_value_bind(
        builder: &mut QueryBuilder<Self>,
        value: RowValue,
        column: &TableColumn,
    ) -> Result<(), TableStoreError>;

    /// Bind a typed `NULL` for the column's type (Postgres needs the concrete
    /// `Option::<T>::None` per type so `$N` infers correctly).
    fn push_null_bind(
        builder: &mut QueryBuilder<Self>,
        column: &TableColumn,
    ) -> Result<(), TableStoreError>;

    /// Affected-row count of a write result. `sqlx` exposes `rows_affected` only as
    /// an inherent method on each backend's `QueryResult`, not via a shared trait,
    /// so the one-line accessor is delegated here.
    fn rows_affected(result: &Self::QueryResult) -> u64;

    /// Render one `SELECT`-list column. Postgres casts JSON/Timestamp to `::text`
    /// so they decode as `String`; SQLite stores them as text already, so it just
    /// pushes the quoted column. (Reading is the inverse of `push_row_value_bind`.)
    fn push_select_column(builder: &mut QueryBuilder<Self>, column: &TableColumn);

    /// Decode one fetched column into a `RowValue`. The one genuinely dialect-
    /// specific read: Postgres has a native `BOOLEAN`, SQLite stores booleans as
    /// `INTEGER` and decodes `value != 0`.
    fn row_value(row: &Self::Row, column: &TableColumn) -> Result<RowValue, TableStoreError>;
}

pub(crate) async fn begin_read_model_tx<DB: SqlxReadModelBackend>(
    pool: &sqlx::Pool<DB>,
) -> Result<Transaction<'_, DB>, TableStoreError> {
    pool.begin()
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "begin transaction", err))
}

pub(crate) async fn commit_read_model_tx<DB: SqlxReadModelBackend>(
    tx: Transaction<'_, DB>,
) -> Result<(), TableStoreError> {
    tx.commit()
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "commit transaction", err))
}

pub(crate) async fn commit_read_model_write_plan<DB>(
    pool: &sqlx::Pool<DB>,
    plan: TableWritePlan,
) -> Result<TableCommitOutcome, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    validate_sql_write_plan(&plan)?;
    let mut tx = begin_read_model_tx(pool).await?;
    let outcome = apply_read_model_write_plan_in_tx(&mut tx, plan).await?;
    commit_read_model_tx(tx).await?;
    Ok(outcome)
}

pub(crate) async fn apply_read_model_write_plan_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    plan: TableWritePlan,
) -> Result<TableCommitOutcome, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    validate_sql_write_plan(&plan)?;

    for mutation in plan.mutations {
        match mutation {
            TableMutation::UpsertRow(mutation) => {
                upsert_relational_row_in_tx(tx, mutation).await?;
            }
            TableMutation::PatchRow(mutation) => {
                patch_relational_row_in_tx(tx, mutation).await?;
            }
            TableMutation::DeleteRow(mutation) => {
                delete_relational_row_in_tx(tx, mutation).await?;
            }
        }
    }

    Ok(TableCommitOutcome::applied())
}

pub(crate) async fn upsert_relational_row_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    mutation: TableRowMutation,
) -> Result<(), TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    validate_key(mutation.schema, &mutation.key)?;
    validate_row_values(mutation.schema, &mutation.values, true)?;
    validate_values_match_key(mutation.schema, &mutation.key, &mutation.values)?;

    // The common case — upsert without an optimistic-version check — is a single
    // `INSERT ... ON CONFLICT (pk) DO UPDATE` round trip. Only version-checked
    // writes need to observe the current row first.
    if matches!(mutation.mode, RowWriteMode::Upsert)
        && matches!(mutation.expected_version, ExpectedVersion::Any)
    {
        return upsert_relational_row_on_conflict_in_tx(tx, mutation.schema, &mutation.values)
            .await;
    }

    let current_version = row_version_in_tx(tx, mutation.schema, &mutation.key).await?;
    validate_row_expected_version(
        mutation.schema,
        &mutation.key,
        &mutation.expected_version,
        current_version,
    )?;
    if matches!(mutation.mode, RowWriteMode::Insert) && current_version.is_some() {
        return Err(row_concurrency_conflict(
            mutation.schema,
            &mutation.key,
            0,
            current_version.unwrap_or_default(),
        ));
    }

    match current_version {
        Some(expected_version) => {
            let rows_affected = update_relational_row_values_in_tx(
                tx,
                mutation.schema,
                &mutation.key,
                &mutation.values,
                Some(expected_version),
            )
            .await?;
            if rows_affected == 0 {
                let actual = row_version_in_tx(tx, mutation.schema, &mutation.key)
                    .await?
                    .unwrap_or(expected_version);
                return Err(row_concurrency_conflict(
                    mutation.schema,
                    &mutation.key,
                    expected_version,
                    actual,
                ));
            }
        }
        None => {
            insert_relational_row_in_tx(
                tx,
                mutation.schema,
                &mutation.values,
                initial_row_version(),
            )
            .await?;
        }
    }

    Ok(())
}

pub(crate) async fn patch_relational_row_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    mutation: PatchTableRowMutation,
) -> Result<(), TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    validate_key(mutation.schema, &mutation.key)?;

    // `NotExists` is the only shape that has to observe the row before writing;
    // `Any`/`Exact` run the UPDATE directly and only re-read on a miss to tell
    // "not found" apart from a version conflict.
    if matches!(mutation.expected_version, ExpectedVersion::NotExists) {
        let current_version = row_version_in_tx(tx, mutation.schema, &mutation.key).await?;
        validate_row_expected_version(
            mutation.schema,
            &mutation.key,
            &mutation.expected_version,
            current_version,
        )?;
        if !matches!(mutation.mode, PatchMode::InsertMissing) {
            return Err(TableStoreError::NotFound {
                collection: mutation.schema.table_name.clone(),
                id: key_fingerprint(&mutation.key),
            });
        }
        let values = row_values_from_key_and_patch(mutation.schema, &mutation.key, mutation.patch)?;
        return insert_relational_row_in_tx(tx, mutation.schema, &values, initial_row_version())
            .await;
    }

    let expected_version = match mutation.expected_version {
        ExpectedVersion::Exact(expected) => Some(expected),
        _ => None,
    };
    let patch_values =
        patch_values_preserving_key(mutation.schema, &mutation.key, &mutation.patch)?;
    let rows_affected = update_relational_columns_in_tx(
        tx,
        mutation.schema,
        &mutation.key,
        patch_values,
        expected_version,
    )
    .await?;
    if rows_affected == 0 {
        if let Some(expected_version) = expected_version {
            return match row_version_in_tx(tx, mutation.schema, &mutation.key).await? {
                Some(actual) => Err(row_concurrency_conflict(
                    mutation.schema,
                    &mutation.key,
                    expected_version,
                    actual,
                )),
                None => Err(TableStoreError::NotFound {
                    collection: mutation.schema.table_name.clone(),
                    id: key_fingerprint(&mutation.key),
                }),
            };
        }
        if matches!(mutation.mode, PatchMode::InsertMissing) {
            let values =
                row_values_from_key_and_patch(mutation.schema, &mutation.key, mutation.patch)?;
            insert_relational_row_in_tx(tx, mutation.schema, &values, initial_row_version())
                .await?;
        } else {
            return Err(TableStoreError::NotFound {
                collection: mutation.schema.table_name.clone(),
                id: key_fingerprint(&mutation.key),
            });
        }
    }

    Ok(())
}

pub(crate) async fn delete_relational_row_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    mutation: DeleteTableRowMutation,
) -> Result<(), TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    validate_key(mutation.schema, &mutation.key)?;

    match mutation.expected_version {
        // The row must not exist: nothing to delete, but surface a conflict if it does.
        ExpectedVersion::NotExists => {
            let current_version = row_version_in_tx(tx, mutation.schema, &mutation.key).await?;
            validate_row_expected_version(
                mutation.schema,
                &mutation.key,
                &mutation.expected_version,
                current_version,
            )?;
            Ok(())
        }
        // No version check: one DELETE; deleting a missing row is a no-op.
        ExpectedVersion::Any => {
            delete_relational_row_where_version_in_tx(tx, mutation.schema, &mutation.key, None)
                .await?;
            Ok(())
        }
        // Version-checked delete: only re-read on a miss to tell "not found"
        // apart from a version conflict.
        ExpectedVersion::Exact(expected_version) => {
            let rows_affected = delete_relational_row_where_version_in_tx(
                tx,
                mutation.schema,
                &mutation.key,
                Some(expected_version),
            )
            .await?;
            if rows_affected == 0 {
                return match row_version_in_tx(tx, mutation.schema, &mutation.key).await? {
                    Some(actual) => Err(row_concurrency_conflict(
                        mutation.schema,
                        &mutation.key,
                        expected_version,
                        actual,
                    )),
                    None => Err(TableStoreError::NotFound {
                        collection: mutation.schema.table_name.clone(),
                        id: key_fingerprint(&mutation.key),
                    }),
                };
            }
            Ok(())
        }
    }
}

pub(crate) async fn row_version_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    schema: &TableSchema,
    key: &RowKey,
) -> Result<Option<u64>, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let version_column = version_column(schema)?;
    let mut builder = QueryBuilder::<DB>::new("SELECT ");
    builder.push(quote_identifier(version_column));
    builder.push(" FROM ");
    builder.push(quote_identifier(&schema.table_name));
    push_key_predicates(&mut builder, schema, key)?;

    let row = builder
        .build()
        .fetch_optional(&mut **tx)
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "load relational row version", err))?;

    row.map(|row| {
        read_model_u64_from_i64(
            DB::BACKEND,
            row.try_get::<i64, _>(version_column).map_err(|err| {
                read_model_storage_error(DB::BACKEND, "decode relational row version", err)
            })?,
            version_column,
        )
    })
    .transpose()
}

pub(crate) async fn insert_relational_row_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    schema: &TableSchema,
    values: &RowValues,
    version: u64,
) -> Result<(), TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let version_column = version_column(schema)?;
    let write_values = row_write_values(schema, values)?;
    let has_write_values = !write_values.is_empty();
    let mut builder = QueryBuilder::<DB>::new("INSERT INTO ");
    builder.push(quote_identifier(&schema.table_name));
    builder.push(" (");
    for (index, (column, _)) in write_values.iter().enumerate() {
        if index > 0 {
            builder.push(", ");
        }
        builder.push(quote_identifier(&column.column_name));
    }
    if has_write_values {
        builder.push(", ");
    }
    builder.push(quote_identifier(version_column));
    builder.push(") VALUES (");
    for (index, (column, value)) in write_values.into_iter().enumerate() {
        if index > 0 {
            builder.push(", ");
        }
        DB::push_row_value_bind(&mut builder, value, column)?;
    }
    if has_write_values {
        builder.push(", ");
    }
    builder.push_bind(read_model_i64_from_u64(
        DB::BACKEND,
        version,
        version_column,
        DB::INTEGER_STORAGE,
    )?);
    builder.push(")");

    builder
        .build()
        .execute(&mut **tx)
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "insert relational row", err))?;

    Ok(())
}

/// Upsert one row in a single statement: `INSERT ... ON CONFLICT (pk) DO UPDATE
/// SET <non-pk columns> = excluded.<column>, <version> = <version> + 1`.
///
/// Both Postgres and SQLite (≥ 3.35) support `ON CONFLICT` with an explicit
/// column-list target and the `excluded` pseudo-table. New rows start at
/// version 1; conflicting rows bump their version in-database (an increment
/// past `i64::MAX` fails as a storage error).
pub(crate) async fn upsert_relational_row_on_conflict_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    schema: &TableSchema,
    values: &RowValues,
) -> Result<(), TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let version_column = version_column(schema)?;
    let write_values = row_write_values(schema, values)?;
    let mut builder = QueryBuilder::<DB>::new("INSERT INTO ");
    builder.push(quote_identifier(&schema.table_name));
    builder.push(" (");
    for (column, _) in &write_values {
        builder.push(quote_identifier(&column.column_name));
        builder.push(", ");
    }
    builder.push(quote_identifier(version_column));
    builder.push(") VALUES (");
    for (column, value) in write_values.iter().cloned() {
        DB::push_row_value_bind(&mut builder, value, column)?;
        builder.push(", ");
    }
    builder.push_bind(read_model_i64_from_u64(
        DB::BACKEND,
        initial_row_version(),
        version_column,
        DB::INTEGER_STORAGE,
    )?);
    builder.push(") ON CONFLICT (");
    for (index, column_name) in schema.primary_key.columns.iter().enumerate() {
        if index > 0 {
            builder.push(", ");
        }
        builder.push(quote_identifier(column_name));
    }
    builder.push(") DO UPDATE SET ");
    for (column, _) in write_values
        .iter()
        .filter(|(column, _)| !column.primary_key)
    {
        builder.push(quote_identifier(&column.column_name));
        builder.push(" = excluded.");
        builder.push(quote_identifier(&column.column_name));
        builder.push(", ");
    }
    // Qualify the existing-row reference with the table name: inside `DO UPDATE`
    // an unqualified column is ambiguous with `excluded` on Postgres.
    builder.push(quote_identifier(version_column));
    builder.push(" = ");
    builder.push(quote_identifier(&schema.table_name));
    builder.push(".");
    builder.push(quote_identifier(version_column));
    builder.push(" + 1");

    builder
        .build()
        .execute(&mut **tx)
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "upsert relational row", err))?;

    Ok(())
}

pub(crate) async fn update_relational_row_values_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    schema: &TableSchema,
    key: &RowKey,
    values: &RowValues,
    expected_version: Option<u64>,
) -> Result<u64, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let mut write_values = row_write_values(schema, values)?;
    write_values.retain(|(column, _)| !column.primary_key);
    update_relational_columns_in_tx(tx, schema, key, write_values, expected_version).await
}

/// `UPDATE <table> SET <columns>, <version> = <version> + 1 WHERE <pk> [AND
/// <version> = <expected>]`, returning the affected-row count. The version bump
/// happens in-database; an increment past `i64::MAX` fails as a storage error.
pub(crate) async fn update_relational_columns_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    schema: &TableSchema,
    key: &RowKey,
    write_values: Vec<(&TableColumn, RowValue)>,
    expected_version: Option<u64>,
) -> Result<u64, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let version_column = version_column(schema)?;
    let mut builder = QueryBuilder::<DB>::new("UPDATE ");
    builder.push(quote_identifier(&schema.table_name));
    builder.push(" SET ");
    for (column, value) in write_values {
        builder.push(quote_identifier(&column.column_name));
        builder.push(" = ");
        DB::push_row_value_bind(&mut builder, value, column)?;
        builder.push(", ");
    }
    builder.push(quote_identifier(version_column));
    builder.push(" = ");
    builder.push(quote_identifier(version_column));
    builder.push(" + 1");
    push_key_predicates(&mut builder, schema, key)?;
    if let Some(expected_version) = expected_version {
        builder.push(" AND ");
        builder.push(quote_identifier(version_column));
        builder.push(" = ");
        builder.push_bind(read_model_i64_from_u64(
            DB::BACKEND,
            expected_version,
            "expected version",
            DB::INTEGER_STORAGE,
        )?);
    }

    let result = builder
        .build()
        .execute(&mut **tx)
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "update relational row", err))?;
    Ok(DB::rows_affected(&result))
}

pub(crate) async fn delete_relational_row_where_version_in_tx<DB>(
    tx: &mut Transaction<'_, DB>,
    schema: &TableSchema,
    key: &RowKey,
    expected_version: Option<u64>,
) -> Result<u64, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c mut <DB as Database>::Connection: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let mut builder = QueryBuilder::<DB>::new("DELETE FROM ");
    builder.push(quote_identifier(&schema.table_name));
    push_key_predicates(&mut builder, schema, key)?;
    if let Some(version) = expected_version {
        let version_column = version_column(schema)?;
        builder.push(" AND ");
        builder.push(quote_identifier(version_column));
        builder.push(" = ");
        builder.push_bind(read_model_i64_from_u64(
            DB::BACKEND,
            version,
            "expected version",
            DB::INTEGER_STORAGE,
        )?);
    }

    let result = builder
        .build()
        .execute(&mut **tx)
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "delete relational row", err))?;
    Ok(DB::rows_affected(&result))
}

pub(crate) fn push_key_predicates<DB: SqlxReadModelBackend>(
    builder: &mut QueryBuilder<DB>,
    schema: &TableSchema,
    key: &RowKey,
) -> Result<(), TableStoreError> {
    builder.push(" WHERE ");
    for (index, column_name) in schema.primary_key.columns.iter().enumerate() {
        if index > 0 {
            builder.push(" AND ");
        }
        let column = column_by_name(schema, column_name)?;
        let value = key.get(column_name).cloned().ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "read model `{}` row key is missing primary-key column `{}`",
                schema.model_name, column_name
            ))
        })?;
        builder.push(quote_identifier(column_name));
        builder.push(" = ");
        DB::push_row_value_bind(builder, value, column)?;
    }
    Ok(())
}

/// Build the shared `SELECT <columns>, <version> FROM <table>` prefix used by every
/// relational read. Per-column rendering is dialect-specific (`push_select_column`).
pub(crate) fn relational_row_select<DB: SqlxReadModelBackend>(
    schema: &TableSchema,
) -> Result<QueryBuilder<DB>, TableStoreError> {
    let version_column = version_column(schema)?;
    let mut builder = QueryBuilder::<DB>::new("SELECT ");
    for (index, column) in schema.columns.iter().enumerate() {
        if index > 0 {
            builder.push(", ");
        }
        DB::push_select_column(&mut builder, column);
    }
    if !schema.columns.is_empty() {
        builder.push(", ");
    }
    builder.push(quote_identifier(version_column));
    builder.push(" FROM ");
    builder.push(quote_identifier(&schema.table_name));
    Ok(builder)
}

pub(crate) fn push_order_by_primary_key<DB: SqlxReadModelBackend>(
    builder: &mut QueryBuilder<DB>,
    schema: &TableSchema,
) {
    if schema.primary_key.columns.is_empty() {
        return;
    }
    builder.push(" ORDER BY ");
    for (index, column) in schema.primary_key.columns.iter().enumerate() {
        if index > 0 {
            builder.push(", ");
        }
        builder.push(quote_identifier(column));
    }
}

pub(crate) fn row_to_versioned_values<DB>(
    schema: &TableSchema,
    row: &DB::Row,
) -> Result<Versioned<RowValues>, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let mut values = RowValues::new();
    for column in &schema.columns {
        values.insert(column.column_name.clone(), DB::row_value(row, column)?);
    }
    let version_column = version_column(schema)?;
    let version = read_model_u64_from_i64(
        DB::BACKEND,
        row.try_get::<i64, _>(version_column).map_err(|err| {
            read_model_storage_error(DB::BACKEND, "decode relational row version", err)
        })?,
        version_column,
    )?;
    Ok(Versioned {
        data: values,
        version,
    })
}

pub(crate) async fn load_relational_row_by_key<DB>(
    pool: &sqlx::Pool<DB>,
    schema: &TableSchema,
    key: &RowKey,
) -> Result<Option<Versioned<RowValues>>, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c sqlx::Pool<DB>: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    validate_key(schema, key)?;
    let mut builder = relational_row_select::<DB>(schema)?;
    push_key_predicates(&mut builder, schema, key)?;
    let row = builder
        .build()
        .fetch_optional(pool)
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "load relational row", err))?;
    row.map(|row| row_to_versioned_values::<DB>(schema, &row))
        .transpose()
}

pub(crate) async fn load_relationship_rows<DB>(
    pool: &sqlx::Pool<DB>,
    root_schema: &TableSchema,
    root_row: &RowValues,
    spec: &IncludeSpec,
) -> Result<Vec<Versioned<RowValues>>, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c sqlx::Pool<DB>: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    match spec.relationship.kind {
        RelationshipKind::HasMany => load_has_many_rows(pool, root_schema, root_row, spec).await,
        RelationshipKind::BelongsTo => {
            load_belongs_to_rows(pool, root_schema, root_row, spec).await
        }
        RelationshipKind::ManyToMany => Err(TableStoreError::Metadata(format!(
            "many-to-many relationship `{}` includes are not supported yet",
            spec.relationship.field_name
        ))),
    }
}

pub(crate) async fn load_read_model_graph<DB>(
    pool: &sqlx::Pool<DB>,
    schemas: &RwLock<TableSchemaRegistry>,
    request: ReadModelLoadRequest,
    capabilities: ReadModelQueryCapabilities,
) -> Result<ReadModelLoadGraph, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c sqlx::Pool<DB>: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    request.validate_for_query_capabilities(&capabilities)?;

    let (root_schema, include_specs) = resolve_registered_read_model_schemas(schemas, &request)?;
    validate_key(&root_schema, &request.key)?;

    let Some(root) = load_relational_row_by_key(pool, &root_schema, &request.key).await? else {
        return Ok(ReadModelLoadGraph::default());
    };

    let mut includes = BTreeMap::new();
    for spec in include_specs {
        let rows = load_relationship_rows(pool, &root_schema, &root.data, &spec).await?;
        includes.insert(
            spec.name,
            ReadModelIncludeRows {
                relationship: spec.relationship,
                target_schema: spec.target_schema,
                rows,
            },
        );
    }

    Ok(ReadModelLoadGraph {
        root: Some(root),
        includes,
    })
}

async fn load_has_many_rows<DB>(
    pool: &sqlx::Pool<DB>,
    root_schema: &TableSchema,
    root_row: &RowValues,
    spec: &IncludeSpec,
) -> Result<Vec<Versioned<RowValues>>, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c sqlx::Pool<DB>: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let foreign_key = spec.relationship.foreign_key.as_deref().ok_or_else(|| {
        TableStoreError::Metadata(format!(
            "relationship `{}` must declare a foreign key",
            spec.relationship.field_name
        ))
    })?;
    let target_column = crate::table::column_name_for(&spec.target_schema, foreign_key)
        .ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "relationship `{}` foreign key `{}` is not a target column",
                spec.relationship.field_name, foreign_key
            ))
        })?;
    let root_column = crate::table::column_name_for(root_schema, foreign_key)
        .or_else(|| root_schema.primary_key.columns.first().cloned())
        .ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "relationship `{}` has no root key column",
                spec.relationship.field_name
            ))
        })?;
    let root_value = root_row.get(&root_column).ok_or_else(|| {
        TableStoreError::Metadata(format!(
            "read model `{}` root row is missing relationship key `{}`",
            root_schema.model_name, root_column
        ))
    })?;

    load_rows_matching_column(pool, &spec.target_schema, &target_column, root_value).await
}

async fn load_belongs_to_rows<DB>(
    pool: &sqlx::Pool<DB>,
    root_schema: &TableSchema,
    root_row: &RowValues,
    spec: &IncludeSpec,
) -> Result<Vec<Versioned<RowValues>>, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c sqlx::Pool<DB>: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let foreign_key = spec.relationship.foreign_key.as_deref().ok_or_else(|| {
        TableStoreError::Metadata(format!(
            "relationship `{}` must declare a foreign key",
            spec.relationship.field_name
        ))
    })?;
    let source_column =
        crate::table::column_name_for(root_schema, foreign_key).ok_or_else(|| {
            TableStoreError::Metadata(format!(
                "relationship `{}` foreign key `{}` is not a source column",
                spec.relationship.field_name, foreign_key
            ))
        })?;
    let target_column = belongs_to_target_column(&spec.target_schema, &source_column)?;
    let source_value = root_row.get(&source_column).ok_or_else(|| {
        TableStoreError::Metadata(format!(
            "read model `{}` root row is missing relationship key `{}`",
            root_schema.model_name, source_column
        ))
    })?;

    load_rows_matching_column(pool, &spec.target_schema, &target_column, source_value).await
}

async fn load_rows_matching_column<DB>(
    pool: &sqlx::Pool<DB>,
    schema: &TableSchema,
    column_name: &str,
    value: &RowValue,
) -> Result<Vec<Versioned<RowValues>>, TableStoreError>
where
    DB: SqlxReadModelBackend,
    for<'c> &'c sqlx::Pool<DB>: Executor<'c, Database = DB>,
    <DB as Database>::Arguments: IntoArguments<DB>,
    for<'q> i64: Encode<'q, DB> + Type<DB> + sqlx::Decode<'q, DB>,
    for<'r> &'r str: sqlx::ColumnIndex<<DB as Database>::Row>,
{
    let column = column_by_name(schema, column_name)?;
    let mut builder = relational_row_select::<DB>(schema)?;
    builder.push(" WHERE ");
    builder.push(quote_identifier(column_name));
    builder.push(" = ");
    DB::push_row_value_bind(&mut builder, value.clone(), column)?;
    push_order_by_primary_key(&mut builder, schema);

    let rows = builder
        .build()
        .fetch_all(pool)
        .await
        .map_err(|err| read_model_storage_error(DB::BACKEND, "load relationship rows", err))?;

    rows.iter()
        .map(|row| row_to_versioned_values::<DB>(schema, row))
        .collect()
}