Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Notification body construction (5.3.1, 5.8.6): the per-subscription shape, tombstones of deleted attributes and the data members.
3 :
4 : use super::*;
5 :
6 : /// Per-type NGSI-LD-null member and its showChanges previous-member (5.8.6).
7 476 : pub(super) fn null_members(atype: &str) -> (&'static str, Value, &'static str) {
8 476 : let null = Value::String("urn:ngsi-ld:null".into());
9 476 : match atype {
10 476 : "Relationship" => ("object", null, "previousObject"),
11 392 : "LanguageProperty" => (
12 100 : "languageMap",
13 100 : json!({"@none": "urn:ngsi-ld:null"}),
14 100 : "previousLanguageMap",
15 100 : ),
16 292 : "JsonProperty" => ("json", null, "previousJson"),
17 272 : "VocabProperty" => ("vocab", null, "previousVocab"),
18 252 : _ => ("value", null, "previousValue"),
19 : }
20 476 : }
21 :
22 132 : pub(super) fn current_member(atype: &str) -> &'static str {
23 132 : null_members(atype).0
24 132 : }
25 :
26 : /// The deletion tombstone for one former attribute instance (5.8.6 payload
27 : /// forms: typed null + optional datasetId / sysAttrs stamps / previous value).
28 212 : pub(super) fn tombstone(before_inst: &Value, sys: bool, show: bool, now: &str) -> Value {
29 212 : let atype = before_inst
30 212 : .get("type")
31 212 : .and_then(Value::as_str)
32 212 : .unwrap_or("Property");
33 212 : let (member, null_val, prev_member) = null_members(atype);
34 212 : let mut m = Map::new();
35 212 : m.insert("type".into(), Value::String(atype.to_owned()));
36 212 : m.insert(member.into(), null_val);
37 212 : if let Some(ds) = before_inst.get("datasetId") {
38 32 : m.insert("datasetId".into(), ds.clone());
39 180 : }
40 212 : if sys {
41 40 : for k in ["createdAt", "modifiedAt"] {
42 40 : if let Some(v) = before_inst.get(k) {
43 20 : m.insert(k.into(), v.clone());
44 20 : }
45 : }
46 20 : m.insert("deletedAt".into(), Value::String(now.to_owned()));
47 192 : }
48 212 : if show {
49 60 : if let Some(prev) = before_inst.get(member) {
50 60 : m.insert(prev_member.into(), prev.clone());
51 60 : }
52 152 : }
53 212 : Value::Object(m)
54 212 : }
55 :
56 323624 : pub(super) fn instance_ds(inst: &Value) -> Option<&str> {
57 323624 : inst.get("datasetId").and_then(Value::as_str)
58 323624 : }
59 :
60 : /// Instances of `before[attr]` that no longer exist in `after[attr]`
61 : /// (matched by datasetId — instance-level deletions count, 046_22_06).
62 164 : pub(super) fn deleted_instances<'a>(
63 164 : before: &'a Value,
64 164 : after: Option<&Value>,
65 164 : attr: &str,
66 164 : ) -> Vec<&'a Value> {
67 164 : let b: Vec<&Value> = before
68 164 : .get(attr)
69 164 : .and_then(Value::as_array)
70 164 : .map(|a| a.iter().collect())
71 164 : .unwrap_or_default();
72 164 : let a: Vec<&Value> = after
73 164 : .and_then(|d| d.get(attr))
74 164 : .and_then(Value::as_array)
75 164 : .map(|x| x.iter().collect())
76 164 : .unwrap_or_default();
77 164 : b.into_iter()
78 208 : .filter(|bi| !a.iter().any(|ai| instance_ds(ai) == instance_ds(bi)))
79 164 : .collect()
80 164 : }
81 :
82 : pub(crate) struct NotifShape {
83 : pub(crate) repr: crate::repr::Repr,
84 : pub(crate) show_changes: bool,
85 : pub(crate) join: Option<(String, usize)>,
86 : }
87 :
88 5020 : pub(crate) fn notif_shape(sub: &Value, ctx: &Context) -> NotifShape {
89 5020 : let n = sub.get("notification").and_then(Value::as_object);
90 5020 : let format = n
91 5020 : .and_then(|n| n.get("format"))
92 5020 : .and_then(Value::as_str)
93 5020 : .unwrap_or("normalized");
94 5020 : let mut repr = crate::repr::Repr {
95 5020 : sys_attrs: n
96 5020 : .and_then(|n| n.get("sysAttrs"))
97 5020 : .and_then(Value::as_bool)
98 5020 : .unwrap_or(false),
99 5020 : key_values: matches!(format, "keyValues" | "simplified"),
100 5020 : concise: format == "concise",
101 : // Table 5.2.12-1: "Language filter to be applied to the query (clause 4.15)"
102 5020 : lang: sub.get("lang").and_then(Value::as_str).map(str::to_owned),
103 5020 : ..Default::default()
104 : };
105 15060 : let names = |key: &str| -> Option<Vec<String>> {
106 15060 : n.and_then(|n| n.get(key))
107 15060 : .and_then(Value::as_array)
108 15060 : .map(|a| {
109 52 : a.iter()
110 52 : .filter_map(Value::as_str)
111 52 : .map(str::to_owned)
112 52 : .collect()
113 52 : })
114 15060 : };
115 5020 : if let Some(attrs) = names("attributes") {
116 20 : repr.attrs = Some(attrs.iter().map(|a| ctx.expand_key(a)).collect());
117 5000 : }
118 : // 4.21 NGSI-LD Attribute Projection Language: pick/omit values are
119 : // projection language strings, which Table 5.2.14.1-1 requires for the
120 : // notification members too ("a valid attribute projection language string
121 : // as per clause 4.21"). Each term may carry a LinkedEntityTerm
122 : // (`ProjectionTerm = AttrName *1(LinkedEntityTerm)`), which is what
123 : // constrains an Attribute inside a Linked Entity retrieved by join. Building
124 : // the nodes by hand here degraded `refDevice{type}` to a literal Attribute
125 : // name matching nothing, so the term was dropped instead of applied.
126 : // A term that fails to parse is dropped rather than kept flat: for pick that
127 : // withholds the Attribute, which is the safe direction.
128 5020 : let nodes = |list: Vec<String>| -> Vec<crate::repr::ProjNode> {
129 32 : list.iter()
130 82 : .filter_map(|term| crate::repr::parse_projection(term, ctx).ok())
131 32 : .flatten()
132 32 : .collect()
133 32 : };
134 5020 : if let Some(pick) = names("pick") {
135 26 : repr.pick = Some(nodes(pick));
136 4994 : }
137 5020 : if let Some(omit) = names("omit") {
138 6 : repr.omit = Some(nodes(omit));
139 5014 : }
140 5020 : if let Some(ds) = sub.get("datasetId").and_then(Value::as_array) {
141 4 : repr.dataset_id = Some(
142 4 : ds.iter()
143 4 : .filter_map(Value::as_str)
144 4 : .map(str::to_owned)
145 4 : .collect(),
146 4 : );
147 5016 : }
148 5020 : let join = n
149 5020 : .and_then(|n| n.get("join"))
150 5020 : .and_then(Value::as_str)
151 5020 : .filter(|j| *j == "inline" || *j == "flat")
152 5020 : .map(|j| {
153 : // the stored member is bounded at creation, but a Subscription
154 : // written before that bound existed is still on disk, so the
155 : // traversal takes the ceiling from here too
156 56 : let level = (n
157 56 : .and_then(|n| n.get("joinLevel"))
158 56 : .and_then(Value::as_u64)
159 56 : .map_or(1, |n| usize::try_from(n).unwrap_or(usize::MAX)))
160 56 : .min(crate::bounds::MAX_JOIN_LEVEL);
161 56 : (j.to_owned(), level)
162 56 : });
163 : NotifShape {
164 5020 : repr,
165 5020 : show_changes: n
166 5020 : .and_then(|n| n.get("showChanges"))
167 5020 : .and_then(Value::as_bool)
168 5020 : .unwrap_or(false),
169 5020 : join,
170 : }
171 5020 : }
172 :
173 : /// Everything [`build_data`] reads of one Subscription for one change, as a
174 : /// memo key: the members [`notif_shape`] reads, the deleted Attributes its
175 : /// triggers name and the `@context` it compacts with. Subscriptions with
176 : /// equal keys get equal payloads, so 1 000 subscriptions of one view render
177 : /// one. The `@context` counts by identity: the drain's memo hands
178 : /// every Subscription naming one URL the same `Arc`, and a key holds an
179 : /// address only while its `Arc` lives in a matched group, so no other
180 : /// `@context` can take that address in the meantime.
181 9058 : pub(super) fn payload_key(
182 9058 : sub: &Value,
183 9058 : ctx: &Arc<Context>,
184 9058 : deleted: &[String],
185 9058 : entity_deleted: bool,
186 9058 : ) -> String {
187 9058 : let n = sub.get("notification");
188 72464 : let member = |k: &str| n.and_then(|n| n.get(k));
189 9058 : json!([
190 9058 : Arc::as_ptr(ctx).addr(),
191 9058 : member("format"),
192 9058 : member("sysAttrs"),
193 9058 : member("attributes"),
194 9058 : member("pick"),
195 9058 : member("omit"),
196 9058 : member("showChanges"),
197 9058 : member("join"),
198 9058 : member("joinLevel"),
199 9058 : sub.get("lang"),
200 9058 : sub.get("datasetId"),
201 : deleted,
202 : entity_deleted,
203 : ])
204 9058 : .to_string()
205 9058 : }
206 :
207 : /// Build the notification `data` array for one change and one subscription.
208 : #[allow(clippy::too_many_arguments)] // one param per 5.8.6 payload input
209 4956 : pub(super) async fn build_data(
210 4956 : st: &AppState,
211 4956 : tenant: &TenantId,
212 4956 : sub: &Value,
213 4956 : ctx: &Context,
214 4956 : before: Option<&Value>,
215 4956 : after: Option<&Value>,
216 4956 : relevant_deleted_attrs: &[String],
217 4956 : entity_deleted_fired: bool,
218 4956 : now: &str,
219 4956 : ) -> Vec<Value> {
220 4956 : let shape = notif_shape(sub, ctx);
221 4956 : let sys = shape.repr.sys_attrs;
222 4956 : let show = shape.show_changes;
223 4956 : let internal = match after {
224 4924 : Some(a) => {
225 4924 : let mut doc = a.clone();
226 4924 : if show {
227 : // previous* on changed instances (046_31..33). An entity off
228 : // the change feed is an object; one that is not carries
229 : // nothing to decorate and travels on unchanged.
230 68 : if let (Some(b), Some(obj)) = (before, doc.as_object_mut()) {
231 376 : for (k, v) in obj.iter_mut() {
232 376 : if crate::repr::ENTITY_META.contains(&k.as_str()) {
233 240 : continue;
234 136 : }
235 136 : let Some(arr) = v.as_array_mut() else {
236 0 : continue;
237 : };
238 136 : for inst in arr {
239 136 : let Some(bi) = b.get(k).and_then(Value::as_array).and_then(|ba| {
240 132 : ba.iter().find(|x| instance_ds(x) == instance_ds(inst))
241 132 : }) else {
242 4 : continue;
243 : };
244 132 : let atype = inst
245 132 : .get("type")
246 132 : .and_then(Value::as_str)
247 132 : .unwrap_or("Property");
248 132 : let member = current_member(atype);
249 132 : let (_, _, prev_member) = null_members(atype);
250 132 : if bi.get(member) != inst.get(member) {
251 34 : if let (Some(pv), Some(io)) =
252 34 : (bi.get(member).cloned(), inst.as_object_mut())
253 34 : {
254 34 : io.insert(prev_member.into(), pv);
255 34 : }
256 98 : }
257 : }
258 : }
259 4 : }
260 4856 : }
261 : // deletion tombstones appended beside surviving instances
262 4924 : if let Some(b) = before {
263 328 : for attr in relevant_deleted_attrs {
264 164 : let gone = deleted_instances(b, Some(&doc), attr);
265 164 : if gone.is_empty() {
266 0 : continue;
267 164 : }
268 : // one tombstone per deleted instance, each with its
269 : // datasetId, deleteAll included (046_22_08)
270 164 : let tss: Vec<Value> = gone
271 164 : .iter()
272 180 : .map(|di| tombstone(di, sys, show, now))
273 164 : .collect();
274 164 : let Some(target) = doc.as_object_mut() else {
275 0 : continue;
276 : };
277 164 : if let Some(arr) = target
278 164 : .entry(attr.clone())
279 164 : .or_insert_with(|| Value::Array(vec![]))
280 164 : .as_array_mut()
281 164 : {
282 164 : arr.extend(tss);
283 164 : }
284 : }
285 4596 : }
286 4924 : doc
287 : }
288 : None => {
289 : // entity deleted: tombstone entity (046_21) + per-trigger attrs.
290 : // The caller returns early unless one of before/after is there,
291 : // so this arm has a before; without one there is nothing to
292 : // describe and nothing to notify about.
293 32 : let Some(b) = before else {
294 0 : return Vec::new();
295 : };
296 32 : let mut m = Map::new();
297 64 : for k in ["id", "type"] {
298 64 : if let Some(v) = b.get(k) {
299 64 : m.insert(k.into(), v.clone());
300 64 : }
301 : }
302 32 : if sys {
303 16 : for k in ["createdAt", "modifiedAt"] {
304 16 : if let Some(v) = b.get(k) {
305 16 : m.insert(k.into(), v.clone());
306 16 : }
307 : }
308 24 : }
309 32 : m.insert("deletedAt".into(), Value::String(now.to_owned()));
310 : // 5.8.6: an Entity deletion notifies "only the deletedAt system
311 : // property" unless showChanges, then every attribute, tombstoned
312 : // (046_37); attributeDeleted alone names the attributes it fired for
313 32 : let attrs: Vec<String> = match (entity_deleted_fired, show) {
314 12 : (true, true) => attr_keys(b),
315 12 : (true, false) => Vec::new(),
316 8 : (false, _) => relevant_deleted_attrs.to_vec(),
317 : };
318 32 : for attr in attrs {
319 32 : let insts: Vec<Value> = b
320 32 : .get(&attr)
321 32 : .and_then(Value::as_array)
322 32 : .map(|a| a.iter().map(|i| tombstone(i, sys, show, now)).collect())
323 32 : .unwrap_or_default();
324 32 : if !insts.is_empty() {
325 32 : m.insert(attr, Value::Array(insts));
326 32 : }
327 : }
328 32 : Value::Object(m)
329 : }
330 : };
331 4956 : let shaped = crate::repr::apply(&internal, &shape.repr);
332 4956 : let mut main = crate::repr::compact_for(&shape.repr, &shaped, ctx);
333 4956 : let mut data = Vec::new();
334 24 : match &shape.join {
335 56 : Some((mode, level)) if mode == "inline" => {
336 32 : crate::repr::inline_join(st, tenant, ctx, &shape.repr, &mut main, &internal, *level)
337 32 : .await;
338 32 : data.push(main);
339 : }
340 24 : Some((mode, level)) if mode == "flat" => {
341 24 : let main_id = internal.get("id").and_then(Value::as_str).unwrap_or("");
342 24 : let mut linked = std::collections::BTreeMap::new();
343 24 : crate::repr::collect_flat(st, tenant, ctx, &shape.repr, &internal, *level, &mut linked)
344 24 : .await;
345 24 : data.push(main);
346 24 : for (id, (ldoc, lrepr)) in linked {
347 24 : if id != main_id {
348 24 : data.push(crate::repr::compact_for(
349 24 : &lrepr,
350 24 : &crate::repr::apply(&ldoc, &lrepr),
351 24 : ctx,
352 24 : ));
353 24 : }
354 : }
355 : }
356 4900 : _ => data.push(main),
357 : }
358 4956 : data
359 4956 : }
|
