distributed 4.2.0

use crate::table::{
    key_fingerprint, validate_row_values, ExpectedVersion, RowKey, RowValue, RowValues,
    TableAdapterCapabilities, TableColumn, TableSchema, TableStoreError, TableWritePlan,
};

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('"', "\"\""))
}