Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The store seam: ONE closed set of backends behind the
3 : //! memory store's 12-method surface. An enum, not a trait — `mutate<T, E>` is
4 : //! generic (dyn-incompatible), the backend set is closed by design
5 : //! (exactly the implementations the product needs), and match
6 : //! exhaustiveness forces every backend to answer every method.
7 : //!
8 : //! Every method returns `Result<_, NgsiError>`: the memory/file backend never
9 : //! errors; the postgres backend maps sqlx failures to `InternalError` (500) —
10 : //! a DB outage must be a visible 5xx, never a silent 404/409.
11 :
12 : use antares_model::{NgsiError, TenantId};
13 : use serde_json::Value;
14 :
15 : #[cfg(feature = "postgres")]
16 : use super::pg::doc::{DocKind, PgDocStore};
17 : #[cfg(feature = "postgres")]
18 : use super::pg::entity::PgEntityStore;
19 : #[cfg(feature = "postgres")]
20 : use super::pg::temporal::PgTemporalStore;
21 : use super::{ChangeHook, Kind, Store};
22 :
23 : /// 5.5.6: unexpected failures (database errors, timeouts) surface as
24 : /// InternalError. The client-visible detail is deliberately generic —
25 : /// driver internals (SQL text, constraint names, connection state) go to
26 : /// the server log only.
27 : #[cfg(feature = "postgres")]
28 14 : pub(crate) fn db(e: sqlx::Error) -> NgsiError {
29 : // The store's own spec errors travel out through the same sqlx channel
30 : // (the signature is fixed by the callers), so recover them — by VALUE, so
31 : // the variant and therefore the status survive — before the generic
32 : // mapping turns them all into a 500.
33 14 : if let sqlx::Error::Configuration(b) = e {
34 8 : return match b.downcast::<NgsiError>() {
35 6 : Ok(n) => *n,
36 2 : Err(b) => {
37 2 : tracing::error!("database error: {b}");
38 2 : NgsiError::InternalError("database error".into())
39 : }
40 : };
41 6 : }
42 : // The pool handed out no connection inside its acquire timeout: the
43 : // broker is holding every connection it may hold and the caller waited
44 : // the whole wall. That is overload, not a fault — the binding answers it
45 : // 503 with Retry-After (`negotiate::ApiError::Overloaded`), so the detail
46 : // is the constant both ends name.
47 6 : if matches!(e, sqlx::Error::PoolTimedOut) {
48 2 : metrics::counter!("antares_pg_pool_timeouts_total").increment(1);
49 2 : tracing::warn!(
50 : "connection pool exhausted: no connection within the acquire timeout \
51 : (raise ANTARES_PG_POOL, or the request rate is above what this \
52 : database can serve)"
53 : );
54 2 : return NgsiError::InternalError(antares_model::error::DB_OVERLOADED.into());
55 4 : }
56 : // 5.5.2: "database timeouts" are InternalError. SQLSTATE 57014 is the
57 : // session's statement_timeout firing — named in the detail so a wall hit
58 : // reads differently from a broken query in the operator's log.
59 4 : if let sqlx::Error::Database(d) = &e {
60 0 : if d.code().as_deref() == Some("57014") {
61 0 : tracing::warn!("database statement timeout: {d}");
62 0 : return NgsiError::InternalError("database statement timeout".into());
63 0 : }
64 4 : }
65 4 : tracing::error!("database error: {e}");
66 4 : NgsiError::InternalError("database error".into())
67 14 : }
68 :
69 : /// 4.22: the read-boundary "now" — every entity read strips expired docs and
70 : /// instances against this stamp (UTC Z, millisecond precision, the same form
71 : /// the broker writes into system timestamps).
72 22595 : pub(crate) fn now_utc() -> String {
73 22595 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
74 22595 : }
75 :
76 : #[cfg(feature = "postgres")]
77 62345 : fn doc_kind(kind: Kind) -> Result<DocKind, NgsiError> {
78 62345 : match kind {
79 767 : Kind::Subscription => Ok(DocKind::Subscription),
80 20949 : Kind::Registration => Ok(DocKind::Registration),
81 1643 : Kind::CSourceSubscription => Ok(DocKind::CSourceSubscription),
82 18803 : Kind::Snapshot => Ok(DocKind::Snapshot),
83 19055 : Kind::EntityMap => Ok(DocKind::EntityMap),
84 1124 : Kind::DistSub => Ok(DocKind::DistSub),
85 4 : Kind::DeadLetter => Ok(DocKind::DeadLetter),
86 : // Entities and Temporal Representations have their own tables and
87 : // never reach the document store. Every caller routes them to their
88 : // own match arm first; one that does not has mismatched its arms,
89 : // and the mistake stays inside the one request.
90 0 : Kind::Entity | Kind::Temporal => Err(NgsiError::InternalError(format!(
91 0 : "{kind:?} is not a document-store kind"
92 0 : ))),
93 : }
94 62345 : }
95 :
96 : /// Postgres backend bundle: one pool, three table-family stores, plus the
97 : /// change hook (the memory store emits its own; here the seam emits).
98 : #[cfg(feature = "postgres")]
99 : pub struct PgBackend {
100 : pub entities: PgEntityStore,
101 : pub temporal: PgTemporalStore,
102 : pub docs: PgDocStore,
103 : hook: std::sync::RwLock<Option<ChangeHook>>,
104 : /// Server and extension versions, read once at startup
105 : /// (`with_version`); `/q/health` serves this copy rather than querying
106 : /// the database on a polled endpoint.
107 : version: Value,
108 : }
109 :
110 : #[cfg(feature = "postgres")]
111 : impl PgBackend {
112 49 : pub fn new(pool: sqlx::postgres::PgPool) -> Self {
113 49 : Self {
114 49 : entities: PgEntityStore::new(pool.clone()),
115 49 : temporal: PgTemporalStore::new(pool.clone()),
116 49 : docs: PgDocStore::new(pool),
117 49 : hook: std::sync::RwLock::new(None),
118 49 : version: Value::Object(serde_json::Map::new()),
119 49 : }
120 49 : }
121 :
122 : /// Attach what the startup probe read from the server
123 : /// (`pg::version_info`), so `/q/health` can answer it without a query.
124 4 : pub fn with_version(mut self, version: Value) -> Self {
125 4 : self.version = version;
126 4 : self
127 4 : }
128 :
129 : /// Poison recovery (`into_inner`) is deliberate, the same choice the
130 : /// memory arm records: the hook runs real code over attacker-shaped JSON,
131 : /// and a panic inside it must unwind one request, not poison this lock
132 : /// and panic every later entity write until the process restarts.
133 5842 : async fn emit(&self, tenant: &TenantId, before: Option<Value>, after: Option<Value>) {
134 : // cloned out of the lock first: a read guard cannot be held across
135 : // the await the hook now costs.
136 5842 : let hook = self
137 5842 : .hook
138 5842 : .read()
139 5842 : .unwrap_or_else(std::sync::PoisonError::into_inner)
140 5842 : .clone();
141 5842 : if let Some(h) = hook {
142 576 : h(tenant, before, after).await;
143 5266 : }
144 5842 : }
145 : }
146 :
147 : /// What `antares-api` sees. No core crate names redb or sqlx.
148 : // One AnyStore exists per process — variant size difference is irrelevant.
149 : #[allow(clippy::large_enum_variant)]
150 : pub enum AnyStore {
151 : Mem(Store),
152 : #[cfg(feature = "postgres")]
153 : Pg(PgBackend),
154 : }
155 :
156 : /// A row's `id`, or `""` for one without: an id-less row sorts first and is
157 : /// never skipped past, so a keyset walk over it cannot lose it. Only the
158 : /// Postgres arm sorts rows it read whole; the memory arm's maps are already
159 : /// keyed by id.
160 : #[cfg(feature = "postgres")]
161 0 : fn row_id(v: &Value) -> &str {
162 0 : v.get("id").and_then(Value::as_str).unwrap_or_default()
163 0 : }
164 :
165 : impl AnyStore {
166 : /// Readiness ping: can the store answer a trivial request
167 : /// RIGHT NOW? Memory/file are in-process (always ready); the Pg arm runs
168 : /// `SELECT 1` so a lost database (failover, network partition) flips
169 : /// /q/ready to 503 and the Service stops routing to this pod.
170 12 : pub async fn ping(&self) -> Result<(), NgsiError> {
171 12 : match self {
172 12 : AnyStore::Mem(_) => Ok(()),
173 : #[cfg(feature = "postgres")]
174 0 : AnyStore::Pg(p) => async {
175 0 : sqlx::query("SELECT 1")
176 0 : .execute(p.docs.pool())
177 0 : .await
178 0 : .map(|_| ())
179 0 : }
180 0 : .await
181 0 : .map_err(db),
182 : }
183 12 : }
184 :
185 : /// What this store runs on, for `/q/health`. The memory and file modes
186 : /// are one backend with two durability shapes and must not read as the
187 : /// same thing; the Postgres arm serves what the startup probe read.
188 182 : pub fn version_info(&self) -> Value {
189 182 : match self {
190 166 : AnyStore::Mem(s) => serde_json::json!({
191 166 : "engine": if s.shadowed() { "redb" } else { "memory" },
192 : }),
193 : #[cfg(feature = "postgres")]
194 16 : AnyStore::Pg(p) => p.version.clone(),
195 : }
196 182 : }
197 :
198 : /// (Queued writers, peak) behind the single redb committer, reported
199 : /// only by a store that has one: `None` for the Pg arm (Postgres has no
200 : /// single-writer commit queue) and for a pure in-memory store, whose
201 : /// lock depth is not the fsync queue this number stands for.
202 94 : pub fn commit_queue(&self) -> Option<(usize, usize)> {
203 94 : match self {
204 88 : AnyStore::Mem(s) => s.shadowed().then(|| s.commit_queue()),
205 : #[cfg(feature = "postgres")]
206 6 : AnyStore::Pg(_) => None,
207 : }
208 94 : }
209 :
210 : /// The last step of the drain: close the connection pool so in-flight
211 : /// transactions finish and the server sees a clean disconnect instead of
212 : /// N abandoned backends. A no-op for the memory/file arm, whose durability
213 : /// is already commit-before-ack — there is nothing buffered to lose.
214 24 : pub async fn close(&self) {
215 24 : match self {
216 20 : AnyStore::Mem(_) => {}
217 : #[cfg(feature = "postgres")]
218 4 : AnyStore::Pg(p) => p.docs.pool().close().await,
219 : }
220 24 : }
221 :
222 566 : pub fn set_change_hook(&self, h: ChangeHook) {
223 566 : match self {
224 560 : AnyStore::Mem(s) => s.set_change_hook(h),
225 : #[cfg(feature = "postgres")]
226 6 : AnyStore::Pg(p) => {
227 6 : *p.hook
228 6 : .write()
229 6 : .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(h)
230 : }
231 : }
232 566 : }
233 :
234 : /// Turn the same-tx outbox producer on (bus=nats). The memory arm has
235 : /// no outbox — the broker's wiring rejects bus=nats without a Pg store,
236 : /// so this is unreachable there by construction.
237 2 : pub fn set_outbox(
238 2 : &self,
239 2 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] on: bool,
240 2 : ) {
241 2 : match self {
242 0 : AnyStore::Mem(_) => {}
243 : #[cfg(feature = "postgres")]
244 2 : AnyStore::Pg(p) => p.entities.set_outbox(on),
245 : }
246 2 : }
247 :
248 : /// Outbox drain: oldest-first page of pending rows `(seq, tenant, event)`.
249 744 : pub async fn outbox_peek(
250 744 : &self,
251 744 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] limit: i64,
252 744 : ) -> Result<Vec<(i64, String, Value)>, NgsiError> {
253 744 : match self {
254 8 : AnyStore::Mem(_) => Ok(Vec::new()),
255 : #[cfg(feature = "postgres")]
256 736 : AnyStore::Pg(p) => super::pg::outbox::peek(p.docs.pool(), limit)
257 736 : .await
258 736 : .map_err(db),
259 : }
260 744 : }
261 :
262 : /// Outbox drain: delete EXACTLY the published rows (never a blanket
263 : /// `seq <= max`, which loses a row committing between peek and ack).
264 0 : pub async fn outbox_ack(
265 0 : &self,
266 0 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] seqs: &[i64],
267 0 : ) -> Result<u64, NgsiError> {
268 0 : match self {
269 0 : AnyStore::Mem(_) => Ok(0),
270 : #[cfg(feature = "postgres")]
271 0 : AnyStore::Pg(p) => super::pg::outbox::ack(p.docs.pool(), seqs)
272 0 : .await
273 0 : .map_err(db),
274 : }
275 0 : }
276 :
277 : /// Outbox drain: keep the rows of events the bus could not carry whole.
278 : /// The memory arm has no outbox, so nothing is ever retained there.
279 0 : pub async fn outbox_retain(
280 0 : &self,
281 0 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] tenant: &TenantId,
282 0 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] seqs: &[i64],
283 0 : ) -> Result<u64, NgsiError> {
284 0 : match self {
285 0 : AnyStore::Mem(_) => Ok(0),
286 : #[cfg(feature = "postgres")]
287 0 : AnyStore::Pg(p) => super::pg::outbox::retain(p.docs.pool(), tenant, seqs)
288 0 : .await
289 0 : .map_err(db),
290 : }
291 0 : }
292 :
293 : /// The whole event behind a claim-check reference, `None` once reaped.
294 2 : pub async fn outbox_event(
295 2 : &self,
296 2 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] seq: i64,
297 2 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] tenant: &TenantId,
298 2 : ) -> Result<Option<Value>, NgsiError> {
299 2 : match self {
300 0 : AnyStore::Mem(_) => Ok(None),
301 : #[cfg(feature = "postgres")]
302 2 : AnyStore::Pg(p) => super::pg::outbox::event(p.docs.pool(), seq, tenant)
303 2 : .await
304 2 : .map_err(db),
305 : }
306 2 : }
307 :
308 : /// Create one document of `kind`; `false` when that id is already taken.
309 18927 : pub async fn create(
310 18927 : &self,
311 18927 : tenant: &TenantId,
312 18927 : kind: Kind,
313 18927 : id: &str,
314 18927 : doc: Value,
315 18927 : ) -> Result<bool, NgsiError> {
316 18927 : match self {
317 17047 : AnyStore::Mem(s) => Ok(s.create(tenant, kind, id, doc).await),
318 : #[cfg(feature = "postgres")]
319 1880 : AnyStore::Pg(p) => {
320 1880 : let created = match kind {
321 688 : Kind::Entity => p.entities.create(tenant, id, &doc).await.map_err(db)?,
322 126 : Kind::Temporal => p.temporal.create(tenant, id, &doc).await.map_err(db)?,
323 : _ => {
324 1066 : let dk = doc_kind(kind)?;
325 1066 : p.docs.create(tenant, dk, id, &doc).await.map_err(db)?
326 : }
327 : };
328 1880 : if created && kind == Kind::Entity {
329 686 : p.emit(tenant, None, Some(doc)).await;
330 1194 : }
331 1880 : Ok(created)
332 : }
333 : }
334 18927 : }
335 :
336 : /// Batch create (entities only): one multi-row statement on the Pg
337 : /// arm, per-item loop on the memory arm. Created-flags in input order.
338 66 : pub async fn batch_create(
339 66 : &self,
340 66 : tenant: &TenantId,
341 66 : items: Vec<(String, Value)>,
342 66 : ) -> Result<Vec<bool>, NgsiError> {
343 66 : match self {
344 62 : AnyStore::Mem(s) => {
345 62 : let mut out = Vec::with_capacity(items.len());
346 4936 : for (id, doc) in items {
347 4936 : out.push(s.create(tenant, Kind::Entity, &id, doc).await);
348 : }
349 62 : Ok(out)
350 : }
351 : #[cfg(feature = "postgres")]
352 4 : AnyStore::Pg(p) => {
353 4 : let flags = p.entities.batch_create(tenant, &items).await.map_err(db)?;
354 4 : for ((_, doc), created) in items.into_iter().zip(&flags) {
355 4 : if *created {
356 4 : p.emit(tenant, None, Some(doc)).await;
357 0 : }
358 : }
359 4 : Ok(flags)
360 : }
361 : }
362 66 : }
363 :
364 : /// Batch delete (entities only): deleted-flags in input order; a
365 : /// duplicate id in the input deletes once and 404s the second time,
366 : /// matching the per-item loop's semantics (5.5.11.4).
367 198 : pub async fn batch_delete(
368 198 : &self,
369 198 : tenant: &TenantId,
370 198 : ids: &[String],
371 198 : ) -> Result<Vec<bool>, NgsiError> {
372 198 : match self {
373 110 : AnyStore::Mem(s) => {
374 110 : let mut out = Vec::with_capacity(ids.len());
375 4244 : for id in ids {
376 4244 : out.push(s.delete(tenant, Kind::Entity, id).await);
377 : }
378 110 : Ok(out)
379 : }
380 : #[cfg(feature = "postgres")]
381 88 : AnyStore::Pg(p) => {
382 88 : let deleted = p.entities.batch_delete(tenant, ids).await.map_err(db)?;
383 88 : let mut prev: std::collections::HashMap<String, Value> =
384 88 : deleted.into_iter().collect();
385 88 : let mut out = Vec::with_capacity(ids.len());
386 1356 : for id in ids {
387 1356 : match prev.remove(id) {
388 1206 : Some(before) => {
389 1206 : p.emit(tenant, Some(before), None).await;
390 1206 : out.push(true);
391 : }
392 150 : None => out.push(false),
393 : }
394 : }
395 88 : Ok(out)
396 : }
397 : }
398 198 : }
399 :
400 15114 : pub async fn upsert(
401 15114 : &self,
402 15114 : tenant: &TenantId,
403 15114 : kind: Kind,
404 15114 : id: &str,
405 15114 : doc: Value,
406 15114 : ) -> Result<bool, NgsiError> {
407 15114 : match self {
408 12674 : AnyStore::Mem(s) => Ok(s.upsert(tenant, kind, id, doc).await),
409 : #[cfg(feature = "postgres")]
410 2440 : AnyStore::Pg(p) => match kind {
411 : Kind::Entity => {
412 : // replace-or-create without a pre-read: try the replace
413 : // first (captures the true before-image under the row
414 : // lock), fall back to create, and on a lost create race
415 : // replace after all.
416 2420 : let mut prev: Option<Value> = None;
417 : // one shape, two places: a closure cannot hold the await
418 : // and an async closure is not a stable language feature,
419 : // so the replace is written once and expanded twice
420 : macro_rules! replace {
421 : () => {
422 : p.entities
423 2 : .mutate(tenant, id, |d| {
424 2 : prev = Some(d.clone());
425 2 : *d = doc.clone();
426 2 : Ok::<(), std::convert::Infallible>(())
427 2 : })
428 : .await
429 : .map_err(db)?
430 : .is_some()
431 : };
432 : }
433 2420 : let mut existed = replace!();
434 2420 : if !existed {
435 2418 : if p.entities.create(tenant, id, &doc).await.map_err(db)? {
436 2418 : existed = false;
437 2418 : } else {
438 : // lost the create race — replace instead
439 0 : existed = replace!();
440 : }
441 2 : }
442 2420 : p.emit(tenant, prev, Some(doc)).await;
443 2420 : Ok(existed)
444 : }
445 : Kind::Temporal => {
446 0 : if p.temporal.create(tenant, id, &doc).await.map_err(db)? {
447 0 : Ok(false)
448 : } else {
449 0 : p.temporal
450 0 : .mutate(tenant, id, |d| {
451 0 : *d = doc.clone();
452 0 : Ok::<(), std::convert::Infallible>(())
453 0 : })
454 0 : .await
455 0 : .map_err(db)?;
456 0 : Ok(true)
457 : }
458 : }
459 : _ => {
460 20 : let dk = doc_kind(kind)?;
461 20 : p.docs.upsert(tenant, dk, id, &doc).await.map_err(db)
462 : }
463 : },
464 : }
465 15114 : }
466 :
467 20671 : pub async fn get(
468 20671 : &self,
469 20671 : tenant: &TenantId,
470 20671 : kind: Kind,
471 20671 : id: &str,
472 20671 : ) -> Result<Option<Value>, NgsiError> {
473 20671 : let doc = match self {
474 18849 : AnyStore::Mem(s) => s.get(tenant, kind, id),
475 : #[cfg(feature = "postgres")]
476 1822 : AnyStore::Pg(p) => match kind {
477 126 : Kind::Entity => p.entities.get(tenant, id).await.map_err(db)?,
478 96 : Kind::Temporal => p.temporal.get(tenant, id).await.map_err(db)?,
479 1600 : _ => p.docs.get(tenant, doc_kind(kind)?, id).await.map_err(db)?,
480 : },
481 : };
482 : // 4.22: an expired entity is invalid context — a read serves it to
483 : // no one, whichever arm stored it (the Pg sweep lags by design).
484 20671 : if kind == Kind::Entity {
485 11338 : let now = now_utc();
486 11338 : if let Some(mut d) = doc {
487 9876 : if crate::store::filter::strip_expired(&mut d, &now) {
488 14 : return Ok(None);
489 9862 : }
490 9862 : return Ok(Some(d));
491 1462 : }
492 1462 : return Ok(None);
493 9333 : }
494 9333 : Ok(doc)
495 20671 : }
496 :
497 10599 : pub async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> Result<bool, NgsiError> {
498 10599 : match self {
499 6219 : AnyStore::Mem(s) => Ok(s.delete(tenant, kind, id).await),
500 : #[cfg(feature = "postgres")]
501 4380 : AnyStore::Pg(p) => match kind {
502 : Kind::Entity => {
503 : // the before-image comes from the DELETE's own RETURNING —
504 : // same transaction, never a separate racy read
505 1350 : let prev = p.entities.delete(tenant, id).await.map_err(db)?;
506 1350 : let hit = prev.is_some();
507 1350 : if hit {
508 1278 : p.emit(tenant, prev, None).await;
509 72 : }
510 1350 : Ok(hit)
511 : }
512 2534 : Kind::Temporal => p.temporal.delete(tenant, id).await.map_err(db),
513 496 : _ => p.docs.delete(tenant, doc_kind(kind)?, id).await.map_err(db),
514 : },
515 : }
516 10599 : }
517 :
518 : /// Delete one entity only if `keep` accepts the stored document, read
519 : /// and removed under one lock (see the driver trait).
520 2252 : pub async fn delete_entity_if(
521 2252 : &self,
522 2252 : tenant: &TenantId,
523 2252 : id: &str,
524 2252 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
525 2252 : ) -> Result<bool, NgsiError> {
526 : // 4.22 is applied to the document `keep` judges, exactly as `get`
527 : // applies it: an expired instance is not there to be matched on.
528 2252 : let now = now_utc();
529 2252 : let judge = |d: &Value| {
530 472 : let mut d = d.clone();
531 472 : !crate::store::filter::strip_expired(&mut d, &now) && keep(&d)
532 472 : };
533 2252 : match self {
534 1164 : AnyStore::Mem(s) => Ok(s.delete_if(tenant, id, &judge).await.is_some()),
535 : #[cfg(feature = "postgres")]
536 1088 : AnyStore::Pg(p) => {
537 : // the before-image comes from the DELETE's own RETURNING —
538 : // same transaction, never a separate racy read
539 1088 : let prev = p.entities.delete_if(tenant, id, &judge).await.map_err(db)?;
540 1088 : let hit = prev.is_some();
541 1088 : if hit {
542 214 : p.emit(tenant, prev, None).await;
543 874 : }
544 1088 : Ok(hit)
545 : }
546 : }
547 2252 : }
548 :
549 10207 : pub async fn list(&self, tenant: &TenantId, kind: Kind) -> Result<Vec<Value>, NgsiError> {
550 10207 : let mut rows = match self {
551 7576 : AnyStore::Mem(s) => s.list(tenant, kind),
552 : #[cfg(feature = "postgres")]
553 2631 : AnyStore::Pg(p) => match kind {
554 18 : Kind::Entity => p.entities.list(tenant).await.map_err(db)?,
555 4 : Kind::Temporal => p.temporal.list(tenant).await.map_err(db)?,
556 2609 : _ => p.docs.list(tenant, doc_kind(kind)?).await.map_err(db)?,
557 : },
558 : };
559 10207 : if kind == Kind::Entity {
560 71 : let now = now_utc();
561 5370 : rows.retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
562 10136 : }
563 10207 : Ok(rows)
564 10207 : }
565 :
566 : /// One id-ordered page of documents (see `CurrentStateDriver::list_page`).
567 61928 : pub async fn list_page(
568 61928 : &self,
569 61928 : tenant: &TenantId,
570 61928 : kind: Kind,
571 61928 : after: Option<&str>,
572 61928 : limit: usize,
573 61928 : ) -> Result<Vec<Value>, NgsiError> {
574 61928 : match self {
575 5983 : AnyStore::Mem(s) => Ok(s.list_page(tenant, kind, after, limit)),
576 : #[cfg(feature = "postgres")]
577 55945 : AnyStore::Pg(p) => match kind {
578 : // Not doc kinds: they live in their own tables, behind their
579 : // own readers, and `doc_kind` refuses them.
580 : Kind::Entity => {
581 9 : let mut rows = p
582 9 : .entities
583 9 : .list_page(tenant, after, i64::try_from(limit).unwrap_or(i64::MAX))
584 9 : .await
585 9 : .map_err(db)?;
586 : // 4.22: the statement excluded expired ENTITIES, so no
587 : // document leaves this page and its length still means
588 : // what the caller reads it as; expired INSTANCES are
589 : // stripped here, at the read boundary, as everywhere else.
590 9 : let now = now_utc();
591 1014 : rows.retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
592 9 : Ok(rows)
593 : }
594 : // ponytail: temporal is sliced from the whole list, so a
595 : // tenant large enough still refuses this read for volume; the
596 : // upgrade is a keyset reader over the instance rows.
597 : // Temporal has no keyset reader of its own: its documents are
598 : // reconstructed from the instance rows, where the volume that
599 : // needs bounding is instances and not entities — the one kind
600 : // that does not yet honour `list_page`'s contract, and the
601 : // reason nothing internal walks temporal state.
602 : Kind::Temporal => {
603 0 : let mut rows = AnyStore::list(self, tenant, kind).await?;
604 0 : rows.sort_by(|a, b| row_id(a).cmp(row_id(b)));
605 0 : Ok(rows
606 0 : .into_iter()
607 0 : .skip_while(|r| after.is_some_and(|a| row_id(r) <= a))
608 0 : .take(limit)
609 0 : .collect())
610 : }
611 55936 : _ => p
612 55936 : .docs
613 55936 : .list_page(
614 55936 : tenant,
615 55936 : doc_kind(kind)?,
616 55936 : after,
617 55936 : i64::try_from(limit).unwrap_or(i64::MAX),
618 : )
619 55936 : .await
620 55936 : .map_err(db),
621 : },
622 : }
623 61928 : }
624 :
625 : /// One id-ordered window of documents and the total (see
626 : /// `CurrentStateDriver::list_slice`).
627 192 : pub async fn list_slice(
628 192 : &self,
629 192 : tenant: &TenantId,
630 192 : kind: Kind,
631 192 : offset: usize,
632 192 : limit: usize,
633 192 : ) -> Result<(Vec<Value>, usize), NgsiError> {
634 192 : match self {
635 130 : AnyStore::Mem(s) => Ok(s.list_slice(tenant, kind, offset, limit)),
636 : #[cfg(feature = "postgres")]
637 62 : AnyStore::Pg(p) => match kind {
638 : // Not doc kinds; sliced from the whole list, like `list_page`.
639 : Kind::Entity | Kind::Temporal => {
640 0 : let mut rows = AnyStore::list(self, tenant, kind).await?;
641 0 : rows.sort_by(|a, b| row_id(a).cmp(row_id(b)));
642 0 : let total = rows.len();
643 0 : Ok((rows.into_iter().skip(offset).take(limit).collect(), total))
644 : }
645 : _ => {
646 62 : let (page, total) = p
647 62 : .docs
648 62 : .list_slice(
649 62 : tenant,
650 62 : doc_kind(kind)?,
651 62 : i64::try_from(offset).unwrap_or(i64::MAX),
652 62 : i64::try_from(limit).unwrap_or(i64::MAX),
653 : )
654 62 : .await
655 62 : .map_err(db)?;
656 62 : Ok((page, usize::try_from(total).unwrap_or(usize::MAX)))
657 : }
658 : },
659 : }
660 192 : }
661 :
662 : /// 5.12 registration candidates for these entity ids / types. The Pg arm
663 : /// reads the `csource_index` rows (an indexed narrowing); `memory`/`file`
664 : /// have nothing to push into and return the same snapshot `list` does.
665 : /// Either way the result is a SUPERSET — the caller's matcher decides
666 : /// every 5.12 condition, this only avoids reading registrations that
667 : /// cannot match on id or type. `types` must be expanded plain type IRIs
668 : /// (what the registration write stored); a caller holding terms or a 4.17
669 : /// selection expression passes `None` and narrows on ids alone.
670 15924 : pub async fn matching_registrations(
671 15924 : &self,
672 15924 : tenant: &TenantId,
673 15924 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] ids: Option<&[String]>,
674 15924 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] types: Option<&[String]>,
675 15924 : ) -> Result<Vec<Value>, NgsiError> {
676 15924 : match self {
677 12462 : AnyStore::Mem(s) => Ok(s.list(tenant, Kind::Registration)),
678 : #[cfg(feature = "postgres")]
679 3462 : AnyStore::Pg(p) => p
680 3462 : .docs
681 3462 : .matching_registrations(tenant, ids, types)
682 3462 : .await
683 3462 : .map_err(db),
684 : }
685 15924 : }
686 :
687 : /// Query Entities with the filter pushed down where the backend
688 : /// can take it. `memory`/`file` have nothing to push into — their
689 : /// entities are already in RAM — so they return the same snapshot `list`
690 : /// does (never `decided`, never `paged`). Either way the caller applies
691 : /// the exact filter afterwards unless the outcome says SQL already did.
692 1742 : pub async fn query_entities(
693 1742 : &self,
694 1742 : tenant: &TenantId,
695 1742 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))]
696 1742 : f: &crate::store::filter::EntityFilter<'_>,
697 1742 : ) -> Result<crate::store::filter::QueryOutcome, NgsiError> {
698 1742 : let mut outcome = match self {
699 1494 : AnyStore::Mem(s) => crate::store::filter::QueryOutcome {
700 1494 : rows: s.list(tenant, Kind::Entity),
701 1494 : decided: false,
702 1494 : paged: false,
703 1494 : total: None,
704 1494 : },
705 : #[cfg(feature = "postgres")]
706 248 : AnyStore::Pg(p) => p.entities.query(tenant, f).await.map_err(db)?,
707 : };
708 : // 4.22: the Pg arm already excludes expired ENTITIES in SQL (so
709 : // paging/totals stay exact); instance stripping — and the whole job
710 : // on the memory arm — happens here at the read boundary.
711 1742 : let now = now_utc();
712 1742 : outcome
713 1742 : .rows
714 19902 : .retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
715 1742 : Ok(outcome)
716 1742 : }
717 :
718 : /// Query Temporal Evolution with entity narrowing, instance-window
719 : /// pruning AND entity paging pushed down. Same
720 : /// contract: the API's window() is the arbiter; the memory arm returns
721 : /// the full snapshot, never paged.
722 566 : pub async fn query_temporal(
723 566 : &self,
724 566 : tenant: &TenantId,
725 566 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))]
726 566 : f: &crate::store::filter::TemporalFilter<'_>,
727 566 : ) -> Result<crate::store::filter::TemporalOutcome, NgsiError> {
728 566 : let mut outcome = match self {
729 336 : AnyStore::Mem(s) => crate::store::filter::TemporalOutcome {
730 336 : rows: s.list(tenant, Kind::Temporal),
731 336 : paged: false,
732 336 : total: None,
733 336 : aggregated: false,
734 336 : },
735 : #[cfg(feature = "postgres")]
736 230 : AnyStore::Pg(p) => p.temporal.query(tenant, f).await.map_err(db)?,
737 : };
738 : // 4.22 on temporal reads: the Pg arm already dropped expired ENTITIES in
739 : // SQL (paging exact); this strips expired attribute INSTANCES, and does
740 : // the whole job on the memory arm (no SQL to push into).
741 566 : let now = now_utc();
742 566 : outcome
743 566 : .rows
744 1540 : .retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
745 566 : Ok(outcome)
746 566 : }
747 :
748 : /// Auto-recording fast path: append instances to an
749 : /// entity's temporal evolution, creating the meta shell on first touch.
750 : /// Pg: pure multi-row INSERT — no history read, no doc rewrite. Memory:
751 : /// the create-or-extend the mirror always did, under the store lock.
752 : /// `shell` carries the meta members; `additions` maps attr IRI →
753 : /// instance array (instanceIds already stamped by the caller).
754 : ///
755 : /// Both arms record only for an entity that still exists: 5.6.6 deletes
756 : /// the entity and then the temporal evolution recorded for it, so an
757 : /// append overlapping the delete must not recreate history nothing will
758 : /// ever clean again. An instance marked `temporal_only` never holds the
759 : /// entities (they live in another backend) and records unconditionally;
760 : /// there the delete overlap stays a window the temporal delete closes.
761 5348 : pub async fn temporal_append(
762 5348 : &self,
763 5348 : tenant: &TenantId,
764 5348 : id: &str,
765 5348 : shell: &Value,
766 5348 : additions: &Value,
767 5348 : ) -> Result<(), NgsiError> {
768 5348 : match self {
769 5104 : AnyStore::Mem(s) => {
770 5104 : if !s.temporal_only && s.get(tenant, Kind::Entity, id).is_none() {
771 6 : return Ok(());
772 5098 : }
773 5098 : if s.get(tenant, Kind::Temporal, id).is_none() {
774 : // loser of a concurrent create race just extends below
775 5018 : let _ = s.create(tenant, Kind::Temporal, id, shell.clone()).await;
776 80 : }
777 5098 : s.mutate(tenant, Kind::Temporal, id, |doc| {
778 5098 : let target = doc.as_object_mut().ok_or(())?;
779 5098 : if let Some(adds) = additions.as_object() {
780 5296 : for (k, v) in adds {
781 5296 : let incoming: Vec<Value> = v.as_array().cloned().unwrap_or_default();
782 5296 : match target.get_mut(k).and_then(Value::as_array_mut) {
783 : // same instanceId = the same instance corrected
784 : // (the pg arm's ON CONFLICT DO UPDATE)
785 74 : Some(cur) => {
786 74 : for inst in incoming {
787 74 : let iid = inst.get("instanceId");
788 90 : match cur.iter_mut().find(|c| c.get("instanceId") == iid) {
789 10 : Some(slot) => *slot = inst,
790 64 : None => cur.push(inst),
791 : }
792 : }
793 : }
794 5222 : None => {
795 5222 : target.insert(k.clone(), Value::Array(incoming));
796 5222 : }
797 : }
798 : }
799 0 : }
800 5098 : Ok::<(), ()>(())
801 5098 : })
802 5098 : .await;
803 5098 : Ok(())
804 : }
805 : #[cfg(feature = "postgres")]
806 244 : AnyStore::Pg(p) => p
807 244 : .temporal
808 244 : .append(tenant, id, shell, additions)
809 244 : .await
810 244 : .map_err(db),
811 : }
812 5348 : }
813 :
814 : /// Mark this instance as the temporal half of a mixed deployment: the
815 : /// entities live in another backend, so appends never look for them here.
816 2 : pub fn temporal_only(mut self) -> Self {
817 2 : match &mut self {
818 2 : AnyStore::Mem(s) => s.temporal_only = true,
819 : #[cfg(feature = "postgres")]
820 0 : AnyStore::Pg(p) => p.temporal.temporal_only = true,
821 : }
822 2 : self
823 2 : }
824 :
825 : /// Retrieve Temporal Evolution with the same instance pruning.
826 272 : pub async fn get_temporal(
827 272 : &self,
828 272 : tenant: &TenantId,
829 272 : id: &str,
830 272 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))]
831 272 : f: &crate::store::filter::TemporalFilter<'_>,
832 272 : ) -> Result<Option<Value>, NgsiError> {
833 272 : let mut doc = match self {
834 250 : AnyStore::Mem(s) => s.get(tenant, Kind::Temporal, id),
835 : #[cfg(feature = "postgres")]
836 22 : AnyStore::Pg(p) => p.temporal.get_range(tenant, id, f).await.map_err(db)?,
837 : };
838 : // 4.22: expired entity → None (Pg already did this in SQL); otherwise
839 : // strip expired instances.
840 272 : if let Some(d) = &mut doc {
841 234 : let now = now_utc();
842 234 : if crate::store::filter::strip_expired(d, &now) {
843 0 : return Ok(None);
844 234 : }
845 38 : }
846 272 : Ok(doc)
847 272 : }
848 :
849 4745 : pub async fn mutate<T, E>(
850 4745 : &self,
851 4745 : tenant: &TenantId,
852 4745 : kind: Kind,
853 4745 : id: &str,
854 4745 : f: impl FnOnce(&mut Value) -> Result<T, E>,
855 4745 : ) -> Result<Option<Result<T, E>>, NgsiError> {
856 4745 : match self {
857 4539 : AnyStore::Mem(s) => Ok(s.mutate(tenant, kind, id, f).await),
858 : #[cfg(feature = "postgres")]
859 206 : AnyStore::Pg(p) => match kind {
860 : Kind::Entity => {
861 : // before/after captured for the change hook's
862 : // prev_payload INSIDE the row lock — a before-image read
863 : // in its own transaction can belong to a different version
864 : // than the one the lock serialized on.
865 40 : let mut before: Option<Value> = None;
866 40 : let mut after: Option<Value> = None;
867 40 : let r = p
868 40 : .entities
869 40 : .mutate(tenant, id, |d| {
870 36 : before = Some(d.clone());
871 36 : let r = f(d);
872 36 : if r.is_ok() {
873 26 : after = Some(d.clone());
874 26 : }
875 36 : r
876 36 : })
877 40 : .await
878 40 : .map_err(db)?;
879 40 : if let (Some(Ok(_)), Some(a)) = (&r, after) {
880 26 : if before.as_ref() != Some(&a) {
881 26 : p.emit(tenant, before, Some(a)).await;
882 0 : }
883 14 : }
884 40 : Ok(r)
885 : }
886 12 : Kind::Temporal => p.temporal.mutate(tenant, id, f).await.map_err(db),
887 : _ => {
888 : // FOR UPDATE + UPDATE in one tx: a bookkeeping writeback
889 : // racing a DELETE must never resurrect the row (047_06).
890 154 : let dk = doc_kind(kind)?;
891 154 : match p.docs.mutate(tenant, dk, id, f).await.map_err(db)? {
892 70 : Some(r) => Ok(Some(r)),
893 84 : None => Ok(None),
894 : }
895 : }
896 : },
897 : }
898 4745 : }
899 :
900 : /// Batch upsert with REPLACE semantics for entities:
901 : /// one statement + one transaction on the Pg arm, per-item loop on the
902 : /// memory arm. Created-flags in input order.
903 36 : pub async fn batch_upsert(
904 36 : &self,
905 36 : tenant: &TenantId,
906 36 : items: Vec<(String, Value)>,
907 36 : ) -> Result<Vec<bool>, NgsiError> {
908 36 : match self {
909 34 : AnyStore::Mem(s) => {
910 34 : let mut out = Vec::with_capacity(items.len());
911 48 : for (id, doc) in items {
912 48 : out.push(!s.upsert(tenant, Kind::Entity, &id, doc).await);
913 : }
914 34 : Ok(out)
915 : }
916 : #[cfg(feature = "postgres")]
917 2 : AnyStore::Pg(p) => {
918 2 : let out = p
919 2 : .entities
920 2 : .batch_upsert_replace(tenant, &items)
921 2 : .await
922 2 : .map_err(db)?;
923 4 : for ((_, doc), (_, prev)) in items.iter().zip(&out) {
924 4 : p.emit(tenant, prev.clone(), Some(doc.clone())).await;
925 : }
926 2 : Ok(out.into_iter().map(|(created, _)| created).collect())
927 : }
928 : }
929 36 : }
930 :
931 : /// Batch read-modify-write for entities: one transaction + one ordered
932 : /// lock set + one multi-row writeback on the Pg arm; per-item mutate on
933 : /// the memory arm. Results align with `ids` (`None` = absent).
934 42 : pub async fn batch_mutate<E>(
935 42 : &self,
936 42 : tenant: &TenantId,
937 42 : ids: &[String],
938 42 : mut f: impl FnMut(&str, &mut Value) -> Result<(), E>,
939 42 : ) -> Result<Vec<Option<Result<(), E>>>, NgsiError> {
940 42 : match self {
941 40 : AnyStore::Mem(s) => {
942 40 : let mut out = Vec::with_capacity(ids.len());
943 4074 : for id in ids {
944 4074 : out.push(s.mutate(tenant, Kind::Entity, id, |d| f(id, d)).await);
945 : }
946 40 : Ok(out)
947 : }
948 : #[cfg(feature = "postgres")]
949 2 : AnyStore::Pg(p) => {
950 : // hook images captured inside the lock, same as single mutate
951 2 : let mut images: Vec<(Value, Value)> = Vec::new();
952 2 : let r = p
953 2 : .entities
954 4 : .batch_mutate(tenant, ids, |id, d| {
955 4 : let before = d.clone();
956 4 : let r = f(id, d);
957 4 : if r.is_ok() && *d != before {
958 4 : images.push((before, d.clone()));
959 4 : }
960 4 : r
961 4 : })
962 2 : .await
963 2 : .map_err(db)?;
964 4 : for (before, after) in images {
965 4 : p.emit(tenant, Some(before), Some(after)).await;
966 : }
967 2 : Ok(r)
968 : }
969 : }
970 42 : }
971 :
972 : /// 4.22 GC for the memory/file arm (the Pg arm's sweep lives in the
973 : /// maintenance job, mode-switched in the broker). Returns reaped count.
974 776 : pub fn sweep_expired(&self) -> usize {
975 776 : match self {
976 382 : AnyStore::Mem(s) => s.sweep_expired(&now_utc()),
977 : // 4.22 reaping on the Pg arm is `pg::maintenance`'s job
978 : // (`reap_expired_entities` / `reap_expired_instances`), which
979 : // deletes in one indexed statement instead of walking the store;
980 : // reads refuse an expired document either way.
981 : #[cfg(feature = "postgres")]
982 394 : AnyStore::Pg(_) => 0,
983 : }
984 776 : }
985 :
986 : /// 5.5.10: does the Tenant exist? The default Tenant "implicitly exists";
987 : /// others exist once implicitly created by a create operation.
988 : ///
989 : /// It answers for CLIENT tenants. A tenant the broker minted for itself
990 : /// does hold a row — every write claims one, and that row is what keeps
991 : /// the tenant enumerable to `subscription_tenants` across a restart — but
992 : /// nothing asks this about one: the wall that asks sits outside the 6.3.22
993 : /// snapshot scoping and reads the header the caller sent, which cannot
994 : /// name an internal tenant. What a Snapshot's synthetic tenant holds is
995 : /// asserted by the Snapshot document; the row only accounts for it, and
996 : /// `tenant_ids` keeps it out of the customer-account listing.
997 354 : pub async fn tenant_exists(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
998 354 : if tenant.as_str() == TenantId::DEFAULT {
999 14 : return Ok(true);
1000 340 : }
1001 340 : match self {
1002 292 : AnyStore::Mem(s) => Ok(s.tenant_exists(tenant)),
1003 : #[cfg(feature = "postgres")]
1004 48 : AnyStore::Pg(p) => {
1005 48 : async {
1006 48 : let row =
1007 48 : sqlx::query_scalar::<_, i32>("SELECT 1 FROM tenants WHERE tenant_id = $1")
1008 48 : .bind(tenant.as_str())
1009 48 : .fetch_optional(p.docs.pool())
1010 48 : .await
1011 48 : .map_err(db)?;
1012 48 : Ok(row.is_some())
1013 48 : }
1014 48 : .await
1015 : }
1016 : }
1017 354 : }
1018 :
1019 : /// Every tenant name the backend knows, sorted. One cheap query: at the
1020 : /// 10 000-tenant target (ADR-0001) an inventory that counted every kind
1021 : /// of every tenant would be 7 counts per tenant on one transaction, so
1022 : /// the counts live in `tenant_stats_one` and are paid per lookup.
1023 808 : pub async fn tenant_ids(&self) -> Result<Vec<String>, NgsiError> {
1024 808 : match self {
1025 410 : AnyStore::Mem(s) => Ok(s.tenant_ids()),
1026 : #[cfg(feature = "postgres")]
1027 398 : AnyStore::Pg(p) => {
1028 398 : async {
1029 398 : let mut rows: Vec<String> =
1030 398 : sqlx::query_scalar("SELECT tenant_id FROM tenants ORDER BY 1")
1031 398 : .fetch_all(p.docs.pool())
1032 398 : .await
1033 398 : .map_err(db)?;
1034 : // The broker's own tenants claim a row like any other
1035 : // write, because that row is also the enumeration the
1036 : // notification paths walk. The inventory is the list of
1037 : // customer accounts, so they are filtered on the way out.
1038 19509 : rows.retain(|t| !TenantId::is_reserved_str(t));
1039 : // 5.5.10: the default Tenant implicitly exists, whether or
1040 : // not a row was ever written for it.
1041 4370 : if !rows.iter().any(|t| t == TenantId::DEFAULT) {
1042 0 : rows.push(TenantId::DEFAULT.to_owned());
1043 0 : rows.sort();
1044 398 : }
1045 398 : Ok(rows)
1046 398 : }
1047 398 : .await
1048 : }
1049 : }
1050 808 : }
1051 :
1052 : /// What one tenant holds; `None` when it does not exist (5.5.10 keeps
1053 : /// the default Tenant existing even when empty).
1054 26 : pub async fn tenant_stats_one(
1055 26 : &self,
1056 26 : tenant: &TenantId,
1057 26 : ) -> Result<Option<antares_store::TenantStats>, NgsiError> {
1058 26 : if !self.tenant_exists(tenant).await? {
1059 4 : return Ok(None);
1060 22 : }
1061 22 : match self {
1062 20 : AnyStore::Mem(s) => Ok(Some(s.tenant_stats_one(tenant))),
1063 : #[cfg(feature = "postgres")]
1064 2 : AnyStore::Pg(p) => {
1065 2 : async {
1066 2 : let mut tx = p.docs.pool().begin().await.map_err(db)?;
1067 : // the tenant setting is what the RLS-guarded counts see
1068 2 : crate::store::pg::set_tenant(&mut tx, tenant)
1069 2 : .await
1070 2 : .map_err(db)?;
1071 2 : let created_at: Option<String> = sqlx::query_scalar(
1072 2 : "SELECT created_at::text FROM tenants WHERE tenant_id = $1",
1073 2 : )
1074 2 : .bind(tenant.as_str())
1075 2 : .fetch_optional(&mut *tx)
1076 2 : .await
1077 2 : .map_err(db)?;
1078 2 : let c: (i64, i64, i64, i64, i64, i64, i64) = sqlx::query_as(
1079 2 : "SELECT (SELECT count(*) FROM entities WHERE tenant_id = $1),
1080 2 : (SELECT count(*) FROM subscriptions WHERE tenant_id = $1),
1081 2 : (SELECT count(*) FROM csource_registrations WHERE tenant_id = $1),
1082 2 : (SELECT count(*) FROM csource_subscriptions WHERE tenant_id = $1),
1083 2 : (SELECT count(*) FROM snapshots WHERE tenant_id = $1),
1084 2 : (SELECT count(*) FROM entity_map_docs WHERE tenant_id = $1),
1085 2 : (SELECT count(*) FROM dist_subs WHERE tenant_id = $1)",
1086 2 : )
1087 2 : .bind(tenant.as_str())
1088 2 : .fetch_one(&mut *tx)
1089 2 : .await
1090 2 : .map_err(db)?;
1091 2 : tx.commit().await.map_err(db)?;
1092 2 : Ok(Some(antares_store::TenantStats {
1093 2 : tenant: tenant.as_str().to_owned(),
1094 2 : created_at,
1095 2 : entities: c.0 as u64,
1096 2 : subscriptions: c.1 as u64,
1097 2 : registrations: c.2 as u64,
1098 2 : csource_subscriptions: c.3 as u64,
1099 2 : snapshots: c.4 as u64,
1100 2 : entity_maps: c.5 as u64,
1101 2 : dist_subs: c.6 as u64,
1102 2 : }))
1103 2 : }
1104 2 : .await
1105 : }
1106 : }
1107 26 : }
1108 :
1109 : /// Purge the current-state half of one tenant in one transaction;
1110 : /// `false` when the tenant did not exist. The default tenant's row stays.
1111 76 : pub async fn purge_tenant(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
1112 76 : match self {
1113 58 : AnyStore::Mem(s) => Ok(s.purge_tenant(tenant)),
1114 : #[cfg(feature = "postgres")]
1115 18 : AnyStore::Pg(p) => {
1116 18 : async {
1117 18 : let mut tx = p.docs.pool().begin().await.map_err(db)?;
1118 18 : crate::store::pg::set_tenant(&mut tx, tenant)
1119 18 : .await
1120 18 : .map_err(db)?;
1121 18 : let known = sqlx::query_scalar::<_, i32>(
1122 18 : "SELECT 1 FROM tenants WHERE tenant_id = $1 FOR UPDATE",
1123 18 : )
1124 18 : .bind(tenant.as_str())
1125 18 : .fetch_optional(&mut *tx)
1126 18 : .await
1127 18 : .map_err(db)?
1128 18 : .is_some();
1129 18 : if !known {
1130 2 : return Ok(false);
1131 16 : }
1132 176 : for table in [
1133 16 : "entities",
1134 16 : "subscriptions",
1135 16 : "csource_subscriptions",
1136 16 : "csource_registrations",
1137 16 : "csource_index",
1138 16 : "outbox",
1139 16 : "snapshots",
1140 16 : "entity_map_docs",
1141 16 : "dist_subs",
1142 16 : "dead_letters",
1143 16 : // ADR-0021: `tenant_id` is GENERATED from the row's kind
1144 16 : // and owner, so a Cached row's is NULL and this DELETE
1145 16 : // takes only what the Tenant stored itself.
1146 16 : "jsonld_contexts",
1147 16 : ] {
1148 176 : sqlx::query(sqlx::AssertSqlSafe(format!(
1149 176 : "DELETE FROM {table} WHERE tenant_id = $1"
1150 176 : )))
1151 176 : .bind(tenant.as_str())
1152 176 : .execute(&mut *tx)
1153 176 : .await
1154 176 : .map_err(db)?;
1155 : }
1156 16 : if tenant.as_str() != TenantId::DEFAULT {
1157 16 : sqlx::query("DELETE FROM tenants WHERE tenant_id = $1")
1158 16 : .bind(tenant.as_str())
1159 16 : .execute(&mut *tx)
1160 16 : .await
1161 16 : .map_err(db)?;
1162 0 : }
1163 16 : tx.commit().await.map_err(db)?;
1164 16 : Ok(true)
1165 18 : }
1166 18 : .await
1167 : }
1168 : }
1169 76 : }
1170 :
1171 692 : pub async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError> {
1172 692 : match self {
1173 666 : AnyStore::Mem(s) => Ok(s.subscription_tenants()),
1174 : // Pg answers the superset the contract allows: every known
1175 : // tenant. Narrowing it would need a cross-tenant read of
1176 : // `subscriptions`, and that table is deliberately outside the
1177 : // `antares.service` escape — a subscription document carries
1178 : // endpoint.receiverInfo, i.e. the credentials the notification
1179 : // is sent with. The callers list per tenant under set_tenant
1180 : // afterwards, so the extra names cost one empty list each.
1181 : #[cfg(feature = "postgres")]
1182 26 : AnyStore::Pg(p) => {
1183 26 : async {
1184 26 : let rows =
1185 26 : sqlx::query_scalar::<_, String>("SELECT tenant_id FROM tenants ORDER BY 1")
1186 26 : .fetch_all(p.docs.pool())
1187 26 : .await
1188 26 : .map_err(db)?;
1189 26 : Ok(rows)
1190 26 : }
1191 26 : .await
1192 : }
1193 : }
1194 692 : }
1195 :
1196 650 : pub async fn context_put(
1197 650 : &self,
1198 650 : tenant: Option<&TenantId>,
1199 650 : id: &str,
1200 650 : doc: Value,
1201 650 : ) -> Result<(), NgsiError> {
1202 : // ADR-0021, the write half no `WHERE` on an existing row can state:
1203 : // the row that ARRIVES must belong to the caller too, or a Tenant
1204 : // could store term mappings under another Tenant's name and decide
1205 : // what that Tenant's payloads mean (5.5.7). Postgres says the same
1206 : // thing as the policy's WITH CHECK; this holds under the roles that
1207 : // bypass RLS.
1208 650 : if !antares_store::context_row_visible(&doc, tenant) {
1209 6 : return Err(NgsiError::InternalError(
1210 6 : "@context belongs to another tenant".into(),
1211 6 : ));
1212 644 : }
1213 644 : match self {
1214 486 : AnyStore::Mem(s) => s.context_put(tenant, id, doc),
1215 : #[cfg(feature = "postgres")]
1216 158 : AnyStore::Pg(p) => {
1217 158 : let kind = doc
1218 158 : .get("kind")
1219 158 : .and_then(Value::as_str)
1220 158 : .unwrap_or("Cached")
1221 158 : .to_owned();
1222 158 : p.docs
1223 158 : .context_put(tenant, id, &doc, &kind)
1224 158 : .await
1225 158 : .map_err(db)
1226 : }
1227 : }
1228 650 : }
1229 :
1230 1176 : pub async fn context_get(
1231 1176 : &self,
1232 1176 : tenant: Option<&TenantId>,
1233 1176 : id: &str,
1234 1176 : ) -> Result<Option<Value>, NgsiError> {
1235 1176 : match self {
1236 854 : AnyStore::Mem(s) => Ok(s.context_get(tenant, id)),
1237 : #[cfg(feature = "postgres")]
1238 322 : AnyStore::Pg(p) => p.docs.context_get(tenant, id).await.map_err(db),
1239 : }
1240 1176 : }
1241 :
1242 222 : pub async fn context_delete(
1243 222 : &self,
1244 222 : tenant: Option<&TenantId>,
1245 222 : id: &str,
1246 222 : ) -> Result<bool, NgsiError> {
1247 222 : match self {
1248 124 : AnyStore::Mem(s) => Ok(s.context_delete(tenant, id)),
1249 : #[cfg(feature = "postgres")]
1250 98 : AnyStore::Pg(p) => p.docs.context_delete(tenant, id).await.map_err(db),
1251 : }
1252 222 : }
1253 :
1254 218 : pub async fn context_list_meta(
1255 218 : &self,
1256 218 : tenant: Option<&TenantId>,
1257 218 : ) -> Result<Vec<Value>, NgsiError> {
1258 218 : match self {
1259 152 : AnyStore::Mem(s) => Ok(s.context_list_meta(tenant)),
1260 : #[cfg(feature = "postgres")]
1261 66 : AnyStore::Pg(p) => p.docs.context_list_meta(tenant).await.map_err(db),
1262 : }
1263 218 : }
1264 : }
1265 :
1266 : #[cfg(all(test, feature = "postgres"))]
1267 : mod db_error_tests {
1268 : #[allow(unused_imports)]
1269 : use antares_model::NgsiError;
1270 :
1271 : /// An acquire timeout is overload, not a fault, and the HTTP binding
1272 : /// recognises it by this exact detail. If the two ends ever spell it
1273 : /// differently the broker answers 500 to a condition a client could
1274 : /// have retried, so the constant is asserted rather than the words.
1275 : #[test]
1276 2 : fn a_pool_timeout_is_marked_as_overload() {
1277 2 : let pd = super::db(sqlx::Error::PoolTimedOut).to_problem_details();
1278 2 : assert_eq!(pd.detail, antares_model::error::DB_OVERLOADED);
1279 2 : assert_eq!(pd.title, "InternalError");
1280 : // and no other sqlx error borrows the mark
1281 2 : assert_ne!(
1282 2 : super::db(sqlx::Error::RowNotFound)
1283 2 : .to_problem_details()
1284 : .detail,
1285 : antares_model::error::DB_OVERLOADED
1286 : );
1287 2 : }
1288 :
1289 : /// 5.5.6 InternalError: the RFC 7807 `detail` a client sees must be
1290 : /// generic — driver internals (SQL text, row counts, connection
1291 : /// strings) belong in the server log, never in the response body.
1292 : #[test]
1293 2 : fn db_error_detail_is_generic() {
1294 2 : let pd = super::db(sqlx::Error::RowNotFound).to_problem_details();
1295 2 : assert_eq!(pd.detail, "database error");
1296 2 : assert_eq!(pd.title, "InternalError");
1297 2 : assert!(
1298 2 : !pd.detail.contains("no rows"),
1299 : "sqlx internals leaked into the client-visible detail: {}",
1300 : pd.detail
1301 : );
1302 : // a configuration error that is NOT one of ours stays generic too
1303 2 : let pd = super::db(sqlx::Error::Configuration("boom".into())).to_problem_details();
1304 2 : assert_eq!(pd.detail, "database error");
1305 2 : assert!(!pd.detail.contains("boom"), "{}", pd.detail);
1306 2 : }
1307 :
1308 : /// A spec error the store raised itself travels out through the driver
1309 : /// error channel. Rebuilding it as a fixed variant forces every one of
1310 : /// them to that variant's status — a 400 BadRequestData would reach the
1311 : /// client as a 403.
1312 : #[test]
1313 2 : fn a_store_raised_spec_error_keeps_its_own_status() {
1314 6 : for (err, kind, status) in [
1315 2 : (NgsiError::BadRequestData("x".into()), "BadRequestData", 400),
1316 2 : (NgsiError::TooManyResults("x".into()), "TooManyResults", 403),
1317 2 : (NgsiError::AlreadyExists("x".into()), "AlreadyExists", 409),
1318 2 : ] {
1319 6 : let out = super::db(sqlx::Error::Configuration(Box::new(err)));
1320 6 : assert_eq!(out.kind(), kind);
1321 6 : assert_eq!(out.status(), status, "{kind} lost its status");
1322 : }
1323 2 : }
1324 : }
1325 :
1326 : // The driver seam: `AnyStore` carries both driver interfaces, delegating to
1327 : // the inherent methods above. New backends implement the traits directly —
1328 : // this enum stays an implementation detail of the built-in backends, no
1329 : // longer the API surface.
1330 : #[async_trait::async_trait]
1331 : impl antares_store::CurrentStateDriver for AnyStore {
1332 12 : async fn ping(&self) -> Result<(), NgsiError> {
1333 : AnyStore::ping(self).await
1334 12 : }
1335 90 : fn commit_queue(&self) -> Option<(usize, usize)> {
1336 90 : AnyStore::commit_queue(self)
1337 90 : }
1338 90 : fn version_info(&self) -> Value {
1339 90 : AnyStore::version_info(self)
1340 90 : }
1341 16 : async fn close(&self) {
1342 : AnyStore::close(self).await;
1343 16 : }
1344 566 : fn set_change_hook(&self, h: super::ChangeHook) {
1345 566 : AnyStore::set_change_hook(self, h);
1346 566 : }
1347 0 : fn set_outbox(&self, on: bool) {
1348 0 : AnyStore::set_outbox(self, on);
1349 0 : }
1350 744 : async fn outbox_peek(&self, limit: i64) -> Result<Vec<(i64, String, Value)>, NgsiError> {
1351 : AnyStore::outbox_peek(self, limit).await
1352 744 : }
1353 0 : async fn outbox_ack(&self, seqs: &[i64]) -> Result<u64, NgsiError> {
1354 : AnyStore::outbox_ack(self, seqs).await
1355 0 : }
1356 0 : async fn outbox_retain(&self, tenant: &TenantId, seqs: &[i64]) -> Result<u64, NgsiError> {
1357 : AnyStore::outbox_retain(self, tenant, seqs).await
1358 0 : }
1359 2 : async fn outbox_event(&self, seq: i64, tenant: &TenantId) -> Result<Option<Value>, NgsiError> {
1360 : AnyStore::outbox_event(self, seq, tenant).await
1361 2 : }
1362 : async fn create(
1363 : &self,
1364 : tenant: &TenantId,
1365 : kind: Kind,
1366 : id: &str,
1367 : doc: Value,
1368 18539 : ) -> Result<bool, NgsiError> {
1369 : AnyStore::create(self, tenant, kind, id, doc).await
1370 18539 : }
1371 : async fn batch_create(
1372 : &self,
1373 : tenant: &TenantId,
1374 : items: Vec<(String, Value)>,
1375 66 : ) -> Result<Vec<bool>, NgsiError> {
1376 : AnyStore::batch_create(self, tenant, items).await
1377 66 : }
1378 : async fn batch_delete(
1379 : &self,
1380 : tenant: &TenantId,
1381 : ids: &[String],
1382 198 : ) -> Result<Vec<bool>, NgsiError> {
1383 : AnyStore::batch_delete(self, tenant, ids).await
1384 198 : }
1385 : async fn batch_upsert(
1386 : &self,
1387 : tenant: &TenantId,
1388 : items: Vec<(String, Value)>,
1389 36 : ) -> Result<Vec<bool>, NgsiError> {
1390 : AnyStore::batch_upsert(self, tenant, items).await
1391 36 : }
1392 : async fn upsert(
1393 : &self,
1394 : tenant: &TenantId,
1395 : kind: Kind,
1396 : id: &str,
1397 : doc: Value,
1398 15098 : ) -> Result<bool, NgsiError> {
1399 : AnyStore::upsert(self, tenant, kind, id, doc).await
1400 15098 : }
1401 : async fn get(
1402 : &self,
1403 : tenant: &TenantId,
1404 : kind: Kind,
1405 : id: &str,
1406 20335 : ) -> Result<Option<Value>, NgsiError> {
1407 : AnyStore::get(self, tenant, kind, id).await
1408 20335 : }
1409 2549 : async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> Result<bool, NgsiError> {
1410 : AnyStore::delete(self, tenant, kind, id).await
1411 2549 : }
1412 : async fn delete_entity_if(
1413 : &self,
1414 : tenant: &TenantId,
1415 : id: &str,
1416 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
1417 2252 : ) -> Result<bool, NgsiError> {
1418 : AnyStore::delete_entity_if(self, tenant, id, keep).await
1419 2252 : }
1420 10191 : async fn list(&self, tenant: &TenantId, kind: Kind) -> Result<Vec<Value>, NgsiError> {
1421 : AnyStore::list(self, tenant, kind).await
1422 10191 : }
1423 : async fn list_page(
1424 : &self,
1425 : tenant: &TenantId,
1426 : kind: Kind,
1427 : after: Option<&str>,
1428 : limit: usize,
1429 61928 : ) -> Result<Vec<Value>, NgsiError> {
1430 : AnyStore::list_page(self, tenant, kind, after, limit).await
1431 61928 : }
1432 : async fn list_slice(
1433 : &self,
1434 : tenant: &TenantId,
1435 : kind: Kind,
1436 : offset: usize,
1437 : limit: usize,
1438 192 : ) -> Result<(Vec<Value>, usize), NgsiError> {
1439 : AnyStore::list_slice(self, tenant, kind, offset, limit).await
1440 192 : }
1441 : async fn matching_registrations(
1442 : &self,
1443 : tenant: &TenantId,
1444 : ids: Option<&[String]>,
1445 : types: Option<&[String]>,
1446 15924 : ) -> Result<Vec<Value>, NgsiError> {
1447 : AnyStore::matching_registrations(self, tenant, ids, types).await
1448 15924 : }
1449 : async fn query_entities(
1450 : &self,
1451 : tenant: &TenantId,
1452 : f: &crate::store::filter::EntityFilter<'_>,
1453 1658 : ) -> Result<crate::store::filter::QueryOutcome, NgsiError> {
1454 : AnyStore::query_entities(self, tenant, f).await
1455 1658 : }
1456 : async fn mutate_boxed<'a>(
1457 : &self,
1458 : tenant: &TenantId,
1459 : kind: Kind,
1460 : id: &str,
1461 : f: antares_store::MutateFn<'a>,
1462 4567 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1463 : AnyStore::mutate(self, tenant, kind, id, f).await
1464 4567 : }
1465 : async fn batch_mutate_boxed<'a>(
1466 : &self,
1467 : tenant: &TenantId,
1468 : ids: &[String],
1469 : mut f: antares_store::BatchMutateFn<'a>,
1470 42 : ) -> Result<Vec<Option<Result<(), ()>>>, NgsiError> {
1471 4052 : AnyStore::batch_mutate(self, tenant, ids, |id, v| f(id, v)).await
1472 42 : }
1473 : /// The Postgres arm answers this in ONE statement (see
1474 : /// `pg::doc::record_delivery`); every other arm runs the shared rule as a
1475 : /// `mutate`, which is what the one statement is a translation of.
1476 : async fn record_delivery(
1477 : &self,
1478 : tenant: &TenantId,
1479 : kind: Kind,
1480 : id: &str,
1481 : now: &str,
1482 402 : ) -> Result<Option<antares_store::Delivery>, NgsiError> {
1483 : // `doc_kind` exists only with the Postgres arm, so its call lives
1484 : // inside the gate with it: outside the feature this function is the
1485 : // shared rule and nothing else.
1486 : #[cfg(feature = "postgres")]
1487 : {
1488 : if let (AnyStore::Pg(p), Ok(dk)) = (self, doc_kind(kind)) {
1489 : let found = p
1490 : .docs
1491 : .record_delivery(tenant, dk, id, now)
1492 : .await
1493 : .map_err(db)?;
1494 : return Ok(
1495 10 : found.map(|(doc, prev_success)| antares_store::Delivery { doc, prev_success })
1496 : );
1497 : }
1498 : }
1499 : antares_store::record_delivery_via_mutate(self, tenant, kind, id, now).await
1500 402 : }
1501 776 : async fn sweep_expired(&self) -> usize {
1502 : AnyStore::sweep_expired(self)
1503 776 : }
1504 322 : async fn tenant_exists(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
1505 : AnyStore::tenant_exists(self, tenant).await
1506 322 : }
1507 690 : async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError> {
1508 : AnyStore::subscription_tenants(self).await
1509 690 : }
1510 804 : async fn tenant_ids(&self) -> Result<Vec<String>, NgsiError> {
1511 : AnyStore::tenant_ids(self).await
1512 804 : }
1513 : async fn tenant_stats_one(
1514 : &self,
1515 : tenant: &TenantId,
1516 22 : ) -> Result<Option<antares_store::TenantStats>, NgsiError> {
1517 : AnyStore::tenant_stats_one(self, tenant).await
1518 22 : }
1519 76 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
1520 : AnyStore::purge_tenant(self, tenant).await
1521 76 : }
1522 : async fn context_put(
1523 : &self,
1524 : tenant: Option<&TenantId>,
1525 : id: &str,
1526 : doc: Value,
1527 650 : ) -> Result<(), NgsiError> {
1528 : AnyStore::context_put(self, tenant, id, doc).await
1529 650 : }
1530 : async fn context_get(
1531 : &self,
1532 : tenant: Option<&TenantId>,
1533 : id: &str,
1534 1176 : ) -> Result<Option<Value>, NgsiError> {
1535 : AnyStore::context_get(self, tenant, id).await
1536 1176 : }
1537 222 : async fn context_delete(&self, tenant: Option<&TenantId>, id: &str) -> Result<bool, NgsiError> {
1538 : AnyStore::context_delete(self, tenant, id).await
1539 222 : }
1540 218 : async fn context_list_meta(&self, tenant: Option<&TenantId>) -> Result<Vec<Value>, NgsiError> {
1541 : AnyStore::context_list_meta(self, tenant).await
1542 218 : }
1543 : }
1544 :
1545 : #[async_trait::async_trait]
1546 : impl antares_store::TemporalDriver for AnyStore {
1547 8 : async fn close(&self) {
1548 : AnyStore::close(self).await;
1549 8 : }
1550 84 : fn version_info(&self) -> Value {
1551 84 : AnyStore::version_info(self)
1552 84 : }
1553 22 : async fn attr_instance_count(&self, tenant: &TenantId) -> Result<u64, NgsiError> {
1554 : match self {
1555 : AnyStore::Mem(s) => Ok(s.attr_instance_count(tenant)),
1556 : #[cfg(feature = "postgres")]
1557 : AnyStore::Pg(p) => {
1558 2 : async {
1559 2 : let mut tx = p.docs.pool().begin().await.map_err(db)?;
1560 2 : crate::store::pg::set_tenant(&mut tx, tenant)
1561 2 : .await
1562 2 : .map_err(db)?;
1563 2 : let n: i64 = sqlx::query_scalar(
1564 2 : "SELECT count(*) FROM attr_instances WHERE tenant_id = $1",
1565 2 : )
1566 2 : .bind(tenant.as_str())
1567 2 : .fetch_one(&mut *tx)
1568 2 : .await
1569 2 : .map_err(db)?;
1570 2 : Ok(n as u64)
1571 2 : }
1572 : .await
1573 : }
1574 : }
1575 22 : }
1576 72 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<(), NgsiError> {
1577 : match self {
1578 : AnyStore::Mem(s) => {
1579 : s.purge_kinds(tenant, &[Kind::Temporal]);
1580 : Ok(())
1581 : }
1582 : #[cfg(feature = "postgres")]
1583 : AnyStore::Pg(p) => {
1584 14 : async {
1585 14 : let mut tx = p.docs.pool().begin().await.map_err(db)?;
1586 14 : crate::store::pg::set_tenant(&mut tx, tenant)
1587 14 : .await
1588 14 : .map_err(db)?;
1589 28 : for table in ["attr_instances", "temporal_entities"] {
1590 28 : sqlx::query(sqlx::AssertSqlSafe(format!(
1591 28 : "DELETE FROM {table} WHERE tenant_id = $1"
1592 28 : )))
1593 28 : .bind(tenant.as_str())
1594 28 : .execute(&mut *tx)
1595 28 : .await
1596 28 : .map_err(db)?;
1597 : }
1598 14 : tx.commit().await.map_err(db)
1599 14 : }
1600 : .await
1601 : }
1602 : }
1603 72 : }
1604 : async fn temporal_append(
1605 : &self,
1606 : tenant: &TenantId,
1607 : id: &str,
1608 : shell: &Value,
1609 : additions: &Value,
1610 5340 : ) -> Result<(), NgsiError> {
1611 : AnyStore::temporal_append(self, tenant, id, shell, additions).await
1612 5340 : }
1613 : /// Only the Postgres arm pages in SQL, and only an exact prefilter
1614 : /// makes that page the evaluator's page; the memory arm answers
1615 : /// `paged: false` and the caller pages.
1616 212 : fn q_pushdown_exact(
1617 212 : &self,
1618 212 : q: &antares_ql::QNode,
1619 212 : range: Option<&crate::store::filter::InstanceRange<'_>>,
1620 212 : expand: &dyn Fn(&str) -> String,
1621 212 : ) -> bool {
1622 212 : match self {
1623 : // the memory arm pages nothing, so the filter is not its business
1624 : AnyStore::Mem(_) => {
1625 114 : let _ = (q, range, expand);
1626 114 : false
1627 : }
1628 : #[cfg(feature = "postgres")]
1629 98 : AnyStore::Pg(_) => crate::compile::qprefilter::prefilter_exact(q, range, expand),
1630 : }
1631 212 : }
1632 : async fn query_temporal(
1633 : &self,
1634 : tenant: &TenantId,
1635 : f: &crate::store::filter::TemporalFilter<'_>,
1636 522 : ) -> Result<crate::store::filter::TemporalOutcome, NgsiError> {
1637 : AnyStore::query_temporal(self, tenant, f).await
1638 522 : }
1639 : async fn get_temporal(
1640 : &self,
1641 : tenant: &TenantId,
1642 : id: &str,
1643 : f: &crate::store::filter::TemporalFilter<'_>,
1644 268 : ) -> Result<Option<Value>, NgsiError> {
1645 : AnyStore::get_temporal(self, tenant, id, f).await
1646 268 : }
1647 332 : async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, NgsiError> {
1648 : AnyStore::get(self, tenant, Kind::Temporal, id).await
1649 332 : }
1650 296 : async fn create(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError> {
1651 : AnyStore::create(self, tenant, Kind::Temporal, id, doc).await
1652 296 : }
1653 16 : async fn upsert(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError> {
1654 : AnyStore::upsert(self, tenant, Kind::Temporal, id, doc).await
1655 16 : }
1656 7936 : async fn delete(&self, tenant: &TenantId, id: &str) -> Result<bool, NgsiError> {
1657 : AnyStore::delete(self, tenant, Kind::Temporal, id).await
1658 7936 : }
1659 14 : async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, NgsiError> {
1660 : AnyStore::list(self, tenant, Kind::Temporal).await
1661 14 : }
1662 : async fn mutate_boxed<'a>(
1663 : &self,
1664 : tenant: &TenantId,
1665 : id: &str,
1666 : f: antares_store::MutateFn<'a>,
1667 178 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1668 : AnyStore::mutate(self, tenant, Kind::Temporal, id, f).await
1669 178 : }
1670 : }
1671 :
1672 : #[cfg(test)]
1673 : mod temporal_only_tests {
1674 : use super::*;
1675 : use antares_model::TenantId;
1676 :
1677 4 : async fn append(store: &AnyStore) -> Option<Value> {
1678 4 : let t = TenantId::new("combo").expect("tenant");
1679 4 : let shell = serde_json::json!({"id": "urn:a", "type": ["T"]});
1680 4 : let adds = serde_json::json!({"speed": [{"instanceId": "urn:i:1", "value": 1}]});
1681 4 : store
1682 4 : .temporal_append(&t, "urn:a", &shell, &adds)
1683 4 : .await
1684 4 : .expect("append");
1685 4 : store.get(&t, Kind::Temporal, "urn:a").await.expect("get")
1686 4 : }
1687 :
1688 : /// A shared instance records only for an entity it holds (5.6.6 delete
1689 : /// overlap); the temporal half of a mixed deployment records
1690 : /// unconditionally, since the entities live elsewhere.
1691 : #[tokio::test]
1692 2 : async fn a_temporal_only_instance_records_without_the_entity() {
1693 2 : assert!(append(&AnyStore::Mem(Store::default())).await.is_none());
1694 2 : let doc = append(&AnyStore::Mem(Store::default()).temporal_only())
1695 2 : .await
1696 2 : .expect("recorded");
1697 2 : assert_eq!(doc["speed"][0]["value"], 1);
1698 2 : }
1699 : }
|
