Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! PgStore slice two: the doc-table kinds —
3 : //! `subscriptions`, `csource_registrations`, `csource_subscriptions` — plus
4 : //! cross-tenant `jsonld_contexts`. Same sync-facade shape as `pg_entity`.
5 : //!
6 : //! The v0 interchange form stores ONE doc per resource; the bookkeeping
7 : //! columns (`expires_at`, `is_active`, `times_sent`, `last_*`) are EXTRACTED
8 : //! from the doc on every write, so the row stays the truth while the
9 : //! API layer keeps its doc-shaped view until the cutover completes.
10 :
11 : use antares_model::operations::{group_members, DEFAULT_OPERATION_GROUP, OPERATION_NAMES};
12 : use antares_model::TenantId;
13 : use serde_json::Value;
14 : use sqlx::postgres::PgPool;
15 : use sqlx::Row;
16 :
17 : use super::entity::{check_ceiling, MAX_UNDECIDED_ROWS};
18 :
19 : /// Which doc table a resource kind lives in.
20 : #[derive(Clone, Copy, Debug)]
21 : pub enum DocKind {
22 : Subscription,
23 : Registration,
24 : CSourceSubscription,
25 : Snapshot,
26 : EntityMap,
27 : DistSub,
28 : DeadLetter,
29 : View,
30 : }
31 :
32 : impl DocKind {
33 607317 : fn table(self) -> &'static str {
34 607317 : match self {
35 140196 : DocKind::Subscription => "subscriptions",
36 173463 : DocKind::Registration => "csource_registrations",
37 14511 : DocKind::CSourceSubscription => "csource_subscriptions",
38 137312 : DocKind::Snapshot => "snapshots",
39 137504 : DocKind::EntityMap => "entity_map_docs",
40 4292 : DocKind::DistSub => "dist_subs",
41 13 : DocKind::DeadLetter => "dead_letters",
42 26 : DocKind::View => "views",
43 : }
44 607317 : }
45 605720 : fn doc_column(self) -> &'static str {
46 605720 : match self {
47 154264 : DocKind::Subscription | DocKind::CSourceSubscription => "subscription",
48 172947 : DocKind::Registration => "registration",
49 : DocKind::Snapshot
50 : | DocKind::EntityMap
51 : | DocKind::DistSub
52 : | DocKind::DeadLetter
53 278509 : | DocKind::View => "doc",
54 : }
55 605720 : }
56 47052 : fn has_bookkeeping(self) -> bool {
57 47052 : matches!(self, DocKind::Subscription | DocKind::CSourceSubscription)
58 47052 : }
59 : }
60 :
61 : /// Bookkeeping columns, derived from the doc (5.2.14.2 output members).
62 1164 : fn bookkeeping(
63 1164 : doc: &Value,
64 1164 : ) -> (
65 1164 : Option<String>,
66 1164 : bool,
67 1164 : i64,
68 1164 : Option<String>,
69 1164 : Option<String>,
70 1164 : Option<String>,
71 1164 : ) {
72 : // 4.6.3: a comma seconds-fraction is legal in a request; every one of
73 : // these becomes a `::timestamptz` bind.
74 4656 : let s = |v: Option<&Value>| {
75 4656 : v.and_then(Value::as_str)
76 4656 : .map(|t| antares_store::filter::canonical_datetime(t).into_owned())
77 4656 : };
78 1164 : let n = doc.get("notification");
79 : (
80 1164 : s(doc.get("expiresAt")),
81 1164 : doc.get("isActive") != Some(&Value::Bool(false)),
82 1164 : n.and_then(|n| n.get("timesSent"))
83 1164 : .and_then(Value::as_i64)
84 1164 : .unwrap_or(0),
85 1164 : s(n.and_then(|n| n.get("lastNotification"))),
86 1164 : s(n.and_then(|n| n.get("lastSuccess"))),
87 1164 : s(n.and_then(|n| n.get("lastFailure"))),
88 : )
89 1164 : }
90 :
91 : // ---- csource_index maintenance ---------------------------------------------
92 : // The flattened federation match table, rebuilt in Rust inside the same
93 : // transaction as every registration write (no triggers). Deleting a
94 : // registration cleans its rows via the FK ON DELETE CASCADE.
95 :
96 : // A registration's operations are stored as a bitmask over Table 4.20-1
97 : // (`antares_model::operations::OPERATION_NAMES`), bit position = index in
98 : // that list. The list's order is append-only for exactly this reason: a
99 : // bitmask is stored data, so renumbering it is a migration, not an edit.
100 : //
101 : // The bit position being stored data also means appending a 64th operation
102 : // would overflow the shift below and write a mask no later migration could
103 : // distinguish from a real one. Fail at compile time on the day it is
104 : // appended, not on a corrupted row later.
105 : const _: () = assert!(OPERATION_NAMES.len() < 64, "ops bitmask is i64");
106 :
107 : /// Registration `operations` → bitmask; absent defaults to federationOps
108 : /// (5.2.9).
109 63052 : pub fn ops_mask(reg: &Value) -> i64 {
110 63052 : let names: Vec<&str> = reg
111 63052 : .get("operations")
112 63052 : .and_then(Value::as_array)
113 63052 : .map(|a| a.iter().filter_map(Value::as_str).collect())
114 63052 : .unwrap_or_else(|| vec![DEFAULT_OPERATION_GROUP]);
115 63052 : let mut mask = 0i64;
116 1201550 : let mut set = |op: &str| {
117 39066214 : if let Some(bit) = OPERATION_NAMES.iter().position(|o| *o == op) {
118 1201548 : mask |= 1 << bit;
119 1201548 : }
120 1201550 : };
121 64910 : for n in names {
122 64910 : match group_members(n) {
123 1199608 : Some(members) => members.iter().for_each(|m| set(m)),
124 1942 : None => set(n),
125 : }
126 : }
127 63052 : mask
128 63052 : }
129 :
130 63042 : fn mode_code(reg: &Value) -> i16 {
131 63042 : match reg.get("mode").and_then(Value::as_str) {
132 800 : Some("auxiliary") => 0,
133 784 : Some("redirect") => 2,
134 434 : Some("exclusive") => 3,
135 62396 : _ => 1, // inclusive is the default (5.2.9)
136 : }
137 63042 : }
138 :
139 : /// Hard ceiling on rows one registration may explode into. The API caps
140 : /// cardinality at the validation boundary, but a document written
141 : /// through any other path — a restored dump, a future importer — must not be
142 : /// able to drive this quadratically. Truncating loses federation matches for
143 : /// an absurd registration; OOM loses the process.
144 : pub const MAX_INDEX_ROWS: usize = 10_000;
145 :
146 : /// Explode one registration document into csource_index rows: each
147 : /// RegistrationInfo element yields entities × (propertyNames ∪
148 : /// relationshipNames) rows, with NULL placeholders when a dimension is
149 : /// absent — the Scorpio csourceinformation shape minus the 46
150 : /// boolean columns. Attribute/type names are stored as they appear in the
151 : /// document; canonical-IRI storage lands with the SQL matching path.
152 : /// NGSI-LD 2.0 readiness: when propertyNames/relationshipNames
153 : /// merge into attributeNames, the migration is a coalesce of the two name
154 : /// columns into one attribute_name column — no reshape.
155 22270 : pub fn index_rows(reg: &Value) -> Vec<Value> {
156 22270 : let endpoint = reg
157 22270 : .get("endpoint")
158 22270 : .and_then(Value::as_str)
159 22270 : .unwrap_or_default();
160 63042 : let common = |entity: Option<&Value>, prop: Option<&str>, rel: Option<&str>| {
161 63042 : serde_json::json!({
162 63042 : "entity_id": entity.and_then(|e| e.get("id")).and_then(Value::as_str),
163 63042 : "id_pattern": entity.and_then(|e| e.get("idPattern")).and_then(Value::as_str),
164 63042 : "entity_type": entity.and_then(|e| e.get("type")).and_then(Value::as_str),
165 63042 : "property_name": prop,
166 63042 : "relationship_name": rel,
167 : // A registration carries its geo scope as a RAW GeoJSON
168 : // geometry under `location` (not instance-wrapped like an entity
169 : // attribute) — see antares_api::csource::csr_matches_subscription,
170 : // which hands exactly this value to `matches_geometry`.
171 63042 : "location": reg.get("location").filter(|g| g.get("type").is_some())
172 63042 : .map(|g| g.to_string()),
173 63042 : "scopes": reg.get("scope").map(|s| match s {
174 4 : Value::String(one) => vec![one.clone()],
175 0 : Value::Array(a) => a.iter().filter_map(Value::as_str).map(str::to_owned).collect(),
176 0 : _ => vec![],
177 4 : }),
178 : // 4.6.3 comma fraction: this row's `expires_at` is cast with a
179 : // bare `::timestamptz` in the INSERT below.
180 63042 : "expires_at": reg.get("expiresAt").and_then(Value::as_str)
181 63042 : .map(antares_store::filter::canonical_datetime),
182 63042 : "endpoint": endpoint,
183 63042 : "mode": mode_code(reg),
184 63042 : "ops": ops_mask(reg),
185 63042 : "tenant_at_peer": reg.get("tenant").and_then(Value::as_str),
186 63042 : "headers": reg.get("contextSourceInfo"),
187 : // Table 5.2.9-1 names this member `contextSourceAlias` — the
188 : // peer's tenant-specific loop pseudonym. (`hostAlias` is the
189 : // prose spelling 6.3.18 uses and the csource_index column name;
190 : // it is not a payload member and was never sent by any client.)
191 63042 : "host_alias": reg.get("contextSourceAlias").and_then(Value::as_str),
192 : })
193 63042 : };
194 22270 : let mut rows = Vec::new();
195 22270 : let infos = reg
196 22270 : .get("information")
197 22270 : .and_then(Value::as_array)
198 22270 : .cloned()
199 22270 : .unwrap_or_default();
200 22270 : for info in &infos {
201 4670 : let names = |k: &str| -> Vec<String> {
202 4670 : info.get(k)
203 4670 : .and_then(Value::as_array)
204 4670 : .map(|a| {
205 692 : a.iter()
206 692 : .filter_map(Value::as_str)
207 692 : .map(str::to_owned)
208 692 : .collect()
209 692 : })
210 4670 : .unwrap_or_default()
211 4670 : };
212 : // TS 104 175 9.4.2: attributeNames, else the deprecated pair. The
213 : // table has no column for a name of either kind, so an
214 : // attributeNames entry is indexed as a property_name: the index is
215 : // a flattened copy, the matchers read the registration itself.
216 2336 : let (props, rels) = match info.get("attributeNames") {
217 2 : Some(_) => (names("attributeNames"), Vec::new()),
218 2334 : None => (names("propertyNames"), names("relationshipNames")),
219 : };
220 2336 : let entities: Vec<Option<&Value>> = match info.get("entities").and_then(Value::as_array) {
221 2332 : Some(a) if !a.is_empty() => a.iter().map(Some).collect(),
222 4 : _ => vec![None],
223 : };
224 22478 : for ent in entities {
225 : // Checked per ROW, not per entity: one `information` element with
226 : // one entity and a million propertyNames never reaches a
227 : // per-entity check twice, so the vector grew unbounded — the OOM
228 : // this ceiling exists to prevent.
229 43040 : for (p, r) in std::iter::once((None, None))
230 22478 : .filter(|_| props.is_empty() && rels.is_empty())
231 22478 : .chain(props.iter().map(|p| (Some(p.as_str()), None)))
232 22478 : .chain(rels.iter().map(|r| (None, Some(r.as_str()))))
233 : {
234 43040 : if rows.len() >= MAX_INDEX_ROWS {
235 4 : tracing::warn!(
236 : "registration explodes past {MAX_INDEX_ROWS} index rows; truncating"
237 : );
238 4 : return rows;
239 43036 : }
240 43036 : rows.push(common(ent, p, r));
241 : }
242 : }
243 : }
244 22266 : if infos.is_empty() {
245 20006 : rows.push(common(None, None, None));
246 20006 : }
247 22266 : rows
248 22270 : }
249 :
250 : /// Rebuild one registration's `csource_index` rows (delete + multi-row
251 : /// insert) inside the CALLER's transaction, so the extracted match rows are
252 : /// never a version behind the document. Shared by `upsert` and `mutate` —
253 : /// the row lock lives in here, so no caller can do it atomically itself.
254 : ///
255 : /// The geometry goes through `try_geomfromgeojson` (0001_init.sql): a
256 : /// location PostGIS cannot parse leaves the column NULL instead of aborting
257 : /// the write.
258 21416 : async fn rebuild_csource_index(
259 21416 : tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
260 21416 : tenant: &TenantId,
261 21416 : id: &str,
262 21416 : doc: &Value,
263 21416 : ) -> Result<(), sqlx::Error> {
264 21416 : sqlx::query("DELETE FROM csource_index WHERE tenant_id = $1 AND registration_id = $2")
265 21416 : .bind(tenant.as_str())
266 21416 : .bind(id)
267 21416 : .execute(&mut **tx)
268 21416 : .await?;
269 21416 : sqlx::query(
270 21416 : "INSERT INTO csource_index
271 21416 : (tenant_id, registration_id, entity_id, id_pattern, entity_type,
272 21416 : property_name, relationship_name, scopes, expires_at, endpoint,
273 21416 : mode, ops, tenant_at_peer, headers, host_alias, location)
274 21416 : SELECT $1, $2, e->>'entity_id', e->>'id_pattern', e->>'entity_type',
275 21416 : e->>'property_name', e->>'relationship_name',
276 21416 : CASE WHEN e->'scopes' = 'null'::jsonb THEN NULL
277 21416 : ELSE ARRAY(SELECT jsonb_array_elements_text(e->'scopes')) END,
278 21416 : (e->>'expires_at')::timestamptz, e->>'endpoint',
279 21416 : (e->>'mode')::smallint, (e->>'ops')::bigint,
280 21416 : e->>'tenant_at_peer', e->'headers', e->>'host_alias',
281 21416 : CASE WHEN ST_IsValid(try_geomfromgeojson(e->>'location'))
282 21416 : THEN try_geomfromgeojson(e->>'location') END
283 21416 : FROM jsonb_array_elements($3::jsonb) AS e",
284 21416 : )
285 21416 : .bind(tenant.as_str())
286 21416 : .bind(id)
287 21416 : .bind(Value::Array(index_rows(doc)))
288 21416 : .execute(&mut **tx)
289 21416 : .await
290 21416 : .map(|_| ())
291 21416 : }
292 :
293 : /// The INSERT every doc write shares — same columns, same binds, differing
294 : /// only in the caller's `ON CONFLICT …` tail. `None` = the tail took a
295 : /// DO NOTHING path (the row was already there).
296 22732 : async fn insert_doc(
297 22732 : tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
298 22732 : tenant: &TenantId,
299 22732 : kind: DocKind,
300 22732 : id: &str,
301 22732 : doc: &Value,
302 22732 : conflict: &str,
303 22732 : ) -> Result<Option<bool>, sqlx::Error> {
304 22732 : let table = kind.table();
305 22732 : let col = kind.doc_column();
306 22732 : let head = if kind.has_bookkeeping() {
307 631 : format!(
308 : "INSERT INTO {table} (tenant_id, id, {col}, context, expires_at, is_active,
309 : times_sent, last_notification, last_success, last_failure)
310 : VALUES ($1, $2, $3, $4, $5::timestamptz, $6, $7,
311 : $8::timestamptz, $9::timestamptz, $10::timestamptz)"
312 : )
313 : } else {
314 22101 : format!("INSERT INTO {table} (tenant_id, id, {col}) VALUES ($1, $2, $3)")
315 : };
316 : // literals from `DocKind` plus the caller's literal tail, values bound
317 22732 : let mut q = sqlx::query(sqlx::AssertSqlSafe(format!("{head}{conflict}")))
318 22732 : .bind(tenant.as_str())
319 22732 : .bind(id)
320 22732 : .bind(doc);
321 : let bk;
322 22732 : if kind.has_bookkeeping() {
323 631 : let context = doc
324 631 : .get("@context")
325 631 : .cloned()
326 631 : .unwrap_or(Value::Object(Default::default()));
327 631 : bk = (bookkeeping(doc), context);
328 631 : let ((expires, active, sent, last_n, last_s, last_f), context) = &bk;
329 631 : q = q
330 631 : .bind(context)
331 631 : .bind(expires)
332 631 : .bind(*active)
333 631 : .bind(*sent)
334 631 : .bind(last_n)
335 631 : .bind(last_s)
336 631 : .bind(last_f);
337 22101 : }
338 22732 : Ok(q.fetch_optional(&mut **tx)
339 22732 : .await?
340 22732 : .map(|r| r.get::<bool, _>(0)))
341 22732 : }
342 :
343 : /// The row a sealed credential is bound to ([`super::seal::Row`]).
344 10826633 : fn seal_row<'a>(kind: DocKind, tenant: &'a TenantId, id: &'a str) -> super::seal::Row<'a> {
345 10826633 : super::seal::Row {
346 10826633 : kind,
347 10826633 : tenant: tenant.as_str(),
348 10826633 : id,
349 10826633 : }
350 10826633 : }
351 :
352 : pub struct PgDocStore {
353 : pool: PgPool,
354 : /// `ANTARES_SECRET_KEY`: seals the notification credentials.
355 : key: Option<std::sync::Arc<super::seal::SecretKey>>,
356 : }
357 :
358 : /// Arm the Row-Level Security policy on `jsonld_contexts` for one call.
359 : ///
360 : /// `antares.tenant` is transaction-scoped, so it is set inside the caller's
361 : /// own transaction. `None` leaves it unset, and `current_setting` then
362 : /// returns NULL: the policy's `tenant_id = current_setting(...)` arm is NULL
363 : /// for every row and only `tenant_id IS NULL` — the Cached rows — remains.
364 : /// That is what the boot warm and the Cached write-through want.
365 11274 : async fn set_context_tenant(
366 11274 : tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
367 11274 : tenant: Option<&TenantId>,
368 11274 : ) -> Result<(), sqlx::Error> {
369 11274 : match tenant {
370 11236 : Some(t) => super::set_tenant(tx, t).await,
371 38 : None => Ok(()),
372 : }
373 11274 : }
374 :
375 : impl PgDocStore {
376 142 : pub fn new(pool: PgPool) -> Self {
377 142 : Self { pool, key: None }
378 142 : }
379 :
380 : /// Seal every notification credential this store writes under `key`
381 : /// ([`super::seal`]).
382 10 : pub fn with_secret_key(mut self, key: super::seal::SecretKey) -> Self {
383 10 : self.key = Some(std::sync::Arc::new(key));
384 10 : self
385 10 : }
386 :
387 10803368 : fn opened(
388 10803368 : &self,
389 10803368 : kind: DocKind,
390 10803368 : tenant: &TenantId,
391 10803368 : id: &str,
392 10803368 : mut doc: Value,
393 10803368 : ) -> Result<Value, sqlx::Error> {
394 10803368 : super::seal::open(self.key.as_deref(), seal_row(kind, tenant, id), &mut doc)?;
395 10803365 : Ok(doc)
396 10803368 : }
397 :
398 23265 : fn sealed<'a>(
399 23265 : &self,
400 23265 : kind: DocKind,
401 23265 : tenant: &TenantId,
402 23265 : id: &str,
403 23265 : doc: &'a Value,
404 23265 : ) -> Result<std::borrow::Cow<'a, Value>, sqlx::Error> {
405 23265 : Ok(super::seal::sealed(
406 23265 : self.key.as_deref(),
407 23265 : seal_row(kind, tenant, id),
408 23265 : doc,
409 0 : )?)
410 23265 : }
411 :
412 : /// Rows of `SELECT {doc}, id`, each credential opened for its own row.
413 577390 : fn opened_all(
414 577390 : &self,
415 577390 : kind: DocKind,
416 577390 : tenant: &TenantId,
417 577390 : rows: Vec<sqlx::postgres::PgRow>,
418 577390 : ) -> Result<Vec<Value>, sqlx::Error> {
419 577390 : rows.into_iter()
420 10800040 : .map(|r| self.opened(kind, tenant, &r.get::<String, _>(1), r.get::<Value, _>(0)))
421 577390 : .collect()
422 577390 : }
423 :
424 2091 : pub fn pool(&self) -> &PgPool {
425 2091 : &self.pool
426 2091 : }
427 :
428 : /// Create one doc: `Ok(false)` when a document with that id already
429 : /// exists. 5.8.1.4 (and 5.9.2.4 for registrations): "If the NGSI-LD
430 : /// endpoint already knows about this Subscription, as there is an
431 : /// existing Subscription whose id (URI) is equivalent, an error of type
432 : /// AlreadyExists shall be raised."
433 : ///
434 : /// ONE statement, so the answer comes from the unique constraint itself:
435 : /// a read-then-write would let two concurrent creates of the same
436 : /// client-supplied id both report created, and the second would silently
437 : /// overwrite the first.
438 22582 : pub async fn create(
439 22582 : &self,
440 22582 : tenant: &TenantId,
441 22582 : kind: DocKind,
442 22582 : id: &str,
443 22582 : doc: &Value,
444 22582 : ) -> Result<bool, sqlx::Error> {
445 22582 : let conflict = " ON CONFLICT (tenant_id, id) DO NOTHING RETURNING true AS created";
446 22582 : let mut tx = super::begin(&self.pool).await?;
447 22582 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
448 22582 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
449 22582 : let stored = self.sealed(kind, tenant, id, doc)?;
450 22582 : let created = insert_doc(&mut tx, tenant, kind, id, &stored, conflict)
451 22582 : .await?
452 22582 : .is_some();
453 : // a losing INSERT must not rebuild the winner's index rows
454 22582 : if created && matches!(kind, DocKind::Registration) {
455 21380 : rebuild_csource_index(&mut tx, tenant, id, doc).await?;
456 1202 : }
457 22582 : tx.commit().await?;
458 22582 : Ok(created)
459 22582 : }
460 :
461 : /// Upsert one doc, refreshing the extracted columns. `Ok(true)` = it
462 : /// existed before.
463 150 : pub async fn upsert(
464 150 : &self,
465 150 : tenant: &TenantId,
466 150 : kind: DocKind,
467 150 : id: &str,
468 150 : doc: &Value,
469 150 : ) -> Result<bool, sqlx::Error> {
470 150 : let col = kind.doc_column();
471 150 : let conflict = if kind.has_bookkeeping() {
472 108 : format!(
473 : " ON CONFLICT (tenant_id, id) DO UPDATE SET {col} = EXCLUDED.{col},
474 : context = EXCLUDED.context, expires_at = EXCLUDED.expires_at,
475 : is_active = EXCLUDED.is_active, times_sent = EXCLUDED.times_sent,
476 : last_notification = EXCLUDED.last_notification,
477 : last_success = EXCLUDED.last_success, last_failure = EXCLUDED.last_failure
478 : RETURNING (xmax <> 0) AS existed"
479 : )
480 : } else {
481 42 : format!(
482 : " ON CONFLICT (tenant_id, id) DO UPDATE SET {col} = EXCLUDED.{col}
483 : RETURNING (xmax <> 0) AS existed"
484 : )
485 : };
486 150 : let mut tx = super::begin(&self.pool).await?;
487 150 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
488 150 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
489 : // INSERT … ON CONFLICT DO UPDATE … RETURNING always answers with
490 : // the row; no row means the statement stopped being an upsert.
491 150 : let stored = self.sealed(kind, tenant, id, doc)?;
492 150 : let existed = insert_doc(&mut tx, tenant, kind, id, &stored, &conflict)
493 150 : .await?
494 150 : .ok_or(sqlx::Error::RowNotFound)?;
495 150 : if matches!(kind, DocKind::Registration) {
496 26 : rebuild_csource_index(&mut tx, tenant, id, doc).await?;
497 124 : }
498 150 : tx.commit().await?;
499 150 : Ok(existed)
500 150 : }
501 :
502 4221 : pub async fn get(
503 4221 : &self,
504 4221 : tenant: &TenantId,
505 4221 : kind: DocKind,
506 4221 : id: &str,
507 4221 : ) -> Result<Option<Value>, sqlx::Error> {
508 4221 : let sql = format!(
509 : "SELECT {} FROM {} WHERE tenant_id = $1 AND id = $2",
510 4221 : kind.doc_column(),
511 4221 : kind.table()
512 : );
513 4221 : let mut tx = super::begin(&self.pool).await?;
514 4221 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
515 4221 : let row = sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
516 4221 : .bind(tenant.as_str())
517 4221 : .bind(id)
518 4221 : .fetch_optional(&mut *tx)
519 4221 : .await?;
520 4221 : tx.commit().await?;
521 4221 : row.map(|r| self.opened(kind, tenant, id, r.get::<Value, _>(0)))
522 4221 : .transpose()
523 4221 : }
524 :
525 : /// Read-modify-write in ONE transaction under the row lock (the entity
526 : /// pattern applied to doc kinds): `SELECT … FOR UPDATE` → apply → `UPDATE`.
527 : /// A missing row returns `None` and is NEVER inserted — a concurrent
528 : /// DELETE must win, not be resurrected by a bookkeeping writeback
529 : /// (the 047_06 leftover-subscription bug).
530 : /// 5.2.14.2 delivery bookkeeping as ONE statement.
531 : ///
532 : /// The generic `mutate` takes an arbitrary closure, so it has to hold the
533 : /// row under `FOR UPDATE` across a network round trip while Rust runs.
534 : /// This mutation is fixed, so it can be expressed in SQL and the lock
535 : /// lives only as long as the UPDATE. That is the point: at fan-out every
536 : /// delivery on one subscription contends for the same row, so the lock
537 : /// hold time is what serializes delivery, not the statement count.
538 : ///
539 : /// The pre-image comes back through a locking sub-select, not a plain
540 : /// CTE. Both are read once per statement, but a CTE is answered from the
541 : /// snapshot the statement started with: when another attempt on the same
542 : /// subscription commits while this one waits for the row, the CTE hands
543 : /// back the value from BEFORE that commit, and the rollback a failed
544 : /// attempt performs then rewinds `lastSuccess` past a delivery that
545 : /// succeeded. `FOR UPDATE` waits for the same row lock the UPDATE needs
546 : /// and re-reads the row it actually gets, so the pre-image is the value
547 : /// this statement overwrote.
548 : ///
549 : /// `notification` is a mandatory member (5.2.12) and `jsonb_set` on a
550 : /// path with no parent is a no-op, so a document that somehow lacks it is
551 : /// left alone rather than grown a synthetic one.
552 247 : pub async fn record_delivery(
553 247 : &self,
554 247 : tenant: &TenantId,
555 247 : kind: DocKind,
556 247 : id: &str,
557 247 : now: &str,
558 247 : ) -> Result<Option<(Value, Option<Value>)>, sqlx::Error> {
559 247 : let mut found = self.record_deliveries(tenant, kind, &[(id, now)]).await?;
560 247 : Ok(found.remove(id))
561 247 : }
562 :
563 : /// [`Self::record_delivery`] for distinct subscriptions in ONE statement:
564 : /// the rows are locked in id order, so two batches over
565 : /// overlapping subscriptions queue behind each other instead of
566 : /// deadlocking, and each pre-image is still the value its own row lock
567 : /// re-read. A subscription that is gone has no entry in the answer.
568 257 : pub async fn record_deliveries(
569 257 : &self,
570 257 : tenant: &TenantId,
571 257 : kind: DocKind,
572 257 : attempts: &[(&str, &str)],
573 257 : ) -> Result<std::collections::HashMap<String, (Value, Option<Value>)>, sqlx::Error> {
574 257 : if attempts.is_empty() {
575 0 : return Ok(std::collections::HashMap::new());
576 257 : }
577 257 : let table = kind.table();
578 257 : let col = kind.doc_column();
579 257 : let (ids, stamps): (Vec<&str>, Vec<&str>) = attempts.iter().copied().unzip();
580 : // literals from DocKind; every value is bound
581 257 : let sql = format!(
582 : "UPDATE {table} AS d SET
583 : {col} = jsonb_set(jsonb_set(jsonb_set(jsonb_set(
584 : d.{col} - 'status',
585 : '{{notification,timesSent}}',
586 : to_jsonb(COALESCE((d.{col} #>> '{{notification,timesSent}}')::bigint, 0) + 1)),
587 : '{{notification,lastNotification}}', to_jsonb(prev.stamp)),
588 : '{{notification,lastSuccess}}', to_jsonb(prev.stamp)),
589 : '{{notification,status}}', '\"ok\"'::jsonb),
590 : times_sent = COALESCE(d.times_sent, 0) + 1,
591 : last_notification = prev.stamp::timestamptz,
592 : last_success = prev.stamp::timestamptz
593 : FROM (SELECT s.id AS locked_id,
594 : s.{col} #> '{{notification,lastSuccess}}' AS ls,
595 : a.stamp
596 : FROM {table} AS s
597 : JOIN unnest($2::text[], $3::text[]) AS a(id, stamp) ON s.id = a.id
598 : WHERE s.tenant_id = $1
599 : ORDER BY s.id
600 : FOR UPDATE OF s) AS prev
601 : WHERE d.tenant_id = $1 AND d.id = prev.locked_id
602 : RETURNING d.id, d.{col}, prev.ls"
603 : );
604 257 : let mut tx = super::begin(&self.pool).await?;
605 257 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
606 257 : let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
607 257 : .bind(tenant.as_str())
608 257 : .bind(&ids)
609 257 : .bind(&stamps)
610 257 : .fetch_all(&mut *tx)
611 257 : .await?;
612 257 : tx.commit().await?;
613 257 : rows.into_iter()
614 347 : .map(|r| {
615 347 : let id = r.get::<String, _>(0);
616 347 : let doc = self.opened(kind, tenant, &id, r.get(1))?;
617 347 : Ok((id, (doc, r.get(2))))
618 347 : })
619 257 : .collect()
620 257 : }
621 :
622 905 : pub async fn mutate<T, E>(
623 905 : &self,
624 905 : tenant: &TenantId,
625 905 : kind: DocKind,
626 905 : id: &str,
627 905 : f: impl FnOnce(&mut Value) -> Result<T, E>,
628 905 : ) -> Result<Option<Result<T, E>>, sqlx::Error> {
629 905 : let col = kind.doc_column();
630 905 : let table = kind.table();
631 905 : let select =
632 905 : format!("SELECT {col} FROM {table} WHERE tenant_id = $1 AND id = $2 FOR UPDATE");
633 905 : let update = if kind.has_bookkeeping() {
634 70 : format!(
635 : "UPDATE {table} SET {col} = $3, expires_at = $4::timestamptz,
636 : is_active = $5, times_sent = $6, last_notification = $7::timestamptz,
637 : last_success = $8::timestamptz, last_failure = $9::timestamptz,
638 : context = $10
639 : WHERE tenant_id = $1 AND id = $2"
640 : )
641 : } else {
642 835 : format!("UPDATE {table} SET {col} = $3 WHERE tenant_id = $1 AND id = $2")
643 : };
644 905 : async move {
645 905 : let mut tx = super::begin(&self.pool).await?;
646 905 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
647 905 : let row = sqlx::query(sqlx::AssertSqlSafe(select.clone()))
648 905 : .bind(tenant.as_str())
649 905 : .bind(id)
650 905 : .fetch_optional(&mut *tx)
651 905 : .await?;
652 905 : let Some(row) = row else {
653 372 : return Ok(None);
654 : };
655 533 : let mut doc = self.opened(kind, tenant, id, row.get(0))?;
656 : // The match rows are a pure function of the document, so the
657 : // rows this write would produce can be compared with the ones
658 : // already stored before deciding to touch them.
659 533 : let index_before = matches!(kind, DocKind::Registration).then(|| index_rows(&doc));
660 533 : match f(&mut doc) {
661 533 : Ok(t) => {
662 533 : let context = doc
663 533 : .get("@context")
664 533 : .cloned()
665 533 : .unwrap_or(Value::Object(Default::default()));
666 533 : let (expires, active, sent, last_n, last_s, last_f) = bookkeeping(&doc);
667 533 : let stored = self.sealed(kind, tenant, id, &doc)?;
668 533 : let mut q = sqlx::query(sqlx::AssertSqlSafe(update.clone()))
669 533 : .bind(tenant.as_str())
670 533 : .bind(id)
671 533 : .bind(stored.as_ref());
672 533 : if kind.has_bookkeeping() {
673 46 : q = q
674 46 : .bind(&expires)
675 46 : .bind(active)
676 46 : .bind(sent)
677 46 : .bind(&last_n)
678 46 : .bind(&last_s)
679 46 : .bind(&last_f)
680 46 : .bind(&context);
681 487 : }
682 533 : q.execute(&mut *tx).await?;
683 : // 5.9.3 Update Registration: a patch may flip the mode,
684 : // rewrite `information` or move the endpoint, so the
685 : // extracted match rows have to be rebuilt with the doc —
686 : // in this transaction, under the same row lock. A write
687 : // that moves none of them rebuilds nothing: Table
688 : // 5.2.9-2's forward counters are booked on every
689 : // distributed operation, and the delete + re-insert
690 : // those would drag along rewrites the whole index of a
691 : // busy registration for a document the matcher reads
692 : // identically.
693 533 : if index_before.is_some_and(|before| before != index_rows(&doc)) {
694 10 : rebuild_csource_index(&mut tx, tenant, id, &doc).await?;
695 523 : }
696 533 : tx.commit().await?;
697 533 : Ok(Some(Ok(t)))
698 : }
699 : // closure rejected the change: nothing written, lock released
700 0 : Err(e) => Ok(Some(Err(e))),
701 : }
702 905 : }
703 905 : .await
704 905 : }
705 :
706 1652 : pub async fn delete(
707 1652 : &self,
708 1652 : tenant: &TenantId,
709 1652 : kind: DocKind,
710 1652 : id: &str,
711 1652 : ) -> Result<bool, sqlx::Error> {
712 1652 : let sql = format!(
713 : "DELETE FROM {} WHERE tenant_id = $1 AND id = $2",
714 1652 : kind.table()
715 : );
716 1652 : let mut tx = super::begin(&self.pool).await?;
717 1652 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
718 1652 : let done = sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
719 1652 : .bind(tenant.as_str())
720 1652 : .bind(id)
721 1652 : .execute(&mut *tx)
722 1652 : .await?
723 1652 : .rows_affected();
724 1652 : tx.commit().await?;
725 1652 : Ok(done == 1)
726 1652 : }
727 :
728 : /// One id-ordered page of docs: ids strictly greater than `after`, at
729 : /// most `limit`.
730 : ///
731 : /// No ceiling, deliberately. `list`'s `MAX_UNDECIDED_ROWS` exists so a
732 : /// large tenant cannot be materialized into one `Vec`; a page bounds
733 : /// that by construction, so refusing here would only break the readers
734 : /// that must see every row — and one tenant's stored volume would
735 : /// decide whether another tenant's subscriptions are ever matched.
736 : /// Keyset, not OFFSET: the walk runs against a table being written to.
737 : /// `after = None` is bound as the empty string rather than as a second
738 : /// statement, which keeps the primary-key range scan; it is below every
739 : /// id because a document id is a URI and a URI is never empty. A row
740 : /// stored with an empty id would be returned by `list` and skipped by
741 : /// every page of this walk.
742 577256 : pub async fn list_page(
743 577256 : &self,
744 577256 : tenant: &TenantId,
745 577256 : kind: DocKind,
746 577256 : after: Option<&str>,
747 577256 : limit: i64,
748 577256 : ) -> Result<Vec<Value>, sqlx::Error> {
749 577256 : let sql = format!(
750 : "SELECT {}, id FROM {} WHERE tenant_id = $1 AND id > $2 ORDER BY id LIMIT $3",
751 577256 : kind.doc_column(),
752 577256 : kind.table()
753 : );
754 577256 : let mut tx = super::begin(&self.pool).await?;
755 577207 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
756 577207 : let rows = sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
757 577207 : .bind(tenant.as_str())
758 577207 : .bind(after.unwrap_or(""))
759 577207 : .bind(limit)
760 577207 : .fetch_all(&mut *tx)
761 577207 : .await?;
762 577207 : tx.commit().await?;
763 577207 : self.opened_all(kind, tenant, rows)
764 577256 : }
765 :
766 : /// One id-ordered window of docs and the size of the whole set:
767 : /// elements `offset..offset + limit`, plus the count 6.3.10 reports.
768 : ///
769 : /// No ceiling, and no `check_ceiling`, for the same reason `list_page`
770 : /// has none: the window bounds the result by construction, so refusing
771 : /// here would only refuse a page the client is entitled to. Both
772 : /// statements run in one transaction under the same `set_tenant`, so
773 : /// the count and the page describe the same set — a count taken outside
774 : /// it could report a total no page of this read ever adds up to.
775 85 : pub async fn list_slice(
776 85 : &self,
777 85 : tenant: &TenantId,
778 85 : kind: DocKind,
779 85 : offset: i64,
780 85 : limit: i64,
781 85 : ) -> Result<(Vec<Value>, i64), sqlx::Error> {
782 85 : let page_sql = format!(
783 : "SELECT {}, id FROM {} WHERE tenant_id = $1 ORDER BY id LIMIT $2 OFFSET $3",
784 85 : kind.doc_column(),
785 85 : kind.table()
786 : );
787 85 : let count_sql = format!("SELECT count(*) FROM {} WHERE tenant_id = $1", kind.table());
788 85 : let mut tx = super::begin(&self.pool).await?;
789 85 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
790 85 : let total: i64 = sqlx::query_scalar(sqlx::AssertSqlSafe(count_sql))
791 85 : .bind(tenant.as_str())
792 85 : .fetch_one(&mut *tx)
793 85 : .await?;
794 85 : let rows = sqlx::query(sqlx::AssertSqlSafe(page_sql))
795 85 : .bind(tenant.as_str())
796 85 : .bind(limit)
797 85 : .bind(offset)
798 85 : .fetch_all(&mut *tx)
799 85 : .await?;
800 85 : tx.commit().await?;
801 85 : Ok((self.opened_all(kind, tenant, rows)?, total))
802 85 : }
803 :
804 : /// Every doc's id and the first 64 bits of the md5 of its text, under
805 : /// `list`'s ceiling: the bodies stay in the database. `views`
806 : /// keeps the digest in its generated column (migration 0026);
807 : /// the other tables compute it here.
808 8 : pub async fn list_digests(
809 8 : &self,
810 8 : tenant: &TenantId,
811 8 : kind: DocKind,
812 8 : ) -> Result<Vec<(String, u64)>, sqlx::Error> {
813 8 : let select = match kind {
814 8 : DocKind::View => "id, digest".to_owned(),
815 0 : _ => format!("id, left(md5({}::text), 16)", kind.doc_column()),
816 : };
817 8 : self.every_row(tenant, kind, &select)
818 8 : .await?
819 8 : .into_iter()
820 14 : .map(|r| {
821 14 : let hex: String = r.get(1);
822 14 : u64::from_str_radix(&hex, 16)
823 14 : .map(|d| (r.get::<String, _>(0), d))
824 14 : .map_err(|e| sqlx::Error::Decode(Box::new(e)))
825 14 : })
826 8 : .collect()
827 8 : }
828 :
829 : /// Every doc of one kind for one tenant, id-ordered.
830 98 : pub async fn list(&self, tenant: &TenantId, kind: DocKind) -> Result<Vec<Value>, sqlx::Error> {
831 98 : let select = format!("{}, id", kind.doc_column());
832 98 : let rows = self.every_row(tenant, kind, &select).await?;
833 98 : self.opened_all(kind, tenant, rows)
834 98 : }
835 :
836 : /// Seal every notification credential still stored in plaintext (rows
837 : /// written before `ANTARES_SECRET_KEY` was set). Run at boot;
838 : /// without a key it seals nothing. Each row is rewritten through
839 : /// [`Self::mutate`], under its row lock, so a concurrent write is never
840 : /// lost. Answers how many rows it sealed.
841 5 : pub async fn seal_plaintext(&self) -> Result<u64, sqlx::Error> {
842 5 : if self.key.is_none() {
843 1 : return Ok(0);
844 4 : }
845 : // every tenant row, the broker's own included, and the default
846 : // Tenant, which exists without one (5.5.10)
847 4 : let mut tenants: Vec<String> = sqlx::query_scalar("SELECT tenant_id FROM tenants")
848 4 : .fetch_all(&self.pool)
849 4 : .await?;
850 4 : if !tenants.iter().any(|t| t == TenantId::DEFAULT) {
851 0 : tenants.push(TenantId::DEFAULT.to_owned());
852 4 : }
853 4 : let mut sealed = 0u64;
854 16 : for kind in [
855 4 : DocKind::Subscription,
856 4 : DocKind::CSourceSubscription,
857 4 : DocKind::Snapshot,
858 4 : DocKind::DeadLetter,
859 4 : ] {
860 16 : let Some(path) = super::seal::plaintext_path(kind) else {
861 0 : continue;
862 : };
863 16 : let sql = format!(
864 : "SELECT id FROM {} WHERE tenant_id = $1 AND jsonb_path_exists({}, $2::jsonpath)
865 : ORDER BY id",
866 16 : kind.table(),
867 16 : kind.doc_column()
868 : );
869 2276 : for raw in &tenants {
870 2276 : let tenant = TenantId::try_from(raw.clone())
871 2276 : .map_err(|e| sqlx::Error::Configuration(Box::new(e)))?;
872 2276 : let mut tx = super::begin(&self.pool).await?;
873 2276 : crate::store::pg::set_tenant(&mut tx, &tenant).await?;
874 2276 : let ids: Vec<String> = sqlx::query_scalar(sqlx::AssertSqlSafe(sql.clone()))
875 2276 : .bind(tenant.as_str())
876 2276 : .bind(path)
877 2276 : .fetch_all(&mut *tx)
878 2276 : .await?;
879 2276 : tx.commit().await?;
880 2276 : for id in ids {
881 5 : let written = self
882 5 : .mutate(&tenant, kind, &id, |_| {
883 5 : Ok::<(), std::convert::Infallible>(())
884 5 : })
885 5 : .await?;
886 5 : sealed = sealed.saturating_add(u64::from(written.is_some()));
887 : }
888 : }
889 : }
890 4 : Ok(sealed)
891 5 : }
892 :
893 : /// `select` of every row of one kind for one tenant, id-ordered.
894 : ///
895 : /// Bounded: without a LIMIT this statement materializes a whole tenant's
896 : /// subscriptions/registrations into one `Vec`, which at the 100 000-per-
897 : /// broker target is the broker's memory, not the database's. A tenant that
898 : /// reaches the ceiling is refused with TooManyResults (5.5.6) rather than
899 : /// served a silent prefix.
900 106 : async fn every_row(
901 106 : &self,
902 106 : tenant: &TenantId,
903 106 : kind: DocKind,
904 106 : select: &str,
905 106 : ) -> Result<Vec<sqlx::postgres::PgRow>, sqlx::Error> {
906 106 : let sql = format!(
907 : "SELECT {select} FROM {} WHERE tenant_id = $1 ORDER BY id LIMIT $2",
908 106 : kind.table()
909 : );
910 106 : let mut tx = super::begin(&self.pool).await?;
911 106 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
912 106 : let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
913 106 : .bind(tenant.as_str())
914 106 : .bind(MAX_UNDECIDED_ROWS)
915 106 : .fetch_all(&mut *tx)
916 106 : .await?;
917 106 : tx.commit().await?;
918 106 : check_ceiling(false, rows.len(), MAX_UNDECIDED_ROWS)?;
919 106 : Ok(rows)
920 106 : }
921 :
922 : /// 5.12: the registrations that may take part in an operation on these
923 : /// entity ids / types, read through the `csource_index` rows every
924 : /// registration write maintains — the alternative is listing every
925 : /// registration document of the tenant and scanning it in Rust, which at
926 : /// the 100 000-registrations target is a full table read per federated
927 : /// request.
928 : ///
929 : /// The narrowing is one-directional, exactly like the entity pushdown: SQL
930 : /// may only REMOVE rows the caller's matcher would reject anyway, and the
931 : /// matcher stays the arbiter of every other 5.12 condition (csf, geo,
932 : /// intervals, datasetId, the Via chain, the idPattern regex). Hence an
933 : /// index dimension left NULL is unconstrained and always survives, and an
934 : /// `idPattern` row survives every id query. `None` means "do not narrow on
935 : /// that dimension". A registration whose explosion hit `MAX_INDEX_ROWS`
936 : /// carries only the rows that fit — the same truncation the index write
937 : /// already accepts.
938 : ///
939 : /// `types` must be EXPANDED plain type IRIs, because that is what the
940 : /// registration write stored (each EntityInfo `type` goes through
941 : /// `expand_key` before the index row is built). A term, or a 4.17 Entity
942 : /// Type Selection expression, matches no stored value and would narrow away
943 : /// registrations that do match — a caller holding either passes `None`.
944 : ///
945 : /// Bounded like every other read (5.5.6): a tenant whose candidate set
946 : /// reaches the ceiling is refused rather than served a silent prefix.
947 510 : pub async fn matching_registrations(
948 510 : &self,
949 510 : tenant: &TenantId,
950 510 : ids: Option<&[String]>,
951 510 : types: Option<&[String]>,
952 510 : ) -> Result<Vec<Value>, sqlx::Error> {
953 : // An absent dimension is OMITTED from the statement rather than bound
954 : // as NULL and escaped in SQL: `$2 IS NULL OR …` is unfoldable in a
955 : // GENERIC plan, and Postgres switches a repeatedly executed prepared
956 : // statement to one — measured as a sequential scan of csource_index,
957 : // exactly the read this function exists to avoid.
958 510 : let mut wheres = String::new();
959 510 : let mut n: usize = 1; // $1 = tenant_id
960 510 : if types.is_some() {
961 388 : n = n.saturating_add(1);
962 388 : wheres.push_str(&format!(
963 388 : " AND (x.entity_type IS NULL OR x.entity_type = ANY(${n}))"
964 388 : ));
965 388 : }
966 510 : if ids.is_some() {
967 438 : n = n.saturating_add(1);
968 438 : wheres.push_str(&format!(
969 438 : " AND (x.entity_id IS NULL OR x.id_pattern IS NOT NULL \
970 438 : OR x.entity_id = ANY(${n}))"
971 438 : ));
972 438 : }
973 : // literals from this function plus `$n` placeholders — no caller text
974 510 : let sql = format!(
975 : "SELECT DISTINCT r.id, r.registration
976 : FROM csource_registrations r
977 : JOIN csource_index x
978 : ON x.tenant_id = r.tenant_id AND x.registration_id = r.id
979 : WHERE r.tenant_id = $1{wheres}
980 : ORDER BY r.id LIMIT ${}",
981 510 : n.saturating_add(1)
982 : );
983 510 : let mut tx = super::begin(&self.pool).await?;
984 510 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
985 510 : let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.clone())).bind(tenant.as_str());
986 510 : if let Some(types) = types {
987 388 : q = q.bind(super::storable(types));
988 388 : }
989 510 : if let Some(ids) = ids {
990 438 : q = q.bind(super::storable(ids));
991 438 : }
992 510 : let rows = q.bind(MAX_UNDECIDED_ROWS).fetch_all(&mut *tx).await?;
993 510 : tx.commit().await?;
994 510 : check_ceiling(false, rows.len(), MAX_UNDECIDED_ROWS)?;
995 30766 : Ok(rows.into_iter().map(|r| r.get::<Value, _>(1)).collect())
996 510 : }
997 :
998 : /// Bookkeeping columns straight from the row (test hook: rows are truth).
999 : #[cfg(any(test, feature = "test-kit"))]
1000 2 : pub async fn status_row(
1001 2 : &self,
1002 2 : tenant: &TenantId,
1003 2 : kind: DocKind,
1004 2 : id: &str,
1005 2 : ) -> Result<Option<(bool, i64)>, sqlx::Error> {
1006 2 : let sql = format!(
1007 : "SELECT is_active, times_sent FROM {} WHERE tenant_id = $1 AND id = $2",
1008 2 : kind.table()
1009 : );
1010 2 : let mut tx = super::begin(&self.pool).await?;
1011 2 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1012 2 : let row = sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
1013 2 : .bind(tenant.as_str())
1014 2 : .bind(id)
1015 2 : .fetch_optional(&mut *tx)
1016 2 : .await?;
1017 2 : tx.commit().await?;
1018 2 : Ok(row.map(|r| (r.get(0), r.get(1))))
1019 2 : }
1020 :
1021 : // jsonldContexts — a `tenant_id` GENERATED from the row's kind and its
1022 : // "owner" member, under the Row-Level Security policy 0006 installs
1023 : // (ADR-0021). A Cached row's tenant is NULL and every Tenant reaches it;
1024 : // every other kind is reachable only by the Tenant that stored it.
1025 : //
1026 : // Both halves of the belt, as everywhere else in this store: the explicit
1027 : // `tenant_id IS NULL OR tenant_id = $n` predicate on every statement, and
1028 : // the policy behind it. The predicate is what holds under the roles that
1029 : // bypass RLS (a superuser, a BYPASSRLS role); the policy is what holds
1030 : // when a future statement forgets the predicate. The GUC is
1031 : // transaction-scoped, so each call below opens its own transaction and
1032 : // sets it — and leaves it unset for `tenant: None`, which is the
1033 : // policy's "rows belonging to no Tenant".
1034 : ///
1035 : /// 5.13.1: "Implementations shall periodically invalidate the 'Cached'
1036 : /// @contexts." A Cached row is written per distinct external URL a request
1037 : /// references — client-controlled input — and the broker warms every
1038 : /// stored row at startup, so an insert that pushes the cache past its
1039 : /// ceiling evicts the oldest Cached rows. `Hosted` and
1040 : /// `ImplicitlyCreated` rows are resources the broker serves on demand
1041 : /// (5.13.2, 5.13.4), not cache, and are never evicted.
1042 206 : pub async fn context_put(
1043 206 : &self,
1044 206 : tenant: Option<&TenantId>,
1045 206 : id: &str,
1046 206 : doc: &Value,
1047 206 : kind: &str,
1048 206 : ) -> Result<(), sqlx::Error> {
1049 206 : let mut tx = self.pool.begin().await?;
1050 206 : set_context_tenant(&mut tx, tenant).await?;
1051 : // `xmax` is zero on a fresh row and the locking transaction's id
1052 : // on the conflict path: the eviction then runs only when the table
1053 : // actually grew, never on a usage bump rewriting a row in place.
1054 : // The `WHERE` on the conflict path is the write half of the rule:
1055 : // a row another Tenant owns is not replaced, the statement returns
1056 : // no row, and `fetch_one` raises rather than silently doing
1057 : // nothing. Nothing in the broker reaches it — every id is minted
1058 : // by the caller — so it is a backstop, and it is the one the
1059 : // driver contract probes.
1060 206 : let inserted: bool = sqlx::query_scalar(
1061 206 : "INSERT INTO jsonld_contexts (id, body, kind) VALUES ($1, $2, $3)
1062 206 : ON CONFLICT (id) DO UPDATE SET body = EXCLUDED.body, kind = EXCLUDED.kind
1063 206 : WHERE jsonld_contexts.tenant_id IS NOT DISTINCT FROM $4
1064 206 : OR jsonld_contexts.tenant_id IS NULL
1065 206 : RETURNING xmax::text = '0'",
1066 206 : )
1067 206 : .bind(id)
1068 206 : .bind(doc)
1069 206 : .bind(kind)
1070 206 : .bind(tenant.map(TenantId::as_str))
1071 206 : .fetch_one(&mut *tx)
1072 206 : .await?;
1073 206 : if inserted && kind == "Cached" {
1074 88 : sqlx::query(
1075 88 : "DELETE FROM jsonld_contexts WHERE id IN (
1076 88 : SELECT id FROM jsonld_contexts WHERE kind = 'Cached'
1077 88 : ORDER BY created_at DESC, id DESC OFFSET $1)",
1078 88 : )
1079 88 : .bind(crate::store::MAX_CACHED_CONTEXTS as i64)
1080 88 : .execute(&mut *tx)
1081 88 : .await?;
1082 118 : }
1083 206 : tx.commit().await?;
1084 206 : Ok(())
1085 206 : }
1086 :
1087 462 : pub async fn context_get(
1088 462 : &self,
1089 462 : tenant: Option<&TenantId>,
1090 462 : id: &str,
1091 462 : ) -> Result<Option<Value>, sqlx::Error> {
1092 462 : let mut tx = self.pool.begin().await?;
1093 462 : set_context_tenant(&mut tx, tenant).await?;
1094 462 : let row = sqlx::query(
1095 462 : "SELECT body FROM jsonld_contexts
1096 462 : WHERE id = $1 AND (tenant_id IS NULL OR tenant_id = $2)",
1097 462 : )
1098 462 : .bind(id)
1099 462 : .bind(tenant.map(TenantId::as_str))
1100 462 : .fetch_optional(&mut *tx)
1101 462 : .await?;
1102 462 : tx.commit().await?;
1103 462 : Ok(row.map(|r| r.get::<Value, _>(0)))
1104 462 : }
1105 :
1106 : /// Read-modify-write of one row under `FOR UPDATE`: `None` when it is
1107 : /// absent or another Tenant's, `Some(Err)` when `f` refused (rolled back).
1108 10328 : pub async fn context_mutate(
1109 10328 : &self,
1110 10328 : tenant: Option<&TenantId>,
1111 10328 : id: &str,
1112 10328 : f: impl FnOnce(&mut Value) -> Result<(), ()> + Send,
1113 10328 : ) -> Result<Option<Result<(), ()>>, sqlx::Error> {
1114 10328 : let mut tx = self.pool.begin().await?;
1115 10328 : set_context_tenant(&mut tx, tenant).await?;
1116 10328 : let row = sqlx::query(
1117 10328 : "SELECT body FROM jsonld_contexts
1118 10328 : WHERE id = $1 AND (tenant_id IS NULL OR tenant_id = $2) FOR UPDATE",
1119 10328 : )
1120 10328 : .bind(id)
1121 10328 : .bind(tenant.map(TenantId::as_str))
1122 10328 : .fetch_optional(&mut *tx)
1123 10328 : .await?;
1124 10328 : let Some(row) = row else {
1125 86 : return Ok(None);
1126 : };
1127 10242 : let mut doc: Value = row.get(0);
1128 10242 : if f(&mut doc).is_err() {
1129 0 : return Ok(Some(Err(())));
1130 10242 : }
1131 10242 : sqlx::query("UPDATE jsonld_contexts SET body = $2 WHERE id = $1")
1132 10242 : .bind(id)
1133 10242 : .bind(&doc)
1134 10242 : .execute(&mut *tx)
1135 10242 : .await?;
1136 10242 : tx.commit().await?;
1137 10242 : Ok(Some(Ok(())))
1138 10328 : }
1139 :
1140 192 : pub async fn context_delete(
1141 192 : &self,
1142 192 : tenant: Option<&TenantId>,
1143 192 : id: &str,
1144 192 : ) -> Result<bool, sqlx::Error> {
1145 192 : let mut tx = self.pool.begin().await?;
1146 192 : set_context_tenant(&mut tx, tenant).await?;
1147 192 : let gone = sqlx::query(
1148 192 : "DELETE FROM jsonld_contexts
1149 192 : WHERE id = $1 AND (tenant_id IS NULL OR tenant_id = $2)",
1150 192 : )
1151 192 : .bind(id)
1152 192 : .bind(tenant.map(TenantId::as_str))
1153 192 : .execute(&mut *tx)
1154 192 : .await?
1155 192 : .rows_affected()
1156 : == 1;
1157 192 : tx.commit().await?;
1158 192 : Ok(gone)
1159 192 : }
1160 :
1161 : /// Every row without its `@context` document: `- 'body'` drops that
1162 : /// member in the database, so the bodies are never on the wire, never
1163 : /// decoded and never resident. A row's body may be 5 MiB and only the
1164 : /// `Cached` rows are capped in number, so selecting whole rows here was
1165 : /// a multi-gigabyte read on the boot path.
1166 86 : pub async fn context_list_meta(
1167 86 : &self,
1168 86 : tenant: Option<&TenantId>,
1169 86 : ) -> Result<Vec<Value>, sqlx::Error> {
1170 86 : let mut tx = self.pool.begin().await?;
1171 86 : set_context_tenant(&mut tx, tenant).await?;
1172 86 : let rows = sqlx::query(
1173 86 : "SELECT body - 'body' FROM jsonld_contexts
1174 86 : WHERE tenant_id IS NULL OR tenant_id = $1 ORDER BY id",
1175 86 : )
1176 86 : .bind(tenant.map(TenantId::as_str))
1177 86 : .fetch_all(&mut *tx)
1178 86 : .await?;
1179 86 : tx.commit().await?;
1180 514 : Ok(rows.into_iter().map(|r| r.get::<Value, _>(0)).collect())
1181 86 : }
1182 : }
1183 :
1184 : #[cfg(test)]
1185 : mod tests {
1186 : use super::*;
1187 : use serde_json::json;
1188 :
1189 : #[test]
1190 2 : fn ops_mask_expands_groups_and_defaults() {
1191 : // absent operations = federationOps (5.2.9)
1192 2 : let default_mask = ops_mask(&json!({}));
1193 2 : let fed_mask = ops_mask(&json!({"operations": ["federationOps"]}));
1194 2 : assert_eq!(default_mask, fed_mask);
1195 1306 : let bit = |op: &str| 1i64 << OPERATION_NAMES.iter().position(|o| *o == op).expect(op);
1196 : // every member of the group, not a sample: the index and the
1197 : // registration validator read one Table 4.20-2, so a member the
1198 : // vocabulary gains or loses has to move this mask with it
1199 38 : for op in group_members("federationOps").expect("federationOps") {
1200 38 : assert_ne!(fed_mask & bit(op), 0, "{op} is in federationOps");
1201 : }
1202 2 : assert_eq!(
1203 2 : fed_mask & bit("createEntity"),
1204 : 0,
1205 : "provision op not in federationOps"
1206 : );
1207 2 : let m = ops_mask(&json!({"operations": ["createEntity", "retrieveOps"]}));
1208 2 : assert_ne!(m & bit("createEntity"), 0);
1209 2 : assert_ne!(m & bit("queryEntity"), 0, "retrieveOps expands");
1210 2 : assert_eq!(m & bit("deleteEntity"), 0);
1211 2 : assert_eq!(ops_mask(&json!({"operations": ["notARealOp"]})), 0);
1212 2 : }
1213 :
1214 : /// Table 4.20-2 defines `redirectionOps` as 23 operations — the provision
1215 : /// and retrieve set PLUS type/attribute introspection, the EntityMap
1216 : /// operations and `retrieveContextSourceIdentity`. A short group writes a
1217 : /// mask that stops those requests being redirected at all.
1218 : #[test]
1219 2 : fn redirection_ops_expands_to_the_whole_table_4_20_2_group() {
1220 2 : let m = ops_mask(&json!({"operations": ["redirectionOps"]}));
1221 1164 : let bit = |op: &str| 1i64 << OPERATION_NAMES.iter().position(|o| *o == op).expect(op);
1222 32 : for op in [
1223 2 : "createEntity",
1224 2 : "deleteEntity",
1225 2 : "purgeEntity",
1226 2 : "retrieveEntity",
1227 2 : "queryEntity",
1228 2 : "retrieveEntityTypes",
1229 2 : "retrieveEntityTypeDetails",
1230 2 : "retrieveEntityTypeInfo",
1231 2 : "retrieveAttrTypes",
1232 2 : "retrieveAttrTypeDetails",
1233 2 : "retrieveAttrTypeInfo",
1234 2 : "retrieveEntityMap",
1235 2 : "updateEntityMap",
1236 2 : "deleteEntityMap",
1237 2 : "createEntityMapQueryEntity",
1238 2 : "retrieveContextSourceIdentity",
1239 2 : ] {
1240 32 : assert_ne!(m & bit(op), 0, "redirectionOps is missing {op}");
1241 : }
1242 2 : assert_eq!(m.count_ones(), 23, "Table 4.20-2 lists 23 operations");
1243 : // and NOT the members the table leaves out
1244 10 : for op in [
1245 2 : "createSubscription",
1246 2 : "queryBatch",
1247 2 : "createBatch",
1248 2 : "retrieveTemporal",
1249 2 : "createEntityMapQueryTemporal",
1250 2 : ] {
1251 10 : assert_eq!(m & bit(op), 0, "{op} is not a redirectionOp");
1252 : }
1253 2 : }
1254 :
1255 : /// The bit position IS stored data (`csource_index.ops` is a `bigint`), so
1256 : /// the list can only grow to 63 entries; at 64 the shift silently writes a
1257 : /// wrong mask no migration could tell from a real one.
1258 : #[test]
1259 2 : fn the_operation_list_still_fits_the_bitmask() {
1260 2 : assert!(OPERATION_NAMES.len() < 64, "ops bitmask is i64");
1261 : // every name is unique: a duplicate would give one operation two bits
1262 2 : let mut seen = OPERATION_NAMES.to_vec();
1263 2 : seen.sort_unstable();
1264 2 : seen.dedup();
1265 2 : assert_eq!(
1266 2 : seen.len(),
1267 2 : OPERATION_NAMES.len(),
1268 : "duplicate operation name"
1269 : );
1270 2 : }
1271 :
1272 : /// The explosion ceiling has to bound the shape its own doc comment names
1273 : /// — a document written through a path with no cardinality validation (a
1274 : /// restored dump, an importer). ONE entity with a huge `propertyNames` is
1275 : /// exactly that shape, and a per-entity check never sees it twice.
1276 : #[test]
1277 2 : fn the_index_ceiling_bounds_one_entity_too() {
1278 21000 : let names: Vec<String> = (0..MAX_INDEX_ROWS + 500).map(|i| format!("p{i}")).collect();
1279 2 : let reg = json!({
1280 2 : "endpoint": "http://cs.example:9090",
1281 2 : "information": [{"entities": [{"type": "T"}], "propertyNames": names}]
1282 : });
1283 2 : assert_eq!(index_rows(®).len(), MAX_INDEX_ROWS);
1284 : // and across many entities, where it already held
1285 2 : let many: Vec<Value> = (0..MAX_INDEX_ROWS + 500)
1286 21000 : .map(|i| json!({"id": format!("urn:e:{i}")}))
1287 2 : .collect();
1288 2 : let reg = json!({"endpoint": "e", "information": [{"entities": many}]});
1289 2 : assert_eq!(index_rows(®).len(), MAX_INDEX_ROWS);
1290 2 : }
1291 :
1292 : #[test]
1293 2 : fn index_rows_explode_information() {
1294 : // 2 entities × (1 property + 1 relationship) = 4 rows
1295 2 : let reg = json!({
1296 2 : "endpoint": "http://cs.example:9090",
1297 2 : "mode": "exclusive",
1298 2 : "information": [{
1299 2 : "entities": [
1300 2 : {"id": "urn:a", "type": "T1"},
1301 2 : {"idPattern": "urn:.*", "type": "T2"}
1302 : ],
1303 2 : "propertyNames": ["speed"],
1304 2 : "relationshipNames": ["isParked"]
1305 : }],
1306 2 : "expiresAt": "2030-01-01T00:00:00Z"
1307 : });
1308 2 : let rows = index_rows(®);
1309 2 : assert_eq!(rows.len(), 4);
1310 2 : assert!(rows
1311 2 : .iter()
1312 8 : .all(|r| r["endpoint"] == "http://cs.example:9090"
1313 8 : && r["mode"] == 3
1314 8 : && r["expires_at"] == "2030-01-01T00:00:00Z"));
1315 2 : assert!(rows
1316 2 : .iter()
1317 2 : .any(|r| r["entity_id"] == "urn:a" && r["property_name"] == "speed"));
1318 2 : assert!(rows
1319 2 : .iter()
1320 8 : .any(|r| r["id_pattern"] == "urn:.*" && r["relationship_name"] == "isParked"));
1321 : // attribute-less info: one row per entity, both attr columns NULL
1322 2 : let bare = json!({"endpoint": "e", "information": [{"entities": [{"type": "T"}]}]});
1323 2 : let rows = index_rows(&bare);
1324 2 : assert_eq!(rows.len(), 1);
1325 2 : assert!(rows[0]["property_name"].is_null() && rows[0]["relationship_name"].is_null());
1326 2 : }
1327 :
1328 : /// TS 104 175 9.4.2: `attributeNames` scopes the rows, and the
1329 : /// deprecated pair beside it is not indexed.
1330 : #[test]
1331 2 : fn attribute_names_are_indexed_and_the_old_pair_beside_them_is_not() {
1332 2 : let reg = json!({
1333 2 : "endpoint": "e",
1334 2 : "information": [{"entities": [{"type": "T"}],
1335 2 : "attributeNames": ["speed", "owner"],
1336 2 : "propertyNames": ["brand"]}]
1337 : });
1338 2 : let rows = index_rows(®);
1339 2 : let mut names: Vec<&str> = rows
1340 2 : .iter()
1341 4 : .filter_map(|r| r["property_name"].as_str())
1342 2 : .collect();
1343 2 : names.sort_unstable();
1344 2 : assert_eq!(names, ["owner", "speed"]);
1345 2 : }
1346 : }
|
