vantage_vista/

source.rs

use std::pin::Pin;
#[cfg(feature = "rhai")]
use vantage_rhai::rhai;

use async_trait::async_trait;
use ciborium::Value as CborValue;
use futures_core::Stream;
use indexmap::IndexMap;
use vantage_core::{Result, VantageError, error};
use vantage_types::Record;

use crate::{
    aggregate::AggregateSpec,
    capabilities::VistaCapabilities,
    column::Column,
    reference::{ContainedSpec, Reference},
    sort::SortDirection,
    vista::Vista,
};

/// A single change observed on the underlying set by a live subscription.
///
/// This is the *push* counterpart to `list_vista_values`: a driver whose
/// backend can stream changes (SurrealDB LIVE, Postgres `LISTEN/NOTIFY`,
/// a Mongo change stream) emits one of these per affected row. The `value`
/// carried on `Inserted`/`Updated` is the record in the **same projected,
/// id-keyed shape** the driver returns from `list_vista_values`, so consumers
/// can drop it straight into a cache without re-reading.
#[derive(Debug, Clone)]
pub enum VistaChange {
    /// A row entered the set.
    Inserted {
        id: String,
        value: Record<CborValue>,
    },
    /// An existing row's contents changed.
    Updated {
        id: String,
        value: Record<CborValue>,
    },
    /// A row left the set.
    Deleted { id: String },
    /// "Something changed, but I can't say what" — a coarse invalidation. The
    /// consumer should reconcile by re-reading the whole set. This is what a
    /// payload-less push (Postgres `LISTEN/NOTIFY`) can offer; drivers that
    /// carry the row (SurrealDB LIVE) emit the fine-grained variants instead.
    Invalidated,
}

impl VistaChange {
    /// The id of the affected row, or `None` for a coarse [`Invalidated`](Self::Invalidated).
    pub fn id(&self) -> Option<&str> {
        match self {
            VistaChange::Inserted { id, .. }
            | VistaChange::Updated { id, .. }
            | VistaChange::Deleted { id } => Some(id),
            VistaChange::Invalidated => None,
        }
    }
}

/// A stream of [`VistaChange`]s from a live subscription. `'static` and `Send`
/// so it can be handed to a background task.
pub type VistaChangeStream = Pin<Box<dyn Stream<Item = Result<VistaChange>> + Send>>;

/// Per-driver executor for a `Vista`.
///
/// Implementations live in driver crates (vantage-sqlite, vantage-mongodb,
/// vantage-aws, etc.). Each method receives `&Vista` so the driver can read
/// the current condition state, columns, and other metadata.
///
/// `Id = String` and `Value = ciborium::Value` at this boundary, so every
/// driver's native id (Mongo `ObjectId`, Surreal `Thing`, …) stringifies
/// here. Methods are named with the `_vista_` infix to mirror
/// `TableSource`'s `_table_` convention; `Vista`'s `ValueSet` impls
/// delegate by stripping the infix.
///
/// `id: &String` (rather than `&str`) is intentional: the upstream
/// `vantage_dataset::ValueSet` trait family fixes `Id = String` and uses
/// `&Self::Id` in its signatures, so impls receive `&String` and forward
/// it through unchanged.
#[async_trait]
#[allow(clippy::ptr_arg)]
pub trait TableShell: Send + Sync + 'static {
    // ---- Schema --------------------------------------------------------------
    //
    // The shell owns the schema. `Vista` is a thin wrapper that forwards its
    // metadata accessors here. No defaults — every impl must answer (an empty
    // schema is a deliberate choice the impl declares explicitly).

    fn columns(&self) -> &IndexMap<String, Column>;

    fn references(&self) -> &IndexMap<String, Reference>;

    fn id_column(&self) -> Option<&str>;

    // ---- ReadableValueSet delegates ----------------------------------------

    async fn list_vista_values(&self, vista: &Vista)
    -> Result<IndexMap<String, Record<CborValue>>>;

    async fn get_vista_value(
        &self,
        vista: &Vista,
        id: &String,
    ) -> Result<Option<Record<CborValue>>>;

    /// Fetch one record by id, with the caller's existing (cheap) record
    /// available to drivers that can use it (e.g. a cmd detail script reading
    /// list-pass columns). The default ignores `row` and delegates to
    /// [`get_vista_value`](Self::get_vista_value); only drivers that benefit
    /// override it.
    async fn get_vista_value_with_row(
        &self,
        vista: &Vista,
        id: &String,
        _row: &Record<CborValue>,
    ) -> Result<Option<Record<CborValue>>> {
        self.get_vista_value(vista, id).await
    }

    async fn get_vista_some_value(
        &self,
        vista: &Vista,
    ) -> Result<Option<(String, Record<CborValue>)>>;

    /// Default implementation wraps `list_vista_values`. Drivers with native
    /// streaming (cursor-based queries, paginated REST APIs) override.
    #[allow(clippy::type_complexity)]
    fn stream_vista_values<'a>(
        &'a self,
        vista: &'a Vista,
    ) -> Pin<Box<dyn Stream<Item = Result<(String, Record<CborValue>)>> + Send + 'a>>
    where
        Self: Sync,
    {
        Box::pin(async_stream::stream! {
            match self.list_vista_values(vista).await {
                Ok(map) => {
                    for item in map {
                        yield Ok(item);
                    }
                }
                Err(e) => yield Err(e),
            }
        })
    }

    // ---- WritableValueSet delegates ----------------------------------------
    //
    // Default impls return a typed VantageError via `default_error` — drivers
    // override only what they actually support. The matching `VistaCapabilities`
    // flag must be set to `true` for any method the driver implements; if the
    // flag is `true` but the trait method falls through to the default,
    // `default_error` produces an `Unimplemented`-kind error (placeholder
    // detected). If the flag is `false`, it produces `Unsupported`. Both are
    // emitted as tracing events at construction.

    async fn insert_vista_value(
        &self,
        _vista: &Vista,
        _id: &String,
        _record: &Record<CborValue>,
    ) -> Result<Record<CborValue>> {
        Err(self.default_error("insert_vista_value", "can_insert"))
    }

    async fn replace_vista_value(
        &self,
        _vista: &Vista,
        _id: &String,
        _record: &Record<CborValue>,
    ) -> Result<Record<CborValue>> {
        Err(self.default_error("replace_vista_value", "can_update"))
    }

    /// Insert row `id`, or replace it if it exists. The default replaces and
    /// falls back to insert on `NotFound`; stores with a native upsert override it.
    async fn upsert_vista_value(
        &self,
        vista: &Vista,
        id: &String,
        record: &Record<CborValue>,
    ) -> Result<Record<CborValue>> {
        match self.replace_vista_value(vista, id, record).await {
            Err(e) if e.is_not_found() => self.insert_vista_value(vista, id, record).await,
            other => other,
        }
    }

    async fn patch_vista_value(
        &self,
        _vista: &Vista,
        _id: &String,
        _partial: &Record<CborValue>,
    ) -> Result<Record<CborValue>> {
        Err(self.default_error("patch_vista_value", "can_update"))
    }

    async fn delete_vista_value(&self, _vista: &Vista, _id: &String) -> Result<()> {
        Err(self.default_error("delete_vista_value", "can_delete"))
    }

    async fn delete_vista_all_values(&self, _vista: &Vista) -> Result<()> {
        Err(self.default_error("delete_vista_all_values", "can_delete"))
    }

    // ---- InsertableValueSet delegate ---------------------------------------

    async fn insert_vista_return_id_value(
        &self,
        _vista: &Vista,
        _record: &Record<CborValue>,
    ) -> Result<String> {
        Err(self.default_error("insert_vista_return_id_value", "can_insert"))
    }

    // ---- Bulk import -------------------------------------------------------

    /// Store `records` (id → record) in one driver-native operation —
    /// SQL COPY, Surreal batch insert — and return how many were
    /// **newly inserted**. An id the table already held counts zero,
    /// whether the driver's batch skips it or upserts it: the number
    /// travels to a user as "n records created", and the per-record
    /// fallback in `Dio::import_values` counts the same way. A driver
    /// that cannot tell new from existing must not advertise
    /// [`can_import`](VistaCapabilities::can_import).
    ///
    /// All-or-nothing is the other half of the contract: a driver that
    /// cannot make the batch atomic must not advertise it either; the
    /// caller then falls back to per-record inserts where partial
    /// progress is honest and reportable.
    async fn import_vista_values(
        &self,
        _vista: &Vista,
        _records: &IndexMap<String, Record<CborValue>>,
    ) -> Result<usize> {
        Err(self.default_error("import_vista_values", "can_import"))
    }

    // ---- Aggregates --------------------------------------------------------

    /// Default impl falls back to `list_vista_values` — drivers with native
    /// count (`SELECT COUNT(*)`, etc.) override.
    async fn get_vista_count(&self, vista: &Vista) -> Result<i64> {
        Ok(self.list_vista_values(vista).await?.len() as i64)
    }

    /// Derive a **new vista** that reduces this one — the driver's equivalent
    /// of selecting from a subquery.
    ///
    /// An aggregation is not a value, it is a different set: `count(*)` yields
    /// one row, `GROUP BY` yields one per group, and either can then be
    /// conditioned, ordered or counted like any other set. Returning a
    /// [`Vista`] is what lets every consumer keep the single shape it already
    /// handles instead of growing a scalar special case.
    ///
    /// `Ok(None)` means *this driver cannot answer this request* — not that
    /// the result is empty. The caller then reduces locally, which is a
    /// different question (the rows it holds, not every row that matches) with
    /// a different answer, so "can't" must never collapse into a number.
    ///
    /// **Narrow before aggregating.** Conditions belong to the source, applied
    /// with `add_eq_condition` before this call — the order SQL uses, where the
    /// filter is the inner query and the aggregate selects from its result.
    /// Narrowing reports its own failure, so a driver is never handed a filter
    /// it would silently ignore.
    ///
    /// **The returned vista's capabilities describe the DERIVED set, not the
    /// source.** In particular it must not advertise condition support unless
    /// the driver really implements it: adding a condition to an aggregate is
    /// `HAVING`, a different operation over different values, and inheriting
    /// the source's flag would promise a filter that silently does nothing.
    /// An aggregator holding its entire output in memory is the exception —
    /// it can filter what it produced, and may say so.
    ///
    /// This is construction, not a query — nothing is fetched until someone
    /// lists the returned vista.
    fn aggregate_vista(&self, _vista: &Vista, _spec: &AggregateSpec) -> Result<Option<Vista>> {
        Ok(None)
    }

    // ---- Conditions --------------------------------------------------------

    /// Translate `field == value` into the driver's native condition type and
    /// apply it to the wrapped table. The default impl returns `Unimplemented`
    /// — every driver is expected to override.
    ///
    /// `value` is the universal CBOR carrier; the driver picks the appropriate
    /// translation (e.g. `cbor_to_bson` for Mongo, `cbor → AnyCsvType` for CSV).
    fn add_eq_condition(&mut self, _field: &str, _value: &CborValue) -> Result<()> {
        Err(error!(
            format!(
                "add_eq_condition not implemented for '{}'",
                std::any::type_name::<Self>()
            ),
            method = "add_eq_condition",
            source_type = std::any::type_name::<Self>()
        )
        .mark_unimplemented()
        .traced())
    }

    /// Translate `field <op> value` into the driver's native condition and
    /// apply it. The default routes `Eq` to [`add_eq_condition`](Self::add_eq_condition)
    /// (so every driver gets equality for free) and returns `Unimplemented`
    /// for every richer operator. Drivers whose query language expresses the
    /// operators (SQL, SurrealDB) override this and advertise
    /// [`can_filter_operators`](crate::VistaCapabilities::can_filter_operators);
    /// consumers that see `false` skip the call and filter locally instead.
    fn add_op_condition(
        &mut self,
        field: &str,
        op: crate::FilterOp,
        value: &CborValue,
    ) -> Result<()> {
        match op {
            crate::FilterOp::Eq => self.add_eq_condition(field, value),
            _ => Err(error!(
                format!(
                    "add_op_condition operator {:?} not implemented for '{}'",
                    op,
                    std::any::type_name::<Self>()
                ),
                method = "add_op_condition",
                operator = format!("{:?}", op),
                source_type = std::any::type_name::<Self>()
            )
            .mark_unimplemented()
            .traced()),
        }
    }

    /// Push a driver-native condition into the wrapped table. The
    /// caller boxes the condition as `dyn Any` and the driver
    /// downcasts to its own `T::Condition`. Used by YAML-driven
    /// relation traversal, where the factory constructs a
    /// `DeferredFn`-bearing condition outside the value-set surface
    /// (which only accepts scalar eq) and pushes it through this
    /// channel. Default is `Unimplemented`.
    fn add_raw_condition(
        &mut self,
        _condition: Box<dyn std::any::Any + Send + Sync>,
    ) -> Result<()> {
        Err(error!(
            format!(
                "add_raw_condition not implemented for '{}'",
                std::any::type_name::<Self>()
            ),
            method = "add_raw_condition",
            source_type = std::any::type_name::<Self>()
        )
        .mark_unimplemented()
        .traced())
    }

    // ---- Pagination --------------------------------------------------------

    /// Declare how many records constitute one page. Used by both
    /// [`fetch_page`](Self::fetch_page) and [`fetch_next`](Self::fetch_next).
    /// Default returns `default_error("set_page_size", "can_set_page_size")`.
    fn set_page_size(&mut self, _size: usize) -> Result<()> {
        Err(self.default_error("set_page_size", "can_set_page_size"))
    }

    /// Fetch a specific page (1-based) using offset-style pagination. The
    /// per-page count comes from the most recent
    /// [`set_page_size`](Self::set_page_size).
    ///
    /// Drivers without random-access pagination (DynamoDB, most token-paginated
    /// REST APIs) leave the default in place, which produces `Unsupported`.
    /// Callers should branch on `vista.capabilities().can_fetch_page` first.
    async fn fetch_page(
        &self,
        _vista: &Vista,
        _page: usize,
    ) -> Result<Vec<(String, Record<CborValue>)>> {
        Err(self.default_error("fetch_page", "can_fetch_page"))
    }

    /// Cursor-style chain fetch. Pass `None` on the first call; pass the
    /// previous call's returned token on subsequent calls. Returned token is
    /// `None` when the result set is exhausted.
    ///
    /// The token is **driver-private** — its shape is whatever the backend
    /// finds convenient (DynamoDB `LastEvaluatedKey` as a CBOR map, REST
    /// `nextToken` as `CborValue::Text`, offset-based as `CborValue::Integer`).
    /// Consumers treat it as opaque and round-trip it back unchanged.
    ///
    /// Default returns `default_error("fetch_next", "can_fetch_next")`.
    async fn fetch_next(
        &self,
        _vista: &Vista,
        _token: Option<CborValue>,
    ) -> Result<(Vec<(String, Record<CborValue>)>, Option<CborValue>)> {
        Err(self.default_error("fetch_next", "can_fetch_next"))
    }

    /// Fetch the half-open row window `[offset, offset + limit)` in the
    /// source's natural order. Offset-style like [`fetch_page`](Self::fetch_page)
    /// but addressed by absolute row index rather than page number, so it
    /// maps directly onto a diorama `on_load_chunk` `Range<usize>` — which
    /// is *not* guaranteed page-aligned. This is the primitive a paged,
    /// lazily-loaded grid drives on scroll.
    ///
    /// Drivers leave the default in place (producing `Unsupported`) until
    /// they implement it; callers branch on
    /// `vista.capabilities().can_fetch_window` first. Default returns
    /// `default_error("fetch_window", "can_fetch_window")`.
    async fn fetch_window(
        &self,
        _vista: &Vista,
        _offset: usize,
        _limit: usize,
    ) -> Result<Vec<(String, Record<CborValue>)>> {
        Err(self.default_error("fetch_window", "can_fetch_window"))
    }

    /// [`fetch_window`](Self::fetch_window), plus the grand total of matching
    /// rows when this fetch already learned it.
    ///
    /// Paged sources typically report the total in every response envelope,
    /// alongside the window's rows. A caller needing both — a lazily-loaded
    /// grid sizing its scrollbar — would otherwise pay a second round trip for
    /// a number the first reply already carried.
    ///
    /// Drivers that can answer override this. The default delegates and
    /// reports `None`, so no existing driver changes and no caller is told a
    /// total exists when it doesn't. `None` means "this fetch didn't say",
    /// never "zero".
    async fn fetch_window_counted(
        &self,
        vista: &Vista,
        offset: usize,
        limit: usize,
    ) -> Result<(Vec<(String, Record<CborValue>)>, Option<i64>)> {
        Ok((self.fetch_window(vista, offset, limit).await?, None))
    }

    // ---- Quicksearch -------------------------------------------------------

    /// Apply a quicksearch filter — a single string the driver fans out across
    /// the columns it considers searchable (typically those flagged
    /// [`SEARCHABLE`](crate::flags::SEARCHABLE), but each driver decides).
    ///
    /// **Replace semantics**: calling `add_search` again wipes the previous
    /// search filter before applying the new one. Default produces
    /// `Unimplemented` (when `can_search: true`) or `Unsupported` (when
    /// `can_search: false`).
    fn add_search(&mut self, _text: &str) -> Result<()> {
        Err(self.default_error("add_search", "can_search"))
    }

    /// Drop the search filter previously applied via
    /// [`add_search`](Self::add_search). Default mirrors `add_search`.
    fn clear_search(&mut self) -> Result<()> {
        Err(self.default_error("clear_search", "can_search"))
    }

    // ---- Ordering ----------------------------------------------------------

    /// Push a single ORDER BY clause onto the wrapped table.
    ///
    /// Vista's `add_order` is replace-semantics: the driver shell should clear
    /// any previously-set order before pushing the new one. Default produces
    /// `Unimplemented` (when `can_order: true`) or `Unsupported` (when
    /// `can_order: false`).
    fn add_order(&mut self, _field: &str, _dir: SortDirection) -> Result<()> {
        Err(self.default_error("add_order", "can_order"))
    }

    /// Wipe every order clause. Default mirrors [`add_order`](Self::add_order).
    fn clear_orders(&mut self) -> Result<()> {
        Err(self.default_error("clear_orders", "can_order"))
    }

    // ---- Cloning -----------------------------------------------------------

    /// Produce an independent copy of this shell, or `None` if the driver can't
    /// be cloned cheaply. The copy must share the backing store / connection
    /// (typically `Arc`) but own its own query state (conditions / order /
    /// search) so a caller can narrow it — set an ORDER BY, add a WHERE — without
    /// disturbing the original. This is how a consumer builds a per-view ordered
    /// Vista to fetch from: `clone_shell()` → `add_order(...)` → `fetch_window`.
    ///
    /// Default `None`: drivers opt in only where a clone is genuinely cheap
    /// (query state is small; the store is `Arc`-shared). Callers that get `None`
    /// fall back to reading the shared shell and ordering client-side.
    fn clone_shell(&self) -> Option<Box<dyn TableShell>> {
        None
    }

    // ---- References --------------------------------------------------------

    /// Resolve a same-persistence relation using a known source row, returning
    /// the related table as a new `Vista`.
    ///
    /// Drivers override by forwarding into the wrapped typed `Table`'s
    /// `get_ref_from_row::<EmptyEntity>(relation, &native_row)` and then
    /// wrapping the result back as a `Vista` through the driver's factory.
    /// The default returns `Unimplemented`. Cross-persistence refs are
    /// handled one layer up by `vantage-vista-factory`'s `VistaCatalog`,
    /// never here.
    fn get_ref(&self, relation: &str, _row: &Record<CborValue>) -> Result<Vista> {
        Err(error!(
            format!(
                "get_ref not implemented for '{}'",
                std::any::type_name::<Self>()
            ),
            method = "get_ref",
            relation = relation,
            source_type = std::any::type_name::<Self>()
        )
        .mark_unimplemented()
        .traced())
    }

    /// Build the **bare** target of a same-persistence relation as a `Vista` —
    /// the table a new related row would be inserted into, with no join
    /// condition applied. Used by Vista's nested insert to reach a has-one /
    /// has-many child's destination.
    ///
    /// Drivers override by forwarding into the wrapped typed `Table`'s
    /// `get_ref_target_erased(relation)` and wrapping the result back
    /// through the driver's factory — the same path as [`get_ref`](Self::get_ref)
    /// minus the row-derived condition. The default returns `Unimplemented`;
    /// cross-persistence relations are rejected at the `Vista` layer before
    /// this is reached.
    fn get_ref_target(&self, relation: &str) -> Result<Vista> {
        Err(error!(
            format!(
                "get_ref_target not implemented for '{}'",
                std::any::type_name::<Self>()
            ),
            method = "get_ref_target",
            relation = relation,
            source_type = std::any::type_name::<Self>()
        )
        .mark_unimplemented()
        .traced())
    }

    /// Contained (embedded-in-row) relations this shell exposes, keyed by name.
    /// Default empty — only shells that model embedded objects/arrays override.
    fn contained(&self) -> &IndexMap<String, ContainedSpec> {
        static EMPTY: std::sync::OnceLock<IndexMap<String, ContainedSpec>> =
            std::sync::OnceLock::new();
        EMPTY.get_or_init(IndexMap::new)
    }

    /// Resolve a contained relation against a known parent `row`, returning the
    /// embedded records as a sub-`Vista`. Writes to that sub-Vista patch the
    /// host column of `row`'s record back through the shell. Default returns
    /// `Unimplemented`; shells override to seed [`crate::build_contained_vista`]
    /// with a writeback that patches the parent.
    fn get_contained_ref(&self, relation: &str, _row: &Record<CborValue>) -> Result<Vista> {
        Err(error!(
            format!(
                "get_contained_ref not implemented for '{}'",
                std::any::type_name::<Self>()
            ),
            method = "get_contained_ref",
            relation = relation,
            source_type = std::any::type_name::<Self>()
        )
        .mark_unimplemented()
        .traced())
    }

    /// Names + cardinalities of the shell's same-persistence references.
    /// Derived from [`references`](Self::references) by default; impls
    /// should rarely need to override.
    fn get_ref_kinds(&self) -> Vec<(String, crate::reference::ReferenceKind)> {
        self.references()
            .iter()
            .map(|(name, r)| (name.clone(), r.kind))
            .collect()
    }

    // ---- Identity ----------------------------------------------------------

    /// Short human label for the underlying driver (e.g. `"csv"`, `"sqlite"`,
    /// `"postgres"`, `"mongodb"`). Used for diagnostics and CLI output.
    /// Drivers should override; the default is a placeholder.
    fn driver_name(&self) -> &'static str {
        "unknown"
    }

    // ---- Preview -----------------------------------------------------------

    /// Render the query this shell would send, **without sending it**.
    ///
    /// Free-form and driver-shaped: SQL answers with `sql`, GraphQL with a
    /// `query` document, a cmd datasource with the argv it would spawn. The
    /// only convention is a `driver` key naming the driver, so a reader can
    /// tell what the rest of the object means.
    ///
    /// There is no default and no `Result`. No default, so a new driver is
    /// forced to answer rather than inheriting silence; no `Result`, because
    /// "there is no query" is an answer, not a failure — a driver that
    /// generates its rows in-process says so in a `note` and is done.
    ///
    /// # Contract
    ///
    /// **Synchronous, and performs no I/O.** That is what makes preview safe
    /// to expose where fetching is not. A value known only at fetch time (an
    /// unresolved foreign-key narrowing) renders as a placeholder; it is never
    /// awaited.
    ///
    /// **Lossy is acceptable, wrong is not.** A preview that omits a detail
    /// should say so in a `note`. A preview showing a filter the driver would
    /// not actually send is a bug.
    fn preview_query(&self, vista: &Vista) -> serde_json::Value;

    // ---- Scripting ---------------------------------------------------------

    /// Contribute backend-specific vocabulary to a Rhai engine, registered
    /// before vantage-vista's `DataVocab`. Backends with an expression engine
    /// (SurrealDB, SQL) override this to register `ident`/`==`/`fx`/graph
    /// constructors plus a `with_condition(<backend expr>)` verb on the table
    /// handle (see `Handle::with_base_vista`) that routes a boxed native
    /// condition through [`add_raw_condition`](Self::add_raw_condition).
    ///
    /// Default is a no-op: engine-less datasources (CSV/Mongo/REST) still get
    /// the table handle verbs and only lose the vendor expression syntax.
    #[cfg(feature = "rhai")]
    fn register_rhai_extensions(&self, _engine: &mut rhai::Engine) {}

    /// Vendor constants a script sees beside `row`/`self` — the per-evaluation
    /// half of [`register_rhai_extensions`](Self::register_rhai_extensions).
    /// SurrealDB pushes `me`, the current-record anchor. A value here is a
    /// scope variable, never an engine hook, so it composes with the host's
    /// own resolver.
    ///
    /// Default: `env` unchanged.
    #[cfg(feature = "rhai")]
    fn rhai_env(&self, env: vantage_rhai::Env) -> vantage_rhai::Env {
        env
    }

    // ---- Live subscription -------------------------------------------------

    /// Subscribe to changes on the set and stream them as [`VistaChange`]s.
    ///
    /// Drivers whose backend can push changes (SurrealDB LIVE, Postgres
    /// `LISTEN/NOTIFY`) override this and advertise
    /// [`can_subscribe`](VistaCapabilities::can_subscribe). The row-bearing
    /// variants carry the record in the same projected shape as
    /// [`list_vista_values`](Self::list_vista_values), so a consumer can apply
    /// them to a cache directly; [`VistaChange::Invalidated`] carries nothing at
    /// all and means "re-read the set". The default produces `Unimplemented` (when
    /// `can_subscribe: true`) or `Unsupported` (when `false`); callers branch on
    /// `vista.capabilities().can_subscribe` first.
    ///
    /// # The subscription contract
    ///
    /// Callers always pass the full `vista`, and drivers deliver on a best-effort
    /// basis. A consumer must not need to know whether a given backend filters
    /// row-, select- or table-wide — that is what keeps consumer code identical
    /// across drivers, and lets a driver tighten its scope later without
    /// breaking anyone. Four promises hold for every implementation:
    ///
    /// 1. **The stream may be coarser than the vista.** Subscribing table-wide
    ///    and letting the consumer discard what it doesn't want is a valid
    ///    implementation; `vista` is a hint about what's interesting, not a
    ///    filter the driver is obliged to apply.
    /// 2. **Payload rows may fall outside the vista's conditions**, precisely
    ///    because of (1). Either the driver reconciles (SurrealDB re-reads each
    ///    notified id *through* the vista's conditions, so a row that no longer
    ///    matches surfaces as [`VistaChange::Deleted`]) or the consumer must.
    ///    Never assume an `Inserted`/`Updated` row belongs in the set.
    /// 3. **[`VistaChange::Invalidated`] means "re-read everything".** It carries
    ///    no id and implies nothing about how much changed — a driver with no row
    ///    payload to offer may emit it for every single write.
    /// 4. **Stream end is normal, not an error.** Connections drop and sessions
    ///    expire; consumers resubscribe (with backoff) and reconcile the gap. A
    ///    driver need not reconnect internally.
    ///
    /// Delivery is not guaranteed even while subscribed — see
    /// [`can_subscribe`](VistaCapabilities::can_subscribe).
    async fn watch_vista(&self, _vista: &Vista) -> Result<VistaChangeStream> {
        Err(self.default_error("watch_vista", "can_subscribe"))
    }

    // ---- Capability advertisement -----------------------------------------

    fn capabilities(&self) -> &VistaCapabilities;

    /// Look up a capability flag by name. Used by `default_error` to decide
    /// between `Unsupported` and `Unimplemented`. Drivers don't normally
    /// need to override this.
    fn capability_flag(&self, name: &str) -> bool {
        let caps = self.capabilities();
        match name {
            "can_count" => caps.can_count,
            "can_insert" => caps.can_insert,
            "can_update" => caps.can_update,
            "can_delete" => caps.can_delete,
            "can_import" => caps.can_import,
            "can_subscribe" => caps.can_subscribe,
            "can_invalidate" => caps.can_invalidate,
            "can_order" => caps.can_order,
            "can_search" => caps.can_search,
            "can_set_page_size" => caps.can_set_page_size,
            "can_fetch_page" => caps.can_fetch_page,
            "can_fetch_next" => caps.can_fetch_next,
            "can_fetch_window" => caps.can_fetch_window,
            "can_traverse_to_record" => caps.can_traverse_to_record,
            "can_traverse_to_set" => caps.can_traverse_to_set,
            "can_build_ref_via_script" => caps.can_build_ref_via_script,
            "can_traverse_in_columns" => caps.can_traverse_in_columns,
            _ => false,
        }
    }

    /// Build the standard error returned by default trait method impls.
    ///
    /// Picks the kind based on the capability flag: a `true` flag means the
    /// driver advertised support but didn't override the method (placeholder
    /// → `Unimplemented`); a `false` flag means the driver honestly doesn't
    /// claim the op (caller should have checked → `Unsupported`).
    ///
    /// Only the `Unimplemented` kind traces at error level — it's a driver
    /// bug. An `Unsupported` refusal is a legitimate answer to a caller
    /// probing a capability (e.g. an exploratory data script calling
    /// `set_page_size` on a cache-mode vista): the error value carries the
    /// full message to the caller, so it logs at debug only.
    fn default_error(&self, method: &str, capability: &str) -> VantageError {
        let source_type = std::any::type_name::<Self>();
        if self.capability_flag(capability) {
            error!(
                format!(
                    "'{}' is advertised as VistaCapability for '{}' but implementation for '{}' is missing",
                    capability, source_type, method
                ),
                method = method,
                capability = capability,
                source_type = source_type
            )
            .mark_unimplemented().traced()
        } else {
            error!(
                format!(
                    "'{}' is not supported by '{}'; '{}' refused",
                    capability, source_type, method
                ),
                method = method,
                capability = capability,
                source_type = source_type
            )
            .mark_unsupported()
            .traced_debug()
        }
    }
}