pharos_runtime_database/
snapshot.rs

1//! Portable validation and ordering for atomic operator snapshot restoration.
2
3use crate::operations::{
4    AppSchema, DatabaseError, Entity, FieldType, Mutation, Predicate, Record, SnapshotCollector,
5    SnapshotLimits, WriteValue, decimal_cents, entity, operation_fingerprint, validate_field_value,
6};
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use std::collections::{BTreeMap, BTreeSet};
10
11/// Default operator bounds keep bulk work outside request-sized transactions.
12pub const OPERATOR_LIMITS: SnapshotLimits = SnapshotLimits {
13    max_records: 100_000,
14    max_bytes: 64 * 1024 * 1024,
15};
16
17/// A backend-independent operator artifact tied to its logical migration contract.
18#[derive(Debug, Clone, Serialize, Deserialize)]
19#[serde(deny_unknown_fields)]
20pub struct DatabaseSnapshot {
21    pub format: String,
22    pub app_id: String,
23    pub contract_fingerprint: String,
24    pub records: Vec<Record>,
25}
26
27impl DatabaseSnapshot {
28    /// Captures one consistent record view after checking the selected migration ledger.
29    pub fn capture(
30        adapter: &dyn crate::operations::DatabaseOperations,
31        config: &crate::DbBackendConfig,
32        contract: &crate::contract::AppDataContract,
33    ) -> Result<Self, DatabaseError> {
34        crate::contract::validate_contract(contract)?;
35        adapter.verify_contract(config, contract)?;
36        let records = adapter.export_snapshot(config, &contract.schema, OPERATOR_LIMITS)?;
37        Ok(Self {
38            format: "pharos.database.snapshot.v1".to_owned(),
39            app_id: contract.schema.app_id.clone(),
40            contract_fingerprint: operation_fingerprint(contract)?,
41            records,
42        })
43    }
44
45    /// Checks artifact identity before replacing app records through the configured adapter.
46    pub fn restore(
47        &self,
48        adapter: &dyn crate::operations::DatabaseOperations,
49        config: &crate::DbBackendConfig,
50        contract: &crate::contract::AppDataContract,
51        counts: &[RestoreCount],
52    ) -> Result<(), DatabaseError> {
53        crate::contract::validate_contract(contract)?;
54        if self.format != "pharos.database.snapshot.v1"
55            || self.app_id != contract.schema.app_id
56            || self.contract_fingerprint != operation_fingerprint(contract)?
57        {
58            return Err(DatabaseError::InvalidContract);
59        }
60        adapter.verify_contract(config, contract)?;
61        adapter.restore_snapshot(
62            config,
63            &contract.schema,
64            &self.records,
65            OPERATOR_LIMITS,
66            counts,
67        )
68    }
69
70    /// Encodes a bounded portable archive without exposing connection configuration.
71    pub fn encode(&self) -> Result<String, DatabaseError> {
72        let text = serde_json::to_string(self).map_err(|_| DatabaseError::InvalidInput)?;
73        if text.len() as u64 > OPERATOR_LIMITS.max_bytes {
74            return Err(DatabaseError::CapacityExceeded);
75        }
76        Ok(text)
77    }
78
79    /// Rejects oversized or malformed input before allocating parsed record structures.
80    pub fn decode(bytes: &[u8]) -> Result<Self, DatabaseError> {
81        if bytes.len() as u64 > OPERATOR_LIMITS.max_bytes {
82            return Err(DatabaseError::CapacityExceeded);
83        }
84        serde_json::from_slice(bytes).map_err(|_| DatabaseError::InvalidInput)
85    }
86}
87
88/// An optional precondition checked against the transaction's current snapshot.
89#[derive(Debug, Clone)]
90pub struct RestoreCount {
91    pub entity: String,
92    pub expected: u64,
93}
94
95/// Rejects malformed or oversized restore input before an adapter clones it or opens storage.
96pub fn validate_restore_records(
97    schema: &AppSchema,
98    records: &[Record],
99    limits: SnapshotLimits,
100) -> Result<(), DatabaseError> {
101    validated_order(schema, records, limits).map(|_| ())
102}
103
104/// A restore plan may execute only in the transaction that read its current records.
105/// It preserves numeric identities and checks references before any write.
106pub fn restore_plan(
107    schema: &AppSchema,
108    current: &[Record],
109    restored: &[Record],
110    limits: SnapshotLimits,
111    counts: &[RestoreCount],
112) -> Result<Vec<Mutation>, DatabaseError> {
113    if counts.len() > schema.entities.len() {
114        return Err(DatabaseError::InvalidInput);
115    }
116    let mut checked = BTreeSet::new();
117    for condition in counts {
118        entity(schema, &condition.entity)?;
119        if !checked.insert(&condition.entity) {
120            return Err(DatabaseError::InvalidInput);
121        }
122        if current
123            .iter()
124            .filter(|record| record.entity == condition.entity)
125            .count() as u64
126            != condition.expected
127        {
128            return Err(DatabaseError::Conflict);
129        }
130    }
131    let current_order = validated_order(schema, current, limits)?;
132    let restored_order = validated_order(schema, restored, limits)?;
133    let mut steps = Vec::new();
134    // A restore replaces the complete validated snapshot inside the adapter's
135    // transaction. Remove old unique values before inserting replacements, so
136    // exchanged unique keys and referenced natural keys do not collide.
137    for declaration in &schema.entities {
138        let values = declaration
139            .fields
140            .iter()
141            .filter(|field| field.nullable && field.references.is_some())
142            .map(|field| (field.name.clone(), WriteValue::Literal(Value::Null)))
143            .collect::<BTreeMap<_, _>>();
144        if !values.is_empty() {
145            let identity = declaration
146                .identity
147                .fields()
148                .first()
149                .copied()
150                .ok_or(DatabaseError::InvalidContract)?
151                .to_owned();
152            steps.push(Mutation::Update {
153                entity: declaration.name.clone(),
154                filter: Predicate::NotEqual(identity, Value::Null),
155                values,
156                expected_rows: None,
157            });
158        }
159    }
160    for index in current_order.into_iter().rev() {
161        let record = &current[index];
162        steps.push(Mutation::Delete {
163            entity: record.entity.clone(),
164            filter: identity_filter(entity(schema, &record.entity)?, record)?,
165            expected_rows: Some(1),
166        });
167    }
168    for index in restored_order {
169        let record = &restored[index];
170        let declaration = entity(schema, &record.entity)?;
171        let values = declaration
172            .fields
173            .iter()
174            .map(|field| {
175                let value = if field.nullable && field.references.is_some() {
176                    Value::Null
177                } else {
178                    record
179                        .fields
180                        .get(&field.name)
181                        .cloned()
182                        .ok_or(DatabaseError::InvalidInput)?
183                };
184                Ok((field.name.clone(), WriteValue::Literal(value)))
185            })
186            .collect::<Result<BTreeMap<_, _>, DatabaseError>>()?;
187        steps.push(Mutation::Insert {
188            entity: record.entity.clone(),
189            values,
190        });
191    }
192    // Every restored parent now exists; restore nullable links, including cycles.
193    for record in restored {
194        let declaration = entity(schema, &record.entity)?;
195        let values = declaration
196            .fields
197            .iter()
198            .filter(|field| {
199                field.nullable
200                    && field.references.is_some()
201                    && record
202                        .fields
203                        .get(&field.name)
204                        .is_some_and(|value| !value.is_null())
205            })
206            .map(|field| {
207                Ok((
208                    field.name.clone(),
209                    WriteValue::Literal(
210                        record
211                            .fields
212                            .get(&field.name)
213                            .cloned()
214                            .ok_or(DatabaseError::InvalidInput)?,
215                    ),
216                ))
217            })
218            .collect::<Result<BTreeMap<_, _>, DatabaseError>>()?;
219        if !values.is_empty() {
220            steps.push(Mutation::Update {
221                entity: record.entity.clone(),
222                filter: identity_filter(declaration, record)?,
223                values,
224                expected_rows: Some(1),
225            });
226        }
227    }
228    Ok(steps)
229}
230
231/// Resolves a complete primary identity without accepting undeclared fields.
232fn identity_filter(declaration: &Entity, record: &Record) -> Result<Predicate, DatabaseError> {
233    let fields = declaration.identity.fields();
234    let filters = fields
235        .into_iter()
236        .map(|field| {
237            Ok(Predicate::Equal(
238                field.to_owned(),
239                record
240                    .fields
241                    .get(field)
242                    .cloned()
243                    .ok_or(DatabaseError::InvalidInput)?,
244            ))
245        })
246        .collect::<Result<Vec<_>, DatabaseError>>()?;
247    Ok(Predicate::All(filters))
248}
249
250/// Normalizes exact decimal and timestamp keys before matching logical references.
251fn value_key(kind: FieldType, value: &Value) -> Result<String, DatabaseError> {
252    match kind {
253        FieldType::Decimal => operation_fingerprint(&decimal_cents(value)?),
254        FieldType::Timestamp => {
255            let timestamp = chrono::DateTime::parse_from_rfc3339(
256                value.as_str().ok_or(DatabaseError::InvalidInput)?,
257            )
258            .map_err(|_| DatabaseError::InvalidInput)?;
259            operation_fingerprint(&(timestamp.timestamp(), timestamp.timestamp_subsec_nanos()))
260        }
261        _ => operation_fingerprint(value),
262    }
263}
264
265/// Validates bounded records, uniqueness, and references in O((rows + links) log rows).
266/// Nullable links can cycle; non-null dependencies must admit an insertion order.
267fn validated_order(
268    schema: &AppSchema,
269    records: &[Record],
270    limits: SnapshotLimits,
271) -> Result<Vec<usize>, DatabaseError> {
272    let _ = SnapshotCollector::new(schema, limits)?;
273    if records.len() > limits.max_records as usize {
274        return Err(DatabaseError::CapacityExceeded);
275    }
276    let mut bytes = 2_u64;
277    let mut keys = BTreeMap::new();
278    let mut identities = BTreeSet::new();
279    let mut unique_keys = BTreeSet::new();
280    let referenced = schema
281        .entities
282        .iter()
283        .flat_map(|declaration| declaration.fields.iter())
284        .filter_map(|field| field.references.as_ref())
285        .cloned()
286        .collect::<BTreeSet<_>>();
287    for (index, record) in records.iter().enumerate() {
288        let declaration = entity(schema, &record.entity)?;
289        if record.fields.len() != declaration.fields.len() {
290            return Err(DatabaseError::InvalidInput);
291        }
292        bytes = bytes
293            .checked_add(
294                serde_json::to_vec(record)
295                    .map_err(|_| DatabaseError::InvalidInput)?
296                    .len() as u64
297                    + 1,
298            )
299            .ok_or(DatabaseError::CapacityExceeded)?;
300        if bytes > limits.max_bytes {
301            return Err(DatabaseError::CapacityExceeded);
302        }
303        for field in &declaration.fields {
304            let value = record
305                .fields
306                .get(&field.name)
307                .ok_or(DatabaseError::InvalidInput)?;
308            validate_field_value(field, value)?;
309            if !value.is_null()
310                && referenced.contains(&(record.entity.clone(), field.name.clone()))
311                && keys
312                    .insert(
313                        (
314                            record.entity.clone(),
315                            field.name.clone(),
316                            value_key(field.kind, value)?,
317                        ),
318                        index,
319                    )
320                    .is_some()
321            {
322                return Err(DatabaseError::InvalidInput);
323            }
324        }
325        let identity = declaration
326            .identity
327            .fields()
328            .into_iter()
329            .map(|name| {
330                let field = declaration
331                    .fields
332                    .iter()
333                    .find(|field| field.name == name)
334                    .ok_or(DatabaseError::InvalidContract)?;
335                value_key(
336                    field.kind,
337                    record.fields.get(name).ok_or(DatabaseError::InvalidInput)?,
338                )
339            })
340            .collect::<Result<Vec<_>, DatabaseError>>()?;
341        if !identities.insert((record.entity.clone(), identity)) {
342            return Err(DatabaseError::InvalidInput);
343        }
344        for fields in &declaration.unique {
345            if fields
346                .iter()
347                .any(|name| record.fields.get(name).is_none_or(Value::is_null))
348            {
349                continue;
350            }
351            let key = fields
352                .iter()
353                .map(|name| {
354                    let field = declaration
355                        .fields
356                        .iter()
357                        .find(|field| field.name == *name)
358                        .ok_or(DatabaseError::InvalidContract)?;
359                    value_key(
360                        field.kind,
361                        record.fields.get(name).ok_or(DatabaseError::InvalidInput)?,
362                    )
363                })
364                .collect::<Result<Vec<_>, DatabaseError>>()?;
365            if !unique_keys.insert((record.entity.clone(), fields.clone(), key)) {
366                return Err(DatabaseError::InvalidInput);
367            }
368        }
369    }
370    let mut dependencies = vec![BTreeSet::new(); records.len()];
371    let mut children = vec![Vec::new(); records.len()];
372    for (index, record) in records.iter().enumerate() {
373        for field in &entity(schema, &record.entity)?.fields {
374            let Some((parent_entity, parent_field)) = &field.references else {
375                continue;
376            };
377            let value = record
378                .fields
379                .get(&field.name)
380                .ok_or(DatabaseError::InvalidInput)?;
381            if value.is_null() {
382                continue;
383            }
384            let parent = *keys
385                .get(&(
386                    parent_entity.clone(),
387                    parent_field.clone(),
388                    value_key(field.kind, value)?,
389                ))
390                .ok_or(DatabaseError::InvalidInput)?;
391            if !field.nullable && parent != index {
392                dependencies[index].insert(parent);
393            }
394        }
395        for parent in &dependencies[index] {
396            children[*parent].push(index);
397        }
398    }
399    let mut ready = dependencies
400        .iter()
401        .enumerate()
402        .filter_map(|(index, parents)| parents.is_empty().then_some(index))
403        .collect::<BTreeSet<_>>();
404    let mut order = Vec::with_capacity(records.len());
405    while let Some(index) = ready.pop_first() {
406        order.push(index);
407        for child in &children[index] {
408            dependencies[*child].remove(&index);
409            if dependencies[*child].is_empty() {
410                ready.insert(*child);
411            }
412        }
413    }
414    if order.len() != records.len() {
415        return Err(DatabaseError::InvalidInput);
416    }
417    Ok(order)
418}
pharos_runtime_database/snapshot.rs

pharos_runtime_database/
snapshot.rs
