vantage_table/traits/

table_source.rs

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};

/// Trait for table data sources that defines column type separate from execution
/// TableSource represents a data source that can create and manage tables
#[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;

    /// 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"
        ))
    }

    /// 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 for a table (e.g., searches across searchable fields)
    ///
    /// Different vendors implement search differently:
    /// - SQL: `field LIKE '%value%'` (returns Expression)
    /// - SurrealDB: `field CONTAINS 'value'` (returns Expression)
    /// - MongoDB: `{ field: { $regex: 'value' } }` (returns MongoCondition)
    ///
    /// The implementation should search across appropriate fields in the table.
    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 as Record value (for WritableValueSet implementation)
    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 as Record value (for WritableValueSet implementation)
    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;

    /// Patch a record as Record value (for WritableValueSet implementation)
    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 (for WritableValueSet implementation)
    async fn delete_table_value<E>(&self, table: &Table<Self, E>, id: &Self::Id) -> Result<()>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Delete all records (for WritableValueSet implementation)
    async fn delete_table_all_values<E>(&self, table: &Table<Self, E>) -> Result<()>
    where
        E: Entity<Self::Value>,
        Self: Sized;

    /// Insert a record and return generated ID (for InsertableValueSet implementation)
    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` for embedding correlated subqueries inside SELECT
    /// expressions (e.g. `(SELECT COUNT(*) FROM order WHERE order.client_id = client.id)`).
    ///
    /// SQL backends produce table-qualified equality; non-SQL backends may not support
    /// correlated subqueries and should leave the default (which panics).
    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")
    }

    /// 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;
}