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