gaman_core/
replay.rs

1//! Shared migration replay, source tracking, dependency, and naming helpers.
2
3use std::collections::{BTreeSet, HashMap};
4
5use crate::graphs::MigrationGraph;
6use crate::migrations::Migration;
7use crate::operations::Operation;
8use crate::states::types::EntityKind;
9use crate::states::{ReplayError, Schema};
10
11/// Schema replay output plus source metadata used to calculate dependencies.
12#[derive(Debug)]
13pub struct ReplaySources {
14    /// Schema after replaying the selected migration order.
15    pub schema: Schema,
16    /// Last migration id seen for each migration namespace.
17    pub last_per_ns: HashMap<String, String>,
18    /// Namespace that most recently produced each entity.
19    pub entity_ns: HashMap<(EntityKind, String), String>,
20}
21
22/// Replays migration graphs into schema state with contextual replay errors.
23#[derive(Debug)]
24pub struct ReplayEngine<'a> {
25    graph: &'a MigrationGraph,
26}
27
28impl<'a> ReplayEngine<'a> {
29    /// Create a replay helper over a migration graph.
30    pub fn new(graph: &'a MigrationGraph) -> Self {
31        Self { graph }
32    }
33
34    /// Replay the provided migration ids into schema state.
35    pub fn replay_ids(&self, ids: &[String]) -> Result<Schema, ReplayError> {
36        let mut schema = Schema::default();
37        for id in ids {
38            let migration = self
39                .graph
40                .get(id)
41                .ok_or_else(|| ReplayError::MigrationNotFound(id.clone()))?;
42            Self::apply_migration(&mut schema, migration)?;
43        }
44        Ok(schema)
45    }
46
47    /// Replay migration ids and return source metadata for dependency planning.
48    pub fn replay_with_sources(
49        &self,
50        ordered_ids: &[String],
51    ) -> Result<ReplaySources, ReplayError> {
52        let mut schema = Schema::default();
53        let mut last_per_ns = HashMap::new();
54        let mut entity_ns = HashMap::new();
55
56        for id in ordered_ids {
57            let migration = self
58                .graph
59                .get(id)
60                .ok_or_else(|| ReplayError::MigrationNotFound(id.clone()))?;
61            Self::apply_migration(&mut schema, migration)?;
62            let ns = namespace_of(id).to_string();
63            for entity in migration.get_entities() {
64                entity_ns.insert(entity, ns.clone());
65            }
66            last_per_ns.insert(ns, id.clone());
67        }
68
69        Ok(ReplaySources {
70            schema,
71            last_per_ns,
72            entity_ns,
73        })
74    }
75
76    /// Apply one migration to a schema with migration/operation error context.
77    pub fn apply_migration(schema: &mut Schema, migration: &Migration) -> Result<(), ReplayError> {
78        for (index, operation) in migration.operations.iter().enumerate() {
79            schema
80                .apply(operation)
81                .map_err(|inner| ReplayError::WithContext {
82                    migration: migration.id.clone(),
83                    op_num: index + 1,
84                    inner: Box::new(inner),
85                })?;
86        }
87        Ok(())
88    }
89}
90
91/// Return the namespace prefix of a migration id, or an empty string for root migrations.
92pub fn namespace_of(id: &str) -> &str {
93    match id.rfind('/') {
94        Some(pos) => &id[..pos],
95        None => "",
96    }
97}
98
99/// Compute migration dependencies from operation entity references and replay source metadata.
100pub fn compute_deps(
101    ops: &[Operation],
102    last_per_ns: &HashMap<String, String>,
103    entity_ns: &HashMap<(EntityKind, String), String>,
104) -> Vec<String> {
105    let mut namespaces = BTreeSet::new();
106    namespaces.insert(String::new());
107
108    for op in ops {
109        for entity in op_entities(op) {
110            if let Some(ns) = entity_ns.get(&entity) {
111                namespaces.insert(ns.clone());
112            }
113        }
114    }
115
116    namespaces
117        .iter()
118        .filter_map(|ns| last_per_ns.get(ns).cloned())
119        .collect()
120}
121
122/// Derive a deterministic migration name from the primary operation targets.
123pub fn deterministic_name_from_ops(ops: &[Operation]) -> String {
124    let mut labels: Vec<&str> = Vec::new();
125    for op in ops {
126        if let Some(label) = op.entity_label()
127            && !labels.contains(&label)
128        {
129            labels.push(label);
130        }
131    }
132    match labels.as_slice() {
133        [] => "changes".to_string(),
134        [a] => sanitize_id_part(a),
135        [a, b] => format!("{}_{}", sanitize_id_part(a), sanitize_id_part(b)),
136        [a, _, ..] => format!("{}_changes", sanitize_id_part(a)),
137    }
138}
139
140fn sanitize_id_part(value: &str) -> String {
141    let sanitized: String = value
142        .chars()
143        .map(|c| {
144            if c.is_ascii_lowercase() || c.is_ascii_digit() {
145                c
146            } else if c.is_ascii_alphabetic() {
147                c.to_ascii_lowercase()
148            } else {
149                '_'
150            }
151        })
152        .collect();
153    let trimmed = sanitized.trim_matches('_');
154    if trimmed.is_empty() {
155        "changes".to_string()
156    } else {
157        trimmed.to_string()
158    }
159}
160
161fn op_entities(op: &Operation) -> Vec<(EntityKind, String)> {
162    op.touched_entities()
163}
gaman_core/replay.rs

gaman_core/
replay.rs
