git.iko.soy/reshape
on main help sign in
reshape/src/lib.rs
526 lines 19 kB history raw
1use crate::{
2 migrations::{Migration, MigrationContext},
3 schema::Schema,
4};
6use anyhow::{anyhow, Context};
7use colored::*;
8use db::{Conn, DbConn, DbLocker};
9use postgres::Config;
10use schema::Table;
12mod db;
13mod helpers;
14pub mod migrations;
15mod schema;
16mod state;
18pub use crate::state::State;
20pub struct Reshape {
21 db: DbLocker,
24impl Reshape {
25 pub fn new(connection_string: &str) -> anyhow::Result<Reshape> {
26 let config: Config = connection_string.parse()?;
27 Self::new_with_config(&config)
28 }
30 pub fn new_with_options(
31 host: &str,
32 port: u16,
33 database: &str,
34 username: &str,
35 password: &str,
36 ) -> anyhow::Result<Reshape> {
37 let mut config = Config::new();
38 config
39 .host(host)
40 .port(port)
41 .user(username)
42 .dbname(database)
43 .password(password);
45 Self::new_with_config(&config)
46 }
48 fn new_with_config(config: &Config) -> anyhow::Result<Reshape> {
49 let db = DbLocker::connect(config)?;
50 Ok(Reshape { db })
51 }
53 pub fn migrate(
54 &mut self,
55 migrations: impl IntoIterator<Item = Migration>,
56 ) -> anyhow::Result<()> {
57 self.db.lock(|db| {
58 let mut state = State::load(db)?;
59 migrate(db, &mut state, migrations)
60 })
61 }
63 pub fn complete(&mut self) -> anyhow::Result<()> {
64 self.db.lock(|db| {
65 let mut state = State::load(db)?;
66 complete(db, &mut state)
67 })
68 }
70 pub fn abort(&mut self) -> anyhow::Result<()> {
71 self.db.lock(|db| {
72 let mut state = State::load(db)?;
73 abort(db, &mut state)
74 })
75 }
77 pub fn remove(&mut self) -> anyhow::Result<()> {
78 self.db.lock(|db| {
79 let mut state = State::load(db)?;
81 // Remove migration schemas and views
82 if let Some(current_migration) = &state::current_migration(db)? {
83 db.run(&format!(
84 "DROP SCHEMA IF EXISTS {} CASCADE",
85 schema_name_for_migration(current_migration)
86 ))?;
87 }
89 if let State::InProgress { migrations } = &state {
90 let target_migration = migrations.last().unwrap().name.to_string();
91 db.run(&format!(
92 "DROP SCHEMA IF EXISTS {} CASCADE",
93 schema_name_for_migration(&target_migration)
94 ))?;
95 }
97 // Remove all tables
98 let schema = Schema::new();
99 for table in schema.get_tables(db)? {
100 db.run(&format!(
101 r#"
102 DROP TABLE IF EXISTS "{}" CASCADE
103 "#,
104 table.real_name
105 ))?;
106 }
108 // Remove all enums
109 let enums: Vec<String> = db
110 .query("SELECT typname FROM pg_type WHERE typcategory = 'E'")?
111 .iter()
112 .map(|row| row.get("typname"))
113 .collect();
114 for enum_type in enums {
115 db.run(&format!("DROP TYPE {}", enum_type))?;
116 }
118 // Reset state
119 state.clear(db)?;
121 println!("Reshape and all data has been removed");
123 Ok(())
124 })
125 }
128pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
129 migrations
130 .last()
131 .map(|migration| schema_name_for_migration(&migration.name))
134pub fn schema_query_for_migration(migration_name: &str) -> String {
135 let schema_name = schema_name_for_migration(migration_name);
136 format!("SET search_path TO {}", schema_name)
139fn schema_name_for_migration(migration_name: &str) -> String {
140 format!("migration_{}", migration_name)
143fn migrate(
144 db: &mut DbConn,
145 state: &mut State,
146 migrations: impl IntoIterator<Item = Migration>,
147) -> anyhow::Result<()> {
148 // Make sure no migration is in progress
149 if let State::InProgress { .. } = &state {
150 println!(
151 "Migration already in progress, please complete using 'reshape migration complete'"
152 );
153 return Ok(());
154 }
156 if let State::Completing { .. } = &state {
157 println!(
158 "Migration already in progress and has started completion, please finish using 'reshape migration complete'"
159 );
160 return Ok(());
161 }
163 // Determine which migrations need to be applied by comparing the provided migrations
164 // with the already applied ones stored in the state. This will throw an error if the
165 // two sets of migrations don't agree, for example if a new migration has been added
166 // in between two existing ones.
167 let remaining_migrations = state::remaining_migrations(db, migrations)?;
168 if remaining_migrations.is_empty() {
169 println!("No migrations left to apply");
170 return Ok(());
171 }
173 // If we have already started applying some migrations we need to ensure that
174 // they are the same ones we want to apply now
175 if let State::Applying {
176 migrations: existing_migrations,
177 } = &state
178 {
179 if existing_migrations != &remaining_migrations {
180 return Err(anyhow!(
181 "a previous migration seems to have failed without cleaning up. Please run `reshape migration abort` and then run migrate again."
182 ));
183 }
184 }
186 // Move to the "Applying" state which is necessary as we can't run the migrations
187 // and state update as a single transaction. If a migration unexpectedly fails without
188 // automatically aborting, this state saves us from dangling migrations. It forces the user
189 // to either run migrate again (which works as all migrations are idempotent) or abort.
190 state.applying(remaining_migrations.clone());
191 state.save(db)?;
193 println!("Applying {} migrations\n", remaining_migrations.len());
195 let target_migration = remaining_migrations.last().unwrap().name.to_string();
196 helpers::set_up_helpers(db, &target_migration).context("failed to set up helpers")?;
198 let mut new_schema = Schema::new();
199 let mut last_migration_index = usize::MAX;
200 let mut last_action_index = usize::MAX;
201 let mut result: anyhow::Result<()> = Ok(());
203 'outer: for (migration_index, migration) in remaining_migrations.iter().enumerate() {
204 println!("Migrating '{}':", migration.name);
205 last_migration_index = migration_index;
207 for (action_index, action) in migration.actions.iter().enumerate() {
208 last_action_index = action_index;
210 let description = action.describe();
211 print!(" + {} ", description);
213 let ctx =
214 MigrationContext::new(migration_index, action_index, state::current_migration(db)?);
215 result = action
216 .run(&ctx, db, &new_schema)
217 .with_context(|| format!("failed to {}", description));
219 if result.is_ok() {
220 action.update_schema(&ctx, &mut new_schema);
221 println!("{}", "done".green());
222 } else {
223 println!("{}", "failed".red());
224 break 'outer;
225 }
226 }
228 println!();
229 }
231 // If a migration failed, we abort all the migrations that were applied
232 if let Err(err) = result {
233 println!("A migration failed, aborting migrations that have already been applied");
235 // Set to the Aborting state. This is to ensure that the failed
236 // migration is fully aborted and nothing is left dangling.
237 // If the abort is interrupted for any reason, the user can try again
238 // by running `reshape migration abort`.
239 state.aborting(
240 remaining_migrations.clone(),
241 last_migration_index + 1,
242 last_action_index + 1,
243 );
245 // Abort will only
246 abort(db, state)?;
248 return Err(err);
249 }
251 // Create schema and views for migration
252 create_schema_for_migration(db, &target_migration, &new_schema)
253 .with_context(|| format!("failed to create schema for migration {}", target_migration))?;
255 // Update state once migrations have been performed
256 state.in_progress(remaining_migrations);
257 state.save(db).context("failed to save in-progress state")?;
259 println!("Migrations have been applied and the new schema is ready for use:");
260 println!(
261 " - Run '{}' from your application to use the latest schema",
262 schema_query_for_migration(&target_migration)
263 );
264 println!(
265 " - Run 'reshape migration complete' once your application has been updated and the previous schema is no longer in use"
266 );
267 Ok(())
270fn complete(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
271 // Make sure a migration is in progress
272 let (remaining_migrations, starting_migration_index, starting_action_index) = match state.clone() {
273 State::InProgress { migrations } => {
274 // Move into the Completing state. Once in this state,
275 // the migration can't be aborted and must be completed.
276 state.completing(migrations.clone(), 0, 0);
277 state.save(db).context("failed to save state")?;
279 (migrations, 0, 0)
280 },
281 State::Completing {
282 migrations,
283 current_migration_index,
284 current_action_index
285 } => (migrations, current_migration_index, current_action_index),
286 State::Aborting { .. } => {
287 return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape migration abort`."))
288 }
289 State::Applying { .. } => {
290 return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))
291 }
292 State::Idle => {
293 println!("No migration in progress");
294 return Ok(());
295 }
296 };
298 // Remove previous migration's schema
299 if let Some(current_migration) = &state::current_migration(db)? {
300 db.run(&format!(
301 "DROP SCHEMA IF EXISTS {} CASCADE",
302 schema_name_for_migration(current_migration)
303 ))
304 .context("failed to remove previous migration's schema")?;
305 }
307 for (migration_index, migration) in remaining_migrations.iter().enumerate() {
308 // Skip all the migrations which have already been completed
309 if migration_index < starting_migration_index {
310 continue;
311 }
313 println!("Completing '{}':", migration.name);
315 for (action_index, action) in migration.actions.iter().enumerate() {
316 // Skip all actions which have already been completed
317 if migration_index == starting_migration_index && action_index < starting_action_index {
318 continue;
319 }
321 let description = action.describe();
322 print!(" + {} ", description);
324 let ctx =
325 MigrationContext::new(migration_index, action_index, state::current_migration(db)?);
327 // Update state to indicate that this action has been completed.
328 // We won't save this new state until after the action has completed.
329 state.completing(
330 remaining_migrations.clone(),
331 migration_index + 1,
332 action_index + 1,
333 );
335 // This did_save check is necessary because of the borrow checker.
336 // The Transaction which might be returned from action.complete
337 // contains a mutable reference to self.db. We need the Transaction
338 // to be dropped before we can save the state using self.db instead,
339 // which we achieve here by limiting the lifetime of the Transaction
340 // with a new block.
341 let did_save = {
342 let result = action
343 .complete(&ctx, db)
344 .with_context(|| format!("failed to complete migration {}", migration.name))
345 .with_context(|| format!("failed to complete action: {}", description));
347 let maybe_transaction = match result {
348 Ok(maybe_transaction) => {
349 println!("{}", "done".green());
350 maybe_transaction
351 }
352 Err(e) => {
353 println!("{}", "failed".red());
354 return Err(e);
355 }
356 };
358 // Update state with which migrations and actions have been completed.
359 // Each action can create and return a transaction if they need atomicity.
360 // We use this transaction to update the state to ensure the action only completes.
361 // once.
362 // We want to use a single transaction for each action to keep the length of
363 // the transaction as short as possible. Wherever possible, we don't want to
364 // use a transaction at all.
365 if let Some(mut transaction) = maybe_transaction {
366 state
367 .save(&mut transaction)
368 .context("failed to save state after completing action")?;
369 transaction
370 .commit()
371 .context("failed to commit transaction")?;
373 true
374 } else {
375 false
376 }
377 };
379 // If the action didn't return a transaction we save the state normally instead
380 if !did_save {
381 state
382 .save(db)
383 .context("failed to save state after completing action")?;
384 }
385 }
387 println!();
388 }
390 // Remove helpers which are no longer in use
391 helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
393 state
394 .complete(db)
395 .context("failed to update state as completed")?;
397 Ok(())
400fn abort(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
401 let (remaining_migrations, last_migration_index, last_action_index) = match state.clone() {
402 State::InProgress { migrations } | State::Applying { migrations } => {
403 // Set to the Aborting state. Once this is done, the migration has to
404 // be fully aborted and can't be completed.
405 state.aborting(migrations.clone(), 0, 0);
406 state.save(db)?;
408 (migrations, usize::MAX, usize::MAX)
409 }
410 State::Aborting {
411 migrations,
412 last_migration_index,
413 last_action_index,
414 } => (migrations, last_migration_index, last_action_index),
415 State::Completing { .. } => {
416 return Err(anyhow!("migration completion has already been started. Please run `reshape migration complete` again to finish it."));
417 }
418 State::Idle => {
419 println!("No migration is in progress");
420 return Ok(());
421 }
422 };
424 // Remove new migration's schema
425 let target_migration = remaining_migrations.last().unwrap().name.to_string();
426 let schema_name = schema_name_for_migration(&target_migration);
427 db.run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
428 .with_context(|| format!("failed to drop schema {}", schema_name))?;
430 // Abort all pending migrations
431 // Abort all migrations in reverse order
432 for (migration_index, migration) in remaining_migrations.iter().enumerate().rev() {
433 // Skip migrations which shouldn't be aborted
434 // The reason can be that they have already been aborted or that
435 // the migration was never applied in the first place.
436 if migration_index >= last_migration_index {
437 continue;
438 }
440 print!("Aborting '{}' ", migration.name);
442 for (action_index, action) in migration.actions.iter().enumerate().rev() {
443 // Skip actions which shouldn't be aborted
444 // The reason can be that they have already been aborted or that
445 // the action was never applied in the first place.
446 if migration_index == last_migration_index - 1 && action_index >= last_action_index {
447 continue;
448 }
450 let ctx =
451 MigrationContext::new(migration_index, action_index, state::current_migration(db)?);
452 action
453 .abort(&ctx, db)
454 .with_context(|| format!("failed to abort migration {}", migration.name))
455 .with_context(|| format!("failed to abort action: {}", action.describe()))?;
457 // Update state with which migrations and actions have been aborted.
458 // We don't need to run this in a transaction as aborts are idempotent.
459 state.aborting(remaining_migrations.to_vec(), migration_index, action_index);
460 state.save(db).context("failed to save state")?;
461 }
463 println!("{}", "done".green());
464 }
466 helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
468 *state = State::Idle;
469 state.save(db).context("failed to save state")?;
471 Ok(())
474fn create_schema_for_migration(
475 db: &mut DbConn,
476 migration_name: &str,
477 schema: &Schema,
478) -> anyhow::Result<()> {
479 // Create schema for migration
480 let schema_name = schema_name_for_migration(migration_name);
481 db.run(&format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name))
482 .with_context(|| {
483 format!(
484 "failed to create schema {} for migration {}",
485 schema_name, migration_name
486 )
487 })?;
489 // Create views inside schema
490 for table in schema.get_tables(db)? {
491 create_view_for_table(db, &table, &schema_name)?;
492 }
494 Ok(())
497fn create_view_for_table(db: &mut impl Conn, table: &Table, schema: &str) -> anyhow::Result<()> {
498 let select_columns: Vec<String> = table
499 .columns
500 .iter()
501 .map(|column| {
502 format!(
503 r#"
504 "{real_name}" AS "{alias}"
505 "#,
506 real_name = column.real_name,
507 alias = column.name,
508 )
509 })
510 .collect();
512 db.run(&format!(
513 r#"
514 CREATE OR REPLACE VIEW {schema}."{view_name}" AS
515 SELECT {columns}
516 FROM "{table_name}"
517 "#,
518 schema = schema,
519 table_name = table.real_name,
520 view_name = table.name,
521 columns = select_columns.join(","),
522 ))
523 .with_context(|| format!("failed to create view for table {}", table.name))?;
525 Ok(())