pharos_runtime_database/
state_delta.rs

1//! Scoped writes for legacy worker state changes during logical adoption.
2
3use std::collections::{BTreeMap, BTreeSet};
4
5use serde_json::Value;
6
7use crate::DbBackendConfig;
8use crate::operations::{
9    AppSchema, DatabaseError, DatabaseOperations, Mutation, Predicate, Transaction, WriteValue,
10};
11
12const WORKER_SECTIONS: &[(&str, &str, Option<&str>)] = &[
13    (
14        "notification_deliveries",
15        "notification_delivery",
16        Some("delivery_status"),
17    ),
18    (
19        "integration_deliveries",
20        "integration_delivery",
21        Some("delivery_status"),
22    ),
23    (
24        "integration_delivery_attempts",
25        "integration_delivery_attempt",
26        None,
27    ),
28    ("job_executions", "job_execution", Some("execution_status")),
29    ("job_execution_attempts", "job_execution_attempt", None),
30    (
31        "workflow_execution_events",
32        "workflow_execution_event",
33        None,
34    ),
35];
36
37/// Reconciles a scoped worker result into a local JSON read cache without
38/// replacing unrelated app sections or treating the cache as database authority.
39pub fn merge_scoped_worker_cache(
40    cache: &Value,
41    before: &Value,
42    after: &Value,
43) -> Result<Value, DatabaseError> {
44    if !matches!(
45        before.get("_worker_scope").and_then(Value::as_str),
46        Some("job" | "queue")
47    ) {
48        return Err(DatabaseError::InvalidInput);
49    }
50    let mut merged = cache.clone();
51    let root = merged.as_object_mut().ok_or(DatabaseError::InvalidInput)?;
52    for &(section, _, _) in WORKER_SECTIONS {
53        if before.get(section).is_none() {
54            continue;
55        }
56        let incoming = after
57            .get(section)
58            .and_then(Value::as_array)
59            .ok_or(DatabaseError::InvalidInput)?;
60        let rows = root
61            .entry(section)
62            .or_insert_with(|| Value::Array(Vec::new()))
63            .as_array_mut()
64            .ok_or(DatabaseError::InvalidInput)?;
65        for item in incoming {
66            let id = item
67                .get("id")
68                .and_then(Value::as_i64)
69                .filter(|id| *id > 0)
70                .ok_or(DatabaseError::InvalidInput)?;
71            if before
72                .get(section)
73                .and_then(Value::as_array)
74                .is_some_and(|rows| {
75                    rows.iter()
76                        .any(|row| row.get("id").and_then(Value::as_i64) == Some(id) && row == item)
77                })
78            {
79                continue;
80            }
81            if let Some(current) = rows
82                .iter_mut()
83                .find(|row| row.get("id").and_then(Value::as_i64) == Some(id))
84            {
85                let current_fields = current.as_object_mut().ok_or(DatabaseError::InvalidInput)?;
86                let changes = item.as_object().ok_or(DatabaseError::InvalidInput)?;
87                current_fields.extend(
88                    changes
89                        .iter()
90                        .filter(|(name, _)| !transient_worker_field(section, name))
91                        .map(|(name, value)| (name.clone(), value.clone())),
92                );
93            } else {
94                let fields = item.as_object().ok_or(DatabaseError::InvalidInput)?;
95                rows.push(Value::Object(
96                    fields
97                        .iter()
98                        .filter(|(name, _)| !transient_worker_field(section, name))
99                        .map(|(name, value)| (name.clone(), value.clone()))
100                        .collect(),
101                ));
102            }
103        }
104    }
105    Ok(merged)
106}
107
108/// Excludes delivery credentials and display aliases from the derived cache.
109fn transient_worker_field(section: &str, name: &str) -> bool {
110    matches!(name, "secret" | "delivery_kind" | "timestamp_header")
111        || (section == "notification_deliveries" && name == "adapter_kind")
112}
113
114/// Persists only rows changed by one worker tick. An unrelated account, receipt,
115/// or delivery written after the worker read its snapshot is never rewritten.
116pub fn apply_worker_delta(
117    adapter: &dyn DatabaseOperations,
118    config: &DbBackendConfig,
119    schema: &AppSchema,
120    before: &Value,
121    after: &Value,
122) -> Result<Value, DatabaseError> {
123    let mut steps = Vec::new();
124    let claims = before.get("_worker_claims").and_then(Value::as_array);
125    if before.get("_worker_claims") != after.get("_worker_claims") {
126        return Err(DatabaseError::InvalidInput);
127    }
128    for claim in claims.into_iter().flatten() {
129        let entity = claim
130            .get("entity")
131            .and_then(Value::as_str)
132            .ok_or(DatabaseError::InvalidInput)?;
133        if !WORKER_SECTIONS.iter().any(|(_, name, _)| *name == entity) {
134            return Err(DatabaseError::InvalidInput);
135        }
136        let mut filters = vec![Predicate::Equal(
137            "id".to_owned(),
138            claim
139                .get("id")
140                .cloned()
141                .ok_or(DatabaseError::InvalidInput)?,
142        )];
143        filters.extend(
144            claim
145                .get("fields")
146                .and_then(Value::as_object)
147                .ok_or(DatabaseError::InvalidInput)?
148                .iter()
149                .map(|(field, value)| Predicate::Equal(field.clone(), value.clone())),
150        );
151        steps.push(Mutation::RequireCount {
152            entity: entity.to_owned(),
153            filter: Predicate::All(filters),
154            minimum: 1,
155            maximum: Some(1),
156        });
157    }
158    let mut inserted = Vec::new();
159    for &(section, entity_name, status_field) in WORKER_SECTIONS {
160        let Some(entity) = schema
161            .entities
162            .iter()
163            .find(|entity| entity.name == entity_name)
164        else {
165            continue;
166        };
167        let known_fields = entity
168            .fields
169            .iter()
170            .map(|field| field.name.as_str())
171            .collect::<BTreeSet<_>>();
172        let old_rows = before.get(section).and_then(Value::as_array);
173        let new_rows = after.get(section).and_then(Value::as_array);
174        let mut old_by_id = BTreeMap::new();
175        for row in old_rows.into_iter().flatten() {
176            let id = row
177                .get("id")
178                .and_then(Value::as_i64)
179                .ok_or(DatabaseError::InvalidInput)?;
180            if old_by_id.insert(id, row).is_some() {
181                return Err(DatabaseError::InvalidInput);
182            }
183        }
184        for row in new_rows.into_iter().flatten() {
185            let id = row
186                .get("id")
187                .and_then(Value::as_i64)
188                .ok_or(DatabaseError::InvalidInput)?;
189            let old = old_by_id.remove(&id);
190            let Some(fields) = row.as_object() else {
191                return Err(DatabaseError::InvalidInput);
192            };
193            let mut values = BTreeMap::new();
194            for (name, value) in fields {
195                let target = if name == "timestamp_header" {
196                    "signature_timestamp_header"
197                } else {
198                    name.as_str()
199                };
200                if target == "id" {
201                    continue;
202                }
203                let previous = old.and_then(|old| old.get(name));
204                if old.is_some() && previous == Some(value) {
205                    continue;
206                }
207                if !known_fields.contains(target) {
208                    return Err(DatabaseError::InvalidContract);
209                }
210                let declaration = entity
211                    .fields
212                    .iter()
213                    .find(|field| field.name == target)
214                    .ok_or(DatabaseError::InvalidContract)?;
215                let value = match (declaration.kind, value.as_str()) {
216                    (crate::operations::FieldType::Timestamp, Some("")) if declaration.nullable => {
217                        Value::Null
218                    }
219                    (crate::operations::FieldType::Timestamp, Some(text))
220                        if text.bytes().all(|byte| byte.is_ascii_digit()) =>
221                    {
222                        let epoch = text
223                            .parse::<i64>()
224                            .map_err(|_| DatabaseError::InvalidInput)?;
225                        let time = chrono::DateTime::<chrono::Utc>::from_timestamp(epoch, 0)
226                            .ok_or(DatabaseError::InvalidInput)?;
227                        Value::String(time.to_rfc3339())
228                    }
229                    _ => value.clone(),
230                };
231                values.insert(target.to_owned(), WriteValue::Literal(value));
232            }
233            if let Some(old) = old {
234                if !values.is_empty() {
235                    let mut filter = vec![Predicate::Equal("id".to_owned(), Value::from(id))];
236                    let claimed = claims.into_iter().flatten().find(|claim| {
237                        claim.get("entity").and_then(Value::as_str) == Some(entity_name)
238                            && claim.get("id").and_then(Value::as_i64) == Some(id)
239                    });
240                    if let Some(status) = status_field.and_then(|field| {
241                        claimed
242                            .and_then(|claim| claim.get("fields"))
243                            .and_then(|fields| fields.get(field))
244                            .or_else(|| old.get(field))
245                    }) {
246                        filter.push(Predicate::Equal(
247                            status_field.unwrap().to_owned(),
248                            status.clone(),
249                        ));
250                    }
251                    steps.push(Mutation::Update {
252                        entity: entity_name.to_owned(),
253                        filter: Predicate::All(filter),
254                        values,
255                        expected_rows: Some(1),
256                    });
257                }
258            } else {
259                inserted.push((section, id, steps.len()));
260                steps.push(Mutation::Insert {
261                    entity: entity_name.to_owned(),
262                    values,
263                });
264            }
265        }
266        if !old_by_id.is_empty() {
267            return Err(DatabaseError::InvalidInput);
268        }
269    }
270    if steps.is_empty() {
271        return Ok(after.clone());
272    }
273    let result = adapter.transact(
274        config,
275        schema,
276        &Transaction {
277            steps,
278            idempotency_key: None,
279        },
280    )?;
281    let mut corrected = after.clone();
282    for (section, provisional_id, step_index) in inserted {
283        let actual_id = result
284            .generated_ids
285            .get(step_index)
286            .copied()
287            .flatten()
288            .filter(|id| *id > 0)
289            .ok_or(DatabaseError::StorageFailure)?;
290        let row = corrected
291            .get_mut(section)
292            .and_then(Value::as_array_mut)
293            .and_then(|rows| {
294                rows.iter_mut()
295                    .find(|row| row.get("id").and_then(Value::as_i64) == Some(provisional_id))
296            })
297            .ok_or(DatabaseError::StorageFailure)?;
298        row["id"] = Value::from(actual_id);
299    }
300    Ok(corrected)
301}
pharos_runtime_database/state_delta.rs

pharos_runtime_database/
state_delta.rs
