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 antares_ql::{CmpOp, QNode, QValue};
16 : use antares_store::ChangeVersion;
17 : use serde_json::Value;
18 : use sqlx::postgres::PgPool;
19 : use sqlx::Row;
20 :
21 : pub struct PgEntityStore {
22 : pool: PgPool,
23 : /// When on, every entity write enqueues its change event into the
24 : /// outbox INSIDE the write transaction (a crash between commit and
25 : /// publish can never lose an event). Off by default: with `bus = local`
26 : /// events flow through the in-process hook and undrained rows would only
27 : /// grow the table. The broker turns this on when `bus = nats`.
28 : outbox: std::sync::atomic::AtomicBool,
29 : /// When on, the batch writes (create, upsert, the batch read-modify-write)
30 : /// commit with `synchronous_commit = off`, set LOCAL to their own
31 : /// transaction: a crash of Postgres loses at most the last
32 : /// ~3 x `wal_writer_delay` of committed batches, never their atomicity
33 : /// or the order of the WAL. Notifications already sent for a lost batch
34 : /// stay sent. Off by default; the broker sets it from
35 : /// `ANTARES_PG_BATCH_ASYNC_COMMIT` for a feed that rewrites itself every
36 : /// second, where the next batch restores what a crash lost.
37 : async_batch: std::sync::atomic::AtomicBool,
38 : /// The pool the batch writes begin on: `pool` itself, or the write
39 : /// reserve (`ANTARES_PG_WRITE_RESERVE`) when the broker set one,
40 : /// whose connections no read takes.
41 : batch_pool: PgPool,
42 : }
43 :
44 : // EntityFilter/Page/QueryOutcome live in `store::filter` (pure data,
45 : // shared with the wasm32 build); re-exported here so existing paths hold.
46 : pub use crate::store::filter::{EntityFilter, Page, QueryOutcome};
47 :
48 : /// A bound value. Enumerated because the bind list is built dynamically while
49 : /// the SQL is assembled — the alternative would be string interpolation.
50 : enum Bind {
51 : Text(String),
52 : TextArr(Vec<String>),
53 : /// jsonpath; bound as text and cast with `$n::jsonpath` in the SQL
54 : Path(String),
55 : /// a distance in metres (geoquery `near`)
56 : Num(f64),
57 : /// LIMIT/OFFSET (pagination pushdown)
58 : Int(i64),
59 : }
60 :
61 : /// The internal doc's members that become extracted columns.
62 : pub(crate) struct Extracted {
63 : types: Vec<String>,
64 : scopes: Option<Vec<String>>,
65 : created: String,
66 : modified: String,
67 : expires: Option<String>,
68 : /// The default GeoProperty as GeoJSON text, for `ST_GeomFromGeoJSON`.
69 : /// `None` (→ SQL NULL) whenever it cannot be represented as ONE geometry.
70 : location: Option<String>,
71 : /// True when the doc CARRIES the default GeoProperty but `location` could
72 : /// not be extracted (multi-instance, non-GeoJSON value). The compiled
73 : /// geoquery ORs on this column — index-shaped (BitmapOr), unlike the old
74 : /// `location IS NULL OR …` guard which forced a sequential scan — and
75 : /// rows without any geoproperty are excluded in SQL, which is exact.
76 : location_ambiguous: bool,
77 : /// The earliest `expiresAt` of any Attribute instance (4.22), so the
78 : /// sweep finds the rows to prune by index instead of reading every doc.
79 : attrs_expire: Option<String>,
80 : }
81 :
82 : /// The four members every stored kind extracts the same way: `type` and
83 : /// `scope` as arrays whether the doc spells them as one string or many, and
84 : /// the two stamps. The temporal tables carry exactly these, entities carry
85 : /// them plus expiry and location, so this is the one implementation of the
86 : /// shared half — two copies is how one of them drifts.
87 : ///
88 : /// 4.6.3 allows a comma seconds-fraction in requests; the stamps feed
89 : /// `::timestamptz` casts, which refuse it, so they are canonicalised here and
90 : /// every extracted column holds the instant the client meant.
91 46825 : pub(super) fn types_scopes_stamps(
92 46825 : doc: &Value,
93 46825 : ) -> (Vec<String>, Option<Vec<String>>, String, String) {
94 51861 : let as_vec = |v: &Value| -> Vec<String> {
95 51861 : match v {
96 20631 : Value::String(s) => vec![s.clone()],
97 31228 : Value::Array(a) => a
98 31228 : .iter()
99 34290 : .filter_map(|x| x.as_str().map(str::to_owned))
100 31228 : .collect(),
101 2 : _ => vec![],
102 : }
103 51861 : };
104 93650 : let ts = |k: &str| {
105 93650 : doc.get(k)
106 93650 : .and_then(Value::as_str)
107 93650 : .map(|s| antares_store::filter::canonical_datetime(s).into_owned())
108 93650 : };
109 46825 : let now = || "1970-01-01T00:00:00Z".to_owned(); // caller always stamps; belt only
110 46825 : (
111 46825 : doc.get("type").map(&as_vec).unwrap_or_default(),
112 46825 : doc.get("scope").map(&as_vec),
113 46825 : ts("createdAt").unwrap_or_else(now),
114 46825 : ts("modifiedAt").unwrap_or_else(now),
115 46825 : )
116 46825 : }
117 :
118 41798 : fn extract(doc: &Value) -> Extracted {
119 41798 : let (types, scopes, created, modified) = types_scopes_stamps(doc);
120 41798 : let ts = |k: &str| {
121 41798 : doc.get(k)
122 41798 : .and_then(Value::as_str)
123 41798 : .map(|s| antares_store::filter::canonical_datetime(s).into_owned())
124 41798 : };
125 41798 : let location = crate::compile::geo::extract_location(doc);
126 41798 : let location_ambiguous =
127 41798 : location.is_none() && doc.get(crate::compile::geo::LOCATION_IRI).is_some();
128 41798 : Extracted {
129 41798 : types,
130 41798 : scopes,
131 41798 : created,
132 41798 : modified,
133 41798 : expires: ts("expiresAt"),
134 41798 : location,
135 41798 : location_ambiguous,
136 41798 : attrs_expire: first_instance_expiry(doc),
137 41798 : }
138 41798 : }
139 :
140 : /// The earliest instance `expiresAt` of an internal doc's Attributes,
141 : /// canonicalised for the `::timestamptz` cast; an Entity member never counts.
142 41798 : fn first_instance_expiry(doc: &Value) -> Option<String> {
143 41798 : doc.as_object()?
144 41798 : .iter()
145 204256 : .filter(|(k, _)| !antares_model::ENTITY_META_KEYS.contains(&k.as_str()))
146 54735 : .filter_map(|(_, v)| v.as_array())
147 41798 : .flatten()
148 43687 : .filter_map(|inst| inst.get("expiresAt").and_then(Value::as_str))
149 41798 : .min_by_key(|s| antares_model::dt_key(s))
150 41798 : .map(|s| antares_store::filter::canonical_datetime(s).into_owned())
151 41798 : }
152 :
153 : /// The [`ChangeVersion`] of a delete whose `RETURNING` row is `r` (version
154 : /// at 2, `created_at::text` at 3): one past the version it removed, so it
155 : /// orders after that row's last write.
156 2763 : fn deleted_version(r: &sqlx::postgres::PgRow) -> ChangeVersion {
157 2763 : ChangeVersion {
158 2763 : incarnation: r.get(3),
159 2763 : version: r.get::<i64, _>(2).saturating_add(1),
160 2763 : }
161 2763 : }
162 :
163 : /// The outbox row's event JSON: what the drain turns into a
164 : /// `ChangeEvent`. Field names are the bus crate's serde names; `seq` and the
165 : /// claim check are the drain's business.
166 : #[allow(clippy::too_many_arguments)] // one param per outbox event field
167 22 : fn change_event(
168 22 : tenant: &TenantId,
169 22 : op: &str,
170 22 : id: &str,
171 22 : types: &[String],
172 22 : prev: Option<&Value>,
173 22 : next: Option<&Value>,
174 22 : version: i64,
175 22 : incarnation: &str,
176 22 : ) -> Value {
177 : // changed_attrs = top-level attribute IRIs that differ between the
178 : // before- and after-images (meta members excluded). Create lists every
179 : // attr, delete lists every prior attr.
180 44 : fn keys(v: Option<&Value>) -> Vec<&str> {
181 44 : v.and_then(Value::as_object)
182 44 : .map(|o| {
183 22 : o.keys()
184 22 : .map(String::as_str)
185 120 : .filter(|k| !antares_model::is_meta(k))
186 22 : .collect()
187 22 : })
188 44 : .unwrap_or_default()
189 44 : }
190 22 : let mut changed: Vec<&str> = Vec::new();
191 34 : for k in keys(prev).into_iter().chain(keys(next)) {
192 34 : if !changed.contains(&k) && prev.and_then(|p| p.get(k)) != next.and_then(|n| n.get(k)) {
193 20 : changed.push(k);
194 20 : }
195 : }
196 22 : serde_json::json!({
197 22 : "tenant": tenant.as_str(),
198 22 : "entity_id": id,
199 22 : "types": types,
200 22 : "op": op,
201 22 : "changed_attrs": changed,
202 22 : "payload": next,
203 22 : "prev_payload": prev,
204 22 : "version": version,
205 22 : "incarnation": incarnation,
206 : })
207 22 : }
208 :
209 : #[allow(clippy::too_many_arguments)] // one param per outbox event field
210 4 : async fn enqueue_change(
211 4 : tx: &mut sqlx::postgres::PgConnection,
212 4 : tenant: &TenantId,
213 4 : op: &str,
214 4 : id: &str,
215 4 : types: &[String],
216 4 : prev: Option<&Value>,
217 4 : next: Option<&Value>,
218 4 : version: i64,
219 4 : incarnation: &str,
220 4 : ) -> Result<(), sqlx::Error> {
221 4 : let ev = change_event(tenant, op, id, types, prev, next, version, incarnation);
222 4 : super::outbox::enqueue(tx, tenant, &ev).await.map(|_| ())
223 4 : }
224 :
225 : /// Rows one statement may materialize when SQL narrowed the match set but
226 : /// did not decide it. 5.5.6 leaves the threshold to the implementation;
227 : /// this is it for the entity query path, and it is the only bound a request
228 : /// that carries no page (idPattern, federation, orderBy) has at all.
229 : pub const MAX_UNDECIDED_ROWS: i64 = 10_000;
230 :
231 : /// The column `q=` is compiled over: the stored document, or, for a row
232 : /// holding an instance past its `expiresAt` (`attrs_expire_at` past, rare
233 : /// until the sweep rewrites it), the document 4.22 lets a read see: expired
234 : /// top-level instances dropped, an Attribute left with none removed, as
235 : /// `filter::strip_expired` does in memory. A live row pays
236 : /// only the CASE test.
237 : const LIVE_ENTITY: &str = "(CASE WHEN attrs_expire_at IS NULL OR attrs_expire_at >= now() \
238 : THEN entity ELSE (SELECT coalesce(jsonb_object_agg(l.k, l.v), '{}'::jsonb) FROM \
239 : (SELECT x.k, CASE WHEN jsonb_typeof(x.v) = 'array' AND strpos(x.k, ':') > 0 \
240 : THEN (SELECT jsonb_agg(i.v) FROM jsonb_array_elements(x.v) AS i(v) \
241 : WHERE jsonb_typeof(i.v -> 'expiresAt') IS DISTINCT FROM 'string' \
242 : OR try_timestamptz(i.v ->> 'expiresAt') IS NULL \
243 : OR try_timestamptz(i.v ->> 'expiresAt') >= now()) \
244 : ELSE x.v END AS v FROM jsonb_each(entity) AS x(k, v)) AS l \
245 : WHERE l.v IS NOT NULL) END)";
246 :
247 : /// Distinct types a type selection may split into arms; a longer list keeps
248 : /// the plain containment, which reads the Tenant.
249 : const MAX_TYPE_ARMS: usize = 16;
250 :
251 : /// A type selection read as leakproof index arms. `types @> $n` cannot be an
252 : /// index condition: `arraycontains` is not leakproof, so under the tables'
253 : /// RLS Postgres keeps it out of every index condition and pages the whole
254 : /// Tenant through `entities_pkey` (18 req/s; the
255 : /// same for `type=A,B`). The split reads one arm per type the selection
256 : /// names, the Entities whose first type it is (`i_entities_first_type_id`),
257 : /// plus one arm for the few of two or more types whose first type is none
258 : /// of them (`i_entities_multi_type_id`). The arms are disjoint (an Entity
259 : /// has one first type) and their union is exactly the selection. Only a
260 : /// group of one type gets first-type arms: a match of an AND-group (`A;B`)
261 : /// has two types, so the multi-type arm reaches it, where a first-type arm
262 : /// walked every Entity of A to keep none (53 000 s on a deployment).
263 : struct TypeSplit {
264 : /// Bind of each type a one-type group names, one first-type arm each.
265 : firsts: Vec<usize>,
266 : /// Bind of those types as an array, `None` when every group is an
267 : /// AND-group and the multi-type arm alone reads the selection.
268 : named: Option<usize>,
269 : /// The selection itself, `types @> $g` per AND-group, ORed.
270 : selection: String,
271 : /// A selection of one type is decided by `types[1] = $t` alone.
272 : exact_first: bool,
273 : }
274 :
275 : /// The full WHERE of each arm the statement reads: one, or the arms of a
276 : /// `TypeSplit`.
277 5230 : fn arms(wheres: &[String], split: Option<&TypeSplit>) -> Vec<String> {
278 5230 : let base = wheres.join(" AND ");
279 5230 : let Some(s) = split else {
280 2903 : return vec![base];
281 : };
282 2327 : let mut arms: Vec<String> = s
283 2327 : .firsts
284 2327 : .iter()
285 2601 : .map(|n| match s.exact_first {
286 1259 : true => format!("{base} AND types[1] = ${n}"),
287 1336 : false => format!("{base} AND types[1] = ${n} AND {}", s.selection),
288 2595 : })
289 2327 : .collect();
290 2327 : arms.push(match s.named {
291 2129 : Some(n) => format!(
292 : "{base} AND cardinality(types) > 1 AND types[1] <> ALL(${n}) AND {}",
293 : s.selection
294 : ),
295 198 : None => format!("{base} AND cardinality(types) > 1 AND {}", s.selection),
296 : });
297 2327 : arms
298 5230 : }
299 :
300 : /// Does `q` require the Entity to have no `location`, by a top-level
301 : /// `!location` term? Then `location IS NULL` holds for every match: a stored
302 : /// geometry comes from an instance's `value` or `object` (`extract`), which
303 : /// `!location` resolves as present. The conjunct is leakproof, so
304 : /// it reaches `i_entities_type_unlocated` (0018) under RLS, where the negated
305 : /// `@?` read the whole type; the `@?` stays and decides.
306 2129 : fn unlocated(q: &QNode, expand: &dyn Fn(&str) -> String) -> bool {
307 2129 : match q {
308 : QNode::Exists {
309 325 : path,
310 : negated: true,
311 : } => {
312 325 : path.links.is_empty()
313 325 : && path.bracket.is_none()
314 321 : && matches!(path.path.as_slice(), [term] if expand(term) == crate::compile::geo::LOCATION_IRI)
315 : }
316 965 : QNode::And(all) => all.iter().any(|n| unlocated(n, expand)),
317 1331 : _ => false,
318 : }
319 2129 : }
320 :
321 : /// The `name` the core @context gives an Entity (`@vocab` + `name`), the one
322 : /// attribute `entities.name_key` (0020) keys.
323 : /// One IRI; a `name` another @context maps elsewhere filters as before.
324 : const NAME_IRI: &str = "https://uri.etsi.org/ngsi-ld/default-context/name";
325 :
326 : /// The string a top-level `name=="…"` term of `q` requires. Every match then
327 : /// has `name_key` equal to it or NULL (0020: NULL is each shape the key
328 : /// cannot hold), so `(name_key = $n OR name_key IS NULL)` narrows by a
329 : /// leakproof btree under RLS where the `@?` read the whole Tenant;
330 : /// the `@?` stays and decides. A NUL is no Postgres text: no narrowing.
331 2445 : fn named<'q>(q: &'q QNode, expand: &dyn Fn(&str) -> String) -> Option<&'q str> {
332 255 : match q {
333 : QNode::Cmp {
334 255 : path,
335 : op: CmpOp::Eq,
336 255 : value: QValue::Str(want),
337 255 : } if path.links.is_empty()
338 249 : && path.bracket.is_none()
339 234 : && !want.contains('\0')
340 234 : && matches!(path.path.as_slice(), [term] if expand(term) == NAME_IRI) =>
341 : {
342 23 : Some(want)
343 : }
344 1275 : QNode::And(all) => all.iter().find_map(|n| named(n, expand)),
345 1875 : _ => None,
346 : }
347 2445 : }
348 :
349 : /// The Entity's own members, which `entity_value_rows` (0024) gives no row:
350 : /// an `attrs=` naming one is not narrowed.
351 : const OWN_MEMBERS: &[&str] = &[
352 : "id",
353 : "type",
354 : "scope",
355 : "@context",
356 : "createdAt",
357 : "modifiedAt",
358 : "deletedAt",
359 : "expiresAt",
360 : ];
361 :
362 : /// The longest string `entity_values.txt` (0024) keys, in bytes.
363 : const MAX_KEYED_TEXT: usize = 1024;
364 :
365 : /// What a top-level `q` term requires of its Attribute's `entity_values`
366 : /// rows (0024). Every bound is non-strict: a stored number is rounded to
367 : /// float8, monotonically, and the query's f64 is the literal's own value.
368 : #[derive(Debug, PartialEq)]
369 : enum ValueKey<'q> {
370 : Eq(f64),
371 : AtLeast(f64),
372 : AtMost(f64),
373 : Between(f64, f64),
374 : Text(&'q str),
375 : }
376 :
377 635 : fn value_key(op: CmpOp, want: &QValue) -> Option<ValueKey<'_>> {
378 635 : Some(match (op, want) {
379 56 : (CmpOp::Eq, QValue::Num(n)) => ValueKey::Eq(*n),
380 235 : (CmpOp::Gt | CmpOp::Ge, QValue::Num(n)) => ValueKey::AtLeast(*n),
381 31 : (CmpOp::Lt | CmpOp::Le, QValue::Num(n)) => ValueKey::AtMost(*n),
382 13 : (CmpOp::Eq, QValue::Range(lo, hi)) => match (lo.as_ref(), hi.as_ref()) {
383 11 : (QValue::Num(a), QValue::Num(b)) => ValueKey::Between(*a, *b),
384 2 : _ => return None,
385 : },
386 169 : (CmpOp::Eq, QValue::Str(s)) if !s.contains('\0') => ValueKey::Text(s),
387 131 : _ => return None,
388 : })
389 635 : }
390 :
391 : /// The first top-level `q` term `entity_values` keys: its Attribute's IRI
392 : /// and the rows a match holds. Call it only for a `q` that compiled: then
393 : /// the term's `@?` is exact and every row it matches has such a row.
394 1630 : fn valued<'q>(q: &'q QNode, expand: &dyn Fn(&str) -> String) -> Option<(String, ValueKey<'q>)> {
395 639 : match q {
396 639 : QNode::Cmp { path, op, value } if path.links.is_empty() && path.bracket.is_none() => {
397 637 : let [term] = path.path.as_slice() else {
398 2 : return None;
399 : };
400 635 : Some((expand(term), value_key(*op, value)?)).filter(|(iri, _)| !iri.contains('\0'))
401 : }
402 693 : QNode::And(all) => all.iter().find_map(|n| valued(n, expand)),
403 607 : _ => None,
404 : }
405 1630 : }
406 :
407 : /// One past the most Entity ids an `entity_values` probe (0024) hands the
408 : /// page as `id = ANY(...)`. A term matching more is no narrowing
409 : /// worth its primary key reads (about 7 µs each under RLS): the page reads
410 : /// as before, and the probe cost at most this many index entries.
411 : const PROBE_ROWS: i64 = 1001;
412 :
413 : /// A read of `entity_values` (0024) for the ids a term's matches are among,
414 : /// `$1` the Tenant: its statement and binds.
415 : struct Probe {
416 : sql: String,
417 : binds: Vec<Bind>,
418 : }
419 :
420 : impl Probe {
421 733 : fn new(tenant: &TenantId, attr: Bind, cond: &str, mut binds: Vec<Bind>) -> Self {
422 733 : binds.splice(0..0, [Bind::Text(tenant.as_str().to_owned()), attr]);
423 733 : binds.push(Bind::Int(PROBE_ROWS));
424 733 : let sql = format!(
425 : "SELECT id FROM entity_values WHERE tenant_id = $1 AND {cond} LIMIT ${}",
426 733 : binds.len()
427 : );
428 733 : Self { sql, binds }
429 733 : }
430 : }
431 :
432 : /// `valued`'s term as a probe: the rows that satisfy it, or hold 'NaN', the
433 : /// mark of an Attribute whose values were not keyed (0024). `f >= x` keeps
434 : /// 'NaN', which sorts above every number.
435 502 : fn value_probe(tenant: &TenantId, iri: String, key: ValueKey<'_>) -> Probe {
436 502 : let (cond, binds) = match key {
437 56 : ValueKey::Eq(n) => ("f = ANY(ARRAY[$3, 'NaN'::float8])", vec![Bind::Num(n)]),
438 235 : ValueKey::AtLeast(n) => ("f >= $3", vec![Bind::Num(n)]),
439 31 : ValueKey::AtMost(n) => ("(f <= $3 OR f = 'NaN')", vec![Bind::Num(n)]),
440 11 : ValueKey::Between(a, b) => (
441 11 : "(f BETWEEN $3 AND $4 OR f = 'NaN')",
442 11 : vec![Bind::Num(a), Bind::Num(b)],
443 11 : ),
444 169 : ValueKey::Text(s) if s.len() > MAX_KEYED_TEXT => ("f = 'NaN'", vec![]),
445 167 : ValueKey::Text(s) => ("(txt = $3 OR f = 'NaN')", vec![Bind::Text(s.to_owned())]),
446 : };
447 502 : Probe::new(
448 502 : tenant,
449 502 : Bind::Text(iri),
450 502 : &format!("attr = $2 AND {cond}"),
451 502 : binds,
452 : )
453 502 : }
454 :
455 : /// `attrs=` as a probe of presence rows (0024), `None` when it names one of
456 : /// the Entity's own members, which have none.
457 324 : fn attrs_probe(tenant: &TenantId, attrs: &[String]) -> Option<Probe> {
458 396 : if attrs.iter().any(|a| OWN_MEMBERS.contains(&a.as_str())) {
459 93 : return None;
460 231 : }
461 231 : let names = Bind::TextArr(super::storable(attrs));
462 231 : Some(Probe::new(tenant, names, "attr = ANY($2)", vec![]))
463 324 : }
464 :
465 : /// The distinct ids `p` reads, `None` when there are `PROBE_ROWS` rows or
466 : /// more. Unprepared, as the page statement.
467 715 : async fn probe_ids(
468 715 : tx: &mut sqlx::Transaction<'static, sqlx::Postgres>,
469 715 : p: &Probe,
470 715 : ) -> Result<Option<Vec<String>>, sqlx::Error> {
471 715 : let mut qy =
472 715 : sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(p.sql.clone())).persistent(false);
473 2640 : for b in &p.binds {
474 2640 : qy = match b {
475 1366 : Bind::Text(s) | Bind::Path(s) => qy.bind(s),
476 229 : Bind::TextArr(v) => qy.bind(v),
477 330 : Bind::Num(n) => qy.bind(n),
478 715 : Bind::Int(n) => qy.bind(n),
479 : };
480 : }
481 715 : let mut ids = qy.fetch_all(&mut **tx).await?;
482 715 : if i64::try_from(ids.len()).map_or(true, |n| n >= PROBE_ROWS) {
483 2 : return Ok(None);
484 713 : }
485 713 : ids.sort_unstable();
486 713 : ids.dedup();
487 713 : Ok(Some(ids))
488 715 : }
489 :
490 : /// The type selection's binds and SQL: `Ok(split)` for at most
491 : /// `MAX_TYPE_ARMS` distinct types, `Err(where)` (the plain containment, or
492 : /// `false` when no group can match) otherwise. A group naming a type with a
493 : /// NUL needs a type no entity has.
494 1960 : fn type_selection(groups: &[Vec<String>], binds: &mut Vec<Bind>) -> Result<TypeSplit, String> {
495 1960 : let groups: Vec<&Vec<String>> = groups
496 1960 : .iter()
497 2869 : .filter(|g| super::storable(g).len() == g.len())
498 1960 : .collect();
499 : // the type of each group naming one (`A;A` is `A`): a group of two
500 : // types is read by the multi-type arm alone
501 1960 : let mut named: Vec<&String> = groups
502 1960 : .iter()
503 3714 : .filter_map(|g| g.first().filter(|f| g.iter().all(|t| t == *f)))
504 1960 : .collect();
505 1960 : named.sort_unstable();
506 1960 : named.dedup();
507 1960 : let ors: Vec<String> = groups
508 1960 : .iter()
509 2866 : .map(|g| {
510 2866 : binds.push(Bind::TextArr((*g).clone()));
511 2866 : format!("types @> ${}", binds.len())
512 2866 : })
513 1960 : .collect();
514 1960 : if ors.is_empty() {
515 3 : return Err("false".to_owned());
516 1957 : }
517 1957 : let selection = format!("({})", ors.join(" OR "));
518 : // an empty group selects every type, which no set of arms can name
519 2832 : if named.len() > MAX_TYPE_ARMS || groups.iter().any(|g| g.is_empty()) {
520 4 : return Err(selection);
521 1953 : }
522 1953 : let firsts = named
523 1953 : .iter()
524 2116 : .map(|t| {
525 2108 : binds.push(Bind::Text((*t).clone()));
526 2108 : binds.len()
527 2108 : })
528 1953 : .collect();
529 1953 : let named = (!named.is_empty()).then(|| {
530 1791 : binds.push(Bind::TextArr(named.into_iter().cloned().collect()));
531 1791 : binds.len()
532 1791 : });
533 : Ok(TypeSplit {
534 1953 : firsts,
535 1953 : named,
536 1953 : selection,
537 1953 : exact_first: matches!(groups.as_slice(), [g] if g.len() == 1),
538 : })
539 1960 : }
540 :
541 : /// count=true's total: a count per arm over exactly the page's WHERE, never
542 : /// a window over the page statement (that kept every matching row, body and
543 : /// all, until the last one was read).
544 382 : fn count_sql(wheres: &[String], split: Option<&TypeSplit>) -> String {
545 382 : let counts: Vec<String> = arms(wheres, split)
546 382 : .iter()
547 869 : .map(|w| format!("(SELECT count(*) FROM entities WHERE {w})"))
548 382 : .collect();
549 382 : format!("SELECT {}", counts.join(" + "))
550 382 : }
551 :
552 : /// `id_prefixes` as SQL: the bounding byte range of every prefix, which the
553 : /// C-collated id index (0027) answers in its order, ANDed with the OR of
554 : /// each prefix's own range. Prefixes with a NUL match nothing (`storable`).
555 28 : fn prefix_ranges(prefixes: &[String], binds: &mut Vec<Bind>) -> String {
556 : const C: &str = "id COLLATE \"C\"";
557 40 : let mut ps: Vec<&String> = prefixes.iter().filter(|p| !p.contains('\0')).collect();
558 28 : ps.sort_unstable();
559 28 : let Some(lo) = ps.first() else {
560 6 : return "false".to_owned();
561 : };
562 58 : let mut range = |lo: &str, hi: Option<String>| {
563 58 : binds.push(Bind::Text(lo.to_owned()));
564 58 : let mut sql = format!("{C} >= ${}", binds.len());
565 58 : if let Some(hi) = hi {
566 58 : binds.push(Bind::Text(hi));
567 58 : sql.push_str(&format!(" AND {C} < ${}", binds.len()));
568 58 : }
569 58 : sql
570 58 : };
571 : // unbounded above when one prefix has no end
572 22 : let hi_end = ps
573 22 : .iter()
574 36 : .map(|p| antares_store::filter::prefix_end(p))
575 22 : .collect::<Option<Vec<_>>>()
576 22 : .and_then(|ends| ends.into_iter().max());
577 22 : let bound = range(lo, hi_end);
578 22 : let each: Vec<String> = ps
579 22 : .iter()
580 36 : .map(|p| format!("({})", range(p, antares_store::filter::prefix_end(p))))
581 22 : .collect();
582 22 : format!("({bound} AND ({}))", each.join(" OR "))
583 28 : }
584 :
585 : /// The statement, split out of `query` so its pagination shape is
586 : /// assertable without a database. Either LIMIT/OFFSET over an exactly
587 : /// decided set or a `query_candidates` walk (`true`; one row past the page answers "is there a next
588 : /// page", the total is `count_sql`'s) or the flat `MAX_UNDECIDED_ROWS`
589 : /// safety LIMIT — never no LIMIT. A split reads each arm in id order up to
590 : /// what the page can need and merges them.
591 4840 : fn query_sql(
592 4840 : select: &str,
593 4840 : wheres: &[String],
594 4840 : split: Option<&TypeSplit>,
595 4840 : page: Option<&Page>,
596 4840 : decided: bool,
597 4840 : order: &str,
598 4840 : binds: &mut Vec<Bind>,
599 4840 : ) -> (String, bool) {
600 4840 : let page = page.filter(|_| decided);
601 : // Saturating: the caller's limit is only bounded by `i64::MAX`
602 : // (`ANTARES_DISCOVERY_SCAN_MAX` clamps to exactly that), and one past
603 : // the largest page is still the largest page.
604 4840 : let limit = page.map_or(MAX_UNDECIDED_ROWS, |p| p.limit.saturating_add(1));
605 4840 : binds.push(Bind::Int(limit));
606 4840 : let lim = binds.len();
607 4840 : let offset = page.map_or(String::new(), |p| {
608 1639 : binds.push(Bind::Int(p.offset));
609 1639 : format!(" OFFSET ${}", binds.len())
610 1639 : });
611 4840 : let arms = arms(wheres, split);
612 : // A deep page skips its OFFSET over ids alone, which the type indexes
613 : // hand back without a heap visit (0013's INCLUDE), and reads only the
614 : // page's documents by primary key (page 50 of a 180k-entity
615 : // type walked 5 000 jsonb rows for 100).
616 4840 : if let Some(p) = page.filter(|p| p.offset > 0) {
617 403 : let ids = match arms.as_slice() {
618 57 : [w] => {
619 57 : format!("SELECT id FROM entities WHERE {w} ORDER BY {order} LIMIT ${lim}{offset}")
620 : }
621 : _ => {
622 346 : binds.push(Bind::Int(limit.saturating_add(p.offset)));
623 346 : let each = binds.len();
624 346 : let parts: Vec<String> = arms
625 346 : .iter()
626 839 : .map(|w| {
627 839 : format!(
628 : "(SELECT id FROM entities WHERE {w} ORDER BY {order} LIMIT ${each})"
629 : )
630 839 : })
631 346 : .collect();
632 346 : format!(
633 : "SELECT id FROM ({}) AS arms ORDER BY {order} LIMIT ${lim}{offset}",
634 346 : parts.join(" UNION ALL ")
635 : )
636 : }
637 : };
638 403 : return (
639 403 : format!(
640 403 : "SELECT {select} AS entity FROM entities \
641 403 : WHERE tenant_id = $1 AND id IN ({ids}) ORDER BY {order}"
642 403 : ),
643 403 : true,
644 403 : );
645 4437 : }
646 4437 : let sql = match arms.as_slice() {
647 2998 : [w] => format!(
648 : "SELECT {select} AS entity FROM entities WHERE {w} ORDER BY {order} LIMIT ${lim}{offset}"
649 : ),
650 : _ => {
651 1439 : let each = match page {
652 816 : Some(p) => {
653 816 : binds.push(Bind::Int(limit.saturating_add(p.offset)));
654 816 : binds.len()
655 : }
656 623 : None => lim,
657 : };
658 1439 : let parts: Vec<String> = arms
659 1439 : .iter()
660 3046 : .map(|w| {
661 3046 : format!("(SELECT id, entity FROM entities WHERE {w} ORDER BY {order} LIMIT ${each})")
662 3046 : })
663 1439 : .collect();
664 1439 : format!(
665 : "SELECT {select} AS entity FROM ({}) AS entities \
666 : ORDER BY {order} LIMIT ${lim}{offset}",
667 1439 : parts.join(" UNION ALL ")
668 : )
669 : }
670 : };
671 4437 : (sql, page.is_some())
672 4840 : }
673 :
674 : /// The total a pushed page shows by itself, `None` when only `count_sql`
675 : /// can tell it. Uncounted, the smallest total consistent with "a next page
676 : /// exists", enough for the next/prev links and never shown as a count;
677 : /// counted, a last page that holds rows (or the first, empty one).
678 1645 : fn page_total(p: &Page, rows: usize, more: bool) -> Option<i64> {
679 1645 : let seen = p.offset.saturating_add(rows as i64);
680 1645 : match (p.count, more) {
681 682 : (false, _) => Some(seen.saturating_add(i64::from(more))),
682 687 : (true, false) if rows > 0 || p.offset == 0 => Some(seen),
683 397 : (true, _) => None,
684 : }
685 1645 : }
686 :
687 : /// Did the safety LIMIT actually cut the match set? A statement that came
688 : /// back full was truncated at a bound nobody chose, and the caller's
689 : /// evaluator still has rows to drop, so no complete page can be built from
690 : /// it. 5.5.6: "When a query operation is producing so many results that can
691 : /// potentially exhaust client or server resources … implementations shall
692 : /// raise an error of type TooManyResults." Serving the truncated prefix
693 : /// would under-report the answer instead.
694 5983 : pub(crate) fn check_ceiling(paged: bool, rows: usize, ceiling: i64) -> Result<(), sqlx::Error> {
695 5983 : if paged || (rows as i64) < ceiling {
696 5977 : return Ok(());
697 6 : }
698 6 : Err(sqlx::Error::Configuration(Box::new(
699 6 : NgsiError::TooManyResults(format!(
700 6 : "query matched more than {ceiling} entities before filtering; \
701 6 : narrow it or request a smaller offset"
702 6 : )),
703 6 : )))
704 5983 : }
705 :
706 : /// Recover a spec error the store raised itself from the driver error it
707 : /// travels in (the store's only error channel is `sqlx::Error`). `None` =
708 : /// a genuine driver failure, which stays an InternalError at the seam.
709 12 : pub fn ngsi_error(e: &sqlx::Error) -> Option<&NgsiError> {
710 12 : match e {
711 8 : sqlx::Error::Configuration(b) => b.downcast_ref::<NgsiError>(),
712 4 : _ => None,
713 : }
714 12 : }
715 :
716 : impl PgEntityStore {
717 166 : pub fn new(pool: PgPool) -> Self {
718 166 : Self {
719 166 : batch_pool: pool.clone(),
720 166 : pool,
721 166 : outbox: std::sync::atomic::AtomicBool::new(false),
722 166 : async_batch: std::sync::atomic::AtomicBool::new(false),
723 166 : }
724 166 : }
725 :
726 : /// Run the batch writes on `reserve` instead of the shared pool.
727 1 : pub fn with_batch_pool(mut self, reserve: PgPool) -> Self {
728 1 : self.batch_pool = reserve;
729 1 : self
730 1 : }
731 :
732 : /// Asynchronous commit for the batch writes (see the field).
733 4 : pub fn set_async_batch_commit(&self, on: bool) {
734 4 : self.async_batch
735 4 : .store(on, std::sync::atomic::Ordering::Relaxed);
736 4 : }
737 :
738 : /// A batch write's transaction, with `synchronous_commit = off` set
739 : /// LOCAL to it when [`Self::set_async_batch_commit`] is on.
740 220 : async fn begin_batch(&self) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, sqlx::Error> {
741 220 : let mut tx = super::begin(&self.batch_pool).await?;
742 219 : if self.async_batch.load(std::sync::atomic::Ordering::Relaxed) {
743 4 : sqlx::query("SET LOCAL synchronous_commit = off")
744 4 : .execute(&mut *tx)
745 4 : .await?;
746 215 : }
747 219 : Ok(tx)
748 220 : }
749 :
750 : /// Outbox producer switch — the broker enables this exactly when `bus=nats`.
751 2 : pub fn set_outbox(&self, on: bool) {
752 2 : self.outbox.store(on, std::sync::atomic::Ordering::Relaxed);
753 2 : }
754 :
755 19845 : fn outbox_on(&self) -> bool {
756 19845 : self.outbox.load(std::sync::atomic::Ordering::Relaxed)
757 19845 : }
758 :
759 : /// 5.6.1-shaped create: `false` when the id already exists (→ 409).
760 : ///
761 : /// 4.22: "expiresAt is defined as the system temporal Property at which a
762 : /// certain Entity, Property or Relationship shall become invalid" — an
763 : /// entity whose expiry has passed is already invalid, whatever the
764 : /// reaping lag, so it must not 409 a create that reads (and GETs) as
765 : /// absent. The conflict clause replaces such a row and reports CREATED;
766 : /// a live row still takes the DO NOTHING path and `rows_affected() == 0`.
767 8749 : pub async fn create(
768 8749 : &self,
769 8749 : tenant: &TenantId,
770 8749 : id: &str,
771 8749 : doc: &Value,
772 8749 : ) -> Result<bool, sqlx::Error> {
773 8749 : Ok(self.create_stamped(tenant, id, doc).await?.is_some())
774 8749 : }
775 :
776 : /// [`Self::create`] with the created row's [`ChangeVersion`] for the
777 : /// change hook, `None` when the id is taken.
778 13413 : pub async fn create_stamped(
779 13413 : &self,
780 13413 : tenant: &TenantId,
781 13413 : id: &str,
782 13413 : doc: &Value,
783 13413 : ) -> Result<Option<ChangeVersion>, sqlx::Error> {
784 13413 : let e = extract(doc);
785 13413 : let mut tx = super::begin(&self.pool).await?;
786 13413 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
787 13413 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
788 13413 : let created = sqlx::query(
789 13413 : "INSERT INTO entities
790 13413 : (tenant_id, id, entity, types, scopes, created_at, modified_at, expires_at,
791 13413 : location, location_ambiguous, attrs_expire_at)
792 13413 : VALUES ($1, $2, $3, $4, $5, $6::timestamptz, $7::timestamptz, $8::timestamptz,
793 13413 : CASE WHEN ST_IsValid(try_geomfromgeojson($9))
794 13413 : THEN try_geomfromgeojson($9) END,
795 13413 : $10 OR ($9 IS NOT NULL
796 13413 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson($9)), false)),
797 13413 : try_timestamptz($11))
798 13413 : ON CONFLICT (tenant_id, id) DO UPDATE SET
799 13413 : entity = EXCLUDED.entity, types = EXCLUDED.types,
800 13413 : scopes = EXCLUDED.scopes, created_at = EXCLUDED.created_at,
801 13413 : modified_at = EXCLUDED.modified_at, expires_at = EXCLUDED.expires_at,
802 13413 : location = EXCLUDED.location,
803 13413 : location_ambiguous = EXCLUDED.location_ambiguous,
804 13413 : attrs_expire_at = EXCLUDED.attrs_expire_at,
805 13413 : version = 1
806 13413 : WHERE entities.expires_at IS NOT NULL AND entities.expires_at <= now()
807 13413 : RETURNING created_at::text",
808 13413 : )
809 13413 : .bind(tenant.as_str())
810 13413 : .bind(id)
811 13413 : .bind(doc)
812 13413 : .bind(&e.types)
813 13413 : .bind(&e.scopes)
814 13413 : .bind(&e.created)
815 13413 : .bind(&e.modified)
816 13413 : .bind(&e.expires)
817 13413 : .bind(&e.location)
818 13413 : .bind(e.location_ambiguous)
819 13413 : .bind(&e.attrs_expire)
820 13413 : .fetch_optional(&mut *tx)
821 13413 : .await?
822 13413 : .map(|r| ChangeVersion {
823 13401 : incarnation: r.get(0),
824 : version: 1,
825 13401 : });
826 13413 : if created.is_some() && self.outbox_on() {
827 4 : enqueue_change(
828 4 : &mut tx,
829 4 : tenant,
830 4 : "create",
831 4 : id,
832 4 : &e.types,
833 4 : None,
834 4 : Some(doc),
835 4 : 1,
836 4 : &e.created,
837 4 : )
838 4 : .await?;
839 13409 : }
840 13413 : tx.commit().await?;
841 13413 : Ok(created)
842 13413 : }
843 :
844 1320 : pub async fn get(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, sqlx::Error> {
845 1320 : let mut tx = super::begin(&self.pool).await?;
846 1320 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
847 : // 4.22: expired rows are invalid context until the sweep reaps them
848 1320 : let row = sqlx::query(
849 1320 : "SELECT entity FROM entities WHERE tenant_id = $1 AND id = $2
850 1320 : AND (expires_at IS NULL OR expires_at > now())",
851 1320 : )
852 1320 : .bind(tenant.as_str())
853 1320 : .bind(id)
854 1320 : .fetch_optional(&mut *tx)
855 1320 : .await?;
856 1320 : tx.commit().await?;
857 1320 : Ok(row.map(|r| r.get::<Value, _>(0)))
858 1320 : }
859 :
860 : /// The types the Tenant's live Entities carry, sorted: the
861 : /// names from the type catalogue the writes keep (migration 0012),
862 : /// each kept while the types index finds a live Entity of it.
863 : /// No `entity` document and no pass over the Tenant's rows.
864 86 : pub async fn type_names(&self, tenant: &TenantId) -> Result<Vec<String>, sqlx::Error> {
865 86 : self.catalogued(tenant, None).await
866 76 : }
867 :
868 : /// The catalogued types a live Entity carries, beside one of `any_of`
869 : /// when given. 4.22: expired rows are invalid context until the sweep
870 : /// reaps them.
871 164 : async fn catalogued(
872 164 : &self,
873 164 : tenant: &TenantId,
874 164 : any_of: Option<&[String]>,
875 164 : ) -> Result<Vec<String>, sqlx::Error> {
876 164 : let mut tx = super::begin(&self.pool).await?;
877 157 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
878 : // A type is live when it is some live Entity's first type, which a
879 : // skip scan of the first-type btree reads as one index probe per
880 : // type, or else, beside another, on an Entity of two or more types,
881 : // which the partial GIN alone indexes (0012). The plain types GIN
882 : // built a bitmap of every Entity of each type: 18 ms for five types
883 : // of 20 000, and a generic plan chose a sequential scan for an
884 : // EXISTS on the btree; this is 0.1 ms either way.
885 : const ALL: &str = "WITH RECURSIVE firsts AS (
886 : SELECT (SELECT min(e.types[1]) FROM entities e WHERE e.tenant_id = $1
887 : AND (e.expires_at IS NULL OR e.expires_at > now())) AS ty
888 : UNION ALL
889 : SELECT (SELECT min(e.types[1]) FROM entities e
890 : WHERE e.tenant_id = $1 AND e.types[1] > f.ty
891 : AND (e.expires_at IS NULL OR e.expires_at > now()))
892 : FROM firsts f WHERE f.ty IS NOT NULL)
893 : SELECT c.type FROM entity_types c WHERE c.tenant_id = $1
894 : AND (c.type IN (SELECT ty FROM firsts WHERE ty IS NOT NULL)
895 : OR EXISTS (SELECT 1 FROM entities e WHERE e.tenant_id = c.tenant_id
896 : AND cardinality(e.types) > 1 AND e.types @> ARRAY[c.type]
897 : AND (e.expires_at IS NULL OR e.expires_at > now())))
898 : ORDER BY c.type";
899 : const BESIDE: &str = "SELECT c.type FROM entity_types c WHERE c.tenant_id = $1
900 : AND EXISTS (SELECT 1 FROM entities e WHERE e.tenant_id = c.tenant_id
901 : AND cardinality(e.types) > 1
902 : AND e.types @> ARRAY[c.type] AND e.types && $2
903 : AND (e.expires_at IS NULL OR e.expires_at > now()))
904 : ORDER BY c.type";
905 156 : let names = match any_of {
906 : None => {
907 78 : sqlx::query_scalar(ALL)
908 78 : .bind(tenant.as_str())
909 78 : .fetch_all(&mut *tx)
910 78 : .await?
911 : }
912 78 : Some(of) => {
913 78 : sqlx::query_scalar(BESIDE)
914 78 : .bind(tenant.as_str())
915 78 : .bind(super::storable(of))
916 78 : .fetch_all(&mut *tx)
917 78 : .await?
918 : }
919 : };
920 156 : tx.commit().await?;
921 154 : Ok(names)
922 154 : }
923 :
924 : /// Every (type, member name) pair the Tenant's Entities ever carried,
925 : /// sorted: the attribute catalogue the writes keep (migration
926 : /// 0025). A pair whose last carrier went stays listed; the reader checks
927 : /// each against a live Entity, as it does a catalogued type.
928 68 : pub async fn attr_pairs(
929 68 : &self,
930 68 : tenant: &TenantId,
931 68 : ) -> Result<Vec<(String, String)>, sqlx::Error> {
932 68 : let mut tx = super::begin(&self.pool).await?;
933 68 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
934 68 : let pairs = sqlx::query_as(
935 68 : "SELECT type, attr FROM entity_attrs WHERE tenant_id = $1 ORDER BY type, attr",
936 68 : )
937 68 : .bind(tenant.as_str())
938 68 : .fetch_all(&mut *tx)
939 68 : .await?;
940 68 : tx.commit().await?;
941 68 : Ok(pairs)
942 68 : }
943 :
944 : /// The types of the Tenant's live Entities of two or more types that carry
945 : /// one of `any_of`, sorted: one probe of the multi-type index per
946 : /// catalogued type.
947 78 : pub async fn types_beside(
948 78 : &self,
949 78 : tenant: &TenantId,
950 78 : any_of: &[String],
951 78 : ) -> Result<Vec<String>, sqlx::Error> {
952 78 : self.catalogued(tenant, Some(any_of)).await
953 78 : }
954 :
955 : /// [`antares_store::CurrentStateDriver::entity_id_segments`] over the C-collated id
956 : /// index (0027): one seek per segment, the next one past the segment's
957 : /// own bound, so the cost follows the segments, not the ids.
958 : /// Expired Entities count too: a segment only narrows, never decides.
959 22 : pub async fn id_segments(
960 22 : &self,
961 22 : tenant: &TenantId,
962 22 : lead: &str,
963 22 : sep: char,
964 22 : max: usize,
965 22 : ) -> Result<Option<Vec<String>>, sqlx::Error> {
966 : use antares_store::filter::prefix_end;
967 22 : let Some(end) = prefix_end(lead).filter(|_| !lead.contains('\0')) else {
968 0 : return Ok(None);
969 : };
970 22 : let mut tx = super::begin(&self.pool).await?;
971 22 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
972 22 : let mut from = lead.to_owned();
973 22 : let mut out = Vec::new();
974 : // a step per segment and per id that holds no `sep` after `lead`
975 22 : for _ in 0..=max {
976 56 : let id: Option<String> = sqlx::query_scalar(
977 56 : "SELECT id FROM entities WHERE tenant_id = $1
978 56 : AND id COLLATE \"C\" >= $2 AND id COLLATE \"C\" < $3
979 56 : ORDER BY id COLLATE \"C\" LIMIT 1",
980 56 : )
981 56 : .bind(tenant.as_str())
982 56 : .bind(&from)
983 56 : .bind(&end)
984 56 : .fetch_optional(&mut *tx)
985 56 : .await?;
986 56 : let Some(id) = id else {
987 20 : tx.commit().await?;
988 20 : return Ok(Some(out));
989 : };
990 36 : let at = id.get(lead.len()..).and_then(|rest| rest.find(sep));
991 36 : from = match at.map(|k| &id[..lead.len().saturating_add(k)]) {
992 32 : Some(seg) => {
993 32 : out.push(seg.to_owned());
994 32 : match prefix_end(&format!("{seg}{sep}")) {
995 32 : Some(next) => next,
996 0 : None => break,
997 : }
998 : }
999 : // no id is greater than `id` and less than this one
1000 4 : None => format!("{id}\u{1}"),
1001 : };
1002 : }
1003 2 : tx.commit().await?;
1004 2 : Ok(None)
1005 22 : }
1006 :
1007 : /// Several Entities in one transaction: one per id that exists,
1008 : /// keyed by id. A write's Relationship targets are read this way, so a
1009 : /// fragment with five Relationships costs one BEGIN, not five.
1010 0 : pub async fn get_many(
1011 0 : &self,
1012 0 : tenant: &TenantId,
1013 0 : ids: &[&str],
1014 0 : ) -> Result<std::collections::HashMap<String, Value>, sqlx::Error> {
1015 0 : if ids.is_empty() {
1016 0 : return Ok(std::collections::HashMap::new());
1017 0 : }
1018 0 : let mut tx = super::begin(&self.pool).await?;
1019 0 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1020 : // 4.22: expired rows are invalid context until the sweep reaps them
1021 0 : let rows = sqlx::query(
1022 0 : "SELECT id, entity FROM entities WHERE tenant_id = $1 AND id = ANY($2)
1023 0 : AND (expires_at IS NULL OR expires_at > now())",
1024 0 : )
1025 0 : .bind(tenant.as_str())
1026 0 : .bind(ids)
1027 0 : .fetch_all(&mut *tx)
1028 0 : .await?;
1029 0 : tx.commit().await?;
1030 0 : Ok(rows
1031 0 : .into_iter()
1032 0 : .map(|r| (r.get::<String, _>(0), r.get::<Value, _>(1)))
1033 0 : .collect())
1034 0 : }
1035 :
1036 : /// Returns the deleted document (the before-image, captured in the same
1037 : /// transaction as the DELETE — never re-read outside it), `None` = absent.
1038 153 : pub async fn delete(&self, tenant: &TenantId, id: &str) -> Result<Option<Value>, sqlx::Error> {
1039 153 : Ok(self.delete_stamped(tenant, id).await?.map(|(prev, _)| prev))
1040 153 : }
1041 :
1042 : /// [`Self::delete`] with the delete's [`ChangeVersion`]: one past the
1043 : /// deleted row's version, in its incarnation.
1044 1549 : pub async fn delete_stamped(
1045 1549 : &self,
1046 1549 : tenant: &TenantId,
1047 1549 : id: &str,
1048 1549 : ) -> Result<Option<(Value, ChangeVersion)>, sqlx::Error> {
1049 1549 : let mut tx = super::begin(&self.pool).await?;
1050 1549 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1051 : // 4.22: an already-expired row is invalid, so deleting it is a
1052 : // 404 exactly as retrieving it is — never a 204 for an entity the
1053 : // API has stopped serving.
1054 1549 : let row = sqlx::query(
1055 1549 : "DELETE FROM entities WHERE tenant_id = $1 AND id = $2
1056 1549 : AND (expires_at IS NULL OR expires_at > now())
1057 1549 : RETURNING entity, types, version, created_at::text",
1058 1549 : )
1059 1549 : .bind(tenant.as_str())
1060 1549 : .bind(id)
1061 1549 : .fetch_optional(&mut *tx)
1062 1549 : .await?;
1063 1549 : let mut prev_out = None;
1064 1549 : if let Some(r) = &row {
1065 1357 : let prev: Value = r.get(0);
1066 1357 : if self.outbox_on() {
1067 0 : let types: Vec<String> = r.get(1);
1068 0 : enqueue_change(
1069 0 : &mut tx,
1070 0 : tenant,
1071 0 : "delete",
1072 0 : id,
1073 0 : &types,
1074 0 : Some(&prev),
1075 0 : None,
1076 0 : r.get::<i64, _>(2),
1077 0 : r.get::<&str, _>(3),
1078 0 : )
1079 0 : .await?;
1080 1357 : }
1081 1357 : prev_out = Some((prev, deleted_version(r)));
1082 192 : }
1083 1549 : tx.commit().await?;
1084 1549 : Ok(prev_out)
1085 1549 : }
1086 :
1087 : /// Delete one entity only if `keep` accepts the stored document, under
1088 : /// one lock: the DELETE itself takes the row lock and produces the
1089 : /// document that decides, and a refusal rolls the transaction back
1090 : /// instead of committing it. A `SELECT` followed by a `DELETE` would
1091 : /// leave the window this exists to close — between them the row can be
1092 : /// deleted and recreated under the same id, and the delete then lands on
1093 : /// a document nobody inspected.
1094 : ///
1095 : /// 4.22 as in `delete`: an already-expired row is invalid, so it is
1096 : /// absent here too.
1097 0 : pub async fn delete_if(
1098 0 : &self,
1099 0 : tenant: &TenantId,
1100 0 : id: &str,
1101 0 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
1102 0 : ) -> Result<Option<Value>, sqlx::Error> {
1103 0 : Ok(self
1104 0 : .delete_if_stamped(tenant, id, keep)
1105 0 : .await?
1106 0 : .map(|(prev, _)| prev))
1107 0 : }
1108 :
1109 : /// [`Self::delete_if`] with the delete's [`ChangeVersion`].
1110 1506 : pub async fn delete_if_stamped(
1111 1506 : &self,
1112 1506 : tenant: &TenantId,
1113 1506 : id: &str,
1114 1506 : keep: &(dyn for<'v> Fn(&'v Value) -> bool + Sync),
1115 1506 : ) -> Result<Option<(Value, ChangeVersion)>, sqlx::Error> {
1116 1506 : let mut tx = super::begin(&self.pool).await?;
1117 1506 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1118 1506 : let row = sqlx::query(
1119 1506 : "DELETE FROM entities WHERE tenant_id = $1 AND id = $2
1120 1506 : AND (expires_at IS NULL OR expires_at > now())
1121 1506 : RETURNING entity, types, version, created_at::text",
1122 1506 : )
1123 1506 : .bind(tenant.as_str())
1124 1506 : .bind(id)
1125 1506 : .fetch_optional(&mut *tx)
1126 1506 : .await?;
1127 1506 : let Some(r) = &row else {
1128 96 : tx.commit().await?;
1129 96 : return Ok(None);
1130 : };
1131 1410 : let prev: Value = r.get(0);
1132 1410 : if !keep(&prev) {
1133 4 : tx.rollback().await?;
1134 4 : return Ok(None);
1135 1406 : }
1136 1406 : if self.outbox_on() {
1137 0 : let types: Vec<String> = r.get(1);
1138 0 : enqueue_change(
1139 0 : &mut tx,
1140 0 : tenant,
1141 0 : "delete",
1142 0 : id,
1143 0 : &types,
1144 0 : Some(&prev),
1145 0 : None,
1146 0 : r.get::<i64, _>(2),
1147 0 : r.get::<&str, _>(3),
1148 0 : )
1149 0 : .await?;
1150 1406 : }
1151 1406 : let version = deleted_version(r);
1152 1406 : tx.commit().await?;
1153 1406 : Ok(Some((prev, version)))
1154 1506 : }
1155 :
1156 : /// Query pushdown. The predicates that compile EXACTLY go to
1157 : /// Postgres; everything else is simply left out of the WHERE clause, so
1158 : /// the result is always a superset of the answer and the caller's
1159 : /// in-memory evaluator remains the arbiter. That is the property that
1160 : /// makes store modes agree: SQL removes rows, it never decides them.
1161 : ///
1162 : /// Every value here is a bind. The only text this function
1163 : /// concatenates is its own operators and `$n` placeholders.
1164 : ///
1165 : /// Every statement it builds is bounded: an exactly decided query by the
1166 : /// caller's page, everything else by `MAX_UNDECIDED_ROWS`. A set that
1167 : /// reaches that ceiling is refused with TooManyResults (5.5.6) rather
1168 : /// than returned as a prefix the caller would page over as if complete.
1169 4519 : pub async fn query(
1170 4519 : &self,
1171 4519 : tenant: &TenantId,
1172 4519 : f: &EntityFilter<'_>,
1173 4519 : ) -> Result<QueryOutcome, sqlx::Error> {
1174 4519 : self.query_in(tenant, f, false).await
1175 4519 : }
1176 :
1177 : /// `query` for a caller that re-decides every row and walks the match
1178 : /// set in id-ordered pages: `f.page` is pushed over the candidates even
1179 : /// when SQL cannot decide them (`paged` without `decided`), so a page of
1180 : /// a geoquery reads that page's candidates, not up to
1181 : /// `MAX_UNDECIDED_ROWS`.
1182 303 : pub async fn query_candidates(
1183 303 : &self,
1184 303 : tenant: &TenantId,
1185 303 : f: &EntityFilter<'_>,
1186 303 : ) -> Result<QueryOutcome, sqlx::Error> {
1187 303 : self.query_in(tenant, f, true).await
1188 303 : }
1189 :
1190 4822 : async fn query_in(
1191 4822 : &self,
1192 4822 : tenant: &TenantId,
1193 4822 : f: &EntityFilter<'_>,
1194 4822 : candidates: bool,
1195 4822 : ) -> Result<QueryOutcome, sqlx::Error> {
1196 4822 : let mut binds: Vec<Bind> = vec![Bind::Text(tenant.as_str().to_owned())];
1197 4822 : let mut wheres = vec![
1198 4822 : "tenant_id = $1".to_owned(),
1199 : // 4.22: expired entities never enter a result (or a page/total)
1200 4822 : "(expires_at IS NULL OR expires_at > now())".to_owned(),
1201 : ];
1202 : // Exactness: ids/types/attrs translate exactly by
1203 : // construction; q is exact IF it compiles (the compiler's contract);
1204 : // scopeQ is documented loose-or-equal, geo has a metric residual
1205 : // (`near` geography vs haversine) — both therefore forfeit
1206 : // `decided`, they only narrow.
1207 4822 : let mut decided = true;
1208 4822 : let mut probes = Vec::new();
1209 :
1210 4822 : if let Some(ids) = f.ids {
1211 240 : binds.push(Bind::TextArr(super::storable(ids)));
1212 240 : wheres.push(format!("id = ANY(${})", binds.len()));
1213 4582 : }
1214 : // exact id prefixes as byte ranges of the C-collated id
1215 : // index (0027). Their bounding range is that index's condition and
1216 : // its order the walk's, so a probe reads the range and never walks
1217 : // a whole type in id order for a few late matches; each prefix's
1218 : // own range decides.
1219 4822 : let order = match f.id_prefixes {
1220 22 : Some(ps) => {
1221 22 : wheres.push(prefix_ranges(ps, &mut binds));
1222 22 : "id COLLATE \"C\""
1223 : }
1224 4800 : None => "id",
1225 : };
1226 : // keyset: the pkey (tenant_id, id) seeks straight to the next page
1227 4822 : if let Some(after) = f.after_id {
1228 4 : binds.push(Bind::Text(after.to_owned()));
1229 4 : wheres.push(format!("{order} > ${}", binds.len()));
1230 4818 : }
1231 : // 5.2.33 idPattern: the literal the regex forces is a plain string
1232 : // test on the id column, exact as a narrowing; the regex itself stays
1233 : // with the caller, so `decided` is untouched.
1234 4822 : if let Some(l) = f.id_literal {
1235 66 : if l.text.contains('\0') {
1236 2 : // no stored id holds a NUL (`storable`)
1237 2 : wheres.push("false".to_owned());
1238 2 : } else {
1239 64 : binds.push(Bind::Text(l.text.to_owned()));
1240 64 : wheres.push(if l.anchored {
1241 13 : format!("starts_with(id, ${})", binds.len())
1242 : } else {
1243 51 : format!("position(${} in id) > 0", binds.len())
1244 : });
1245 : }
1246 4756 : }
1247 : // a rule's idPatterns, exact where each reads alike as an ARE;
1248 : // a non-leakproof regex under RLS filters the rows the
1249 : // tenant, type and prefix keys found, it reaches no index.
1250 : // `regexp_like`, not `~`: the entity query is planned with its
1251 : // values, and `~`'s estimator reads every fixed character of a
1252 : // dataset pattern as a 1-in-5 filter, so it expects about one match
1253 : // and sorts the whole prefix range (164k Trees, 10 s on a deployment) instead
1254 : // of reading the C-collated id index in order up to the page's LIMIT.
1255 : // A function has no estimator and keeps the default third.
1256 4822 : if let Some(ps) = f.id_patterns {
1257 26 : match ps
1258 26 : .iter()
1259 26 : .map(|p| antares_store::filter::id_pattern_sql(p))
1260 26 : .collect::<Option<Vec<_>>>()
1261 : {
1262 24 : Some(res) => {
1263 24 : let any: Vec<String> = res
1264 24 : .into_iter()
1265 24 : .map(|re| {
1266 24 : binds.push(Bind::Text(re));
1267 24 : format!("regexp_like(id COLLATE \"C\", ${})", binds.len())
1268 24 : })
1269 24 : .collect();
1270 24 : wheres.push(match any.as_slice() {
1271 24 : [] => "false".to_owned(),
1272 24 : _ => format!("({})", any.join(" OR ")),
1273 : });
1274 : }
1275 2 : None => decided = false,
1276 : }
1277 4796 : }
1278 : // OR of AND-groups, mirroring the Entity Type Selection Language
1279 : // (4.17) the caller already parsed, read as `TypeSplit` arms.
1280 4822 : let mut split = None;
1281 4822 : if let Some(groups) = f.types.filter(|g| !g.is_empty()) {
1282 1940 : match type_selection(groups, &mut binds) {
1283 1939 : Ok(s) => split = Some(s),
1284 1 : Err(w) => wheres.push(w),
1285 : }
1286 2882 : }
1287 : // `attrs`: the entity holds at least one of them NOW — the
1288 : // evaluator's `has_live_attr` after 4.22 stripped expired instances.
1289 : // `?|` alone also matched an Attribute whose every instance expired.
1290 : // Only a row with an expired instance (`attrs_expire_at` past, rare)
1291 : // pays for the per-instance check, so the predicate stays decided
1292 : // and the page stays pushed down.
1293 4822 : if let Some(attrs) = f.attrs {
1294 306 : binds.push(Bind::TextArr(super::storable(attrs)));
1295 306 : let n = binds.len();
1296 306 : wheres.push(format!(
1297 306 : "entity ?| ${n} AND (attrs_expire_at IS NULL OR attrs_expire_at >= now() \
1298 306 : OR EXISTS (SELECT 1 FROM unnest(${n}::text[]) AS w(a) \
1299 306 : WHERE entity ? w.a AND (jsonb_typeof(entity -> w.a) <> 'array' \
1300 306 : OR strpos(w.a, ':') = 0 OR jsonb_array_length(entity -> w.a) = 0 \
1301 306 : OR EXISTS (SELECT 1 FROM jsonb_array_elements(entity -> w.a) AS i(v) \
1302 306 : WHERE jsonb_typeof(i.v -> 'expiresAt') IS DISTINCT FROM 'string' \
1303 306 : OR try_timestamptz(i.v ->> 'expiresAt') IS NULL \
1304 306 : OR try_timestamptz(i.v ->> 'expiresAt') >= now()))))"
1305 306 : ));
1306 306 : probes.extend(attrs_probe(tenant, attrs));
1307 4516 : }
1308 4822 : if let Some(node) = f.q {
1309 1146 : match crate::compile::q::compile_q(
1310 1146 : node,
1311 1146 : LIVE_ENTITY,
1312 1146 : binds.len().saturating_add(1),
1313 1146 : f.expand,
1314 1146 : ) {
1315 899 : Some(c) => {
1316 899 : wheres.push(c.sql);
1317 899 : binds.extend(c.binds.into_iter().map(Bind::Path));
1318 899 : if let Some((iri, key)) = valued(node, f.expand) {
1319 486 : probes.push(value_probe(tenant, iri, key));
1320 556 : }
1321 : }
1322 247 : None => decided = false,
1323 : }
1324 1146 : if unlocated(node, f.expand) {
1325 256 : wheres.push("location IS NULL".to_owned());
1326 890 : }
1327 1146 : if let Some(name) = named(node, f.expand) {
1328 17 : binds.push(Bind::Text(name.to_owned()));
1329 17 : wheres.push(format!("(name_key = ${} OR name_key IS NULL)", binds.len()));
1330 1129 : }
1331 3676 : }
1332 4822 : if let Some(sq) = f.scope_q {
1333 308 : decided = false;
1334 305 : if let Some(c) =
1335 308 : crate::compile::scope::compile_scope_q(sq, "scopes", binds.len().saturating_add(1))
1336 305 : {
1337 305 : wheres.push(c.sql);
1338 305 : binds.extend(c.binds.into_iter().map(Bind::Text));
1339 305 : }
1340 4514 : }
1341 4822 : let mut tx = self
1342 4822 : .begin_probed(tenant, &probes, &mut wheres, &mut binds)
1343 4822 : .await?;
1344 4820 : for spec in f.geo {
1345 1617 : decided = false;
1346 : // A client may send a self-intersecting polygon. GEOS raises on
1347 : // one (`side location conflict`), which would turn a query into a
1348 : // 500 in `postgres` mode while `memory` mode answers happily from
1349 : // the evaluator. Probing validity once here keeps the two modes
1350 : // identical: invalid ⇒ no pushdown, evaluator decides. Stored
1351 : // geometries can't be invalid — the write path NULLs those. The
1352 : // probe runs on the query's own transaction, and a failed probe
1353 : // fails the query: a pool timeout answers 503
1354 : // instead of costing the pushdown.
1355 1617 : if geometry_is_valid(&mut tx, spec).await? {
1356 1613 : if let Some(c) = crate::compile::geo::compile_geo(
1357 1613 : spec,
1358 1613 : "location",
1359 1613 : binds.len().saturating_add(1),
1360 1613 : ) {
1361 1609 : wheres.push(c.sql);
1362 1609 : // geo binds first, then the numeric ones, then the cell
1363 1609 : // cover — the order `compile_geo` numbered them in.
1364 1609 : binds.extend(c.geo_binds.into_iter().map(Bind::Text));
1365 1609 : binds.extend(c.num_binds.into_iter().map(Bind::Num));
1366 1609 : binds.extend(c.cell_binds.into_iter().map(Bind::Text));
1367 1609 : }
1368 4 : }
1369 : }
1370 :
1371 : // every bind up to here belongs to the WHERE clause — the count-only
1372 : // fallback statement below reuses exactly this prefix
1373 4820 : let where_binds = binds.len();
1374 :
1375 : // Projection pushdown, only once SQL decides row membership: the
1376 : // kept doc must only need to feed `repr::apply`, never a re-check.
1377 : // `pick` keeps listed attrs + every non-attribute member (attribute
1378 : // keys are expanded IRIs — `http…` — so core members never match the
1379 : // LIKE and always survive; a non-http attr IRI merely stays
1380 : // unprojected, which is the safe direction). `omit` drops exactly the
1381 : // listed top-level IRIs.
1382 4820 : let mut select = "entity".to_owned();
1383 4820 : if decided {
1384 2902 : if let Some(keep) = f.keep_attrs {
1385 148 : binds.push(Bind::TextArr(super::storable(keep)));
1386 148 : select = format!(
1387 148 : "(SELECT COALESCE(jsonb_object_agg(t.k, t.v), '{{}}'::jsonb) \
1388 148 : FROM jsonb_each(entity) AS t(k, v) \
1389 148 : WHERE t.k NOT LIKE 'http%' OR t.k = ANY(${}))",
1390 148 : binds.len()
1391 148 : );
1392 2754 : } else if let Some(drop) = f.drop_attrs {
1393 12 : binds.push(Bind::TextArr(super::storable(drop)));
1394 12 : select = format!(
1395 12 : "(SELECT COALESCE(jsonb_object_agg(t.k, t.v), '{{}}'::jsonb) \
1396 12 : FROM jsonb_each(entity) AS t(k, v) \
1397 12 : WHERE NOT (t.k = ANY(${})))",
1398 12 : binds.len()
1399 12 : );
1400 2742 : }
1401 1918 : }
1402 :
1403 : // Pagination pushdown: ORDER BY id is the store's default order
1404 : // either way; the total for count= comes from `count_sql` below,
1405 : // only when the page itself cannot tell it. When the page cannot be pushed the statement still carries
1406 : // the safety LIMIT — without one, a single undecided request (any
1407 : // `scopeQ`, any `georel`, a `q=` shape the compiler declined)
1408 : // materializes the whole tenant into the Vec below.
1409 4820 : let ceiling = MAX_UNDECIDED_ROWS;
1410 4820 : let (sql, paged) = query_sql(
1411 4820 : &select,
1412 4820 : &wheres,
1413 4820 : split.as_ref(),
1414 4820 : f.page.as_ref(),
1415 4820 : decided || candidates,
1416 4820 : order,
1417 4820 : &mut binds,
1418 : );
1419 : // sqlx 0.9 makes dynamic SQL opt-in. The assertion holds by
1420 : // construction: `sql` is built from this function's own literals
1421 : // plus `$n` placeholders — no caller-supplied text reaches it.
1422 : // The audit lives here, next to the builder.
1423 : // Unprepared: a prepared statement goes to a generic plan after five
1424 : // runs on a connection, and that plan cannot see a selective value:
1425 : // 1.6 s instead of 1.5 ms at 100 000 entities when a GIN still
1426 : // existed. The text varies with every q, geo and
1427 : // projection shape, so the statement cache saved little anyway.
1428 4820 : let mut qy = sqlx::query(sqlx::AssertSqlSafe(sql.clone())).persistent(false);
1429 58137 : for b in &binds {
1430 58137 : qy = match b {
1431 44177 : Bind::Text(s) | Bind::Path(s) => qy.bind(s),
1432 6010 : Bind::TextArr(v) => qy.bind(v),
1433 343 : Bind::Num(n) => qy.bind(n),
1434 7607 : Bind::Int(n) => qy.bind(n),
1435 : };
1436 : }
1437 : // Stream rows, decode each to its Value and
1438 : // drop the PgRow — the full row set never sits in memory twice.
1439 4820 : let mut docs: Vec<Value> = Vec::new();
1440 4820 : let mut total: Option<i64> = None;
1441 : // candidates are no match set to count
1442 4820 : let counted = paged && decided && f.page.as_ref().is_some_and(|p| p.count);
1443 : {
1444 : use futures_util::TryStreamExt;
1445 4820 : let mut stream = qy.fetch(&mut *tx);
1446 42908 : while let Some(row) = stream.try_next().await? {
1447 38088 : docs.push(row.get::<Value, _>(0));
1448 38088 : }
1449 : }
1450 4820 : check_ceiling(paged, docs.len(), ceiling)?;
1451 : // the extra row says a next page exists. Uncounted, the total
1452 : // handed back is the smallest one consistent with that, enough for
1453 : // the next/prev links and never shown as a count; counted, a last
1454 : // page that holds rows is the total itself.
1455 4818 : if let Some(p) = f.page.as_ref().filter(|_| paged) {
1456 1627 : let more = docs.len() as i64 > p.limit;
1457 1627 : docs.truncate(p.limit as usize);
1458 1627 : total = page_total(p, docs.len(), more).filter(|_| decided);
1459 3191 : }
1460 : // count=true with more rows past the page, or an off-the-end page:
1461 : // count the match set under the same WHERE, binds and transaction
1462 4818 : if counted && total.is_none() {
1463 : // unprepared for the same reason as the statement above
1464 378 : let mut cq = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(count_sql(
1465 378 : &wheres,
1466 378 : split.as_ref(),
1467 378 : )))
1468 378 : .persistent(false);
1469 : // same wheres ⇒ same bind prefix; stop before the
1470 : // projection/page binds, which the count statement lacks
1471 2000 : for b in binds.iter().take(where_binds) {
1472 2000 : cq = match b {
1473 911 : Bind::Text(s) | Bind::Path(s) => cq.bind(s),
1474 1089 : Bind::TextArr(v) => cq.bind(v),
1475 0 : Bind::Num(n) => cq.bind(n),
1476 0 : Bind::Int(n) => cq.bind(n),
1477 : };
1478 : }
1479 378 : total = Some(cq.fetch_one(&mut *tx).await?);
1480 4440 : }
1481 4818 : tx.commit().await?;
1482 4818 : Ok(QueryOutcome {
1483 4818 : rows: docs,
1484 4818 : decided,
1485 4818 : paged,
1486 4818 : total,
1487 4818 : })
1488 4822 : }
1489 :
1490 : /// The page's transaction with the Tenant set, and each probe's ids
1491 : /// ANDed to `wheres` as `id = ANY(...)`. A probed page reads in
1492 : /// one snapshot (repeatable read), so no write between the probe and the
1493 : /// page drops a match.
1494 4822 : async fn begin_probed(
1495 4822 : &self,
1496 4822 : tenant: &TenantId,
1497 4822 : probes: &[Probe],
1498 4822 : wheres: &mut Vec<String>,
1499 4822 : binds: &mut Vec<Bind>,
1500 4822 : ) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, sqlx::Error> {
1501 4822 : let mut tx = super::begin(&self.pool).await?;
1502 4820 : if !probes.is_empty() {
1503 715 : sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY")
1504 715 : .execute(&mut *tx)
1505 715 : .await?;
1506 4105 : }
1507 4820 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1508 4820 : for p in probes {
1509 715 : if let Some(ids) = probe_ids(&mut tx, p).await? {
1510 713 : binds.push(Bind::TextArr(ids));
1511 713 : wheres.push(format!("id = ANY(${})", binds.len()));
1512 713 : }
1513 : }
1514 4820 : Ok(tx)
1515 4822 : }
1516 :
1517 : /// Id-ordered snapshot for one tenant (the v0 `list` shape — still the
1518 : /// path for every non-entity kind and for callers with no filter).
1519 : ///
1520 : /// Bounded like every other read: the statement carries the same
1521 : /// `MAX_UNDECIDED_ROWS` safety LIMIT, and a tenant that reaches it is
1522 : /// refused with TooManyResults (5.5.6) rather than served a silent
1523 : /// prefix.
1524 36 : pub async fn list(&self, tenant: &TenantId) -> Result<Vec<Value>, sqlx::Error> {
1525 36 : let mut tx = super::begin(&self.pool).await?;
1526 36 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1527 36 : let mut docs: Vec<Value> = Vec::new();
1528 : {
1529 : use futures_util::TryStreamExt;
1530 36 : let mut stream = sqlx::query(
1531 : "SELECT entity FROM entities WHERE tenant_id = $1
1532 : AND (expires_at IS NULL OR expires_at > now())
1533 : ORDER BY id LIMIT $2",
1534 : )
1535 36 : .bind(tenant.as_str())
1536 36 : .bind(MAX_UNDECIDED_ROWS)
1537 36 : .fetch(&mut *tx);
1538 21306 : while let Some(row) = stream.try_next().await? {
1539 21270 : docs.push(row.get::<Value, _>(0));
1540 21270 : }
1541 : }
1542 36 : tx.commit().await?;
1543 36 : check_ceiling(false, docs.len(), MAX_UNDECIDED_ROWS)?;
1544 34 : Ok(docs)
1545 36 : }
1546 :
1547 : /// One id-ordered page of entities: ids strictly greater than `after`,
1548 : /// at most `limit`.
1549 : ///
1550 : /// No ceiling, and no fold of the tenant to build the page.
1551 : /// `MAX_UNDECIDED_ROWS` on `list` exists so a large tenant cannot be
1552 : /// materialized into one `Vec`; a page bounds that by construction, so
1553 : /// refusing here would only break the readers that must see every row
1554 : /// and have no TooManyResults to raise — 5.9.2.4's registration-vs-entity
1555 : /// conflict check among them. Keyset over the primary key, not OFFSET:
1556 : /// the walk runs against a table being written to. `after = None` is
1557 : /// bound as the empty string, which is below every id because an
1558 : /// Entity id is a URI and a URI is never empty.
1559 : ///
1560 : /// 4.22 at the entity level is in the statement, so a full page is a
1561 : /// full page: the caller reads a short page as the end of the tenant,
1562 : /// and a filter applied after the LIMIT would end the walk early.
1563 66 : pub async fn list_page(
1564 66 : &self,
1565 66 : tenant: &TenantId,
1566 66 : after: Option<&str>,
1567 66 : limit: i64,
1568 66 : ) -> Result<Vec<Value>, sqlx::Error> {
1569 66 : let mut tx = super::begin(&self.pool).await?;
1570 66 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1571 66 : let rows = sqlx::query(
1572 66 : "SELECT entity FROM entities WHERE tenant_id = $1 AND id > $2
1573 66 : AND (expires_at IS NULL OR expires_at > now())
1574 66 : ORDER BY id LIMIT $3",
1575 66 : )
1576 66 : .bind(tenant.as_str())
1577 66 : .bind(after.unwrap_or(""))
1578 66 : .bind(limit)
1579 66 : .fetch_all(&mut *tx)
1580 66 : .await?;
1581 66 : tx.commit().await?;
1582 4036 : Ok(rows.into_iter().map(|r| r.get::<Value, _>(0)).collect())
1583 66 : }
1584 :
1585 : /// One id-ordered `LIMIT`/`OFFSET` window of the Tenant's live Entities
1586 : /// and their count, in one snapshot: the window bounds the read, so no
1587 : /// row ceiling applies (`CurrentStateDriver::list_slice`).
1588 10 : pub async fn list_slice(
1589 10 : &self,
1590 10 : tenant: &TenantId,
1591 10 : offset: i64,
1592 10 : limit: i64,
1593 10 : ) -> Result<(Vec<Value>, i64), sqlx::Error> {
1594 10 : let mut tx = super::begin(&self.pool).await?;
1595 10 : sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY")
1596 10 : .execute(&mut *tx)
1597 10 : .await?;
1598 10 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1599 10 : let rows = sqlx::query(
1600 10 : "SELECT entity FROM entities WHERE tenant_id = $1
1601 10 : AND (expires_at IS NULL OR expires_at > now())
1602 10 : ORDER BY id LIMIT $2 OFFSET $3",
1603 10 : )
1604 10 : .bind(tenant.as_str())
1605 10 : .bind(limit)
1606 10 : .bind(offset)
1607 10 : .fetch_all(&mut *tx)
1608 10 : .await?;
1609 10 : let total: i64 = sqlx::query_scalar(
1610 10 : "SELECT count(*) FROM entities WHERE tenant_id = $1
1611 10 : AND (expires_at IS NULL OR expires_at > now())",
1612 10 : )
1613 10 : .bind(tenant.as_str())
1614 10 : .fetch_one(&mut *tx)
1615 10 : .await?;
1616 10 : tx.commit().await?;
1617 : Ok((
1618 18 : rows.into_iter().map(|r| r.get::<Value, _>(0)).collect(),
1619 10 : total,
1620 : ))
1621 10 : }
1622 :
1623 : /// Read-modify-write: row lock via `SELECT … FOR NO KEY UPDATE`, closure
1624 : /// applied in Rust, `version` bumped under the lock. Two racing PATCHes
1625 : /// serialize in Postgres, neither is lost. `Ok(None)` = entity absent.
1626 : ///
1627 : /// No write changes the key (tenant_id, id), so the lock is the one the
1628 : /// UPDATE itself takes: it serializes writers and leaves the history
1629 : /// append's `FOR KEY SHARE` alone (a `FOR UPDATE` here waited on
1630 : /// every in-flight append and made the append wait on it in turn, until
1631 : /// `lock_timeout` dropped the history of a whole drained request).
1632 178 : pub async fn mutate<T, E>(
1633 178 : &self,
1634 178 : tenant: &TenantId,
1635 178 : id: &str,
1636 178 : f: impl FnOnce(&mut Value) -> Result<T, E>,
1637 178 : ) -> Result<Option<Result<T, E>>, sqlx::Error> {
1638 178 : Ok(self
1639 178 : .mutate_stamped(tenant, id, f)
1640 178 : .await?
1641 178 : .map(|r| r.map(|(t, _)| t)))
1642 178 : }
1643 :
1644 : /// [`Self::mutate`] with the version the write left, read under its
1645 : /// row lock.
1646 3066 : pub async fn mutate_stamped<T, E>(
1647 3066 : &self,
1648 3066 : tenant: &TenantId,
1649 3066 : id: &str,
1650 3066 : f: impl FnOnce(&mut Value) -> Result<T, E>,
1651 3066 : ) -> Result<Option<Result<(T, ChangeVersion), E>>, sqlx::Error> {
1652 3066 : let mut tx = super::begin(&self.pool).await?;
1653 3066 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1654 : // 4.22: an expired row is invalid, so a patch of it is a 404 —
1655 : // the same answer the retrieve it would follow already gives.
1656 3066 : let row = sqlx::query(
1657 3066 : "SELECT entity FROM entities WHERE tenant_id = $1 AND id = $2
1658 3066 : AND (expires_at IS NULL OR expires_at > now()) FOR NO KEY UPDATE",
1659 3066 : )
1660 3066 : .bind(tenant.as_str())
1661 3066 : .bind(id)
1662 3066 : .fetch_optional(&mut *tx)
1663 3066 : .await?;
1664 3066 : let Some(row) = row else {
1665 2444 : tx.commit().await?;
1666 2444 : return Ok(None);
1667 : };
1668 622 : let mut doc: Value = row.get(0);
1669 622 : let before = self.outbox_on().then(|| doc.clone());
1670 622 : match f(&mut doc) {
1671 602 : Ok(t) => {
1672 602 : let e = extract(&doc);
1673 602 : let updated = sqlx::query(
1674 602 : "UPDATE entities SET entity = $3, types = $4, scopes = $5,
1675 602 : modified_at = $6::timestamptz, expires_at = $7::timestamptz,
1676 602 : location = CASE WHEN ST_IsValid(try_geomfromgeojson($8))
1677 602 : THEN try_geomfromgeojson($8) END,
1678 602 : location_ambiguous = $9 OR ($8 IS NOT NULL
1679 602 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson($8)), false)),
1680 602 : attrs_expire_at = try_timestamptz($10),
1681 602 : version = version + 1
1682 602 : WHERE tenant_id = $1 AND id = $2
1683 602 : RETURNING version, created_at::text",
1684 602 : )
1685 602 : .bind(tenant.as_str())
1686 602 : .bind(id)
1687 602 : .bind(&doc)
1688 602 : .bind(&e.types)
1689 602 : .bind(&e.scopes)
1690 602 : .bind(&e.modified)
1691 602 : .bind(&e.expires)
1692 602 : .bind(&e.location)
1693 602 : .bind(e.location_ambiguous)
1694 602 : .bind(&e.attrs_expire)
1695 602 : .fetch_one(&mut *tx)
1696 602 : .await?;
1697 602 : if let Some(before) = &before {
1698 0 : enqueue_change(
1699 0 : &mut tx,
1700 0 : tenant,
1701 0 : "update",
1702 0 : id,
1703 0 : &e.types,
1704 0 : Some(before),
1705 0 : Some(&doc),
1706 0 : updated.get::<i64, _>(0),
1707 0 : updated.get::<&str, _>(1),
1708 0 : )
1709 0 : .await?;
1710 602 : }
1711 602 : tx.commit().await?;
1712 602 : Ok(Some(Ok((
1713 602 : t,
1714 602 : ChangeVersion {
1715 602 : version: updated.get(0),
1716 602 : incarnation: updated.get(1),
1717 602 : },
1718 602 : ))))
1719 : }
1720 20 : Err(e) => {
1721 20 : tx.rollback().await?;
1722 20 : Ok(Some(Err(e)))
1723 : }
1724 : }
1725 3066 : }
1726 :
1727 : /// Batch create: ONE multi-row INSERT for the whole batch (the jsonb
1728 : /// elements form of UNNEST), one transaction, one commit.
1729 : /// Returns a created-flag per input item, input order preserved.
1730 : /// Duplicate ids within one batch are pre-deduped here: `ON CONFLICT`
1731 : /// raises "cannot affect row a second time" otherwise — the later
1732 : /// duplicate reports `false` (5.5.11.1: the first instance wins).
1733 : ///
1734 : /// 5.6.7.4: "For each of the NGSI-LD Entities included in the input
1735 : /// Array execute the behaviour defined by clause 5.6.1, but limited to a
1736 : /// local operation" — so the conflict clause is `create`'s, and an
1737 : /// entity past its `expiresAt` is absent (4.22) and gets created over
1738 : /// rather than reported as an id that already exists.
1739 18 : pub async fn batch_create(
1740 18 : &self,
1741 18 : tenant: &TenantId,
1742 18 : items: &[(String, Value)],
1743 18 : ) -> Result<Vec<bool>, sqlx::Error> {
1744 18 : Ok(self
1745 18 : .batch_create_stamped(tenant, items)
1746 18 : .await?
1747 18 : .into_iter()
1748 20012 : .map(|v| v.is_some())
1749 18 : .collect())
1750 18 : }
1751 :
1752 : /// [`Self::batch_create`] with each created row's [`ChangeVersion`].
1753 54 : pub async fn batch_create_stamped(
1754 54 : &self,
1755 54 : tenant: &TenantId,
1756 54 : items: &[(String, Value)],
1757 54 : ) -> Result<Vec<Option<ChangeVersion>>, sqlx::Error> {
1758 54 : let mut seen = std::collections::HashSet::new();
1759 54 : let mut payload = Vec::new();
1760 20066 : for (id, doc) in items {
1761 20066 : if seen.insert(id.as_str()) {
1762 20062 : let e = extract(doc);
1763 20062 : payload.push(serde_json::json!({
1764 20062 : "id": id, "doc": doc, "types": e.types, "scopes": e.scopes,
1765 20062 : "created": e.created, "modified": e.modified, "expires": e.expires,
1766 20062 : "location": e.location, "loc_ambiguous": e.location_ambiguous,
1767 20062 : "attrs_expire": e.attrs_expire,
1768 20062 : }));
1769 20062 : }
1770 : }
1771 54 : let mut tx = self.begin_batch().await?;
1772 54 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1773 54 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
1774 54 : let rows = sqlx::query(
1775 54 : "INSERT INTO entities
1776 54 : (tenant_id, id, entity, types, scopes, created_at, modified_at, expires_at,
1777 54 : location, location_ambiguous, attrs_expire_at)
1778 54 : SELECT $1, e->>'id', e->'doc',
1779 54 : ARRAY(SELECT jsonb_array_elements_text(e->'types')),
1780 54 : CASE WHEN e->'scopes' = 'null'::jsonb THEN NULL
1781 54 : ELSE ARRAY(SELECT jsonb_array_elements_text(e->'scopes')) END,
1782 54 : (e->>'created')::timestamptz, (e->>'modified')::timestamptz,
1783 54 : (e->>'expires')::timestamptz,
1784 54 : CASE WHEN ST_IsValid(try_geomfromgeojson(e->>'location'))
1785 54 : THEN try_geomfromgeojson(e->>'location') END,
1786 54 : COALESCE((e->>'loc_ambiguous')::bool, false)
1787 54 : OR (e->>'location' IS NOT NULL
1788 54 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson(e->>'location')), false)),
1789 54 : try_timestamptz(e->>'attrs_expire')
1790 54 : FROM jsonb_array_elements($2::jsonb) AS e
1791 54 : ON CONFLICT (tenant_id, id) DO UPDATE SET
1792 54 : entity = EXCLUDED.entity, types = EXCLUDED.types,
1793 54 : scopes = EXCLUDED.scopes, created_at = EXCLUDED.created_at,
1794 54 : modified_at = EXCLUDED.modified_at,
1795 54 : expires_at = EXCLUDED.expires_at, location = EXCLUDED.location,
1796 54 : location_ambiguous = EXCLUDED.location_ambiguous,
1797 54 : attrs_expire_at = EXCLUDED.attrs_expire_at,
1798 54 : version = 1
1799 54 : WHERE entities.expires_at IS NOT NULL AND entities.expires_at <= now()
1800 54 : RETURNING id, created_at::text",
1801 54 : )
1802 54 : .bind(tenant.as_str())
1803 54 : .bind(Value::Array(payload))
1804 54 : .fetch_all(&mut *tx)
1805 54 : .await?;
1806 54 : let mut created_now: std::collections::HashMap<String, String> = rows
1807 54 : .into_iter()
1808 20050 : .map(|r| (r.get::<String, _>(0), r.get::<String, _>(1)))
1809 54 : .collect();
1810 54 : if self.outbox_on() {
1811 0 : let mut seen_ev = std::collections::HashSet::new();
1812 0 : let mut events = Vec::new();
1813 0 : for (id, doc) in items {
1814 0 : if created_now.contains_key(id.as_str()) && seen_ev.insert(id.as_str()) {
1815 0 : let e = extract(doc);
1816 0 : events.push(change_event(
1817 0 : tenant,
1818 0 : "create",
1819 0 : id,
1820 0 : &e.types,
1821 0 : None,
1822 0 : Some(doc),
1823 0 : 1,
1824 0 : &e.created,
1825 0 : ));
1826 0 : }
1827 : }
1828 0 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
1829 54 : }
1830 54 : tx.commit().await?;
1831 : // consume-once: a duplicate of a created id still reports false
1832 54 : Ok(items
1833 54 : .iter()
1834 20066 : .map(|(id, _)| {
1835 20066 : created_now
1836 20066 : .remove(id.as_str())
1837 20066 : .map(|incarnation| ChangeVersion {
1838 20050 : incarnation,
1839 : version: 1,
1840 20050 : })
1841 20066 : })
1842 54 : .collect())
1843 54 : }
1844 :
1845 : /// Batch delete: ONE statement, returning each deleted row's previous
1846 : /// document (the change-hook before-image).
1847 : ///
1848 : /// 5.6.10.4: "For each of the NGSI-LD Entity IDs included in the input
1849 : /// Array execute the behaviour defined by clause 5.6.6, but limited to a
1850 : /// local operation" — so an entity past its `expiresAt` is absent (4.22)
1851 : /// and is not deleted here either, exactly as `delete` refuses it.
1852 10 : pub async fn batch_delete(
1853 10 : &self,
1854 10 : tenant: &TenantId,
1855 10 : ids: &[String],
1856 10 : ) -> Result<Vec<(String, Value)>, sqlx::Error> {
1857 10 : Ok(self
1858 10 : .batch_delete_stamped(tenant, ids)
1859 10 : .await?
1860 10 : .into_iter()
1861 10 : .map(|(id, prev, _)| (id, prev))
1862 10 : .collect())
1863 10 : }
1864 :
1865 : /// [`Self::batch_delete`] with each delete's [`ChangeVersion`].
1866 176 : pub async fn batch_delete_stamped(
1867 176 : &self,
1868 176 : tenant: &TenantId,
1869 176 : ids: &[String],
1870 176 : ) -> Result<Vec<(String, Value, ChangeVersion)>, sqlx::Error> {
1871 176 : let mut tx = super::begin(&self.pool).await?;
1872 176 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1873 176 : let rows = sqlx::query(
1874 176 : "DELETE FROM entities WHERE tenant_id = $1 AND id = ANY($2)
1875 176 : AND (expires_at IS NULL OR expires_at > now())
1876 176 : RETURNING id, entity, types, version, created_at::text",
1877 176 : )
1878 176 : .bind(tenant.as_str())
1879 176 : .bind(ids)
1880 176 : .fetch_all(&mut *tx)
1881 176 : .await?;
1882 176 : if self.outbox_on() {
1883 0 : let events: Vec<Value> = rows
1884 0 : .iter()
1885 0 : .map(|r| {
1886 0 : let id: String = r.get(0);
1887 0 : let prev: Value = r.get(1);
1888 0 : let types: Vec<String> = r.get(2);
1889 0 : change_event(
1890 0 : tenant,
1891 0 : "delete",
1892 0 : &id,
1893 0 : &types,
1894 0 : Some(&prev),
1895 0 : None,
1896 0 : r.get::<i64, _>(3),
1897 0 : r.get::<&str, _>(4),
1898 : )
1899 0 : })
1900 0 : .collect();
1901 0 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
1902 176 : }
1903 176 : tx.commit().await?;
1904 176 : Ok(rows
1905 176 : .iter()
1906 1468 : .map(|r| {
1907 1468 : let v = ChangeVersion {
1908 1468 : incarnation: r.get(4),
1909 1468 : version: r.get::<i64, _>(3).saturating_add(1),
1910 1468 : };
1911 1468 : (r.get::<String, _>(0), r.get::<Value, _>(1), v)
1912 1468 : })
1913 176 : .collect())
1914 176 : }
1915 :
1916 : /// Batch upsert in REPLACE semantics — ONE
1917 : /// `INSERT … ON CONFLICT DO UPDATE` for the whole batch, one transaction.
1918 : /// Returns a created-flag per input item (input order; duplicates of one
1919 : /// id report the first outcome). Before-images for the change events are
1920 : /// captured by a single `FOR NO KEY UPDATE` select in the same
1921 : /// transaction (see [`Self::mutate`]). Returns per input item: (created?,
1922 : /// before-image) — the before-image is for the caller's change hook and
1923 : /// comes from the same-tx lock.
1924 : ///
1925 : /// 5.6.8.4: "Create the Entity locally if it does not exist (i.e. no
1926 : /// Entity with the same Entity ID is present) executing the behaviour
1927 : /// defined by clause 5.6.1, but limited to a local operation" — an
1928 : /// entity past its `expiresAt` does not exist (4.22), so it takes the
1929 : /// creation branch: fresh `created_at`, `version` back to 1 and a
1930 : /// created-flag, which is what splits 201 from 204 at the API.
1931 24 : pub async fn batch_upsert_replace(
1932 24 : &self,
1933 24 : tenant: &TenantId,
1934 24 : items: &[(String, Value)],
1935 24 : ) -> Result<Vec<(bool, Option<Value>)>, sqlx::Error> {
1936 24 : Ok(self
1937 24 : .batch_upsert_replace_stamped(tenant, items)
1938 24 : .await?
1939 23 : .into_iter()
1940 2657 : .map(|(created, prev, _)| (created, prev))
1941 23 : .collect())
1942 24 : }
1943 :
1944 : /// [`Self::batch_upsert_replace`] with each written row's
1945 : /// [`ChangeVersion`].
1946 82 : pub async fn batch_upsert_replace_stamped(
1947 82 : &self,
1948 82 : tenant: &TenantId,
1949 82 : items: &[(String, Value)],
1950 82 : ) -> Result<Vec<(bool, Option<Value>, Option<ChangeVersion>)>, sqlx::Error> {
1951 82 : let mut seen = std::collections::HashSet::new();
1952 82 : let mut payload = Vec::new();
1953 82 : let mut ids = Vec::new();
1954 2756 : for (id, doc) in items {
1955 2756 : if seen.insert(id.as_str()) {
1956 2752 : let e = extract(doc);
1957 2752 : ids.push(id.clone());
1958 2752 : payload.push(serde_json::json!({
1959 2752 : "id": id, "doc": doc, "types": e.types, "scopes": e.scopes,
1960 2752 : "created": e.created, "modified": e.modified, "expires": e.expires,
1961 2752 : "location": e.location, "loc_ambiguous": e.location_ambiguous,
1962 2752 : "attrs_expire": e.attrs_expire,
1963 2752 : }));
1964 2752 : }
1965 : }
1966 82 : let mut tx = self.begin_batch().await?;
1967 81 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
1968 81 : crate::store::pg::claim_tenant(&mut tx, tenant).await?;
1969 : // lock + before-images in one statement (ordered: stable lock order)
1970 81 : let prev_rows = sqlx::query(
1971 81 : "SELECT id, entity,
1972 81 : (expires_at IS NOT NULL AND expires_at <= now()) AS expired
1973 81 : FROM entities WHERE tenant_id = $1 AND id = ANY($2)
1974 81 : ORDER BY id FOR NO KEY UPDATE",
1975 81 : )
1976 81 : .bind(tenant.as_str())
1977 81 : .bind(&ids)
1978 81 : .fetch_all(&mut *tx)
1979 81 : .await?;
1980 : // 4.22: an expired row is already invalid, so it is not a
1981 : // before-image — the upsert that replaces it is a creation and
1982 : // its change event has nothing before it. The lock still covers
1983 : // the row, so the stable lock order is unchanged.
1984 81 : let prevs: std::collections::HashMap<String, Value> = prev_rows
1985 81 : .into_iter()
1986 86 : .filter(|r| !r.get::<bool, _>(2))
1987 84 : .map(|r| (r.get::<String, _>(0), r.get::<Value, _>(1)))
1988 81 : .collect();
1989 81 : let rows = sqlx::query(
1990 81 : "INSERT INTO entities
1991 81 : (tenant_id, id, entity, types, scopes, created_at, modified_at, expires_at,
1992 81 : location, location_ambiguous, attrs_expire_at)
1993 81 : SELECT $1, e->>'id', e->'doc',
1994 81 : ARRAY(SELECT jsonb_array_elements_text(e->'types')),
1995 81 : CASE WHEN e->'scopes' = 'null'::jsonb THEN NULL
1996 81 : ELSE ARRAY(SELECT jsonb_array_elements_text(e->'scopes')) END,
1997 81 : (e->>'created')::timestamptz, (e->>'modified')::timestamptz,
1998 81 : (e->>'expires')::timestamptz,
1999 81 : CASE WHEN ST_IsValid(try_geomfromgeojson(e->>'location'))
2000 81 : THEN try_geomfromgeojson(e->>'location') END,
2001 81 : COALESCE((e->>'loc_ambiguous')::bool, false)
2002 81 : OR (e->>'location' IS NOT NULL
2003 81 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson(e->>'location')), false)),
2004 81 : try_timestamptz(e->>'attrs_expire')
2005 81 : FROM jsonb_array_elements($2::jsonb) AS e
2006 81 : ON CONFLICT (tenant_id, id) DO UPDATE SET
2007 81 : entity = EXCLUDED.entity, types = EXCLUDED.types,
2008 81 : scopes = EXCLUDED.scopes, modified_at = EXCLUDED.modified_at,
2009 81 : expires_at = EXCLUDED.expires_at, location = EXCLUDED.location,
2010 81 : location_ambiguous = EXCLUDED.location_ambiguous,
2011 81 : attrs_expire_at = EXCLUDED.attrs_expire_at,
2012 81 : created_at = CASE WHEN entities.expires_at IS NOT NULL
2013 81 : AND entities.expires_at <= now()
2014 81 : THEN EXCLUDED.created_at
2015 81 : ELSE entities.created_at END,
2016 81 : version = CASE WHEN entities.expires_at IS NOT NULL
2017 81 : AND entities.expires_at <= now()
2018 81 : THEN 1 ELSE entities.version + 1 END
2019 81 : RETURNING id, (xmax = 0 OR version = 1) AS inserted,
2020 81 : version, created_at::text",
2021 81 : )
2022 81 : .bind(tenant.as_str())
2023 81 : .bind(Value::Array(payload))
2024 81 : .fetch_all(&mut *tx)
2025 81 : .await?;
2026 81 : let mut created: std::collections::HashMap<String, (bool, ChangeVersion)> =
2027 81 : std::collections::HashMap::new();
2028 81 : let mut events = Vec::new();
2029 2751 : for r in &rows {
2030 2751 : let id: String = r.get(0);
2031 2751 : let inserted: bool = r.get(1);
2032 2751 : let version = ChangeVersion {
2033 2751 : version: r.get(2),
2034 2751 : incarnation: r.get(3),
2035 2751 : };
2036 : // the id came out of `items`, so the lookup answers; an
2037 : // outbox row for a document that is not there would carry no
2038 : // payload, so there is nothing to emit either way
2039 0 : if let (true, Some(doc)) = (
2040 2751 : self.outbox_on(),
2041 1091677 : items.iter().find(|(i, _)| *i == id).map(|(_, d)| d),
2042 : ) {
2043 0 : let e = extract(doc);
2044 0 : events.push(change_event(
2045 0 : tenant,
2046 0 : if inserted { "create" } else { "update" },
2047 0 : &id,
2048 0 : &e.types,
2049 0 : prevs.get(&id),
2050 0 : Some(doc),
2051 0 : r.get::<i64, _>(2),
2052 0 : r.get::<&str, _>(3),
2053 : ));
2054 2751 : }
2055 2751 : created.insert(id, (inserted, version));
2056 : }
2057 81 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
2058 81 : tx.commit().await?;
2059 81 : Ok(items
2060 81 : .iter()
2061 2755 : .map(|(id, _)| {
2062 2755 : let (inserted, version) = created
2063 2755 : .get(id)
2064 2755 : .map_or((false, None), |(c, v)| (*c, Some(v.clone())));
2065 2755 : (inserted, prevs.get(id).cloned(), version)
2066 2755 : })
2067 81 : .collect())
2068 82 : }
2069 :
2070 : /// Batch read-modify-write — ONE transaction,
2071 : /// all rows locked in a single ordered `FOR NO KEY UPDATE` select (see
2072 : /// [`Self::mutate`]), the closure
2073 : /// applied per doc in Rust, and ONE multi-row UPDATE writeback. Per-item
2074 : /// results align with `ids`: `None` = absent, `Some(Err)` = the closure
2075 : /// rejected that item (its write is skipped, the rest proceed).
2076 18 : pub async fn batch_mutate<E>(
2077 18 : &self,
2078 18 : tenant: &TenantId,
2079 18 : ids: &[String],
2080 18 : f: impl FnMut(&str, &mut Value) -> Result<(), E>,
2081 18 : ) -> Result<Vec<Option<Result<(), E>>>, sqlx::Error> {
2082 18 : Ok(self.batch_mutate_stamped(tenant, ids, f).await?.0)
2083 18 : }
2084 :
2085 : /// [`Self::batch_mutate`] with the [`ChangeVersion`] of each row written.
2086 : #[allow(clippy::type_complexity)] // the plain results and their versions
2087 84 : pub async fn batch_mutate_stamped<E>(
2088 84 : &self,
2089 84 : tenant: &TenantId,
2090 84 : ids: &[String],
2091 84 : mut f: impl FnMut(&str, &mut Value) -> Result<(), E>,
2092 84 : ) -> Result<
2093 84 : (
2094 84 : Vec<Option<Result<(), E>>>,
2095 84 : std::collections::HashMap<String, ChangeVersion>,
2096 84 : ),
2097 84 : sqlx::Error,
2098 84 : > {
2099 84 : let mut tx = self.begin_batch().await?;
2100 84 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
2101 84 : let rows = sqlx::query(
2102 84 : // 4.22: an expired entity is invalid, so it is not a row this
2103 84 : // update can find — 5.6.9.4 is clause 5.6.3 run per item, and
2104 84 : // 5.6.3 on an absent entity is ResourceNotFound. Without the
2105 84 : // predicate the batch reported the id as updated, wrote to
2106 84 : // it, and emitted a change event for an entity every read
2107 84 : // refuses. ORDER BY id keeps the lock order stable.
2108 84 : "SELECT id, entity FROM entities
2109 84 : WHERE tenant_id = $1 AND id = ANY($2)
2110 84 : AND (expires_at IS NULL OR expires_at > now())
2111 84 : ORDER BY id FOR NO KEY UPDATE",
2112 84 : )
2113 84 : .bind(tenant.as_str())
2114 84 : .bind(ids)
2115 84 : .fetch_all(&mut *tx)
2116 84 : .await?;
2117 84 : let mut docs: std::collections::HashMap<String, Value> = rows
2118 84 : .into_iter()
2119 6934 : .map(|r| (r.get::<String, _>(0), r.get::<Value, _>(1)))
2120 84 : .collect();
2121 84 : let mut results: Vec<Option<Result<(), E>>> = Vec::with_capacity(ids.len());
2122 : // 5.6.9.4 is clause 5.6.3 run per item, so a repeated id applies
2123 : // its closure once per item onto the SAME document and the last
2124 : // application is the state the row must end at. The writeback
2125 : // below is one `UPDATE … FROM`, which updates a target row once
2126 : // from whichever source row the join happens to reach, so the
2127 : // statement is fed one row per id — the document as the whole
2128 : // batch left it — rather than one per input item.
2129 84 : let mut write_ids: std::collections::HashSet<&str> = std::collections::HashSet::new();
2130 84 : let mut changed: Vec<(String, Value, Value)> = Vec::new(); // id, before, after
2131 6952 : for id in ids {
2132 6952 : match docs.get_mut(id) {
2133 16 : None => results.push(None),
2134 6936 : Some(doc) => {
2135 6936 : let before = doc.clone();
2136 6936 : match f(id, doc) {
2137 2 : Err(e) => {
2138 2 : *doc = before; // closure failed: discard its edits
2139 2 : results.push(Some(Err(e)));
2140 2 : }
2141 : Ok(()) => {
2142 6934 : if *doc != before {
2143 4932 : write_ids.insert(id.as_str());
2144 4932 : changed.push((id.clone(), before, doc.clone()));
2145 4932 : }
2146 6934 : results.push(Some(Ok(())));
2147 : }
2148 : }
2149 : }
2150 : }
2151 : }
2152 84 : let payload: Vec<Value> = write_ids
2153 84 : .iter()
2154 4932 : .filter_map(|id| docs.get(*id).map(|doc| (*id, doc)))
2155 4932 : .map(|(id, doc)| {
2156 4932 : let e = extract(doc);
2157 4932 : serde_json::json!({
2158 4932 : "id": id, "doc": doc, "types": e.types,
2159 4932 : "scopes": e.scopes, "modified": e.modified,
2160 4932 : "expires": e.expires, "location": e.location,
2161 4932 : "loc_ambiguous": e.location_ambiguous,
2162 4932 : "attrs_expire": e.attrs_expire,
2163 : })
2164 4932 : })
2165 84 : .collect();
2166 84 : let mut versions = std::collections::HashMap::new();
2167 84 : if !payload.is_empty() {
2168 78 : let updated = sqlx::query(
2169 78 : "UPDATE entities t SET
2170 78 : entity = e->'doc',
2171 78 : types = ARRAY(SELECT jsonb_array_elements_text(e->'types')),
2172 78 : scopes = CASE WHEN e->'scopes' = 'null'::jsonb THEN NULL
2173 78 : ELSE ARRAY(SELECT jsonb_array_elements_text(e->'scopes')) END,
2174 78 : modified_at = (e->>'modified')::timestamptz,
2175 78 : expires_at = (e->>'expires')::timestamptz,
2176 78 : location = CASE WHEN ST_IsValid(try_geomfromgeojson(e->>'location'))
2177 78 : THEN try_geomfromgeojson(e->>'location') END,
2178 78 : location_ambiguous = COALESCE((e->>'loc_ambiguous')::bool, false)
2179 78 : OR (e->>'location' IS NOT NULL
2180 78 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson(e->>'location')), false)),
2181 78 : attrs_expire_at = try_timestamptz(e->>'attrs_expire'),
2182 78 : version = t.version + 1
2183 78 : FROM jsonb_array_elements($2::jsonb) AS e
2184 78 : WHERE t.tenant_id = $1 AND t.id = e->>'id'
2185 78 : RETURNING t.id, t.version, t.created_at::text",
2186 78 : )
2187 78 : .bind(tenant.as_str())
2188 78 : .bind(Value::Array(payload))
2189 78 : .fetch_all(&mut *tx)
2190 78 : .await?;
2191 78 : versions = updated
2192 78 : .into_iter()
2193 4932 : .map(|r| {
2194 4932 : (
2195 4932 : r.get::<String, _>(0),
2196 4932 : ChangeVersion {
2197 4932 : version: r.get(1),
2198 4932 : incarnation: r.get(2),
2199 4932 : },
2200 4932 : )
2201 4932 : })
2202 78 : .collect();
2203 78 : if self.outbox_on() {
2204 0 : let events: Vec<Value> = changed
2205 0 : .iter()
2206 0 : .filter_map(|(id, before, after)| {
2207 0 : let e = extract(after);
2208 0 : let v = versions.get(id)?;
2209 0 : Some(change_event(
2210 0 : tenant,
2211 0 : "update",
2212 0 : id,
2213 0 : &e.types,
2214 0 : Some(before),
2215 0 : Some(after),
2216 0 : v.version,
2217 0 : &v.incarnation,
2218 0 : ))
2219 0 : })
2220 0 : .collect();
2221 0 : super::outbox::enqueue_many(&mut tx, tenant, &events).await?;
2222 78 : }
2223 6 : }
2224 84 : tx.commit().await?;
2225 84 : Ok((results, versions))
2226 84 : }
2227 :
2228 : /// Current row version (test hook for the version-monotonicity assertions).
2229 : #[cfg(any(test, feature = "test-kit"))]
2230 8 : pub async fn version(&self, tenant: &TenantId, id: &str) -> Result<Option<i64>, sqlx::Error> {
2231 8 : let mut tx = super::begin(&self.pool).await?;
2232 8 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
2233 8 : let row = sqlx::query("SELECT version FROM entities WHERE tenant_id = $1 AND id = $2")
2234 8 : .bind(tenant.as_str())
2235 8 : .bind(id)
2236 8 : .fetch_optional(&mut *tx)
2237 8 : .await?;
2238 8 : tx.commit().await?;
2239 8 : Ok(row.map(|r| r.get::<i64, _>(0)))
2240 8 : }
2241 : }
2242 :
2243 : /// 4.22 for the current state: drop every Attribute instance whose
2244 : /// `expiresAt` passed from the stored docs, at most `batch` of them, inside
2245 : /// the sweep's transaction `tx` (service escape on). The rows come by the
2246 : /// `attrs_expire_at` index; each Tenant's are locked and rewritten under that
2247 : /// Tenant, as 0005 grants the escape no UPDATE. `modified_at` stays: an
2248 : /// expiry is no modification, so no change event is owed either. A row a
2249 : /// writer holds is skipped and taken on the next tick. Returns the rows
2250 : /// rewritten.
2251 1007 : pub(super) async fn prune_expired_current(
2252 1007 : tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
2253 1007 : batch: i64,
2254 1007 : ) -> Result<u64, sqlx::Error> {
2255 1007 : let due: Vec<(String, String)> = sqlx::query_as(
2256 1007 : "SELECT tenant_id, id FROM entities
2257 1007 : WHERE attrs_expire_at IS NOT NULL AND attrs_expire_at < now()
2258 1007 : ORDER BY attrs_expire_at LIMIT $1",
2259 1007 : )
2260 1007 : .bind(batch)
2261 1007 : .fetch_all(&mut **tx)
2262 1007 : .await?;
2263 1007 : let mut by_tenant: std::collections::BTreeMap<&str, Vec<&str>> =
2264 1007 : std::collections::BTreeMap::new();
2265 1007 : for (tenant, id) in &due {
2266 19 : by_tenant.entry(tenant).or_default().push(id);
2267 19 : }
2268 1007 : let mut n = 0u64;
2269 1007 : for (tenant, ids) in by_tenant {
2270 16 : sqlx::query(crate::SET_TENANT_SQL)
2271 16 : .bind(tenant)
2272 16 : .execute(&mut **tx)
2273 16 : .await?;
2274 16 : let rows = sqlx::query(
2275 16 : "SELECT id, entity,
2276 16 : to_char(now() AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US\"Z\"')
2277 16 : FROM entities
2278 16 : WHERE tenant_id = $1 AND id = ANY($2::text[]) AND attrs_expire_at < now()
2279 16 : FOR NO KEY UPDATE SKIP LOCKED",
2280 16 : )
2281 16 : .bind(tenant)
2282 16 : .bind(&ids)
2283 16 : .fetch_all(&mut **tx)
2284 16 : .await?;
2285 17 : for r in rows {
2286 17 : let id: String = r.get(0);
2287 17 : let mut doc: Value = r.get(1);
2288 17 : let now: String = r.get(2);
2289 17 : crate::store::mem::prune_expired_instances(&mut doc, &now);
2290 17 : let e = extract(&doc);
2291 17 : let rewritten = sqlx::query(
2292 17 : "UPDATE entities SET entity = $3,
2293 17 : location = CASE WHEN ST_IsValid(try_geomfromgeojson($4))
2294 17 : THEN try_geomfromgeojson($4) END,
2295 17 : location_ambiguous = $5 OR ($4 IS NOT NULL
2296 17 : AND NOT COALESCE(ST_IsValid(try_geomfromgeojson($4)), false)),
2297 17 : attrs_expire_at = try_timestamptz($6),
2298 17 : version = version + 1
2299 17 : WHERE tenant_id = $1 AND id = $2",
2300 17 : )
2301 17 : .bind(tenant)
2302 17 : .bind(&id)
2303 17 : .bind(&doc)
2304 17 : .bind(&e.location)
2305 17 : .bind(e.location_ambiguous)
2306 17 : .bind(&e.attrs_expire)
2307 17 : .execute(&mut **tx)
2308 17 : .await?
2309 17 : .rows_affected();
2310 17 : n = n.saturating_add(rewritten);
2311 : }
2312 : }
2313 1007 : Ok(n)
2314 1007 : }
2315 :
2316 : /// One cheap probe on `tx`: is the client's query geometry valid?
2317 : /// Unparseable GeoJSON counts as invalid — same outcome, no pushdown — and
2318 : /// reaches that answer through `try_geomfromgeojson` (0001_init.sql), so only
2319 : /// the connection can make the probe fail.
2320 1617 : async fn geometry_is_valid(
2321 1617 : tx: &mut sqlx::PgConnection,
2322 1617 : spec: &crate::compile::geo::GeoSpec<'_>,
2323 1617 : ) -> Result<bool, sqlx::Error> {
2324 1617 : let geojson = serde_json::to_string(&serde_json::json!({
2325 1617 : "type": spec.geometry, "coordinates": spec.coordinates
2326 1617 : }))
2327 1617 : .unwrap_or_default();
2328 1617 : sqlx::query_scalar::<_, bool>("SELECT COALESCE(ST_IsValid(try_geomfromgeojson($1)), false)")
2329 1617 : .bind(&geojson)
2330 1617 : .fetch_one(&mut *tx)
2331 1617 : .await
2332 1617 : }
2333 :
2334 : #[cfg(test)]
2335 : mod tests {
2336 : use super::*;
2337 : use serde_json::json;
2338 :
2339 18 : fn wheres() -> Vec<String> {
2340 18 : vec!["tenant_id = $1".to_owned()]
2341 18 : }
2342 :
2343 : /// only a `!location` every match must satisfy narrows to
2344 : /// `location IS NULL`; an alternative, a sub-path or another
2345 : /// geoproperty does not.
2346 : #[test]
2347 2 : fn only_a_required_missing_location_narrows_to_no_geometry() {
2348 8 : let expand = |t: &str| match t {
2349 8 : "location" => crate::compile::geo::LOCATION_IRI.to_owned(),
2350 2 : t => format!("https://x/{t}"),
2351 8 : };
2352 18 : let narrows = |q: &str| unlocated(&antares_ql::parse_q(q).expect("q parses"), &expand);
2353 6 : for q in ["!location", "a>1;!location", "(a==1;!location);b"] {
2354 6 : assert!(narrows(q), "{q}");
2355 : }
2356 12 : for q in [
2357 2 : "location",
2358 2 : "!location|a",
2359 2 : "!location.observedAt",
2360 2 : "!place",
2361 2 : "a;b",
2362 2 : "!location[x]",
2363 2 : ] {
2364 12 : assert!(!narrows(q), "{q}");
2365 : }
2366 2 : }
2367 :
2368 : /// only a `name=="…"` string every match must equal narrows to
2369 : /// `name_key`; an alternative, another operator or value, a sub-path or
2370 : /// another attribute does not.
2371 : /// the top-level terms `entity_values` keys, each probing the
2372 : /// rows a match must hold; everything else is left to the `@?`.
2373 : #[test]
2374 2 : fn a_top_level_value_term_probes_entity_values() {
2375 28 : let expand = |t: &str| format!("https://x/{t}");
2376 2 : let tenant = TenantId::new("t").expect("tenant");
2377 2 : let long = "é".repeat(600);
2378 2 : let long_q = format!(r#"b=="{long}""#);
2379 16 : for (q, attr, cond, nums, text) in [
2380 2 : (
2381 2 : "a==42",
2382 2 : "https://x/a",
2383 2 : "f = ANY(ARRAY[$3, 'NaN'::float8])",
2384 2 : vec![42.0],
2385 2 : None,
2386 2 : ),
2387 2 : ("a>1.5", "https://x/a", "f >= $3", vec![1.5], None),
2388 2 : (
2389 2 : "x==1;a>=2",
2390 2 : "https://x/x",
2391 2 : "f = ANY(ARRAY[$3, 'NaN'::float8])",
2392 2 : vec![1.0],
2393 2 : None,
2394 2 : ),
2395 2 : (
2396 2 : "a<0",
2397 2 : "https://x/a",
2398 2 : "(f <= $3 OR f = 'NaN')",
2399 2 : vec![0.0],
2400 2 : None,
2401 2 : ),
2402 2 : (
2403 2 : "a<=-1",
2404 2 : "https://x/a",
2405 2 : "(f <= $3 OR f = 'NaN')",
2406 2 : vec![-1.0],
2407 2 : None,
2408 2 : ),
2409 2 : (
2410 2 : "a==1..3",
2411 2 : "https://x/a",
2412 2 : "(f BETWEEN $3 AND $4 OR f = 'NaN')",
2413 2 : vec![1.0, 3.0],
2414 2 : None,
2415 2 : ),
2416 2 : (
2417 2 : r#"a=="x""#,
2418 2 : "https://x/a",
2419 2 : "(txt = $3 OR f = 'NaN')",
2420 2 : vec![],
2421 2 : Some("x"),
2422 2 : ),
2423 2 : (long_q.as_str(), "https://x/b", "f = 'NaN'", vec![], None),
2424 2 : ] {
2425 16 : let node = antares_ql::parse_q(q).expect("q parses");
2426 16 : let (iri, key) = valued(&node, &expand).expect(q);
2427 16 : assert_eq!(iri, attr, "{q}");
2428 16 : let p = value_probe(&tenant, iri, key);
2429 16 : let limit = p.binds.len();
2430 16 : assert_eq!(
2431 : p.sql,
2432 16 : format!("SELECT id FROM entity_values WHERE tenant_id = $1 AND attr = $2 AND {cond} LIMIT ${limit}"),
2433 : "{q}"
2434 : );
2435 16 : let texts: Vec<&str> = p
2436 16 : .binds
2437 16 : .iter()
2438 64 : .filter_map(|b| match b {
2439 34 : Bind::Text(t) => Some(t.as_str()),
2440 30 : _ => None,
2441 64 : })
2442 16 : .collect();
2443 16 : assert_eq!(
2444 : texts,
2445 16 : ["t", attr].into_iter().chain(text).collect::<Vec<_>>(),
2446 : "{q}"
2447 : );
2448 16 : let got_nums: Vec<f64> = p
2449 16 : .binds
2450 16 : .iter()
2451 64 : .filter_map(|b| match b {
2452 14 : Bind::Num(n) => Some(*n),
2453 50 : _ => None,
2454 64 : })
2455 16 : .collect();
2456 16 : assert_eq!(got_nums, nums, "{q}");
2457 16 : assert!(matches!(p.binds.last(), Some(Bind::Int(PROBE_ROWS))), "{q}");
2458 : }
2459 22 : for q in [
2460 2 : "a",
2461 2 : "!a",
2462 2 : "a!=1",
2463 2 : "a==1|b==2",
2464 2 : "a==true",
2465 2 : "a==1,2",
2466 2 : r#"a=="x".."y""#,
2467 2 : r#"a~="x""#,
2468 2 : "a.sub==1",
2469 2 : "a[k]==1",
2470 2 : r#"a>"x""#,
2471 2 : ] {
2472 22 : let node = antares_ql::parse_q(q).expect("q parses");
2473 22 : assert_eq!(valued(&node, &expand), None, "{q}");
2474 : }
2475 2 : }
2476 :
2477 : /// `attrs=` probes presence rows unless it names one of the
2478 : /// Entity's own members, which have none.
2479 : #[test]
2480 2 : fn attrs_probe_presence_rows_unless_they_name_an_own_member() {
2481 2 : let tenant = TenantId::new("t").expect("tenant");
2482 2 : let attrs = ["https://x/a".to_owned(), "https://x/b".to_owned()];
2483 2 : let p = attrs_probe(&tenant, &attrs).expect("probe");
2484 2 : assert_eq!(
2485 : p.sql,
2486 : "SELECT id FROM entity_values WHERE tenant_id = $1 AND attr = ANY($2) LIMIT $3"
2487 : );
2488 2 : assert!(matches!(p.binds.get(1), Some(Bind::TextArr(v)) if v.as_slice() == attrs));
2489 16 : for own in OWN_MEMBERS {
2490 16 : let attrs = ["https://x/a".to_owned(), (*own).to_owned()];
2491 16 : assert!(attrs_probe(&tenant, &attrs).is_none(), "{own}");
2492 : }
2493 2 : }
2494 :
2495 : #[test]
2496 2 : fn only_a_required_name_string_narrows_to_the_name_key() {
2497 8 : let expand = |t: &str| match t {
2498 8 : "name" => NAME_IRI.to_owned(),
2499 2 : t => format!("https://x/{t}"),
2500 8 : };
2501 2 : let wants =
2502 24 : |q: &str| named(&antares_ql::parse_q(q).expect("q parses"), &expand).map(str::to_owned);
2503 6 : for (q, want) in [
2504 2 : (r#"name=="Stredočeský""#, "Stredočeský"),
2505 2 : (r#"a>1;name=="x""#, "x"),
2506 2 : (r#"(a==1;name=="");b"#, ""),
2507 2 : ] {
2508 6 : assert_eq!(wants(q).as_deref(), Some(want), "{q}");
2509 : }
2510 18 : for q in [
2511 2 : "name",
2512 2 : r#"name!="x""#,
2513 2 : r#"name=="x"|a==1"#,
2514 2 : "name==1",
2515 2 : r#"name=="x","y""#,
2516 2 : r#"name~="x""#,
2517 2 : r#"name.sub=="x""#,
2518 2 : r#"name[x]=="x""#,
2519 2 : r#"title=="x""#,
2520 2 : ] {
2521 18 : assert_eq!(wants(q), None, "{q}");
2522 : }
2523 2 : }
2524 :
2525 : /// `location_ambiguous` is the single bit that keeps geoquery pushdown
2526 : /// honest: `compile_geo` ORs it into every predicate so a row whose
2527 : /// default GeoProperty could not be reduced to ONE geometry still reaches
2528 : /// the evaluator. It must be true exactly when the doc CARRIES the
2529 : /// geoproperty and extraction refused — a row with no geoproperty at all
2530 : /// can never match a geoquery and is excluded in SQL, which is the whole
2531 : /// point of the column.
2532 : #[test]
2533 2 : fn location_ambiguous_is_carried_but_unextractable() {
2534 2 : let loc = crate::compile::geo::LOCATION_IRI;
2535 2 : let point = json!({"type": "Point", "coordinates": [2.29, 48.85]});
2536 :
2537 : // one default GeoProperty instance with a geometry: extracted, exact
2538 2 : let e = extract(&json!({"id": "urn:x", loc: [{"value": point}]}));
2539 2 : assert!(e.location.is_some());
2540 2 : assert!(!e.location_ambiguous);
2541 :
2542 : // multi-instance: no single geometry, so the evaluator arbitrates
2543 2 : let e = extract(&json!({"id": "urn:x", loc: [{"value": point}, {"value": point}]}));
2544 2 : assert!(e.location.is_none());
2545 2 : assert!(
2546 2 : e.location_ambiguous,
2547 : "a multi-instance location must reach the evaluator"
2548 : );
2549 :
2550 : // a GeometryCollection and a non-GeoJSON value are the same case
2551 6 : for v in [
2552 2 : json!({"type": "GeometryCollection", "geometries": []}),
2553 2 : json!("somewhere"),
2554 2 : json!({"coordinates": [1, 2]}),
2555 2 : ] {
2556 6 : let e = extract(&json!({"id": "urn:x", loc: [{"value": v}]}));
2557 6 : assert!(e.location.is_none(), "{v} must not become a geometry");
2558 6 : assert!(e.location_ambiguous, "{v} must reach the evaluator");
2559 : }
2560 :
2561 : // NO geoproperty: not ambiguous — the row is excluded in SQL
2562 2 : let e = extract(&json!({"id": "urn:x", "https://a/speed": [{"value": 1}]}));
2563 2 : assert!(e.location.is_none());
2564 2 : assert!(
2565 2 : !e.location_ambiguous,
2566 : "a row without a location must stay excludable in SQL"
2567 : );
2568 2 : }
2569 :
2570 : /// The rest of `extract`: `type`/`scope` accept the string form and the
2571 : /// array form, an absent `scope` is SQL NULL (never an empty array, which
2572 : /// would mean "scoped to nothing"), and a missing system timestamp falls
2573 : /// back rather than failing the write.
2574 : #[test]
2575 2 : fn extract_reads_types_scopes_and_stamps() {
2576 2 : let e = extract(&json!({"id": "urn:x", "type": "T", "scope": "/a"}));
2577 2 : assert_eq!(e.types, ["T"]);
2578 2 : assert_eq!(e.scopes.as_deref(), Some(&["/a".to_owned()][..]));
2579 2 : let e = extract(&json!({"id": "urn:x", "type": ["T", "U"], "scope": ["/a", "/b"]}));
2580 2 : assert_eq!(e.types, ["T", "U"]);
2581 2 : assert_eq!(e.scopes.expect("scopes").len(), 2);
2582 2 : let e = extract(&json!({"id": "urn:x"}));
2583 2 : assert!(e.types.is_empty());
2584 2 : assert!(e.scopes.is_none(), "absent scope must be SQL NULL");
2585 2 : assert!(e.expires.is_none());
2586 2 : assert_eq!(e.created, "1970-01-01T00:00:00Z");
2587 : // a non-string, non-array scope is present but names nothing
2588 2 : let e = extract(&json!({"id": "urn:x", "scope": 7}));
2589 2 : assert_eq!(e.scopes.as_deref(), Some(&[][..]));
2590 2 : }
2591 :
2592 : /// The outbox row IS the wire contract the drain deserializes, and the
2593 : /// drain DELETES any row it cannot decode — a renamed key loses the event
2594 : /// silently. Pin the key set and the operation vocabulary here, where the
2595 : /// producer lives.
2596 : #[test]
2597 2 : fn the_change_event_carries_exactly_the_wire_keys() {
2598 2 : let t = TenantId::default();
2599 2 : let doc = json!({"id": "urn:e", "https://a/speed": [{"value": 1}]});
2600 2 : let ev = change_event(
2601 2 : &t,
2602 2 : "create",
2603 2 : "urn:e",
2604 2 : &["T".to_owned()],
2605 2 : None,
2606 2 : Some(&doc),
2607 : 1,
2608 2 : "inc",
2609 : );
2610 2 : let keys: Vec<&str> = ev
2611 2 : .as_object()
2612 2 : .expect("object")
2613 2 : .keys()
2614 2 : .map(String::as_str)
2615 2 : .collect();
2616 2 : assert_eq!(
2617 : keys,
2618 : [
2619 : "changed_attrs",
2620 : "entity_id",
2621 : "incarnation",
2622 : "op",
2623 : "payload",
2624 : "prev_payload",
2625 : "tenant",
2626 : "types",
2627 : "version",
2628 : ]
2629 : );
2630 2 : assert_eq!(ev["op"], "create");
2631 2 : assert_eq!(ev["tenant"], t.as_str());
2632 2 : assert!(ev["prev_payload"].is_null(), "a create has no before-image");
2633 6 : for op in ["create", "update", "delete"] {
2634 6 : assert_eq!(
2635 6 : change_event(&t, op, "urn:e", &[], None, None, 1, "inc")["op"],
2636 : op
2637 : );
2638 : }
2639 2 : }
2640 :
2641 : /// `changed_attrs` names the top-level ATTRIBUTES that differ. A create
2642 : /// lists every attribute, a delete lists every prior one — and, the half
2643 : /// that actually bounds notification traffic, an attribute present in
2644 : /// both images with an EQUAL value is not listed, nor is any system
2645 : /// member that changed on every write.
2646 : #[test]
2647 2 : fn changed_attrs_lists_the_differences_and_nothing_else() {
2648 2 : let t = TenantId::default();
2649 10 : let ev = |prev: Option<&Value>, next: Option<&Value>| {
2650 10 : change_event(&t, "update", "urn:e", &[], prev, next, 1, "inc")["changed_attrs"].clone()
2651 10 : };
2652 2 : let a = json!({"id": "urn:e", "type": ["T"], "createdAt": "t0", "modifiedAt": "t0",
2653 2 : "https://a/x": [{"value": 1}], "https://a/y": [{"value": 2}]});
2654 2 : let b = json!({"id": "urn:e", "type": ["T"], "createdAt": "t0", "modifiedAt": "t9",
2655 2 : "https://a/x": [{"value": 1}], "https://a/y": [{"value": 3}]});
2656 2 : assert_eq!(ev(Some(&a), Some(&b)), json!(["https://a/y"]));
2657 2 : assert_eq!(ev(None, Some(&a)), json!(["https://a/x", "https://a/y"]));
2658 2 : assert_eq!(ev(Some(&a), None), json!(["https://a/x", "https://a/y"]));
2659 : // identical images change nothing at all
2660 2 : assert_eq!(ev(Some(&a), Some(&a)), json!([]));
2661 : // and no system member is ever named
2662 2 : let ids = json!({"id": "urn:other", "type": ["U"], "scope": "/s",
2663 2 : "modifiedAt": "t1", "deletedAt": "t1", "expiresAt": "t1"});
2664 2 : assert_eq!(
2665 2 : ev(Some(&a), Some(&ids)),
2666 2 : json!(["https://a/x", "https://a/y"])
2667 : );
2668 2 : }
2669 :
2670 10 : fn page(offset: i64, limit: i64) -> Page {
2671 10 : Page {
2672 10 : offset,
2673 10 : limit,
2674 10 : count: true,
2675 10 : }
2676 10 : }
2677 :
2678 : /// 5.5.9.1: a query SQL decided exactly keeps the pushed page — LIMIT
2679 : /// and OFFSET. count=true carries no window either: the window
2680 : /// kept every match, body and all, before the LIMIT could cut.
2681 : #[test]
2682 2 : fn a_decided_page_pushes_limit_and_offset() {
2683 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2684 2 : let (sql, paged) = query_sql(
2685 2 : "entity",
2686 2 : &wheres(),
2687 2 : None,
2688 2 : Some(&page(20, 10)),
2689 2 : true,
2690 2 : "id",
2691 2 : &mut binds,
2692 2 : );
2693 2 : assert!(paged);
2694 2 : assert_eq!(
2695 : sql,
2696 : "SELECT entity AS entity FROM entities WHERE tenant_id = $1 AND id IN (\
2697 : SELECT id FROM entities WHERE tenant_id = $1 ORDER BY id LIMIT $2 OFFSET $3) ORDER BY id"
2698 : );
2699 2 : assert!(!sql.contains("OVER"), "{sql}");
2700 2 : assert!(matches!(binds[1], Bind::Int(11)));
2701 2 : assert!(matches!(binds[2], Bind::Int(20)));
2702 : // the first page reads its documents directly
2703 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2704 2 : let (sql, _) = query_sql(
2705 2 : "entity",
2706 2 : &wheres(),
2707 2 : None,
2708 2 : Some(&page(0, 10)),
2709 2 : true,
2710 2 : "id",
2711 2 : &mut binds,
2712 2 : );
2713 2 : assert!(
2714 2 : sql.starts_with("SELECT entity AS entity FROM entities WHERE tenant_id"),
2715 : "{sql}"
2716 : );
2717 2 : assert!(sql.ends_with("ORDER BY id LIMIT $2 OFFSET $3"), "{sql}");
2718 2 : }
2719 :
2720 : /// one named type reads two leakproof index arms, each in id
2721 : /// order and bounded by what the page can need, never `types @>` alone
2722 : /// (which RLS keeps out of every index condition).
2723 : /// id prefixes are one bounding byte range ANDed with each
2724 : /// prefix's own, and a page over them orders by bytes, the C-collated
2725 : /// index's order; no prefix selects nothing.
2726 : #[test]
2727 2 : fn id_prefixes_are_byte_ranges_in_byte_order() {
2728 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2729 2 : let ps = ["b:".to_owned(), "a:".to_owned(), "z\0".to_owned()];
2730 2 : let w = prefix_ranges(&ps, &mut binds);
2731 2 : let c = "id COLLATE \"C\"";
2732 2 : assert_eq!(
2733 : w,
2734 2 : format!(
2735 : "({c} >= $2 AND {c} < $3 AND (({c} >= $4 AND {c} < $5) OR ({c} >= $6 AND {c} < $7)))"
2736 : )
2737 : );
2738 2 : let texts: Vec<&str> = binds
2739 2 : .iter()
2740 14 : .filter_map(|b| match b {
2741 14 : Bind::Text(t) => Some(t.as_str()),
2742 0 : _ => None,
2743 14 : })
2744 2 : .collect();
2745 2 : assert_eq!(texts, ["t", "a:", "b;", "a:", "a;", "b:", "b;"]);
2746 2 : assert_eq!(prefix_ranges(&[], &mut binds), "false");
2747 2 : assert_eq!(prefix_ranges(&["\0".to_owned()], &mut binds), "false");
2748 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2749 2 : let (sql, _) = query_sql(
2750 2 : "entity",
2751 2 : &["tenant_id = $1".to_owned()],
2752 2 : None,
2753 2 : Some(&page(0, 10)),
2754 2 : true,
2755 2 : c,
2756 2 : &mut binds,
2757 2 : );
2758 2 : assert_eq!(
2759 : sql,
2760 2 : format!(
2761 : "SELECT entity AS entity FROM entities WHERE tenant_id = $1 ORDER BY {c} LIMIT $2 OFFSET $3"
2762 : )
2763 : );
2764 2 : }
2765 :
2766 : #[test]
2767 2 : fn one_type_pages_through_two_bounded_arms() {
2768 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2769 2 : let split = type_selection(&[vec!["T".to_owned()]], &mut binds).expect("split");
2770 2 : assert!(matches!(&binds[2], Bind::Text(t) if t == "T"));
2771 2 : let (sql, paged) = query_sql(
2772 2 : "entity",
2773 2 : &wheres(),
2774 2 : Some(&split),
2775 2 : Some(&page(20, 10)),
2776 2 : true,
2777 2 : "id",
2778 2 : &mut binds,
2779 2 : );
2780 2 : assert!(paged);
2781 2 : assert_eq!(
2782 : sql,
2783 : "SELECT entity AS entity FROM entities WHERE tenant_id = $1 AND id IN (SELECT id FROM (\
2784 : (SELECT id FROM entities WHERE tenant_id = $1 AND types[1] = $3 \
2785 : ORDER BY id LIMIT $7) UNION ALL \
2786 : (SELECT id FROM entities WHERE tenant_id = $1 AND cardinality(types) > 1 \
2787 : AND types[1] <> ALL($4) AND (types @> $2) ORDER BY id LIMIT $7)) AS arms \
2788 : ORDER BY id LIMIT $5 OFFSET $6) ORDER BY id"
2789 : );
2790 2 : assert!(matches!(binds[4], Bind::Int(11)));
2791 2 : assert!(matches!(binds[5], Bind::Int(20)));
2792 2 : assert!(
2793 2 : matches!(binds[6], Bind::Int(31)),
2794 : "each arm: offset + limit + 1"
2795 : );
2796 : // the undecided ceiling bounds each arm and the merge alike
2797 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2798 2 : let split = type_selection(&[vec!["T".to_owned()]], &mut binds).expect("split");
2799 2 : let (sql, paged) = query_sql(
2800 2 : "entity",
2801 2 : &wheres(),
2802 2 : Some(&split),
2803 2 : None,
2804 2 : false,
2805 2 : "id",
2806 2 : &mut binds,
2807 2 : );
2808 2 : assert!(!paged);
2809 2 : assert_eq!(sql.matches("LIMIT $5").count(), 3, "{sql}");
2810 2 : assert!(!sql.contains("OFFSET"), "{sql}");
2811 2 : }
2812 :
2813 : /// `type=A,B` and `type=A;B` read one first-type arm per named
2814 : /// type, each filtered by the whole selection unless one type decides
2815 : /// it, plus the multi-type arm; a selection the arms cannot name keeps
2816 : /// the plain containment.
2817 : #[test]
2818 2 : fn a_type_list_reads_one_arm_per_type() {
2819 28 : let t = |s: &str| s.to_owned();
2820 2 : let mut binds = vec![Bind::Text(t("t"))];
2821 2 : let split = type_selection(&[vec![t("B")], vec![t("A")]], &mut binds).expect("split");
2822 2 : let w = arms(&[t("tenant_id = $1")], Some(&split));
2823 2 : assert_eq!(
2824 : w,
2825 : [
2826 : "tenant_id = $1 AND types[1] = $4 AND (types @> $2 OR types @> $3)",
2827 : "tenant_id = $1 AND types[1] = $5 AND (types @> $2 OR types @> $3)",
2828 : "tenant_id = $1 AND cardinality(types) > 1 AND types[1] <> ALL($6) \
2829 : AND (types @> $2 OR types @> $3)",
2830 : ]
2831 : );
2832 2 : assert!(matches!(&binds[3], Bind::Text(x) if x == "A"), "sorted");
2833 2 : assert!(matches!(&binds[5], Bind::TextArr(v) if v == &[t("A"), t("B")]));
2834 :
2835 2 : let mut binds = vec![Bind::Text(t("t"))];
2836 2 : let split = type_selection(&[vec![t("A"), t("B")]], &mut binds).expect("split");
2837 2 : assert!(!split.exact_first, "A;B is not decided by the first type");
2838 :
2839 2 : let mut binds = vec![Bind::Text(t("t"))];
2840 34 : let many: Vec<Vec<String>> = (0..=MAX_TYPE_ARMS).map(|i| vec![format!("T{i}")]).collect();
2841 2 : let w = type_selection(&many, &mut binds)
2842 2 : .err()
2843 2 : .expect("too many arms");
2844 2 : assert!(w.starts_with("(types @> $2 OR"), "{w}");
2845 2 : let mut binds = vec![Bind::Text(t("t"))];
2846 2 : assert!(type_selection(&[vec![t("A")], vec![]], &mut binds).is_err());
2847 2 : let mut binds = vec![Bind::Text(t("t"))];
2848 2 : assert_eq!(
2849 2 : type_selection(&[vec![t("A\0")]], &mut binds)
2850 2 : .err()
2851 2 : .as_deref(),
2852 : Some("false")
2853 : );
2854 2 : }
2855 :
2856 : /// a match of `A;B` has two types, so the multi-type arm alone
2857 : /// reads it; a first-type arm walked every Entity of A and kept none.
2858 : /// A one-type group beside it (`A,B;C`) keeps its first-type arm, and
2859 : /// `A;A` is the one type A.
2860 : #[test]
2861 2 : fn an_and_group_reads_the_multi_type_arm_only() {
2862 28 : let t = |s: &str| s.to_owned();
2863 2 : let mut binds = vec![Bind::Text(t("t"))];
2864 2 : let split = type_selection(&[vec![t("A"), t("B")]], &mut binds).expect("split");
2865 2 : assert_eq!(
2866 2 : arms(&[t("tenant_id = $1")], Some(&split)),
2867 : ["tenant_id = $1 AND cardinality(types) > 1 AND (types @> $2)"]
2868 : );
2869 2 : assert_eq!(binds.len(), 2, "no first-type binds");
2870 :
2871 2 : let mut binds = vec![Bind::Text(t("t"))];
2872 2 : let split =
2873 2 : type_selection(&[vec![t("A")], vec![t("B"), t("C")]], &mut binds).expect("split");
2874 2 : assert_eq!(
2875 2 : arms(&[t("tenant_id = $1")], Some(&split)),
2876 : [
2877 : "tenant_id = $1 AND types[1] = $4 AND (types @> $2 OR types @> $3)",
2878 : "tenant_id = $1 AND cardinality(types) > 1 AND types[1] <> ALL($5) \
2879 : AND (types @> $2 OR types @> $3)",
2880 : ]
2881 : );
2882 2 : assert!(matches!(&binds[3], Bind::Text(x) if x == "A"));
2883 2 : assert!(matches!(&binds[4], Bind::TextArr(v) if v == &[t("A")]));
2884 :
2885 2 : let mut binds = vec![Bind::Text(t("t"))];
2886 2 : let split = type_selection(&[vec![t("A"), t("A")]], &mut binds).expect("split");
2887 2 : assert_eq!(
2888 2 : arms(&[t("tenant_id = $1")], Some(&split)).len(),
2889 : 2,
2890 : "A;A is A"
2891 : );
2892 2 : }
2893 :
2894 : /// count=false never asks for a count; count=true asks only when the
2895 : /// page cannot show the total: more rows past it, or an off-the-end page.
2896 : #[test]
2897 2 : fn only_a_counted_page_that_cannot_show_its_total_asks_for_a_count() {
2898 2 : let p = |offset, count| Page {
2899 18 : offset,
2900 : limit: 10,
2901 18 : count,
2902 18 : };
2903 8 : for (offset, rows, more) in [(0, 10, true), (20, 0, false), (20, 3, false), (0, 0, false)] {
2904 8 : assert!(page_total(&p(offset, false), rows, more).is_some());
2905 : }
2906 2 : assert_eq!(page_total(&p(20, false), 10, true), Some(31));
2907 2 : assert_eq!(page_total(&p(0, true), 10, true), None);
2908 2 : assert_eq!(page_total(&p(20, true), 0, false), None, "off the end");
2909 2 : assert_eq!(page_total(&p(20, true), 3, false), Some(23), "last page");
2910 2 : assert_eq!(
2911 2 : page_total(&p(0, true), 0, false),
2912 : Some(0),
2913 : "nothing matches"
2914 : );
2915 2 : }
2916 :
2917 : /// The count runs over exactly the page's arms: the same WHERE and the
2918 : /// same binds, a count per arm, no window.
2919 : #[test]
2920 2 : fn the_count_reads_the_same_arms_as_the_page() {
2921 2 : let w = vec!["tenant_id = $1".to_owned(), "entity ?| $4".to_owned()];
2922 2 : assert_eq!(
2923 2 : count_sql(&w, None),
2924 : "SELECT (SELECT count(*) FROM entities WHERE tenant_id = $1 AND entity ?| $4)"
2925 : );
2926 2 : let split = TypeSplit {
2927 2 : firsts: vec![2],
2928 2 : named: Some(3),
2929 2 : selection: "(types @> $5)".to_owned(),
2930 2 : exact_first: true,
2931 2 : };
2932 2 : let sql = count_sql(&w, Some(&split));
2933 2 : assert_eq!(
2934 : sql,
2935 : "SELECT (SELECT count(*) FROM entities WHERE tenant_id = $1 AND entity ?| $4 \
2936 : AND types[1] = $2) + (SELECT count(*) FROM entities WHERE tenant_id = $1 \
2937 : AND entity ?| $4 AND cardinality(types) > 1 AND types[1] <> ALL($3) \
2938 : AND (types @> $5))"
2939 : );
2940 2 : }
2941 :
2942 : /// 6.3.10: without count=true no total is owed, so the statement carries
2943 : /// no window count (which forced a walk of the whole match set) and
2944 : /// fetches one row past the page to learn whether a next page exists.
2945 : #[test]
2946 2 : fn an_uncounted_page_fetches_one_extra_row_and_no_window_total() {
2947 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2948 2 : let p = Page {
2949 2 : offset: 20,
2950 2 : limit: 10,
2951 2 : count: false,
2952 2 : };
2953 2 : let (sql, paged) = query_sql("entity", &wheres(), None, Some(&p), true, "id", &mut binds);
2954 2 : assert!(paged);
2955 2 : assert!(!sql.contains("count(*) OVER ()"), "{sql}");
2956 2 : assert!(sql.contains("LIMIT $2 OFFSET $3"), "{sql}");
2957 2 : assert!(matches!(binds[1], Bind::Int(11)), "limit + 1 must be bound");
2958 2 : assert!(matches!(binds[2], Bind::Int(20)));
2959 2 : }
2960 :
2961 : /// The largest page a caller can build still has to produce a usable
2962 : /// LIMIT. `types_attrs.rs` clamps its discovery scan to `i64::MAX` on
2963 : /// purpose so the cast is safe; the uncounted path then binds
2964 : /// `limit + 1`, which is where that clamp stops being enough. The
2965 : /// addition overflows: a panic in a debug build, and in release a wrap
2966 : /// to `i64::MIN`, which Postgres refuses as a negative LIMIT.
2967 : #[test]
2968 2 : fn the_largest_page_still_binds_a_usable_limit() {
2969 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2970 2 : let p = Page {
2971 2 : offset: 0,
2972 2 : limit: i64::MAX,
2973 2 : count: false,
2974 2 : };
2975 2 : let (sql, paged) = query_sql("entity", &wheres(), None, Some(&p), true, "id", &mut binds);
2976 2 : assert!(paged);
2977 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
2978 2 : assert!(
2979 2 : matches!(binds[1], Bind::Int(i64::MAX)),
2980 : "the bound limit must stay positive"
2981 : );
2982 2 : }
2983 :
2984 : /// The undecided path must never leave the statement unbounded: SQL only
2985 : /// narrowed the set (`scopeQ`, `georel`, a `q=` the compiler declined),
2986 : /// so the rows come back to be filtered — under a LIMIT, and without an
2987 : /// OFFSET, which would page over the wrong set.
2988 : #[test]
2989 2 : fn an_undecided_query_carries_a_safety_limit_and_no_offset() {
2990 2 : let mut binds = vec![Bind::Text("t".to_owned())];
2991 2 : let (sql, paged) = query_sql(
2992 2 : "entity",
2993 2 : &wheres(),
2994 2 : None,
2995 2 : Some(&page(20, 10)),
2996 2 : false,
2997 2 : "id",
2998 2 : &mut binds,
2999 2 : );
3000 2 : assert!(!paged);
3001 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
3002 2 : assert!(!sql.contains("OFFSET"), "offset cannot be pushed: {sql}");
3003 2 : assert!(!sql.contains("count(*) OVER ()"), "{sql}");
3004 : // the candidate ceiling, not the page: the caller's evaluator still
3005 : // filters these rows, so a page-sized fetch would refuse legitimate
3006 : // queries whenever candidates outnumber matches
3007 2 : assert!(matches!(binds[1], Bind::Int(MAX_UNDECIDED_ROWS)));
3008 2 : }
3009 :
3010 : /// A caller with no page (idPattern, federation, orderBy) hands the store
3011 : /// no bound at all — the ceiling is then the only one.
3012 : #[test]
3013 2 : fn a_query_without_a_page_falls_back_to_the_ceiling() {
3014 2 : let mut binds = vec![Bind::Text("t".to_owned())];
3015 2 : let (sql, paged) = query_sql("entity", &wheres(), None, None, false, "id", &mut binds);
3016 2 : assert!(!paged);
3017 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
3018 2 : assert!(matches!(binds[1], Bind::Int(MAX_UNDECIDED_ROWS)));
3019 : // an exactly decided query with no page is materialized whole too
3020 2 : let mut binds = vec![Bind::Text("t".to_owned())];
3021 2 : let (sql, _) = query_sql("entity", &wheres(), None, None, true, "id", &mut binds);
3022 2 : assert!(sql.contains("LIMIT $2"), "{sql}");
3023 2 : }
3024 :
3025 : /// 5.5.6 / Table 6.3.2-1: reaching the ceiling means the statement was
3026 : /// truncated, so the page built from it would silently under-report —
3027 : /// the operation is refused with TooManyResults (403) instead.
3028 : #[test]
3029 2 : fn a_ceiling_hit_is_too_many_results_not_a_short_page() {
3030 2 : assert!(check_ceiling(false, 30, 31).is_ok(), "under the ceiling");
3031 2 : let err = check_ceiling(false, 31, 31).expect_err("ceiling reached");
3032 2 : let ngsi = ngsi_error(&err).expect("the spec error travels in the driver error");
3033 2 : assert_eq!(ngsi.kind(), "TooManyResults");
3034 2 : assert_eq!(ngsi.status(), 403);
3035 : // a pushed page is exact by construction: a full page is a page, not
3036 : // a truncation
3037 2 : assert!(check_ceiling(true, 1000, 1000).is_ok());
3038 2 : }
3039 :
3040 : /// A real driver failure must stay a driver failure — only the store's
3041 : /// own spec errors are recovered, or every 500 would become a 403.
3042 : #[test]
3043 2 : fn a_driver_error_is_not_mistaken_for_a_spec_error() {
3044 2 : assert!(ngsi_error(&sqlx::Error::RowNotFound).is_none());
3045 2 : assert!(ngsi_error(&sqlx::Error::PoolTimedOut).is_none());
3046 2 : assert!(ngsi_error(&sqlx::Error::Configuration("boom".into())).is_none());
3047 2 : }
3048 : }
|
