pharos_runtime_database/
document.rs

1//! Atomic document operations for apps that declare document-shaped persistence.
2
3use crate::DbBackendConfig;
4use crate::operations::{
5    AppSchema, DatabaseError, DatabaseOperations, FieldType, Mutation, Predicate, Query,
6    Transaction, WriteValue,
7};
8use serde_json::Value;
9use std::collections::BTreeMap;
10
11const MAX_DOCUMENT_BYTES: usize = 1024 * 1024;
12
13/// Checks only declared document queue sections, without relational projection.
14pub fn has_due_work(value: &Value) -> bool {
15    let Ok(now) = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) else {
16        return false;
17    };
18    let timestamp = |value: Option<&Value>| {
19        value.and_then(Value::as_str).and_then(|text| {
20            text.parse::<u64>().ok().or_else(|| {
21                chrono::DateTime::parse_from_rfc3339(text)
22                    .ok()
23                    .and_then(|date| u64::try_from(date.timestamp()).ok())
24            })
25        })
26    };
27    [
28        "notification_deliveries",
29        "integration_deliveries",
30        "local_file_deliveries",
31    ]
32    .iter()
33    .any(|section| {
34        value
35            .get(*section)
36            .and_then(Value::as_array)
37            .is_some_and(|rows| {
38                rows.iter().any(|row| {
39                    row.get("delivery_status").and_then(Value::as_str) == Some("queued")
40                        || (row.get("delivery_status").and_then(Value::as_str) == Some("retryable")
41                            && timestamp(row.get("next_attempt_at"))
42                                .is_some_and(|due| due <= now.as_secs()))
43                })
44            })
45    }) || value
46        .get("job_executions")
47        .and_then(Value::as_array)
48        .is_some_and(|rows| {
49            rows.iter().any(|row| {
50                row.get("execution_status").and_then(Value::as_str) == Some("queued")
51                    && timestamp(row.get("available_at")).is_none_or(|due| due <= now.as_secs())
52            })
53        })
54}
55
56/// Holds the authoritative representation used by a subsequent compare-and-write.
57#[derive(Clone)]
58pub struct DocumentSnapshot {
59    pub value: Value,
60    original: Value,
61    entity: String,
62    field: String,
63    identity: Predicate,
64    kind: FieldType,
65}
66
67/// Reads exactly one declared document without exporting any other app entities.
68pub fn read_document(
69    adapter: &dyn DatabaseOperations,
70    config: &DbBackendConfig,
71    schema: &AppSchema,
72    entity: &str,
73    identity: Predicate,
74    field: &str,
75) -> Result<DocumentSnapshot, DatabaseError> {
76    let declaration = schema
77        .entities
78        .iter()
79        .find(|entry| entry.name == entity)
80        .ok_or(DatabaseError::InvalidContract)?;
81    let identity_terms = match &identity {
82        Predicate::Equal(_, _) => vec![&identity],
83        Predicate::All(terms) => terms.iter().collect(),
84        _ => return Err(DatabaseError::InvalidInput),
85    };
86    let mut identity_values = BTreeMap::new();
87    for term in identity_terms {
88        let Predicate::Equal(name, value) = term else {
89            return Err(DatabaseError::InvalidInput);
90        };
91        if value.is_null()
92            || value.is_array()
93            || value.is_object()
94            || identity_values.insert(name.as_str(), value).is_some()
95        {
96            return Err(DatabaseError::InvalidInput);
97        }
98    }
99    let identity_fields = declaration.identity.fields();
100    if identity_values.len() != identity_fields.len()
101        || identity_fields
102            .iter()
103            .any(|name| !identity_values.contains_key(name))
104    {
105        return Err(DatabaseError::InvalidInput);
106    }
107    let kind = declaration
108        .fields
109        .iter()
110        .find(|entry| entry.name == field)
111        .ok_or(DatabaseError::InvalidContract)?
112        .kind;
113    if !matches!(kind, FieldType::Text | FieldType::Json) {
114        return Err(DatabaseError::InvalidContract);
115    }
116    let mut rows = adapter.query(
117        config,
118        schema,
119        &Query {
120            entity: entity.to_owned(),
121            fields: vec![field.to_owned()],
122            filter: Some(identity.clone()),
123            order: Vec::new(),
124            limit: 2,
125            offset: 0,
126        },
127    )?;
128    if rows.len() != 1 {
129        return Err(if rows.is_empty() {
130            DatabaseError::MissingRecord
131        } else {
132            DatabaseError::Conflict
133        });
134    }
135    let original = rows
136        .pop()
137        .and_then(|mut row| row.remove(field))
138        .ok_or(DatabaseError::StorageFailure)?;
139    let value = match kind {
140        FieldType::Text => {
141            let text = original.as_str().ok_or(DatabaseError::StorageFailure)?;
142            if text.len() > MAX_DOCUMENT_BYTES {
143                return Err(DatabaseError::CapacityExceeded);
144            }
145            serde_json::from_str(text).map_err(|_| DatabaseError::StorageFailure)?
146        }
147        FieldType::Json => original.clone(),
148        _ => return Err(DatabaseError::InvalidContract),
149    };
150    if !value.is_object()
151        || serde_json::to_vec(&value)
152            .map_err(|_| DatabaseError::StorageFailure)?
153            .len()
154            > MAX_DOCUMENT_BYTES
155    {
156        return Err(DatabaseError::InvalidInput);
157    }
158    Ok(DocumentSnapshot {
159        value,
160        original,
161        entity: entity.to_owned(),
162        field: field.to_owned(),
163        identity,
164        kind,
165    })
166}
167
168/// Commits one changed document atomically and rejects a stale snapshot.
169pub fn write_document(
170    adapter: &dyn DatabaseOperations,
171    config: &DbBackendConfig,
172    schema: &AppSchema,
173    snapshot: &DocumentSnapshot,
174    value: &Value,
175) -> Result<(), DatabaseError> {
176    if live_worker_lease(&snapshot.value)? {
177        return Err(DatabaseError::Conflict);
178    }
179    if value.get("_database_worker_lease") != snapshot.value.get("_database_worker_lease") {
180        return Err(DatabaseError::InvalidInput);
181    }
182    compare_and_write(adapter, config, schema, snapshot, value)
183}
184
185/// Claims document-backed work before any external delivery. App writes reject
186/// an active lease, and a crashed worker releases ownership through expiry.
187pub fn claim_worker(
188    adapter: &dyn DatabaseOperations,
189    config: &DbBackendConfig,
190    schema: &AppSchema,
191    snapshot: &DocumentSnapshot,
192) -> Result<DocumentSnapshot, DatabaseError> {
193    if live_worker_lease(&snapshot.value)? {
194        return Err(DatabaseError::Conflict);
195    }
196    let now = std::time::SystemTime::now()
197        .duration_since(std::time::UNIX_EPOCH)
198        .map_err(|_| DatabaseError::Unavailable)?;
199    static NEXT_TOKEN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
200    let token = format!(
201        "{}:{}:{}",
202        std::process::id(),
203        now.as_nanos(),
204        NEXT_TOKEN.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
205    );
206    let mut value = snapshot.value.clone();
207    value
208        .as_object_mut()
209        .ok_or(DatabaseError::InvalidInput)?
210        .insert(
211            "_database_worker_lease".to_owned(),
212            serde_json::json!({"token": token, "expires_at": now.as_secs().saturating_add(120)}),
213        );
214    compare_and_write(adapter, config, schema, snapshot, &value)?;
215    let original = if snapshot.kind == FieldType::Text {
216        Value::String(serde_json::to_string(&value).map_err(|_| DatabaseError::InvalidInput)?)
217    } else {
218        value.clone()
219    };
220    Ok(DocumentSnapshot {
221        value,
222        original,
223        entity: snapshot.entity.clone(),
224        field: snapshot.field.clone(),
225        identity: snapshot.identity.clone(),
226        kind: snapshot.kind,
227    })
228}
229
230/// Completes only the claimed document version, atomically releasing its lease.
231pub fn complete_worker(
232    adapter: &dyn DatabaseOperations,
233    config: &DbBackendConfig,
234    schema: &AppSchema,
235    snapshot: &DocumentSnapshot,
236    value: &Value,
237) -> Result<(), DatabaseError> {
238    if !live_worker_lease(&snapshot.value)?
239        || value.get("_database_worker_lease") != snapshot.value.get("_database_worker_lease")
240    {
241        return Err(DatabaseError::Conflict);
242    }
243    let mut completed = value.clone();
244    completed
245        .as_object_mut()
246        .ok_or(DatabaseError::InvalidInput)?
247        .remove("_database_worker_lease");
248    compare_and_write(adapter, config, schema, snapshot, &completed)
249}
250
251/// Tests persisted lease ownership using the shared epoch-clock representation.
252fn live_worker_lease(value: &Value) -> Result<bool, DatabaseError> {
253    let Some(lease) = value.get("_database_worker_lease") else {
254        return Ok(false);
255    };
256    if lease
257        .get("token")
258        .and_then(Value::as_str)
259        .is_none_or(str::is_empty)
260    {
261        return Err(DatabaseError::StorageFailure);
262    }
263    let expiry = lease
264        .get("expires_at")
265        .and_then(Value::as_u64)
266        .ok_or(DatabaseError::StorageFailure)?;
267    let now = std::time::SystemTime::now()
268        .duration_since(std::time::UNIX_EPOCH)
269        .map_err(|_| DatabaseError::Unavailable)?
270        .as_secs();
271    Ok(expiry > now)
272}
273
274/// Writes the exact previously read representation through one guarded mutation.
275fn compare_and_write(
276    adapter: &dyn DatabaseOperations,
277    config: &DbBackendConfig,
278    schema: &AppSchema,
279    snapshot: &DocumentSnapshot,
280    value: &Value,
281) -> Result<(), DatabaseError> {
282    if !value.is_object() {
283        return Err(DatabaseError::InvalidInput);
284    }
285    let encoded = serde_json::to_string(value).map_err(|_| DatabaseError::InvalidInput)?;
286    if encoded.len() > MAX_DOCUMENT_BYTES {
287        return Err(DatabaseError::CapacityExceeded);
288    }
289    if value == &snapshot.value {
290        return Ok(());
291    }
292    let stored = if snapshot.kind == FieldType::Text {
293        Value::String(encoded)
294    } else {
295        value.clone()
296    };
297    adapter.transact(
298        config,
299        schema,
300        &Transaction {
301            steps: vec![Mutation::Update {
302                entity: snapshot.entity.clone(),
303                filter: Predicate::All(vec![
304                    snapshot.identity.clone(),
305                    Predicate::Equal(snapshot.field.clone(), snapshot.original.clone()),
306                ]),
307                values: BTreeMap::from([(snapshot.field.clone(), WriteValue::Literal(stored))]),
308                expected_rows: Some(1),
309            }],
310            idempotency_key: None,
311        },
312    )?;
313    Ok(())
314}
pharos_runtime_database/document.rs

pharos_runtime_database/
document.rs
