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