git.iko.soy/reshape
on iko/more-quotes help sign in
reshape/src/lib.rs
523 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 = MigrationContext::new(migration_index, action_index);
214 result = action
215 .run(&ctx, db, &new_schema)
216 .with_context(|| format!("failed to {}", description));
218 if result.is_ok() {
219 action.update_schema(&ctx, &mut new_schema);
220 println!("{}", "done".green());
221 } else {
222 println!("{}", "failed".red());
223 break 'outer;
224 }
225 }
227 println!();
228 }
230 // If a migration failed, we abort all the migrations that were applied
231 if let Err(err) = result {
232 println!("A migration failed, aborting migrations that have already been applied");
234 // Set to the Aborting state. This is to ensure that the failed
235 // migration is fully aborted and nothing is left dangling.
236 // If the abort is interrupted for any reason, the user can try again
237 // by running `reshape migration abort`.
238 state.aborting(
239 remaining_migrations.clone(),
240 last_migration_index + 1,
241 last_action_index + 1,
242 );
244 // Abort will only
245 abort(db, state)?;
247 return Err(err);
248 }
250 // Create schema and views for migration
251 create_schema_for_migration(db, &target_migration, &new_schema)
252 .with_context(|| format!("failed to create schema for migration {}", target_migration))?;
254 // Update state once migrations have been performed
255 state.in_progress(remaining_migrations);
256 state.save(db).context("failed to save in-progress state")?;
258 println!("Migrations have been applied and the new schema is ready for use:");
259 println!(
260 " - Run '{}' from your application to use the latest schema",
261 schema_query_for_migration(&target_migration)
262 );
263 println!(
264 " - Run 'reshape migration complete' once your application has been updated and the previous schema is no longer in use"
265 );
266 Ok(())
269fn complete(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
270 // Make sure a migration is in progress
271 let (remaining_migrations, starting_migration_index, starting_action_index) = match state.clone() {
272 State::InProgress { migrations } => {
273 // Move into the Completing state. Once in this state,
274 // the migration can't be aborted and must be completed.
275 state.completing(migrations.clone(), 0, 0);
276 state.save(db).context("failed to save state")?;
278 (migrations, 0, 0)
279 },
280 State::Completing {
281 migrations,
282 current_migration_index,
283 current_action_index
284 } => (migrations, current_migration_index, current_action_index),
285 State::Aborting { .. } => {
286 return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape migration abort`."))
287 }
288 State::Applying { .. } => {
289 return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))
290 }
291 State::Idle => {
292 println!("No migration in progress");
293 return Ok(());
294 }
295 };
297 // Remove previous migration's schema
298 if let Some(current_migration) = &state::current_migration(db)? {
299 db.run(&format!(
300 "DROP SCHEMA IF EXISTS {} CASCADE",
301 schema_name_for_migration(current_migration)
302 ))
303 .context("failed to remove previous migration's schema")?;
304 }
306 for (migration_index, migration) in remaining_migrations.iter().enumerate() {
307 // Skip all the migrations which have already been completed
308 if migration_index < starting_migration_index {
309 continue;
310 }
312 println!("Completing '{}':", migration.name);
314 for (action_index, action) in migration.actions.iter().enumerate() {
315 // Skip all actions which have already been completed
316 if migration_index == starting_migration_index && action_index < starting_action_index {
317 continue;
318 }
320 let description = action.describe();
321 print!(" + {} ", description);
323 let ctx = MigrationContext::new(migration_index, action_index);
325 // Update state to indicate that this action has been completed.
326 // We won't save this new state until after the action has completed.
327 state.completing(
328 remaining_migrations.clone(),
329 migration_index + 1,
330 action_index + 1,
331 );
333 // This did_save check is necessary because of the borrow checker.
334 // The Transaction which might be returned from action.complete
335 // contains a mutable reference to self.db. We need the Transaction
336 // to be dropped before we can save the state using self.db instead,
337 // which we achieve here by limiting the lifetime of the Transaction
338 // with a new block.
339 let did_save = {
340 let result = action
341 .complete(&ctx, db)
342 .with_context(|| format!("failed to complete migration {}", migration.name))
343 .with_context(|| format!("failed to complete action: {}", description));
345 let maybe_transaction = match result {
346 Ok(maybe_transaction) => {
347 println!("{}", "done".green());
348 maybe_transaction
349 }
350 Err(e) => {
351 println!("{}", "failed".red());
352 return Err(e);
353 }
354 };
356 // Update state with which migrations and actions have been completed.
357 // Each action can create and return a transaction if they need atomicity.
358 // We use this transaction to update the state to ensure the action only completes.
359 // once.
360 // We want to use a single transaction for each action to keep the length of
361 // the transaction as short as possible. Wherever possible, we don't want to
362 // use a transaction at all.
363 if let Some(mut transaction) = maybe_transaction {
364 state
365 .save(&mut transaction)
366 .context("failed to save state after completing action")?;
367 transaction
368 .commit()
369 .context("failed to commit transaction")?;
371 true
372 } else {
373 false
374 }
375 };
377 // If the action didn't return a transaction we save the state normally instead
378 if !did_save {
379 state
380 .save(db)
381 .context("failed to save state after completing action")?;
382 }
383 }
385 println!();
386 }
388 // Remove helpers which are no longer in use
389 helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
391 state
392 .complete(db)
393 .context("failed to update state as completed")?;
395 Ok(())
398fn abort(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {
399 let (remaining_migrations, last_migration_index, last_action_index) = match state.clone() {
400 State::InProgress { migrations } | State::Applying { migrations } => {
401 // Set to the Aborting state. Once this is done, the migration has to
402 // be fully aborted and can't be completed.
403 state.aborting(migrations.clone(), 0, 0);
404 state.save(db)?;
406 (migrations, usize::MAX, usize::MAX)
407 }
408 State::Aborting {
409 migrations,
410 last_migration_index,
411 last_action_index,
412 } => (migrations, last_migration_index, last_action_index),
413 State::Completing { .. } => {
414 return Err(anyhow!("migration completion has already been started. Please run `reshape migration complete` again to finish it."));
415 }
416 State::Idle => {
417 println!("No migration is in progress");
418 return Ok(());
419 }
420 };
422 // Remove new migration's schema
423 let target_migration = remaining_migrations.last().unwrap().name.to_string();
424 let schema_name = schema_name_for_migration(&target_migration);
425 db.run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
426 .with_context(|| format!("failed to drop schema {}", schema_name))?;
428 // Abort all pending migrations
429 // Abort all migrations in reverse order
430 for (migration_index, migration) in remaining_migrations.iter().enumerate().rev() {
431 // Skip migrations which shouldn't be aborted
432 // The reason can be that they have already been aborted or that
433 // the migration was never applied in the first place.
434 if migration_index >= last_migration_index {
435 continue;
436 }
438 print!("Aborting '{}' ", migration.name);
440 for (action_index, action) in migration.actions.iter().enumerate().rev() {
441 // Skip actions which shouldn't be aborted
442 // The reason can be that they have already been aborted or that
443 // the action was never applied in the first place.
444 if migration_index == last_migration_index - 1 && action_index >= last_action_index {
445 continue;
446 }
448 let ctx = MigrationContext::new(migration_index, action_index);
449 action
450 .abort(&ctx, db)
451 .with_context(|| format!("failed to abort migration {}", migration.name))
452 .with_context(|| format!("failed to abort action: {}", action.describe()))?;
454 // Update state with which migrations and actions have been aborted.
455 // We don't need to run this in a transaction as aborts are idempotent.
456 state.aborting(remaining_migrations.to_vec(), migration_index, action_index);
457 state.save(db).context("failed to save state")?;
458 }
460 println!("{}", "done".green());
461 }
463 helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
465 *state = State::Idle;
466 state.save(db).context("failed to save state")?;
468 Ok(())
471fn create_schema_for_migration(
472 db: &mut DbConn,
473 migration_name: &str,
474 schema: &Schema,
475) -> anyhow::Result<()> {
476 // Create schema for migration
477 let schema_name = schema_name_for_migration(migration_name);
478 db.run(&format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name))
479 .with_context(|| {
480 format!(
481 "failed to create schema {} for migration {}",
482 schema_name, migration_name
483 )
484 })?;
486 // Create views inside schema
487 for table in schema.get_tables(db)? {
488 create_view_for_table(db, &table, &schema_name)?;
489 }
491 Ok(())
494fn create_view_for_table(db: &mut impl Conn, table: &Table, schema: &str) -> anyhow::Result<()> {
495 let select_columns: Vec<String> = table
496 .columns
497 .iter()
498 .map(|column| {
499 format!(
500 r#"
501 "{real_name}" AS "{alias}"
502 "#,
503 real_name = column.real_name,
504 alias = column.name,
505 )
506 })
507 .collect();
509 db.run(&format!(
510 r#"
511 CREATE OR REPLACE VIEW {schema}."{view_name}" AS
512 SELECT {columns}
513 FROM "{table_name}"
514 "#,
515 schema = schema,
516 table_name = table.real_name,
517 view_name = table.name,
518 columns = select_columns.join(","),
519 ))
520 .with_context(|| format!("failed to create view for table {}", table.name))?;
522 Ok(())