vantage-table 3.0.0

use std::hash::Hash;
use std::pin::Pin;

use async_trait::async_trait;
use futures_core::Stream;
use indexmap::IndexMap;
use vantage_dataset::traits::Result;
use vantage_expressions::{
    Expression,
    traits::associated_expressions::AssociatedExpression,
    traits::datasource::{DataSource, ExprDataSource},
    traits::expressive::ExpressiveEnum,
};
use vantage_types::{Entity, Record};

use crate::{column::core::ColumnType, table::Table, traits::column_like::ColumnLike};

/// A data source that holds tables: it names their column, value, id,
/// source and condition types, and runs `Table`'s reads and writes.
#[async_trait]
pub trait TableSource: DataSource + Clone + 'static {
    type Column<Type>: ColumnLike<Type> + Clone
    where
        Type: ColumnType;
    type AnyType: ColumnType;
    type Value: Clone + Send + Sync + 'static;
    type Id: Send + Sync + Clone + Hash + Eq + 'static;

    /// How this backend names its source. Most backends use `String`;
    /// SQL/SurrealDB backends use `SelectSource<Self::Select>` so a table can
    /// be sourced from an arbitrary sub-`SELECT`. See [`crate::source`].
    type Source: crate::traits::table_source_spec::TableSourceSpec;

    /// The condition type stored by `Table`. SQL/SurrealDB backends use
    /// `Expression<Self::Value>`; document-oriented backends like MongoDB
    /// can use a native filter type (e.g. `bson::Document`).
    type Condition: Clone + Send + Sync + 'static;

    /// Build a textual `field == value` condition.
    ///
    /// Generic UIs (e.g. CLI argument parsers) only have strings on
    /// hand, so they need a way to construct a `Self::Condition` without
    /// knowing the backend's native expression type. Backends that
    /// don't support raw text filtering can leave the default, which
    /// returns an error.
    fn eq_condition(field: &str, value: &str) -> Result<Self::Condition> {
        let _ = (field, value);
        Err(vantage_core::error!(
            "eq_condition not implemented for this TableSource"
        ))
    }

    /// Build a `field == value` condition with a typed `Self::Value`.
    ///
    /// Sibling of [`TableSource::eq_condition`] for the case where the caller already
    /// has a native-typed value in hand (e.g. `Reference::resolve_from_row`
    /// pulling a join value out of a `Record<Self::Value>`). Avoids going
    /// through string serialization and avoids the `Expression → T::Condition`
    /// coercion that some document backends only support via a panic stub.
    ///
    /// Backends that participate in row-based reference traversal override;
    /// the default returns an error so existing impls compile.
    fn eq_value_condition(&self, field: &str, value: Self::Value) -> Result<Self::Condition> {
        let _ = (field, value);
        Err(vantage_core::error!(
            "eq_value_condition not implemented for this TableSource"
        ))
    }

    /// Build a `field IN (values)` condition from native-typed values: the
    /// list-valued sibling of [`Self::eq_value_condition`].
    /// [`Self::condition_column`] names `field` for it. An empty `values`
    /// matches no row. The default returns an error.
    fn in_values_condition(
        &self,
        field: &str,
        values: Vec<Self::Value>,
    ) -> Result<Self::Condition> {
        let _ = (field, values);
        Err(vantage_core::error!(
            "in_values_condition not implemented for this TableSource"
        ))
    }

    /// When `condition` is a literal `column = value`, return the pair.
    ///
    /// [`Table::add_condition`] registers it as a set invariant: an insert or
    /// replace through the table fills an absent column, a patch may not change
    /// it. Expression backends use [`crate::conditions::literal_equality`];
    /// document backends read their own filter shape. Default `None`: the
    /// condition still narrows, it just fills nothing.
    fn condition_equality(&self, condition: &Self::Condition) -> Option<(String, Self::Value)> {
        let _ = condition;
        None
    }

    /// `condition` as text for people, e.g. `client_id = 7`: what a log entry
    /// records for a bulk delete. Not for parsing; use
    /// [`Self::condition_column`] / [`Self::condition_equality`] in code.
    /// A part resolved only at query time renders as a placeholder.
    fn describe_condition(&self, condition: &Self::Condition) -> String {
        let _ = condition;
        "<condition>".to_string()
    }

    /// The column `condition` narrows to values, for the shapes this backend
    /// can read (`col = v`, `col IN (…)`, comparisons, `LIKE`), with a value
    /// — not a column or an arbitrary expression — on the right. `None` when
    /// it can't tell — a raw expression, a combination, a negation (`!=`,
    /// `NOT IN`, `NOT LIKE`), `col = other_col` — so a check that requires a
    /// condition on some column fails safe.
    fn condition_column(&self, condition: &Self::Condition) -> Option<String> {
        let _ = condition;
        None
    }

    /// Whether every by-id write (`insert_table_value`, `replace_table_value`,
    /// `patch_table_value`, `delete_table_value`, `insert_table_return_id_value`)
    /// honours the write contract over `table.conditions()`: it only touches
    /// rows in the set, and only leaves rows that still belong to it. A source
    /// that returns `false` is refused any by-id write through a conditioned
    /// table: `Table` refuses it with `Unsupported` before reaching the source.
    /// Default `false`.
    fn can_confine_writes(&self) -> bool {
        false
    }

    /// Coerce a reference join value into the backend's native id form.
    ///
    /// Reference traversal (`resolve_from_row`) reads the join value raw out
    /// of the parent row and uses it for both the narrowing eq-condition and
    /// the target's insert invariant. On scalar-FK backends any value compares
    /// as-is, so the default is identity. Backends with a richer id type
    /// override: SurrealDB re-tags a `"table:key"` string — a record id that
    /// round-tripped through JSON or a script — back into a record id, so the
    /// condition renders a record literal (and an inserted child stores a
    /// link), not a quoted string that matches nothing.
    fn coerce_reference_value(&self, value: Self::Value) -> Self::Value {
        value
    }

    /// Create a new column with the given name
    fn create_column<Type: ColumnType>(&self, name: &str) -> Self::Column<Type>;

    /// Convert a typed column to type-erased column
    fn to_any_column<Type: ColumnType>(
        &self,
        column: Self::Column<Type>,
    ) -> Self::Column<Self::AnyType>;

    /// Attempt to convert a type-erased column back to typed column
    fn convert_any_column<Type: ColumnType>(
        &self,
        any_column: Self::Column<Self::AnyType>,
    ) -> Option<Self::Column<Type>>;

    /// Create an expression from a template and parameters, similar to Expression::new
    fn expr(
        &self,
        template: impl Into<String>,
        parameters: Vec<ExpressiveEnum<Self::Value>>,
    ) -> Expression<Self::Value>;

    /// Create a search condition matching `search_value` across the columns
    /// of [`Table::search_columns`]: those flagged `Searchable`, or every
    /// column when none is. A backend may leave out columns it can't filter
    /// on (dotted paths, imported columns). This is a *server-side*
    /// capability — it pushes the match down to the backend's query engine.
    ///
    /// Different vendors implement search differently:
    /// - SQL: `field LIKE '%value%'` (returns Expression)
    /// - SurrealDB: `field CONTAINS 'value'` (returns Expression)
    /// - MongoDB: `{ field: { $regex: 'value' } }` (returns MongoCondition)
    ///
    /// **Contract for backends that cannot search server-side** (e.g. CSV, redb):
    /// return a condition that yields an `ErrorKind::Unsupported` error when it is
    /// *resolved* — never a silent match-all (which leaks every row as if filtered)
    /// and never a panic. Searching loaded/in-memory data is the Lens/Diorama
    /// layer's responsibility, not the backend's.
    fn search_table_condition<E>(
        &self,
        table: &Table<Self, E>,
        search_value: &str,
    ) -> Self::Condition
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Get all data from a table as Record values with IDs (for ReadableValueSet implementation)
    async fn list_table_values<E>(
        &self,
        table: &Table<Self, E>,
    ) -> Result<IndexMap<Self::Id, Record<Self::Value>>>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Get a single record by ID as Record value (for ReadableValueSet implementation).
    ///
    /// Returns `Ok(None)` when no record exists with the given ID.
    async fn get_table_value<E>(
        &self,
        table: &Table<Self, E>,
        id: &Self::Id,
    ) -> Result<Option<Record<Self::Value>>>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Get some data from a table as Record value with ID (for ReadableValueSet implementation)
    async fn get_table_some_value<E>(
        &self,
        table: &Table<Self, E>,
    ) -> Result<Option<(Self::Id, Record<Self::Value>)>>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Get count of records in the table
    async fn get_table_count<E>(&self, table: &Table<Self, E>) -> Result<i64>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Get sum of a column in the table (returns native value type)
    async fn get_table_sum<E>(
        &self,
        table: &Table<Self, E>,
        column: &Self::Column<Self::AnyType>,
    ) -> Result<Self::Value>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Get maximum value of a column in the table (returns native value type)
    async fn get_table_max<E>(
        &self,
        table: &Table<Self, E>,
        column: &Self::Column<Self::AnyType>,
    ) -> Result<Self::Value>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Get minimum value of a column in the table (returns native value type)
    async fn get_table_min<E>(
        &self,
        table: &Table<Self, E>,
        column: &Self::Column<Self::AnyType>,
    ) -> Result<Self::Value>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Insert a record by ID. **Confined and idempotent:** when the id is
    /// already stored in `table.conditions()`, return the stored row unchanged;
    /// when it is held by a row outside the set, fail with `Conflict`; a new
    /// row must fit the conditions, else `Conflict` and nothing is stored.
    /// Refuse with `Unsupported` when a condition can be neither applied nor
    /// evaluated. See `WritableValueSet` for the table.
    async fn insert_table_value<E>(
        &self,
        table: &Table<Self, E>,
        id: &Self::Id,
        record: &Record<Self::Value>,
    ) -> Result<Record<Self::Value>>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Replace a record by ID, creating it when missing. **Confined and
    /// idempotent:** an id held by a row outside `table.conditions()` fails
    /// with `Conflict`; the written row must fit the conditions, else
    /// `Conflict` and nothing changes. Refuse with `Unsupported` when a
    /// condition can be neither applied nor evaluated. See `WritableValueSet`
    /// for the table.
    async fn replace_table_value<E>(
        &self,
        table: &Table<Self, E>,
        id: &Self::Id,
        record: &Record<Self::Value>,
    ) -> Result<Record<Self::Value>>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Merge `partial` into the row with this ID. **Confined:** a missing row
    /// or one outside `table.conditions()` is `NotFound` (marked); the merged
    /// row must still fit the conditions, else `Conflict` and nothing changes.
    /// Refuse with `Unsupported` when a condition can be neither applied nor
    /// evaluated. See `WritableValueSet` for the table.
    async fn patch_table_value<E>(
        &self,
        table: &Table<Self, E>,
        id: &Self::Id,
        partial: &Record<Self::Value>,
    ) -> Result<Record<Self::Value>>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Delete a record by ID. **Confined and idempotent:** delete only when the
    /// row is in `table.conditions()`; a missing row or a row outside the set
    /// is `Ok(())` (no lookup is needed — SQL: `DELETE … WHERE id = ? AND
    /// <conditions>`). Refuse with `Unsupported` when a condition can be
    /// neither applied nor evaluated. See `WritableValueSet` for the table.
    async fn delete_table_value<E>(&self, table: &Table<Self, E>, id: &Self::Id) -> Result<()>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Delete every record the table represents (for WritableValueSet).
    ///
    /// **Honour `table.conditions()`.** A conditioned table is a SUBSET, and
    /// every read path already treats it as one, so deleting it must delete
    /// that subset. An implementation that builds its statement from
    /// `table_name()` alone silently truncates the table instead — the caller
    /// narrowed to one row, and lost all of them. An unconditioned table
    /// still deletes everything, which is the only case this looks like.
    async fn delete_table_all_values<E>(&self, table: &Table<Self, E>) -> Result<()>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Insert a record under a generated ID and return it. **Confined, not
    /// retry-safe:** the record must fit `table.conditions()`, else `Conflict`
    /// and nothing is stored; a retry after a lost reply stores a second row.
    /// Refuse with `Unsupported` when a condition can be neither applied nor
    /// evaluated.
    async fn insert_table_return_id_value<E>(
        &self,
        table: &Table<Self, E>,
        record: &Record<Self::Value>,
    ) -> Result<Self::Id>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Stream all records from a table as (Id, Record) pairs.
    ///
    /// Default implementation wraps `list_table_values` into a stream.
    /// Backends with native streaming (e.g. REST APIs with pagination)
    /// can override this to yield records incrementally.
    #[allow(clippy::type_complexity)]
    fn stream_table_values<'a, E>(
        &'a self,
        table: &Table<Self, E>,
    ) -> Pin<Box<dyn Stream<Item = Result<(Self::Id, Record<Self::Value>)>> + Send + 'a>>
    where
        E: Entity<Self::Value> + 'a,
        Self: Sized,
    {
        let table = table.clone();
        Box::pin(async_stream::stream! {
            let records = self.list_table_values(&table).await;
            match records {
                Ok(map) => {
                    for item in map {
                        yield Ok(item);
                    }
                }
                Err(e) => yield Err(e),
            }
        })
    }

    /// Build a condition for "target_field IN (values of source_column from source_table)".
    ///
    /// Used by the relationship traversal system (`get_ref_as`) to filter
    /// a target table by foreign key values from a source table.
    ///
    /// Each backend implements this natively:
    /// - SQL backends build a subquery: `"id" IN (SELECT "bakery_id" FROM "client" WHERE ...)`
    /// - MongoDB builds a deferred `{ field: { "$in": [...] } }` document
    /// - CSV fetches values in memory and builds an IN condition
    fn related_in_condition<SourceE: Entity<Self::Value> + 'static>(
        &self,
        target_field: &str,
        source_table: &Table<Self, SourceE>,
        source_column: &str,
    ) -> Self::Condition
    where
        Self: Sized;

    /// Build a correlated condition: `target_table.target_field = source_table.source_column`.
    ///
    /// Used by `get_subquery_as` / `get_subquery_erased` for embedding correlated
    /// subqueries inside SELECT expressions (e.g.
    /// `(SELECT COUNT(*) FROM order WHERE order.client_id = client.id)`), and by
    /// the generic implicit-reference lowering. SQL and SurrealDB backends
    /// produce table-qualified equality; backends without correlated subqueries
    /// leave the default (which panics — they must also keep
    /// [`supports_traversal`](Self::supports_traversal) `false`).
    fn related_correlated_condition(
        &self,
        target_table: &str,
        target_field: &str,
        source_table: &str,
        source_column: &str,
    ) -> Self::Condition {
        let _ = (target_table, target_field, source_table, source_column);
        unimplemented!("correlated subqueries not supported by this backend")
    }

    /// Whether this backend can lower a dotted active column
    /// (`"country.name"`) into its own query — a nested correlated subquery for
    /// SQL, a native idiom path for SurrealDB, `$lookup` stages for MongoDB.
    /// Backends that can build none of these (e.g. CSV, REST) leave the
    /// default `false`, and
    /// [`Table::with_active_columns`](crate::table::Table::with_active_columns)
    /// rejects dotted names against them at build time rather than at fetch time.
    fn supports_traversal(&self) -> bool {
        false
    }

    /// Build a backend-native path expression for a dotted active column,
    /// tried before the generic correlated-subquery chain.
    ///
    /// `hops` holds each traversed relation's **foreign-key/link field** (not
    /// its registry name — a relation named `owner` over a link field `client`
    /// must lower to `client.…`), and `column` the final field. The default
    /// returns `None`, so backends fall back to the generic chain built on
    /// [`related_correlated_condition`](Self::related_correlated_condition).
    /// SurrealDB overrides this to emit an idiom path (`batch.golf_course.name`,
    /// each segment escaped separately); multi-hop comes for free. The returned
    /// expression is unaliased — `select()` aliases it to the dotted name.
    fn traversal_path_expr(&self, hops: &[&str], column: &str) -> Option<Expression<Self::Value>> {
        let _ = (hops, column);
        None
    }

    /// Return an associated expression that, when resolved, yields all values
    /// of the given typed column from this table (respecting current conditions).
    ///
    /// For query-language backends, this can be a subquery expression.
    /// For simple backends (CSV), this uses a `DeferredFn` that loads data
    /// and extracts the column values at execution time.
    ///
    /// The returned `AssociatedExpression` can be:
    /// - Executed directly: `.get().await -> Result<Vec<Type>>`
    /// - Composed into expressions: used via `Expressive` trait in `in_()` conditions
    ///
    /// ```rust,ignore
    /// let fk_col = source.get_column::<String>("bakery_id").unwrap();
    /// let fk_values = source.data_source().column_table_values_expr(&source, &fk_col);
    /// // Execute: let ids = fk_values.get().await?;
    /// // Or compose: target.add_condition(target["id"].in_((fk_values)));
    /// ```
    fn column_table_values_expr<'a, E, Type: ColumnType>(
        &'a self,
        table: &Table<Self, E>,
        column: &Self::Column<Type>,
    ) -> AssociatedExpression<'a, Self, Self::Value, Vec<Type>>
    where
        E: Entity<Self::Value> + 'static,
        Self: ExprDataSource<Self::Value> + Sized;
}