pharos_runtime_database/
conversion.rs

1//! Lossless value conversion for guarded logical field migrations.
2//! Adapters use the same preflight before changing any physical representation.
3
4use crate::operations::{
5    AppSchema, DatabaseError, Field, FieldType, SchemaChange, canonical_stored_timestamp,
6    decimal_cents, decimal_from_cents, entity, validate_field_value, validate_schema,
7};
8use serde_json::Value;
9
10/// Reuses lossless conversion while separating an explicitly retired physical check
11/// from the logical values admitted before and after the migration.
12pub fn normalize_field_change(
13    schema: &AppSchema,
14    change: &SchemaChange,
15) -> Result<(SchemaChange, Option<Field>), DatabaseError> {
16    let SchemaChange::RetireFieldConstraint {
17        entity: name,
18        field: key,
19        constraint,
20    } = change
21    else {
22        return match change {
23            SchemaChange::ConvertField { .. } => Ok((change.clone(), None)),
24            _ => Err(DatabaseError::InvalidContract),
25        };
26    };
27    let declaration = entity(schema, name)?;
28    let current = declaration
29        .fields
30        .iter()
31        .find(|field| field.name == *key)
32        .ok_or(DatabaseError::InvalidContract)?;
33    if declaration.identity.fields().contains(&key.as_str())
34        || current.constraints.contains(constraint)
35    {
36        return Err(DatabaseError::InvalidContract);
37    }
38    let mut retired = current.clone();
39    retired.constraints.push(constraint.clone());
40    let mut probe = schema.clone();
41    *probe
42        .entities
43        .iter_mut()
44        .find(|entity| entity.name == *name)
45        .unwrap()
46        .fields
47        .iter_mut()
48        .find(|field| field.name == *key)
49        .unwrap() = retired.clone();
50    validate_schema(&probe)?;
51    Ok((
52        SchemaChange::ConvertField {
53            entity: name.clone(),
54            previous: current.clone(),
55            field: current.clone(),
56        },
57        Some(retired),
58    ))
59}
60
61/// Converts one admitted stored value, rejecting truncation, rounding and invented defaults.
62pub fn convert_value(
63    previous: &Field,
64    target: &Field,
65    value: &Value,
66) -> Result<Value, DatabaseError> {
67    validate_field_value(previous, value).map_err(|_| DatabaseError::Conflict)?;
68    if value.is_null() {
69        validate_field_value(target, value).map_err(|_| DatabaseError::Conflict)?;
70        return Ok(Value::Null);
71    }
72    let converted = if previous.kind == target.kind {
73        value.clone()
74    } else {
75        match target.kind {
76            FieldType::Text => Value::String(match previous.kind {
77                FieldType::Text | FieldType::Decimal | FieldType::Timestamp => {
78                    value.as_str().ok_or(DatabaseError::Conflict)?.to_owned()
79                }
80                FieldType::Binary => {
81                    let bytes = value
82                        .as_array()
83                        .ok_or(DatabaseError::Conflict)?
84                        .iter()
85                        .map(|v| {
86                            v.as_u64()
87                                .and_then(|v| u8::try_from(v).ok())
88                                .ok_or(DatabaseError::Conflict)
89                        })
90                        .collect::<Result<Vec<_>, _>>()?;
91                    String::from_utf8(bytes).map_err(|_| DatabaseError::Conflict)?
92                }
93                _ => serde_json::to_string(value).map_err(|_| DatabaseError::Conflict)?,
94            }),
95            FieldType::Integer => {
96                let integer = match previous.kind {
97                    FieldType::Text => value
98                        .as_str()
99                        .ok_or(DatabaseError::Conflict)?
100                        .parse::<i64>()
101                        .map_err(|_| DatabaseError::Conflict)?,
102                    FieldType::Decimal => {
103                        let cents = decimal_cents(value).map_err(|_| DatabaseError::Conflict)?;
104                        if cents % 100 != 0 {
105                            return Err(DatabaseError::Conflict);
106                        }
107                        cents / 100
108                    }
109                    FieldType::Boolean => {
110                        i64::from(value.as_bool().ok_or(DatabaseError::Conflict)?)
111                    }
112                    FieldType::Json => value.as_i64().ok_or(DatabaseError::Conflict)?,
113                    _ => return Err(DatabaseError::Conflict),
114                };
115                Value::from(integer)
116            }
117            FieldType::Decimal => {
118                let value = match previous.kind {
119                    FieldType::Integer | FieldType::Json => value.clone(),
120                    FieldType::Text => {
121                        Value::String(value.as_str().ok_or(DatabaseError::Conflict)?.to_owned())
122                    }
123                    FieldType::Boolean => {
124                        Value::from(i64::from(value.as_bool().ok_or(DatabaseError::Conflict)?))
125                    }
126                    _ => return Err(DatabaseError::Conflict),
127                };
128                Value::String(decimal_from_cents(
129                    decimal_cents(&value).map_err(|_| DatabaseError::Conflict)?,
130                ))
131            }
132            FieldType::Boolean => {
133                let boolean = match previous.kind {
134                    FieldType::Integer | FieldType::Json => match value {
135                        Value::Bool(value) => *value,
136                        _ => match value.as_i64() {
137                            Some(0) => false,
138                            Some(1) => true,
139                            _ => return Err(DatabaseError::Conflict),
140                        },
141                    },
142                    FieldType::Decimal => {
143                        match decimal_cents(value).map_err(|_| DatabaseError::Conflict)? {
144                            0 => false,
145                            100 => true,
146                            _ => return Err(DatabaseError::Conflict),
147                        }
148                    }
149                    FieldType::Text => match value.as_str().ok_or(DatabaseError::Conflict)? {
150                        "false" | "0" => false,
151                        "true" | "1" => true,
152                        _ => return Err(DatabaseError::Conflict),
153                    },
154                    _ => return Err(DatabaseError::Conflict),
155                };
156                Value::Bool(boolean)
157            }
158            FieldType::Timestamp => Value::String(
159                canonical_stored_timestamp(value.as_str().ok_or(DatabaseError::Conflict)?)
160                    .map_err(|_| DatabaseError::Conflict)?,
161            ),
162            FieldType::Json => match previous.kind {
163                FieldType::Text => {
164                    serde_json::from_str(value.as_str().ok_or(DatabaseError::Conflict)?)
165                        .map_err(|_| DatabaseError::Conflict)?
166                }
167                _ => value.clone(),
168            },
169            FieldType::Binary => match previous.kind {
170                FieldType::Text => Value::Array(
171                    value
172                        .as_str()
173                        .ok_or(DatabaseError::Conflict)?
174                        .bytes()
175                        .map(Value::from)
176                        .collect(),
177                ),
178                FieldType::Json => value.clone(),
179                _ => return Err(DatabaseError::Conflict),
180            },
181        }
182    };
183    validate_field_value(target, &converted).map_err(|_| DatabaseError::Conflict)?;
184    Ok(converted)
185}
186
187/// Validates a field transition and reconstructs the schema used to read historical values.
188pub fn previous_schema(
189    schema: &AppSchema,
190    change: &SchemaChange,
191) -> Result<AppSchema, DatabaseError> {
192    let SchemaChange::ConvertField {
193        entity: name,
194        previous,
195        field,
196    } = change
197    else {
198        return Err(DatabaseError::InvalidContract);
199    };
200    validate_schema(schema)?;
201    let declaration = entity(schema, name)?;
202    if previous.name != field.name
203        || declaration
204            .fields
205            .iter()
206            .find(|item| item.name == field.name)
207            != Some(field)
208        || declaration.identity.fields().contains(&field.name.as_str())
209        || previous.references != field.references
210        || previous.on_delete != field.on_delete
211        || (previous.kind != field.kind
212            && (previous.references.is_some()
213                || schema.entities.iter().any(|entity| {
214                    entity.fields.iter().any(|item| {
215                        item.references
216                            .as_ref()
217                            .is_some_and(|(parent, key)| parent == name && key == &field.name)
218                    })
219                })))
220    {
221        return Err(DatabaseError::InvalidContract);
222    }
223    let mut before = schema.clone();
224    *before
225        .entities
226        .iter_mut()
227        .find(|entity| entity.name == *name)
228        .unwrap()
229        .fields
230        .iter_mut()
231        .find(|item| item.name == field.name)
232        .unwrap() = previous.clone();
233    validate_schema(&before)?;
234    Ok(before)
235}
236
237/// Replays a declaration transition without changing earlier migration identities.
238pub fn apply_declaration(
239    declaration: &mut crate::operations::Entity,
240    previous: &Field,
241    field: &Field,
242) -> Result<(), DatabaseError> {
243    if previous.name != field.name || declaration.identity.fields().contains(&field.name.as_str()) {
244        return Err(DatabaseError::InvalidContract);
245    }
246    let stored = declaration
247        .fields
248        .iter_mut()
249        .find(|item| item.name == previous.name)
250        .ok_or(DatabaseError::InvalidContract)?;
251    if stored != previous {
252        return Err(DatabaseError::InvalidContract);
253    }
254    *stored = field.clone();
255    Ok(())
256}
257
258/// Names an internal staging entity deterministically so interrupted migration can resume.
259pub fn staging_name(change: &SchemaChange) -> Result<String, DatabaseError> {
260    let hash = crate::operations::operation_fingerprint(change)?;
261    Ok(format!("_convert_stage_{}", &hash[..24]))
262}
263
264/// Builds a complete target-row staging entity to validate fields and cross-field indexes.
265pub fn staging_schema(
266    schema: &AppSchema,
267    change: &SchemaChange,
268) -> Result<AppSchema, DatabaseError> {
269    previous_schema(schema, change)?;
270    let SchemaChange::ConvertField { entity: name, .. } = change else {
271        return Err(DatabaseError::InvalidContract);
272    };
273    let source = entity(schema, name)?;
274    let mut stage = source.clone();
275    stage.name = staging_name(change)?;
276    for item in &mut stage.fields {
277        item.references = None;
278        item.on_delete = None;
279        item.default = None;
280    }
281    // A complete target row validates cross-field unique/index guarantees before
282    // any adapter changes the populated source representation.
283    for (number, index) in stage.conditional_indexes.iter_mut().enumerate() {
284        index.name = format!("{}_index_{number}", stage.name);
285    }
286    let staged = AppSchema {
287        app_id: schema.app_id.clone(),
288        entities: vec![stage],
289    };
290    validate_schema(&staged)?;
291    Ok(staged)
292}
293
294/// Reads bounded conversion pages in stable identity order without exporting an app.
295pub fn query(
296    schema: &AppSchema,
297    change: &SchemaChange,
298    after: Option<Value>,
299) -> Result<crate::operations::Query, DatabaseError> {
300    let before = previous_schema(schema, change)?;
301    let SchemaChange::ConvertField { entity: name, .. } = change else {
302        return Err(DatabaseError::InvalidContract);
303    };
304    let declaration = entity(&before, name)?;
305    let (filter, order) = crate::operations::identity_page(declaration, after)?;
306    let fields = declaration
307        .fields
308        .iter()
309        .map(|field| field.name.clone())
310        .collect();
311    Ok(crate::operations::Query {
312        entity: name.clone(),
313        fields,
314        filter,
315        order,
316        limit: 1000,
317        offset: 0,
318    })
319}
320
321/// Stages typed values through ordinary safe parameter binding without mutating source records.
322pub fn statements(
323    dialect: crate::relational::Dialect,
324    schema: &AppSchema,
325    change: &SchemaChange,
326    rows: &[serde_json::Map<String, Value>],
327) -> Result<Vec<crate::relational::Statement>, DatabaseError> {
328    use crate::operations::{Mutation, WriteValue};
329    let staged = staging_schema(schema, change)?;
330    let SchemaChange::ConvertField {
331        previous, field, ..
332    } = change
333    else {
334        return Err(DatabaseError::InvalidContract);
335    };
336    let stage = &staged.entities[0];
337    rows.iter()
338        .map(|row| {
339            let old = row
340                .get(&previous.name)
341                .ok_or(DatabaseError::StorageFailure)?;
342            let value = convert_value(previous, field, old)?;
343            let mut values = stage
344                .fields
345                .iter()
346                .map(|item| {
347                    row.get(&item.name)
348                        .cloned()
349                        .map(|value| (item.name.clone(), WriteValue::Literal(value)))
350                        .ok_or(DatabaseError::StorageFailure)
351                })
352                .collect::<Result<std::collections::BTreeMap<_, _>, _>>()?;
353            values.insert(field.name.clone(), WriteValue::Literal(value));
354            crate::relational::compile_mutation(
355                dialect,
356                &staged,
357                &Mutation::Insert {
358                    entity: stage.name.clone(),
359                    values,
360                },
361                &[],
362            )
363        })
364        .collect()
365}
366
367/// Reconstructs the declared schema after one step, excluding later schema changes.
368/// Conversion must not inspect fields or tables that the same migration has not created yet.
369pub fn schema_after_step(
370    final_schema: &AppSchema,
371    migration: &crate::operations::Migration,
372    index: usize,
373) -> Result<AppSchema, DatabaseError> {
374    use crate::operations::Identity;
375    if migration.app_id != final_schema.app_id || index >= migration.changes.len() {
376        return Err(DatabaseError::InvalidContract);
377    }
378    let mut schema = final_schema.clone();
379    for change in migration.changes[index + 1..].iter().rev() {
380        match change {
381            SchemaChange::CreateEntity(created) => {
382                schema.entities.retain(|item| item.name != created.name)
383            }
384            SchemaChange::ConvertField {
385                entity: name,
386                previous,
387                field,
388            } => {
389                let declaration = schema
390                    .entities
391                    .iter_mut()
392                    .find(|item| item.name == *name)
393                    .ok_or(DatabaseError::InvalidContract)?;
394                let target = declaration
395                    .fields
396                    .iter_mut()
397                    .find(|item| item.name == field.name)
398                    .ok_or(DatabaseError::InvalidContract)?;
399                if target != field {
400                    return Err(DatabaseError::InvalidContract);
401                }
402                *target = previous.clone();
403            }
404            SchemaChange::AddField {
405                entity: name,
406                field,
407            } => schema
408                .entities
409                .iter_mut()
410                .find(|item| item.name == *name)
411                .ok_or(DatabaseError::InvalidContract)?
412                .fields
413                .retain(|item| item.name != field.name),
414            SchemaChange::AddUnique {
415                entity: name,
416                fields,
417            } => schema
418                .entities
419                .iter_mut()
420                .find(|item| item.name == *name)
421                .ok_or(DatabaseError::InvalidContract)?
422                .unique
423                .retain(|item| item != fields),
424            SchemaChange::AddIndex {
425                entity: name,
426                fields,
427            } => schema
428                .entities
429                .iter_mut()
430                .find(|item| item.name == *name)
431                .ok_or(DatabaseError::InvalidContract)?
432                .indexes
433                .retain(|item| item != fields),
434            SchemaChange::AddConditionalIndex {
435                entity: name,
436                index,
437            } => schema
438                .entities
439                .iter_mut()
440                .find(|item| item.name == *name)
441                .ok_or(DatabaseError::InvalidContract)?
442                .conditional_indexes
443                .retain(|item| item != index),
444            SchemaChange::WidenIntegerToDecimal {
445                entity: name,
446                field,
447            } => {
448                schema
449                    .entities
450                    .iter_mut()
451                    .find(|item| item.name == *name)
452                    .ok_or(DatabaseError::InvalidContract)?
453                    .fields
454                    .iter_mut()
455                    .find(|item| item.name == *field)
456                    .ok_or(DatabaseError::InvalidContract)?
457                    .kind = FieldType::Integer
458            }
459            SchemaChange::PromoteIdentity {
460                entity: name,
461                field,
462            } => {
463                schema
464                    .entities
465                    .iter_mut()
466                    .find(|item| item.name == *name)
467                    .ok_or(DatabaseError::InvalidContract)?
468                    .identity = Identity::Composite(vec![field.clone()])
469            }
470            SchemaChange::SeedFrom { .. }
471            | SchemaChange::RetireFieldConstraint { .. }
472            | SchemaChange::Seed { .. }
473            | SchemaChange::Backfill { .. }
474            | SchemaChange::CopyJsonRecords { .. }
475            | SchemaChange::CopyField { .. }
476            | SchemaChange::DeleteWhere { .. } => {}
477        }
478    }
479    validate_schema(&schema)?;
480    Ok(schema)
481}
482
483#[cfg(test)]
484mod tests {
485    use super::*;
486    use serde_json::json;
487
488    /// Creates an unconstrained portable field for conversion semantics.
489    fn field(kind: FieldType) -> Field {
490        Field {
491            name: "value".to_owned(),
492            kind,
493            nullable: true,
494            max_length: None,
495            default: None,
496            references: None,
497            on_delete: None,
498            constraints: Vec::new(),
499        }
500    }
501
502    /// Retirement cannot weaken a declared guarantee or reject values admitted by the current schema.
503    #[test]
504    fn retirement_checks_authority_without_narrowing_current_values() {
505        use crate::operations::{Entity, FieldConstraint, Identity};
506        let mut id = field(FieldType::Integer);
507        id.name = "id".into();
508        id.nullable = false;
509        let mut schema = AppSchema {
510            app_id: "retirement".into(),
511            entities: vec![Entity {
512                name: "records".into(),
513                identity: Identity::AutoInteger("id".into()),
514                fields: vec![id, field(FieldType::Text)],
515                unique: vec![],
516                indexes: vec![],
517                conditional_indexes: vec![],
518            }],
519        };
520        let constraint = FieldConstraint::Allowed(vec![json!("old")]);
521        let change = SchemaChange::RetireFieldConstraint {
522            entity: "records".into(),
523            field: "value".into(),
524            constraint: constraint.clone(),
525        };
526        let (normalized, retired) = normalize_field_change(&schema, &change).unwrap();
527        assert_eq!(retired.unwrap().constraints, vec![constraint.clone()]);
528        let SchemaChange::ConvertField {
529            previous, field, ..
530        } = normalized
531        else {
532            panic!("lossless conversion")
533        };
534        assert_eq!(
535            convert_value(&previous, &field, &json!("new")),
536            Ok(json!("new"))
537        );
538        schema.entities[0].fields[1].constraints.push(constraint);
539        assert_eq!(
540            normalize_field_change(&schema, &change),
541            Err(DatabaseError::InvalidContract)
542        );
543    }
544
545    #[test]
546    fn conversion_is_exact_and_rejects_invented_values() {
547        let cases = [
548            (FieldType::Text, FieldType::Integer, json!("42"), json!(42)),
549            (
550                FieldType::Integer,
551                FieldType::Decimal,
552                json!(42),
553                json!("42.00"),
554            ),
555            (
556                FieldType::Decimal,
557                FieldType::Integer,
558                json!("-42.00"),
559                json!(-42),
560            ),
561            (
562                FieldType::Text,
563                FieldType::Boolean,
564                json!("true"),
565                json!(true),
566            ),
567            (
568                FieldType::Boolean,
569                FieldType::Text,
570                json!(false),
571                json!("false"),
572            ),
573            (
574                FieldType::Text,
575                FieldType::Binary,
576                json!("é"),
577                json!([195, 169]),
578            ),
579            (
580                FieldType::Binary,
581                FieldType::Text,
582                json!([195, 169]),
583                json!("é"),
584            ),
585            (
586                FieldType::Text,
587                FieldType::Json,
588                json!("{\"count\":2}"),
589                json!({"count":2}),
590            ),
591            (
592                FieldType::Json,
593                FieldType::Text,
594                json!({"count":2}),
595                json!("{\"count\":2}"),
596            ),
597        ];
598        for (old, new, value, expected) in cases {
599            assert_eq!(
600                convert_value(&field(old), &field(new), &value),
601                Ok(expected)
602            );
603        }
604        for (old, new, value) in [
605            (FieldType::Text, FieldType::Integer, json!("")),
606            (
607                FieldType::Text,
608                FieldType::Integer,
609                json!("9223372036854775808"),
610            ),
611            (FieldType::Decimal, FieldType::Integer, json!("1.01")),
612            (FieldType::Text, FieldType::Decimal, json!("1.001")),
613            (FieldType::Integer, FieldType::Boolean, json!(2)),
614            (FieldType::Binary, FieldType::Text, json!([255])),
615            (FieldType::Text, FieldType::Json, json!("not JSON")),
616        ] {
617            assert_eq!(
618                convert_value(&field(old), &field(new), &value),
619                Err(DatabaseError::Conflict)
620            );
621        }
622    }
623
624    #[test]
625    fn conversion_applies_target_constraints_and_nullability() {
626        let mut target = field(FieldType::Text);
627        target.max_length = Some(2);
628        assert_eq!(
629            convert_value(&field(FieldType::Integer), &target, &json!(123)),
630            Err(DatabaseError::Conflict)
631        );
632        target.nullable = false;
633        assert_eq!(
634            convert_value(&field(FieldType::Text), &target, &Value::Null),
635            Err(DatabaseError::Conflict)
636        );
637        target.default = Some(crate::operations::FieldDefault::Literal(json!("ok")));
638        assert_eq!(
639            convert_value(&field(FieldType::Text), &target, &Value::Null),
640            Err(DatabaseError::Conflict)
641        );
642        target
643            .constraints
644            .push(crate::operations::FieldConstraint::Allowed(vec![json!(
645                "12"
646            )]));
647        assert_eq!(
648            convert_value(&field(FieldType::Integer), &target, &json!(12)),
649            Ok(json!("12"))
650        );
651        assert_eq!(
652            convert_value(&field(FieldType::Integer), &target, &json!(13)),
653            Err(DatabaseError::Conflict)
654        );
655    }
656}
pharos_runtime_database/conversion.rs

pharos_runtime_database/
conversion.rs
