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 : #[cfg(feature = "postgres")]
23 : use antares_store::ChangeVersion;
24 :
25 : /// 5.5.6: unexpected failures (database errors, timeouts) surface as
26 : /// InternalError. The client-visible detail is deliberately generic —
27 : /// driver internals (SQL text, constraint names, connection state) go to
28 : /// the server log only.
29 : #[cfg(feature = "postgres")]
30 69 : pub(crate) fn db(e: sqlx::Error) -> NgsiError {
31 : // The store's own spec errors travel out through the same sqlx channel
32 : // (the signature is fixed by the callers), so recover them — by VALUE, so
33 : // the variant and therefore the status survive — before the generic
34 : // mapping turns them all into a 500.
35 69 : if let sqlx::Error::Configuration(b) = e {
36 8 : return match b.downcast::<NgsiError>() {
37 6 : Ok(n) => *n,
38 2 : Err(b) => {
39 2 : tracing::error!("database error: {b}");
40 2 : NgsiError::InternalError("database error".into())
41 : }
42 : };
43 61 : }
44 : // The pool handed out no connection inside its acquire timeout: the
45 : // broker is holding every connection it may hold and the caller waited
46 : // the whole wall. That is overload, not a fault — the binding answers it
47 : // 503 with Retry-After (`negotiate::ApiError::Overloaded`), so the detail
48 : // is the constant both ends name.
49 61 : if matches!(e, sqlx::Error::PoolTimedOut) {
50 2 : metrics::counter!("antares_pg_pool_timeouts_total").increment(1);
51 2 : tracing::warn!(
52 : "connection pool exhausted: no connection within the acquire timeout \
53 : (raise ANTARES_PG_POOL, or the request rate is above what this \
54 : database can serve)"
55 : );
56 2 : return NgsiError::InternalError(antares_model::error::DB_OVERLOADED.into());
57 59 : }
58 : // 5.5.2: "database timeouts" are InternalError. SQLSTATE 57014 is the
59 : // session's statement_timeout firing — named in the detail so a wall hit
60 : // reads differently from a broken query in the operator's log.
61 59 : if let sqlx::Error::Database(d) = &e {
62 6 : if d.code().as_deref() == Some("57014") {
63 6 : tracing::warn!("database statement timeout: {d}");
64 6 : return NgsiError::InternalError("database statement timeout".into());
65 0 : }
66 53 : }
67 53 : tracing::error!("database error: {e}");
68 53 : NgsiError::InternalError("database error".into())
69 69 : }
70 :
71 : /// 4.22: the read-boundary "now" — every entity read strips expired docs and
72 : /// instances against this stamp (UTC Z, millisecond precision, the same form
73 : /// the broker writes into system timestamps).
74 68924 : pub(crate) fn now_utc() -> String {
75 68924 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
76 68924 : }
77 :
78 : #[cfg(feature = "postgres")]
79 611697 : fn doc_kind(kind: Kind) -> Result<DocKind, NgsiError> {
80 611697 : match kind {
81 144647 : Kind::Subscription => Ok(DocKind::Subscription),
82 173395 : Kind::Registration => Ok(DocKind::Registration),
83 14552 : Kind::CSourceSubscription => Ok(DocKind::CSourceSubscription),
84 137299 : Kind::Snapshot => Ok(DocKind::Snapshot),
85 137504 : Kind::EntityMap => Ok(DocKind::EntityMap),
86 4292 : Kind::DistSub => Ok(DocKind::DistSub),
87 4 : Kind::DeadLetter => Ok(DocKind::DeadLetter),
88 4 : Kind::View => Ok(DocKind::View),
89 : // Entities and Temporal Representations have their own tables and
90 : // never reach the document store. Every caller routes them to their
91 : // own match arm first; one that does not has mismatched its arms,
92 : // and the mistake stays inside the one request.
93 0 : Kind::Entity | Kind::Temporal => Err(NgsiError::InternalError(format!(
94 0 : "{kind:?} is not a document-store kind"
95 0 : ))),
96 : }
97 611697 : }
98 :
99 : /// A transaction scoped to one tenant: begun, then RLS set for it first.
100 : #[cfg(feature = "postgres")]
101 82 : async fn tenant_tx(
102 82 : p: &PgBackend,
103 82 : tenant: &TenantId,
104 82 : ) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, NgsiError> {
105 82 : let mut tx = p.docs.pool().begin().await.map_err(db)?;
106 82 : crate::store::pg::set_tenant(&mut tx, tenant)
107 82 : .await
108 82 : .map_err(db)?;
109 82 : Ok(tx)
110 82 : }
111 :
112 : /// Postgres backend bundle: one pool, three table-family stores, plus the
113 : /// change hook (the memory store emits its own; here the seam emits).
114 : #[cfg(feature = "postgres")]
115 : pub struct PgBackend {
116 : pub entities: PgEntityStore,
117 : pub temporal: PgTemporalStore,
118 : pub docs: PgDocStore,
119 : hook: std::sync::RwLock<Option<ChangeHook>>,
120 : /// Server and extension versions, read once at startup
121 : /// (`with_version`); `/q/health` serves this copy rather than querying
122 : /// the database on a polled endpoint.
123 : version: Value,
124 : }
125 :
126 : #[cfg(feature = "postgres")]
127 : impl PgBackend {
128 106 : pub fn new(pool: sqlx::postgres::PgPool) -> Self {
129 106 : Self {
130 106 : entities: PgEntityStore::new(pool.clone()),
131 106 : temporal: PgTemporalStore::new(pool.clone()),
132 106 : docs: PgDocStore::new(pool),
133 106 : hook: std::sync::RwLock::new(None),
134 106 : version: Value::Object(serde_json::Map::new()),
135 106 : }
136 106 : }
137 :
138 : /// Seal the notification credentials under `ANTARES_SECRET_KEY`
139 : /// ([`super::pg::seal`]).
140 4 : pub fn with_secret_key(mut self, key: super::pg::seal::SecretKey) -> Self {
141 4 : self.docs = self.docs.with_secret_key(key);
142 4 : self
143 4 : }
144 :
145 : /// Give the batch writes their own pool (`ANTARES_PG_WRITE_RESERVE`):
146 : /// reads keep the main pool, so a flood of them cannot queue
147 : /// ahead of a pipeline's batch upsert.
148 0 : pub fn with_write_reserve(mut self, reserve: sqlx::postgres::PgPool) -> Self {
149 0 : self.entities = self.entities.with_batch_pool(reserve);
150 0 : self
151 0 : }
152 :
153 : /// Attach what the startup probe read from the server
154 : /// (`pg::version_info`), so `/q/health` can answer it without a query.
155 4 : pub fn with_version(mut self, version: Value) -> Self {
156 4 : self.version = version;
157 4 : self
158 4 : }
159 :
160 : /// Poison recovery (`into_inner`) is deliberate, the same choice the
161 : /// memory arm records: the hook runs real code over attacker-shaped JSON,
162 : /// and a panic inside it must unwind one request, not poison this lock
163 : /// and panic every later entity write until the process restarts.
164 9508 : async fn emit(
165 9508 : &self,
166 9508 : tenant: &TenantId,
167 9508 : before: Option<Value>,
168 9508 : after: Option<Value>,
169 9508 : version: ChangeVersion,
170 9508 : ) {
171 : // cloned out of the lock first: a read guard cannot be held across
172 : // the await the hook now costs.
173 9508 : let hook = self
174 9508 : .hook
175 9508 : .read()
176 9508 : .unwrap_or_else(std::sync::PoisonError::into_inner)
177 9508 : .clone();
178 9508 : if let Some(h) = hook {
179 4120 : h(tenant, before, after, version).await;
180 5388 : }
181 9508 : }
182 : }
183 :
184 : /// Instances per Attribute of one temporal document: every array member
185 : /// whose items are all objects (an Entity's own `type` list is not one).
186 5916 : fn instance_counts(doc: &Value) -> std::collections::HashMap<String, usize> {
187 5916 : doc.as_object()
188 5916 : .into_iter()
189 5916 : .flatten()
190 28998 : .filter_map(|(k, v)| {
191 28998 : let l = v.as_array().filter(|l| l.iter().all(Value::is_object))?;
192 11180 : Some((k.clone(), l.len()))
193 28998 : })
194 5916 : .collect()
195 5916 : }
196 :
197 : /// The memory/file store's history cap ([`Store::max_instances`]): the
198 : /// write that leaves an Attribute of `after` past `cap` instances, more than
199 : /// `was` held, is refused. One already past it that the write does not grow
200 : /// (a lowered cap) still takes corrections and deletions.
201 242840 : fn past_instance_cap(
202 242840 : was: &std::collections::HashMap<String, usize>,
203 242840 : after: &Value,
204 242840 : cap: usize,
205 242840 : ) -> Result<(), NgsiError> {
206 242840 : if cap == 0 {
207 238940 : return Ok(());
208 3900 : }
209 3900 : match instance_counts(after)
210 3900 : .into_iter()
211 9720 : .find(|(k, n)| *n > cap && *n > was.get(k).copied().unwrap_or(0))
212 : {
213 : // A cap the client can do something about, refused like the batch
214 : // cap (400 BadRequestData): the same request will not fit later
215 : // either, until older instances are deleted or reaped.
216 12 : Some((attr, n)) => Err(NgsiError::BadRequestData(format!(
217 12 : "the history of Attribute {attr} would hold {n} instances, more than the \
218 12 : {cap} instances this broker keeps per Attribute; delete older instances first"
219 12 : ))),
220 3888 : None => Ok(()),
221 : }
222 242840 : }
223 :
224 : /// What `antares-api` sees. No core crate names redb or sqlx.
225 : // One AnyStore exists per process — variant size difference is irrelevant.
226 : #[allow(clippy::large_enum_variant)]
227 : pub enum AnyStore {
228 : Mem(Store),
229 : #[cfg(feature = "postgres")]
230 : Pg(PgBackend),
231 : }
232 :
233 : /// A row's `id`, or `""` for one without: an id-less row sorts first and is
234 : /// never skipped past, so a keyset walk over it cannot lose it. Only the
235 : /// Postgres arm sorts rows it read whole; the memory arm's maps are already
236 : /// keyed by id.
237 : #[cfg(feature = "postgres")]
238 0 : fn row_id(v: &Value) -> &str {
239 0 : v.get("id").and_then(Value::as_str).unwrap_or_default()
240 0 : }
241 :
242 : /// The bound on one size scan of a tenant (`GET /q/tenants/{tenant}`):
243 : /// the history table runs to tens of GB, and a size is not worth a
244 : /// longer hold on a connection.
245 : #[cfg(feature = "postgres")]
246 : const SIZE_SCAN_TIMEOUT: &str = "SET LOCAL statement_timeout = '5s'";
247 :
248 : /// The attribute instances of one tenant, both history tables.
249 : #[cfg(feature = "postgres")]
250 8 : async fn count_instances(
251 8 : tx: &mut sqlx::Transaction<'static, sqlx::Postgres>,
252 8 : tenant: &TenantId,
253 8 : ) -> Result<u64, NgsiError> {
254 8 : let n: i64 = sqlx::query_scalar(
255 8 : "SELECT (SELECT count(*) FROM attr_instances WHERE tenant_id = $1)
256 8 : + (SELECT count(*) FROM attr_rows WHERE tenant_id = $1)",
257 8 : )
258 8 : .bind(tenant.as_str())
259 8 : .fetch_one(&mut **tx)
260 8 : .await
261 8 : .map_err(db)?;
262 6 : Ok(n.max(0) as u64)
263 8 : }
264 :
265 : impl AnyStore {
266 : /// Readiness ping: can the store answer a trivial request
267 : /// RIGHT NOW? Memory/file are in-process (always ready); the Pg arm runs
268 : /// `SELECT 1` so a lost database (failover, network partition) flips
269 : /// /q/ready to 503 and the Service stops routing to this pod.
270 16 : pub async fn ping(&self) -> Result<(), NgsiError> {
271 16 : match self {
272 16 : AnyStore::Mem(_) => Ok(()),
273 : #[cfg(feature = "postgres")]
274 0 : AnyStore::Pg(p) => async {
275 0 : sqlx::query("SELECT 1")
276 0 : .execute(p.docs.pool())
277 0 : .await
278 0 : .map(|_| ())
279 0 : }
280 0 : .await
281 0 : .map_err(db),
282 : }
283 16 : }
284 :
285 : /// What this store runs on, for `/q/health`. The memory and file modes
286 : /// are one backend with two durability shapes and must not read as the
287 : /// same thing; the Postgres arm serves what the startup probe read.
288 174 : pub fn version_info(&self) -> Value {
289 174 : match self {
290 158 : AnyStore::Mem(s) => serde_json::json!({
291 158 : "engine": if s.shadowed() { "redb" } else { "memory" },
292 : }),
293 : #[cfg(feature = "postgres")]
294 16 : AnyStore::Pg(p) => p.version.clone(),
295 : }
296 174 : }
297 :
298 : /// (Queued writers, peak) behind the single redb committer, reported
299 : /// only by a store that has one: `None` for the Pg arm (Postgres has no
300 : /// single-writer commit queue) and for a pure in-memory store, whose
301 : /// lock depth is not the fsync queue this number stands for.
302 90 : pub fn commit_queue(&self) -> Option<(usize, usize)> {
303 90 : match self {
304 84 : AnyStore::Mem(s) => s.shadowed().then(|| s.commit_queue()),
305 : #[cfg(feature = "postgres")]
306 6 : AnyStore::Pg(_) => None,
307 : }
308 90 : }
309 :
310 : /// The last step of the drain: close the connection pool so in-flight
311 : /// transactions finish and the server sees a clean disconnect instead of
312 : /// N abandoned backends. A no-op for the memory/file arm, whose durability
313 : /// is already commit-before-ack — there is nothing buffered to lose.
314 26 : pub async fn close(&self) {
315 26 : match self {
316 22 : AnyStore::Mem(_) => {}
317 : #[cfg(feature = "postgres")]
318 4 : AnyStore::Pg(p) => p.docs.pool().close().await,
319 : }
320 26 : }
321 :
322 1292 : pub fn set_change_hook(&self, h: ChangeHook) {
323 1292 : match self {
324 1258 : AnyStore::Mem(s) => s.set_change_hook(h),
325 : #[cfg(feature = "postgres")]
326 34 : AnyStore::Pg(p) => {
327 34 : *p.hook
328 34 : .write()
329 34 : .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(h)
330 : }
331 : }
332 1292 : }
333 :
334 : /// Turn the same-tx outbox producer on (bus=nats). The memory arm has
335 : /// no outbox — the broker's wiring rejects bus=nats without a Pg store,
336 : /// so this is unreachable there by construction.
337 2 : pub fn set_outbox(
338 2 : &self,
339 2 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] on: bool,
340 2 : ) {
341 2 : match self {
342 0 : AnyStore::Mem(_) => {}
343 : #[cfg(feature = "postgres")]
344 2 : AnyStore::Pg(p) => p.entities.set_outbox(on),
345 : }
346 2 : }
347 :
348 : /// Seal the notification credentials a release before
349 : /// `ANTARES_SECRET_KEY` stored in plaintext; the count of rows
350 : /// sealed. The memory arm seals nothing.
351 0 : pub async fn seal_plaintext_credentials(&self) -> Result<u64, NgsiError> {
352 0 : match self {
353 0 : AnyStore::Mem(_) => Ok(0),
354 : #[cfg(feature = "postgres")]
355 0 : AnyStore::Pg(p) => p.docs.seal_plaintext().await.map_err(db),
356 : }
357 0 : }
358 :
359 : /// Asynchronous commit for the batch writes (ANTARES_PG_BATCH_ASYNC_COMMIT). The memory arm has no commit to relax.
360 2 : pub fn set_async_batch_commit(
361 2 : &self,
362 2 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] on: bool,
363 2 : ) {
364 2 : match self {
365 0 : AnyStore::Mem(_) => {}
366 : #[cfg(feature = "postgres")]
367 2 : AnyStore::Pg(p) => p.entities.set_async_batch_commit(on),
368 : }
369 2 : }
370 :
371 : /// Outbox drain: oldest-first page of pending rows `(seq, tenant, event)`.
372 730 : pub async fn outbox_peek(
373 730 : &self,
374 730 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] limit: i64,
375 730 : ) -> Result<Vec<(i64, String, Value)>, NgsiError> {
376 730 : match self {
377 8 : AnyStore::Mem(_) => Ok(Vec::new()),
378 : #[cfg(feature = "postgres")]
379 722 : AnyStore::Pg(p) => super::pg::outbox::peek(p.docs.pool(), limit)
380 722 : .await
381 722 : .map_err(db),
382 : }
383 730 : }
384 :
385 : /// Outbox drain: delete EXACTLY the published rows (never a blanket
386 : /// `seq <= max`, which loses a row committing between peek and ack).
387 0 : pub async fn outbox_ack(
388 0 : &self,
389 0 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] seqs: &[i64],
390 0 : ) -> Result<u64, NgsiError> {
391 0 : match self {
392 0 : AnyStore::Mem(_) => Ok(0),
393 : #[cfg(feature = "postgres")]
394 0 : AnyStore::Pg(p) => super::pg::outbox::ack(p.docs.pool(), seqs)
395 0 : .await
396 0 : .map_err(db),
397 : }
398 0 : }
399 :
400 : /// Outbox drain: keep the rows of events the bus could not carry whole.
401 : /// The memory arm has no outbox, so nothing is ever retained there.
402 0 : pub async fn outbox_retain(
403 0 : &self,
404 0 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] tenant: &TenantId,
405 0 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] seqs: &[i64],
406 0 : ) -> Result<u64, NgsiError> {
407 0 : match self {
408 0 : AnyStore::Mem(_) => Ok(0),
409 : #[cfg(feature = "postgres")]
410 0 : AnyStore::Pg(p) => super::pg::outbox::retain(p.docs.pool(), tenant, seqs)
411 0 : .await
412 0 : .map_err(db),
413 : }
414 0 : }
415 :
416 : /// The whole event behind a claim-check reference, `None` once reaped.
417 2 : pub async fn outbox_event(
418 2 : &self,
419 2 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] seq: i64,
420 2 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] tenant: &TenantId,
421 2 : ) -> Result<Option<Value>, NgsiError> {
422 2 : match self {
423 0 : AnyStore::Mem(_) => Ok(None),
424 : #[cfg(feature = "postgres")]
425 2 : AnyStore::Pg(p) => super::pg::outbox::event(p.docs.pool(), seq, tenant)
426 2 : .await
427 2 : .map_err(db),
428 : }
429 2 : }
430 :
431 : /// Create one document of `kind`; `false` when that id is already taken.
432 162148 : pub async fn create(
433 162148 : &self,
434 162148 : tenant: &TenantId,
435 162148 : kind: Kind,
436 162148 : id: &str,
437 162148 : doc: Value,
438 162148 : ) -> Result<bool, NgsiError> {
439 162148 : match self {
440 136812 : AnyStore::Mem(s) => {
441 136812 : if kind == Kind::Temporal {
442 806 : past_instance_cap(&Default::default(), &doc, s.max_instances)?;
443 136006 : }
444 136810 : Ok(s.create(tenant, kind, id, doc).await)
445 : }
446 : #[cfg(feature = "postgres")]
447 25336 : AnyStore::Pg(p) => match kind {
448 : Kind::Entity => {
449 2244 : let created = p
450 2244 : .entities
451 2244 : .create_stamped(tenant, id, &doc)
452 2244 : .await
453 2244 : .map_err(db)?;
454 2244 : let hit = created.is_some();
455 2244 : if let Some(v) = created {
456 2234 : p.emit(tenant, None, Some(doc), v).await;
457 10 : }
458 2244 : Ok(hit)
459 : }
460 514 : Kind::Temporal => p.temporal.create(tenant, id, &doc).await.map_err(db),
461 : _ => {
462 22578 : let dk = doc_kind(kind)?;
463 22578 : p.docs.create(tenant, dk, id, &doc).await.map_err(db)
464 : }
465 : },
466 : }
467 162148 : }
468 :
469 : /// Batch create (entities only): one multi-row statement on the Pg
470 : /// arm, per-item loop on the memory arm. Created-flags in input order.
471 156 : pub async fn batch_create(
472 156 : &self,
473 156 : tenant: &TenantId,
474 156 : items: Vec<(String, Value)>,
475 156 : ) -> Result<Vec<bool>, NgsiError> {
476 156 : match self {
477 120 : AnyStore::Mem(s) => {
478 120 : let mut out = Vec::with_capacity(items.len());
479 5174 : for (id, doc) in items {
480 5174 : out.push(s.create(tenant, Kind::Entity, &id, doc).await);
481 : }
482 120 : Ok(out)
483 : }
484 : #[cfg(feature = "postgres")]
485 36 : AnyStore::Pg(p) => {
486 36 : let created = p
487 36 : .entities
488 36 : .batch_create_stamped(tenant, &items)
489 36 : .await
490 36 : .map_err(db)?;
491 36 : let flags = created.iter().map(Option::is_some).collect();
492 54 : for ((_, doc), created) in items.into_iter().zip(created) {
493 54 : if let Some(v) = created {
494 42 : p.emit(tenant, None, Some(doc), v).await;
495 12 : }
496 : }
497 36 : Ok(flags)
498 : }
499 : }
500 156 : }
501 :
502 : /// Batch delete (entities only): deleted-flags in input order; a
503 : /// duplicate id in the input deletes once and 404s the second time,
504 : /// matching the per-item loop's semantics (5.5.11.4).
505 390 : pub async fn batch_delete(
506 390 : &self,
507 390 : tenant: &TenantId,
508 390 : ids: &[String],
509 390 : ) -> Result<Vec<bool>, NgsiError> {
510 390 : match self {
511 224 : AnyStore::Mem(s) => {
512 224 : let mut out = Vec::with_capacity(ids.len());
513 7194 : for id in ids {
514 7194 : out.push(s.delete(tenant, Kind::Entity, id).await);
515 : }
516 224 : Ok(out)
517 : }
518 : #[cfg(feature = "postgres")]
519 166 : AnyStore::Pg(p) => {
520 166 : let deleted = p
521 166 : .entities
522 166 : .batch_delete_stamped(tenant, ids)
523 166 : .await
524 166 : .map_err(db)?;
525 166 : let mut prev: std::collections::HashMap<String, (Value, ChangeVersion)> = deleted
526 166 : .into_iter()
527 1460 : .map(|(id, before, v)| (id, (before, v)))
528 166 : .collect();
529 166 : let mut out = Vec::with_capacity(ids.len());
530 1484 : for id in ids {
531 1484 : match prev.remove(id) {
532 1460 : Some((before, v)) => {
533 1460 : p.emit(tenant, Some(before), None, v).await;
534 1460 : out.push(true);
535 : }
536 24 : None => out.push(false),
537 : }
538 : }
539 166 : Ok(out)
540 : }
541 : }
542 390 : }
543 :
544 20822 : pub async fn upsert(
545 20822 : &self,
546 20822 : tenant: &TenantId,
547 20822 : kind: Kind,
548 20822 : id: &str,
549 20822 : doc: Value,
550 20822 : ) -> Result<bool, NgsiError> {
551 20822 : match self {
552 18364 : AnyStore::Mem(s) => {
553 18364 : if kind == Kind::Temporal {
554 52 : past_instance_cap(&Default::default(), &doc, s.max_instances)?;
555 18312 : }
556 18364 : Ok(s.upsert(tenant, kind, id, doc).await)
557 : }
558 : #[cfg(feature = "postgres")]
559 2458 : AnyStore::Pg(p) => match kind {
560 : Kind::Entity => {
561 : // replace-or-create without a pre-read: try the replace
562 : // first (captures the true before-image under the row
563 : // lock), fall back to create, and on a lost create race
564 : // replace after all.
565 2434 : let mut prev: Option<Value> = None;
566 2434 : let mut version = ChangeVersion::default();
567 : // one shape, two places: a closure cannot hold the await
568 : // and an async closure is not a stable language feature,
569 : // so the replace is written once and expanded twice
570 : macro_rules! replace {
571 : () => {
572 : match p
573 : .entities
574 14 : .mutate_stamped(tenant, id, |d| {
575 14 : prev = Some(d.clone());
576 14 : *d = doc.clone();
577 14 : Ok::<(), std::convert::Infallible>(())
578 14 : })
579 : .await
580 : .map_err(db)?
581 : {
582 : Some(Ok(((), v))) => {
583 : version = v;
584 : true
585 : }
586 : Some(Err(never)) => match never {},
587 : None => false,
588 : }
589 : };
590 : }
591 2434 : let mut existed = replace!();
592 2434 : if !existed {
593 2420 : if let Some(v) = p
594 2420 : .entities
595 2420 : .create_stamped(tenant, id, &doc)
596 2420 : .await
597 2420 : .map_err(db)?
598 2420 : {
599 2420 : version = v;
600 2420 : existed = false;
601 2420 : } else {
602 : // lost the create race — replace instead
603 0 : existed = replace!();
604 : }
605 14 : }
606 2434 : p.emit(tenant, prev, Some(doc), version).await;
607 2434 : Ok(existed)
608 : }
609 : Kind::Temporal => {
610 4 : if p.temporal.create(tenant, id, &doc).await.map_err(db)? {
611 0 : Ok(false)
612 : } else {
613 4 : p.temporal
614 4 : .mutate(tenant, id, |d| {
615 4 : *d = doc.clone();
616 4 : Ok::<(), std::convert::Infallible>(())
617 4 : })
618 4 : .await
619 4 : .map_err(db)?;
620 4 : Ok(true)
621 : }
622 : }
623 : _ => {
624 20 : let dk = doc_kind(kind)?;
625 20 : p.docs.upsert(tenant, dk, id, &doc).await.map_err(db)
626 : }
627 : },
628 : }
629 20822 : }
630 :
631 32284 : pub async fn get(
632 32284 : &self,
633 32284 : tenant: &TenantId,
634 32284 : kind: Kind,
635 32284 : id: &str,
636 32284 : ) -> Result<Option<Value>, NgsiError> {
637 32284 : let doc = match self {
638 26293 : AnyStore::Mem(s) => s.get(tenant, kind, id),
639 : #[cfg(feature = "postgres")]
640 5991 : AnyStore::Pg(p) => match kind {
641 1274 : Kind::Entity => p.entities.get(tenant, id).await.map_err(db)?,
642 524 : Kind::Temporal => p.temporal.get(tenant, id).await.map_err(db)?,
643 4193 : _ => p.docs.get(tenant, doc_kind(kind)?, id).await.map_err(db)?,
644 : },
645 : };
646 : // 4.22: an expired entity is invalid context — a read serves it to
647 : // no one, whichever arm stored it (the Pg sweep lags by design).
648 32284 : if kind == Kind::Entity {
649 14846 : let now = now_utc();
650 14846 : if let Some(mut d) = doc {
651 12620 : if crate::store::filter::strip_expired(&mut d, &now) {
652 16 : return Ok(None);
653 12604 : }
654 12604 : return Ok(Some(d));
655 2226 : }
656 2226 : return Ok(None);
657 17438 : }
658 17438 : Ok(doc)
659 32284 : }
660 :
661 : /// [`Self::get`] of several Entities, in one transaction on Postgres;
662 : /// an absent or expired id has no entry.
663 72 : pub async fn get_entities(
664 72 : &self,
665 72 : tenant: &TenantId,
666 72 : ids: &[&str],
667 72 : ) -> Result<std::collections::HashMap<String, Value>, NgsiError> {
668 72 : let found: std::collections::HashMap<String, Value> = match self {
669 72 : AnyStore::Mem(s) => ids
670 72 : .iter()
671 270 : .filter_map(|id| Some(((*id).to_owned(), s.get(tenant, Kind::Entity, id)?)))
672 72 : .collect(),
673 : #[cfg(feature = "postgres")]
674 0 : AnyStore::Pg(p) => p.entities.get_many(tenant, ids).await.map_err(db)?,
675 : };
676 72 : let now = now_utc();
677 72 : Ok(found
678 72 : .into_iter()
679 72 : .filter_map(|(id, mut d)| {
680 56 : (!crate::store::filter::strip_expired(&mut d, &now)).then_some((id, d))
681 56 : })
682 72 : .collect())
683 72 : }
684 :
685 20116 : pub async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> Result<bool, NgsiError> {
686 20116 : match self {
687 12472 : AnyStore::Mem(s) => Ok(s.delete(tenant, kind, id).await),
688 : #[cfg(feature = "postgres")]
689 7644 : AnyStore::Pg(p) => match kind {
690 : Kind::Entity => {
691 : // the before-image comes from the DELETE's own RETURNING —
692 : // same transaction, never a separate racy read
693 1396 : let prev = p.entities.delete_stamped(tenant, id).await.map_err(db)?;
694 1396 : let hit = prev.is_some();
695 1396 : if let Some((before, v)) = prev {
696 1322 : p.emit(tenant, Some(before), None, v).await;
697 74 : }
698 1396 : Ok(hit)
699 : }
700 4664 : Kind::Temporal => p.temporal.delete(tenant, id).await.map_err(db),
701 1584 : _ => p.docs.delete(tenant, doc_kind(kind)?, id).await.map_err(db),
702 : },
703 : }
704 20116 : }
705 :
706 : /// Delete one entity only if `keep` accepts the stored document, read
707 : /// and removed under one lock (see the driver trait).
708 3126 : pub async fn delete_entity_if(
709 3126 : &self,
710 3126 : tenant: &TenantId,
711 3126 : id: &str,
712 3126 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
713 3126 : ) -> Result<bool, NgsiError> {
714 : // 4.22 is applied to the document `keep` judges, exactly as `get`
715 : // applies it: an expired instance is not there to be matched on.
716 3126 : let now = now_utc();
717 3126 : let judge = |d: &Value| {
718 2886 : let mut d = d.clone();
719 2886 : !crate::store::filter::strip_expired(&mut d, &now) && keep(&d)
720 2886 : };
721 3126 : match self {
722 1620 : AnyStore::Mem(s) => Ok(s.delete_if(tenant, id, &judge).await.is_some()),
723 : #[cfg(feature = "postgres")]
724 1506 : AnyStore::Pg(p) => {
725 : // the before-image comes from the DELETE's own RETURNING —
726 : // same transaction, never a separate racy read
727 1506 : let prev = p
728 1506 : .entities
729 1506 : .delete_if_stamped(tenant, id, &judge)
730 1506 : .await
731 1506 : .map_err(db)?;
732 1506 : let hit = prev.is_some();
733 1506 : if let Some((before, v)) = prev {
734 1406 : p.emit(tenant, Some(before), None, v).await;
735 100 : }
736 1506 : Ok(hit)
737 : }
738 : }
739 3126 : }
740 :
741 5212 : pub async fn list(&self, tenant: &TenantId, kind: Kind) -> Result<Vec<Value>, NgsiError> {
742 5212 : let mut rows = match self {
743 5078 : AnyStore::Mem(s) => s.list(tenant, kind),
744 : #[cfg(feature = "postgres")]
745 134 : AnyStore::Pg(p) => match kind {
746 32 : Kind::Entity => p.entities.list(tenant).await.map_err(db)?,
747 12 : Kind::Temporal => p.temporal.list(tenant).await.map_err(db)?,
748 90 : _ => p.docs.list(tenant, doc_kind(kind)?).await.map_err(db)?,
749 : },
750 : };
751 5212 : if kind == Kind::Entity {
752 111 : let now = now_utc();
753 8223 : rows.retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
754 5101 : }
755 5212 : Ok(rows)
756 5212 : }
757 :
758 : /// Every document's id and digest (see `CurrentStateDriver::list_digests`):
759 : /// a Postgres doc kind digests in the database, so no body leaves it.
760 0 : pub async fn list_digests(
761 0 : &self,
762 0 : tenant: &TenantId,
763 0 : kind: Kind,
764 0 : ) -> Result<Vec<(String, u64)>, NgsiError> {
765 : #[cfg(feature = "postgres")]
766 0 : if let AnyStore::Pg(p) = self {
767 0 : if !matches!(kind, Kind::Entity | Kind::Temporal) {
768 0 : return p
769 0 : .docs
770 0 : .list_digests(tenant, doc_kind(kind)?)
771 0 : .await
772 0 : .map_err(db);
773 0 : }
774 0 : }
775 0 : Ok(antares_store::digests(self.list(tenant, kind).await?))
776 0 : }
777 :
778 : /// One id-ordered page of documents (see `CurrentStateDriver::list_page`).
779 617492 : pub async fn list_page(
780 617492 : &self,
781 617492 : tenant: &TenantId,
782 617492 : kind: Kind,
783 617492 : after: Option<&str>,
784 617492 : limit: usize,
785 617492 : ) -> Result<Vec<Value>, NgsiError> {
786 617492 : match self {
787 40213 : AnyStore::Mem(s) => Ok(s.list_page(tenant, kind, after, limit)),
788 : #[cfg(feature = "postgres")]
789 577279 : AnyStore::Pg(p) => match kind {
790 : // Not doc kinds: they live in their own tables, behind their
791 : // own readers, and `doc_kind` refuses them.
792 : Kind::Entity => {
793 60 : let mut rows = p
794 60 : .entities
795 60 : .list_page(tenant, after, i64::try_from(limit).unwrap_or(i64::MAX))
796 60 : .await
797 60 : .map_err(db)?;
798 : // 4.22: the statement excluded expired ENTITIES, so no
799 : // document leaves this page and its length still means
800 : // what the caller reads it as; expired INSTANCES are
801 : // stripped here, at the read boundary, as everywhere else.
802 60 : let now = now_utc();
803 2026 : rows.retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
804 60 : Ok(rows)
805 : }
806 : // ponytail: temporal is sliced from the whole list, so a
807 : // tenant large enough still refuses this read for volume; the
808 : // upgrade is a keyset reader over the instance rows.
809 : // Temporal has no keyset reader of its own: its documents are
810 : // reconstructed from the instance rows, where the volume that
811 : // needs bounding is instances and not entities — the one kind
812 : // that does not yet honour `list_page`'s contract, and the
813 : // reason nothing internal walks temporal state.
814 : Kind::Temporal => {
815 0 : let mut rows = AnyStore::list(self, tenant, kind).await?;
816 0 : rows.sort_by(|a, b| row_id(a).cmp(row_id(b)));
817 0 : Ok(rows
818 0 : .into_iter()
819 0 : .skip_while(|r| after.is_some_and(|a| row_id(r) <= a))
820 0 : .take(limit)
821 0 : .collect())
822 : }
823 577219 : _ => p
824 577219 : .docs
825 577219 : .list_page(
826 577219 : tenant,
827 577219 : doc_kind(kind)?,
828 577219 : after,
829 577219 : i64::try_from(limit).unwrap_or(i64::MAX),
830 : )
831 577219 : .await
832 577219 : .map_err(db),
833 : },
834 : }
835 617492 : }
836 :
837 : /// One id-ordered window of documents and the total (see
838 : /// `CurrentStateDriver::list_slice`).
839 288 : pub async fn list_slice(
840 288 : &self,
841 288 : tenant: &TenantId,
842 288 : kind: Kind,
843 288 : offset: usize,
844 288 : limit: usize,
845 288 : ) -> Result<(Vec<Value>, usize), NgsiError> {
846 288 : match self {
847 194 : AnyStore::Mem(s) => Ok(s.list_slice(tenant, kind, offset, limit)),
848 : #[cfg(feature = "postgres")]
849 94 : AnyStore::Pg(p) => match kind {
850 : // Not doc kinds: each reads its own window in SQL, so the
851 : // whole-list ceiling never refuses a page.
852 : Kind::Entity => {
853 10 : let (mut rows, total) = p
854 10 : .entities
855 10 : .list_slice(
856 10 : tenant,
857 10 : i64::try_from(offset).unwrap_or(i64::MAX),
858 10 : i64::try_from(limit).unwrap_or(i64::MAX),
859 10 : )
860 10 : .await
861 10 : .map_err(db)?;
862 : // 4.22 at the read boundary, as `list_page` does
863 10 : let now = now_utc();
864 18 : rows.retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
865 10 : Ok((rows, usize::try_from(total).unwrap_or(usize::MAX)))
866 : }
867 : // Temporal is sliced from the whole list, as in
868 : // `list_page`: its documents are rebuilt from instance rows
869 : // and the paged temporal query skips shells without history;
870 : // the upgrade is a keyset reader over temporal_entities.
871 : Kind::Temporal => {
872 0 : let mut rows = AnyStore::list(self, tenant, kind).await?;
873 0 : rows.sort_by(|a, b| row_id(a).cmp(row_id(b)));
874 0 : let total = rows.len();
875 0 : Ok((rows.into_iter().skip(offset).take(limit).collect(), total))
876 : }
877 : _ => {
878 84 : let (page, total) = p
879 84 : .docs
880 84 : .list_slice(
881 84 : tenant,
882 84 : doc_kind(kind)?,
883 84 : i64::try_from(offset).unwrap_or(i64::MAX),
884 84 : i64::try_from(limit).unwrap_or(i64::MAX),
885 : )
886 84 : .await
887 84 : .map_err(db)?;
888 84 : Ok((page, usize::try_from(total).unwrap_or(usize::MAX)))
889 : }
890 : },
891 : }
892 288 : }
893 :
894 : /// 5.12 registration candidates for these entity ids / types. The Pg arm
895 : /// reads the `csource_index` rows (an indexed narrowing); `memory`/`file`
896 : /// have nothing to push into and return the same snapshot `list` does.
897 : /// Either way the result is a SUPERSET — the caller's matcher decides
898 : /// every 5.12 condition, this only avoids reading registrations that
899 : /// cannot match on id or type. `types` must be expanded plain type IRIs
900 : /// (what the registration write stored); a caller holding terms or a 4.17
901 : /// selection expression passes `None` and narrows on ids alone.
902 4855 : pub async fn matching_registrations(
903 4855 : &self,
904 4855 : tenant: &TenantId,
905 4855 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] ids: Option<&[String]>,
906 4855 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))] types: Option<&[String]>,
907 4855 : ) -> Result<Vec<Value>, NgsiError> {
908 4855 : match self {
909 4355 : AnyStore::Mem(s) => Ok(s.list(tenant, Kind::Registration)),
910 : #[cfg(feature = "postgres")]
911 500 : AnyStore::Pg(p) => p
912 500 : .docs
913 500 : .matching_registrations(tenant, ids, types)
914 500 : .await
915 500 : .map_err(db),
916 : }
917 4855 : }
918 :
919 : /// Query Entities with the filter pushed down where the backend
920 : /// can take it: SQL for Postgres, the borrowed documents for
921 : /// `memory`/`file` (`Store::query_entities`). Either way the caller
922 : /// applies the exact filter afterwards unless the outcome says the store
923 : /// already did.
924 3829 : pub async fn query_entities(
925 3829 : &self,
926 3829 : tenant: &TenantId,
927 3829 : f: &crate::store::filter::EntityFilter<'_>,
928 3829 : ) -> Result<crate::store::filter::QueryOutcome, NgsiError> {
929 3829 : self.query_paged(tenant, f, false).await
930 3829 : }
931 :
932 : /// `Store::query_candidates`: the Pg arm pages the candidates it cannot
933 : /// decide; the memory arm answers as `query_entities`.
934 372 : pub async fn query_candidates(
935 372 : &self,
936 372 : tenant: &TenantId,
937 372 : f: &crate::store::filter::EntityFilter<'_>,
938 372 : ) -> Result<crate::store::filter::QueryOutcome, NgsiError> {
939 372 : self.query_paged(tenant, f, true).await
940 372 : }
941 :
942 4201 : async fn query_paged(
943 4201 : &self,
944 4201 : tenant: &TenantId,
945 4201 : f: &crate::store::filter::EntityFilter<'_>,
946 4201 : candidates: bool,
947 4201 : ) -> Result<crate::store::filter::QueryOutcome, NgsiError> {
948 4201 : let mut outcome = match self {
949 2989 : AnyStore::Mem(s) => {
950 2989 : let _ = candidates;
951 2989 : s.query_entities(tenant, f)
952 : }
953 : #[cfg(feature = "postgres")]
954 1212 : AnyStore::Pg(p) => match candidates {
955 32 : true => p.entities.query_candidates(tenant, f).await,
956 1180 : false => p.entities.query(tenant, f).await,
957 : }
958 1212 : .map_err(db)?,
959 : };
960 : // 4.22: the Pg arm already excludes expired ENTITIES in SQL (so
961 : // paging/totals stay exact); instance stripping — and the whole job
962 : // on the memory arm — happens here at the read boundary.
963 4201 : let now = now_utc();
964 4201 : outcome
965 4201 : .rows
966 39842 : .retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
967 4201 : Ok(outcome)
968 4201 : }
969 :
970 : /// The Entity types of the Tenant's live Entities, sorted, read without
971 : /// the documents: the `types` column on Postgres, the held
972 : /// documents' `type` on the memory arm.
973 182 : pub async fn entity_type_names(&self, tenant: &TenantId) -> Result<Vec<String>, NgsiError> {
974 182 : match self {
975 138 : AnyStore::Mem(s) => Ok(s.entity_type_names(tenant, &now_utc())),
976 : #[cfg(feature = "postgres")]
977 44 : AnyStore::Pg(p) => p.entities.type_names(tenant).await.map_err(db),
978 : }
979 182 : }
980 :
981 : /// The (type, member name) pairs of the Tenant's attribute catalogue,
982 : /// sorted: table `entity_attrs` on Postgres, which may still
983 : /// list a pair no live Entity carries; the live documents in memory.
984 152 : pub async fn entity_attr_pairs(
985 152 : &self,
986 152 : tenant: &TenantId,
987 152 : ) -> Result<Vec<(String, String)>, NgsiError> {
988 152 : match self {
989 96 : AnyStore::Mem(s) => Ok(s.entity_attr_pairs(tenant, &now_utc())),
990 : #[cfg(feature = "postgres")]
991 56 : AnyStore::Pg(p) => p.entities.attr_pairs(tenant).await.map_err(db),
992 : }
993 152 : }
994 :
995 : /// The types of the Tenant's live Entities that carry one of `any_of`,
996 : /// sorted: the types index on Postgres, one pass in memory.
997 88 : pub async fn entity_types_beside(
998 88 : &self,
999 88 : tenant: &TenantId,
1000 88 : any_of: &[String],
1001 88 : ) -> Result<Vec<String>, NgsiError> {
1002 88 : match self {
1003 16 : AnyStore::Mem(s) => Ok(s.entity_types_beside(tenant, any_of, &now_utc())),
1004 : #[cfg(feature = "postgres")]
1005 72 : AnyStore::Pg(p) => p.entities.types_beside(tenant, any_of).await.map_err(db),
1006 : }
1007 88 : }
1008 :
1009 : /// The id segments the Pg arm reads by its C-collated id index;
1010 : /// the memory arm keeps none and its probes narrow by literal.
1011 16 : pub async fn entity_id_segments(
1012 16 : &self,
1013 16 : tenant: &TenantId,
1014 16 : lead: &str,
1015 16 : sep: char,
1016 16 : max: usize,
1017 16 : ) -> Result<Option<Vec<String>>, NgsiError> {
1018 16 : match self {
1019 : AnyStore::Mem(_) => {
1020 0 : let _ = (tenant, lead, sep, max);
1021 0 : Ok(None)
1022 : }
1023 : #[cfg(feature = "postgres")]
1024 16 : AnyStore::Pg(p) => p
1025 16 : .entities
1026 16 : .id_segments(tenant, lead, sep, max)
1027 16 : .await
1028 16 : .map_err(db),
1029 : }
1030 16 : }
1031 :
1032 : /// Query Temporal Evolution with entity narrowing, instance-window
1033 : /// pruning AND entity paging pushed down. Same
1034 : /// contract: the API's window() is the arbiter; the memory arm returns
1035 : /// the full snapshot, never paged.
1036 1092 : pub async fn query_temporal(
1037 1092 : &self,
1038 1092 : tenant: &TenantId,
1039 1092 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))]
1040 1092 : f: &crate::store::filter::TemporalFilter<'_>,
1041 1092 : ) -> Result<crate::store::filter::TemporalOutcome, NgsiError> {
1042 1092 : let mut outcome = match self {
1043 690 : AnyStore::Mem(s) => crate::store::filter::TemporalOutcome {
1044 690 : rows: s.list(tenant, Kind::Temporal),
1045 690 : paged: false,
1046 690 : total: None,
1047 690 : aggregated: false,
1048 690 : },
1049 : #[cfg(feature = "postgres")]
1050 402 : AnyStore::Pg(p) => p.temporal.query(tenant, f).await.map_err(db)?,
1051 : };
1052 : // 4.22 on temporal reads: the Pg arm already dropped expired ENTITIES in
1053 : // SQL (paging exact); this strips expired attribute INSTANCES, and does
1054 : // the whole job on the memory arm (no SQL to push into).
1055 1092 : let now = now_utc();
1056 1092 : outcome
1057 1092 : .rows
1058 2440 : .retain_mut(|d| !crate::store::filter::strip_expired(d, &now));
1059 1092 : Ok(outcome)
1060 1092 : }
1061 :
1062 : /// Auto-recording fast path: append instances to an
1063 : /// entity's temporal evolution, creating the meta shell on first touch.
1064 : /// Pg: pure multi-row INSERT — no history read, no doc rewrite. Memory:
1065 : /// the create-or-extend the mirror always did, under the store lock.
1066 : /// `shell` carries the meta members; `additions` maps attr IRI →
1067 : /// instance array (instanceIds already stamped by the caller).
1068 : ///
1069 : /// Both arms record only for an entity that still exists: 5.6.6 deletes
1070 : /// the entity and then the temporal evolution recorded for it, so an
1071 : /// append overlapping the delete must not recreate history nothing will
1072 : /// ever clean again. An instance marked `temporal_only` never holds the
1073 : /// entities (they live in another backend) and records unconditionally;
1074 : /// there the delete overlap stays a window the temporal delete closes.
1075 121920 : pub async fn temporal_append(
1076 121920 : &self,
1077 121920 : tenant: &TenantId,
1078 121920 : id: &str,
1079 121920 : shell: &Value,
1080 121920 : additions: &Value,
1081 121920 : ) -> Result<(), NgsiError> {
1082 121920 : match self {
1083 121874 : AnyStore::Mem(s) => {
1084 121874 : if !s.temporal_only && s.get(tenant, Kind::Entity, id).is_none() {
1085 6 : return Ok(());
1086 121868 : }
1087 121868 : if s.get(tenant, Kind::Temporal, id).is_none() {
1088 : // the cap the extension below applies, judged before the
1089 : // shell is stored: refused there, the empty shell stayed
1090 119844 : past_instance_cap(&Default::default(), additions, s.max_instances)?;
1091 : // loser of a concurrent create race just extends below
1092 119842 : let _ = s.create(tenant, Kind::Temporal, id, shell.clone()).await;
1093 2024 : }
1094 121866 : let cap = s.max_instances;
1095 121866 : let mut refused = None;
1096 121866 : s.mutate(tenant, Kind::Temporal, id, |doc| {
1097 121866 : let was = if cap > 0 {
1098 1742 : instance_counts(doc)
1099 : } else {
1100 120124 : Default::default()
1101 : };
1102 121866 : let target = doc.as_object_mut().ok_or(())?;
1103 : // 4.8: the Entity's modifiedAt is its last write, an
1104 : // Attribute's update included; never moved backwards
1105 121866 : if let Some(m) = shell.get("modifiedAt").and_then(Value::as_str) {
1106 9720 : let stale = target
1107 9720 : .get("modifiedAt")
1108 9720 : .and_then(Value::as_str)
1109 9720 : .is_none_or(|cur| super::mem::earlier(cur, m));
1110 9720 : if stale {
1111 1761 : target.insert("modifiedAt".into(), Value::String(m.to_owned()));
1112 7959 : }
1113 112146 : }
1114 121866 : if let Some(adds) = additions.as_object() {
1115 126632 : for (k, v) in adds {
1116 126632 : let mut incoming: Vec<Value> =
1117 126632 : v.as_array().cloned().unwrap_or_default();
1118 : // The memory store keeps no provenance and no
1119 : // versions; Postgres does.
1120 127002 : incoming.iter_mut().for_each(|i| {
1121 127002 : antares_store::take_provenance(i);
1122 127002 : });
1123 126632 : match target.get_mut(k).and_then(Value::as_array_mut) {
1124 : // same instanceId = the same instance corrected
1125 : // (the pg arm's ON CONFLICT DO UPDATE)
1126 1954 : Some(cur) => {
1127 1962 : for inst in incoming {
1128 1962 : let iid = inst.get("instanceId");
1129 143430 : match cur.iter_mut().find(|c| c.get("instanceId") == iid) {
1130 36 : Some(slot) => *slot = inst,
1131 1926 : None => cur.push(inst),
1132 : }
1133 : }
1134 : }
1135 124678 : None => {
1136 124678 : target.insert(k.clone(), Value::Array(incoming));
1137 124678 : }
1138 : }
1139 : }
1140 0 : }
1141 121866 : past_instance_cap(&was, doc, cap).map_err(|e| {
1142 4 : refused = Some(e);
1143 4 : })
1144 121866 : })
1145 121866 : .await;
1146 121866 : refused.map_or(Ok(()), Err)
1147 : }
1148 : #[cfg(feature = "postgres")]
1149 46 : AnyStore::Pg(p) => p
1150 46 : .temporal
1151 46 : .append(tenant, id, shell, additions)
1152 46 : .await
1153 46 : .map_err(db),
1154 : }
1155 121920 : }
1156 :
1157 : /// Mark this instance as the temporal half of a mixed deployment: the
1158 : /// entities live in another backend, so appends never look for them here.
1159 10 : pub fn temporal_only(mut self) -> Self {
1160 10 : match &mut self {
1161 10 : AnyStore::Mem(s) => s.temporal_only = true,
1162 : #[cfg(feature = "postgres")]
1163 0 : AnyStore::Pg(p) => p.temporal.temporal_only = true,
1164 : }
1165 10 : self
1166 10 : }
1167 :
1168 : /// Cap the instances one Attribute's history may hold in the memory/file
1169 : /// store (`0` = none); the Postgres arm keeps history in rows and has no
1170 : /// such document to bound.
1171 : #[cfg_attr(not(feature = "postgres"), allow(irrefutable_let_patterns))]
1172 31 : pub fn max_instances(mut self, cap: usize) -> Self {
1173 31 : if let AnyStore::Mem(s) = &mut self {
1174 31 : s.max_instances = cap;
1175 31 : }
1176 31 : self
1177 31 : }
1178 :
1179 : /// Retrieve Temporal Evolution with the same instance pruning.
1180 1036 : pub async fn get_temporal(
1181 1036 : &self,
1182 1036 : tenant: &TenantId,
1183 1036 : id: &str,
1184 1036 : #[cfg_attr(not(feature = "postgres"), allow(unused_variables))]
1185 1036 : f: &crate::store::filter::TemporalFilter<'_>,
1186 1036 : ) -> Result<Option<Value>, NgsiError> {
1187 1036 : let mut doc = match self {
1188 768 : AnyStore::Mem(s) => s.get(tenant, Kind::Temporal, id),
1189 : #[cfg(feature = "postgres")]
1190 268 : AnyStore::Pg(p) => p.temporal.get_range(tenant, id, f).await.map_err(db)?,
1191 : };
1192 : // 4.22: expired entity → None (Pg already did this in SQL); otherwise
1193 : // strip expired instances.
1194 1036 : if let Some(d) = &mut doc {
1195 972 : let now = now_utc();
1196 972 : if crate::store::filter::strip_expired(d, &now) {
1197 0 : return Ok(None);
1198 972 : }
1199 64 : }
1200 1036 : Ok(doc)
1201 1036 : }
1202 :
1203 25826 : pub async fn mutate<T, E>(
1204 25826 : &self,
1205 25826 : tenant: &TenantId,
1206 25826 : kind: Kind,
1207 25826 : id: &str,
1208 25826 : f: impl FnOnce(&mut Value) -> Result<T, E>,
1209 25826 : ) -> Result<Option<Result<T, E>>, NgsiError> {
1210 24181 : match self {
1211 24181 : AnyStore::Mem(s) if kind == Kind::Temporal && s.max_instances > 0 => {
1212 : // The cap's refusal is no error of the closure: it aborts the
1213 : // write the same way and comes back as the store's own.
1214 : enum Refused<E> {
1215 : Closure(E),
1216 : Cap(NgsiError),
1217 : }
1218 298 : let cap = s.max_instances;
1219 298 : let r = s
1220 298 : .mutate(tenant, kind, id, |doc| {
1221 274 : let was = instance_counts(doc);
1222 274 : let t = f(doc).map_err(Refused::Closure)?;
1223 272 : past_instance_cap(&was, doc, cap).map_err(Refused::Cap)?;
1224 268 : Ok(t)
1225 274 : })
1226 298 : .await;
1227 6 : match r {
1228 24 : None => Ok(None),
1229 268 : Some(Ok(t)) => Ok(Some(Ok(t))),
1230 2 : Some(Err(Refused::Closure(e))) => Ok(Some(Err(e))),
1231 4 : Some(Err(Refused::Cap(e))) => Err(e),
1232 : }
1233 : }
1234 23883 : AnyStore::Mem(s) => Ok(s.mutate(tenant, kind, id, f).await),
1235 : #[cfg(feature = "postgres")]
1236 1645 : AnyStore::Pg(p) => match kind {
1237 : Kind::Entity => {
1238 : // before/after captured for the change hook's
1239 : // prev_payload INSIDE the row lock — a before-image read
1240 : // in its own transaction can belong to a different version
1241 : // than the one the lock serialized on.
1242 454 : let mut before: Option<Value> = None;
1243 454 : let mut after: Option<Value> = None;
1244 454 : let r = p
1245 454 : .entities
1246 454 : .mutate_stamped(tenant, id, |d| {
1247 436 : before = Some(d.clone());
1248 436 : let r = f(d);
1249 436 : if r.is_ok() {
1250 418 : after = Some(d.clone());
1251 418 : }
1252 436 : r
1253 436 : })
1254 454 : .await
1255 454 : .map_err(db)?;
1256 454 : let (r, version) = match r {
1257 418 : Some(Ok((t, v))) => (Some(Ok(t)), Some(v)),
1258 18 : Some(Err(e)) => (Some(Err(e)), None),
1259 18 : None => (None, None),
1260 : };
1261 454 : if let (Some(v), Some(a)) = (version, after) {
1262 418 : if before.as_ref() != Some(&a) {
1263 402 : p.emit(tenant, before, Some(a), v).await;
1264 16 : }
1265 36 : }
1266 454 : Ok(r)
1267 : }
1268 302 : Kind::Temporal => p.temporal.mutate(tenant, id, f).await.map_err(db),
1269 : _ => {
1270 : // FOR UPDATE + UPDATE in one tx: a bookkeeping writeback
1271 : // racing a DELETE must never resurrect the row (047_06).
1272 889 : let dk = doc_kind(kind)?;
1273 889 : match p.docs.mutate(tenant, dk, id, f).await.map_err(db)? {
1274 519 : Some(r) => Ok(Some(r)),
1275 370 : None => Ok(None),
1276 : }
1277 : }
1278 : },
1279 : }
1280 25826 : }
1281 :
1282 : /// Batch upsert with REPLACE semantics for entities:
1283 : /// one statement + one transaction on the Pg arm, per-item loop on the
1284 : /// memory arm. Created-flags in input order.
1285 180 : pub async fn batch_upsert(
1286 180 : &self,
1287 180 : tenant: &TenantId,
1288 180 : items: Vec<(String, Value)>,
1289 180 : ) -> Result<Vec<bool>, NgsiError> {
1290 : // Each id once per call, on every arm: Postgres refuses an ON CONFLICT
1291 : // row affected twice and the memory arm would keep the last, so a
1292 : // repeat is the caller's bug (the API splits repeats into rounds).
1293 180 : let mut seen = std::collections::HashSet::with_capacity(items.len());
1294 280 : if let Some((id, _)) = items.iter().find(|(id, _)| !seen.insert(id.as_str())) {
1295 6 : return Err(NgsiError::InternalError(format!(
1296 6 : "batch_upsert names {id} twice"
1297 6 : )));
1298 174 : }
1299 174 : match self {
1300 116 : AnyStore::Mem(s) => {
1301 116 : let mut out = Vec::with_capacity(items.len());
1302 170 : for (id, doc) in items {
1303 170 : out.push(!s.upsert(tenant, Kind::Entity, &id, doc).await);
1304 : }
1305 116 : Ok(out)
1306 : }
1307 : #[cfg(feature = "postgres")]
1308 58 : AnyStore::Pg(p) => {
1309 58 : let out = p
1310 58 : .entities
1311 58 : .batch_upsert_replace_stamped(tenant, &items)
1312 58 : .await
1313 58 : .map_err(db)?;
1314 98 : for ((_, doc), (_, prev, v)) in items.iter().zip(&out) {
1315 98 : p.emit(
1316 98 : tenant,
1317 98 : prev.clone(),
1318 98 : Some(doc.clone()),
1319 98 : v.clone().unwrap_or_default(),
1320 98 : )
1321 98 : .await;
1322 : }
1323 58 : Ok(out.into_iter().map(|(created, _, _)| created).collect())
1324 : }
1325 : }
1326 180 : }
1327 :
1328 : /// Batch read-modify-write for entities: one transaction + one ordered
1329 : /// lock set + one multi-row writeback on the Pg arm; per-item mutate on
1330 : /// the memory arm. Results align with `ids` (`None` = absent).
1331 236 : pub async fn batch_mutate<E>(
1332 236 : &self,
1333 236 : tenant: &TenantId,
1334 236 : ids: &[String],
1335 236 : mut f: impl FnMut(&str, &mut Value) -> Result<(), E>,
1336 236 : ) -> Result<Vec<Option<Result<(), E>>>, NgsiError> {
1337 236 : match self {
1338 170 : AnyStore::Mem(s) => {
1339 170 : let mut out = Vec::with_capacity(ids.len());
1340 8258 : for id in ids {
1341 8258 : out.push(s.mutate(tenant, Kind::Entity, id, |d| f(id, d)).await);
1342 : }
1343 170 : Ok(out)
1344 : }
1345 : #[cfg(feature = "postgres")]
1346 66 : AnyStore::Pg(p) => {
1347 : // hook images captured inside the lock, same as single mutate
1348 66 : let mut images: Vec<(String, Value, Value)> = Vec::new();
1349 66 : let (r, versions) = p
1350 66 : .entities
1351 112 : .batch_mutate_stamped(tenant, ids, |id, d| {
1352 112 : let before = d.clone();
1353 112 : let r = f(id, d);
1354 112 : if r.is_ok() && *d != before {
1355 110 : images.push((id.to_owned(), before, d.clone()));
1356 110 : }
1357 112 : r
1358 112 : })
1359 66 : .await
1360 66 : .map_err(db)?;
1361 110 : for (id, before, after) in images {
1362 110 : let v = versions.get(&id).cloned().unwrap_or_default();
1363 110 : p.emit(tenant, Some(before), Some(after), v).await;
1364 : }
1365 66 : Ok(r)
1366 : }
1367 : }
1368 236 : }
1369 :
1370 : /// 4.22 GC for the memory/file arm (the Pg arm's sweep lives in the
1371 : /// maintenance job, mode-switched in the broker). Returns reaped count.
1372 1958 : pub fn sweep_expired(&self) -> usize {
1373 1958 : match self {
1374 962 : AnyStore::Mem(s) => s.sweep_expired(&now_utc()),
1375 : // 4.22 reaping on the Pg arm is `pg::maintenance`'s job
1376 : // (`reap_expired_entities` / `reap_expired_instances`), which
1377 : // deletes in one indexed statement instead of walking the store;
1378 : // reads refuse an expired document either way.
1379 : #[cfg(feature = "postgres")]
1380 996 : AnyStore::Pg(_) => 0,
1381 : }
1382 1958 : }
1383 :
1384 : /// 5.5.10: does the Tenant exist? The default Tenant "implicitly exists";
1385 : /// others exist once implicitly created by a create operation.
1386 : ///
1387 : /// It answers for CLIENT tenants. A tenant the broker minted for itself
1388 : /// does hold a row — every write claims one, and that row is what keeps
1389 : /// the tenant enumerable to `subscription_tenants` across a restart — but
1390 : /// nothing asks this about one: the wall that asks sits outside the 6.3.22
1391 : /// snapshot scoping and reads the header the caller sent, which cannot
1392 : /// name an internal tenant. What a Snapshot's synthetic tenant holds is
1393 : /// asserted by the Snapshot document; the row only accounts for it, and
1394 : /// `tenant_ids` keeps it out of the customer-account listing.
1395 348 : pub async fn tenant_exists(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
1396 348 : if tenant.as_str() == TenantId::DEFAULT {
1397 14 : return Ok(true);
1398 334 : }
1399 334 : match self {
1400 252 : AnyStore::Mem(s) => Ok(s.tenant_exists(tenant)),
1401 : #[cfg(feature = "postgres")]
1402 82 : AnyStore::Pg(p) => {
1403 82 : async {
1404 82 : let row =
1405 82 : sqlx::query_scalar::<_, i32>("SELECT 1 FROM tenants WHERE tenant_id = $1")
1406 82 : .bind(tenant.as_str())
1407 82 : .fetch_optional(p.docs.pool())
1408 82 : .await
1409 82 : .map_err(db)?;
1410 82 : Ok(row.is_some())
1411 82 : }
1412 82 : .await
1413 : }
1414 : }
1415 348 : }
1416 :
1417 : /// Every tenant name the backend knows, sorted. One cheap query: at the
1418 : /// 10 000-tenant target (ADR-0001) an inventory that counted every kind
1419 : /// of every tenant would be 7 counts per tenant on one transaction, so
1420 : /// the counts live in `tenant_stats_one` and are paid per lookup.
1421 2004 : pub async fn tenant_ids(&self) -> Result<Vec<String>, NgsiError> {
1422 2004 : match self {
1423 1004 : AnyStore::Mem(s) => Ok(s.tenant_ids()),
1424 : #[cfg(feature = "postgres")]
1425 1000 : AnyStore::Pg(p) => {
1426 1000 : async {
1427 1000 : let mut rows: Vec<String> =
1428 1000 : sqlx::query_scalar("SELECT tenant_id FROM tenants ORDER BY 1")
1429 1000 : .fetch_all(p.docs.pool())
1430 1000 : .await
1431 1000 : .map_err(db)?;
1432 : // The broker's own tenants claim a row like any other
1433 : // write, because that row is also the enumeration the
1434 : // notification paths walk. The inventory is the list of
1435 : // customer accounts, so they are filtered on the way out.
1436 140284 : rows.retain(|t| !TenantId::is_reserved_str(t));
1437 : // 5.5.10: the default Tenant implicitly exists, whether or
1438 : // not a row was ever written for it.
1439 12992 : if !rows.iter().any(|t| t == TenantId::DEFAULT) {
1440 0 : rows.push(TenantId::DEFAULT.to_owned());
1441 0 : rows.sort();
1442 1000 : }
1443 1000 : Ok(rows)
1444 1000 : }
1445 1000 : .await
1446 : }
1447 : }
1448 2004 : }
1449 :
1450 : /// What one tenant holds; `None` when it does not exist (5.5.10 keeps
1451 : /// the default Tenant existing even when empty).
1452 44 : pub async fn tenant_stats_one(
1453 44 : &self,
1454 44 : tenant: &TenantId,
1455 44 : ) -> Result<Option<antares_store::TenantStats>, NgsiError> {
1456 44 : if !self.tenant_exists(tenant).await? {
1457 10 : return Ok(None);
1458 34 : }
1459 34 : match self {
1460 32 : AnyStore::Mem(s) => Ok(Some(s.tenant_stats_one(tenant))),
1461 : #[cfg(feature = "postgres")]
1462 2 : AnyStore::Pg(p) => {
1463 2 : async {
1464 2 : let mut tx = tenant_tx(p, tenant).await?;
1465 : // the tenant setting is what the RLS-guarded counts see
1466 2 : let created_at: Option<String> = sqlx::query_scalar(
1467 2 : "SELECT created_at::text FROM tenants WHERE tenant_id = $1",
1468 2 : )
1469 2 : .bind(tenant.as_str())
1470 2 : .fetch_optional(&mut *tx)
1471 2 : .await
1472 2 : .map_err(db)?;
1473 2 : let c: (i64, i64, i64, i64, i64, i64, i64) = sqlx::query_as(
1474 2 : "SELECT (SELECT count(*) FROM entities WHERE tenant_id = $1),
1475 2 : (SELECT count(*) FROM subscriptions WHERE tenant_id = $1),
1476 2 : (SELECT count(*) FROM csource_registrations WHERE tenant_id = $1),
1477 2 : (SELECT count(*) FROM csource_subscriptions WHERE tenant_id = $1),
1478 2 : (SELECT count(*) FROM snapshots WHERE tenant_id = $1),
1479 2 : (SELECT count(*) FROM entity_map_docs WHERE tenant_id = $1),
1480 2 : (SELECT count(*) FROM dist_subs WHERE tenant_id = $1)",
1481 2 : )
1482 2 : .bind(tenant.as_str())
1483 2 : .fetch_one(&mut *tx)
1484 2 : .await
1485 2 : .map_err(db)?;
1486 2 : tx.commit().await.map_err(db)?;
1487 2 : Ok(Some(antares_store::TenantStats {
1488 2 : tenant: tenant.as_str().to_owned(),
1489 2 : created_at,
1490 2 : entities: c.0 as u64,
1491 2 : subscriptions: c.1 as u64,
1492 2 : registrations: c.2 as u64,
1493 2 : csource_subscriptions: c.3 as u64,
1494 2 : snapshots: c.4 as u64,
1495 2 : entity_maps: c.5 as u64,
1496 2 : dist_subs: c.6 as u64,
1497 2 : }))
1498 2 : }
1499 2 : .await
1500 : }
1501 : }
1502 44 : }
1503 :
1504 : /// Purge the current-state half of one tenant in one transaction;
1505 : /// `false` when the tenant did not exist. The default tenant's row stays.
1506 116 : pub async fn purge_tenant(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
1507 116 : match self {
1508 86 : AnyStore::Mem(s) => Ok(s.purge_tenant(tenant)),
1509 : #[cfg(feature = "postgres")]
1510 30 : AnyStore::Pg(p) => {
1511 30 : async {
1512 30 : let mut tx = tenant_tx(p, tenant).await?;
1513 30 : let known = sqlx::query_scalar::<_, i32>(
1514 30 : "SELECT 1 FROM tenants WHERE tenant_id = $1 FOR UPDATE",
1515 30 : )
1516 30 : .bind(tenant.as_str())
1517 30 : .fetch_optional(&mut *tx)
1518 30 : .await
1519 30 : .map_err(db)?
1520 30 : .is_some();
1521 30 : if !known {
1522 2 : return Ok(false);
1523 28 : }
1524 420 : for table in [
1525 28 : "entities",
1526 28 : // emptied by the delete trigger of `entities` (0024)
1527 28 : "entity_values",
1528 28 : "subscriptions",
1529 28 : "csource_subscriptions",
1530 28 : "csource_registrations",
1531 28 : "csource_index",
1532 28 : "outbox",
1533 28 : "snapshots",
1534 28 : "entity_map_docs",
1535 28 : "dist_subs",
1536 28 : "dead_letters",
1537 28 : "views",
1538 28 : "entity_types",
1539 28 : "entity_attrs",
1540 28 : // ADR-0021: `tenant_id` is GENERATED from the row's kind
1541 28 : // and owner, so a Cached row's is NULL and this DELETE
1542 28 : // takes only what the Tenant stored itself.
1543 28 : "jsonld_contexts",
1544 28 : ] {
1545 420 : sqlx::query(sqlx::AssertSqlSafe(format!(
1546 420 : "DELETE FROM {table} WHERE tenant_id = $1"
1547 420 : )))
1548 420 : .bind(tenant.as_str())
1549 420 : .execute(&mut *tx)
1550 420 : .await
1551 420 : .map_err(db)?;
1552 : }
1553 28 : if tenant.as_str() != TenantId::DEFAULT {
1554 28 : sqlx::query("DELETE FROM tenants WHERE tenant_id = $1")
1555 28 : .bind(tenant.as_str())
1556 28 : .execute(&mut *tx)
1557 28 : .await
1558 28 : .map_err(db)?;
1559 0 : }
1560 28 : tx.commit().await.map_err(db)?;
1561 28 : Ok(true)
1562 30 : }
1563 30 : .await
1564 : }
1565 : }
1566 116 : }
1567 :
1568 3155 : pub async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError> {
1569 3155 : match self {
1570 2958 : AnyStore::Mem(s) => Ok(s.subscription_tenants()),
1571 : // Pg answers the superset the contract allows: every known
1572 : // tenant. Narrowing it would need a cross-tenant read of
1573 : // `subscriptions`, and that table is deliberately outside the
1574 : // `antares.service` escape — a subscription document carries
1575 : // endpoint.receiverInfo, i.e. the credentials the notification
1576 : // is sent with. The callers list per tenant under set_tenant
1577 : // afterwards, so the extra names cost one empty list each.
1578 : #[cfg(feature = "postgres")]
1579 197 : AnyStore::Pg(p) => {
1580 197 : async {
1581 197 : let rows =
1582 197 : sqlx::query_scalar::<_, String>("SELECT tenant_id FROM tenants ORDER BY 1")
1583 197 : .fetch_all(p.docs.pool())
1584 197 : .await
1585 197 : .map_err(db)?;
1586 197 : Ok(rows)
1587 197 : }
1588 197 : .await
1589 : }
1590 : }
1591 3155 : }
1592 :
1593 688 : pub async fn context_put(
1594 688 : &self,
1595 688 : tenant: Option<&TenantId>,
1596 688 : id: &str,
1597 688 : doc: Value,
1598 688 : ) -> Result<(), NgsiError> {
1599 : // ADR-0021, the write half no `WHERE` on an existing row can state:
1600 : // the row that ARRIVES must belong to the caller too, or a Tenant
1601 : // could store term mappings under another Tenant's name and decide
1602 : // what that Tenant's payloads mean (5.5.7). Postgres says the same
1603 : // thing as the policy's WITH CHECK; this holds under the roles that
1604 : // bypass RLS.
1605 688 : if !antares_store::context_row_visible(&doc, tenant) {
1606 12 : return Err(NgsiError::InternalError(
1607 12 : "@context belongs to another tenant".into(),
1608 12 : ));
1609 676 : }
1610 : // 5.13.1 names three kinds and the row's kind decides who reaches it
1611 : // (ADR-0021): a row without one would be Cached to Postgres and
1612 : // nothing at all to the memory arm.
1613 676 : if doc.get("kind").and_then(Value::as_str).is_none() {
1614 0 : return Err(NgsiError::InternalError(
1615 0 : "@context row without a kind".into(),
1616 0 : ));
1617 676 : }
1618 676 : match self {
1619 478 : AnyStore::Mem(s) => s.context_put(tenant, id, doc),
1620 : #[cfg(feature = "postgres")]
1621 198 : AnyStore::Pg(p) => {
1622 198 : let kind = doc
1623 198 : .get("kind")
1624 198 : .and_then(Value::as_str)
1625 198 : .unwrap_or("Cached")
1626 198 : .to_owned();
1627 198 : p.docs
1628 198 : .context_put(tenant, id, &doc, &kind)
1629 198 : .await
1630 198 : .map_err(db)
1631 : }
1632 : }
1633 688 : }
1634 :
1635 1350 : pub async fn context_get(
1636 1350 : &self,
1637 1350 : tenant: Option<&TenantId>,
1638 1350 : id: &str,
1639 1350 : ) -> Result<Option<Value>, NgsiError> {
1640 1350 : match self {
1641 902 : AnyStore::Mem(s) => Ok(s.context_get(tenant, id)),
1642 : #[cfg(feature = "postgres")]
1643 448 : AnyStore::Pg(p) => p.docs.context_get(tenant, id).await.map_err(db),
1644 : }
1645 1350 : }
1646 :
1647 21058 : pub async fn context_mutate(
1648 21058 : &self,
1649 21058 : tenant: Option<&TenantId>,
1650 21058 : id: &str,
1651 21058 : f: antares_store::MutateFn<'_>,
1652 21058 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1653 : // the row a closure leaves must still be one this caller may store,
1654 : // and keep its kind: the same two checks context_put makes
1655 21058 : let guarded = move |doc: &mut Value| {
1656 20716 : let kind = doc.get("kind").cloned();
1657 20716 : f(doc)?;
1658 20716 : if doc.get("kind") != kind.as_ref() || !antares_store::context_row_visible(doc, tenant)
1659 : {
1660 0 : return Err(());
1661 20716 : }
1662 20716 : Ok(())
1663 20716 : };
1664 21058 : match self {
1665 10730 : AnyStore::Mem(s) => Ok(s.context_mutate(tenant, id, guarded)),
1666 : #[cfg(feature = "postgres")]
1667 10328 : AnyStore::Pg(p) => p.docs.context_mutate(tenant, id, guarded).await.map_err(db),
1668 : }
1669 21058 : }
1670 :
1671 412 : pub async fn context_delete(
1672 412 : &self,
1673 412 : tenant: Option<&TenantId>,
1674 412 : id: &str,
1675 412 : ) -> Result<bool, NgsiError> {
1676 412 : match self {
1677 226 : AnyStore::Mem(s) => Ok(s.context_delete(tenant, id)),
1678 : #[cfg(feature = "postgres")]
1679 186 : AnyStore::Pg(p) => p.docs.context_delete(tenant, id).await.map_err(db),
1680 : }
1681 412 : }
1682 :
1683 282 : pub async fn context_list_meta(
1684 282 : &self,
1685 282 : tenant: Option<&TenantId>,
1686 282 : ) -> Result<Vec<Value>, NgsiError> {
1687 282 : match self {
1688 198 : AnyStore::Mem(s) => Ok(s.context_list_meta(tenant)),
1689 : #[cfg(feature = "postgres")]
1690 84 : AnyStore::Pg(p) => p.docs.context_list_meta(tenant).await.map_err(db),
1691 : }
1692 282 : }
1693 : }
1694 :
1695 : #[cfg(all(test, feature = "postgres"))]
1696 : mod db_error_tests {
1697 : #[allow(unused_imports)]
1698 : use antares_model::NgsiError;
1699 :
1700 : /// An acquire timeout is overload, not a fault, and the HTTP binding
1701 : /// recognises it by this exact detail. If the two ends ever spell it
1702 : /// differently the broker answers 500 to a condition a client could
1703 : /// have retried, so the constant is asserted rather than the words.
1704 : #[test]
1705 2 : fn a_pool_timeout_is_marked_as_overload() {
1706 2 : let pd = super::db(sqlx::Error::PoolTimedOut).to_problem_details();
1707 2 : assert_eq!(pd.detail, antares_model::error::DB_OVERLOADED);
1708 2 : assert_eq!(pd.title, "InternalError");
1709 : // and no other sqlx error borrows the mark
1710 2 : assert_ne!(
1711 2 : super::db(sqlx::Error::RowNotFound)
1712 2 : .to_problem_details()
1713 : .detail,
1714 : antares_model::error::DB_OVERLOADED
1715 : );
1716 2 : }
1717 :
1718 : /// 5.5.6 InternalError: the RFC 7807 `detail` a client sees must be
1719 : /// generic — driver internals (SQL text, row counts, connection
1720 : /// strings) belong in the server log, never in the response body.
1721 : #[test]
1722 2 : fn db_error_detail_is_generic() {
1723 2 : let pd = super::db(sqlx::Error::RowNotFound).to_problem_details();
1724 2 : assert_eq!(pd.detail, "database error");
1725 2 : assert_eq!(pd.title, "InternalError");
1726 2 : assert!(
1727 2 : !pd.detail.contains("no rows"),
1728 : "sqlx internals leaked into the client-visible detail: {}",
1729 : pd.detail
1730 : );
1731 : // a configuration error that is NOT one of ours stays generic too
1732 2 : let pd = super::db(sqlx::Error::Configuration("boom".into())).to_problem_details();
1733 2 : assert_eq!(pd.detail, "database error");
1734 2 : assert!(!pd.detail.contains("boom"), "{}", pd.detail);
1735 2 : }
1736 :
1737 : /// A spec error the store raised itself travels out through the driver
1738 : /// error channel. Rebuilding it as a fixed variant forces every one of
1739 : /// them to that variant's status — a 400 BadRequestData would reach the
1740 : /// client as a 403.
1741 : #[test]
1742 2 : fn a_store_raised_spec_error_keeps_its_own_status() {
1743 6 : for (err, kind, status) in [
1744 2 : (NgsiError::BadRequestData("x".into()), "BadRequestData", 400),
1745 2 : (NgsiError::TooManyResults("x".into()), "TooManyResults", 403),
1746 2 : (NgsiError::AlreadyExists("x".into()), "AlreadyExists", 409),
1747 2 : ] {
1748 6 : let out = super::db(sqlx::Error::Configuration(Box::new(err)));
1749 6 : assert_eq!(out.kind(), kind);
1750 6 : assert_eq!(out.status(), status, "{kind} lost its status");
1751 : }
1752 2 : }
1753 : }
1754 :
1755 : // The driver seam: `AnyStore` carries both driver interfaces, delegating to
1756 : // the inherent methods above. New backends implement the traits directly —
1757 : // this enum stays an implementation detail of the built-in backends, no
1758 : // longer the API surface.
1759 : /// The most Entities one history transaction appends. Each append
1760 : /// holds `FOR KEY SHARE` on its Entity until the commit, which only a DELETE
1761 : /// of that Entity waits for (writes lock `FOR NO KEY UPDATE`); a
1762 : /// 1000-member page in one transaction would hold such a delete for the whole
1763 : /// append. Small requests, a live vehicle feed's, are one transaction.
1764 : // A fixed chunk; a measured one if lock waits show.
1765 : #[cfg(feature = "postgres")]
1766 : const HISTORY_CHUNK: usize = 64;
1767 :
1768 : #[async_trait::async_trait]
1769 : impl antares_store::CurrentStateDriver for AnyStore {
1770 16 : async fn ping(&self) -> Result<(), NgsiError> {
1771 : AnyStore::ping(self).await
1772 16 : }
1773 86 : fn commit_queue(&self) -> Option<(usize, usize)> {
1774 86 : AnyStore::commit_queue(self)
1775 86 : }
1776 86 : fn version_info(&self) -> Value {
1777 86 : AnyStore::version_info(self)
1778 86 : }
1779 16 : async fn close(&self) {
1780 : AnyStore::close(self).await;
1781 16 : }
1782 1290 : fn set_change_hook(&self, h: super::ChangeHook) {
1783 1290 : AnyStore::set_change_hook(self, h);
1784 1290 : }
1785 0 : fn set_outbox(&self, on: bool) {
1786 0 : AnyStore::set_outbox(self, on);
1787 0 : }
1788 730 : async fn outbox_peek(&self, limit: i64) -> Result<Vec<(i64, String, Value)>, NgsiError> {
1789 : AnyStore::outbox_peek(self, limit).await
1790 730 : }
1791 0 : async fn outbox_ack(&self, seqs: &[i64]) -> Result<u64, NgsiError> {
1792 : AnyStore::outbox_ack(self, seqs).await
1793 0 : }
1794 0 : async fn outbox_retain(&self, tenant: &TenantId, seqs: &[i64]) -> Result<u64, NgsiError> {
1795 : AnyStore::outbox_retain(self, tenant, seqs).await
1796 0 : }
1797 2 : async fn outbox_event(&self, seq: i64, tenant: &TenantId) -> Result<Option<Value>, NgsiError> {
1798 : AnyStore::outbox_event(self, seq, tenant).await
1799 2 : }
1800 : async fn create(
1801 : &self,
1802 : tenant: &TenantId,
1803 : kind: Kind,
1804 : id: &str,
1805 : doc: Value,
1806 160682 : ) -> Result<bool, NgsiError> {
1807 : AnyStore::create(self, tenant, kind, id, doc).await
1808 160682 : }
1809 : async fn batch_create(
1810 : &self,
1811 : tenant: &TenantId,
1812 : items: Vec<(String, Value)>,
1813 156 : ) -> Result<Vec<bool>, NgsiError> {
1814 : AnyStore::batch_create(self, tenant, items).await
1815 156 : }
1816 : async fn batch_delete(
1817 : &self,
1818 : tenant: &TenantId,
1819 : ids: &[String],
1820 390 : ) -> Result<Vec<bool>, NgsiError> {
1821 : AnyStore::batch_delete(self, tenant, ids).await
1822 390 : }
1823 : async fn batch_upsert(
1824 : &self,
1825 : tenant: &TenantId,
1826 : items: Vec<(String, Value)>,
1827 178 : ) -> Result<Vec<bool>, NgsiError> {
1828 : AnyStore::batch_upsert(self, tenant, items).await
1829 178 : }
1830 : async fn upsert(
1831 : &self,
1832 : tenant: &TenantId,
1833 : kind: Kind,
1834 : id: &str,
1835 : doc: Value,
1836 20760 : ) -> Result<bool, NgsiError> {
1837 : AnyStore::upsert(self, tenant, kind, id, doc).await
1838 20760 : }
1839 : async fn get(
1840 : &self,
1841 : tenant: &TenantId,
1842 : kind: Kind,
1843 : id: &str,
1844 30592 : ) -> Result<Option<Value>, NgsiError> {
1845 : AnyStore::get(self, tenant, kind, id).await
1846 30592 : }
1847 : async fn get_entities(
1848 : &self,
1849 : tenant: &TenantId,
1850 : ids: &[&str],
1851 72 : ) -> Result<std::collections::HashMap<String, Value>, NgsiError> {
1852 : AnyStore::get_entities(self, tenant, ids).await
1853 72 : }
1854 4960 : async fn delete(&self, tenant: &TenantId, kind: Kind, id: &str) -> Result<bool, NgsiError> {
1855 : AnyStore::delete(self, tenant, kind, id).await
1856 4960 : }
1857 : async fn delete_entity_if(
1858 : &self,
1859 : tenant: &TenantId,
1860 : id: &str,
1861 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
1862 3126 : ) -> Result<bool, NgsiError> {
1863 : AnyStore::delete_entity_if(self, tenant, id, keep).await
1864 3126 : }
1865 5160 : async fn list(&self, tenant: &TenantId, kind: Kind) -> Result<Vec<Value>, NgsiError> {
1866 : AnyStore::list(self, tenant, kind).await
1867 5160 : }
1868 : async fn list_digests(
1869 : &self,
1870 : tenant: &TenantId,
1871 : kind: Kind,
1872 0 : ) -> Result<Vec<(String, u64)>, NgsiError> {
1873 : AnyStore::list_digests(self, tenant, kind).await
1874 0 : }
1875 : async fn list_page(
1876 : &self,
1877 : tenant: &TenantId,
1878 : kind: Kind,
1879 : after: Option<&str>,
1880 : limit: usize,
1881 617492 : ) -> Result<Vec<Value>, NgsiError> {
1882 : AnyStore::list_page(self, tenant, kind, after, limit).await
1883 617492 : }
1884 : async fn list_slice(
1885 : &self,
1886 : tenant: &TenantId,
1887 : kind: Kind,
1888 : offset: usize,
1889 : limit: usize,
1890 286 : ) -> Result<(Vec<Value>, usize), NgsiError> {
1891 : AnyStore::list_slice(self, tenant, kind, offset, limit).await
1892 286 : }
1893 : async fn matching_registrations(
1894 : &self,
1895 : tenant: &TenantId,
1896 : ids: Option<&[String]>,
1897 : types: Option<&[String]>,
1898 4855 : ) -> Result<Vec<Value>, NgsiError> {
1899 : AnyStore::matching_registrations(self, tenant, ids, types).await
1900 4855 : }
1901 : async fn query_entities(
1902 : &self,
1903 : tenant: &TenantId,
1904 : f: &crate::store::filter::EntityFilter<'_>,
1905 3723 : ) -> Result<crate::store::filter::QueryOutcome, NgsiError> {
1906 : AnyStore::query_entities(self, tenant, f).await
1907 3723 : }
1908 : async fn query_candidates(
1909 : &self,
1910 : tenant: &TenantId,
1911 : f: &crate::store::filter::EntityFilter<'_>,
1912 372 : ) -> Result<crate::store::filter::QueryOutcome, NgsiError> {
1913 : AnyStore::query_candidates(self, tenant, f).await
1914 372 : }
1915 182 : async fn entity_type_names(&self, tenant: &TenantId) -> Result<Option<Vec<String>>, NgsiError> {
1916 : AnyStore::entity_type_names(self, tenant).await.map(Some)
1917 182 : }
1918 : async fn entity_attr_pairs(
1919 : &self,
1920 : tenant: &TenantId,
1921 152 : ) -> Result<Option<Vec<(String, String)>>, NgsiError> {
1922 : AnyStore::entity_attr_pairs(self, tenant).await.map(Some)
1923 152 : }
1924 : async fn entity_types_beside(
1925 : &self,
1926 : tenant: &TenantId,
1927 : any_of: &[String],
1928 84 : ) -> Result<Option<Vec<String>>, NgsiError> {
1929 : AnyStore::entity_types_beside(self, tenant, any_of)
1930 : .await
1931 : .map(Some)
1932 84 : }
1933 : async fn entity_id_segments(
1934 : &self,
1935 : tenant: &TenantId,
1936 : lead: &str,
1937 : sep: char,
1938 : max: usize,
1939 16 : ) -> Result<Option<Vec<String>>, NgsiError> {
1940 : AnyStore::entity_id_segments(self, tenant, lead, sep, max).await
1941 16 : }
1942 : async fn mutate_boxed<'a>(
1943 : &self,
1944 : tenant: &TenantId,
1945 : kind: Kind,
1946 : id: &str,
1947 : f: antares_store::MutateFn<'a>,
1948 24976 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
1949 : AnyStore::mutate(self, tenant, kind, id, f).await
1950 24976 : }
1951 : async fn batch_mutate_boxed<'a>(
1952 : &self,
1953 : tenant: &TenantId,
1954 : ids: &[String],
1955 : mut f: antares_store::BatchMutateFn<'a>,
1956 236 : ) -> Result<Vec<Option<Result<(), ()>>>, NgsiError> {
1957 8324 : AnyStore::batch_mutate(self, tenant, ids, |id, v| f(id, v)).await
1958 236 : }
1959 : /// The Postgres arm answers this in ONE statement (see
1960 : /// `pg::doc::record_delivery`); every other arm runs the shared rule as a
1961 : /// `mutate`, which is what the one statement is a translation of.
1962 : async fn record_delivery(
1963 : &self,
1964 : tenant: &TenantId,
1965 : kind: Kind,
1966 : id: &str,
1967 : now: &str,
1968 5004 : ) -> Result<Option<antares_store::Delivery>, NgsiError> {
1969 : // `doc_kind` exists only with the Postgres arm, so its call lives
1970 : // inside the gate with it: outside the feature this function is the
1971 : // shared rule and nothing else.
1972 : #[cfg(feature = "postgres")]
1973 : {
1974 : if let (AnyStore::Pg(p), Ok(dk)) = (self, doc_kind(kind)) {
1975 : let found = p
1976 : .docs
1977 : .record_delivery(tenant, dk, id, now)
1978 : .await
1979 : .map_err(db)?;
1980 : return Ok(
1981 238 : found.map(|(doc, prev_success)| antares_store::Delivery { doc, prev_success })
1982 : );
1983 : }
1984 : }
1985 : antares_store::record_delivery_via_mutate(self, tenant, kind, id, now).await
1986 5004 : }
1987 : /// The Postgres arm books every attempt in ONE statement (see
1988 : /// `pg::doc::record_deliveries`); every other arm books them one by one.
1989 : async fn record_deliveries(
1990 : &self,
1991 : tenant: &TenantId,
1992 : kind: Kind,
1993 : attempts: &[(&str, &str)],
1994 36 : ) -> Result<Vec<Option<antares_store::Delivery>>, NgsiError> {
1995 : #[cfg(feature = "postgres")]
1996 : {
1997 : if let (AnyStore::Pg(p), Ok(dk)) = (self, doc_kind(kind)) {
1998 : let mut found = p
1999 : .docs
2000 : .record_deliveries(tenant, dk, attempts)
2001 : .await
2002 : .map_err(db)?;
2003 : return Ok(attempts
2004 : .iter()
2005 106 : .map(|(id, _)| {
2006 106 : found
2007 106 : .remove(*id)
2008 106 : .map(|(doc, prev_success)| antares_store::Delivery {
2009 102 : doc,
2010 102 : prev_success,
2011 102 : })
2012 106 : })
2013 : .collect());
2014 : }
2015 : }
2016 : let mut out = Vec::with_capacity(attempts.len());
2017 : for (id, now) in attempts {
2018 : out.push(antares_store::record_delivery_via_mutate(self, tenant, kind, id, now).await?);
2019 : }
2020 : Ok(out)
2021 36 : }
2022 1956 : async fn sweep_expired(&self) -> usize {
2023 : AnyStore::sweep_expired(self)
2024 1956 : }
2025 : async fn retain_current(
2026 : &self,
2027 : tenant: &TenantId,
2028 : type_iri: &str,
2029 : before: &str,
2030 : batch: usize,
2031 : dry_run: bool,
2032 16 : ) -> Result<u64, NgsiError> {
2033 : match self {
2034 : AnyStore::Mem(s) => Ok(s.retain_current(tenant, type_iri, before, batch, dry_run)),
2035 : #[cfg(feature = "postgres")]
2036 : AnyStore::Pg(p) => crate::store::pg::maintenance::retain_current(
2037 : p.docs.pool(),
2038 : tenant.as_str(),
2039 : type_iri,
2040 : before,
2041 : batch,
2042 : dry_run,
2043 : )
2044 : .await
2045 : .map_err(db),
2046 : }
2047 16 : }
2048 298 : async fn tenant_exists(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
2049 : AnyStore::tenant_exists(self, tenant).await
2050 298 : }
2051 3153 : async fn subscription_tenants(&self) -> Result<Vec<String>, NgsiError> {
2052 : AnyStore::subscription_tenants(self).await
2053 3153 : }
2054 2000 : async fn tenant_ids(&self) -> Result<Vec<String>, NgsiError> {
2055 : AnyStore::tenant_ids(self).await
2056 2000 : }
2057 : async fn history_overrides(
2058 : &self,
2059 : limit: usize,
2060 4 : ) -> Result<Vec<(String, String, Value)>, NgsiError> {
2061 : match self {
2062 : AnyStore::Mem(s) => Ok(s.history_overrides(limit)),
2063 : #[cfg(feature = "postgres")]
2064 : AnyStore::Pg(p) => {
2065 : crate::store::pg::maintenance::history_overrides(p.docs.pool(), limit)
2066 : .await
2067 : .map_err(db)
2068 : }
2069 : }
2070 4 : }
2071 : async fn tenant_stats_one(
2072 : &self,
2073 : tenant: &TenantId,
2074 40 : ) -> Result<Option<antares_store::TenantStats>, NgsiError> {
2075 : AnyStore::tenant_stats_one(self, tenant).await
2076 40 : }
2077 : async fn tenant_bytes(
2078 : &self,
2079 : tenant: &TenantId,
2080 38 : ) -> Result<antares_store::TenantBytes, NgsiError> {
2081 : match self {
2082 : AnyStore::Mem(s) => Ok(s.tenant_bytes(tenant)),
2083 : #[cfg(feature = "postgres")]
2084 : AnyStore::Pg(p) => {
2085 8 : async {
2086 8 : let mut tx = tenant_tx(p, tenant).await?;
2087 : // a scan of the tenant: bounded, a slow one answers no size
2088 8 : sqlx::query(SIZE_SCAN_TIMEOUT)
2089 8 : .execute(&mut *tx)
2090 8 : .await
2091 8 : .map_err(db)?;
2092 : // the stored row, TOAST included, by the first type the
2093 : // write recorded
2094 8 : let rows: Vec<(String, i64)> = sqlx::query_as(
2095 8 : "SELECT coalesce(types[1], ''), sum(pg_column_size(e.*))::bigint
2096 8 : FROM entities e WHERE tenant_id = $1 GROUP BY 1",
2097 8 : )
2098 8 : .bind(tenant.as_str())
2099 8 : .fetch_all(&mut *tx)
2100 8 : .await
2101 8 : .map_err(db)?;
2102 6 : let mut out = antares_store::TenantBytes::default();
2103 6 : for (ty, n) in rows {
2104 6 : let n = n.max(0) as u64;
2105 6 : out.entities = out.entities.saturating_add(n);
2106 6 : out.by_type.insert(ty, n);
2107 6 : }
2108 6 : Ok(out)
2109 8 : }
2110 : .await
2111 : }
2112 : }
2113 38 : }
2114 116 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<bool, NgsiError> {
2115 : AnyStore::purge_tenant(self, tenant).await
2116 116 : }
2117 : async fn context_put(
2118 : &self,
2119 : tenant: Option<&TenantId>,
2120 : id: &str,
2121 : doc: Value,
2122 688 : ) -> Result<(), NgsiError> {
2123 : AnyStore::context_put(self, tenant, id, doc).await
2124 688 : }
2125 : async fn context_get(
2126 : &self,
2127 : tenant: Option<&TenantId>,
2128 : id: &str,
2129 1350 : ) -> Result<Option<Value>, NgsiError> {
2130 : AnyStore::context_get(self, tenant, id).await
2131 1350 : }
2132 : async fn context_mutate<'a>(
2133 : &self,
2134 : tenant: Option<&TenantId>,
2135 : id: &str,
2136 : f: antares_store::MutateFn<'a>,
2137 21058 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
2138 : AnyStore::context_mutate(self, tenant, id, f).await
2139 21058 : }
2140 412 : async fn context_delete(&self, tenant: Option<&TenantId>, id: &str) -> Result<bool, NgsiError> {
2141 : AnyStore::context_delete(self, tenant, id).await
2142 412 : }
2143 282 : async fn context_list_meta(&self, tenant: Option<&TenantId>) -> Result<Vec<Value>, NgsiError> {
2144 : AnyStore::context_list_meta(self, tenant).await
2145 282 : }
2146 : }
2147 :
2148 : #[async_trait::async_trait]
2149 : impl antares_store::TemporalDriver for AnyStore {
2150 8 : async fn close(&self) {
2151 : AnyStore::close(self).await;
2152 8 : }
2153 0 : fn set_short_history(&self, short: antares_store::ShortHistory) {
2154 0 : match self {
2155 : // Memory has no WAL to save
2156 0 : AnyStore::Mem(_) => drop(short),
2157 : #[cfg(feature = "postgres")]
2158 0 : AnyStore::Pg(p) => p.temporal.set_short_history(short),
2159 : }
2160 0 : }
2161 : async fn versions(
2162 : &self,
2163 : tenant: &TenantId,
2164 : entity_id: &str,
2165 0 : ) -> Result<serde_json::Map<String, Value>, NgsiError> {
2166 : match self {
2167 : // Memory keeps no versions or provenance
2168 : AnyStore::Mem(_) => {
2169 : let _ = (tenant, entity_id);
2170 : Ok(serde_json::Map::new())
2171 : }
2172 : #[cfg(feature = "postgres")]
2173 : AnyStore::Pg(p) => p.temporal.versions(tenant, entity_id).await.map_err(db),
2174 : }
2175 0 : }
2176 : async fn erase(
2177 : &self,
2178 : tenant: &TenantId,
2179 : entity_id: &str,
2180 : attrs: Option<&[String]>,
2181 : before: Option<&str>,
2182 0 : ) -> Result<u64, NgsiError> {
2183 : match self {
2184 : AnyStore::Mem(_) => {
2185 0 : let instant = |t: &str| {
2186 0 : chrono::DateTime::parse_from_rfc3339(
2187 0 : &antares_store::filter::canonical_datetime(t),
2188 : )
2189 0 : .ok()
2190 0 : };
2191 : // an instance goes when it was observed at or before the cut;
2192 : // a cut that is no DateTime erases nothing
2193 : let cut = before.map(instant);
2194 0 : let goes = |i: &Value| match cut {
2195 0 : None => true,
2196 0 : Some(cut) => {
2197 0 : let at = i
2198 0 : .get("observedAt")
2199 0 : .and_then(Value::as_str)
2200 0 : .and_then(instant);
2201 0 : cut.zip(at).is_some_and(|(cut, at)| at <= cut)
2202 : }
2203 0 : };
2204 0 : let erased = AnyStore::mutate(self, tenant, Kind::Temporal, entity_id, |doc| {
2205 0 : let o = doc.as_object_mut().ok_or(())?;
2206 0 : let mut n = 0u64;
2207 0 : for (k, v) in o.iter_mut() {
2208 0 : let named = attrs.is_none_or(|a| a.contains(k));
2209 0 : let Some(list) = v.as_array_mut().filter(|_| named) else {
2210 0 : continue;
2211 : };
2212 0 : if !list.iter().all(Value::is_object) {
2213 0 : continue; // the Entity's own `type` list
2214 0 : }
2215 0 : let was = list.len();
2216 0 : list.retain(|i| !goes(i));
2217 0 : let gone = was.saturating_sub(list.len());
2218 0 : n = n.saturating_add(u64::try_from(gone).unwrap_or(u64::MAX));
2219 : }
2220 0 : let emptied = o
2221 0 : .values()
2222 0 : .filter_map(Value::as_array)
2223 0 : .all(|l| l.is_empty() || !l.iter().all(Value::is_object));
2224 0 : Ok::<_, ()>((n, emptied))
2225 0 : })
2226 : .await?;
2227 : let Some(Ok((n, emptied))) = erased else {
2228 : return Ok(0);
2229 : };
2230 : if attrs.is_none() && emptied {
2231 : AnyStore::delete(self, tenant, Kind::Temporal, entity_id).await?;
2232 : }
2233 : Ok(n)
2234 : }
2235 : #[cfg(feature = "postgres")]
2236 : AnyStore::Pg(p) => p
2237 : .temporal
2238 : .erase(tenant, entity_id, attrs, before)
2239 : .await
2240 : .map_err(db),
2241 : }
2242 0 : }
2243 : async fn retain_history(
2244 : &self,
2245 : tenant: &TenantId,
2246 : scope: antares_store::HistoryScope<'_>,
2247 : before: &str,
2248 : batch: usize,
2249 : dry_run: bool,
2250 32 : ) -> Result<u64, NgsiError> {
2251 : match self {
2252 : AnyStore::Mem(s) => Ok(s.retain_history(tenant, scope, before, batch, dry_run)),
2253 : #[cfg(feature = "postgres")]
2254 : AnyStore::Pg(p) => crate::store::pg::maintenance::retain_history(
2255 : p.docs.pool(),
2256 : tenant.as_str(),
2257 : scope,
2258 : before,
2259 : batch,
2260 : dry_run,
2261 : )
2262 : .await
2263 : .map_err(db),
2264 : }
2265 32 : }
2266 : async fn drop_retained_history(
2267 : &self,
2268 : floors: &[(String, String)],
2269 0 : ) -> Result<Vec<String>, NgsiError> {
2270 : match self {
2271 : // memory keeps no partitions
2272 : AnyStore::Mem(_) => {
2273 0 : let _ = floors;
2274 : Ok(Vec::new())
2275 : }
2276 : #[cfg(feature = "postgres")]
2277 : AnyStore::Pg(p) => {
2278 : crate::store::pg::maintenance::drop_retained_partitions(p.docs.pool(), floors)
2279 : .await
2280 : .map_err(db)
2281 : }
2282 : }
2283 0 : }
2284 80 : fn version_info(&self) -> Value {
2285 80 : AnyStore::version_info(self)
2286 80 : }
2287 6 : async fn attr_instance_count(&self, tenant: &TenantId) -> Result<u64, NgsiError> {
2288 : match self {
2289 : AnyStore::Mem(s) => Ok(s.attr_instance_count(tenant)),
2290 : #[cfg(feature = "postgres")]
2291 : AnyStore::Pg(p) => {
2292 4 : async {
2293 4 : let mut tx = tenant_tx(p, tenant).await?;
2294 4 : count_instances(&mut tx, tenant).await
2295 4 : }
2296 : .await
2297 : }
2298 : }
2299 6 : }
2300 : async fn attr_instance_count_within(
2301 : &self,
2302 : tenant: &TenantId,
2303 : bound: std::time::Duration,
2304 34 : ) -> Result<u64, NgsiError> {
2305 : match self {
2306 : AnyStore::Mem(s) => {
2307 : // the memory store counts at once: no bound to apply
2308 : let _ = bound;
2309 : Ok(s.attr_instance_count(tenant))
2310 : }
2311 : #[cfg(feature = "postgres")]
2312 : AnyStore::Pg(p) => {
2313 4 : async {
2314 4 : let mut tx = tenant_tx(p, tenant).await?;
2315 : // a scan of the tenant's history: a slow one answers no count
2316 4 : sqlx::query("SELECT set_config('statement_timeout', $1, true)")
2317 4 : .bind(format!("{}ms", bound.as_millis().max(1)))
2318 4 : .execute(&mut *tx)
2319 4 : .await
2320 4 : .map_err(db)?;
2321 4 : count_instances(&mut tx, tenant).await
2322 4 : }
2323 : .await
2324 : }
2325 : }
2326 34 : }
2327 36 : async fn attr_instance_bytes(&self, tenant: &TenantId) -> Result<u64, NgsiError> {
2328 : match self {
2329 : AnyStore::Mem(s) => Ok(s.attr_instance_bytes(tenant)),
2330 : #[cfg(feature = "postgres")]
2331 : AnyStore::Pg(p) => {
2332 8 : async {
2333 8 : let mut tx = tenant_tx(p, tenant).await?;
2334 8 : sqlx::query(SIZE_SCAN_TIMEOUT)
2335 8 : .execute(&mut *tx)
2336 8 : .await
2337 8 : .map_err(db)?;
2338 8 : let n: i64 = sqlx::query_scalar(
2339 8 : "SELECT (SELECT coalesce(sum(pg_column_size(a.*)), 0)
2340 8 : FROM attr_instances a WHERE tenant_id = $1)::bigint
2341 8 : + (SELECT coalesce(sum(pg_column_size(r.*)), 0)
2342 8 : FROM attr_rows r WHERE tenant_id = $1)::bigint",
2343 8 : )
2344 8 : .bind(tenant.as_str())
2345 8 : .fetch_one(&mut *tx)
2346 8 : .await
2347 8 : .map_err(db)?;
2348 6 : Ok(n.max(0) as u64)
2349 8 : }
2350 : .await
2351 : }
2352 : }
2353 36 : }
2354 112 : async fn purge_tenant(&self, tenant: &TenantId) -> Result<(), NgsiError> {
2355 : match self {
2356 : AnyStore::Mem(s) => {
2357 : s.purge_kinds(tenant, &[Kind::Temporal]);
2358 : Ok(())
2359 : }
2360 : #[cfg(feature = "postgres")]
2361 : AnyStore::Pg(p) => {
2362 26 : async {
2363 26 : let mut tx = tenant_tx(p, tenant).await?;
2364 : // whole statements, not a spliced table name: the
2365 : // attr_instances tenant-predicate check reads literals
2366 130 : for sql in [
2367 26 : "DELETE FROM attr_instances WHERE tenant_id = $1",
2368 26 : "DELETE FROM attr_rows WHERE tenant_id = $1",
2369 26 : "DELETE FROM iri_ids WHERE tenant_id = $1",
2370 26 : "DELETE FROM attr_versions WHERE tenant_id = $1",
2371 26 : "DELETE FROM temporal_entities WHERE tenant_id = $1",
2372 26 : ] {
2373 130 : sqlx::query(sql)
2374 130 : .bind(tenant.as_str())
2375 130 : .execute(&mut *tx)
2376 130 : .await
2377 130 : .map_err(db)?;
2378 : }
2379 26 : tx.commit().await.map_err(db)?;
2380 26 : p.temporal.forget_tenant(tenant);
2381 26 : Ok(())
2382 26 : }
2383 : .await
2384 : }
2385 : }
2386 112 : }
2387 : async fn temporal_append(
2388 : &self,
2389 : tenant: &TenantId,
2390 : id: &str,
2391 : shell: &Value,
2392 : additions: &Value,
2393 121884 : ) -> Result<(), NgsiError> {
2394 : AnyStore::temporal_append(self, tenant, id, shell, additions).await
2395 121884 : }
2396 : /// On Postgres a drained request is one transaction per
2397 : /// `HISTORY_CHUNK` Entities; a scope change, and the memory
2398 : /// arm, take the default's append per Entity.
2399 123506 : async fn event_list(&self, evs: &[antares_store::TemporalEvent]) -> Result<(), NgsiError> {
2400 : #[cfg(feature = "postgres")]
2401 : if let (AnyStore::Pg(p), Some(appends)) = (self, antares_store::fold_appends(evs)) {
2402 : // One chunk that fails (a lock timeout) loses its own Entities'
2403 : // history, never the rest of the request's.
2404 : let mut first = None;
2405 : for chunk in appends.chunks(HISTORY_CHUNK) {
2406 : if let Err(e) = p.temporal.append_many(chunk).await {
2407 : first.get_or_insert(e);
2408 : }
2409 : }
2410 0 : return first.map_or(Ok(()), |e| Err(db(e)));
2411 : }
2412 : antares_store::drain_events(self, evs).await
2413 123506 : }
2414 : /// Only the Postgres arm pages in SQL, and only an exact prefilter
2415 : /// makes that page the evaluator's page; the memory arm answers
2416 : /// `paged: false` and the caller pages.
2417 246 : fn q_pushdown_exact(
2418 246 : &self,
2419 246 : q: &antares_ql::QNode,
2420 246 : range: Option<&crate::store::filter::InstanceRange<'_>>,
2421 246 : expand: &dyn Fn(&str) -> String,
2422 246 : ) -> bool {
2423 246 : match self {
2424 : // the memory arm pages nothing, so the filter is not its business
2425 : AnyStore::Mem(_) => {
2426 146 : let _ = (q, range, expand);
2427 146 : false
2428 : }
2429 : #[cfg(feature = "postgres")]
2430 100 : AnyStore::Pg(_) => crate::compile::qprefilter::prefilter_exact(q, range, expand),
2431 : }
2432 246 : }
2433 : async fn query_temporal(
2434 : &self,
2435 : tenant: &TenantId,
2436 : f: &crate::store::filter::TemporalFilter<'_>,
2437 1046 : ) -> Result<crate::store::filter::TemporalOutcome, NgsiError> {
2438 : AnyStore::query_temporal(self, tenant, f).await
2439 1046 : }
2440 : async fn get_temporal(
2441 : &self,
2442 : tenant: &TenantId,
2443 : id: &str,
2444 : f: &crate::store::filter::TemporalFilter<'_>,
2445 1032 : ) -> Result<Option<Value>, NgsiError> {
2446 : AnyStore::get_temporal(self, tenant, id, f).await
2447 1032 : }
2448 1604 : async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, NgsiError> {
2449 : AnyStore::get(self, tenant, Kind::Temporal, id).await
2450 1604 : }
2451 1256 : async fn create(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError> {
2452 : AnyStore::create(self, tenant, Kind::Temporal, id, doc).await
2453 1256 : }
2454 56 : async fn upsert(&self, tenant: &TenantId, id: &str, doc: Value) -> Result<bool, NgsiError> {
2455 : AnyStore::upsert(self, tenant, Kind::Temporal, id, doc).await
2456 56 : }
2457 15032 : async fn delete(&self, tenant: &TenantId, id: &str) -> Result<bool, NgsiError> {
2458 : AnyStore::delete(self, tenant, Kind::Temporal, id).await
2459 15032 : }
2460 50 : async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, NgsiError> {
2461 : AnyStore::list(self, tenant, Kind::Temporal).await
2462 50 : }
2463 : async fn mutate_boxed<'a>(
2464 : &self,
2465 : tenant: &TenantId,
2466 : id: &str,
2467 : f: antares_store::MutateFn<'a>,
2468 848 : ) -> Result<Option<Result<(), ()>>, NgsiError> {
2469 : // Postgres takes the provenance into its column in decompose; the
2470 : // memory store keeps none, so neither arm answers it as data.
2471 : let keeps_none = matches!(self, AnyStore::Mem(_));
2472 730 : AnyStore::mutate(self, tenant, Kind::Temporal, id, move |doc: &mut Value| {
2473 730 : let done = f(doc);
2474 730 : if keeps_none {
2475 454 : antares_store::strip_provenance(doc);
2476 548 : }
2477 730 : done
2478 730 : })
2479 : .await
2480 848 : }
2481 : }
2482 :
2483 : #[cfg(test)]
2484 : mod temporal_only_tests {
2485 : use super::*;
2486 : use antares_model::TenantId;
2487 :
2488 4 : async fn append(store: &AnyStore) -> Option<Value> {
2489 4 : let t = TenantId::new("combo").expect("tenant");
2490 4 : let shell = serde_json::json!({"id": "urn:a", "type": ["T"]});
2491 4 : let adds = serde_json::json!({"speed": [{"instanceId": "urn:i:1", "value": 1}]});
2492 4 : store
2493 4 : .temporal_append(&t, "urn:a", &shell, &adds)
2494 4 : .await
2495 4 : .expect("append");
2496 4 : store.get(&t, Kind::Temporal, "urn:a").await.expect("get")
2497 4 : }
2498 :
2499 : /// A shared instance records only for an entity it holds (5.6.6 delete
2500 : /// overlap); the temporal half of a mixed deployment records
2501 : /// unconditionally, since the entities live elsewhere.
2502 : #[tokio::test]
2503 2 : async fn a_temporal_only_instance_records_without_the_entity() {
2504 2 : assert!(append(&AnyStore::Mem(Store::default())).await.is_none());
2505 2 : let doc = append(&AnyStore::Mem(Store::default()).temporal_only())
2506 2 : .await
2507 2 : .expect("recorded");
2508 2 : assert_eq!(doc["speed"][0]["value"], 1);
2509 2 : }
2510 : }
2511 :
2512 : #[cfg(test)]
2513 : mod instance_cap_tests {
2514 : use super::*;
2515 : use antares_model::TenantId;
2516 : use antares_store::TemporalDriverExt;
2517 :
2518 46 : fn t() -> TenantId {
2519 46 : TenantId::new("capped").expect("tenant")
2520 46 : }
2521 :
2522 42 : fn inst(i: usize) -> Value {
2523 42 : serde_json::json!({"instanceId": format!("urn:i:{i}"), "type": "Property", "value": i,
2524 42 : "observedAt": format!("2026-01-01T00:00:{:02}Z", i % 60)})
2525 42 : }
2526 :
2527 20 : async fn append(store: &AnyStore, i: usize) -> Result<(), NgsiError> {
2528 20 : let shell = serde_json::json!({"id": "urn:a", "type": ["T"]});
2529 20 : let adds = serde_json::json!({"speed": [inst(i)]});
2530 20 : store.temporal_append(&t(), "urn:a", &shell, &adds).await
2531 20 : }
2532 :
2533 12 : async fn speed(store: &AnyStore) -> usize {
2534 12 : store
2535 12 : .get(&t(), Kind::Temporal, "urn:a")
2536 12 : .await
2537 12 : .expect("get")
2538 12 : .and_then(|d| d["speed"].as_array().map(Vec::len))
2539 12 : .unwrap_or(0)
2540 12 : }
2541 :
2542 : /// One Attribute's history in the memory/file store is bounded: the
2543 : /// append past the cap is refused with an error that says why, and
2544 : /// nothing of it is written. The auto-recording of Entity writes
2545 : /// takes this path.
2546 : #[tokio::test]
2547 2 : async fn an_append_past_the_instance_cap_is_refused_and_writes_nothing() {
2548 2 : let store = AnyStore::Mem(Store::default())
2549 2 : .temporal_only()
2550 2 : .max_instances(3);
2551 6 : for i in 0..3 {
2552 6 : append(&store, i).await.expect("under the cap");
2553 : }
2554 2 : assert_eq!(speed(&store).await, 3);
2555 2 : let refused = append(&store, 3).await;
2556 2 : assert!(
2557 2 : matches!(&refused, Err(NgsiError::BadRequestData(m)) if m.contains("3 instances")),
2558 : "{refused:?}"
2559 : );
2560 2 : assert_eq!(speed(&store).await, 3, "the refused append wrote");
2561 : // a correction of a held instance adds none and still lands
2562 2 : append(&store, 2)
2563 2 : .await
2564 2 : .expect("same instanceId, same count");
2565 : // no cap, no refusal
2566 2 : let open = AnyStore::Mem(Store::default()).temporal_only();
2567 10 : for i in 0..5 {
2568 10 : append(&open, i).await.expect("uncapped");
2569 2 : }
2570 2 : assert_eq!(speed(&open).await, 5);
2571 2 : }
2572 :
2573 : /// A first append past the cap is refused before anything is stored:
2574 : /// no attribute-less Temporal Entity is left behind it.
2575 : #[tokio::test]
2576 2 : async fn a_refused_first_append_leaves_no_empty_temporal_shell() {
2577 2 : let store = AnyStore::Mem(Store::default())
2578 2 : .temporal_only()
2579 2 : .max_instances(2);
2580 2 : let shell = serde_json::json!({"id": "urn:a", "type": ["T"]});
2581 2 : let adds = serde_json::json!({"speed": [inst(0), inst(1), inst(2)]});
2582 2 : assert!(store
2583 2 : .temporal_append(&t(), "urn:a", &shell, &adds)
2584 2 : .await
2585 2 : .is_err());
2586 2 : assert!(store
2587 2 : .get(&t(), Kind::Temporal, "urn:a")
2588 2 : .await
2589 2 : .expect("get")
2590 2 : .is_none());
2591 2 : }
2592 :
2593 : /// The temporal API's own writes (5.6.11 create, 5.6.12 add) go through
2594 : /// create and mutate: the same cap, the same refusal.
2595 : #[tokio::test]
2596 2 : async fn the_temporal_api_writes_hit_the_same_cap() {
2597 : use antares_store::TemporalDriver;
2598 2 : let store = AnyStore::Mem(Store::default())
2599 2 : .temporal_only()
2600 2 : .max_instances(3);
2601 4 : let doc = |n: usize| {
2602 4 : serde_json::json!({"id": "urn:a", "type": ["T"],
2603 4 : "speed": (0..n).map(inst).collect::<Vec<_>>()})
2604 4 : };
2605 2 : let refused = TemporalDriver::create(&store, &t(), "urn:a", doc(4)).await;
2606 2 : assert!(
2607 2 : matches!(refused, Err(NgsiError::BadRequestData(_))),
2608 : "{refused:?}"
2609 : );
2610 2 : assert_eq!(speed(&store).await, 0, "the refused create wrote");
2611 2 : assert!(TemporalDriver::create(&store, &t(), "urn:a", doc(3))
2612 2 : .await
2613 2 : .expect("create at the cap"));
2614 2 : let grow = TemporalDriverExt::mutate(&store, &t(), "urn:a", |d| {
2615 2 : if let Some(a) = d["speed"].as_array_mut() {
2616 2 : a.push(inst(9));
2617 2 : }
2618 2 : Ok::<(), ()>(())
2619 2 : })
2620 2 : .await;
2621 2 : assert!(
2622 2 : matches!(grow, Err(NgsiError::BadRequestData(_))),
2623 : "{grow:?}"
2624 : );
2625 2 : assert_eq!(speed(&store).await, 3, "the refused add wrote");
2626 : // the closure's own refusal still reaches its caller as its own
2627 2 : let own = TemporalDriverExt::mutate(&store, &t(), "urn:a", |_| Err::<(), _>("mine")).await;
2628 2 : assert!(matches!(own, Ok(Some(Err("mine")))), "{own:?}");
2629 : // deleting instances of a full history is never refused
2630 2 : let shrink = TemporalDriverExt::mutate(&store, &t(), "urn:a", |d| {
2631 2 : if let Some(a) = d["speed"].as_array_mut() {
2632 2 : a.pop();
2633 2 : }
2634 2 : Ok::<(), ()>(())
2635 2 : })
2636 2 : .await;
2637 2 : assert!(matches!(shrink, Ok(Some(Ok(())))), "{shrink:?}");
2638 2 : assert_eq!(speed(&store).await, 2);
2639 2 : }
2640 : }
|
