pharos_runtime_database/
historical.rs

1//! Bounded historical JSON adoption through the same logical field contract.
2
3use std::collections::BTreeMap;
4
5use serde_json::{Map, Value};
6
7use crate::operations::{
8    AppSchema, DatabaseError, FieldType, Query, SchemaChange, SortKey, WriteValue,
9    canonical_stored_timestamp, entity, validate_field_value,
10};
11
12/// Lowers validated historical rows using ordinary parameterized insert binding.
13pub fn insert_statements(
14    dialect: crate::relational::Dialect,
15    schema: &AppSchema,
16    change: &SchemaChange,
17    sources: Vec<Map<String, Value>>,
18) -> Result<Vec<crate::relational::Statement>, DatabaseError> {
19    let SchemaChange::CopyJsonRecords { target_entity, .. } = change else {
20        return Err(DatabaseError::InvalidContract);
21    };
22    copied_values(schema, change, sources)?
23        .into_iter()
24        .map(|values| {
25            crate::relational::compile_mutation(
26                dialect,
27                schema,
28                &crate::operations::Mutation::Insert {
29                    entity: target_entity.clone(),
30                    values,
31                },
32                &[],
33            )
34        })
35        .collect()
36}
37
38/// Reads historical source documents deterministically without exporting app state.
39pub fn source_query(schema: &AppSchema, change: &SchemaChange) -> Result<Query, DatabaseError> {
40    let SchemaChange::CopyJsonRecords {
41        source_entity,
42        source_field,
43        source_filter,
44        section,
45        target_entity,
46        fields,
47    } = change
48    else {
49        return Err(DatabaseError::InvalidContract);
50    };
51    let source = entity(schema, source_entity)?;
52    crate::contract::validate_json_record_copy(
53        source,
54        source_field,
55        source_filter,
56        section,
57        entity(schema, target_entity)?,
58        fields,
59    )?;
60    Ok(Query {
61        entity: source_entity.clone(),
62        fields: vec![source_field.clone()],
63        filter: Some(source_filter.clone()),
64        order: source
65            .identity
66            .fields()
67            .into_iter()
68            .map(|field| SortKey {
69                field: field.to_owned(),
70                descending: false,
71            })
72            .collect(),
73        limit: 257,
74        offset: 0,
75    })
76}
77
78/// Validates all copied records before adapters write them in the migration transaction.
79/// Byte arrays stay logical binary values; JSON arrays never become binary implicitly.
80pub fn copied_values(
81    schema: &AppSchema,
82    change: &SchemaChange,
83    sources: Vec<Map<String, Value>>,
84) -> Result<Vec<BTreeMap<String, WriteValue>>, DatabaseError> {
85    source_query(schema, change)?;
86    let SchemaChange::CopyJsonRecords {
87        source_field,
88        section,
89        target_entity,
90        fields,
91        ..
92    } = change
93    else {
94        return Err(DatabaseError::InvalidContract);
95    };
96    if sources.len() > 256 {
97        return Err(DatabaseError::CapacityExceeded);
98    }
99    let target = entity(schema, target_entity)?;
100    let mut bytes = 0_usize;
101    let mut output = Vec::new();
102    for mut source in sources {
103        let raw = source
104            .remove(source_field)
105            .ok_or(DatabaseError::InvalidContract)?;
106        let size = match &raw {
107            Value::String(text) => text.len(),
108            _ => serde_json::to_vec(&raw)
109                .map_err(|_| DatabaseError::InvalidInput)?
110                .len(),
111        };
112        bytes = bytes
113            .checked_add(size)
114            .ok_or(DatabaseError::CapacityExceeded)?;
115        if size > 8 * 1024 * 1024 || bytes > 32 * 1024 * 1024 {
116            return Err(DatabaseError::CapacityExceeded);
117        }
118        let value = match raw {
119            Value::String(text) => {
120                serde_json::from_str(&text).map_err(|_| DatabaseError::InvalidInput)?
121            }
122            other => other,
123        };
124        let Some(rows) = value.get(section) else {
125            continue;
126        };
127        let rows = rows.as_array().ok_or(DatabaseError::InvalidInput)?;
128        for row in rows {
129            if output.len() >= 10_000 {
130                return Err(DatabaseError::CapacityExceeded);
131            }
132            let row = row.as_object().ok_or(DatabaseError::InvalidInput)?;
133            let mut selected = BTreeMap::new();
134            for name in fields {
135                let Some(value) = row.get(name) else { continue };
136                let field = target
137                    .fields
138                    .iter()
139                    .find(|field| field.name == *name)
140                    .ok_or(DatabaseError::InvalidContract)?;
141                let value = if field.kind == FieldType::Timestamp {
142                    match value {
143                        Value::String(text) if text.is_empty() && field.nullable => Value::Null,
144                        Value::String(text) => Value::String(canonical_stored_timestamp(text)?),
145                        other => other.clone(),
146                    }
147                } else {
148                    value.clone()
149                };
150                validate_field_value(field, &value)?;
151                selected.insert(name.clone(), WriteValue::Literal(value));
152            }
153            if target
154                .identity
155                .fields()
156                .iter()
157                .any(|field| !selected.contains_key(*field))
158                || target.fields.iter().any(|field| {
159                    !field.nullable
160                        && field.default.is_none()
161                        && !selected.contains_key(&field.name)
162                })
163            {
164                return Err(DatabaseError::InvalidInput);
165            }
166            output.push(selected);
167        }
168    }
169    Ok(output)
170}
pharos_runtime_database/historical.rs

pharos_runtime_database/
historical.rs
