Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Candidate selection and full evaluation of a change against subscriptions: attribute diff, linked-entity and view projection, the per-subscription @context.
3 :
4 : use super::*;
5 :
6 : /// Where the matcher reads candidates from: the indexed mirror (both bus
7 : /// modes wire one), with the store scan only as the never-wired
8 : /// fallback so a missing mirror degrades to correct-but-slow.
9 140905 : pub(super) async fn subs_for(
10 140905 : st: &AppState,
11 140905 : tenant: &TenantId,
12 140905 : types: &[&str],
13 140905 : changed: &[&str],
14 140905 : queued: u64,
15 140905 : ) -> Vec<Arc<Value>> {
16 140905 : match &st.sub_mirror {
17 140903 : Some(m) => m.candidates_before(tenant.as_str(), types, changed, queued),
18 : // The scan reads subscriptions as they are now, so a change
19 : // drained late may reach one made after it; only the never-wired path.
20 : // The scan is the fallback, so its own failure may not be read as
21 : // "this tenant has no subscriptions" either: that is the silence
22 : // the mirror seed refuses to install.
23 2 : None => match st.store.list(tenant, Kind::Subscription).await {
24 2 : Ok(subs) => subs.into_iter().map(Arc::new).collect(),
25 0 : Err(e) => {
26 0 : tracing::error!(
27 : "subscription scan failed for tenant {}: {e}; no candidate matched this change",
28 0 : tenant.as_str()
29 : );
30 0 : Vec::new()
31 : }
32 : },
33 : }
34 140905 : }
35 :
36 : /// Whether two attribute instances are the same once the volatile members
37 : /// — the timestamps the broker itself stamps, and the instance identity —
38 : /// are set aside. Compared in place: this runs per changed attribute per
39 : /// matching subscription per event, and materializing two stripped copies of
40 : /// every JSON tree to throw them away one comparison later is the cost that
41 : /// multiplies by.
42 441006 : pub(super) fn stable_eq(a: &Value, b: &Value) -> bool {
43 2094708 : fn volatile(k: &str) -> bool {
44 2094708 : matches!(k, "createdAt" | "modifiedAt" | "instanceId")
45 2094708 : }
46 441006 : match (a, b) {
47 161636 : (Value::Object(oa), Value::Object(ob)) => {
48 1602116 : let count = |o: &Map<String, Value>| o.keys().filter(|k| !volatile(k)).count();
49 : // by lookup rather than by zipped iteration: which order a
50 : // serde_json Map yields is a cargo feature away from changing,
51 : // and features are additive
52 161636 : count(oa) == count(ob)
53 161440 : && oa
54 161440 : .iter()
55 492592 : .filter(|(k, _)| !volatile(k))
56 172268 : .all(|(k, va)| ob.get(k).is_some_and(|vb| stable_eq(va, vb)))
57 : }
58 108470 : (Value::Array(aa), Value::Array(ab)) => {
59 109706 : aa.len() == ab.len() && aa.iter().zip(ab).all(|(x, y)| stable_eq(x, y))
60 : }
61 170900 : _ => a == b,
62 : }
63 441006 : }
64 :
65 556858 : pub(super) fn attr_keys(doc: &Value) -> Vec<String> {
66 556858 : doc.as_object()
67 556858 : .map(|o| {
68 556858 : o.keys()
69 1257535 : .filter(|k| !crate::repr::ENTITY_META.contains(&k.as_str()))
70 556858 : .cloned()
71 556858 : .collect()
72 556858 : })
73 556858 : .unwrap_or_default()
74 556858 : }
75 :
76 : /// Per-attribute change classification between two internal docs.
77 278423 : pub(super) fn diff(before: Option<&Value>, after: Option<&Value>) -> Vec<(String, ChangeClass)> {
78 278423 : let empty = Value::Object(Map::new());
79 278423 : let b = before.unwrap_or(&empty);
80 278423 : let a = after.unwrap_or(&empty);
81 278423 : let mut keys = attr_keys(b);
82 364846 : for k in attr_keys(a) {
83 364846 : if !keys.contains(&k) {
84 257456 : keys.push(k);
85 257456 : }
86 : }
87 278423 : let mut out = Vec::new();
88 372664 : for k in keys {
89 372664 : let bv = b.get(&k);
90 372664 : let av = a.get(&k);
91 372664 : match (bv, av) {
92 257456 : (None, Some(_)) => out.push((k, ChangeClass::Created)),
93 7818 : (Some(_), None) => out.push((k, ChangeClass::Deleted)),
94 107390 : (Some(x), Some(y)) => {
95 : // instance-level deletion (by datasetId) counts as
96 : // attributeDeleted even when other instances survive (5.8.6)
97 107390 : let bx: Vec<&Value> = x.as_array().map(|a| a.iter().collect()).unwrap_or_default();
98 107390 : let by: Vec<&Value> = y.as_array().map(|a| a.iter().collect()).unwrap_or_default();
99 107390 : let removed = bx
100 107390 : .iter()
101 109174 : .any(|bi| !by.iter().any(|ai| instance_ds(ai) == instance_ds(bi)));
102 107390 : if removed {
103 252 : out.push((k.clone(), ChangeClass::Deleted));
104 284 : let survivors_changed = by.iter().any(|ai| {
105 464 : match bx.iter().find(|bi| instance_ds(bi) == instance_ds(ai)) {
106 284 : Some(bi) => !stable_eq(bi, ai),
107 0 : None => true,
108 : }
109 284 : });
110 252 : if survivors_changed {
111 4 : out.push((k, ChangeClass::Updated));
112 248 : }
113 107138 : } else if !stable_eq(x, y) {
114 103066 : out.push((k, ChangeClass::Updated));
115 104510 : }
116 : }
117 0 : _ => {}
118 : }
119 : }
120 278423 : out
121 278423 : }
122 :
123 : /// One string-valued member of a stored Subscription, absent or of another
124 : /// JSON type reading the same as absent.
125 11111 : pub(super) fn sub_str<'a>(sub: &'a Value, key: &str) -> Option<&'a str> {
126 11111 : sub.get(key).and_then(Value::as_str)
127 11111 : }
128 :
129 : /// 4.9 EXAMPLE 13/14: linked-entity q terms (`attr{path}`) resolve through
130 : /// the local store, same tenant. The evaluator is synchronous and the store
131 : /// is not, so `eval` runs against a cache: a URI the cache misses is
132 : /// recorded, reads as absent for that pass, and is fetched before the next
133 : /// one. The first pass that misses nothing is the answer. Every miss of a
134 : /// pass is fetched together, so the pass count is the depth of the link
135 : /// chain rather than the number of entities it touches.
136 : ///
137 : /// `narrow` is the caller's narrowing (ADR-0020) with the @context its
138 : /// policy q is read against: a hop into an Entity it does not admit reads
139 : /// as a hop into a missing one, and a resolved target loses the Attributes
140 : /// its projection hides before the evaluator sees it.
141 9674 : pub(crate) async fn linked_eval<F>(
142 9674 : st: &AppState,
143 9674 : tenant: &TenantId,
144 9674 : narrow: Option<(&crate::repr::Repr, &Context)>,
145 9674 : mut eval: F,
146 9674 : ) -> bool
147 9674 : where
148 9674 : F: for<'a> FnMut(antares_ql::eval::EntityLookup<'a>) -> bool,
149 9674 : {
150 9674 : let mut cache: HashMap<String, Option<Value>> = HashMap::new();
151 : // The evaluator's own lookup budget bounds how many distinct
152 : // URIs one expression can name, so it bounds the passes too; a chain
153 : // deeper than that reads as unresolved, which is what exhausting the
154 : // budget inside the evaluator already does.
155 9674 : for _ in 0..antares_ql::eval::MAX_Q_LINK_LOOKUPS {
156 9726 : let missed = std::cell::RefCell::new(Vec::new());
157 9726 : let out = eval(&|uri: &str| match cache.get(uri) {
158 52 : Some(doc) => doc.clone(),
159 : None => {
160 52 : missed.borrow_mut().push(uri.to_owned());
161 52 : None
162 : }
163 104 : });
164 9726 : let missed = missed.into_inner();
165 9726 : if missed.is_empty() {
166 9674 : return out;
167 52 : }
168 52 : for uri in missed {
169 52 : let mut doc = st
170 52 : .store
171 52 : .get(tenant, Kind::Entity, &uri)
172 52 : .await
173 52 : .ok()
174 52 : .flatten();
175 52 : if let (Some(d), Some((r, ctx))) = (&doc, narrow) {
176 : // boxed: the admission check evaluates the policy's own q,
177 : // which comes back here
178 28 : let admitted = Box::pin(crate::repr::policy_admits(
179 28 : st,
180 28 : tenant,
181 28 : ctx,
182 28 : r.policy_reach.as_deref(),
183 28 : d,
184 28 : ))
185 28 : .await;
186 28 : doc = admitted.then(|| crate::repr::policy_view(d, r));
187 24 : }
188 52 : cache.insert(uri, doc);
189 : }
190 : }
191 0 : eval(&|uri: &str| cache.get(uri).cloned().flatten())
192 9674 : }
193 :
194 : /// What each Entity of a notification's `data` (and each one it embeds)
195 : /// shows under the expanded `rules` (every Attribute without rules), read
196 : /// on its stored state now, names compacted the way `data` is. An Entity the store no longer holds, that
197 : /// `reach` does not admit, or that no rule holds for, maps to `None`.
198 4 : pub(super) async fn shown_now(
199 4 : st: &AppState,
200 4 : tenant: &TenantId,
201 4 : rules: Option<&crate::policy::Filter>,
202 4 : reach: Option<&crate::repr::PolicyReach>,
203 4 : data: &Value,
204 4 : ctx: &Context,
205 4 : ) -> HashMap<String, Option<crate::policy::Shown<'static>>> {
206 78 : fn ids<'a>(v: &'a Value, out: &mut std::collections::BTreeSet<&'a str>) {
207 78 : match v {
208 20 : Value::Object(o) => {
209 20 : if let (Some(id), Some(_)) = (o.get("id").and_then(Value::as_str), o.get("type")) {
210 4 : out.insert(id);
211 16 : }
212 62 : o.values().for_each(|v| ids(v, out));
213 : }
214 12 : Value::Array(a) => a.iter().for_each(|v| ids(v, out)),
215 50 : _ => {}
216 : }
217 78 : }
218 4 : let mut wanted = std::collections::BTreeSet::new();
219 4 : ids(data, &mut wanted);
220 : // one read for every Entity of the notification, shared with every
221 : // other delivery of the tenant waiting for a read: one by one, a
222 : // folded notification of 256 Entities was 256 round trips under a
223 : // delivery slot, and a fan-out to 1 000 subscriptions of one
224 : // view read the same Entities 1 000 times. One failed read hides every
225 : // Entity of the notification; the failure is logged, not silent.
226 4 : let ids: Vec<String> = wanted.iter().map(|id| (*id).to_owned()).collect();
227 4 : let stored_now = super::coalesce::read_entities(st, tenant, ids).await;
228 4 : let mut out = HashMap::new();
229 4 : for id in wanted {
230 4 : let mut stored = stored_now.as_ref().and_then(|m| m.get(id));
231 4 : if let (Some(d), Some(_)) = (stored, reach) {
232 0 : let ctx = crate::repr::policy_context();
233 0 : if !crate::repr::policy_admits(st, tenant, ctx, reach, d).await {
234 0 : stored = None;
235 0 : }
236 4 : }
237 4 : let shown = stored
238 4 : .and_then(|d| crate::repr::shown(rules, d))
239 4 : .map(|s| s.map_names(|n| ctx.compact_iri(n)));
240 4 : out.insert(id.to_owned(), shown);
241 : }
242 4 : out
243 4 : }
244 :
245 : /// The entry of `shown` for the Entity `e`; `None` when `e` is no Entity.
246 8 : pub(super) fn entity_shown<'a>(
247 8 : shown: &'a HashMap<String, Option<crate::policy::Shown<'static>>>,
248 8 : e: &Value,
249 8 : ) -> Option<Option<&'a crate::policy::Shown<'static>>> {
250 8 : let id = e.get("id").and_then(Value::as_str)?;
251 8 : Some(shown.get(id).and_then(Option::as_ref))
252 8 : }
253 :
254 : /// Strip a compacted notified Entity to what [`shown_now`] says it shows,
255 : /// and every Entity it embeds the same way: an embedded one no rule holds
256 : /// for is not embedded, its Relationship stays.
257 4 : pub(super) fn project_shown(
258 4 : shown: &HashMap<String, Option<crate::policy::Shown<'static>>>,
259 4 : e: &mut Value,
260 4 : key_values: bool,
261 4 : ) {
262 4 : let allowed = entity_shown(shown, e).flatten().cloned();
263 4 : let Some(obj) = e.as_object_mut() else {
264 0 : return;
265 : };
266 4 : let allowed = allowed.unwrap_or_else(crate::policy::Shown::nothing);
267 20 : obj.retain(|k, _| k == "@context" || antares_model::is_meta(k) || allowed.allows(k));
268 12 : for v in obj.values_mut() {
269 12 : let instances: Vec<&mut Value> = match v {
270 0 : Value::Array(a) => a.iter_mut().collect(),
271 12 : other => vec![other],
272 : };
273 12 : for inst in instances {
274 12 : if key_values {
275 0 : if inst.get("id").is_some() && inst.get("type").is_some() {
276 0 : match entity_shown(shown, inst) {
277 0 : Some(Some(_)) => project_shown(shown, inst, key_values),
278 0 : _ => {
279 0 : let id = inst.get("id").cloned().unwrap_or(Value::Null);
280 0 : *inst = id;
281 0 : }
282 : }
283 0 : }
284 0 : continue;
285 12 : }
286 12 : let Some(o) = inst.as_object_mut() else {
287 8 : continue;
288 : };
289 4 : if let Some(j) = o.get("entity") {
290 0 : if matches!(entity_shown(shown, j), Some(Some(_))) {
291 0 : if let Some(j) = o.get_mut("entity") {
292 0 : project_shown(shown, j, key_values);
293 0 : }
294 0 : } else {
295 0 : o.remove("entity");
296 0 : }
297 4 : }
298 4 : if let Some(Value::Array(a)) = o.get_mut("entityList") {
299 0 : a.retain(|j| matches!(entity_shown(shown, j), Some(Some(_))));
300 0 : for j in a.iter_mut() {
301 0 : project_shown(shown, j, key_values);
302 0 : }
303 4 : }
304 : }
305 : }
306 4 : }
307 :
308 : /// ADR-0020: a narrowing reaches every document the answer carries, so the
309 : /// projection runs on each Entity `join=inline` embedded (4.5.23.2): under
310 : /// `entity`/`entityList` normalized, in place of the object URI simplified.
311 18 : pub(super) fn project_joined(f: &crate::policy::Filter, e: &mut Value, key_values: bool) {
312 18 : f.project(e);
313 18 : let Some(obj) = e.as_object_mut() else {
314 0 : return;
315 : };
316 62 : for v in obj.values_mut() {
317 62 : let instances: Vec<&mut Value> = match v {
318 0 : Value::Array(a) => a.iter_mut().collect(),
319 62 : other => vec![other],
320 : };
321 62 : for inst in instances {
322 62 : if key_values {
323 0 : if inst.get("id").is_some() && inst.get("type").is_some() {
324 0 : project_joined(f, inst, key_values);
325 0 : }
326 0 : continue;
327 62 : }
328 62 : let Some(o) = inst.as_object_mut() else {
329 36 : continue;
330 : };
331 52 : for m in ["entity", "entityList"] {
332 52 : match o.get_mut(m) {
333 0 : Some(Value::Array(a)) => {
334 0 : for j in a {
335 0 : project_joined(f, j, key_values);
336 0 : }
337 : }
338 4 : Some(j @ Value::Object(_)) => project_joined(f, j, key_values),
339 48 : _ => {}
340 : }
341 : }
342 : }
343 : }
344 18 : }
345 :
346 : /// Whether the Subscription's q follows a Relationship (4.9
347 : /// LinkedEntityRelation).
348 16 : pub(super) fn linked_condition(sub: &Value) -> bool {
349 16 : sub_str(sub, "q")
350 16 : .is_some_and(|q| antares_ql::regex::q_node(q).is_some_and(|n| n.has_linked_paths()))
351 16 : }
352 :
353 : /// The @context governing a subscription's notifications (5.8.6): the
354 : /// jsonldContext member if set, else the @context of the creating request.
355 6800 : pub(crate) async fn sub_context(st: &AppState, tenant: &TenantId, sub: &Value) -> Arc<Context> {
356 : // Borrowed: the resolver only reads it, and this is on the per-candidate
357 : // path, where cloning an inline @context copied the whole document.
358 6800 : let source = sub.get("jsonldContext").or_else(|| sub.get("__context"));
359 6340 : match source {
360 : // 5.5.10: the Subscription belongs to one Tenant, so the @context it
361 : // names resolves within that Tenant — a Hosted @context another Tenant
362 : // stored (5.13.1) is not in scope here and falls back to the core
363 : // context rather than compacting this Notification against it.
364 6340 : Some(v) if !v.is_null() => st
365 6340 : .loader
366 6340 : .resolve_quiet_for(tenant, v)
367 6340 : .await
368 6340 : .unwrap_or_else(|_| st.loader.core()),
369 460 : _ => st.loader.core(),
370 : }
371 6800 : }
372 :
373 : /// A resolved `@context` per Tenant, then per the `@context` URL the
374 : /// Subscription names. Keyed by Tenant first: 5.5.10 scopes a resolution to
375 : /// one Tenant, so one Tenant's resolved context is never reachable under
376 : /// another's identifier even where both name the same URL.
377 : pub(super) type CtxMemo = HashMap<String, HashMap<String, Arc<Context>>>;
378 :
379 : /// `sub_context` over one drain's memo. Resolving is a cache lookup per call,
380 : /// and this runs once per candidate subscription per change: a drain of
381 : /// `CHANGE_BATCH` changes resolves the same handful of `@context` URLs
382 : /// thousands of times over. The memo lives for the one drain that owns it, so
383 : /// an `@context` a request replaces or deletes meanwhile is picked up by the
384 : /// next drain. A Subscription naming an inline `@context` rather than a URL
385 : /// resolves as before — it has no cheap identity to key on.
386 9078 : pub(super) async fn sub_context_memo(
387 9078 : st: &AppState,
388 9078 : tenant: &TenantId,
389 9078 : tenant_str: &str,
390 9078 : sub: &Value,
391 9078 : memo: &mut CtxMemo,
392 9078 : ) -> Arc<Context> {
393 9078 : let url = sub
394 9078 : .get("jsonldContext")
395 9078 : .or_else(|| sub.get("__context"))
396 9078 : .and_then(Value::as_str);
397 9078 : let Some(url) = url else {
398 264 : return sub_context(st, tenant, sub).await;
399 : };
400 8814 : if let Some(hit) = memo.get(tenant_str).and_then(|by_url| by_url.get(url)) {
401 4126 : return Arc::clone(hit);
402 4688 : }
403 4688 : let ctx = sub_context(st, tenant, sub).await;
404 4688 : memo.entry(tenant_str.to_owned())
405 4688 : .or_default()
406 4688 : .insert(url.to_owned(), Arc::clone(&ctx));
407 4688 : ctx
408 9078 : }
409 :
410 : /// The notification triggers of one subscription, in the form the matcher
411 : /// compares against (Table 5.2.12-1): absent means the default combination
412 : /// `"attributeCreated"` + `"attributeUpdated"`, and `"entityUpdated"` "is
413 : /// equivalent to the combination `"attributeCreated"`, `"attributeUpdated"`
414 : /// and `"attributeDeleted"`" — so it is expanded here, at the single point
415 : /// the list is read, rather than at each comparison.
416 9104 : pub(super) fn triggers_of(sub: &Value) -> Vec<&str> {
417 : // Borrowed from the Subscription, not copied out of it: this runs once
418 : // per candidate per change, and the default arm used to allocate a
419 : // String for each of DEFAULT_TRIGGERS, which are already 'static.
420 9104 : let mut triggers: Vec<&str> = sub
421 9104 : .get("notificationTrigger")
422 9104 : .and_then(Value::as_array)
423 9104 : .map(|a| a.iter().filter_map(Value::as_str).collect())
424 9104 : .unwrap_or_else(|| DEFAULT_TRIGGERS.to_vec());
425 9104 : if triggers.contains(&"entityUpdated") {
426 36 : for t in DEFAULT_TRIGGERS.iter().copied().chain(["attributeDeleted"]) {
427 36 : if !triggers.contains(&t) {
428 24 : triggers.push(t);
429 24 : }
430 : }
431 9092 : }
432 9104 : triggers
433 9104 : }
434 :
435 140905 : pub(super) async fn matches_for(
436 140905 : st: &AppState,
437 140905 : tenant_str: &str,
438 140905 : before: Option<Value>,
439 140905 : after: Option<Value>,
440 140905 : queued: u64,
441 140905 : ctx_memo: &mut CtxMemo,
442 140905 : ) -> Vec<Matched> {
443 140905 : let mut out = Vec::new();
444 : // The tenant of the write that fired the change hook — this broker's own
445 : // value, and a write inside a Snapshot (5.5.15) carries the synthetic one.
446 140905 : let Ok(tenant) = TenantId::new_internal(tenant_str) else {
447 0 : return out;
448 : };
449 140905 : let changes = diff(before.as_ref(), after.as_ref());
450 140905 : let entity_trigger = match (&before, &after) {
451 123244 : (None, Some(_)) => "entityCreated",
452 3371 : (Some(_), None) => "entityDeleted",
453 14290 : _ => "entityUpdated",
454 : };
455 140905 : let eval_doc = after.as_ref().or(before.as_ref());
456 140905 : let Some(eval_doc) = eval_doc else { return out };
457 : // Candidate lookup by the entity's types and the changed attribute
458 : // IRIs — no linear scan over all subscriptions.
459 140905 : let types: Vec<&str> = eval_doc
460 140905 : .get("type")
461 140905 : .and_then(Value::as_array)
462 140905 : .map(|a| a.iter().filter_map(Value::as_str).collect())
463 140905 : .unwrap_or_default();
464 187932 : let changed_keys: Vec<&str> = changes.iter().map(|(k, _)| k.as_str()).collect();
465 140905 : let subs = subs_for(st, &tenant, &types, &changed_keys, queued).await;
466 : // one instant for the change, so equal shapes render one payload
467 140905 : let now = now_iso();
468 140905 : let mut built: HashMap<String, Vec<Value>> = HashMap::new();
469 140905 : for sub in subs {
470 9108 : if !is_active(&sub) || sub.get("timeInterval").is_some() {
471 24 : continue;
472 9084 : }
473 9084 : let narrow = None;
474 9084 : let triggers = triggers_of(&sub);
475 : // which attribute-level changes this sub cares about
476 9084 : let watched: Option<Vec<&str>> = sub
477 9084 : .get("watchedAttributes")
478 9084 : .and_then(Value::as_array)
479 9084 : .map(|a| a.iter().filter_map(Value::as_str).collect());
480 : #[expect(unused_mut)]
481 9084 : let mut relevant: Vec<&(String, ChangeClass)> = changes
482 9084 : .iter()
483 9224 : .filter(|(k, _)| watched.as_ref().is_none_or(|w| w.contains(&k.as_str())))
484 9084 : .collect();
485 9146 : let attr_trigger_fired = relevant.iter().any(|(_, c)| {
486 9146 : let t = match c {
487 8856 : ChangeClass::Created => "attributeCreated",
488 142 : ChangeClass::Updated => "attributeUpdated",
489 148 : ChangeClass::Deleted => "attributeDeleted",
490 : };
491 9146 : triggers.contains(&t)
492 9146 : });
493 9084 : let entity_trigger_fired =
494 9084 : triggers.contains(&entity_trigger) && (watched.is_none() || !relevant.is_empty());
495 9084 : if !attr_trigger_fired && !entity_trigger_fired {
496 22 : continue;
497 9062 : }
498 9062 : let ctx = sub_context_memo(st, &tenant, tenant_str, &sub, ctx_memo).await;
499 9062 : if !selector_match(&sub, eval_doc, &ctx) {
500 4 : continue;
501 9058 : }
502 9074 : if !linked_eval(st, &tenant, narrow, |l| {
503 9074 : conditions_match(&sub, eval_doc, &ctx, l)
504 9074 : })
505 9058 : .await
506 : {
507 14 : continue;
508 9044 : }
509 9044 : if throttled(&sub) {
510 6 : continue;
511 9038 : }
512 9038 : let deleted: Vec<String> = if triggers.contains(&"attributeDeleted") {
513 100 : relevant
514 100 : .iter()
515 100 : .filter(|(_, c)| *c == ChangeClass::Deleted)
516 100 : .map(|(k, _)| k.clone())
517 100 : .collect()
518 : } else {
519 8938 : Vec::new()
520 : };
521 9038 : let entity_deleted_fired = after.is_none() && triggers.contains(&"entityDeleted");
522 9038 : let key = payload_key(&sub, &ctx, &deleted, entity_deleted_fired);
523 9038 : let data = match built.get(&key) {
524 4338 : Some(data) => data.clone(),
525 : None => {
526 4700 : let data = build_data(
527 4700 : st,
528 4700 : &tenant,
529 4700 : &sub,
530 4700 : &ctx,
531 4700 : before.as_ref(),
532 4700 : after.as_ref(),
533 4700 : &deleted,
534 4700 : entity_deleted_fired,
535 4700 : &now,
536 4700 : )
537 4700 : .await;
538 4700 : built.insert(key, data.clone());
539 4700 : data
540 : }
541 : };
542 9038 : out.push(Matched {
543 9038 : tenant: tenant.clone(),
544 9038 : sub,
545 9038 : ctx,
546 9038 : data,
547 9038 : });
548 : }
549 140905 : out
550 140905 : }
551 :
552 : /// The exact Entity ids a subscription's `entities` selector pins down, or
553 : /// `None` when it leaves any of them open. Table 5.2.33-1: `id` is a String
554 : /// or a String[] and "id takes precedence over idPattern", so an entry
555 : /// carrying an id constrains the read exactly — while one entry without an id
556 : /// (a bare type, or an idPattern no store column can answer) admits every id
557 : /// and forfeits the narrowing for the whole OR-ed selector.
558 204 : pub(super) fn selector_ids(sub: &Value) -> Option<Vec<String>> {
559 204 : let sel = sub.get("entities").and_then(Value::as_array)?;
560 192 : let mut ids = Vec::new();
561 200 : for e in sel {
562 200 : match e.get("id") {
563 40 : Some(Value::String(i)) => ids.push(i.clone()),
564 8 : Some(Value::Array(a)) => {
565 16 : for v in a {
566 16 : ids.push(v.as_str()?.to_owned());
567 : }
568 : }
569 152 : _ => return None,
570 : }
571 : }
572 40 : (!ids.is_empty()).then_some(ids)
573 204 : }
574 :
575 : /// A store read on a delivery path has no caller to fail: the sweep is a
576 : /// timer, the fan-out is spawned, and both answer `()`. Silence is what makes
577 : /// a failure dangerous here — a subscription that stops firing because the
578 : /// store could not be read looks exactly like one with nothing to send — so
579 : /// the failure is named and the path continues on the empty set it would
580 : /// have continued on anyway.
581 : /// Every document of `kind` in `tenant` that `keep` accepts, read a page
582 : /// at a time: the whole-tenant `list` carries the client-query
583 : /// row ceiling (5.5.6), and past it a notification path would read nothing
584 : /// and send an empty or no notification. Only the kept documents are held.
585 21941 : pub(crate) async fn docs_where(
586 21941 : st: &AppState,
587 21941 : tenant: &TenantId,
588 21941 : kind: Kind,
589 21941 : keep: impl Fn(&Value) -> bool,
590 21941 : ) -> Result<Vec<Value>, antares_model::NgsiError> {
591 21941 : let mut out = Vec::new();
592 24502 : crate::csource::walk_docs(st, tenant, kind, |doc| {
593 20785 : if keep(&doc) {
594 10661 : out.push(doc);
595 10661 : }
596 20785 : Ok(())
597 20785 : })
598 21941 : .await?;
599 21941 : Ok(out)
600 21941 : }
601 :
602 22524 : pub(super) fn read_or_warn<T>(res: Result<Vec<T>, antares_model::NgsiError>, what: &str) -> Vec<T> {
603 22524 : res.unwrap_or_else(|e| {
604 0 : tracing::warn!("notification path: reading {what} failed: {e}");
605 0 : Vec::new()
606 0 : })
607 22524 : }
|
