use thiserror::Error;
use crate::clarifier::{
Clarifier, ClarifyError, ClarifyResult, Decision, TypeResolution, non_type_decisions,
resolve_unknown_types,
};
use crate::dialects::{Dialect, DialectError};
use crate::diff::{DiffEngine, DiffError};
use crate::graphs::{GraphError, MigrationGraph};
use crate::migrations::Migration;
use crate::replay::{ReplayEngine, ReplaySources, compute_deps, deterministic_name_from_ops};
use crate::sql_plan::{SqlPlanError, SqlPlanRenderer};
use crate::states::{ReplayError, Schema};
#[derive(Debug, Error)]
pub enum OfflineError {
#[error(transparent)]
Graph(#[from] GraphError),
#[error(transparent)]
Diff(#[from] DiffError),
#[error(transparent)]
Clarifier(#[from] ClarifyError),
#[error("migration generation needs clarification input")]
NeedsInput(Vec<crate::clarifier::Clarification>),
#[error(transparent)]
Dialect(#[from] DialectError),
#[error(transparent)]
Replay(#[from] ReplayError),
#[error(transparent)]
SqlPlan(#[from] SqlPlanError),
#[error("schema validation failed: {0}")]
Schema(String),
}
#[derive(Debug, Clone)]
pub struct OfflinePlanner {
dialect: Dialect,
migrations: Vec<Migration>,
}
#[derive(Copy, Clone)]
pub struct EmbeddedMigrations {
pub files: &'static [(&'static str, &'static str)],
pub dir: &'static str,
pub children: &'static [(&'static str, &'static EmbeddedMigrations)],
}
impl OfflinePlanner {
pub fn new(dialect: Dialect) -> Self {
Self {
dialect,
migrations: Vec::new(),
}
}
pub fn from_migrations(mut self, migrations: Vec<Migration>) -> Self {
self.migrations = migrations;
self
}
pub fn replay(&self) -> Result<Schema, OfflineError> {
self.replay_with_sources()?
.schema
.prepare(self.dialect)
.map_err(|err| OfflineError::Schema(err.to_string()))
}
pub fn make_migration(
&self,
desired_schema: Schema,
decisions: &[Decision],
) -> Result<Option<Migration>, OfflineError> {
let replay = self.replay_with_sources()?;
let previous = replay
.schema
.prepare(self.dialect)
.map_err(|err| OfflineError::Schema(err.to_string()))?;
let desired_schema = desired_schema
.prepare(self.dialect)
.map_err(|err| OfflineError::Schema(err.to_string()))?;
let desired_schema =
match resolve_unknown_types(self.dialect, desired_schema, &previous, decisions)? {
TypeResolution::Resolved(schema) => schema,
TypeResolution::NeedsInput(clarifications) => {
return Err(OfflineError::NeedsInput(clarifications));
}
}
.prepare(self.dialect)
.map_err(|err| OfflineError::Schema(err.to_string()))?;
let raw_ops = DiffEngine::new().diff(&desired_schema, &previous, &self.dialect)?;
if raw_ops.is_empty() {
return Ok(None);
}
let op_decisions = non_type_decisions(decisions);
let ops = match Clarifier.process(&raw_ops, &op_decisions)? {
ClarifyResult::NeedsInput(clarifications) => {
return Err(OfflineError::NeedsInput(clarifications));
}
ClarifyResult::Resolved(ops) => ops,
};
let ops = self.dialect.reorder(ops, &previous, &desired_schema);
let id = format!(
"{:04}_{}",
self.next_number(),
deterministic_name_from_ops(&ops)
);
MigrationGraph::validate_id(&id)?;
Ok(Some(Migration {
id,
dependencies: compute_deps(&ops, &replay.last_per_ns, &replay.entity_ns),
operations: ops,
atomic: true,
}))
}
pub fn sql_migrate(&self, migrations: &[Migration]) -> Result<Vec<String>, OfflineError> {
Ok(SqlPlanRenderer::new(self.dialect, self.migrations.clone())?
.render_migrations(migrations)?)
}
fn graph(&self) -> Result<(MigrationGraph, Vec<String>), OfflineError> {
let mut graph = MigrationGraph::new();
for migration in self.migrations.clone() {
graph.add(migration)?;
}
let ordered_ids = graph
.topological_order()?
.into_iter()
.map(str::to_string)
.collect();
Ok((graph, ordered_ids))
}
fn next_number(&self) -> u32 {
self.migrations
.iter()
.filter(|migration| !migration.id.contains('/'))
.filter_map(|migration| migration.id.split('_').next()?.parse::<u32>().ok())
.max()
.map(|number| number + 1)
.unwrap_or(1)
}
fn replay_with_sources(&self) -> Result<ReplaySources, OfflineError> {
let (graph, ordered_ids) = self.graph()?;
Ok(ReplayEngine::new(&graph).replay_with_sources(&ordered_ids)?)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::clarifier::{Answer, ClarificationKind, Decision};
use crate::operations::Operation;
use crate::states::{Column, Schema, Table};
fn users_table(columns: Vec<Column>) -> Table {
Table {
name: "users".to_string(),
schema: None,
primary_key: None,
columns,
foreign_keys: vec![],
indexes: vec![],
constraints: vec![],
triggers: vec![],
options: Default::default(),
}
}
fn id_col() -> Column {
Column {
name: "id".to_string(),
col_type: "integer".to_string(),
nullable: false,
default: None,
primary_key: true,
references: None,
check: None,
generated: None,
}
}
fn email_col() -> Column {
Column {
name: "email".to_string(),
col_type: "text".to_string(),
nullable: true,
default: None,
primary_key: false,
references: None,
check: None,
generated: None,
}
}
fn col(name: &str, col_type: &str) -> Column {
Column {
name: name.to_string(),
col_type: col_type.to_string(),
nullable: true,
default: None,
primary_key: false,
references: None,
check: None,
generated: None,
}
}
#[test]
fn offline_planner_generates_migration_without_database_driver() {
let mut desired = Schema::default();
desired.tables.insert(
"users".to_string(),
users_table(vec![id_col(), email_col()]),
);
let migration = OfflinePlanner::new(Dialect::Postgres)
.make_migration(desired, &[])
.expect("offline planning should succeed")
.expect("migration should be generated");
assert_eq!(migration.id, "0001_users");
assert!(migration.dependencies.is_empty());
assert!(matches!(
migration.operations.as_slice(),
[Operation::CreateTable { table }] if table.name == "users"
));
}
#[test]
fn offline_replay_is_deterministic() {
let migration = Migration {
id: "0001_users".to_string(),
dependencies: vec![],
operations: vec![Operation::CreateTable {
table: users_table(vec![id_col()]),
}],
atomic: true,
};
let planner = OfflinePlanner::new(Dialect::Postgres).from_migrations(vec![migration]);
let first = planner.replay().expect("first replay should succeed");
let second = planner.replay().expect("second replay should succeed");
assert_eq!(first, second);
assert!(first.tables.contains_key("users"));
}
#[test]
fn migration_yaml_round_trips_from_strings() {
let migration = Migration {
id: "0001_users".to_string(),
dependencies: vec![],
operations: vec![Operation::CreateTable {
table: users_table(vec![id_col()]),
}],
atomic: true,
};
let yaml = migration.to_yaml_string().expect("serialize migration");
let mut parsed = Migration::from_yaml_str(&yaml).expect("parse migration");
parsed.id = migration.id.clone();
assert_eq!(parsed.id, migration.id);
assert_eq!(parsed.operations, migration.operations);
}
#[test]
fn make_migration_asks_for_new_unknown_type_before_diff() {
let mut desired = Schema::default();
desired.tables.insert(
"users".to_string(),
users_table(vec![id_col(), col("age", "intger")]),
);
let err = OfflinePlanner::new(Dialect::Postgres)
.make_migration(desired, &[])
.unwrap_err();
let OfflineError::NeedsInput(clarifications) = err else {
panic!("expected needs input");
};
assert_eq!(clarifications.len(), 1);
assert!(matches!(
&clarifications[0].kind,
ClarificationKind::UnknownType { type_name, suggested, .. }
if type_name == "intger" && suggested.contains(&"integer".to_string())
));
}
#[test]
fn make_migration_type_decision_rewrites_desired_schema() {
let mut desired = Schema::default();
desired.tables.insert(
"users".to_string(),
users_table(vec![id_col(), col("age", "intger")]),
);
let decisions = vec![Decision {
clarification_id: "unknown_type:users:age".to_string(),
answer: Answer::UseType("integer".to_string()),
}];
let migration = OfflinePlanner::new(Dialect::Postgres)
.make_migration(desired, &decisions)
.expect("planning should succeed")
.expect("migration should be generated");
let Operation::CreateTable { table } = &migration.operations[0] else {
panic!("expected create table");
};
assert_eq!(table.columns[1].col_type, "integer");
}
#[test]
fn make_migration_trusts_unknown_type_from_replay() {
let previous = Migration {
id: "0001_users".to_string(),
dependencies: vec![],
operations: vec![Operation::CreateTable {
table: users_table(vec![id_col(), col("code", "project_code")]),
}],
atomic: true,
};
let mut desired = Schema::default();
desired.tables.insert(
"users".to_string(),
users_table(vec![
id_col(),
col("code", "project_code"),
col("other_code", "project_code"),
]),
);
let migration = OfflinePlanner::new(Dialect::Postgres)
.from_migrations(vec![previous])
.make_migration(desired, &[])
.expect("trusted replay type should not ask")
.expect("migration should be generated");
assert!(matches!(
migration.operations.as_slice(),
[Operation::AddColumn { column, .. }] if column.col_type == "project_code"
));
}
#[test]
fn make_migration_accepts_known_extension_type_without_prompt() {
let mut desired = Schema::default();
desired.tables.insert(
"users".to_string(),
users_table(vec![id_col(), col("email", "citext")]),
);
let migration = OfflinePlanner::new(Dialect::Postgres)
.make_migration(desired, &[])
.expect("known extension type should not ask")
.expect("migration should be generated");
assert!(matches!(
migration.operations.as_slice(),
[Operation::CreateTable { table }] if table.columns[1].col_type == "citext"
));
}
}