pharos_native/runtime/event_discovery/
model.rs

1//! Bounded event identity, location and existing-record comparisons for the discovery consumer.
2
3use super::{BTreeMap, DiscoveredEvent, Map, Value, normalized_name, public_link};
4use pharos_runtime_database::operations::{DatabaseError, Predicate, Query};
5
6/// Normalizes public source whitespace while rejecting control characters and excessive text.
7fn text(value: &Value, limit: usize) -> Option<String> {
8    let value = value
9        .as_str()?
10        .split_whitespace()
11        .collect::<Vec<_>>()
12        .join(" ");
13    (value.len() <= limit && !value.chars().any(char::is_control)).then_some(value)
14}
15
16/// Preserves explicit source identities across URL/date edits without merging unrelated opaque IDs.
17pub(super) fn event_identity(
18    value: &Value,
19    source: &str,
20    start: u64,
21    venue: &str,
22    title: &str,
23    event_url: &str,
24) -> Option<String> {
25    if let Some(identity) = value
26        .get("@id")
27        .or_else(|| value.get("identifier"))
28        .and_then(Value::as_str)
29        && !identity.is_empty()
30        && identity.len() <= 512
31        && !identity.chars().any(char::is_control)
32    {
33        return if let Some(url) = public_link(identity) {
34            pharos_runtime_database::operations::operation_fingerprint(&("event_identity", url))
35                .ok()
36        } else {
37            pharos_runtime_database::operations::operation_fingerprint(&(source, identity)).ok()
38        };
39    }
40    if value
41        .get("url")
42        .and_then(Value::as_str)
43        .and_then(public_link)
44        .is_some()
45    {
46        pharos_runtime_database::operations::operation_fingerprint(&event_url).ok()
47    } else {
48        pharos_runtime_database::operations::operation_fingerprint(&(
49            source,
50            start,
51            normalized_name(venue),
52            normalized_name(title),
53        ))
54        .ok()
55    }
56}
57
58/// Extracts optional public address/country evidence without inventing missing location fields.
59pub(super) fn location(value: &Value) -> Option<(String, String)> {
60    let Some(address) = value.pointer("/location/address") else {
61        return Some((String::new(), String::new()));
62    };
63    if address.is_string() {
64        return Some((text(address, 500)?, String::new()));
65    }
66    let address = address.as_object()?;
67    let mut parts = Vec::new();
68    for key in [
69        "streetAddress",
70        "addressLocality",
71        "addressRegion",
72        "postalCode",
73    ] {
74        if let Some(value) = address.get(key) {
75            let value = text(value, 300)?;
76            if !value.is_empty() {
77                parts.push(value);
78            }
79        }
80    }
81    let line = parts.join(", ");
82    if line.len() > 500 {
83        return None;
84    }
85    let country = match address.get("addressCountry") {
86        None => String::new(),
87        Some(value) => text(value.get("name").unwrap_or(value), 100)?,
88    };
89    Some((line, country))
90}
91
92/// Keeps at most sixteen explicit organizer names; names never supply contact or credential authority.
93pub(super) fn organizer_names(value: &Value) -> Option<Vec<String>> {
94    let Some(value) = value.get("organizer") else {
95        return Some(Vec::new());
96    };
97    let values = value
98        .as_array()
99        .map(Vec::as_slice)
100        .unwrap_or_else(|| std::slice::from_ref(value));
101    if values.len() > 16 {
102        return None;
103    }
104    let mut names = BTreeMap::new();
105    for value in values {
106        let name = text(value.get("name").unwrap_or(value), 300)?;
107        if !name.is_empty() {
108            names.entry(normalized_name(&name)).or_insert(name);
109        }
110    }
111    Some(names.into_values().collect())
112}
113
114/// Supplies only public event facts; relationship identities and authority metadata are assigned by the host.
115pub(super) fn values(event: &DiscoveredEvent) -> Map<String, Value> {
116    serde_json::json!({"event_key":event.key,"title":event.title,"starts_at":event.starts_at,"ends_at":event.ends_at,"venue":event.venue,"address":event.address,"country":event.country,"timezone":event.timezone,"event_state":event.event_state,"event_url":event.event_url,"ticket_url":event.ticket_url,"status":event.status}).as_object().cloned().unwrap_or_default()
117}
118
119/// Loads all relevant prior events in batches matching the shared predicate's sixty-four-value limit.
120pub(super) fn existing_events(
121    store: &super::super::database::LogicalAppStore,
122    events: &[DiscoveredEvent],
123) -> Result<BTreeMap<String, Map<String, Value>>, DatabaseError> {
124    let entity = pharos_runtime_database::operations::entity(&store.schema, "gig_event")?;
125    let wanted = [
126        "id",
127        "event_key",
128        "title",
129        "artist_row_id",
130        "organizer_id",
131        "source_id",
132        "starts_at",
133        "ends_at",
134        "venue",
135        "address",
136        "country",
137        "timezone",
138        "event_state",
139        "event_url",
140        "ticket_url",
141        "status",
142        "manual_locked",
143        "last_seen_at",
144        "record_updated_at",
145    ];
146    let fields = wanted
147        .into_iter()
148        .filter(|name| entity.fields.iter().any(|field| field.name == *name))
149        .map(str::to_owned)
150        .collect::<Vec<_>>();
151    let mut result = BTreeMap::new();
152    for batch in events.chunks(64) {
153        let rows = store.adapter.query(
154            &store.config,
155            &store.schema,
156            &Query {
157                entity: "gig_event".into(),
158                fields: fields.clone(),
159                filter: Some(Predicate::In(
160                    "event_key".into(),
161                    batch
162                        .iter()
163                        .map(|event| Value::String(event.key.clone()))
164                        .collect(),
165                )),
166                order: Vec::new(),
167                limit: 64,
168                offset: 0,
169            },
170        )?;
171        for row in rows {
172            if let Some(key) = row.get("event_key").and_then(Value::as_str) {
173                result.insert(key.to_owned(), row);
174            }
175        }
176    }
177    Ok(result)
178}
179
180/// Requires review when another source contradicts accepted facts; optional absent facts never erase evidence.
181pub(super) fn conflicts(
182    existing: &Map<String, Value>,
183    event: &DiscoveredEvent,
184    source: i64,
185) -> bool {
186    if existing.get("manual_locked").and_then(Value::as_bool) == Some(true) {
187        return false;
188    }
189    if existing.get("status").and_then(Value::as_str) == Some("cancelled")
190        && !event.state_explicit
191        && existing
192            .get("starts_at")
193            .and_then(Value::as_str)
194            .and_then(super::runtime_timestamp_seconds)
195            != super::runtime_timestamp_seconds(&event.starts_at)
196    {
197        return true;
198    }
199    if existing.get("source_id").and_then(Value::as_i64) == Some(source) {
200        return false;
201    }
202    let same_announcement = super::social_announcements::performance_title(&event.title)
203        .zip(
204            existing
205                .get("title")
206                .and_then(Value::as_str)
207                .and_then(super::social_announcements::performance_title),
208        )
209        .is_some_and(|(incoming, current)| incoming == current);
210    values(event).iter().any(|(field, incoming)| {
211        if same_announcement && matches!(field.as_str(), "title" | "event_url") {
212            return false;
213        }
214        if matches!(field.as_str(), "event_key" | "timezone")
215            || !event.state_explicit && matches!(field.as_str(), "status" | "event_state")
216        {
217            return false;
218        }
219        let Some(current) = existing.get(field) else {
220            return false;
221        };
222        if field == "event_state" && current.as_str() == Some("unknown") {
223            return false;
224        }
225        if incoming.as_str() == Some("") || current.as_str() == Some("") {
226            return false;
227        }
228        if matches!(field.as_str(), "starts_at" | "ends_at") {
229            return incoming.as_str().and_then(super::runtime_timestamp_seconds)
230                != current.as_str().and_then(super::runtime_timestamp_seconds);
231        }
232        if matches!(field.as_str(), "title" | "venue" | "address" | "country") {
233            return incoming.as_str().map(normalized_name) != current.as_str().map(normalized_name);
234        }
235        current != incoming
236    })
237}
238
239/// Preserves primary identities, omitted facts and explicit cancellations while assigning a fresh save revision.
240pub(super) fn write_values(
241    existing: Option<&Map<String, Value>>,
242    event: &DiscoveredEvent,
243    artist: i64,
244    source: i64,
245    now: &str,
246    organizer: Option<i64>,
247) -> Option<Map<String, Value>> {
248    let mut values = values(event);
249    if let Some(existing) = existing {
250        let other_source = existing.get("source_id").and_then(Value::as_i64) != Some(source);
251        for (field, current) in existing {
252            if !values.contains_key(field) || matches!(field.as_str(), "event_key") {
253                continue;
254            }
255            let supplied_empty = values.get(field).and_then(Value::as_str) == Some("");
256            let implicit_state =
257                !event.state_explicit && matches!(field.as_str(), "status" | "event_state");
258            if (other_source || supplied_empty || implicit_state)
259                && !current.is_null()
260                && current.as_str() != Some("")
261            {
262                values.insert(field.clone(), current.clone());
263            }
264        }
265        if !event.state_explicit
266            && existing.get("status").and_then(Value::as_str) != Some("cancelled")
267            && existing
268                .get("starts_at")
269                .and_then(Value::as_str)
270                .and_then(super::runtime_timestamp_seconds)
271                != super::runtime_timestamp_seconds(&event.starts_at)
272        {
273            values.insert("event_state".into(), Value::String("rescheduled".into()));
274        }
275    }
276    values.insert(
277        "artist_row_id".into(),
278        existing
279            .and_then(|row| row.get("artist_row_id"))
280            .cloned()
281            .unwrap_or(Value::from(artist)),
282    );
283    values.insert(
284        "source_id".into(),
285        existing
286            .and_then(|row| row.get("source_id"))
287            .filter(|value| !value.is_null())
288            .cloned()
289            .unwrap_or(Value::from(source)),
290    );
291    values.insert(
292        "organizer_id".into(),
293        existing
294            .and_then(|row| row.get("organizer_id"))
295            .filter(|value| !value.is_null())
296            .cloned()
297            .unwrap_or_else(|| organizer.map(Value::from).unwrap_or(Value::Null)),
298    );
299    values.insert("last_seen_at".into(), Value::String(now.into()));
300    values.insert("manual_locked".into(), Value::Bool(false));
301    values.insert(
302        "record_updated_at".into(),
303        Value::String(super::super::dynamic::random_secret_hex(16).ok()?),
304    );
305    Some(values)
306}
307
308/// Resolves active existing organizers by unambiguous normalized names using public metadata only.
309pub(super) fn known_organizers(
310    store: &super::super::database::LogicalAppStore,
311) -> Result<BTreeMap<String, Map<String, Value>>, DatabaseError> {
312    if !store
313        .schema
314        .entities
315        .iter()
316        .any(|entity| entity.name == "gig_organizer")
317    {
318        return Ok(BTreeMap::new());
319    }
320    let rows = store.adapter.query(
321        &store.config,
322        &store.schema,
323        &Query {
324            entity: "organizer".into(),
325            fields: vec![
326                "id".into(),
327                "name".into(),
328                "crm_status".into(),
329                "record_updated_at".into(),
330            ],
331            filter: Some(Predicate::Equal(
332                "crm_status".into(),
333                Value::String("active".into()),
334            )),
335            order: Vec::new(),
336            limit: 1000,
337            offset: 0,
338        },
339    )?;
340    if rows.len() == 1000 {
341        return Err(DatabaseError::CapacityExceeded);
342    }
343    let mut names: BTreeMap<String, Option<Map<String, Value>>> = BTreeMap::new();
344    for row in rows {
345        if let Some(name) = row.get("name").and_then(Value::as_str) {
346            names
347                .entry(normalized_name(name))
348                .and_modify(|existing| *existing = None)
349                .or_insert(Some(row));
350        }
351    }
352    Ok(names
353        .into_iter()
354        .filter_map(|(name, row)| row.map(|row| (name, row)))
355        .collect())
356}
pharos_native/runtime/event_discovery/model.rs

pharos_native/runtime/event_discovery/
model.rs
