Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! PgStore, first slice: entity CRUD over the `entities`
3 : //! table. Same signatures as the in-memory `Store`, awaited: every method is
4 : //! an `async fn` over sqlx, so a caller waiting on the database holds no
5 : //! thread.
6 : //!
7 : //! Extracted columns are computed in Rust at write time (no triggers):
8 : //! `types`, `scopes`, `created_at`, `modified_at`, `expires_at` and
9 : //! `location`, the default GeoProperty, converted by PostGIS itself from
10 : //! bound GeoJSON text (`ST_GeomFromGeoJSON` rather than a geozero
11 : //! dependency — the DB already owns the conversion, and the value still
12 : //! travels as a bind).
13 :
14 : use antares_model::{NgsiError, TenantId};
15 : use serde_json::Value;
16 : use sqlx::postgres::PgPool;
17 : use sqlx::Row;
18 :
19 : pub struct PgEntityStore {
20 : pool: PgPool,
21 : /// When on, every entity write enqueues its change event into the
22 : /// outbox INSIDE the write transaction (a crash between commit and
23 : /// publish can never lose an event). Off by default: with `bus = local`
24 : /// events flow through the in-process hook and undrained rows would only
25 : /// grow the table. The broker turns this on when `bus = nats`.
26 : outbox: std::sync::atomic::AtomicBool,
27 : }
28 :
29 : // EntityFilter/Page/QueryOutcome live in `store::filter` (pure data,
30 : // shared with the wasm32 build); re-exported here so existing paths hold.
31 : pub use crate::store::filter::{EntityFilter, Page, QueryOutcome};
32 :
33 : /// A bound value. Enumerated because the bind list is built dynamically while
34 : /// the SQL is assembled — the alternative would be string interpolation.
35 : enum Bind {
36 : Text(String),
37 : TextArr(Vec<String>),
38 : /// jsonpath; bound as text and cast with `$n::jsonpath` in the SQL
39 : Path(String),
40 : /// a distance in metres (geoquery `near`)
41 : Num(f64),
42 : /// LIMIT/OFFSET (pagination pushdown)
43 : Int(i64),
44 : }
45 :
46 : /// The internal doc's members that become extracted columns.
47 : pub(crate) struct Extracted {
48 : types: Vec<String>,
49 : scopes: Option<Vec<String>>,
50 : created: String,
51 : modified: String,
52 : expires: Option<String>,
53 : /// The default GeoProperty as GeoJSON text, for `ST_GeomFromGeoJSON`.
54 : /// `None` (→ SQL NULL) whenever it cannot be represented as ONE geometry.
55 : location: Option<String>,
56 : /// True when the doc CARRIES the default GeoProperty but `location` could
57 : /// not be extracted (multi-instance, non-GeoJSON value). The compiled
58 : /// geoquery ORs on this column — index-shaped (BitmapOr), unlike the old
59 : /// `location IS NULL OR …` guard which forced a sequential scan — and
60 : /// rows without any geoproperty are excluded in SQL, which is exact.
61 : location_ambiguous: bool,
62 : }
63 :
64 : /// The four members every stored kind extracts the same way: `type` and
65 : /// `scope` as arrays whether the doc spells them as one string or many, and
66 : /// the two stamps. The temporal tables carry exactly these, entities carry
67 : /// them plus expiry and location, so this is the one implementation of the
68 : /// shared half — two copies is how one of them drifts.
69 : ///
70 : /// 4.6.3 allows a comma seconds-fraction in requests; the stamps feed
71 : /// `::timestamptz` casts, which refuse it, so they are canonicalised here and
72 : /// every extracted column holds the instant the client meant.
73 3677 : pub(super) fn types_scopes_stamps(
74 3677 : doc: &Value,
75 3677 : ) -> (Vec<String>, Option<Vec<String>>, String, String) {
76 3713 : let as_vec = |v: &Value| -> Vec<String> {
77 3713 : match v {
78 179 : Value::String(s) => vec![s.clone()],
79 3532 : Value::Array(a) => a
80 3532 : .iter()
81 3547 : .filter_map(|x| x.as_str().map(str::to_owned))
82 3532 : .collect(),
83 2 : _ => vec![],
84 : }
85 3713 : };
86 7354 : let ts = |k: &str| {
87 7354 : doc.get(k)
88 7354 : .and_then(Value::as_str)
89 7354 : .map(|s| antares_store::filter::canonical_datetime(s).into_owned())
90 7354 : };
91 5044 : let now = || "1970-01-01T00:00:00Z".to_owned(); // caller always stamps; belt only
92 3677 : (
93 3677 : doc.get("type").map(&as_vec).unwrap_or_default(),
94 3677 : doc.get("scope").map(&as_vec),
95 3677 : ts("createdAt").unwrap_or_else(now),
96 3677 : ts("modifiedAt").unwrap_or_else(now),
97 3677 : )
98 3677 : }
99 :
100 3276 : fn extract(doc: &Value) -> Extracted {
101 3276 : let (types, scopes, created, modified) = types_scopes_stamps(doc);
102 3276 : let ts = |k: &str| {
103 3276 : doc.get(k)
104 3276 : .and_then(Value::as_str)
105 3276 : .map(|s| antares_store::filter::canonical_datetime(s).into_owned())
106 3276 : };
107 3276 : let location = crate::compile::geo::extract_location(doc);
108 3276 : let location_ambiguous =
109 3276 : location.is_none() && doc.get(crate::compile::geo::LOCATION_IRI).is_some();
110 3276 : Extracted {
111 3276 : types,
112 3276 : scopes,
113 3276 : created,
114 3276 : modified,
115 3276 : expires: ts("expiresAt"),
116 3276 : location,
117 3276 : location_ambiguous,
118 3276 : }
119 3276 : }
120 :
121 : /// The outbox row's event JSON: what the drain turns into a
122 : /// `ChangeEvent`. Field names are the bus crate's serde names; `seq` and the
123 : /// claim check are the drain's business.
124 : #[allow(clippy::too_many_arguments)] // one param per outbox event field
125 22 : fn change_event(
126 22 : tenant: &TenantId,
127 22 : op: &str,
128 22 : id: &str,
129 22 : types: &[String],
130 22 : prev: Option<&Value>,
131 22 : next: Option<&Value>,
132 22 : version: i64,
133 22 : incarnation: &str,
134 22 : ) -> Value {
135 : // changed_attrs = top-level attribute IRIs that differ between the
136 : // before- and after-images (meta members excluded). Create lists every
137 : // attr, delete lists every prior attr.
138 44 : fn keys(v: Option<&Value>) -> Vec<&str> {
139 44 : v.and_then(Value::as_object)
140 44 : .map(|o| {
141 22 : o.keys()
142 22 : .map(String::as_str)
143 120 : .filter(|k| !antares_model::is_meta(k))
144 22 : .collect()
145 22 : })
146 44 : .unwrap_or_default()
147 44 : }
148 22 : let mut changed: Vec<&str> = Vec::new();
149 34 : for k in keys(prev).into_iter().chain(keys(next)) {
150 34 : if !changed.contains(&k) && prev.and_then(|p| p.get(k)) != next.and_then(|n| n.get(k)) {
151 20 : changed.push(k);
152 20 : }
153 : }
154 22 : serde_json::json!({
155 22 : "tenant": tenant.as_str(),
156 22 : "entity_id": id,
157 22 : "types": types,
158 22 : "op": op,
159 22 : "changed_attrs": changed,
160 22 : "payload": next,
161 22 : "prev_payload": prev,
162 22 : "version": version,
163 22 : "incarnation": incarnation,
164 : })
165 22 : }
166 :
167 : #[allow(clippy::too_many_arguments)] // one param per outbox event field
168 4 : async fn enqueue_change(
169 4 : tx: &mut sqlx::postgres::PgConnection,
170 4 : tenant: &TenantId,
171 4 : op: &str,
172 4 : id: &str,
173 4 : types: &[String],
174 4 : prev: Option<&Value>,
175 4 : next: Option<&Value>,
176 4 : version: i64,
177 4 : incarnation: &str,
178 4 : ) -> Result<(), sqlx::Error> {
179 4 : let ev = change_event(tenant, op, id, types, prev, next, version, incarnation);
180 4 : super::outbox::enqueue(tx, tenant, &ev).await.map(|_| ())
181 4 : }
182 :
183 : /// Rows one statement may materialize when SQL narrowed the match set but
184 : /// did not decide it. 5.5.6 leaves the threshold to the implementation;
185 : /// this is it for the entity query path, and it is the only bound a request
186 : /// that carries no page (idPattern, federation, orderBy) has at all.
187 : pub const MAX_UNDECIDED_ROWS: i64 = 10_000;
188 :
189 : /// The statement, split out of `query` so its pagination shape is
190 : /// assertable without a database. Either LIMIT/OFFSET over an exactly
191 : /// decided set (`true`, `count(*) OVER ()` rides along for the pre-LIMIT
192 : /// total) or the flat `MAX_UNDECIDED_ROWS` safety LIMIT — never no LIMIT.
193 282 : fn query_sql(
194 282 : select: &str,
195 282 : wheres: &[String],
196 282 : page: Option<&Page>,
197 282 : decided: bool,
198 282 : binds: &mut Vec<Bind>,
199 282 : ) -> (String, bool) {
200 282 : let wheres = wheres.join(" AND ");
201 282 : match page.filter(|_| decided) {
202 : // count=true: the window total rides the statement. Otherwise one
203 : // row past the page answers "is there a next page" — the window
204 : // count made Postgres visit the whole match set before LIMIT
205 : // (2.9 s p99 for 20 rows out of 10 000).
206 135 : Some(p) => {
207 135 : let total = if p.count {
208 17 : ", count(*) OVER () AS total"
209 : } else {
210 118 : ""
211 : };
212 : // Saturating: the caller's limit is only bounded by `i64::MAX`
213 : // (`ANTARES_DISCOVERY_SCAN_MAX` clamps to exactly that), and one
214 : // past the largest page is still the largest page.
215 135 : binds.push(Bind::Int(if p.count {
216 17 : p.limit
217 : } else {
218 118 : p.limit.saturating_add(1)
219 : }));
220 135 : let lim = binds.len();
221 135 : binds.push(Bind::Int(p.offset));
222 135 : (
223 135 : format!(
224 135 : "SELECT {select} AS entity{total} \
225 135 : FROM entities WHERE {wheres} ORDER BY id LIMIT ${lim} OFFSET ${}",
226 135 : binds.len()
227 135 : ),
228 135 : true,
229 135 : )
230 : }
231 : None => {
232 147 : binds.push(Bind::Int(MAX_UNDECIDED_ROWS));
233 147 : (
234 147 : format!(
235 147 : "SELECT {select} AS entity FROM entities WHERE {wheres} \
236 147 : ORDER BY id LIMIT ${}",
237 147 : binds.len()
238 147 : ),
239 147 : false,
240 147 : )
241 : }
242 : }
243 282 : }
244 :
245 : /// Did the safety LIMIT actually cut the match set? A statement that came
246 : /// back full was truncated at a bound nobody chose, and the caller's
247 : /// evaluator still has rows to drop, so no complete page can be built from
248 : /// it. 5.5.6: "When a query operation is producing so many results that can
249 : /// potentially exhaust client or server resources … implementations shall
250 : /// raise an error of type TooManyResults." Serving the truncated prefix
251 : /// would under-report the answer instead.
252 6624 : pub(crate) fn check_ceiling(paged: bool, rows: usize, ceiling: i64) -> Result<(), sqlx::Error> {
253 6624 : if paged || (rows as i64) < ceiling {
254 6620 : return Ok(());
255 4 : }
256 4 : Err(sqlx::Error::Configuration(Box::new(
257 4 : NgsiError::TooManyResults(format!(
258 4 : "query matched more than {ceiling} entities before filtering; \
259 4 : narrow it or request a smaller offset"
260 4 : )),
261 4 : )))
262 6624 : }
263 :
264 : /// Recover a spec error the store raised itself from the driver error it
265 : /// travels in (the store's only error channel is `sqlx::Error`). `None` =
266 : /// a genuine driver failure, which stays an InternalError at the seam.
267 10 : pub fn ngsi_error(e: &sqlx::Error) -> Option<&NgsiError> {
268 10 : match e {
269 6 : sqlx::Error::Configuration(b) => b.downcast_ref::<NgsiError>(),
270 4 : _ => None,
271 : }
272 10 : }
273 :
274 : impl PgEntityStore {
275 60 : pub fn new(pool: PgPool) -> Self {
276 60 : Self {
277 60 : pool,
278 60 : outbox: std::sync::atomic::AtomicBool::new(false),
279 60 : }
280 60 : }
281 :
282 : /// Outbox producer switch — the broker enables this exactly when `bus=nats`.
283 2 : pub fn set_outbox(&self, on: bool) {
284 2 : self.outbox.store(on, std::sync::atomic::Ordering::Relaxed);
285 2 : }
286 :
287 4852 : fn outbox_on(&self) -> bool {
288 4852 : self.outbox.load(std::sync::atomic::Ordering::Relaxed)
289 4852 : }
290 :
291 : /// 5.6.1-shaped create: `false` when the id already exists (→ 409).
292 : ///
293 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
294 : /// certain Entity, Property or Relationship shall become invalid" — an
295 : /// entity whose expiry has passed is already invalid, whatever the
296 : /// reaping lag, so it must not 409 a create that reads (and GETs) as
297 : /// absent. The conflict clause replaces such a row and reports CREATED;
298 : /// a live row still takes the DO NOTHING path and `rows_affected() == 0`.
299 3128 : pub async fn create(
300 3128 : &self,
301 3128 : tenant: &TenantId,
302 3128 : id: &str,
303 3128 : doc: &Value,
304 3128 : ) -> Result<bool, sqlx::Error> {
305 3128 : let e = extract(doc);
306 3128 : let mut tx = super::begin(&self.pool).await?;
307 3128 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
308 3128 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
309 3128 : let done = sqlx::query(
310 3128 : "INSERT INTO entities
311 3128 : (tenant_id, id, entity, types, scopes, created_at, modified_at, expires_at,
312 3128 : location, location_ambiguous)
313 3128 : VALUES ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz, $8::timestamptz,
314 3128 : CASE WHEN ST_IsValid(try_geomfromgeojson($9))
315 3128 : THEN try_geomfromgeojson($9) END,
316 3128 : $10 OR ($9 IS NOT NULL
317 3128 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson($9)), false)))
318 3128 : ON CONFLICT (tenant_id, id) DO UPDATE SET
319 3128 : entity = EXCLUDED.entity, types = EXCLUDED.types,
320 3128 : scopes = EXCLUDED.scopes, created_at = EXCLUDED.created_at,
321 3128 : modified_at = EXCLUDED.modified_at, expires_at = EXCLUDED.expires_at,
322 3128 : location = EXCLUDED.location,
323 3128 : location_ambiguous = EXCLUDED.location_ambiguous,
324 3128 : version = 1
325 3128 : WHERE entities.expires_at IS NOT NULL AND entities.expires_at <= now()",
326 3128 : )
327 3128 : .bind(tenant.as_str())
328 3128 : .bind(id)
329 3128 : .bind(doc)
330 3128 : .bind(&e.types)
331 3128 : .bind(&e.scopes)
332 3128 : .bind(&e.created)
333 3128 : .bind(&e.modified)
334 3128 : .bind(&e.expires)
335 3128 : .bind(&e.location)
336 3128 : .bind(e.location_ambiguous)
337 3128 : .execute(&mut *tx)
338 3128 : .await?
339 3128 : .rows_affected();
340 3128 : if done == 1 && self.outbox_on() {
341 4 : enqueue_change(
342 4 : &mut tx,
343 4 : tenant,
344 4 : "create",
345 4 : id,
346 4 : &e.types,
347 4 : None,
348 4 : Some(doc),
349 4 : 1,
350 4 : &e.created,
351 4 : )
352 4 : .await?;
353 3124 : }
354 3128 : tx.commit().await?;
355 3128 : Ok(done == 1)
356 3128 : }
357 :
358 147 : pub async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, sqlx::Error> {
359 147 : let mut tx = super::begin(&self.pool).await?;
360 147 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
361 : // 4.22: expired rows are invalid context until the sweep reaps them
362 147 : let row = sqlx::query(
363 147 : "SELECT entity FROM entities WHERE tenant_id = $1 AND id = $2
364 147 : AND (expires_at IS NULL OR expires_at > now())",
365 147 : )
366 147 : .bind(tenant.as_str())
367 147 : .bind(id)
368 147 : .fetch_optional(&mut *tx)
369 147 : .await?;
370 147 : tx.commit().await?;
371 147 : Ok(row.map(|r| r.get::<Value, _>(0)))
372 147 : }
373 :
374 : /// Returns the deleted document (the before-image, captured in the same
375 : /// transaction as the DELETE — never re-read outside it), `None` = absent.
376 1377 : pub async fn delete(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, sqlx::Error> {
377 1377 : let mut tx = super::begin(&self.pool).await?;
378 1377 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
379 : // 4.22: an already-expired row is invalid, so deleting it is a
380 : // 404 exactly as retrieving it is — never a 204 for an entity the
381 : // API has stopped serving.
382 1377 : let row = sqlx::query(
383 1377 : "DELETE FROM entities WHERE tenant_id = $1 AND id = $2
384 1377 : AND (expires_at IS NULL OR expires_at > now())
385 1377 : RETURNING entity, types, version, created_at::text",
386 1377 : )
387 1377 : .bind(tenant.as_str())
388 1377 : .bind(id)
389 1377 : .fetch_optional(&mut *tx)
390 1377 : .await?;
391 1377 : let mut prev_out = None;
392 1377 : if let Some(r) = &row {
393 1284 : let prev: Value = r.get(0);
394 1284 : if self.outbox_on() {
395 0 : let types: Vec<String> = r.get(1);
396 0 : enqueue_change(
397 0 : &mut tx,
398 0 : tenant,
399 0 : "delete",
400 0 : id,
401 0 : &types,
402 0 : Some(&prev),
403 0 : None,
404 0 : r.get::<i64, _>(2),
405 0 : r.get::<&str, _>(3),
406 0 : )
407 0 : .await?;
408 1284 : }
409 1284 : prev_out = Some(prev);
410 93 : }
411 1377 : tx.commit().await?;
412 1377 : Ok(prev_out)
413 1377 : }
414 :
415 : /// Delete one entity only if `keep` accepts the stored document, under
416 : /// one lock: the DELETE itself takes the row lock and produces the
417 : /// document that decides, and a refusal rolls the transaction back
418 : /// instead of committing it. A `SELECT` followed by a `DELETE` would
419 : /// leave the window this exists to close — between them the row can be
420 : /// deleted and recreated under the same id, and the delete then lands on
421 : /// a document nobody inspected.
422 : ///
423 : /// 4.22 as in `delete`: an already-expired row is invalid, so it is
424 : /// absent here too.
425 1088 : pub async fn delete_if(
426 1088 : &self,
427 1088 : tenant: &TenantId,
428 1088 : id: &str,
429 1088 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
430 1088 : ) -> Result<Option<Value>, sqlx::Error> {
431 1088 : let mut tx = super::begin(&self.pool).await?;
432 1088 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
433 1088 : let row = sqlx::query(
434 1088 : "DELETE FROM entities WHERE tenant_id = $1 AND id = $2
435 1088 : AND (expires_at IS NULL OR expires_at > now())
436 1088 : RETURNING entity, types, version, created_at::text",
437 1088 : )
438 1088 : .bind(tenant.as_str())
439 1088 : .bind(id)
440 1088 : .fetch_optional(&mut *tx)
441 1088 : .await?;
442 1088 : let Some(r) = &row else {
443 870 : tx.commit().await?;
444 870 : return Ok(None);
445 : };
446 218 : let prev: Value = r.get(0);
447 218 : if !keep(&prev) {
448 4 : tx.rollback().await?;
449 4 : return Ok(None);
450 214 : }
451 214 : if self.outbox_on() {
452 0 : let types: Vec<String> = r.get(1);
453 0 : enqueue_change(
454 0 : &mut tx,
455 0 : tenant,
456 0 : "delete",
457 0 : id,
458 0 : &types,
459 0 : Some(&prev),
460 0 : None,
461 0 : r.get::<i64, _>(2),
462 0 : r.get::<&str, _>(3),
463 0 : )
464 0 : .await?;
465 214 : }
466 214 : tx.commit().await?;
467 214 : Ok(Some(prev))
468 1088 : }
469 :
470 : /// Query pushdown. The predicates that compile EXACTLY go to
471 : /// Postgres; everything else is simply left out of the WHERE clause, so
472 : /// the result is always a superset of the answer and the caller's
473 : /// in-memory evaluator remains the arbiter. That is the property that
474 : /// makes store modes agree: SQL removes rows, it never decides them.
475 : ///
476 : /// Every value here is a bind. The only text this function
477 : /// concatenates is its own operators and `$n` placeholders.
478 : ///
479 : /// Every statement it builds is bounded: an exactly decided query by the
480 : /// caller's page, everything else by `MAX_UNDECIDED_ROWS`. A set that
481 : /// reaches that ceiling is refused with TooManyResults (5.5.6) rather
482 : /// than returned as a prefix the caller would page over as if complete.
483 270 : pub async fn query(
484 270 : &self,
485 270 : tenant: &TenantId,
486 270 : f: &EntityFilter<'_>,
487 270 : ) -> Result<QueryOutcome, sqlx::Error> {
488 270 : let mut binds: Vec<Bind> = vec![Bind::Text(tenant.as_str().to_owned())];
489 270 : let mut wheres = vec![
490 270 : "tenant_id = $1".to_owned(),
491 : // 4.22: expired entities never enter a result (or a page/total)
492 270 : "(expires_at IS NULL OR expires_at > now())".to_owned(),
493 : ];
494 : // Exactness: ids/types/attrs translate exactly by
495 : // construction; q is exact IF it compiles (the compiler's contract);
496 : // scopeQ is documented loose-or-equal, geo has a metric residual
497 : // (`near` geography vs haversine) — both therefore forfeit
498 : // `decided`, they only narrow.
499 270 : let mut decided = true;
500 :
501 270 : if let Some(ids) = f.ids {
502 81 : binds.push(Bind::TextArr(ids.iter().map(|s| s.to_string()).collect()));
503 31 : wheres.push(format!("id = ANY(${})", binds.len()));
504 239 : }
505 : // 5.2.33 idPattern: the literal the regex forces is a plain string
506 : // test on the id column, exact as a narrowing; the regex itself stays
507 : // with the caller, so `decided` is untouched.
508 270 : if let Some(l) = f.id_literal {
509 18 : binds.push(Bind::Text(l.text.to_owned()));
510 18 : wheres.push(if l.anchored {
511 4 : format!("starts_with(id, ${})", binds.len())
512 : } else {
513 14 : format!("position(${} in id) > 0", binds.len())
514 : });
515 252 : }
516 : // OR of AND-groups, mirroring the Entity Type Selection Language
517 : // (4.17) the caller already parsed: `types @> ARRAY[…]` per group.
518 270 : if let Some(groups) = f.types {
519 149 : let mut ors = Vec::with_capacity(groups.len());
520 149 : for g in groups {
521 149 : binds.push(Bind::TextArr(g.clone()));
522 149 : ors.push(format!("types @> ${}", binds.len()));
523 149 : }
524 149 : if !ors.is_empty() {
525 149 : wheres.push(format!("({})", ors.join(" OR ")));
526 149 : }
527 121 : }
528 : // `attrs`: the entity carries at least one of them — jsonb `?|`,
529 : // exactly the evaluator's `any(|a| doc.get(a).is_some())`.
530 270 : if let Some(attrs) = f.attrs {
531 1 : binds.push(Bind::TextArr(attrs.to_vec()));
532 1 : wheres.push(format!("entity ?| ${}", binds.len()));
533 269 : }
534 270 : if let Some(node) = f.q {
535 65 : match crate::compile::q::compile_q(node, "entity", binds.len() + 1, f.expand) {
536 51 : Some(c) => {
537 51 : wheres.push(c.sql);
538 51 : binds.extend(c.binds.into_iter().map(Bind::Path));
539 51 : }
540 14 : None => decided = false,
541 : }
542 205 : }
543 270 : if let Some(sq) = f.scope_q {
544 16 : decided = false;
545 16 : if let Some(c) = crate::compile::scope::compile_scope_q(sq, "scopes", binds.len() + 1) {
546 14 : wheres.push(c.sql);
547 14 : binds.extend(c.binds.into_iter().map(Bind::Text));
548 14 : }
549 254 : }
550 270 : if let Some(spec) = f.geo {
551 16 : decided = false;
552 : // A client may send a self-intersecting polygon. GEOS raises on
553 : // one (`side location conflict`), which would turn a query into a
554 : // 500 in `postgres` mode while `memory` mode answers happily from
555 : // the evaluator. Probing validity once here keeps the two modes
556 : // identical: invalid ⇒ no pushdown, evaluator decides. Stored
557 : // geometries can't be invalid — the write path NULLs those.
558 16 : if self.geometry_is_valid(spec).await {
559 16 : if let Some(c) = crate::compile::geo::compile_geo(spec, "location", binds.len() + 1)
560 16 : {
561 16 : wheres.push(c.sql);
562 16 : // geo binds first, then the numeric ones — the order
563 16 : // `compile_geo` numbered its placeholders in.
564 16 : binds.extend(c.geo_binds.into_iter().map(Bind::Text));
565 16 : binds.extend(c.num_binds.into_iter().map(Bind::Num));
566 16 : }
567 0 : }
568 254 : }
569 :
570 : // every bind up to here belongs to the WHERE clause — the count-only
571 : // fallback statement below reuses exactly this prefix
572 270 : let where_binds = binds.len();
573 :
574 : // Projection pushdown, only once SQL decides row membership: the
575 : // kept doc must only need to feed `repr::apply`, never a re-check.
576 : // `pick` keeps listed attrs + every non-attribute member (attribute
577 : // keys are expanded IRIs — `http…` — so core members never match the
578 : // LIKE and always survive; a non-http attr IRI merely stays
579 : // unprojected, which is the safe direction). `omit` drops exactly the
580 : // listed top-level IRIs.
581 270 : let mut select = "entity".to_owned();
582 270 : if decided {
583 224 : if let Some(keep) = f.keep_attrs {
584 2 : binds.push(Bind::TextArr(keep.to_vec()));
585 2 : select = format!(
586 2 : "(SELECT COALESCE(jsonb_object_agg(t.k, t.v), '{{}}'::jsonb) \
587 2 : FROM jsonb_each(entity) AS t(k, v) \
588 2 : WHERE t.k NOT LIKE 'http%' OR t.k = ANY(${}))",
589 2 : binds.len()
590 2 : );
591 222 : } else if let Some(drop) = f.drop_attrs {
592 0 : binds.push(Bind::TextArr(drop.to_vec()));
593 0 : select = format!(
594 0 : "(SELECT COALESCE(jsonb_object_agg(t.k, t.v), '{{}}'::jsonb) \
595 0 : FROM jsonb_each(entity) AS t(k, v) \
596 0 : WHERE NOT (t.k = ANY(${})))",
597 0 : binds.len()
598 0 : );
599 222 : }
600 46 : }
601 :
602 : // Pagination pushdown: ORDER BY id is the store's default order
603 : // either way; `count(*) OVER ()` rides the same statement so the
604 : // caller gets the pre-LIMIT total for count= and the next/prev
605 : // links. When the page cannot be pushed the statement still carries
606 : // the safety LIMIT — without one, a single undecided request (any
607 : // `scopeQ`, any `georel`, a `q=` shape the compiler declined)
608 : // materializes the whole tenant into the Vec below.
609 270 : let ceiling = MAX_UNDECIDED_ROWS;
610 270 : let (sql, paged) = query_sql(&select, &wheres, f.page.as_ref(), decided, &mut binds);
611 270 : let mut tx = super::begin(&self.pool).await?;
612 270 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
613 : // sqlx 0.9 makes dynamic SQL opt-in. The assertion holds by
614 : // construction: `sql` is built from this function's own literals
615 : // plus `$n` placeholders — no caller-supplied text reaches it.
616 : // The audit lives here, next to the builder.
617 270 : let mut qy = sqlx::query(sqlx::AssertSqlSafe(sql.clone()));
618 1450 : for b in &binds {
619 1450 : qy = match b {
620 862 : Bind::Text(s) | Bind::Path(s) => qy.bind(s),
621 183 : Bind::TextArr(v) => qy.bind(v),
622 6 : Bind::Num(n) => qy.bind(n),
623 399 : Bind::Int(n) => qy.bind(n),
624 : };
625 : }
626 : // Stream rows, decode each to its Value and
627 : // drop the PgRow — the full row set never sits in memory twice.
628 270 : let mut docs: Vec<Value> = Vec::new();
629 270 : let mut total: Option<i64> = None;
630 270 : let counted = paged && f.page.as_ref().is_some_and(|p| p.count);
631 : {
632 : use futures_util::TryStreamExt;
633 270 : let mut stream = qy.fetch(&mut *tx);
634 13328 : while let Some(row) = stream.try_next().await? {
635 13058 : if counted && docs.is_empty() {
636 13 : total = Some(row.get::<i64, _>(1));
637 13045 : }
638 13058 : docs.push(row.get::<Value, _>(0));
639 : }
640 : }
641 270 : check_ceiling(paged, docs.len(), ceiling)?;
642 : // uncounted page: the extra row says a next page exists; the
643 : // total handed back is the smallest one consistent with that,
644 : // enough for the next/prev links and never shown as a count
645 269 : if let Some(p) = f.page.as_ref().filter(|_| paged && !counted) {
646 114 : let more = docs.len() as i64 > p.limit;
647 114 : docs.truncate(p.limit as usize);
648 114 : total = Some(p.offset + docs.len() as i64 + i64::from(more));
649 155 : }
650 : // an off-the-end counted page returns zero rows and no window
651 : // total — count the match set separately so links/count stay correct
652 269 : if counted && total.is_none() {
653 2 : let count_sql = format!(
654 : "SELECT count(*) FROM entities WHERE {}",
655 2 : wheres.join(" AND ")
656 : );
657 2 : let mut cq = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(count_sql));
658 : // same wheres ⇒ same bind prefix; stop before the
659 : // projection/page binds, which the count statement lacks
660 4 : for b in binds.iter().take(where_binds) {
661 4 : cq = match b {
662 2 : Bind::Text(s) | Bind::Path(s) => cq.bind(s),
663 2 : Bind::TextArr(v) => cq.bind(v),
664 0 : Bind::Num(n) => cq.bind(n),
665 0 : Bind::Int(n) => cq.bind(n),
666 : };
667 : }
668 2 : total = Some(cq.fetch_one(&mut *tx).await?);
669 267 : }
670 269 : tx.commit().await?;
671 269 : Ok(QueryOutcome {
672 269 : rows: docs,
673 269 : decided,
674 269 : paged,
675 269 : total,
676 269 : })
677 270 : }
678 :
679 : /// One cheap probe: is the client's query geometry OGC-valid? Unparseable
680 : /// GeoJSON counts as invalid — same outcome, no pushdown — and reaches
681 : /// that answer through `try_geomfromgeojson` (0001_init.sql), so the
682 : /// probe itself can never raise.
683 16 : async fn geometry_is_valid(&self, spec: &crate::compile::geo::GeoSpec<'_>) -> bool {
684 16 : let geojson = serde_json::to_string(&serde_json::json!({
685 16 : "type": spec.geometry, "coordinates": spec.coordinates
686 16 : }))
687 16 : .unwrap_or_default();
688 16 : sqlx::query_scalar::<_, bool>("SELECT COALESCE(ST_IsValid(try_geomfromgeojson($1)), false)")
689 16 : .bind(&geojson)
690 16 : .fetch_one(&self.pool)
691 16 : .await
692 16 : .unwrap_or(false)
693 16 : }
694 :
695 : /// Id-ordered snapshot for one tenant (the v0 `list` shape — still the
696 : /// path for every non-entity kind and for callers with no filter).
697 : ///
698 : /// Bounded like every other read: the statement carries the same
699 : /// `MAX_UNDECIDED_ROWS` safety LIMIT, and a tenant that reaches it is
700 : /// refused with TooManyResults (5.5.6) rather than served a silent
701 : /// prefix.
702 20 : pub async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, sqlx::Error> {
703 20 : let mut tx = super::begin(&self.pool).await?;
704 20 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
705 20 : let mut docs: Vec<Value> = Vec::new();
706 : {
707 : use futures_util::TryStreamExt;
708 20 : let mut stream = sqlx::query(
709 : "SELECT entity FROM entities WHERE tenant_id = $1
710 : AND (expires_at IS NULL OR expires_at > now())
711 : ORDER BY id LIMIT $2",
712 : )
713 20 : .bind(tenant.as_str())
714 20 : .bind(MAX_UNDECIDED_ROWS)
715 20 : .fetch(&mut *tx);
716 11261 : while let Some(row) = stream.try_next().await? {
717 11241 : docs.push(row.get::<Value, _>(0));
718 11241 : }
719 : }
720 20 : tx.commit().await?;
721 20 : check_ceiling(false, docs.len(), MAX_UNDECIDED_ROWS)?;
722 19 : Ok(docs)
723 20 : }
724 :
725 : /// One id-ordered page of entities: ids strictly greater than `after`,
726 : /// at most `limit`.
727 : ///
728 : /// No ceiling, and no fold of the tenant to build the page.
729 : /// `MAX_UNDECIDED_ROWS` on `list` exists so a large tenant cannot be
730 : /// materialized into one `Vec`; a page bounds that by construction, so
731 : /// refusing here would only break the readers that must see every row
732 : /// and have no TooManyResults to raise — 5.9.2.4's registration-vs-entity
733 : /// conflict check among them. Keyset over the primary key, not OFFSET:
734 : /// the walk runs against a table being written to. `after = None` is
735 : /// bound as the empty string, which is below every id because an
736 : /// Entity id is a URI and a URI is never empty.
737 : ///
738 : /// 4.22 at the entity level is in the statement, so a full page is a
739 : /// full page: the caller reads a short page as the end of the tenant,
740 : /// and a filter applied after the LIMIT would end the walk early.
741 12 : pub async fn list_page(
742 12 : &self,
743 12 : tenant: &TenantId,
744 12 : after: Option<&str>,
745 12 : limit: i64,
746 12 : ) -> Result<Vec<Value>, sqlx::Error> {
747 12 : let mut tx = super::begin(&self.pool).await?;
748 12 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
749 12 : let rows = sqlx::query(
750 12 : "SELECT entity FROM entities WHERE tenant_id = $1 AND id > $2
751 12 : AND (expires_at IS NULL OR expires_at > now())
752 12 : ORDER BY id LIMIT $3",
753 12 : )
754 12 : .bind(tenant.as_str())
755 12 : .bind(after.unwrap_or(""))
756 12 : .bind(limit)
757 12 : .fetch_all(&mut *tx)
758 12 : .await?;
759 12 : tx.commit().await?;
760 2019 : Ok(rows.into_iter().map(|r| r.get::<Value, _>(0)).collect())
761 12 : }
762 :
763 : /// Read-modify-write: row lock via `SELECT … FOR UPDATE`, closure
764 : /// applied in Rust, `version` bumped under the lock. Two racing PATCHes
765 : /// serialize in Postgres, neither is lost. `Ok(None)` = entity absent.
766 2545 : pub async fn mutate<T, E>(
767 2545 : &self,
768 2545 : tenant: &TenantId,
769 2545 : id: &str,
770 2545 : f: impl FnOnce(&mut Value) -> Result<T, E>,
771 2545 : ) -> Result<Option<Result<T, E>>, sqlx::Error> {
772 2545 : let mut tx = super::begin(&self.pool).await?;
773 2545 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
774 : // 4.22: an expired row is invalid, so a patch of it is a 404 —
775 : // the same answer the retrieve it would follow already gives.
776 2545 : let row = sqlx::query(
777 2545 : "SELECT entity FROM entities WHERE tenant_id = $1 AND id = $2
778 2545 : AND (expires_at IS NULL OR expires_at > now()) FOR UPDATE",
779 2545 : )
780 2545 : .bind(tenant.as_str())
781 2545 : .bind(id)
782 2545 : .fetch_optional(&mut *tx)
783 2545 : .await?;
784 2545 : let Some(row) = row else {
785 2425 : tx.commit().await?;
786 2425 : return Ok(None);
787 : };
788 120 : let mut doc: Value = row.get(0);
789 120 : let before = self.outbox_on().then(|| doc.clone());
790 120 : match f(&mut doc) {
791 109 : Ok(t) => {
792 109 : let e = extract(&doc);
793 109 : let updated = sqlx::query(
794 109 : "UPDATE entities SET entity = $3, types = $4, scopes = $5,
795 109 : modified_at = $6::timestamptz, expires_at = $7::timestamptz,
796 109 : location = CASE WHEN ST_IsValid(try_geomfromgeojson($8))
797 109 : THEN try_geomfromgeojson($8) END,
798 109 : location_ambiguous = $9 OR ($8 IS NOT NULL
799 109 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson($8)), false)),
800 109 : version = version + 1
801 109 : WHERE tenant_id = $1 AND id = $2
802 109 : RETURNING version, created_at::text",
803 109 : )
804 109 : .bind(tenant.as_str())
805 109 : .bind(id)
806 109 : .bind(&doc)
807 109 : .bind(&e.types)
808 109 : .bind(&e.scopes)
809 109 : .bind(&e.modified)
810 109 : .bind(&e.expires)
811 109 : .bind(&e.location)
812 109 : .bind(e.location_ambiguous)
813 109 : .fetch_one(&mut *tx)
814 109 : .await?;
815 109 : if let Some(before) = &before {
816 0 : enqueue_change(
817 0 : &mut tx,
818 0 : tenant,
819 0 : "update",
820 0 : id,
821 0 : &e.types,
822 0 : Some(before),
823 0 : Some(&doc),
824 0 : updated.get::<i64, _>(0),
825 0 : updated.get::<&str, _>(1),
826 0 : )
827 0 : .await?;
828 109 : }
829 109 : tx.commit().await?;
830 109 : Ok(Some(Ok(t)))
831 : }
832 11 : Err(e) => {
833 11 : tx.rollback().await?;
834 11 : Ok(Some(Err(e)))
835 : }
836 : }
837 2545 : }
838 :
839 : /// Batch create: ONE multi-row INSERT for the whole batch (the jsonb
840 : /// elements form of UNNEST), one transaction, one commit.
841 : /// Returns a created-flag per input item, input order preserved.
842 : /// Duplicate ids within one batch are pre-deduped here: `ON CONFLICT`
843 : /// raises "cannot affect row a second time" otherwise — the later
844 : /// duplicate reports `false` (5.5.11.1: the first instance wins).
845 : ///
846 : /// 5.6.7.4: "For each of the NGSI-LD Entities included in the input
847 : /// Array execute the behaviour defined by clause 5.6.1, but limited to a
848 : /// local operation" — so the conflict clause is `create`'s, and an
849 : /// entity past its `expiresAt` is absent (4.22) and gets created over
850 : /// rather than reported as an id that already exists.
851 7 : pub async fn batch_create(
852 7 : &self,
853 7 : tenant: &TenantId,
854 7 : items: &[(String, Value)],
855 7 : ) -> Result<Vec<bool>, sqlx::Error> {
856 7 : let mut seen = std::collections::HashSet::new();
857 7 : let mut payload = Vec::new();
858 9 : for (id, doc) in items {
859 9 : if seen.insert(id.as_str()) {
860 8 : let e = extract(doc);
861 8 : payload.push(serde_json::json!({
862 8 : "id": id, "doc": doc, "types": e.types, "scopes": e.scopes,
863 8 : "created": e.created, "modified": e.modified, "expires": e.expires,
864 8 : "location": e.location, "loc_ambiguous": e.location_ambiguous,
865 8 : }));
866 8 : }
867 : }
868 7 : let mut tx = super::begin(&self.pool).await?;
869 7 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
870 7 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
871 7 : let rows = sqlx::query(
872 7 : "INSERT INTO entities
873 7 : (tenant_id, id, entity, types, scopes, created_at, modified_at, expires_at,
874 7 : location, location_ambiguous)
875 7 : SELECT $1, e->>'id', e->'doc',
876 7 : ARRAY(SELECT jsonb_array_elements_text(e->'types')),
877 7 : CASE WHEN e->'scopes' = 'null'::jsonb THEN NULL
878 7 : ELSE ARRAY(SELECT jsonb_array_elements_text(e->'scopes')) END,
879 7 : (e->>'created')::timestamptz, (e->>'modified')::timestamptz,
880 7 : (e->>'expires')::timestamptz,
881 7 : CASE WHEN ST_IsValid(try_geomfromgeojson(e->>'location'))
882 7 : THEN try_geomfromgeojson(e->>'location') END,
883 7 : COALESCE((e->>'loc_ambiguous')::bool, false)
884 7 : OR (e->>'location' IS NOT NULL
885 7 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson(e->>'location')), false))
886 7 : FROM jsonb_array_elements($2::jsonb) AS e
887 7 : ON CONFLICT (tenant_id, id) DO UPDATE SET
888 7 : entity = EXCLUDED.entity, types = EXCLUDED.types,
889 7 : scopes = EXCLUDED.scopes, created_at = EXCLUDED.created_at,
890 7 : modified_at = EXCLUDED.modified_at,
891 7 : expires_at = EXCLUDED.expires_at, location = EXCLUDED.location,
892 7 : location_ambiguous = EXCLUDED.location_ambiguous,
893 7 : version = 1
894 7 : WHERE entities.expires_at IS NOT NULL AND entities.expires_at <= now()
895 7 : RETURNING id",
896 7 : )
897 7 : .bind(tenant.as_str())
898 7 : .bind(Value::Array(payload))
899 7 : .fetch_all(&mut *tx)
900 7 : .await?;
901 7 : let created_now: std::collections::HashSet<String> =
902 7 : rows.into_iter().map(|r| r.get::<String, _>(0)).collect();
903 7 : if self.outbox_on() {
904 0 : let mut seen_ev = std::collections::HashSet::new();
905 0 : let mut events = Vec::new();
906 0 : for (id, doc) in items {
907 0 : if created_now.contains(id.as_str()) && seen_ev.insert(id.as_str()) {
908 0 : let e = extract(doc);
909 0 : events.push(change_event(
910 0 : tenant,
911 0 : "create",
912 0 : id,
913 0 : &e.types,
914 0 : None,
915 0 : Some(doc),
916 0 : 1,
917 0 : &e.created,
918 0 : ));
919 0 : }
920 : }
921 0 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
922 7 : }
923 7 : tx.commit().await?;
924 7 : let mut created = created_now;
925 : // consume-once: a duplicate of a created id still reports false
926 7 : Ok(items
927 7 : .iter()
928 9 : .map(|(id, _)| created.remove(id.as_str()))
929 7 : .collect())
930 7 : }
931 :
932 : /// Batch delete: ONE statement, returning each deleted row's previous
933 : /// document (the change-hook before-image).
934 : ///
935 : /// 5.6.10.4: "For each of the NGSI-LD Entity IDs included in the input
936 : /// Array execute the behaviour defined by clause 5.6.6, but limited to a
937 : /// local operation" — so an entity past its `expiresAt` is absent (4.22)
938 : /// and is not deleted here either, exactly as `delete` refuses it.
939 93 : pub async fn batch_delete(
940 93 : &self,
941 93 : tenant: &TenantId,
942 93 : ids: &[String],
943 93 : ) -> Result<Vec<(String, Value)>, sqlx::Error> {
944 93 : let mut tx = super::begin(&self.pool).await?;
945 93 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
946 93 : let rows = sqlx::query(
947 93 : "DELETE FROM entities WHERE tenant_id = $1 AND id = ANY($2)
948 93 : AND (expires_at IS NULL OR expires_at > now())
949 93 : RETURNING id, entity, types, version, created_at::text",
950 93 : )
951 93 : .bind(tenant.as_str())
952 93 : .bind(ids)
953 93 : .fetch_all(&mut *tx)
954 93 : .await?;
955 93 : if self.outbox_on() {
956 0 : let events: Vec<Value> = rows
957 0 : .iter()
958 0 : .map(|r| {
959 0 : let id: String = r.get(0);
960 0 : let prev: Value = r.get(1);
961 0 : let types: Vec<String> = r.get(2);
962 0 : change_event(
963 0 : tenant,
964 0 : "delete",
965 0 : &id,
966 0 : &types,
967 0 : Some(&prev),
968 0 : None,
969 0 : r.get::<i64, _>(3),
970 0 : r.get::<&str, _>(4),
971 : )
972 0 : })
973 0 : .collect();
974 0 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
975 93 : }
976 93 : tx.commit().await?;
977 93 : Ok(rows
978 93 : .into_iter()
979 1210 : .map(|r| (r.get::<String, _>(0), r.get::<Value, _>(1)))
980 93 : .collect())
981 93 : }
982 :
983 : /// Batch upsert in REPLACE semantics — ONE
984 : /// `INSERT … ON CONFLICT DO UPDATE` for the whole batch, one transaction.
985 : /// Returns a created-flag per input item (input order; duplicates of one
986 : /// id report the first outcome). Before-images for the change events are
987 : /// captured by a single `FOR UPDATE` select in the same transaction.
988 : /// Returns per input item: (created?, before-image) — the before-image is
989 : /// for the caller's change hook and comes from the same-tx FOR UPDATE.
990 : ///
991 : /// 5.6.8.4: "Create the Entity locally if it does not exist (i.e. no
992 : /// Entity with the same Entity ID is present) executing the behaviour
993 : /// defined by clause 5.6.1, but limited to a local operation" — an
994 : /// entity past its `expiresAt` does not exist (4.22), so it takes the
995 : /// creation branch: fresh `created_at`, `version` back to 1 and a
996 : /// created-flag, which is what splits 201 from 204 at the API.
997 5 : pub async fn batch_upsert_replace(
998 5 : &self,
999 5 : tenant: &TenantId,
1000 5 : items: &[(String, Value)],
1001 5 : ) -> Result<Vec<(bool, Option<Value>)>, sqlx::Error> {
1002 5 : let mut seen = std::collections::HashSet::new();
1003 5 : let mut payload = Vec::new();
1004 5 : let mut ids = Vec::new();
1005 8 : for (id, doc) in items {
1006 8 : if seen.insert(id.as_str()) {
1007 6 : let e = extract(doc);
1008 6 : ids.push(id.clone());
1009 6 : payload.push(serde_json::json!({
1010 6 : "id": id, "doc": doc, "types": e.types, "scopes": e.scopes,
1011 6 : "created": e.created, "modified": e.modified, "expires": e.expires,
1012 6 : "location": e.location, "loc_ambiguous": e.location_ambiguous,
1013 6 : }));
1014 6 : }
1015 : }
1016 5 : let mut tx = super::begin(&self.pool).await?;
1017 5 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1018 5 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
1019 : // lock + before-images in one statement (ordered: stable lock order)
1020 5 : let prev_rows = sqlx::query(
1021 5 : "SELECT id, entity,
1022 5 : (expires_at IS NOT NULL AND expires_at <= now()) AS expired
1023 5 : FROM entities WHERE tenant_id = $1 AND id = ANY($2)
1024 5 : ORDER BY id FOR UPDATE",
1025 5 : )
1026 5 : .bind(tenant.as_str())
1027 5 : .bind(&ids)
1028 5 : .fetch_all(&mut *tx)
1029 5 : .await?;
1030 : // 4.22: an expired row is already invalid, so it is not a
1031 : // before-image — the upsert that replaces it is a creation and
1032 : // its change event has nothing before it. The lock still covers
1033 : // the row, so the stable lock order is unchanged.
1034 5 : let prevs: std::collections::HashMap<String, Value> = prev_rows
1035 5 : .into_iter()
1036 5 : .filter(|r| !r.get::<bool, _>(2))
1037 5 : .map(|r| (r.get::<String, _>(0), r.get::<Value, _>(1)))
1038 5 : .collect();
1039 5 : let rows = sqlx::query(
1040 5 : "INSERT INTO entities
1041 5 : (tenant_id, id, entity, types, scopes, created_at, modified_at, expires_at,
1042 5 : location, location_ambiguous)
1043 5 : SELECT $1, e->>'id', e->'doc',
1044 5 : ARRAY(SELECT jsonb_array_elements_text(e->'types')),
1045 5 : CASE WHEN e->'scopes' = 'null'::jsonb THEN NULL
1046 5 : ELSE ARRAY(SELECT jsonb_array_elements_text(e->'scopes')) END,
1047 5 : (e->>'created')::timestamptz, (e->>'modified')::timestamptz,
1048 5 : (e->>'expires')::timestamptz,
1049 5 : CASE WHEN ST_IsValid(try_geomfromgeojson(e->>'location'))
1050 5 : THEN try_geomfromgeojson(e->>'location') END,
1051 5 : COALESCE((e->>'loc_ambiguous')::bool, false)
1052 5 : OR (e->>'location' IS NOT NULL
1053 5 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson(e->>'location')), false))
1054 5 : FROM jsonb_array_elements($2::jsonb) AS e
1055 5 : ON CONFLICT (tenant_id, id) DO UPDATE SET
1056 5 : entity = EXCLUDED.entity, types = EXCLUDED.types,
1057 5 : scopes = EXCLUDED.scopes, modified_at = EXCLUDED.modified_at,
1058 5 : expires_at = EXCLUDED.expires_at, location = EXCLUDED.location,
1059 5 : location_ambiguous = EXCLUDED.location_ambiguous,
1060 5 : created_at = CASE WHEN entities.expires_at IS NOT NULL
1061 5 : AND entities.expires_at <= now()
1062 5 : THEN EXCLUDED.created_at
1063 5 : ELSE entities.created_at END,
1064 5 : version = CASE WHEN entities.expires_at IS NOT NULL
1065 5 : AND entities.expires_at <= now()
1066 5 : THEN 1 ELSE entities.version + 1 END
1067 5 : RETURNING id, (xmax = 0 OR version = 1) AS inserted,
1068 5 : version, created_at::text",
1069 5 : )
1070 5 : .bind(tenant.as_str())
1071 5 : .bind(Value::Array(payload))
1072 5 : .fetch_all(&mut *tx)
1073 5 : .await?;
1074 5 : let mut created: std::collections::HashMap<String, bool> = std::collections::HashMap::new();
1075 5 : let mut events = Vec::new();
1076 6 : for r in &rows {
1077 6 : let id: String = r.get(0);
1078 6 : let inserted: bool = r.get(1);
1079 : // the id came out of `items`, so the lookup answers; an
1080 : // outbox row for a document that is not there would carry no
1081 : // payload, so there is nothing to emit either way
1082 0 : if let (true, Some(doc)) = (
1083 6 : self.outbox_on(),
1084 8 : items.iter().find(|(i, _)| *i == id).map(|(_, d)| d),
1085 : ) {
1086 0 : let e = extract(doc);
1087 0 : events.push(change_event(
1088 0 : tenant,
1089 0 : if inserted { "create" } else { "update" },
1090 0 : &id,
1091 0 : &e.types,
1092 0 : prevs.get(&id),
1093 0 : Some(doc),
1094 0 : r.get::<i64, _>(2),
1095 0 : r.get::<&str, _>(3),
1096 : ));
1097 6 : }
1098 6 : created.insert(id, inserted);
1099 : }
1100 5 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
1101 5 : tx.commit().await?;
1102 5 : Ok(items
1103 5 : .iter()
1104 8 : .map(|(id, _)| {
1105 8 : (
1106 8 : created.get(id).copied().unwrap_or(false),
1107 8 : prevs.get(id).cloned(),
1108 8 : )
1109 8 : })
1110 5 : .collect())
1111 5 : }
1112 :
1113 : /// Batch read-modify-write — ONE transaction,
1114 : /// all rows locked in a single ordered `FOR UPDATE` select, the closure
1115 : /// applied per doc in Rust, and ONE multi-row UPDATE writeback. Per-item
1116 : /// results align with `ids`: `None` = absent, `Some(Err)` = the closure
1117 : /// rejected that item (its write is skipped, the rest proceed).
1118 5 : pub async fn batch_mutate<E>(
1119 5 : &self,
1120 5 : tenant: &TenantId,
1121 5 : ids: &[String],
1122 5 : mut f: impl FnMut(&str, &mut Value) -> Result<(), E>,
1123 5 : ) -> Result<Vec<Option<Result<(), E>>>, sqlx::Error> {
1124 5 : let mut tx = super::begin(&self.pool).await?;
1125 5 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1126 5 : let rows = sqlx::query(
1127 5 : // 4.22: an expired entity is invalid, so it is not a row this
1128 5 : // update can find — 5.6.9.4 is clause 5.6.3 run per item, and
1129 5 : // 5.6.3 on an absent entity is ResourceNotFound. Without the
1130 5 : // predicate the batch reported the id as updated, wrote to
1131 5 : // it, and emitted a change event for an entity every read
1132 5 : // refuses. ORDER BY id keeps the lock order stable.
1133 5 : "SELECT id, entity FROM entities
1134 5 : WHERE tenant_id = $1 AND id = ANY($2)
1135 5 : AND (expires_at IS NULL OR expires_at > now())
1136 5 : ORDER BY id FOR UPDATE",
1137 5 : )
1138 5 : .bind(tenant.as_str())
1139 5 : .bind(ids)
1140 5 : .fetch_all(&mut *tx)
1141 5 : .await?;
1142 5 : let mut docs: std::collections::HashMap<String, Value> = rows
1143 5 : .into_iter()
1144 6 : .map(|r| (r.get::<String, _>(0), r.get::<Value, _>(1)))
1145 5 : .collect();
1146 5 : let mut results: Vec<Option<Result<(), E>>> = Vec::with_capacity(ids.len());
1147 : // 5.6.9.4 is clause 5.6.3 run per item, so a repeated id applies
1148 : // its closure once per item onto the SAME document and the last
1149 : // application is the state the row must end at. The writeback
1150 : // below is one `UPDATE … FROM`, which updates a target row once
1151 : // from whichever source row the join happens to reach, so the
1152 : // statement is fed one row per id — the document as the whole
1153 : // batch left it — rather than one per input item.
1154 5 : let mut write_ids: std::collections::HashSet<&str> = std::collections::HashSet::new();
1155 5 : let mut changed: Vec<(String, Value, Value)> = Vec::new(); // id, before, after
1156 10 : for id in ids {
1157 10 : match docs.get_mut(id) {
1158 4 : None => results.push(None),
1159 6 : Some(doc) => {
1160 6 : let before = doc.clone();
1161 6 : match f(id, doc) {
1162 0 : Err(e) => {
1163 0 : *doc = before; // closure failed: discard its edits
1164 0 : results.push(Some(Err(e)));
1165 0 : }
1166 : Ok(()) => {
1167 6 : if *doc != before {
1168 5 : write_ids.insert(id.as_str());
1169 5 : changed.push((id.clone(), before, doc.clone()));
1170 5 : }
1171 6 : results.push(Some(Ok(())));
1172 : }
1173 : }
1174 : }
1175 : }
1176 : }
1177 5 : let payload: Vec<Value> = write_ids
1178 5 : .iter()
1179 6 : .filter_map(|id| docs.get(*id).map(|doc| (*id, doc)))
1180 6 : .map(|(id, doc)| {
1181 5 : let e = extract(doc);
1182 5 : serde_json::json!({
1183 5 : "id": id, "doc": doc, "types": e.types,
1184 5 : "scopes": e.scopes, "modified": e.modified,
1185 5 : "expires": e.expires, "location": e.location,
1186 5 : "loc_ambiguous": e.location_ambiguous,
1187 : })
1188 5 : })
1189 5 : .collect();
1190 5 : if !payload.is_empty() {
1191 3 : let updated = sqlx::query(
1192 3 : "UPDATE entities t SET
1193 3 : entity = e->'doc',
1194 3 : types = ARRAY(SELECT jsonb_array_elements_text(e->'types')),
1195 3 : scopes = CASE WHEN e->'scopes' = 'null'::jsonb THEN NULL
1196 3 : ELSE ARRAY(SELECT jsonb_array_elements_text(e->'scopes')) END,
1197 3 : modified_at = (e->>'modified')::timestamptz,
1198 3 : expires_at = (e->>'expires')::timestamptz,
1199 3 : location = CASE WHEN ST_IsValid(try_geomfromgeojson(e->>'location'))
1200 3 : THEN try_geomfromgeojson(e->>'location') END,
1201 3 : location_ambiguous = COALESCE((e->>'loc_ambiguous')::bool, false)
1202 3 : OR (e->>'location' IS NOT NULL
1203 3 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson(e->>'location')), false)),
1204 3 : version = t.version + 1
1205 3 : FROM jsonb_array_elements($2::jsonb) AS e
1206 3 : WHERE t.tenant_id = $1 AND t.id = e->>'id'
1207 3 : RETURNING t.id, t.version, t.created_at::text",
1208 3 : )
1209 3 : .bind(tenant.as_str())
1210 3 : .bind(Value::Array(payload))
1211 3 : .fetch_all(&mut *tx)
1212 3 : .await?;
1213 3 : if self.outbox_on() {
1214 0 : let meta: std::collections::HashMap<String, (i64, String)> = updated
1215 0 : .into_iter()
1216 0 : .map(|r| {
1217 0 : (
1218 0 : r.get::<String, _>(0),
1219 0 : (r.get::<i64, _>(1), r.get::<String, _>(2)),
1220 0 : )
1221 0 : })
1222 0 : .collect();
1223 0 : let events: Vec<Value> = changed
1224 0 : .iter()
1225 0 : .filter_map(|(id, before, after)| {
1226 0 : let e = extract(after);
1227 0 : let (version, incarnation) = meta.get(id)?;
1228 0 : Some(change_event(
1229 0 : tenant,
1230 0 : "update",
1231 0 : id,
1232 0 : &e.types,
1233 0 : Some(before),
1234 0 : Some(after),
1235 0 : *version,
1236 0 : incarnation,
1237 0 : ))
1238 0 : })
1239 0 : .collect();
1240 0 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
1241 3 : }
1242 2 : }
1243 5 : tx.commit().await?;
1244 5 : Ok(results)
1245 5 : }
1246 :
1247 : /// Current row version (test hook for the version-monotonicity assertions).
1248 : #[cfg(any(test, feature = "test-kit"))]
1249 4 : pub async fn version(&self, tenant: &TenantId, id: &str) -> Result<Option<i64>, sqlx::Error> {
1250 4 : let mut tx = super::begin(&self.pool).await?;
1251 4 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1252 4 : let row = sqlx::query("SELECT version FROM entities WHERE tenant_id = $1 AND id = $2")
1253 4 : .bind(tenant.as_str())
1254 4 : .bind(id)
1255 4 : .fetch_optional(&mut *tx)
1256 4 : .await?;
1257 4 : tx.commit().await?;
1258 4 : Ok(row.map(|r| r.get::<i64, _>(0)))
1259 4 : }
1260 : }
1261 :
1262 : #[cfg(test)]
1263 : mod tests {
1264 : use super::*;
1265 : use serde_json::json;
1266 :
1267 12 : fn wheres() -> Vec<String> {
1268 12 : vec!["tenant_id = $1".to_owned()]
1269 12 : }
1270 :
1271 : /// `location_ambiguous` is the single bit that keeps geoquery pushdown
1272 : /// honest: `compile_geo` ORs it into every predicate so a row whose
1273 : /// default GeoProperty could not be reduced to ONE geometry still reaches
1274 : /// the evaluator. It must be true exactly when the doc CARRIES the
1275 : /// geoproperty and extraction refused — a row with no geoproperty at all
1276 : /// can never match a geoquery and is excluded in SQL, which is the whole
1277 : /// point of the column.
1278 : #[test]
1279 2 : fn location_ambiguous_is_carried_but_unextractable() {
1280 2 : let loc = crate::compile::geo::LOCATION_IRI;
1281 2 : let point = json!({"type": "Point", "coordinates": [2.29, 48.85]});
1282 :
1283 : // one default GeoProperty instance with a geometry: extracted, exact
1284 2 : let e = extract(&json!({"id": "urn:x", loc: [{"value": point}]}));
1285 2 : assert!(e.location.is_some());
1286 2 : assert!(!e.location_ambiguous);
1287 :
1288 : // multi-instance: no single geometry, so the evaluator arbitrates
1289 2 : let e = extract(&json!({"id": "urn:x", loc: [{"value": point}, {"value": point}]}));
1290 2 : assert!(e.location.is_none());
1291 2 : assert!(
1292 2 : e.location_ambiguous,
1293 : "a multi-instance location must reach the evaluator"
1294 : );
1295 :
1296 : // a GeometryCollection and a non-GeoJSON value are the same case
1297 6 : for v in [
1298 2 : json!({"type": "GeometryCollection", "geometries": []}),
1299 2 : json!("somewhere"),
1300 2 : json!({"coordinates": [1, 2]}),
1301 2 : ] {
1302 6 : let e = extract(&json!({"id": "urn:x", loc: [{"value": v}]}));
1303 6 : assert!(e.location.is_none(), "{v} must not become a geometry");
1304 6 : assert!(e.location_ambiguous, "{v} must reach the evaluator");
1305 : }
1306 :
1307 : // NO geoproperty: not ambiguous — the row is excluded in SQL
1308 2 : let e = extract(&json!({"id": "urn:x", "https://a/speed": [{"value": 1}]}));
1309 2 : assert!(e.location.is_none());
1310 2 : assert!(
1311 2 : !e.location_ambiguous,
1312 : "a row without a location must stay excludable in SQL"
1313 : );
1314 2 : }
1315 :
1316 : /// The rest of `extract`: `type`/`scope` accept the string form and the
1317 : /// array form, an absent `scope` is SQL NULL (never an empty array, which
1318 : /// would mean "scoped to nothing"), and a missing system timestamp falls
1319 : /// back rather than failing the write.
1320 : #[test]
1321 2 : fn extract_reads_types_scopes_and_stamps() {
1322 2 : let e = extract(&json!({"id": "urn:x", "type": "T", "scope": "/a"}));
1323 2 : assert_eq!(e.types, ["T"]);
1324 2 : assert_eq!(e.scopes.as_deref(), Some(&["/a".to_owned()][..]));
1325 2 : let e = extract(&json!({"id": "urn:x", "type": ["T", "U"], "scope": ["/a", "/b"]}));
1326 2 : assert_eq!(e.types, ["T", "U"]);
1327 2 : assert_eq!(e.scopes.expect("scopes").len(), 2);
1328 2 : let e = extract(&json!({"id": "urn:x"}));
1329 2 : assert!(e.types.is_empty());
1330 2 : assert!(e.scopes.is_none(), "absent scope must be SQL NULL");
1331 2 : assert!(e.expires.is_none());
1332 2 : assert_eq!(e.created, "1970-01-01T00:00:00Z");
1333 : // a non-string, non-array scope is present but names nothing
1334 2 : let e = extract(&json!({"id": "urn:x", "scope": 7}));
1335 2 : assert_eq!(e.scopes.as_deref(), Some(&[][..]));
1336 2 : }
1337 :
1338 : /// The outbox row IS the wire contract the drain deserializes, and the
1339 : /// drain DELETES any row it cannot decode — a renamed key loses the event
1340 : /// silently. Pin the key set and the operation vocabulary here, where the
1341 : /// producer lives.
1342 : #[test]
1343 2 : fn the_change_event_carries_exactly_the_wire_keys() {
1344 2 : let t = TenantId::default();
1345 2 : let doc = json!({"id": "urn:e", "https://a/speed": [{"value": 1}]});
1346 2 : let ev = change_event(
1347 2 : &t,
1348 2 : "create",
1349 2 : "urn:e",
1350 2 : &["T".to_owned()],
1351 2 : None,
1352 2 : Some(&doc),
1353 : 1,
1354 2 : "inc",
1355 : );
1356 2 : let keys: Vec<&str> = ev
1357 2 : .as_object()
1358 2 : .expect("object")
1359 2 : .keys()
1360 2 : .map(String::as_str)
1361 2 : .collect();
1362 2 : assert_eq!(
1363 : keys,
1364 : [
1365 : "changed_attrs",
1366 : "entity_id",
1367 : "incarnation",
1368 : "op",
1369 : "payload",
1370 : "prev_payload",
1371 : "tenant",
1372 : "types",
1373 : "version",
1374 : ]
1375 : );
1376 2 : assert_eq!(ev["op"], "create");
1377 2 : assert_eq!(ev["tenant"], t.as_str());
1378 2 : assert!(ev["prev_payload"].is_null(), "a create has no before-image");
1379 6 : for op in ["create", "update", "delete"] {
1380 6 : assert_eq!(
1381 6 : change_event(&t, op, "urn:e", &[], None, None, 1, "inc")["op"],
1382 : op
1383 : );
1384 : }
1385 2 : }
1386 :
1387 : /// `changed_attrs` names the top-level ATTRIBUTES that differ. A create
1388 : /// lists every attribute, a delete lists every prior one — and, the half
1389 : /// that actually bounds notification traffic, an attribute present in
1390 : /// both images with an EQUAL value is not listed, nor is any system
1391 : /// member that changed on every write.
1392 : #[test]
1393 2 : fn changed_attrs_lists_the_differences_and_nothing_else() {
1394 2 : let t = TenantId::default();
1395 10 : let ev = |prev: Option<&Value>, next: Option<&Value>| {
1396 10 : change_event(&t, "update", "urn:e", &[], prev, next, 1, "inc")["changed_attrs"].clone()
1397 10 : };
1398 2 : let a = json!({"id": "urn:e", "type": ["T"], "createdAt": "t0", "modifiedAt": "t0",
1399 2 : "https://a/x": [{"value": 1}], "https://a/y": [{"value": 2}]});
1400 2 : let b = json!({"id": "urn:e", "type": ["T"], "createdAt": "t0", "modifiedAt": "t9",
1401 2 : "https://a/x": [{"value": 1}], "https://a/y": [{"value": 3}]});
1402 2 : assert_eq!(ev(Some(&a), Some(&b)), json!(["https://a/y"]));
1403 2 : assert_eq!(ev(None, Some(&a)), json!(["https://a/x", "https://a/y"]));
1404 2 : assert_eq!(ev(Some(&a), None), json!(["https://a/x", "https://a/y"]));
1405 : // identical images change nothing at all
1406 2 : assert_eq!(ev(Some(&a), Some(&a)), json!([]));
1407 : // and no system member is ever named
1408 2 : let ids = json!({"id": "urn:other", "type": ["U"], "scope": "/s",
1409 2 : "modifiedAt": "t1", "deletedAt": "t1", "expiresAt": "t1"});
1410 2 : assert_eq!(
1411 2 : ev(Some(&a), Some(&ids)),
1412 2 : json!(["https://a/x", "https://a/y"])
1413 : );
1414 2 : }
1415 :
1416 4 : fn page(offset: i64, limit: i64) -> Page {
1417 4 : Page {
1418 4 : offset,
1419 4 : limit,
1420 4 : count: true,
1421 4 : }
1422 4 : }
1423 :
1424 : /// 5.5.9.1: a query SQL decided exactly keeps the pushed page — LIMIT
1425 : /// and OFFSET, with the window total that feeds count= and the
1426 : /// next/prev pointers.
1427 : #[test]
1428 2 : fn a_decided_page_pushes_limit_and_offset() {
1429 2 : let mut binds = vec![Bind::Text("t".to_owned())];
1430 2 : let (sql, paged) = query_sql("entity", &wheres(), Some(&page(20, 10)), true, &mut binds);
1431 2 : assert!(paged);
1432 2 : assert!(sql.contains("LIMIT $2 OFFSET $3"), "{sql}");
1433 2 : assert!(sql.contains("count(*) OVER ()"), "{sql}");
1434 2 : assert!(matches!(binds[1], Bind::Int(10)));
1435 2 : assert!(matches!(binds[2], Bind::Int(20)));
1436 2 : }
1437 :
1438 : /// 6.3.10: without count=true no total is owed, so the statement carries
1439 : /// no window count (which forced a walk of the whole match set) and
1440 : /// fetches one row past the page to learn whether a next page exists.
1441 : #[test]
1442 2 : fn an_uncounted_page_fetches_one_extra_row_and_no_window_total() {
1443 2 : let mut binds = vec![Bind::Text("t".to_owned())];
1444 2 : let p = Page {
1445 2 : offset: 20,
1446 2 : limit: 10,
1447 2 : count: false,
1448 2 : };
1449 2 : let (sql, paged) = query_sql("entity", &wheres(), Some(&p), true, &mut binds);
1450 2 : assert!(paged);
1451 2 : assert!(!sql.contains("count(*) OVER ()"), "{sql}");
1452 2 : assert!(sql.contains("LIMIT $2 OFFSET $3"), "{sql}");
1453 2 : assert!(matches!(binds[1], Bind::Int(11)), "limit + 1 must be bound");
1454 2 : assert!(matches!(binds[2], Bind::Int(20)));
1455 2 : }
1456 :
1457 : /// The largest page a caller can build still has to produce a usable
1458 : /// LIMIT. `types_attrs.rs` clamps its discovery scan to `i64::MAX` on
1459 : /// purpose so the cast is safe; the uncounted path then binds
1460 : /// `limit + 1`, which is where that clamp stops being enough. The
1461 : /// addition overflows: a panic in a debug build, and in release a wrap
1462 : /// to `i64::MIN`, which Postgres refuses as a negative LIMIT.
1463 : #[test]
1464 2 : fn the_largest_page_still_binds_a_usable_limit() {
1465 2 : let mut binds = vec![Bind::Text("t".to_owned())];
1466 2 : let p = Page {
1467 2 : offset: 0,
1468 2 : limit: i64::MAX,
1469 2 : count: false,
1470 2 : };
1471 2 : let (sql, paged) = query_sql("entity", &wheres(), Some(&p), true, &mut binds);
1472 2 : assert!(paged);
1473 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
1474 2 : assert!(
1475 2 : matches!(binds[1], Bind::Int(i64::MAX)),
1476 : "the bound limit must stay positive"
1477 : );
1478 2 : }
1479 :
1480 : /// The undecided path must never leave the statement unbounded: SQL only
1481 : /// narrowed the set (`scopeQ`, `georel`, a `q=` the compiler declined),
1482 : /// so the rows come back to be filtered — under a LIMIT, and without an
1483 : /// OFFSET, which would page over the wrong set.
1484 : #[test]
1485 2 : fn an_undecided_query_carries_a_safety_limit_and_no_offset() {
1486 2 : let mut binds = vec![Bind::Text("t".to_owned())];
1487 2 : let (sql, paged) = query_sql("entity", &wheres(), Some(&page(20, 10)), false, &mut binds);
1488 2 : assert!(!paged);
1489 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
1490 2 : assert!(!sql.contains("OFFSET"), "offset cannot be pushed: {sql}");
1491 2 : assert!(!sql.contains("count(*) OVER ()"), "{sql}");
1492 : // the candidate ceiling, not the page: the caller's evaluator still
1493 : // filters these rows, so a page-sized fetch would refuse legitimate
1494 : // queries whenever candidates outnumber matches
1495 2 : assert!(matches!(binds[1], Bind::Int(MAX_UNDECIDED_ROWS)));
1496 2 : }
1497 :
1498 : /// A caller with no page (idPattern, federation, orderBy) hands the store
1499 : /// no bound at all — the ceiling is then the only one.
1500 : #[test]
1501 2 : fn a_query_without_a_page_falls_back_to_the_ceiling() {
1502 2 : let mut binds = vec![Bind::Text("t".to_owned())];
1503 2 : let (sql, paged) = query_sql("entity", &wheres(), None, false, &mut binds);
1504 2 : assert!(!paged);
1505 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
1506 2 : assert!(matches!(binds[1], Bind::Int(MAX_UNDECIDED_ROWS)));
1507 : // an exactly decided query with no page is materialized whole too
1508 2 : let mut binds = vec![Bind::Text("t".to_owned())];
1509 2 : let (sql, _) = query_sql("entity", &wheres(), None, true, &mut binds);
1510 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
1511 2 : }
1512 :
1513 : /// 5.5.6 / Table 6.3.2-1: reaching the ceiling means the statement was
1514 : /// truncated, so the page built from it would silently under-report —
1515 : /// the operation is refused with TooManyResults (403) instead.
1516 : #[test]
1517 2 : fn a_ceiling_hit_is_too_many_results_not_a_short_page() {
1518 2 : assert!(check_ceiling(false, 30, 31).is_ok(), "under the ceiling");
1519 2 : let err = check_ceiling(false, 31, 31).expect_err("ceiling reached");
1520 2 : let ngsi = ngsi_error(&err).expect("the spec error travels in the driver error");
1521 2 : assert_eq!(ngsi.kind(), "TooManyResults");
1522 2 : assert_eq!(ngsi.status(), 403);
1523 : // a pushed page is exact by construction: a full page is a page, not
1524 : // a truncation
1525 2 : assert!(check_ceiling(true, 1000, 1000).is_ok());
1526 2 : }
1527 :
1528 : /// A real driver failure must stay a driver failure — only the store's
1529 : /// own spec errors are recovered, or every 500 would become a 403.
1530 : #[test]
1531 2 : fn a_driver_error_is_not_mistaken_for_a_spec_error() {
1532 2 : assert!(ngsi_error(&sqlx::Error::RowNotFound).is_none());
1533 2 : assert!(ngsi_error(&sqlx::Error::PoolTimedOut).is_none());
1534 2 : assert!(ngsi_error(&sqlx::Error::Configuration("boom".into())).is_none());
1535 2 : }
1536 : }
|
