vantage_vista/source.rs
1use std::pin::Pin;
2#[cfg(feature = "rhai")]
3use vantage_rhai::rhai;
4
5use async_trait::async_trait;
6use ciborium::Value as CborValue;
7use futures_core::Stream;
8use indexmap::IndexMap;
9use vantage_core::{Result, VantageError, error};
10use vantage_types::Record;
11
12use crate::{
13 aggregate::AggregateSpec,
14 capabilities::VistaCapabilities,
15 column::Column,
16 reference::{ContainedSpec, Reference},
17 sort::SortDirection,
18 vista::Vista,
19};
20
21/// A single change observed on the underlying set by a live subscription.
22///
23/// This is the *push* counterpart to `list_vista_values`: a driver whose
24/// backend can stream changes (SurrealDB LIVE, Postgres `LISTEN/NOTIFY`,
25/// a Mongo change stream) emits one of these per affected row. The `value`
26/// carried on `Inserted`/`Updated` is the record in the **same projected,
27/// id-keyed shape** the driver returns from `list_vista_values`, so consumers
28/// can drop it straight into a cache without re-reading.
29#[derive(Debug, Clone)]
30pub enum VistaChange {
31 /// A row entered the set.
32 Inserted {
33 id: String,
34 value: Record<CborValue>,
35 },
36 /// An existing row's contents changed.
37 Updated {
38 id: String,
39 value: Record<CborValue>,
40 },
41 /// A row left the set.
42 Deleted { id: String },
43 /// "Something changed, but I can't say what" — a coarse invalidation. The
44 /// consumer should reconcile by re-reading the whole set. This is what a
45 /// payload-less push (Postgres `LISTEN/NOTIFY`) can offer; drivers that
46 /// carry the row (SurrealDB LIVE) emit the fine-grained variants instead.
47 Invalidated,
48}
49
50impl VistaChange {
51 /// The id of the affected row, or `None` for a coarse [`Invalidated`](Self::Invalidated).
52 pub fn id(&self) -> Option<&str> {
53 match self {
54 VistaChange::Inserted { id, .. }
55 | VistaChange::Updated { id, .. }
56 | VistaChange::Deleted { id } => Some(id),
57 VistaChange::Invalidated => None,
58 }
59 }
60}
61
62/// A stream of [`VistaChange`]s from a live subscription. `'static` and `Send`
63/// so it can be handed to a background task.
64pub type VistaChangeStream = Pin<Box<dyn Stream<Item = Result<VistaChange>> + Send>>;
65
66/// The stream [`TableShell::stream_vista_values`] returns.
67pub type VistaRowStream<'a> =
68 Pin<Box<dyn Stream<Item = Result<(String, Record<CborValue>)>> + Send + 'a>>;
69
70/// The rows `list` resolves to, as a stream, or its error as the only item:
71/// how [`TableShell::stream_vista_values`] answers by default.
72pub fn stream_from_list<'a>(
73 list: impl Future<Output = Result<IndexMap<String, Record<CborValue>>>> + Send + 'a,
74) -> VistaRowStream<'a> {
75 Box::pin(async_stream::stream! {
76 match list.await {
77 Ok(map) => {
78 for item in map {
79 yield Ok(item);
80 }
81 }
82 Err(e) => yield Err(e),
83 }
84 })
85}
86
87/// Per-driver executor for a `Vista`.
88///
89/// Implementations live in driver crates (vantage-sqlite, vantage-mongodb,
90/// vantage-aws, etc.). Each method receives `&Vista` so the driver can read
91/// the current condition state, columns, and other metadata.
92///
93/// `Id = String` and `Value = ciborium::Value` at this boundary, so every
94/// driver's native id (Mongo `ObjectId`, Surreal `Thing`, …) stringifies
95/// here. Methods are named with the `_vista_` infix to mirror
96/// `TableSource`'s `_table_` convention; `Vista`'s `ValueSet` impls
97/// delegate by stripping the infix.
98///
99/// `id: &String` (rather than `&str`) is intentional: the upstream
100/// `vantage_dataset::ValueSet` trait family fixes `Id = String` and uses
101/// `&Self::Id` in its signatures, so impls receive `&String` and forward
102/// it through unchanged.
103#[async_trait]
104#[allow(clippy::ptr_arg)]
105pub trait TableShell: Send + Sync + 'static {
106 // ---- Schema --------------------------------------------------------------
107 //
108 // The shell owns the schema. `Vista` is a thin wrapper that forwards its
109 // metadata accessors here. No defaults — every impl must answer (an empty
110 // schema is a deliberate choice the impl declares explicitly).
111
112 fn columns(&self) -> &IndexMap<String, Column>;
113
114 fn references(&self) -> &IndexMap<String, Reference>;
115
116 fn id_column(&self) -> Option<&str>;
117
118 // ---- ReadableValueSet delegates ----------------------------------------
119
120 async fn list_vista_values(&self, vista: &Vista)
121 -> Result<IndexMap<String, Record<CborValue>>>;
122
123 async fn get_vista_value(
124 &self,
125 vista: &Vista,
126 id: &String,
127 ) -> Result<Option<Record<CborValue>>>;
128
129 /// Fetch one record by id, with the caller's existing (cheap) record
130 /// available to drivers that can use it (e.g. a cmd detail script reading
131 /// list-pass columns). The default ignores `row` and delegates to
132 /// [`get_vista_value`](Self::get_vista_value); only drivers that benefit
133 /// override it.
134 async fn get_vista_value_with_row(
135 &self,
136 vista: &Vista,
137 id: &String,
138 _row: &Record<CborValue>,
139 ) -> Result<Option<Record<CborValue>>> {
140 self.get_vista_value(vista, id).await
141 }
142
143 async fn get_vista_some_value(
144 &self,
145 vista: &Vista,
146 ) -> Result<Option<(String, Record<CborValue>)>>;
147
148 /// Default implementation wraps `list_vista_values`. Drivers with native
149 /// streaming (cursor-based queries, paginated REST APIs) override.
150 fn stream_vista_values<'a>(&'a self, vista: &'a Vista) -> VistaRowStream<'a>
151 where
152 Self: Sync,
153 {
154 stream_from_list(self.list_vista_values(vista))
155 }
156
157 // ---- WritableValueSet delegates ----------------------------------------
158 //
159 // Default impls return a typed VantageError via `default_error` — drivers
160 // override only what they actually support. The matching `VistaCapabilities`
161 // flag must be set to `true` for any method the driver implements; if the
162 // flag is `true` but the trait method falls through to the default,
163 // `default_error` produces an `Unimplemented`-kind error (placeholder
164 // detected). If the flag is `false`, it produces `Unsupported`. Both are
165 // emitted as tracing events at construction.
166
167 /// Every write method follows the write contract on
168 /// `vantage_dataset::WritableValueSet` over this shell's own narrowing:
169 /// confined, idempotent, invariant-filling. A shell that implements them
170 /// must advertise [`can_confine_writes`](VistaCapabilities::can_confine_writes);
171 /// one that can't confine leaves it `false` and `Vista` refuses narrowed
172 /// writes.
173 ///
174 /// Insert: an id already in the set returns the stored row untouched; an
175 /// id held outside the set is `Conflict`; a new row is conformed to the
176 /// set's equality conditions.
177 async fn insert_vista_value(
178 &self,
179 _vista: &Vista,
180 _id: &String,
181 _record: &Record<CborValue>,
182 ) -> Result<Record<CborValue>> {
183 Err(self.default_error("insert_vista_value", "can_insert"))
184 }
185
186 /// Replace: a row in the set is replaced, a missing one created, both
187 /// conformed to the set; an id held outside the set is `Conflict`.
188 async fn replace_vista_value(
189 &self,
190 _vista: &Vista,
191 _id: &String,
192 _record: &Record<CborValue>,
193 ) -> Result<Record<CborValue>> {
194 Err(self.default_error("replace_vista_value", "can_update"))
195 }
196
197 /// Insert row `id`, or replace it if it exists. The default replaces and
198 /// falls back to insert on `NotFound`; stores with a native upsert override it.
199 ///
200 /// Upsert: same rows as replace — an id held outside the set is `Conflict`.
201 async fn upsert_vista_value(
202 &self,
203 vista: &Vista,
204 id: &String,
205 record: &Record<CborValue>,
206 ) -> Result<Record<CborValue>> {
207 match self.replace_vista_value(vista, id, record).await {
208 Err(e) if e.is_not_found() => self.insert_vista_value(vista, id, record).await,
209 other => other,
210 }
211 }
212
213 /// Patch: applied to a row in the set; a row outside the set or missing
214 /// is `NotFound`; a result that leaves the set is `Conflict`.
215 async fn patch_vista_value(
216 &self,
217 _vista: &Vista,
218 _id: &String,
219 _partial: &Record<CborValue>,
220 ) -> Result<Record<CborValue>> {
221 Err(self.default_error("patch_vista_value", "can_update"))
222 }
223
224 /// Delete: removes the row only when it is in the set; `Ok(())` whether it
225 /// was in the set, outside it, or missing.
226 async fn delete_vista_value(&self, _vista: &Vista, _id: &String) -> Result<()> {
227 Err(self.default_error("delete_vista_value", "can_delete"))
228 }
229
230 /// Delete all: removes exactly the rows in the set.
231 async fn delete_vista_all_values(&self, _vista: &Vista) -> Result<()> {
232 Err(self.default_error("delete_vista_all_values", "can_delete"))
233 }
234
235 // ---- InsertableValueSet delegate ---------------------------------------
236
237 /// Insert with a backend-made id: the record is conformed to the set like
238 /// insert. Not retry-safe — a retry makes a second row.
239 async fn insert_vista_return_id_value(
240 &self,
241 _vista: &Vista,
242 _record: &Record<CborValue>,
243 ) -> Result<String> {
244 Err(self.default_error("insert_vista_return_id_value", "can_insert"))
245 }
246
247 // ---- Bulk import -------------------------------------------------------
248
249 /// Store `records` (id → record) in one driver-native operation —
250 /// SQL COPY, Surreal batch insert — and return how many were
251 /// **newly inserted**. An id the table already held counts zero,
252 /// whether the driver's batch skips it or upserts it: the number
253 /// travels to a user as "n records created", and the per-record
254 /// fallback in `Dio::import_values` counts the same way. A driver
255 /// that cannot tell new from existing must not advertise
256 /// [`can_import`](VistaCapabilities::can_import).
257 ///
258 /// All-or-nothing is the other half of the contract: a driver that
259 /// cannot make the batch atomic must not advertise it either; the
260 /// caller then falls back to per-record inserts where partial
261 /// progress is honest and reportable.
262 ///
263 /// Import: each record follows the insert row of the write contract.
264 async fn import_vista_values(
265 &self,
266 _vista: &Vista,
267 _records: &IndexMap<String, Record<CborValue>>,
268 ) -> Result<usize> {
269 Err(self.default_error("import_vista_values", "can_import"))
270 }
271
272 // ---- Aggregates --------------------------------------------------------
273
274 /// Default impl falls back to `list_vista_values` — drivers with native
275 /// count (`SELECT COUNT(*)`, etc.) override.
276 async fn get_vista_count(&self, vista: &Vista) -> Result<i64> {
277 Ok(self.list_vista_values(vista).await?.len() as i64)
278 }
279
280 /// Derive a **new vista** that reduces this one — the driver's equivalent
281 /// of selecting from a subquery.
282 ///
283 /// An aggregation is not a value, it is a different set: `count(*)` yields
284 /// one row, `GROUP BY` yields one per group, and either can then be
285 /// conditioned, ordered or counted like any other set. Returning a
286 /// [`Vista`] is what lets every consumer keep the single shape it already
287 /// handles instead of growing a scalar special case.
288 ///
289 /// `Ok(None)` means *this driver cannot answer this request* — not that
290 /// the result is empty. The caller then reduces locally, which is a
291 /// different question (the rows it holds, not every row that matches) with
292 /// a different answer, so "can't" must never collapse into a number.
293 ///
294 /// **Narrow before aggregating.** Conditions belong to the source, applied
295 /// with `add_eq_condition` before this call — the order SQL uses, where the
296 /// filter is the inner query and the aggregate selects from its result.
297 /// Narrowing reports its own failure, so a driver is never handed a filter
298 /// it would silently ignore.
299 ///
300 /// **The returned vista's capabilities describe the DERIVED set, not the
301 /// source.** In particular it must not advertise condition support unless
302 /// the driver really implements it: adding a condition to an aggregate is
303 /// `HAVING`, a different operation over different values, and inheriting
304 /// the source's flag would promise a filter that silently does nothing.
305 /// An aggregator holding its entire output in memory is the exception —
306 /// it can filter what it produced, and may say so.
307 ///
308 /// This is construction, not a query — nothing is fetched until someone
309 /// lists the returned vista.
310 fn aggregate_vista(&self, _vista: &Vista, _spec: &AggregateSpec) -> Result<Option<Vista>> {
311 Ok(None)
312 }
313
314 // ---- Conditions --------------------------------------------------------
315
316 /// Translate `field == value` into the driver's native condition type and
317 /// apply it to the wrapped table. The default impl returns `Unimplemented`
318 /// — every driver is expected to override.
319 ///
320 /// `value` is the universal CBOR carrier; the driver picks the appropriate
321 /// translation (e.g. `cbor_to_bson` for Mongo, `cbor → AnyCsvType` for CSV).
322 fn add_eq_condition(&mut self, _field: &str, _value: &CborValue) -> Result<()> {
323 Err(error!(
324 "add_eq_condition not implemented",
325 method = "add_eq_condition",
326 source_type = std::any::type_name::<Self>()
327 )
328 .mark_unimplemented()
329 .traced())
330 }
331
332 /// Translate `field <op> value` into the driver's native condition and
333 /// apply it. The default routes `Eq` to [`add_eq_condition`](Self::add_eq_condition)
334 /// (so every driver gets equality for free) and returns `Unimplemented`
335 /// for every richer operator. Drivers whose query language expresses the
336 /// operators (SQL, SurrealDB) override this and advertise
337 /// [`can_filter_operators`](crate::VistaCapabilities::can_filter_operators);
338 /// consumers that see `false` skip the call and filter locally instead.
339 fn add_op_condition(
340 &mut self,
341 field: &str,
342 op: crate::FilterOp,
343 value: &CborValue,
344 ) -> Result<()> {
345 match op {
346 crate::FilterOp::Eq => self.add_eq_condition(field, value),
347 _ => Err(error!(
348 "add_op_condition operator not implemented",
349 method = "add_op_condition",
350 operator = format!("{:?}", op),
351 source_type = std::any::type_name::<Self>()
352 )
353 .mark_unimplemented()
354 .traced()),
355 }
356 }
357
358 /// Push a driver-native condition into the wrapped table. The
359 /// caller boxes the condition as `dyn Any` and the driver
360 /// downcasts to its own `T::Condition`. Used by YAML-driven
361 /// relation traversal, where the factory constructs a
362 /// `DeferredFn`-bearing condition outside the value-set surface
363 /// (which only accepts scalar eq) and pushes it through this
364 /// channel. Default is `Unimplemented`.
365 fn add_raw_condition(
366 &mut self,
367 _condition: Box<dyn std::any::Any + Send + Sync>,
368 ) -> Result<()> {
369 Err(error!(
370 "add_raw_condition not implemented",
371 method = "add_raw_condition",
372 source_type = std::any::type_name::<Self>()
373 )
374 .mark_unimplemented()
375 .traced())
376 }
377
378 // ---- Pagination --------------------------------------------------------
379
380 /// Declare how many records constitute one page. Used by both
381 /// [`fetch_page`](Self::fetch_page) and [`fetch_next`](Self::fetch_next).
382 /// Default returns `default_error("set_page_size", "can_set_page_size")`.
383 fn set_page_size(&mut self, _size: usize) -> Result<()> {
384 Err(self.default_error("set_page_size", "can_set_page_size"))
385 }
386
387 /// Fetch a specific page (1-based) using offset-style pagination. The
388 /// per-page count comes from the most recent
389 /// [`set_page_size`](Self::set_page_size).
390 ///
391 /// Drivers without random-access pagination (DynamoDB, most token-paginated
392 /// REST APIs) leave the default in place, which produces `Unsupported`.
393 /// Callers should branch on `vista.capabilities().can_fetch_page` first.
394 async fn fetch_page(
395 &self,
396 _vista: &Vista,
397 _page: usize,
398 ) -> Result<Vec<(String, Record<CborValue>)>> {
399 Err(self.default_error("fetch_page", "can_fetch_page"))
400 }
401
402 /// Cursor-style chain fetch. Pass `None` on the first call; pass the
403 /// previous call's returned token on subsequent calls. Returned token is
404 /// `None` when the result set is exhausted.
405 ///
406 /// The token is **driver-private** — its shape is whatever the backend
407 /// finds convenient (DynamoDB `LastEvaluatedKey` as a CBOR map, REST
408 /// `nextToken` as `CborValue::Text`, offset-based as `CborValue::Integer`).
409 /// Consumers treat it as opaque and round-trip it back unchanged.
410 ///
411 /// Default returns `default_error("fetch_next", "can_fetch_next")`.
412 async fn fetch_next(
413 &self,
414 _vista: &Vista,
415 _token: Option<CborValue>,
416 ) -> Result<(Vec<(String, Record<CborValue>)>, Option<CborValue>)> {
417 Err(self.default_error("fetch_next", "can_fetch_next"))
418 }
419
420 /// Fetch the half-open row window `[offset, offset + limit)` in the
421 /// source's natural order. Offset-style like [`fetch_page`](Self::fetch_page)
422 /// but addressed by absolute row index rather than page number, so it
423 /// maps directly onto a diorama `on_load_chunk` `Range<usize>` — which
424 /// is *not* guaranteed page-aligned. This is the primitive a paged,
425 /// lazily-loaded grid drives on scroll.
426 ///
427 /// Drivers leave the default in place (producing `Unsupported`) until
428 /// they implement it; callers branch on
429 /// `vista.capabilities().can_fetch_window` first. Default returns
430 /// `default_error("fetch_window", "can_fetch_window")`.
431 async fn fetch_window(
432 &self,
433 _vista: &Vista,
434 _offset: usize,
435 _limit: usize,
436 ) -> Result<Vec<(String, Record<CborValue>)>> {
437 Err(self.default_error("fetch_window", "can_fetch_window"))
438 }
439
440 /// [`fetch_window`](Self::fetch_window), plus the grand total of matching
441 /// rows when this fetch already learned it.
442 ///
443 /// Paged sources typically report the total in every response envelope,
444 /// alongside the window's rows. A caller needing both — a lazily-loaded
445 /// grid sizing its scrollbar — would otherwise pay a second round trip for
446 /// a number the first reply already carried.
447 ///
448 /// Drivers that can answer override this. The default delegates and
449 /// reports `None`, so no existing driver changes and no caller is told a
450 /// total exists when it doesn't. `None` means "this fetch didn't say",
451 /// never "zero".
452 async fn fetch_window_counted(
453 &self,
454 vista: &Vista,
455 offset: usize,
456 limit: usize,
457 ) -> Result<(Vec<(String, Record<CborValue>)>, Option<i64>)> {
458 Ok((self.fetch_window(vista, offset, limit).await?, None))
459 }
460
461 // ---- Quicksearch -------------------------------------------------------
462
463 /// Apply a quicksearch filter — a single string the driver fans out across
464 /// the columns it considers searchable (typically those flagged
465 /// [`SEARCHABLE`](crate::flags::SEARCHABLE), but each driver decides).
466 ///
467 /// **Replace semantics**: calling `add_search` again wipes the previous
468 /// search filter before applying the new one. Default produces
469 /// `Unimplemented` (when `can_search: true`) or `Unsupported` (when
470 /// `can_search: false`).
471 fn add_search(&mut self, _text: &str) -> Result<()> {
472 Err(self.default_error("add_search", "can_search"))
473 }
474
475 /// Drop the search filter previously applied via
476 /// [`add_search`](Self::add_search). Default mirrors `add_search`.
477 fn clear_search(&mut self) -> Result<()> {
478 Err(self.default_error("clear_search", "can_search"))
479 }
480
481 // ---- Ordering ----------------------------------------------------------
482
483 /// Push a single ORDER BY clause onto the wrapped table.
484 ///
485 /// Vista's `add_order` is replace-semantics: the driver shell should clear
486 /// any previously-set order before pushing the new one. Default produces
487 /// `Unimplemented` (when `can_order: true`) or `Unsupported` (when
488 /// `can_order: false`).
489 fn add_order(&mut self, _field: &str, _dir: SortDirection) -> Result<()> {
490 Err(self.default_error("add_order", "can_order"))
491 }
492
493 /// Wipe every order clause. Default mirrors [`add_order`](Self::add_order).
494 fn clear_orders(&mut self) -> Result<()> {
495 Err(self.default_error("clear_orders", "can_order"))
496 }
497
498 // ---- Cloning -----------------------------------------------------------
499
500 /// Produce an independent copy of this shell, or `None` if the driver can't
501 /// be cloned cheaply. The copy must share the backing store / connection
502 /// (typically `Arc`) but own its own query state (conditions / order /
503 /// search) so a caller can narrow it — set an ORDER BY, add a WHERE — without
504 /// disturbing the original. This is how a consumer builds a per-view ordered
505 /// Vista to fetch from: `clone_shell()` → `add_order(...)` → `fetch_window`.
506 ///
507 /// Default `None`: drivers opt in only where a clone is genuinely cheap
508 /// (query state is small; the store is `Arc`-shared). Callers that get `None`
509 /// fall back to reading the shared shell and ordering client-side.
510 fn clone_shell(&self) -> Option<Box<dyn TableShell>> {
511 None
512 }
513
514 // ---- References --------------------------------------------------------
515
516 /// Resolve a same-persistence relation using a known source row, returning
517 /// the related table as a new `Vista`.
518 ///
519 /// Drivers override by forwarding into the wrapped typed `Table`'s
520 /// `get_ref_from_row::<EmptyEntity>(relation, &native_row)` and then
521 /// wrapping the result back as a `Vista` through the driver's factory.
522 /// The default returns `Unimplemented`. Cross-persistence refs are
523 /// handled one layer up by `vantage-vista-factory`'s `VistaCatalog`,
524 /// never here.
525 fn get_ref(&self, relation: &str, _row: &Record<CborValue>) -> Result<Vista> {
526 Err(error!(
527 "get_ref not implemented",
528 method = "get_ref",
529 relation = relation,
530 source_type = std::any::type_name::<Self>()
531 )
532 .mark_unimplemented()
533 .traced())
534 }
535
536 /// Build the **bare** target of a same-persistence relation as a `Vista` —
537 /// the table a new related row would be inserted into, with no join
538 /// condition applied. Used by Vista's nested insert to reach a has-one /
539 /// has-many child's destination.
540 ///
541 /// Drivers override by forwarding into the wrapped typed `Table`'s
542 /// `get_ref_target_erased(relation)` and wrapping the result back
543 /// through the driver's factory — the same path as [`get_ref`](Self::get_ref)
544 /// minus the row-derived condition. The default returns `Unimplemented`;
545 /// cross-persistence relations are rejected at the `Vista` layer before
546 /// this is reached.
547 fn get_ref_target(&self, relation: &str) -> Result<Vista> {
548 Err(error!(
549 "get_ref_target not implemented",
550 method = "get_ref_target",
551 relation = relation,
552 source_type = std::any::type_name::<Self>()
553 )
554 .mark_unimplemented()
555 .traced())
556 }
557
558 /// Contained (embedded-in-row) relations this shell exposes, keyed by name.
559 /// Default empty — only shells that model embedded objects/arrays override.
560 fn contained(&self) -> &IndexMap<String, ContainedSpec> {
561 static EMPTY: std::sync::OnceLock<IndexMap<String, ContainedSpec>> =
562 std::sync::OnceLock::new();
563 EMPTY.get_or_init(IndexMap::new)
564 }
565
566 /// Resolve a contained relation against a known parent `row`, returning the
567 /// embedded records as a sub-`Vista`. Writes to that sub-Vista patch the
568 /// host column of `row`'s record back through the shell. Default returns
569 /// `Unimplemented`; shells override to seed [`crate::build_contained_vista`]
570 /// with a writeback that patches the parent.
571 fn get_contained_ref(&self, relation: &str, _row: &Record<CborValue>) -> Result<Vista> {
572 Err(error!(
573 "get_contained_ref not implemented",
574 method = "get_contained_ref",
575 relation = relation,
576 source_type = std::any::type_name::<Self>()
577 )
578 .mark_unimplemented()
579 .traced())
580 }
581
582 /// Names + cardinalities of the shell's same-persistence references.
583 /// Derived from [`references`](Self::references) by default; impls
584 /// should rarely need to override.
585 fn get_ref_kinds(&self) -> Vec<(String, crate::reference::ReferenceKind)> {
586 self.references()
587 .iter()
588 .map(|(name, r)| (name.clone(), r.kind))
589 .collect()
590 }
591
592 // ---- Identity ----------------------------------------------------------
593
594 /// Short human label for the underlying driver (e.g. `"csv"`, `"sqlite"`,
595 /// `"postgres"`, `"mongodb"`). Used for diagnostics and CLI output.
596 /// Drivers should override; the default is a placeholder.
597 fn driver_name(&self) -> &'static str {
598 "unknown"
599 }
600
601 // ---- Preview -----------------------------------------------------------
602
603 /// Render the query this shell would send, **without sending it**.
604 ///
605 /// Free-form and driver-shaped: SQL answers with `sql`, GraphQL with a
606 /// `query` document, a cmd datasource with the argv it would spawn. The
607 /// only convention is a `driver` key naming the driver, so a reader can
608 /// tell what the rest of the object means.
609 ///
610 /// There is no default and no `Result`. No default, so a new driver is
611 /// forced to answer rather than inheriting silence; no `Result`, because
612 /// "there is no query" is an answer, not a failure — a driver that
613 /// generates its rows in-process says so in a `note` and is done.
614 ///
615 /// # Contract
616 ///
617 /// **Synchronous, and performs no I/O.** That is what makes preview safe
618 /// to expose where fetching is not. A value known only at fetch time (an
619 /// unresolved foreign-key narrowing) renders as a placeholder; it is never
620 /// awaited.
621 ///
622 /// **Lossy is acceptable, wrong is not.** A preview that omits a detail
623 /// should say so in a `note`. A preview showing a filter the driver would
624 /// not actually send is a bug.
625 fn preview_query(&self, vista: &Vista) -> serde_json::Value;
626
627 // ---- Scripting ---------------------------------------------------------
628
629 /// Contribute backend-specific vocabulary to a Rhai engine, registered
630 /// before vantage-vista's `DataVocab`. Backends with an expression engine
631 /// (SurrealDB, SQL) override this to register `ident`/`==`/`fx`/graph
632 /// constructors plus a `with_condition(<backend expr>)` verb on the table
633 /// handle (see `Handle::with_base_vista`) that routes a boxed native
634 /// condition through [`add_raw_condition`](Self::add_raw_condition).
635 ///
636 /// Default is a no-op: engine-less datasources (CSV/Mongo/REST) still get
637 /// the table handle verbs and only lose the vendor expression syntax.
638 #[cfg(feature = "rhai")]
639 fn register_rhai_extensions(&self, _engine: &mut rhai::Engine) {}
640
641 /// Vendor constants a script sees beside `row`/`self` — the per-evaluation
642 /// half of [`register_rhai_extensions`](Self::register_rhai_extensions).
643 /// SurrealDB pushes `me`, the current-record anchor. A value here is a
644 /// scope variable, never an engine hook, so it composes with the host's
645 /// own resolver.
646 ///
647 /// Default: `env` unchanged.
648 #[cfg(feature = "rhai")]
649 fn rhai_env(&self, env: vantage_rhai::Env) -> vantage_rhai::Env {
650 env
651 }
652
653 // ---- Live subscription -------------------------------------------------
654
655 /// Subscribe to changes on the set and stream them as [`VistaChange`]s.
656 ///
657 /// Drivers whose backend can push changes (SurrealDB LIVE, Postgres
658 /// `LISTEN/NOTIFY`) override this and advertise
659 /// [`can_subscribe`](VistaCapabilities::can_subscribe). The row-bearing
660 /// variants carry the record in the same projected shape as
661 /// [`list_vista_values`](Self::list_vista_values), so a consumer can apply
662 /// them to a cache directly; [`VistaChange::Invalidated`] carries nothing at
663 /// all and means "re-read the set". The default produces `Unimplemented` (when
664 /// `can_subscribe: true`) or `Unsupported` (when `false`); callers branch on
665 /// `vista.capabilities().can_subscribe` first.
666 ///
667 /// # The subscription contract
668 ///
669 /// Callers always pass the full `vista`, and drivers deliver on a best-effort
670 /// basis. A consumer must not need to know whether a given backend filters
671 /// row-, select- or table-wide — that is what keeps consumer code identical
672 /// across drivers, and lets a driver tighten its scope later without
673 /// breaking anyone. Four promises hold for every implementation:
674 ///
675 /// 1. **The stream may be coarser than the vista.** Subscribing table-wide
676 /// and letting the consumer discard what it doesn't want is a valid
677 /// implementation; `vista` is a hint about what's interesting, not a
678 /// filter the driver is obliged to apply.
679 /// 2. **Payload rows may fall outside the vista's conditions**, precisely
680 /// because of (1). Either the driver reconciles (SurrealDB re-reads each
681 /// notified id *through* the vista's conditions, so a row that no longer
682 /// matches surfaces as [`VistaChange::Deleted`]) or the consumer must.
683 /// Never assume an `Inserted`/`Updated` row belongs in the set.
684 /// 3. **[`VistaChange::Invalidated`] means "re-read everything".** It carries
685 /// no id and implies nothing about how much changed — a driver with no row
686 /// payload to offer may emit it for every single write.
687 /// 4. **Stream end is normal, not an error.** Connections drop and sessions
688 /// expire; consumers resubscribe (with backoff) and reconcile the gap. A
689 /// driver need not reconnect internally.
690 ///
691 /// Delivery is not guaranteed even while subscribed — see
692 /// [`can_subscribe`](VistaCapabilities::can_subscribe).
693 async fn watch_vista(&self, _vista: &Vista) -> Result<VistaChangeStream> {
694 Err(self.default_error("watch_vista", "can_subscribe"))
695 }
696
697 // ---- Capability advertisement -----------------------------------------
698
699 fn capabilities(&self) -> &VistaCapabilities;
700
701 /// Look up a capability flag by name. Used by `default_error` to decide
702 /// between `Unsupported` and `Unimplemented`. Drivers don't normally
703 /// need to override this.
704 fn capability_flag(&self, name: &str) -> bool {
705 let caps = self.capabilities();
706 match name {
707 "can_count" => caps.can_count,
708 "can_insert" => caps.can_insert,
709 "can_update" => caps.can_update,
710 "can_delete" => caps.can_delete,
711 "can_import" => caps.can_import,
712 "can_subscribe" => caps.can_subscribe,
713 "can_invalidate" => caps.can_invalidate,
714 "can_order" => caps.can_order,
715 "can_search" => caps.can_search,
716 "can_set_page_size" => caps.can_set_page_size,
717 "can_fetch_page" => caps.can_fetch_page,
718 "can_fetch_next" => caps.can_fetch_next,
719 "can_fetch_window" => caps.can_fetch_window,
720 "can_traverse_to_record" => caps.can_traverse_to_record,
721 "can_traverse_to_set" => caps.can_traverse_to_set,
722 "can_build_ref_via_script" => caps.can_build_ref_via_script,
723 "can_traverse_in_columns" => caps.can_traverse_in_columns,
724 "can_confine_writes" => caps.can_confine_writes,
725 _ => false,
726 }
727 }
728
729 /// Build the standard error returned by default trait method impls.
730 ///
731 /// Picks the kind based on the capability flag: a `true` flag means the
732 /// driver advertised support but didn't override the method (placeholder
733 /// → `Unimplemented`); a `false` flag means the driver honestly doesn't
734 /// claim the op (caller should have checked → `Unsupported`).
735 ///
736 /// Only the `Unimplemented` kind traces at error level — it's a driver
737 /// bug. An `Unsupported` refusal is a legitimate answer to a caller
738 /// probing a capability (e.g. an exploratory data script calling
739 /// `set_page_size` on a cache-mode vista): the error value carries the
740 /// full message to the caller, so it logs at debug only.
741 fn default_error(&self, method: &str, capability: &str) -> VantageError {
742 let source_type = std::any::type_name::<Self>();
743 if self.capability_flag(capability) {
744 error!(
745 "capability is advertised but its implementation is missing",
746 method = method,
747 capability = capability,
748 source_type = source_type
749 )
750 .mark_unimplemented()
751 .traced()
752 } else {
753 error!(
754 "capability is not supported; method refused",
755 method = method,
756 capability = capability,
757 source_type = source_type
758 )
759 .mark_unsupported()
760 .traced_debug()
761 }
762 }
763}