Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Subscription matching + HTTP notification delivery (5.8.6, 5.3.1).
3 : //!
4 : //! Change detection: the store's change hook feeds every entity write here as
5 : //! a (before, after) pair; attribute-level changes are derived by diffing —
6 : //! one hook point instead of one call per write handler.
7 : //!
8 : //! Candidate lookup is index-shaped. `SubMirror` keeps inverted
9 : //! (tenant, type) and (tenant, watched-attr) maps next to the docs, so one
10 : //! change evaluates only the subscriptions that could possibly fire — never
11 : //! a scan over all of a tenant's subscriptions. Subscriptions the index
12 : //! cannot classify exactly (4.17 type-selection expressions) fall into a
13 : //! `broad` bucket that is always evaluated: the index may over-select,
14 : //! never under-select. Full evaluation (selector/q/geo/scope/triggers)
15 : //! stays the truth for every candidate.
16 :
17 : mod coalesce;
18 : mod csource;
19 : mod delivery;
20 : pub(crate) use delivery::kv_pairs;
21 : mod interval;
22 : mod matching;
23 : mod payload;
24 : mod retry;
25 : #[cfg(test)]
26 : mod tests;
27 : mod throttle;
28 :
29 : use crate::mirror::{index_keys, Change, Keys, Mirror, SubMirror, CSUB_SWEEP_BACKSTOP_MS};
30 : use crate::negotiate::{inject_context, link_header_value};
31 : use crate::state::{now_iso, AppState};
32 : use antares_jsonld::Context;
33 : use antares_model::{dt_key, TenantId};
34 : use antares_store::CurrentStateDriverExt;
35 : use antares_store::Kind;
36 : use antares_store::{TemporalEvent, TemporalOp};
37 : pub use csource::*;
38 : pub(crate) use delivery::*;
39 : pub use interval::*;
40 : pub(crate) use matching::*;
41 : pub(crate) use payload::*;
42 : pub(crate) use retry::*;
43 : use serde_json::{json, Map, Value};
44 : use std::collections::HashMap;
45 : use std::sync::Arc;
46 : pub use throttle::*;
47 :
48 : /// `notification.endpoint.uri` of a Subscription.
49 10354 : pub(crate) fn endpoint_uri(sub: &serde_json::Map<String, serde_json::Value>) -> Option<&str> {
50 10354 : sub.get("notification")?
51 10354 : .get("endpoint")?
52 10354 : .get("uri")?
53 10354 : .as_str()
54 10354 : }
55 :
56 : const DEFAULT_TRIGGERS: &[&str] = &["attributeCreated", "attributeUpdated"];
57 : /// Depth of the change→matcher queue, the same ring size the local bus uses.
58 : /// Held with the other published ceilings so `/q/health` reports it.
59 : use crate::bounds::{CHANGE_QUEUE, DELIVERY_WIDTH};
60 : // the per-tenant slots of the native delivery pool (`tenant_slots`) and its lanes
61 : #[cfg(not(target_arch = "wasm32"))]
62 : use crate::bounds::{DELIVERY_LANE, DELIVERY_WIDTH_PER_TENANT};
63 : static CHANGES_DROPPED: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
64 : static TASK_PANICS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
65 : /// Entities whose newest change version the matcher keeps (two generations
66 : /// of this many): a hook overtakes another only among the writes in flight
67 : /// at once.
68 : const CHANGE_ORDER_KEYS: usize = 16_384;
69 :
70 : /// Hand a batch to the matcher queue: counted as pending on acceptance,
71 : /// counted as dropped when the queue is full. Not the durable outbox
72 : /// (`antares_sql::store::pg::outbox::enqueue`), which is a row inside the
73 : /// caller's transaction: this ring lives in the process and a full one
74 : /// drops, which is why the drop is counted.
75 140284 : fn queue_for_matching(
76 140284 : tx: &tokio::sync::mpsc::Sender<(u64, Vec<Change>)>,
77 140284 : pending: &std::sync::atomic::AtomicUsize,
78 140284 : changes: Vec<Change>,
79 140284 : ) {
80 : // stamped now, after the write: a subscription made later never matches it
81 140284 : if tx.try_send((crate::mirror::tick(), changes)).is_ok() {
82 140284 : pending.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
83 140284 : } else {
84 0 : note_drop();
85 0 : }
86 140284 : }
87 :
88 : /// Changes dropped because the matcher queue was full, since process start.
89 94 : pub(crate) fn changes_dropped() -> u64 {
90 94 : CHANGES_DROPPED.load(std::sync::atomic::Ordering::Relaxed)
91 94 : }
92 :
93 : /// Panics absorbed at the notification-task boundary, since process start.
94 86 : pub(crate) fn task_panics() -> u64 {
95 86 : TASK_PANICS.load(std::sync::atomic::Ordering::Relaxed)
96 86 : }
97 :
98 : /// Entity-level members that are NOT Attributes. Table 5.2.12-1 scopes
99 : /// watchedAttributes to "Properties or Relationships", so the entity's own
100 : /// system members (including 4.22 `expiresAt` and `deletedAt`) must never be
101 : /// diffed as attribute-level changes.
102 : #[derive(Clone, Copy, PartialEq, Debug)]
103 : enum ChangeClass {
104 : Created,
105 : Updated,
106 : Deleted,
107 : }
108 :
109 : /// Fill a mirror from the store, or say why it could not be filled.
110 : ///
111 : /// Every document of every tenant has to be in the mirror before it is
112 : /// installed. `CurrentStateDriver::subscription_tenants` states the rule for
113 : /// the data path — "A SUBSET is a silent outage: a tenant missing here never
114 : /// fires a periodic notification and never reaches the mirror" — and an
115 : /// error absorbed into an empty list is the same subset by another route.
116 : /// A connection failure at startup refuses it, so this is reachable rather
117 : /// than theoretical.
118 : ///
119 : /// Paged, not `list`: 5.5.6 licenses TooManyResults for "a query operation
120 : /// … producing so many results that can potentially exhaust client or
121 : /// server resources", and the seed is not one — it must see every
122 : /// document of every tenant or it is the silent outage above. Reading it
123 : /// through the ceiling `list` carries for client queries made one tenant's
124 : /// stored volume decide whether OTHER tenants are matched at all.
125 : ///
126 : /// One function for every mirror and both bus modes. `bus=local` seeds the
127 : /// subscription mirror here; `bus=nats` seeds a subscription mirror and a
128 : /// registration mirror, and re-seeds the registration one after a consumer
129 : /// gap. Those were a second copy of this walk that kept the ceiling and
130 : /// swallowed the error into an empty list, which is how a rule fixed in one
131 : /// place stayed broken in the others.
132 : ///
133 : /// The domain is `subscription_tenants`, whose contract covers every kind a
134 : /// mirror is built from. A tenant holding nothing of this kind costs one
135 : /// empty page.
136 2586 : pub async fn seed_mirror(
137 2586 : store: &dyn antares_store::CurrentStateDriver,
138 2586 : mirror: &dyn Mirror,
139 2586 : kind: Kind,
140 2586 : ) -> Result<(), antares_model::NgsiError> {
141 2588 : for tenant_str in store.subscription_tenants().await? {
142 : // The store's own enumeration, so the grammar is all that is left to
143 : // check: 5.5.15 permits a Subscription inside a Snapshot, and its
144 : // tenant is the synthetic one no client may name.
145 2243 : let tenant = TenantId::new_internal(&tenant_str)?;
146 2243 : let mut after: Option<String> = None;
147 : loop {
148 2563 : let page = store
149 2563 : .list_page(&tenant, kind, after.as_deref(), SEED_PAGE)
150 2563 : .await?;
151 2563 : let short = page.len() < SEED_PAGE;
152 2563 : let before = after.clone();
153 332932 : for doc in page {
154 332932 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
155 332932 : let id = id.to_owned();
156 332932 : after = Some(id.clone());
157 332932 : mirror.apply(&tenant_str, &id, Some(doc));
158 332932 : }
159 : }
160 : // A short page is the end. A cursor that did not move is also the
161 : // end, and it is the load-bearing half: only a document carrying
162 : // an `id` can advance it, so a full page without one would
163 : // otherwise re-read the same page forever. No write path stores
164 : // such a document — which is exactly why the loop may not depend
165 : // on that staying true.
166 2563 : if short || after == before {
167 2243 : break;
168 320 : }
169 : }
170 : }
171 2586 : Ok(())
172 2586 : }
173 :
174 : /// Documents per mirror-seed page: the peak transient allocation of the
175 : /// walk, paid once per tenant at startup.
176 : const SEED_PAGE: usize = 1_000;
177 :
178 : /// Install the subscription mirror, the store change hook that feeds the
179 : /// matcher, and the background tasks the pipeline needs. Called once at
180 : /// startup, by `crate::wire`.
181 1284 : pub(crate) async fn wire_matcher(state: &mut AppState) {
182 : // bus=local: the same indexed mirror the nats wiring builds, fed
183 : // synchronously by the CUD hook — the matcher never rescans the store.
184 1284 : let mirror = Arc::new(SubMirror::default());
185 1284 : match seed_mirror(state.store.as_ref(), mirror.as_ref(), Kind::Subscription).await {
186 1284 : Ok(()) => state.sub_mirror = Some(mirror.clone()),
187 : // Not installed, so `subs_for` takes the store scan it documents as
188 : // the missing-mirror fallback. Installing what the seed managed to
189 : // read would be the one outcome that is neither correct nor slow:
190 : // the matcher reads candidates from the mirror alone, so a
191 : // subscription absent from it never fires again in this process.
192 0 : Err(e) => tracing::error!(
193 : "subscription mirror seed failed ({e}); \
194 : matching falls back to a store scan per change"
195 : ),
196 : }
197 1284 : let m = mirror.clone();
198 1284 : state.sub_sync = Some(Arc::new(
199 1968 : move |tenant: &TenantId, kind: Kind, id: &str, doc: Option<&Value>| match kind {
200 452 : Kind::CSourceSubscription => m.csub_written(),
201 1516 : _ => m.apply(tenant.as_str(), id, doc.cloned()),
202 1968 : },
203 : ));
204 :
205 : // bus=local: the registration mirror the nats wiring builds, fed in
206 : // process by the registration CUD hook, so a federated query no longer
207 : // reads `csource_registrations` per request. Installed only when the
208 : // hydrate is whole, by the rule the subscription mirror follows above:
209 : // `reg_docs` reads a present mirror as the truth.
210 1284 : let regs = Arc::new(crate::mirror::DocMirror::default());
211 1284 : match seed_mirror(state.store.as_ref(), regs.as_ref(), Kind::Registration).await {
212 : Ok(()) => {
213 1284 : state.reg_mirror = Some(regs.clone());
214 2240 : state.reg_sync = Some(Arc::new(move |tenant: &TenantId, id: &str, doc| {
215 2206 : regs.apply(tenant.as_str(), id, doc.cloned());
216 2206 : }));
217 : }
218 0 : Err(e) => tracing::error!(
219 : "registration mirror hydrate failed ({e}); \
220 : federation matching falls back to a store read per request"
221 : ),
222 : }
223 :
224 : // The queue carries whole before+after payloads and is drained one
225 : // inline delivery at a time, so behind one slow subscriber an unbounded
226 : // queue grows until the process dies. Bounded instead: a full queue drops
227 : // the change and counts it.
228 1284 : let (tx, mut rx) = tokio::sync::mpsc::channel::<(u64, Vec<Change>)>(CHANGE_QUEUE);
229 1284 : let flush_tx = tx.clone();
230 1284 : let flush_pending = state.pending_changes.clone();
231 27254 : state.change_flush = Some(Arc::new(move |changes: Vec<Change>| {
232 27254 : queue_for_matching(&flush_tx, &flush_pending, changes)
233 27254 : }));
234 : // Temporal auto-recording runs SYNCHRONOUSLY on the hook (read-your-writes:
235 : // the ETSI suite queries history immediately after a write); the matcher
236 : // work is handed to the async task below. One choke point for every write.
237 1284 : let st_rec = state.clone();
238 1284 : let hook_pending = state.pending_changes.clone();
239 : // Two writers of one Entity may run their hooks in either order on the
240 : // Postgres arm (S3-18): the matcher drops a change older than one of the
241 : // same Entity it has taken, so no subscriber is handed the older image
242 : // last. Recording keeps every change, late or not: each carries the
243 : // instances only it wrote, and history is ordered by their instants.
244 1284 : let newest = Arc::new(antares_store::NewestSeen::new(CHANGE_ORDER_KEYS));
245 1284 : state.store.set_change_hook(Arc::new(
246 : move |tenant: &TenantId,
247 : before: Option<Value>,
248 : after: Option<Value>,
249 : version: antares_store::ChangeVersion|
250 140892 : -> antares_store::HookFuture<'_> {
251 140892 : let st_rec = st_rec.clone();
252 140892 : let tx = tx.clone();
253 140892 : let hook_pending = hook_pending.clone();
254 140892 : let newest = newest.clone();
255 140892 : Box::pin(async move {
256 140892 : record_temporal_change(&st_rec, tenant, before.as_ref(), after.as_ref()).await;
257 140892 : let id = after
258 140892 : .as_ref()
259 140892 : .or(before.as_ref())
260 140892 : .and_then(|d| d.get("id"))
261 140892 : .and_then(Value::as_str);
262 140892 : if id.is_some_and(|id| !newest.admit(tenant.as_str(), id, &version)) {
263 2 : return;
264 140890 : }
265 : // inside a request the change rides the request's buffer and
266 : // reaches the matcher with the rest of that request's changes
267 113030 : let Some(change) =
268 140890 : crate::history::buffer_change((tenant.as_str().to_owned(), before, after))
269 : else {
270 27860 : return;
271 : };
272 113030 : queue_for_matching(&tx, &hook_pending, vec![change]);
273 140892 : })
274 140892 : },
275 : ));
276 1284 : let st = state.clone();
277 1284 : let pending = state.pending_changes.clone();
278 1284 : crate::spawn_loop(async move {
279 90539 : while let Some((queued, first)) = rx.recv().await {
280 89900 : let mut batch: Vec<(u64, Change)> = first.into_iter().map(|c| (queued, c)).collect();
281 89296 : let mut taken: usize = 1;
282 : // everything already queued behind it rides the same pass
283 140277 : while batch.len() < CHANGE_BATCH {
284 140261 : match rx.try_recv() {
285 50981 : Ok((queued, c)) => {
286 50981 : batch.extend(c.into_iter().map(|c| (queued, c)));
287 50981 : taken = taken.saturating_add(1);
288 : }
289 89280 : Err(_) => break,
290 : }
291 : }
292 89296 : let st = st.clone();
293 : // The batch stays pending until its last group is delivered: the
294 : // groups leave on their subscriptions' lanes, after this drain.
295 89296 : let done = Arc::new(BatchDone {
296 89296 : pending: pending.clone(),
297 89296 : taken,
298 89296 : });
299 89296 : guarded(async move { process_queued(&st, batch, Some(done)).await }).await;
300 : }
301 0 : });
302 1284 : let st = state.clone();
303 1284 : crate::spawn_loop(async move {
304 : loop {
305 : // Tokio's timer natively; the browser's own timer on wasm32
306 : // (tokio time never fires without a reactor there).
307 : #[cfg(not(target_arch = "wasm32"))]
308 10628 : tokio::time::sleep(std::time::Duration::from_millis(500)).await;
309 : #[cfg(target_arch = "wasm32")]
310 : gloo_timers::future::TimeoutFuture::new(500).await;
311 9385 : let st = st.clone();
312 9385 : guarded(async move { interval_tick(&st).await }).await;
313 : }
314 : });
315 1284 : }
316 :
317 : /// Run one pipeline step on its own task so a panic inside it cannot end
318 : /// notification delivery for the whole process: the task boundary absorbs
319 : /// the panic, it is counted and logged, and the caller keeps consuming — a
320 : /// later matching change still notifies (5.8.6). The step is awaited, so
321 : /// delivery stays as serial as it was.
322 : #[cfg(not(target_arch = "wasm32"))]
323 103208 : async fn guarded<F>(fut: F)
324 103208 : where
325 103208 : F: std::future::Future<Output = ()>,
326 103208 : {
327 : use futures_util::FutureExt as _;
328 : // Caught here rather than on a spawned task: a task boundary would demand
329 : // Send + 'static of the step, and the interval step holds state that is
330 : // not Sync. Unwinding in place absorbs the panic just as well and keeps
331 : // the step running on this task, so delivery stays exactly as serial.
332 103208 : if std::panic::AssertUnwindSafe(fut)
333 103208 : .catch_unwind()
334 103208 : .await
335 103157 : .is_err()
336 0 : {
337 0 : note_panic();
338 103157 : }
339 103157 : }
340 :
341 : #[cfg(not(target_arch = "wasm32"))]
342 0 : fn note_panic() {
343 0 : TASK_PANICS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
344 0 : metrics::counter!("antares_notification_task_panics_total").increment(1);
345 0 : tracing::error!("notification pipeline task panicked; this change is lost");
346 0 : }
347 :
348 : /// Wasm32 has no task boundary to catch with (single-threaded executor, and
349 : /// the browser profile aborts on panic); the step runs inline.
350 : #[cfg(target_arch = "wasm32")]
351 : async fn guarded<F>(fut: F)
352 : where
353 : F: std::future::Future<Output = ()>,
354 : {
355 : fut.await;
356 : }
357 :
358 : /// Auto-record a current-state change into the temporal representation
359 : /// (5.6.11). Driven by the store's SYNCHRONOUS change hook, which fires inside
360 : /// every entity write (create, update, partial update, merge, replace, batch)
361 : /// for both the memory and postgres stores — so a new write path records
362 : /// without the handler having to remember. The per-handler `mirror_record`
363 : /// this replaced was the forgettable trap that left Partial Attribute Update
364 : /// (5.6.4) and Replace Attribute (5.6.19) silently unrecorded.
365 : ///
366 : /// Append-only, instance-precise: only the instances that are new or changed
367 : /// (by datasetId, ignoring volatile members) are appended, so a multi-instance
368 : /// attribute does not re-record unchanged datasets. An attribute or instance
369 : /// the write removed is recorded here through `mirror_delete_attr` (the typed
370 : /// null shape is not a plain append); entity DELETION keeps
371 : /// `mirror_delete_entity`, which the delete handlers call.
372 140892 : pub async fn record_temporal_change(
373 140892 : st: &AppState,
374 140892 : tenant: &TenantId,
375 140892 : before: Option<&Value>,
376 140892 : after: Option<&Value>,
377 140892 : ) {
378 140892 : if !st.record_locally() {
379 0 : return;
380 140892 : }
381 140892 : let Some(after) = after else {
382 3376 : return; // entity deletion — handled by mirror_delete_entity
383 : };
384 137516 : let Some(id) = after.get("id").and_then(Value::as_str) else {
385 0 : return;
386 : };
387 137516 : if !crate::history::records_history(st, tenant, after) {
388 6 : return;
389 137510 : }
390 137510 : let mut shell = Map::new();
391 687550 : for k in ["id", "type", "createdAt", "modifiedAt", "scope"] {
392 687550 : if let Some(v) = after.get(k) {
393 324540 : shell.insert(k.to_string(), v.clone());
394 376412 : }
395 : }
396 137510 : let shell = Value::Object(shell);
397 137510 : let event = |op, attr: &str, instance| TemporalEvent {
398 131616 : op,
399 131616 : tenant: tenant.clone(),
400 131616 : entity_id: id.to_owned(),
401 131616 : shell: shell.clone(),
402 131616 : attr: attr.to_owned(),
403 131616 : instance,
404 131616 : };
405 : // 4.5.6: the Scope of a Temporal Evolution is represented as a temporal
406 : // Property whose only sub-properties are the non-reified createdAt,
407 : // modifiedAt, deletedAt and observedAt; when it "is updated as the result
408 : // of a change from the Core API, the observedAt sub-Property should be
409 : // set as a copy of the modifiedAt sub-Property".
410 137510 : let ts = || {
411 488 : after
412 488 : .get("modifiedAt")
413 488 : .and_then(Value::as_str)
414 488 : .map(String::from)
415 488 : .unwrap_or_else(now_iso)
416 488 : };
417 137510 : if before.is_some_and(|b| b.get("scope") != after.get("scope")) {
418 : // 4.18: a deleted scope reads as the NGSI-LD Null
419 76 : if after.get("scope").is_none() {
420 28 : crate::history::mirror_delete_attr(st, tenant, id, "scope", None, &ts()).await;
421 48 : }
422 76 : if let Some(scope) = after.get("scope") {
423 48 : let ts = ts();
424 48 : let inst = json!({
425 48 : "type": "Property",
426 48 : "value": scope.clone(),
427 48 : "instanceId": format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4()),
428 48 : "createdAt": ts, "modifiedAt": ts, "observedAt": ts,
429 : });
430 48 : crate::history::push(st, event(TemporalOp::ScopeChanged, "scope", inst)).await;
431 28 : }
432 137434 : }
433 180660 : for (k, class) in diff(before, Some(after)) {
434 180660 : let op = match class {
435 128714 : ChangeClass::Created => TemporalOp::AttrCreated,
436 51534 : ChangeClass::Updated => TemporalOp::AttrModified,
437 : ChangeClass::Deleted => {
438 : // 4.5.7/4.5.8: each instance the write removed is recorded
439 : // as the NGSI-LD Null with deletedAt, whichever route
440 : // removed it (Delete Attribute, a replace, a merge-patch Null)
441 412 : let ts = ts();
442 432 : for ds in removed_datasets(before.and_then(|b| b.get(&k)), after.get(&k)) {
443 432 : crate::history::mirror_delete_attr(st, tenant, id, &k, ds.as_deref(), &ts)
444 432 : .await;
445 : }
446 412 : continue;
447 : }
448 : };
449 180248 : let Some(av) = after.get(&k) else { continue };
450 : // gate 1, value-change: an unchanged instance produces no event
451 180248 : let types: Vec<&str> = match after.get("type") {
452 6 : Some(Value::String(t)) => vec![t.as_str()],
453 180242 : Some(Value::Array(ts)) => ts.iter().filter_map(Value::as_str).collect(),
454 0 : _ => Vec::new(),
455 : };
456 180248 : let operation = if op == TemporalOp::AttrCreated {
457 128714 : "create"
458 : } else {
459 51534 : "update"
460 : };
461 180248 : let was = before.and_then(|b| b.get(&k));
462 180966 : for mut inst in changed_instances(was, av) {
463 180966 : if !crate::history::admits_change(st, tenant, &types, id, &k, was, &inst) {
464 49398 : continue;
465 131568 : }
466 131568 : if let Some(o) = inst.as_object_mut() {
467 131568 : let iid = instance_id(id, &k, &*o);
468 131568 : o.entry("instanceId".to_owned())
469 131568 : .or_insert_with(|| Value::String(iid));
470 : // Who wrote it, for the store's column; a store
471 : // takes it out before it keeps the instance.
472 131568 : o.insert(
473 131568 : antares_store::PROVENANCE_KEY.to_owned(),
474 131568 : crate::history::provenance(operation),
475 : );
476 0 : }
477 131568 : crate::history::push(st, event(op, &k, inst)).await;
478 : }
479 : }
480 140892 : }
481 :
482 : /// 4.5.7: an instance is the Attribute "at a particular point in time",
483 : /// recorded as its observedAt. The id of an observed instance is therefore
484 : /// derived from (entity, attribute, datasetId, observedAt), so a re-send for
485 : /// the same instant lands on the same row — the temporal store's upsert key —
486 : /// and corrects it instead of appending a duplicate. Without observedAt there
487 : /// is no instant to key on: a fresh random id, append-only.
488 : ///
489 : /// The instant is keyed through `dt_key`, not through the stamp as written.
490 : /// 4.6.3 leaves the seconds fraction optional and accepts a comma separator
491 : /// in requests, and the broker stores a DateTime exactly as the client wrote
492 : /// it, so one instant arrives under several spellings; keying the raw text
493 : /// gave each spelling its own instance and left the correction's target in
494 : /// place beside it — the failure 4.5.7 calls severe "in the case of
495 : /// modification or deletion requests for legal reasons".
496 131652 : pub(crate) fn instance_id(
497 131652 : entity: &str,
498 131652 : attr: &str,
499 131652 : inst: &serde_json::Map<String, Value>,
500 131652 : ) -> String {
501 131652 : let u = match inst.get("observedAt").and_then(Value::as_str) {
502 3602 : Some(at) => {
503 3602 : let ds = inst
504 3602 : .get("datasetId")
505 3602 : .and_then(Value::as_str)
506 3602 : .unwrap_or("@none");
507 3602 : uuid::Uuid::new_v5(
508 3602 : &uuid::Uuid::NAMESPACE_URL,
509 3602 : format!("{entity}\n{attr}\n{ds}\n{}", dt_key(at)).as_bytes(),
510 : )
511 : }
512 128050 : None => uuid::Uuid::new_v4(),
513 : };
514 131652 : format!("urn:ngsi-ld:Instance:{u}")
515 131652 : }
516 :
517 : /// The datasetIds (None: the default instance) of the instances in `before`
518 : /// that `after` no longer holds; every one of them when the Attribute went.
519 412 : fn removed_datasets(before: Option<&Value>, after: Option<&Value>) -> Vec<Option<String>> {
520 412 : let after_arr = after.and_then(Value::as_array);
521 412 : before
522 412 : .and_then(Value::as_array)
523 412 : .into_iter()
524 412 : .flatten()
525 574 : .filter(|bi| {
526 574 : !after_arr.is_some_and(|a| a.iter().any(|ai| instance_ds(ai) == instance_ds(bi)))
527 574 : })
528 432 : .map(|bi| {
529 432 : bi.get("datasetId")
530 432 : .and_then(Value::as_str)
531 432 : .map(String::from)
532 432 : })
533 412 : .collect()
534 412 : }
535 :
536 : /// The instances in `after` that are new or changed vs `before` — matched by
537 : /// datasetId, compared via `stable_eq` (volatile members ignored). A newly
538 : /// created attribute (`before` None) contributes every instance.
539 180248 : fn changed_instances(before: Option<&Value>, after: &Value) -> Vec<Value> {
540 180248 : let before_arr: Vec<&Value> = before
541 180248 : .and_then(Value::as_array)
542 180248 : .map(|a| a.iter().collect())
543 180248 : .unwrap_or_default();
544 180248 : after
545 180248 : .as_array()
546 180248 : .cloned()
547 180248 : .unwrap_or_default()
548 180248 : .into_iter()
549 181038 : .filter(|ai| {
550 181038 : match before_arr
551 181038 : .iter()
552 181038 : .find(|bi| instance_ds(bi) == instance_ds(ai))
553 : {
554 129464 : None => true,
555 51574 : Some(bi) => !stable_eq(bi, ai),
556 : }
557 181038 : })
558 180248 : .collect()
559 180248 : }
560 :
561 : pub(crate) use antares_matcher::{
562 : conditions_match, geo_params, is_active, selector_match, throttled,
563 : };
564 :
565 : pub(crate) use antares_notifier::{redact_userinfo, Outbound};
|
