Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The storage seam: two driver traits every backend implements, plus the
3 : //! backend-neutral types they speak (resource kinds, filter shapes, the
4 : //! store-mode enum). This crate names no backend — redb, sqlx and the
5 : //! browser's OPFS all live behind these traits in their own crates.
6 : //!
7 : //! Current-state and temporal storage are SEPARATE interfaces on purpose:
8 : //! a deployment may run postgres current-state with no temporal store at
9 : //! all, or memory current-state with a database-backed history. A driver
10 : //! that does not support an operation answers with an error (or a benign
11 : //! no-op for internal bookkeeping), never a panic — `NoTemporal` is the
12 : //! canonical instance.
13 : #![cfg_attr(not(test), warn(clippy::expect_used))]
14 : #![deny(missing_docs)]
15 :
16 : #[cfg(feature = "test-kit")]
17 : /// The shared driver contract every backend runs as a test. It asserts, so
18 : /// it panics on a store that breaks an invariant — that is the point.
19 : #[allow(clippy::expect_used)]
20 : pub mod contract;
21 : pub mod filter;
22 :
23 : use antares_model::{NgsiError, TenantId};
24 : use serde_json::Value;
25 :
26 : /// A stored document as the object it is. Every document a driver holds is a
27 : /// JSON object — `contract` asserts the shape on every backend — so a value
28 : /// that is not one means the driver returned something the contract forbids
29 : /// underneath a live request. That fails the one request; it never takes the
30 : /// process down, which is what an unwrap here would do.
31 4392 : pub fn stored_object(doc: &mut Value) -> Result<&mut serde_json::Map<String, Value>, NgsiError> {
32 4392 : doc.as_object_mut()
33 4392 : .ok_or_else(|| NgsiError::InternalError("stored document is not a JSON object".into()))
34 4392 : }
35 :
36 : /// What a [`ChangeHook`] call returns: the hook records temporal history
37 : /// through the temporal driver, which is asynchronous, so the driver that
38 : /// fires the hook awaits it before its own write returns.
39 : pub type HookFuture<'a> = std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'a>>;
40 :
41 : /// Called with (tenant, before, after) on every entity write — the local-mode
42 : /// change feed: create ⇒ (None, Some), delete ⇒ (Some, None). Shared rather
43 : /// than owned: a driver clones it out of its lock before awaiting, since a
44 : /// lock guard cannot be held across an await.
45 : pub type ChangeHook = std::sync::Arc<
46 : dyn for<'a> Fn(&'a TenantId, Option<Value>, Option<Value>) -> HookFuture<'a> + Send + Sync,
47 : >;
48 :
49 : /// Which resource family an operation touches.
50 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
51 : pub enum Kind {
52 : /// 5.2.4 Entity current-state documents.
53 : Entity,
54 : /// 5.2.12 Subscription documents.
55 : Subscription,
56 : /// 5.2.9 Context Source Registration documents.
57 : Registration,
58 : /// Context Source Registration Subscription documents (5.11).
59 : CSourceSubscription,
60 : /// Temporal Representation documents (5.2.5).
61 : Temporal,
62 : /// 5.16 Snapshot status documents (+ the internal synth-tenant index).
63 : Snapshot,
64 : /// 5.14 EntityMap API documents.
65 : EntityMap,
66 : /// 5.8.1.4 distributed-subscription mappings (remote ids per CSR).
67 : DistSub,
68 : /// Notifications a delivery policy gave up on (dead letters), kept
69 : /// under the subscription's tenant for replay.
70 : DeadLetter,
71 : }
72 :
73 : /// The four store backends, decided ONCE at startup and threaded as a value —
74 : /// never re-derived from strings or runtime probes, so a section gated on the
75 : /// wrong mode is unrepresentable.
76 : #[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
77 : pub enum StoreMode {
78 : /// In-process maps; nothing survives a restart.
79 : #[default]
80 : Memory,
81 : /// Embedded single-file store on local disk.
82 : File,
83 : /// PostgreSQL with PostGIS.
84 : Postgres,
85 : /// PostgreSQL with PostGIS and TimescaleDB hypertables for history.
86 : Timescale,
87 : }
88 :
89 : impl StoreMode {
90 : /// Every backend this workspace knows, in the order a listing shows
91 : /// them. One source of truth: `FromStr`, the unknown-mode message and
92 : /// the broker's built-with shelf all read it, so a backend added to the
93 : /// enum cannot go missing from any of them.
94 : pub const ALL: [StoreMode; 4] = [
95 : StoreMode::Memory,
96 : StoreMode::File,
97 : StoreMode::Postgres,
98 : StoreMode::Timescale,
99 : ];
100 :
101 : /// The accepted mode names, `memory|file|postgres|timescale`.
102 4 : pub fn names() -> String {
103 4 : StoreMode::ALL
104 4 : .iter()
105 16 : .map(|m| m.as_str())
106 4 : .collect::<Vec<_>>()
107 4 : .join("|")
108 4 : }
109 :
110 : /// The mode name as accepted by `FromStr` (`memory|file|postgres|timescale`).
111 432 : pub fn as_str(self) -> &'static str {
112 432 : match self {
113 151 : StoreMode::Memory => "memory",
114 107 : StoreMode::File => "file",
115 94 : StoreMode::Postgres => "postgres",
116 80 : StoreMode::Timescale => "timescale",
117 : }
118 432 : }
119 : /// Shared-database modes — the only ones that can back multiple instances.
120 39 : pub fn is_pg(self) -> bool {
121 39 : matches!(self, StoreMode::Postgres | StoreMode::Timescale)
122 39 : }
123 : }
124 :
125 : impl std::str::FromStr for StoreMode {
126 : type Err = String;
127 83 : fn from_str(s: &str) -> Result<Self, Self::Err> {
128 83 : StoreMode::ALL
129 83 : .into_iter()
130 156 : .find(|m| m.as_str() == s)
131 83 : .ok_or_else(|| format!("unknown store mode {s} ({})", StoreMode::names()))
132 83 : }
133 : }
134 :
135 : impl std::fmt::Display for StoreMode {
136 4 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
137 4 : f.write_str(self.as_str())
138 4 : }
139 : }
140 :
141 : /// What one tenant holds, per current-state document kind.
142 : #[derive(Debug, Clone, Default, PartialEq, Eq)]
143 : pub struct TenantStats {
144 : /// The tenant name.
145 : pub tenant: String,
146 : /// When the tenant row was created, if the backend records it.
147 : pub created_at: Option<String>,
148 : /// 5.2.4 entities.
149 : pub entities: u64,
150 : /// 5.2.12 subscriptions.
151 : pub subscriptions: u64,
152 : /// 5.2.9 context source registrations.
153 : pub registrations: u64,
154 : /// Context source registration subscriptions.
155 : pub csource_subscriptions: u64,
156 : /// 5.16 snapshots.
157 : pub snapshots: u64,
158 : /// Entity maps (5.14).
159 : pub entity_maps: u64,
160 : /// Distributed subscriptions this tenant placed at other brokers.
161 : pub dist_subs: u64,
162 : }
163 :
164 : /// The read-modify-write closure as the object-safe seam carries it. The
165 : /// typed `Result<T, E>` travels through [`CurrentStateDriverExt::mutate`]'s
166 : /// side slot; the boxed closure only signals commit (`Ok`) vs reject
167 : /// (`Err`) to the driver.
168 : pub type MutateFn<'a> = Box<dyn FnOnce(&mut Value) -> Result<(), ()> + Send + 'a>;
169 : /// Per-id variant for batch mutation; the driver calls it once per PRESENT
170 : /// id, in input order — the ext trait's error slot depends on that order.
171 : pub type BatchMutateFn<'a> = Box<dyn FnMut(&str, &mut Value) -> Result<(), ()> + Send + 'a>;
172 : /// The delivery stamp expressed as a `mutate`. This is the rule itself —
173 : /// what `timesSent`, `lastNotification`, `lastSuccess` and `status` become
174 : /// after one attempt (5.2.14.2) — so a backend that reimplements
175 : /// [`CurrentStateDriver::record_delivery`] in its own query language is
176 : /// reimplementing THIS, and the two must agree.
177 390 : pub async fn record_delivery_via_mutate(
178 390 : d: &(impl CurrentStateDriver + ?Sized),
179 390 : tenant: &TenantId,
180 390 : kind: Kind,
181 390 : id: &str,
182 390 : now: &str,
183 390 : ) -> Result<Option<Delivery>, NgsiError> {
184 390 : let mut out: Option<Delivery> = None;
185 390 : d.mutate::<(), ()>(tenant, kind, id, |doc| {
186 382 : let mut prev_success = None;
187 382 : if let Some(o) = doc.as_object_mut() {
188 382 : o.remove("status");
189 382 : }
190 382 : if let Some(n) = doc
191 382 : .as_object_mut()
192 382 : .and_then(|o| o.get_mut("notification"))
193 382 : .and_then(Value::as_object_mut)
194 382 : {
195 382 : let sent = n.get("timesSent").and_then(Value::as_i64).unwrap_or(0);
196 382 : n.insert("timesSent".into(), serde_json::json!(sent + 1));
197 382 : n.insert("lastNotification".into(), Value::String(now.to_owned()));
198 382 : prev_success = n.insert("lastSuccess".into(), Value::String(now.to_owned()));
199 382 : n.insert("status".into(), Value::String("ok".into()));
200 382 : }
201 382 : out = Some(Delivery {
202 382 : doc: doc.clone(),
203 382 : prev_success,
204 382 : });
205 382 : Ok(())
206 382 : })
207 390 : .await?;
208 390 : Ok(out)
209 390 : }
210 :
211 : /// The forward stamp expressed as a `mutate`. This is the Table 5.2.9-2
212 : /// rule itself — what `timesSent`, `timesFailed`, `lastSuccess`,
213 : /// `lastFailure` and `status` become after one distributed operation — so a
214 : /// backend that reimplements [`CurrentStateDriver::record_forward`] in its
215 : /// own query language is reimplementing THIS, and the two must agree.
216 : ///
217 : /// `timesSent` moves on EVERY attempt: the table defines it as the number of
218 : /// times the registration "triggered a distributed operation, including
219 : /// failed attempts". A member is written when it first has meaning, which is
220 : /// what cardinality 0..1 and "created on first successful operation" ask for:
221 : /// a registration that has only ever succeeded carries no `timesFailed` and
222 : /// no `lastFailure`, and one that has only ever failed carries no
223 : /// `lastSuccess`. `status` always names the LAST attempt.
224 435 : pub async fn record_forward_via_mutate(
225 435 : d: &(impl CurrentStateDriver + ?Sized),
226 435 : tenant: &TenantId,
227 435 : id: &str,
228 435 : now: &str,
229 435 : ok: bool,
230 435 : ) -> Result<Option<Value>, NgsiError> {
231 435 : let mut out: Option<Value> = None;
232 435 : d.mutate::<(), ()>(tenant, Kind::Registration, id, |doc| {
233 409 : if let Some(o) = doc.as_object_mut() {
234 409 : let sent = o.get("timesSent").and_then(Value::as_i64).unwrap_or(0);
235 409 : o.insert("timesSent".into(), serde_json::json!(sent + 1));
236 409 : if ok {
237 370 : o.insert("lastSuccess".into(), Value::String(now.to_owned()));
238 370 : o.insert("status".into(), Value::String("ok".into()));
239 370 : } else {
240 39 : let failed = o.get("timesFailed").and_then(Value::as_i64).unwrap_or(0);
241 39 : o.insert("timesFailed".into(), serde_json::json!(failed + 1));
242 39 : o.insert("lastFailure".into(), Value::String(now.to_owned()));
243 39 : o.insert("status".into(), Value::String("failed".into()));
244 39 : }
245 0 : }
246 409 : out = Some(doc.clone());
247 409 : Ok(())
248 409 : })
249 435 : .await?;
250 435 : Ok(out)
251 435 : }
252 :
253 : /// What one delivery attempt wrote: the stored subscription as it now
254 : /// stands (the mirror is fed from it) and the `lastSuccess` that was there
255 : /// before, which a failed attempt puts back.
256 : #[derive(Debug, Clone)]
257 : pub struct Delivery {
258 : /// The subscription as it now stands, the shape the mirror is fed.
259 : pub doc: Value,
260 : /// The `notification.lastSuccess` this attempt overwrote, absent when
261 : /// the subscription had never succeeded.
262 : pub prev_success: Option<Value>,
263 : }
264 :
265 : /// ADR-0021: the Tenant a stored `@context` row belongs to, or `None` when it
266 : /// belongs to none.
267 : ///
268 : /// 5.13.1 "Cached" is a copy of a public document the broker downloaded for
269 : /// whoever named its URL, so every Tenant reaches it. "Hosted" and
270 : /// "ImplicitlyCreated" are documents a client stored THROUGH a Tenant, and
271 : /// 5.5.7 makes their term mappings decide what that Tenant's payloads mean;
272 : /// the owner travels in the row. A row written before the member existed
273 : /// reads as the default Tenant's, and so does a row whose `kind` is missing
274 : /// altogether: no writer produces one, and an unrecognised row is safer read
275 : /// as somebody's than as everybody's.
276 8914 : pub fn context_row_owner(row: &Value) -> Option<&str> {
277 8914 : (row["kind"].as_str() != Some("Cached"))
278 8914 : .then(|| row["owner"].as_str().unwrap_or(TenantId::DEFAULT))
279 8914 : }
280 :
281 : /// Whether a call acting for `tenant` — `None` for no Tenant at all — reaches
282 : /// the row. The rule [`context_row_owner`] states, in the form the stores and
283 : /// the API both need.
284 8874 : pub fn context_row_visible(row: &Value, tenant: Option<&TenantId>) -> bool {
285 8874 : match context_row_owner(row) {
286 7290 : None => true,
287 1584 : Some(owner) => tenant.is_some_and(|t| t.as_str() == owner),
288 : }
289 8874 : }
290 :
291 : /// Current-state storage: everything except the temporal evolution.
292 : ///
293 : /// Contract carried over from the enum seam it replaces: every mutate is
294 : /// one transaction under the row lock, and a missing row is `None`, never
295 : /// an insert (a bookkeeping writeback racing a DELETE must not resurrect
296 : /// the row). Backends map their internal failures to
297 : /// `NgsiError::InternalError` with a GENERIC client-visible detail.
298 : #[async_trait::async_trait]
299 : pub trait CurrentStateDriver: Send + Sync {
300 : /// Readiness RIGHT NOW — a lost database flips /q/ready to 503.
301 : async fn ping(&self) -> Result<(), NgsiError>;
302 : /// (Queued writers, peak) of a single-writer commit section, if the
303 : /// backend has one.
304 0 : fn commit_queue(&self) -> Option<(usize, usize)> {
305 0 : None
306 0 : }
307 : /// Drain: finish in-flight work and disconnect cleanly.
308 : async fn close(&self);
309 : /// What this driver runs on, for `/q/health`: engine, server version,
310 : /// extensions — whatever an operator needs to tell two deployments of
311 : /// the same backend name apart. Read from state captured at startup,
312 : /// never by querying on the request: health is polled. An empty object
313 : /// (the default) means the driver has nothing to add to its name.
314 0 : fn version_info(&self) -> Value {
315 0 : Value::Object(serde_json::Map::new())
316 0 : }
317 : /// Installs the (tenant, before, after) hook called on every entity write.
318 : fn set_change_hook(&self, h: ChangeHook);
319 : /// Turn the same-transaction outbox producer on (bus=nats).
320 0 : fn set_outbox(&self, on: bool) {
321 0 : let _ = on;
322 0 : }
323 : /// Outbox drain: oldest-first page of pending rows `(seq, tenant, event)`.
324 0 : async fn outbox_peek(&self, limit: i64) -> Result<Vec<(i64, String, Value)>, NgsiError> {
325 : let _ = limit;
326 : Ok(Vec::new())
327 0 : }
328 : /// Outbox drain: delete EXACTLY the published rows.
329 0 : async fn outbox_ack(&self, seqs: &[i64]) -> Result<u64, NgsiError> {
330 0 : let _ = seqs;
331 : Ok(0)
332 0 : }
333 : /// Outbox drain: keep the rows of events the bus could not carry whole
334 : /// and take them out of the drain's page. The published message holds a
335 : /// claim-check reference; the row holds the bodies it stands for.
336 0 : async fn outbox_retain(&self, tenant: &TenantId, seqs: &[i64]) -> Result<u64, NgsiError> {
337 : let _ = (tenant, seqs);
338 : Ok(0)
339 0 : }
340 : /// The whole event behind a claim-check reference, or `None` once it has
341 : /// been reaped. Scoped to the tenant that wrote it.
342 0 : async fn outbox_event(&self, seq: i64, tenant: &TenantId) -> Result<Option<Value>, NgsiError> {
343 : let _ = (seq, tenant);
344 : Ok(None)
345 0 : }
346 :
347 : /// Insert a document; `false` if the id already exists (nothing written).
348 : async fn create(
349 : &self,
350 : tenant: &TenantId,
351 : kind: Kind,
352 : id: &str,
353 : doc: Value,
354 : ) -> Result<bool, NgsiError>;
355 : /// Batch create (entities only); created-flags in input order.
356 : async fn batch_create(
357 : &self,
358 : tenant: &TenantId,
359 : items: Vec<(String, Value)>,
360 : ) -> Result<Vec<bool>, NgsiError>;
361 : /// Batch delete (entities only); deleted-flags in input order, a
362 : /// duplicate id deletes once and reads absent the second time.
363 : async fn batch_delete(&self, tenant: &TenantId, ids: &[String])
364 : -> Result<Vec<bool>, NgsiError>;
365 : /// Batch upsert with REPLACE semantics (entities only); created-flags in
366 : /// input order.
367 : async fn batch_upsert(
368 : &self,
369 : tenant: &TenantId,
370 : items: Vec<(String, Value)>,
371 : ) -> Result<Vec<bool>, NgsiError>;
372 : /// Insert or replace a document. `true` means a document was ALREADY
373 : /// there and this call replaced it; `false` means this call created it.
374 : /// The polarity is the opposite of [`Self::batch_upsert`], which answers
375 : /// created-flags — the batch path needs them to split 201 from 204
376 : /// (5.6.8) while the single path ignores the value.
377 : async fn upsert(
378 : &self,
379 : tenant: &TenantId,
380 : kind: Kind,
381 : id: &str,
382 : doc: Value,
383 : ) -> Result<bool, NgsiError>;
384 : /// Read one document; `None` if absent.
385 : async fn get(
386 : &self,
387 : tenant: &TenantId,
388 : kind: Kind,
389 : id: &str,
390 : ) -> Result<Option<Value>, NgsiError>;
391 : /// Delete one document; `false` if it was absent.
392 : async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> Result<bool, NgsiError>;
393 : /// Delete one Entity, but only if `keep` accepts the stored document.
394 : /// `false` = absent OR refused; the caller cannot tell the two apart and
395 : /// does not need to, because 5.6.6.4 gives the same answer to both: an
396 : /// Entity the selector excludes "is not known" for this operation.
397 : ///
398 : /// The read and the delete happen under one lock. Required, deliberately:
399 : /// the obvious default is `get` then `delete`, and that pair is the race
400 : /// this method exists to remove — between the two the Entity can be
401 : /// deleted and recreated under the same id, and the delete then lands on
402 : /// a document the caller never inspected. A backend inheriting a default
403 : /// would keep the race with no compile error to say so.
404 : async fn delete_entity_if(
405 : &self,
406 : tenant: &TenantId,
407 : id: &str,
408 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
409 : ) -> Result<bool, NgsiError>;
410 : /// Every document of this kind in the tenant.
411 : ///
412 : /// A backend may refuse a tenant that holds too many to materialize
413 : /// (5.5.6 TooManyResults). That is right for a client query and wrong
414 : /// for a reader that must see ALL of them — use [`Self::list_page`].
415 : async fn list(&self, tenant: &TenantId, kind: Kind) -> Result<Vec<Value>, NgsiError>;
416 : /// One id-ordered page of documents, for the internal readers that must
417 : /// see every one of them and so cannot be refused: ids strictly greater
418 : /// than `after`, at most `limit`. A short page means the end.
419 : ///
420 : /// Keyset, not offset: the caller walks a tenant that is being written
421 : /// to underneath it, where OFFSET skips and repeats rows. The peak cost
422 : /// is one page, not the whole tenant, which is what the row ceiling on
423 : /// `list` was protecting — so this carries no ceiling of its own, and a
424 : /// backend may not refuse it for volume.
425 : ///
426 : /// Required, deliberately: the obvious default is `list` sliced, and
427 : /// `list` is the read that may refuse. A backend inheriting that would
428 : /// silently reacquire the outage this method exists to prevent, with no
429 : /// compile error to say so.
430 : async fn list_page(
431 : &self,
432 : tenant: &TenantId,
433 : kind: Kind,
434 : after: Option<&str>,
435 : limit: usize,
436 : ) -> Result<Vec<Value>, NgsiError>;
437 : /// One id-ordered window of documents and the size of the whole set:
438 : /// elements `offset..offset + limit`, plus the count 6.3.10's `count`
439 : /// parameter reports. `limit` 0 is a legal request for the count alone.
440 : ///
441 : /// 5.5.9.1: "the query resolution mechanisms of the NGSI-LD System shall
442 : /// ensure that only up to a maximum of L NGSI-LD Elements are RETRIEVED
443 : /// and returned to the NGSI-LD client". Reading a whole tenant to serve
444 : /// one page of it is what that sentence rules out, and it is why this
445 : /// carries no row ceiling: the window bounds the result by construction,
446 : /// so the only thing a ceiling could refuse is a page the client is
447 : /// entitled to.
448 : ///
449 : /// Offset, not the keyset of [`Self::list_page`], because this serves
450 : /// 5.5.9.2's `limit`/`offset`, which lets a client "jump to a desired
451 : /// set of elements". A cursor cannot answer that; the two reads exist
452 : /// for different callers and neither replaces the other.
453 : ///
454 : /// Required for the same reason `list_page` is: the obvious default
455 : /// slices `list`, and `list` is the read that may refuse.
456 : async fn list_slice(
457 : &self,
458 : tenant: &TenantId,
459 : kind: Kind,
460 : offset: usize,
461 : limit: usize,
462 : ) -> Result<(Vec<Value>, usize), NgsiError>;
463 : /// Registrations that can match these ids/types (a backend may narrow;
464 : /// returning the full tenant list is always correct).
465 : async fn matching_registrations(
466 : &self,
467 : tenant: &TenantId,
468 : ids: Option<&[String]>,
469 : types: Option<&[String]>,
470 : ) -> Result<Vec<Value>, NgsiError>;
471 : /// Query Entities with the filter pushed down where the backend can
472 : /// take it; the caller re-checks unless the outcome says `decided`.
473 : async fn query_entities(
474 : &self,
475 : tenant: &TenantId,
476 : f: &filter::EntityFilter<'_>,
477 : ) -> Result<filter::QueryOutcome, NgsiError>;
478 : /// Read-modify-write under the row lock; `None` = absent (never an
479 : /// insert), `Some(Err)` = the closure rejected, nothing committed.
480 : async fn mutate_boxed<'a>(
481 : &self,
482 : tenant: &TenantId,
483 : kind: Kind,
484 : id: &str,
485 : f: MutateFn<'a>,
486 : ) -> Result<Option<Result<(), ()>>, NgsiError>;
487 : /// Batch read-modify-write (entities only); results align with `ids`.
488 : async fn batch_mutate_boxed<'a>(
489 : &self,
490 : tenant: &TenantId,
491 : ids: &[String],
492 : f: BatchMutateFn<'a>,
493 : ) -> Result<Vec<Option<Result<(), ()>>>, NgsiError>;
494 : /// 5.2.14.2 delivery bookkeeping: stamp one delivery attempt on a
495 : /// subscription and hand back the stored document. `timesSent` moves by
496 : /// one, `lastNotification` and `lastSuccess` take `now`, and the previous
497 : /// `lastSuccess` comes back so a failed attempt can roll it back.
498 : /// `None` means the row is gone — the subscription was deleted between
499 : /// matching and delivery, and nothing may be sent (5.8.6).
500 : ///
501 : /// The default expresses it as a `mutate`, which is correct everywhere.
502 : /// A backend whose `mutate` locks the row across a network round trip
503 : /// should override it with one statement: at fan-out every delivery on
504 : /// one subscription contends for that row, so the lock hold time — not
505 : /// the statement count — is what serializes delivery.
506 : async fn record_delivery(
507 : &self,
508 : tenant: &TenantId,
509 : kind: Kind,
510 : id: &str,
511 : now: &str,
512 0 : ) -> Result<Option<Delivery>, NgsiError> {
513 : record_delivery_via_mutate(self, tenant, kind, id, now).await
514 0 : }
515 : /// Reap expired docs/instances (backends with their own maintenance job
516 : /// return 0).
517 0 : async fn sweep_expired(&self) -> usize {
518 : 0
519 0 : }
520 : /// Table 5.2.9-2 forward bookkeeping: stamp one distributed operation on
521 : /// a registration and hand back the stored document. `None` means the row
522 : /// is gone — the registration was deleted while its forward was in
523 : /// flight, and a bookkeeping writeback must never resurrect it.
524 : ///
525 : /// The default expresses it as a `mutate`, which is correct everywhere.
526 : // ponytail: one read-modify-write per forward, so a registration every
527 : // request fans out to is one contended row; a backend whose `mutate`
528 : // locks that row across a network round trip can collapse it into one
529 : // statement the way `record_delivery` does, once a profile asks.
530 : async fn record_forward(
531 : &self,
532 : tenant: &TenantId,
533 : id: &str,
534 : now: &str,
535 : ok: bool,
536 435 : ) -> Result<Option<Value>, NgsiError> {
537 : record_forward_via_mutate(self, tenant, id, now, ok).await
538 435 : }
539 : /// 5.5.10: the default Tenant implicitly exists; others once created.
540 : async fn tenant_exists(&self, tenant: &TenantId) -> Result<bool, NgsiError>;
541 : /// The iteration domain of the interval sweep and of every mirror
542 : /// hydration: a tenant holding a Subscription, a Context Source
543 : /// Registration Subscription OR a Registration SHALL appear. A superset
544 : /// is allowed — the callers list per tenant afterwards and an empty list
545 : /// costs nothing — so a backend that cannot narrow the set cheaply may
546 : /// return every tenant it knows. A SUBSET is a silent outage: a tenant
547 : /// missing here never fires a periodic notification and never reaches
548 : /// the mirror.
549 : ///
550 : /// Registrations are in the domain because one of the hydrations fills
551 : /// the REGISTRATION mirror, and the federation path reads that mirror
552 : /// alone whenever it is installed. A domain that stopped at
553 : /// subscription-holding tenants left a tenant with registrations and no
554 : /// subscription forwarding to no Context Source at all.
555 : async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError>;
556 : /// Every tenant the backend knows, sorted. The default Tenant is listed
557 : /// even when empty (5.5.10: it always exists). Names only: at the
558 : /// 10 000-tenant target (ADR-0001) an inventory carrying per-kind counts
559 : /// would cost a count per kind per tenant, so the counts are paid per
560 : /// lookup in `tenant_stats_one`.
561 0 : async fn tenant_ids(&self) -> Result<Vec<String>, NgsiError> {
562 : Err(NgsiError::OperationNotSupported("tenant inventory".into()))
563 0 : }
564 : /// What one tenant holds; `None` when it does not exist.
565 0 : async fn tenant_stats_one(&self, tenant: &TenantId) -> Result<Option<TenantStats>, NgsiError> {
566 0 : let _ = tenant;
567 : Err(NgsiError::OperationNotSupported("tenant inventory".into()))
568 0 : }
569 : /// Remove every current-state document of one tenant; `false` when the
570 : /// tenant did not exist. The default Tenant is emptied but keeps
571 : /// existing.
572 0 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
573 0 : let _ = tenant;
574 : Err(NgsiError::OperationNotSupported("tenant purge".into()))
575 0 : }
576 : /// 5.13 @context documents. `tenant` is the Tenant the call acts for, and
577 : /// `None` means no Tenant at all — the boot warm and the `Cached`
578 : /// write-through, which touch documents downloaded from public URLs.
579 : ///
580 : /// A backend enforces ADR-0021 on it: a `Cached` row belongs to no Tenant
581 : /// and is reachable with any `tenant`, and every other kind is reachable
582 : /// only by the Tenant named in its `owner` member. A backend that ignores
583 : /// the parameter fails the driver contract.
584 : async fn context_put(
585 : &self,
586 : tenant: Option<&TenantId>,
587 : id: &str,
588 : doc: Value,
589 : ) -> Result<(), NgsiError>;
590 : /// Read a stored @context document by id; `None` if absent or owned by
591 : /// another Tenant.
592 : async fn context_get(
593 : &self,
594 : tenant: Option<&TenantId>,
595 : id: &str,
596 : ) -> Result<Option<Value>, NgsiError>;
597 : /// Delete a stored @context document; `false` if it was absent or owned
598 : /// by another Tenant.
599 : async fn context_delete(&self, tenant: Option<&TenantId>, id: &str) -> Result<bool, NgsiError>;
600 : /// Every stored @context document.
601 : /// Every stored `@context` row WITHOUT its `body` member — the url,
602 : /// localId, kind, owner and usage counters, and not the document.
603 : ///
604 : /// There is deliberately no read that returns every row WITH its body.
605 : /// A body is accepted up to `MAX_CONTEXT_BYTES` (5 MiB) and only the
606 : /// `Cached` rows are capped in number, so one such read materializes
607 : /// gigabytes — on the boot path, where it decides whether the broker
608 : /// starts at all. The callers that need one body ask for it by id
609 : /// ([`Self::context_get`]); nothing needs them all at once.
610 : async fn context_list_meta(&self, tenant: Option<&TenantId>) -> Result<Vec<Value>, NgsiError>;
611 : }
612 :
613 : /// The typed-slot lock, taken and released inside one closure call. A
614 : /// poisoned slot is read through: the panic that poisoned it is already on
615 : /// its way out of the driver, and the value behind it is whatever the
616 : /// closure managed to write.
617 1863 : fn lock<T>(m: &std::sync::Mutex<T>) -> std::sync::MutexGuard<'_, T> {
618 1863 : m.lock().unwrap_or_else(|p| p.into_inner())
619 1863 : }
620 :
621 : /// The same, for taking the slot's value once the driver has answered.
622 1905 : fn into_inner<T>(m: std::sync::Mutex<T>) -> T {
623 1905 : m.into_inner().unwrap_or_else(|p| p.into_inner())
624 1905 : }
625 :
626 : /// Typed sugar over the boxed mutate seam — call sites keep their
627 : /// `Result<T, E>` closures; the value crosses the object boundary in a
628 : /// side slot.
629 : pub trait CurrentStateDriverExt {
630 : /// Typed read-modify-write: `None` = absent, `Some(Err(e))` = the closure
631 : /// rejected and nothing was committed.
632 : fn mutate<T: Send, E: Send>(
633 : &self,
634 : tenant: &TenantId,
635 : kind: Kind,
636 : id: &str,
637 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
638 : ) -> impl std::future::Future<Output = Result<Option<Result<T, E>>, NgsiError>> + Send;
639 : /// Typed batch read-modify-write (entities only); results align with `ids`.
640 : fn batch_mutate<E: Send>(
641 : &self,
642 : tenant: &TenantId,
643 : ids: &[String],
644 : f: impl FnMut(&str, &mut Value) -> Result<(), E> + Send,
645 : ) -> impl std::future::Future<Output = Result<Vec<Option<Result<(), E>>>, NgsiError>> + Send;
646 : }
647 :
648 : impl<S: CurrentStateDriver + ?Sized> CurrentStateDriverExt for S {
649 4573 : async fn mutate<T: Send, E: Send>(
650 4573 : &self,
651 4573 : tenant: &TenantId,
652 4573 : kind: Kind,
653 4573 : id: &str,
654 4573 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
655 4573 : ) -> Result<Option<Result<T, E>>, NgsiError> {
656 4573 : let slot = std::sync::Mutex::new(None);
657 4573 : let r = self
658 4573 : .mutate_boxed(
659 4573 : tenant,
660 4573 : kind,
661 4573 : id,
662 4573 : Box::new(|v| {
663 1771 : let r = f(v);
664 1771 : let flag = if r.is_ok() { Ok(()) } else { Err(()) };
665 1771 : *lock(&slot) = Some(r);
666 1771 : flag
667 1771 : }),
668 : )
669 4573 : .await?;
670 4573 : Ok(r.and_then(|_| into_inner(slot)))
671 4573 : }
672 :
673 42 : async fn batch_mutate<E: Send>(
674 42 : &self,
675 42 : tenant: &TenantId,
676 42 : ids: &[String],
677 42 : mut f: impl FnMut(&str, &mut Value) -> Result<(), E> + Send,
678 42 : ) -> Result<Vec<Option<Result<(), E>>>, NgsiError> {
679 : // Errors land in the queue in closure-call order, which the trait
680 : // contract fixes to input order over present ids.
681 42 : let errs = std::sync::Mutex::new(std::collections::VecDeque::new());
682 42 : let r = self
683 42 : .batch_mutate_boxed(
684 42 : tenant,
685 42 : ids,
686 4052 : Box::new(|id, v| match f(id, v) {
687 4052 : Ok(()) => Ok(()),
688 0 : Err(e) => {
689 0 : lock(&errs).push_back(e);
690 0 : Err(())
691 : }
692 4052 : }),
693 : )
694 42 : .await?;
695 42 : let mut errs = into_inner(errs);
696 42 : let mut out = Vec::with_capacity(r.len());
697 4080 : for slot in r {
698 4080 : out.push(match slot {
699 28 : None => None,
700 4052 : Some(Ok(())) => Some(Ok(())),
701 : // one queued error per rejected id is the trait contract; a
702 : // driver that reports more rejections than the closure raised
703 : // has broken it, and the batch fails rather than inventing an
704 : // error for the caller to read.
705 0 : Some(Err(())) => Some(Err(errs.pop_front().ok_or_else(|| {
706 0 : NgsiError::InternalError(
707 0 : "batch driver reported more rejections than were raised".into(),
708 0 : )
709 0 : })?)),
710 : });
711 : }
712 42 : Ok(out)
713 42 : }
714 : }
715 :
716 : /// What a temporal event records.
717 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
718 : pub enum TemporalOp {
719 : /// An Attribute the entity did not carry before (one event per instance).
720 : AttrCreated,
721 : /// A changed instance of an existing Attribute (4.5.6 append).
722 : AttrModified,
723 : /// 4.5.6: the Scope changed through the Core API — recorded as a scope
724 : /// instance whose observedAt copies modifiedAt.
725 : ScopeChanged,
726 : }
727 :
728 : /// One change the write path hands to the temporal seam. Events are
729 : /// produced per attribute INSTANCE (the gate chain and a columnar writer
730 : /// both work per instance) and drained per request, in order.
731 : #[derive(Clone, Debug)]
732 : pub struct TemporalEvent {
733 : /// What the event records.
734 : pub op: TemporalOp,
735 : /// Owning tenant.
736 : pub tenant: TenantId,
737 : /// Id of the entity whose history this instance belongs to.
738 : pub entity_id: String,
739 : /// The entity's meta shell (id, type, createdAt, modifiedAt, scope) as
740 : /// it stood after the write — what `temporal_append` creates on first
741 : /// touch.
742 : pub shell: Value,
743 : /// Expanded Attribute name, or `scope`.
744 : pub attr: String,
745 : /// The instance snapshot: value, datasetId, observedAt, instanceId…
746 : pub instance: Value,
747 : }
748 :
749 : /// Temporal storage: the entity history (Temporal Evolution) plus the raw
750 : /// temporal documents the 5.6.13-5.6.16 edit paths operate on.
751 : ///
752 : /// Internal bookkeeping (snapshot copies, delete cascades) degrades to
753 : /// benign no-ops on a driver without temporal support; the CLIENT-facing
754 : /// operations answer `OperationNotSupported` (422 per CIM 009
755 : /// Table 6.3.2-1) instead.
756 : #[async_trait::async_trait]
757 : pub trait TemporalDriver: Send + Sync {
758 : /// The drain: one call carries a whole request's events, in production
759 : /// order. The default folds consecutive events of one entity into a
760 : /// single `temporal_append` (scope changes go through `mutate`, as
761 : /// 4.5.6 shapes them); a bulk writer overrides this and sees the batch.
762 5334 : async fn event_list(&self, evs: &[TemporalEvent]) -> Result<(), NgsiError> {
763 : let mut i = 0;
764 : while i < evs.len() {
765 : let (tenant, id) = (&evs[i].tenant, evs[i].entity_id.as_str());
766 : let mut additions = serde_json::Map::new();
767 : let mut shell = &evs[i].shell;
768 : let mut j = i;
769 : while j < evs.len() && evs[j].tenant == *tenant && evs[j].entity_id == id {
770 : let ev = &evs[j];
771 : shell = &ev.shell;
772 : if ev.op == TemporalOp::ScopeChanged {
773 : let inst = ev.instance.clone();
774 4 : self.mutate(tenant, id, |doc| {
775 4 : let target = doc.as_object_mut().ok_or(())?;
776 4 : match target.get_mut("scope").and_then(Value::as_array_mut) {
777 2 : Some(arr) if arr.first().is_some_and(Value::is_object) => {
778 0 : arr.push(inst);
779 0 : }
780 4 : _ => {
781 4 : target.insert("scope".into(), Value::Array(vec![inst]));
782 4 : }
783 : }
784 4 : Ok::<(), ()>(())
785 4 : })
786 : .await?;
787 : } else {
788 : if let Some(arr) = additions
789 : .entry(ev.attr.clone())
790 5608 : .or_insert_with(|| Value::Array(Vec::new()))
791 : .as_array_mut()
792 : {
793 : arr.push(ev.instance.clone());
794 : }
795 : }
796 : j += 1;
797 : }
798 : if !additions.is_empty() {
799 : self.temporal_append(tenant, id, shell, &Value::Object(additions))
800 : .await?;
801 : }
802 : i = j;
803 : }
804 : Ok(())
805 5334 : }
806 : /// `false` = this deployment records no history (`NoTemporal`); the
807 : /// write path skips recording entirely.
808 15402 : fn supported(&self) -> bool {
809 15402 : true
810 15402 : }
811 : /// Stored attribute instances of one tenant (inventory); a driver
812 : /// without history reports 0.
813 0 : async fn attr_instance_count(&self, tenant: &TenantId) -> Result<u64, NgsiError> {
814 0 : let _ = tenant;
815 : Ok(0)
816 0 : }
817 : /// Remove the whole history of one tenant; nothing to do without history.
818 0 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<(), NgsiError> {
819 0 : let _ = tenant;
820 : Ok(())
821 0 : }
822 : /// Readiness of the temporal backend; the default is always ready.
823 0 : async fn ping(&self) -> Result<(), NgsiError> {
824 : Ok(())
825 0 : }
826 : /// What this temporal driver runs on, for `/q/health`; the same
827 : /// contract as [`CurrentStateDriver::version_info`].
828 6 : fn version_info(&self) -> Value {
829 6 : Value::Object(serde_json::Map::new())
830 6 : }
831 : /// Drain: finish in-flight work and disconnect cleanly. A temporal
832 : /// driver configured as a backend of its own holds its own pool, and the
833 : /// shutdown path closes it here. May be called more than once — when one
834 : /// instance serves both seams it is closed through each of them — so an
835 : /// implementation makes it idempotent. The default has nothing to close.
836 6 : async fn close(&self) {}
837 : /// Auto-recording fast path: append instances, creating the meta shell
838 : /// on first touch — and only for an entity that still exists (5.6.6
839 : /// deletes history; an append overlapping the delete must not recreate
840 : /// it).
841 : async fn temporal_append(
842 : &self,
843 : tenant: &TenantId,
844 : id: &str,
845 : shell: &Value,
846 : additions: &Value,
847 : ) -> Result<(), NgsiError>;
848 : /// Would `query_temporal` page EXACTLY under this `q` and range — the
849 : /// driver's own entity verdict equal to the evaluator's on every row,
850 : /// so a page it cuts is the page the caller would have cut? A driver
851 : /// that pages after an in-process evaluation, or not at all, answers
852 : /// `false`, and the caller pages the merged result itself (5.7.4.4).
853 0 : fn q_pushdown_exact(
854 0 : &self,
855 0 : _q: &antares_ql::QNode,
856 0 : _range: Option<&filter::InstanceRange<'_>>,
857 0 : _expand: &dyn Fn(&str) -> String,
858 0 : ) -> bool {
859 0 : false
860 0 : }
861 : /// Query Temporal Evolution (5.7.4) with pushdown where possible.
862 : async fn query_temporal(
863 : &self,
864 : tenant: &TenantId,
865 : f: &filter::TemporalFilter<'_>,
866 : ) -> Result<filter::TemporalOutcome, NgsiError>;
867 : /// Retrieve Temporal Evolution (5.7.3) with instance pruning.
868 : async fn get_temporal(
869 : &self,
870 : tenant: &TenantId,
871 : id: &str,
872 : f: &filter::TemporalFilter<'_>,
873 : ) -> Result<Option<Value>, NgsiError>;
874 : /// Raw temporal document access (the 5.6.11-5.6.16 edit/delete paths
875 : /// and internal copies).
876 : async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, NgsiError>;
877 : /// Insert a temporal document; `false` if the id already exists.
878 : async fn create(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError>;
879 : /// Insert or replace a temporal document. `true` means it was ALREADY
880 : /// there and this call replaced it; `false` means this call created it,
881 : /// the same polarity as the current-state seam.
882 : async fn upsert(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError>;
883 : /// Delete an entity's whole history; `false` if it had none.
884 : async fn delete(&self, tenant: &TenantId, id: &str) -> Result<bool, NgsiError>;
885 : /// Every temporal document in the tenant.
886 : async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, NgsiError>;
887 : /// Read-modify-write of one temporal document under its row lock;
888 : /// `None` = absent (never an insert), `Some(Err)` = rejected, not committed.
889 : async fn mutate_boxed<'a>(
890 : &self,
891 : tenant: &TenantId,
892 : id: &str,
893 : f: MutateFn<'a>,
894 : ) -> Result<Option<Result<(), ()>>, NgsiError>;
895 : }
896 :
897 : /// Typed mutate sugar for the temporal seam, same slot trick as
898 : /// [`CurrentStateDriverExt`].
899 : pub trait TemporalDriverExt {
900 : /// Typed read-modify-write of one temporal document: `None` = absent,
901 : /// `Some(Err(e))` = the closure rejected and nothing was committed.
902 : fn mutate<T: Send, E: Send>(
903 : &self,
904 : tenant: &TenantId,
905 : id: &str,
906 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
907 : ) -> impl std::future::Future<Output = Result<Option<Result<T, E>>, NgsiError>> + Send;
908 : }
909 :
910 : impl<S: TemporalDriver + ?Sized> TemporalDriverExt for S {
911 182 : async fn mutate<T: Send, E: Send>(
912 182 : &self,
913 182 : tenant: &TenantId,
914 182 : id: &str,
915 182 : f: impl FnOnce(&mut Value) -> Result<T, E> + Send,
916 182 : ) -> Result<Option<Result<T, E>>, NgsiError> {
917 182 : let slot = std::sync::Mutex::new(None);
918 182 : let r = self
919 182 : .mutate_boxed(
920 182 : tenant,
921 182 : id,
922 182 : Box::new(|v| {
923 92 : let r = f(v);
924 92 : let flag = if r.is_ok() { Ok(()) } else { Err(()) };
925 92 : *lock(&slot) = Some(r);
926 92 : flag
927 92 : }),
928 : )
929 182 : .await?;
930 182 : Ok(r.and_then(|_| into_inner(slot)))
931 182 : }
932 : }
933 :
934 : /// The no-history driver: temporal OFF as a driver choice. Client-facing
935 : /// reads answer `OperationNotSupported`; the recorder and the internal
936 : /// bookkeeping paths degrade to no-ops.
937 : pub struct NoTemporal;
938 :
939 6 : fn unsupported() -> NgsiError {
940 6 : NgsiError::OperationNotSupported("no temporal store is configured".into())
941 6 : }
942 :
943 : #[async_trait::async_trait]
944 : impl TemporalDriver for NoTemporal {
945 12 : fn supported(&self) -> bool {
946 12 : false
947 12 : }
948 : async fn temporal_append(
949 : &self,
950 : _tenant: &TenantId,
951 : _id: &str,
952 : _shell: &Value,
953 : _additions: &Value,
954 2 : ) -> Result<(), NgsiError> {
955 : Ok(())
956 2 : }
957 : async fn query_temporal(
958 : &self,
959 : _tenant: &TenantId,
960 : _f: &filter::TemporalFilter<'_>,
961 4 : ) -> Result<filter::TemporalOutcome, NgsiError> {
962 : Err(unsupported())
963 4 : }
964 : async fn get_temporal(
965 : &self,
966 : _tenant: &TenantId,
967 : _id: &str,
968 : _f: &filter::TemporalFilter<'_>,
969 2 : ) -> Result<Option<Value>, NgsiError> {
970 : Err(unsupported())
971 2 : }
972 2 : async fn get(&self, _tenant: &TenantId, _id: &str) -> Result<Option<Value>, NgsiError> {
973 : Ok(None)
974 2 : }
975 0 : async fn create(&self, _tenant: &TenantId, _id: &str, _doc: Value) -> Result<bool, NgsiError> {
976 : Ok(false)
977 0 : }
978 0 : async fn upsert(&self, _tenant: &TenantId, _id: &str, _doc: Value) -> Result<bool, NgsiError> {
979 : Ok(false)
980 0 : }
981 2 : async fn delete(&self, _tenant: &TenantId, _id: &str) -> Result<bool, NgsiError> {
982 : Ok(false)
983 2 : }
984 2 : async fn list(&self, _tenant: &TenantId) -> Result<Vec<Value>, NgsiError> {
985 : Ok(Vec::new())
986 2 : }
987 : async fn mutate_boxed<'a>(
988 : &self,
989 : _tenant: &TenantId,
990 : _id: &str,
991 : _f: MutateFn<'a>,
992 2 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
993 : Ok(None)
994 2 : }
995 : }
996 :
997 : #[cfg(test)]
998 : mod tests {
999 : use super::*;
1000 :
1001 : /// The mode list is one source of truth: every variant round-trips
1002 : /// through its name, and the message for an unknown name lists all of
1003 : /// them — a backend added to the enum cannot be missing from either.
1004 : #[test]
1005 2 : fn every_store_mode_round_trips_and_the_error_lists_them_all() {
1006 8 : for m in StoreMode::ALL {
1007 8 : let back = m
1008 8 : .as_str()
1009 8 : .parse::<StoreMode>()
1010 8 : .unwrap_or_else(|e| panic!("{m} must parse back: {e}"));
1011 8 : assert_eq!(back, m);
1012 : }
1013 2 : let err = "mongo".parse::<StoreMode>().expect_err("unknown mode");
1014 2 : assert!(err.contains("mongo"), "{err}");
1015 8 : for m in StoreMode::ALL {
1016 8 : assert!(err.contains(m.as_str()), "the message must name {m}: {err}");
1017 : }
1018 2 : }
1019 :
1020 : /// A minimal driver proving the boxed seam round-trips typed results:
1021 : /// present row → the closure's T and E cross intact; absent → None.
1022 : struct OneDoc(std::sync::Mutex<Option<Value>>);
1023 : #[async_trait::async_trait]
1024 : impl CurrentStateDriver for OneDoc {
1025 0 : async fn ping(&self) -> Result<(), NgsiError> {
1026 : Ok(())
1027 0 : }
1028 0 : async fn close(&self) {}
1029 0 : fn set_change_hook(&self, _h: ChangeHook) {}
1030 : async fn create(
1031 : &self,
1032 : _t: &TenantId,
1033 : _k: Kind,
1034 : _id: &str,
1035 : doc: Value,
1036 2 : ) -> Result<bool, NgsiError> {
1037 : *self.0.lock().expect("lock") = Some(doc);
1038 : Ok(true)
1039 2 : }
1040 : async fn batch_create(
1041 : &self,
1042 : _t: &TenantId,
1043 : _items: Vec<(String, Value)>,
1044 0 : ) -> Result<Vec<bool>, NgsiError> {
1045 : unimplemented!()
1046 0 : }
1047 : async fn list_page(
1048 : &self,
1049 : _t: &TenantId,
1050 : _k: Kind,
1051 : _after: Option<&str>,
1052 : _limit: usize,
1053 0 : ) -> Result<Vec<Value>, NgsiError> {
1054 : unimplemented!()
1055 0 : }
1056 : async fn list_slice(
1057 : &self,
1058 : _t: &TenantId,
1059 : _k: Kind,
1060 : _offset: usize,
1061 : _limit: usize,
1062 0 : ) -> Result<(Vec<Value>, usize), NgsiError> {
1063 : unimplemented!()
1064 0 : }
1065 : async fn batch_delete(
1066 : &self,
1067 : _t: &TenantId,
1068 : _ids: &[String],
1069 0 : ) -> Result<Vec<bool>, NgsiError> {
1070 : unimplemented!()
1071 0 : }
1072 : async fn batch_upsert(
1073 : &self,
1074 : _t: &TenantId,
1075 : _items: Vec<(String, Value)>,
1076 0 : ) -> Result<Vec<bool>, NgsiError> {
1077 : unimplemented!()
1078 0 : }
1079 : async fn upsert(
1080 : &self,
1081 : _t: &TenantId,
1082 : _k: Kind,
1083 : _id: &str,
1084 : _doc: Value,
1085 0 : ) -> Result<bool, NgsiError> {
1086 : unimplemented!()
1087 0 : }
1088 : async fn get(
1089 : &self,
1090 : _t: &TenantId,
1091 : _k: Kind,
1092 : _id: &str,
1093 4 : ) -> Result<Option<Value>, NgsiError> {
1094 : Ok(self.0.lock().expect("lock").clone())
1095 4 : }
1096 0 : async fn delete(&self, _t: &TenantId, _k: Kind, _id: &str) -> Result<bool, NgsiError> {
1097 : unimplemented!()
1098 0 : }
1099 : async fn delete_entity_if(
1100 : &self,
1101 : _t: &TenantId,
1102 : _id: &str,
1103 : _keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
1104 0 : ) -> Result<bool, NgsiError> {
1105 : unimplemented!()
1106 0 : }
1107 0 : async fn list(&self, _t: &TenantId, _k: Kind) -> Result<Vec<Value>, NgsiError> {
1108 : unimplemented!()
1109 0 : }
1110 : async fn matching_registrations(
1111 : &self,
1112 : _t: &TenantId,
1113 : _ids: Option<&[String]>,
1114 : _types: Option<&[String]>,
1115 0 : ) -> Result<Vec<Value>, NgsiError> {
1116 : unimplemented!()
1117 0 : }
1118 : async fn query_entities(
1119 : &self,
1120 : _t: &TenantId,
1121 : _f: &filter::EntityFilter<'_>,
1122 0 : ) -> Result<filter::QueryOutcome, NgsiError> {
1123 : unimplemented!()
1124 0 : }
1125 : async fn mutate_boxed<'a>(
1126 : &self,
1127 : _t: &TenantId,
1128 : _k: Kind,
1129 : _id: &str,
1130 : f: MutateFn<'a>,
1131 6 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1132 : let mut guard = self.0.lock().expect("lock");
1133 : match guard.as_mut() {
1134 : None => Ok(None),
1135 : Some(v) => {
1136 : let mut copy = v.clone();
1137 : match f(&mut copy) {
1138 : Ok(()) => {
1139 : *v = copy;
1140 : Ok(Some(Ok(())))
1141 : }
1142 : Err(()) => Ok(Some(Err(()))),
1143 : }
1144 : }
1145 : }
1146 6 : }
1147 : async fn batch_mutate_boxed<'a>(
1148 : &self,
1149 : _t: &TenantId,
1150 : _ids: &[String],
1151 : _f: BatchMutateFn<'a>,
1152 0 : ) -> Result<Vec<Option<Result<(), ()>>>, NgsiError> {
1153 : unimplemented!()
1154 0 : }
1155 0 : async fn tenant_exists(&self, _t: &TenantId) -> Result<bool, NgsiError> {
1156 : Ok(true)
1157 0 : }
1158 0 : async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError> {
1159 : Ok(Vec::new())
1160 0 : }
1161 : async fn context_put(
1162 : &self,
1163 : _t: Option<&TenantId>,
1164 : _id: &str,
1165 : _doc: Value,
1166 0 : ) -> Result<(), NgsiError> {
1167 : Ok(())
1168 0 : }
1169 : async fn context_get(
1170 : &self,
1171 : _t: Option<&TenantId>,
1172 : _id: &str,
1173 0 : ) -> Result<Option<Value>, NgsiError> {
1174 : Ok(None)
1175 0 : }
1176 : async fn context_delete(
1177 : &self,
1178 : _t: Option<&TenantId>,
1179 : _id: &str,
1180 0 : ) -> Result<bool, NgsiError> {
1181 : Ok(false)
1182 0 : }
1183 0 : async fn context_list_meta(&self, _t: Option<&TenantId>) -> Result<Vec<Value>, NgsiError> {
1184 : Ok(Vec::new())
1185 0 : }
1186 : }
1187 :
1188 : #[tokio::test]
1189 2 : async fn typed_mutate_round_trips_through_the_boxed_seam() {
1190 2 : let t = TenantId::new("t").expect("tenant");
1191 2 : let d: std::sync::Arc<dyn CurrentStateDriver> =
1192 2 : std::sync::Arc::new(OneDoc(std::sync::Mutex::new(None)));
1193 : // absent → None, closure never runs
1194 2 : let r = d
1195 2 : .mutate::<u32, &str>(&t, Kind::Entity, "x", |_| panic!("must not run"))
1196 2 : .await
1197 2 : .expect("driver ok");
1198 2 : assert!(r.is_none());
1199 2 : d.create(&t, Kind::Entity, "x", serde_json::json!({"n": 1}))
1200 2 : .await
1201 2 : .expect("create");
1202 : // typed success crosses the boundary AND the write commits
1203 2 : let r = d
1204 2 : .mutate::<u32, &str>(&t, Kind::Entity, "x", |v| {
1205 2 : v["n"] = serde_json::json!(2);
1206 2 : Ok(7)
1207 2 : })
1208 2 : .await
1209 2 : .expect("driver ok");
1210 2 : assert_eq!(r, Some(Ok(7)));
1211 2 : assert_eq!(
1212 2 : d.get(&t, Kind::Entity, "x")
1213 2 : .await
1214 2 : .expect("get")
1215 2 : .expect("doc")["n"],
1216 : 2
1217 : );
1218 : // typed error crosses the boundary AND the write is discarded
1219 2 : let r = d
1220 2 : .mutate::<u32, &str>(&t, Kind::Entity, "x", |v| {
1221 2 : v["n"] = serde_json::json!(99);
1222 2 : Err("rejected")
1223 2 : })
1224 2 : .await
1225 2 : .expect("driver ok");
1226 2 : assert_eq!(r, Some(Err("rejected")));
1227 2 : assert_eq!(
1228 2 : d.get(&t, Kind::Entity, "x")
1229 2 : .await
1230 2 : .expect("get")
1231 2 : .expect("doc")["n"],
1232 2 : 2,
1233 2 : "a rejecting closure must not commit"
1234 2 : );
1235 2 : }
1236 :
1237 : /// A temporal driver that only records what the seam hands it: appends
1238 : /// as (id, additions) in call order, scope mutations on a held doc.
1239 : #[derive(Default)]
1240 : struct Recorder {
1241 : appends: std::sync::Mutex<Vec<(String, Value)>>,
1242 : doc: std::sync::Mutex<Option<Value>>,
1243 : }
1244 : #[async_trait::async_trait]
1245 : impl TemporalDriver for Recorder {
1246 : async fn temporal_append(
1247 : &self,
1248 : _t: &TenantId,
1249 : id: &str,
1250 : _shell: &Value,
1251 : additions: &Value,
1252 6 : ) -> Result<(), NgsiError> {
1253 : self.appends
1254 : .lock()
1255 : .expect("lock")
1256 : .push((id.to_owned(), additions.clone()));
1257 : Ok(())
1258 6 : }
1259 : async fn query_temporal(
1260 : &self,
1261 : _t: &TenantId,
1262 : _f: &filter::TemporalFilter<'_>,
1263 0 : ) -> Result<filter::TemporalOutcome, NgsiError> {
1264 : unimplemented!()
1265 0 : }
1266 : async fn get_temporal(
1267 : &self,
1268 : _t: &TenantId,
1269 : _id: &str,
1270 : _f: &filter::TemporalFilter<'_>,
1271 0 : ) -> Result<Option<Value>, NgsiError> {
1272 : unimplemented!()
1273 0 : }
1274 2 : async fn get(&self, _t: &TenantId, _id: &str) -> Result<Option<Value>, NgsiError> {
1275 : Ok(self.doc.lock().expect("lock").clone())
1276 2 : }
1277 2 : async fn create(&self, _t: &TenantId, _id: &str, doc: Value) -> Result<bool, NgsiError> {
1278 : *self.doc.lock().expect("lock") = Some(doc);
1279 : Ok(true)
1280 2 : }
1281 0 : async fn upsert(&self, _t: &TenantId, _id: &str, _doc: Value) -> Result<bool, NgsiError> {
1282 : unimplemented!()
1283 0 : }
1284 0 : async fn delete(&self, _t: &TenantId, _id: &str) -> Result<bool, NgsiError> {
1285 : unimplemented!()
1286 0 : }
1287 0 : async fn list(&self, _t: &TenantId) -> Result<Vec<Value>, NgsiError> {
1288 : unimplemented!()
1289 0 : }
1290 : async fn mutate_boxed<'a>(
1291 : &self,
1292 : _t: &TenantId,
1293 : _id: &str,
1294 : f: MutateFn<'a>,
1295 2 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1296 : let mut guard = self.doc.lock().expect("lock");
1297 : match guard.as_mut() {
1298 : None => Ok(None),
1299 : Some(v) => Ok(Some(f(v))),
1300 : }
1301 2 : }
1302 : }
1303 :
1304 12 : fn ev(op: TemporalOp, id: &str, attr: &str, n: u32) -> TemporalEvent {
1305 12 : TemporalEvent {
1306 12 : op,
1307 12 : tenant: TenantId::new("t").expect("tenant"),
1308 12 : entity_id: id.into(),
1309 12 : shell: serde_json::json!({"id": id, "type": ["T"]}),
1310 12 : attr: attr.into(),
1311 12 : instance: serde_json::json!({"type": "Property", "value": n}),
1312 12 : }
1313 12 : }
1314 :
1315 : /// The drain folds one request's events into ONE append per entity run
1316 : /// — a 2-attribute entity is one call carrying both, not two — and
1317 : /// keeps production order across entities.
1318 : #[tokio::test]
1319 2 : async fn event_list_folds_a_request_into_one_append_per_entity_run() {
1320 2 : let d = Recorder::default();
1321 2 : d.event_list(&[
1322 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 1),
1323 2 : ev(TemporalOp::AttrCreated, "urn:a", "speed", 2),
1324 2 : ev(TemporalOp::AttrModified, "urn:a", "heading", 3),
1325 2 : ev(TemporalOp::AttrModified, "urn:b", "speed", 4),
1326 2 : ev(TemporalOp::AttrModified, "urn:a", "speed", 5),
1327 2 : ])
1328 2 : .await
1329 2 : .expect("drain ok");
1330 2 : let appends = d.appends.lock().expect("lock").clone();
1331 6 : let ids: Vec<&str> = appends.iter().map(|(id, _)| id.as_str()).collect();
1332 2 : assert_eq!(
1333 : ids,
1334 : ["urn:a", "urn:b", "urn:a"],
1335 : "one append per entity run, in order"
1336 : );
1337 2 : let first = &appends[0].1;
1338 2 : assert_eq!(first["speed"].as_array().map(Vec::len), Some(2), "{first}");
1339 2 : assert_eq!(
1340 2 : first["heading"].as_array().map(Vec::len),
1341 : Some(1),
1342 : "{first}"
1343 : );
1344 2 : assert!(
1345 2 : first.get("value").is_none(),
1346 : "instances live under their attribute: {first}"
1347 : );
1348 2 : assert_eq!(appends[2].1["speed"][0]["value"], 5);
1349 2 : }
1350 :
1351 : /// 4.5.6: a scope change becomes a scope instance on the held temporal
1352 : /// doc (array-of-instances form), not an attribute append.
1353 : #[tokio::test]
1354 2 : async fn event_list_records_scope_changes_as_scope_instances() {
1355 2 : let t = TenantId::new("t").expect("tenant");
1356 2 : let d = Recorder::default();
1357 2 : d.create(
1358 2 : &t,
1359 2 : "urn:a",
1360 2 : serde_json::json!({"id": "urn:a", "scope": "/old"}),
1361 2 : )
1362 2 : .await
1363 2 : .expect("create");
1364 2 : let mut scope = ev(TemporalOp::ScopeChanged, "urn:a", "scope", 0);
1365 2 : scope.instance = serde_json::json!({"type": "Property", "value": "/new",
1366 2 : "observedAt": "2026-01-01T00:00:00Z"});
1367 2 : d.event_list(&[scope]).await.expect("drain ok");
1368 2 : assert!(
1369 2 : d.appends.lock().expect("lock").is_empty(),
1370 : "no attribute append for a scope change"
1371 : );
1372 2 : let doc = d.get(&t, "urn:a").await.expect("get").expect("doc");
1373 2 : assert_eq!(doc["scope"][0]["value"], "/new", "{doc}");
1374 2 : assert_eq!(
1375 2 : doc["scope"].as_array().map(Vec::len),
1376 2 : Some(1),
1377 2 : "the plain scope became an instance array: {doc}"
1378 2 : );
1379 2 : }
1380 :
1381 : #[tokio::test]
1382 2 : async fn no_temporal_degrades_without_panicking() {
1383 2 : let t = TenantId::new("t").expect("tenant");
1384 2 : let d: std::sync::Arc<dyn TemporalDriver> = std::sync::Arc::new(NoTemporal);
1385 2 : assert!(!d.supported());
1386 : // client-facing reads: the spec error, not a panic
1387 2 : let e = match d
1388 2 : .query_temporal(&t, &filter::TemporalFilter::default())
1389 2 : .await
1390 : {
1391 2 : Err(e) => e,
1392 0 : Ok(_) => panic!("query_temporal on NoTemporal must be unsupported"),
1393 : };
1394 2 : assert_eq!(e.status(), 422);
1395 2 : assert_eq!(e.kind(), "OperationNotSupported");
1396 : // internal bookkeeping: benign no-ops
1397 2 : assert!(d
1398 2 : .temporal_append(&t, "x", &Value::Null, &Value::Null)
1399 2 : .await
1400 2 : .is_ok());
1401 2 : assert!(!d.delete(&t, "x").await.expect("ok"));
1402 2 : assert!(d.list(&t).await.expect("ok").is_empty());
1403 2 : assert!(d
1404 2 : .mutate::<(), ()>(&t, "x", |_| Ok(()))
1405 2 : .await
1406 2 : .expect("ok")
1407 2 : .is_none());
1408 2 : }
1409 : }
|
