@@ -5,7 +5,7 @@ use crate::{
55
66use anyhow::{anyhow, Context};
77use colored::*;
8−use db::{Conn, DbConn, DbLocker};
8+use db::{Conn, DbConn};
99use postgres::Config;
1010use schema::Table;
1111
@@ -15,10 +15,11 @@ pub mod migrations;
1515mod schema;
1616mod state;
1717
18−pub use crate::state::State;
18+pub use crate::state::{State, Status};
1919
2020pub struct Reshape {
21− db: DbLocker,
21+ pub state: State,
22+ db: DbConn,
2223}
2324
2425impl Reshape {
@@ -30,7 +31,6 @@ impl Reshape {
3031 pub fn new_with_options(
3132 host: &str,
3233 port: u16,
33− database: &str,
3434 username: &str,
3535 password: &str,
3636 ) -> anyhow::Result<Reshape> {
@@ -39,488 +39,504 @@ impl Reshape {
3939 .host(host)
4040 .port(port)
4141 .user(username)
42− .dbname(database)
4342 .password(password);
4443
4544 Self::new_with_config(&config)
4645 }
4746
4847 fn new_with_config(config: &Config) -> anyhow::Result<Reshape> {
49− let db = DbLocker::connect(config)?;
50− Ok(Reshape { db })
51− }
52−
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− }
48+ let mut db = DbConn::connect(config)?;
49+ let state = State::load(&mut db);
6250
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− })
51+ Ok(Reshape { db, state })
6852 }
6953
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− }
76−
77− pub fn remove(&mut self) -> anyhow::Result<()> {
78− self.db.lock(|db| {
79− let mut state = State::load(db)?;
54+ pub fn migrate<T>(&mut self, migrations: T) -> anyhow::Result<()>
55+ where
56+ T: IntoIterator<Item = Migration>,
57+ {
58+ self.state = State::load(&mut self.db);
8059
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− }
60+ // Make sure no migration is in progress
61+ if let state::Status::InProgress { .. } = &self.state.status {
62+ println!("Migration already in progress, please complete using 'reshape complete'");
63+ return Ok(());
64+ }
8865
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− }
66+ if let state::Status::Completing { .. } = &self.state.status {
67+ println!(
68+ "Migration already in progress and has started completion, please finish using 'reshape complete'"
69+ );
70+ return Ok(());
71+ }
9672
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− }
73+ // Determine which migrations need to be applied by comparing the provided migrations
74+ // with the already applied ones stored in the state. This will throw an error if the
75+ // two sets of migrations don't agree, for example if a new migration has been added
76+ // in between two existing ones.
77+ let current_migration = &self.state.current_migration.clone();
78+ let remaining_migrations = self.state.get_remaining_migrations(migrations)?;
79+ if remaining_migrations.is_empty() {
80+ println!("No migrations left to apply");
81+ return Ok(());
82+ }
10783
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))?;
84+ // If we have already started applying some migrations we need to ensure that
85+ // they are the same ones we want to apply now
86+ if let state::Status::Applying {
87+ migrations: existing_migrations,
88+ } = &self.state.status
89+ {
90+ if existing_migrations != &remaining_migrations {
91+ return Err(anyhow!(
92+ "a previous migration seems to have failed without cleaning up. Please run `reshape abort` and then run migrate again."
93+ ));
11694 }
95+ }
11796
118− // Reset state
119− state.clear(db)?;
97+ // Move to the "Applying" state which is necessary as we can't run the migrations
98+ // and state update as a single transaction. If a migration unexpectedly fails without
99+ // automatically aborting, this state saves us from dangling migrations. It forces the user
100+ // to either run migrate again (which works as all migrations are idempotent) or abort.
101+ self.state.applying(remaining_migrations.clone());
102+ self.state.save(&mut self.db)?;
120103
121− println!("Reshape and all data has been removed");
104+ println!("Applying {} migrations\n", remaining_migrations.len());
122105
123− Ok(())
124− })
125− }
126−}
106+ helpers::set_up_helpers(&mut self.db, current_migration)
107+ .context("failed to set up helpers")?;
127108
128−pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
129− migrations
130− .last()
131− .map(|migration| schema_name_for_migration(&migration.name))
132−}
109+ let mut new_schema = Schema::new();
110+ let mut last_migration_index = usize::MAX;
111+ let mut last_action_index = usize::MAX;
112+ let mut result: anyhow::Result<()> = Ok(());
133113
134−pub 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)
137−}
114+ for (migration_index, migration) in remaining_migrations.iter().enumerate() {
115+ println!("Migrating '{}':", migration.name);
116+ last_migration_index = migration_index;
138117
139−fn schema_name_for_migration(migration_name: &str) -> String {
140− format!("migration_{}", migration_name)
141−}
118+ for (action_index, action) in migration.actions.iter().enumerate() {
119+ last_action_index = action_index;
142120
143−fn 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− }
121+ let description = action.describe();
122+ print!(" + {} ", description);
155123
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− }
124+ let ctx = MigrationContext::new(migration_index, action_index);
125+ result = action
126+ .run(&ctx, &mut self.db, &new_schema)
127+ .with_context(|| format!("failed to {}", description));
162128
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− }
129+ if result.is_ok() {
130+ action.update_schema(&ctx, &mut new_schema);
131+ println!("{}", "done".green());
132+ } else {
133+ println!("{}", "failed".red());
134+ break;
135+ }
136+ }
172137
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− ));
138+ println!();
183139 }
184− }
185140
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)?;
192−
193− println!("Applying {} migrations\n", remaining_migrations.len());
194−
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")?;
197−
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(());
202−
203− 'outer: for (migration_index, migration) in remaining_migrations.iter().enumerate() {
204− println!("Migrating '{}':", migration.name);
205− last_migration_index = migration_index;
206−
207− for (action_index, action) in migration.actions.iter().enumerate() {
208− last_action_index = action_index;
209−
210− let description = action.describe();
211− print!(" + {} ", description);
212−
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));
218−
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− }
141+ // If a migration failed, we abort all the migrations that were applied
142+ if let Err(err) = result {
143+ println!("A migration failed, aborting migrations that have already been applied");
227144
228− println!();
229− }
145+ // Set to the Aborting state. This is to ensure that the failed
146+ // migration is fully aborted and nothing is left dangling.
147+ // If the abort is interrupted for any reason, the user can try again
148+ // by running `reshape abort`.
149+ self.state.aborting(
150+ remaining_migrations.clone(),
151+ last_migration_index + 1,
152+ last_action_index + 1,
153+ );
230154
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");
155+ // Abort will only
156+ self.abort()?;
234157
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− );
158+ return Err(err);
159+ }
244160
245− // Abort will only
246− abort(db, state)?;
161+ // Create schema and views for migration
162+ let target_migration = remaining_migrations.last().unwrap().name.to_string();
163+ self.create_schema_for_migration(&target_migration, &new_schema)
164+ .with_context(|| {
165+ format!("failed to create schema for migration {}", target_migration)
166+ })?;
167+
168+ // Update state once migrations have been performed
169+ self.state.in_progress(remaining_migrations);
170+ self.state
171+ .save(&mut self.db)
172+ .context("failed to save in-progress state")?;
173+
174+ // If we started from a blank slate, we can finish the migration immediately
175+ if current_migration.is_none() {
176+ println!("Automatically completing migrations\n");
177+ self.complete_migration()
178+ .context("failed to automatically complete migrations")?;
179+
180+ println!("Migrations complete:");
181+ println!(
182+ " - Run '{}' from your application to use the latest schema",
183+ schema_query_for_migration(&target_migration)
184+ );
185+ } else {
186+ println!("Migrations have been applied and the new schema is ready for use:");
187+ println!(
188+ " - Run '{}' from your application to use the latest schema",
189+ schema_query_for_migration(&target_migration)
190+ );
191+ println!(
192+ " - Run 'reshape complete' once your application has been updated and the previous schema is no longer in use"
193+ );
194+ }
247195
248− return Err(err);
196+ Ok(())
249197 }
250198
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))?;
254−
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")?;
258−
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(())
268−}
269−
270−fn 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")?;
278−
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− };
297−
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− }
199+ pub fn complete_migration(&mut self) -> anyhow::Result<()> {
200+ // Make sure a migration is in progress
201+ let (remaining_migrations, starting_migration_index, starting_action_index) = match self.state.status.clone() {
202+ state::Status::InProgress { migrations } => {
203+ // Move into the Completing state. Once in this state,
204+ // the migration can't be aborted and must be completed.
205+ self.state.completing(migrations.clone(), 0, 0);
206+ self.state.save(&mut self.db).context("failed to save state")?;
207+
208+ (migrations, 0, 0)
209+ },
210+ state::Status::Completing {
211+ migrations,
212+ current_migration_index,
213+ current_action_index
214+ } => (migrations, current_migration_index, current_action_index),
215+ state::Status::Aborting { .. } => {
216+ return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape abort`."))
217+ }
218+ state::Status::Applying { .. } => {
219+ return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))
220+ }
221+ state::Status::Idle => {
222+ println!("No migration in progress");
223+ return Ok(());
224+ }
225+ };
306226
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;
227+ // Remove previous migration's schema
228+ if let Some(current_migration) = &self.state.current_migration {
229+ self.db
230+ .run(&format!(
231+ "DROP SCHEMA IF EXISTS {} CASCADE",
232+ schema_name_for_migration(current_migration)
233+ ))
234+ .context("failed to remove previous migration's schema")?;
311235 }
312236
313− println!("Completing '{}':", migration.name);
314−
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 {
237+ for (migration_index, migration) in remaining_migrations.iter().enumerate() {
238+ // Skip all the migrations which have already been completed
239+ if migration_index < starting_migration_index {
318240 continue;
319241 }
320242
321− let description = action.describe();
322− print!(" + {} ", description);
243+ println!("Completing '{}':", migration.name);
323244
324− let ctx =
325− MigrationContext::new(migration_index, action_index, state::current_migration(db)?);
245+ for (action_index, action) in migration.actions.iter().enumerate() {
246+ // Skip all actions which have already been completed
247+ if migration_index == starting_migration_index
248+ && action_index < starting_action_index
249+ {
250+ continue;
251+ }
326252
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− );
253+ let description = action.describe();
254+ print!(" + {} ", description);
334255
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));
346−
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);
256+ let ctx = MigrationContext::new(migration_index, action_index);
257+
258+ // Update state to indicate that this action has been completed.
259+ // We won't save this new state until after the action has completed.
260+ self.state.completing(
261+ remaining_migrations.clone(),
262+ migration_index + 1,
263+ action_index + 1,
264+ );
265+
266+ // This did_save check is necessary because of the borrow checker.
267+ // The Transaction which might be returned from action.complete
268+ // contains a mutable reference to self.db. We need the Transaction
269+ // to be dropped before we can save the state using self.db instead,
270+ // which we achieve here by limiting the lifetime of the Transaction
271+ // with a new block.
272+ let did_save = {
273+ let result = action
274+ .complete(&ctx, &mut self.db)
275+ .with_context(|| format!("failed to complete migration {}", migration.name))
276+ .with_context(|| format!("failed to complete action: {}", description));
277+
278+ let maybe_transaction = match result {
279+ Ok(maybe_transaction) => {
280+ println!("{}", "done".green());
281+ maybe_transaction
282+ }
283+ Err(e) => {
284+ println!("{}", "failed".red());
285+ return Err(e);
286+ }
287+ };
288+
289+ // Update state with which migrations and actions have been completed.
290+ // Each action can create and return a transaction if they need atomicity.
291+ // We use this transaction to update the state to ensure the action only completes.
292+ // once.
293+ // We want to use a single transaction for each action to keep the length of
294+ // the transaction as short as possible. Wherever possible, we don't want to
295+ // use a transaction at all.
296+ if let Some(mut transaction) = maybe_transaction {
297+ self.state
298+ .save(&mut transaction)
299+ .context("failed to save state after completing action")?;
300+ transaction
301+ .commit()
302+ .context("failed to commit transaction")?;
303+
304+ true
305+ } else {
306+ false
355307 }
356308 };
357309
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)
310+ // If the action didn't return a transaction we save the state normally instead
311+ if !did_save {
312+ self.state
313+ .save(&mut self.db)
368314 .context("failed to save state after completing action")?;
369− transaction
370− .commit()
371− .context("failed to commit transaction")?;
372−
373− true
374− } else {
375− false
376315 }
377− };
378−
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")?;
384316 }
317+
318+ println!();
385319 }
386320
387− println!();
321+ // Remove helpers which are no longer in use
322+ helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
323+
324+ self.state
325+ .complete()
326+ .context("failed to update state as completed")?;
327+ self.state
328+ .save(&mut self.db)
329+ .context("failed to save state after setting as completed")?;
330+
331+ Ok(())
388332 }
389333
390− // Remove helpers which are no longer in use
391− helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
334+ fn create_schema_for_migration(
335+ &mut self,
336+ migration_name: &str,
337+ schema: &Schema,
338+ ) -> anyhow::Result<()> {
339+ // Create schema for migration
340+ let schema_name = schema_name_for_migration(migration_name);
341+ self.db
342+ .run(&format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name))
343+ .with_context(|| {
344+ format!(
345+ "failed to create schema {} for migration {}",
346+ schema_name, migration_name
347+ )
348+ })?;
349+
350+ // Create views inside schema
351+ for table in schema.get_tables(&mut self.db)? {
352+ Self::create_view_for_table(&mut self.db, &table, &schema_name)?;
353+ }
354+
355+ Ok(())
356+ }
392357
393− state
394− .complete(db)
395− .context("failed to update state as completed")?;
358+ fn create_view_for_table(
359+ db: &mut impl Conn,
360+ table: &Table,
361+ schema: &str,
362+ ) -> anyhow::Result<()> {
363+ let select_columns: Vec<String> = table
364+ .columns
365+ .iter()
366+ .map(|column| {
367+ format!(
368+ r#"
369+ "{real_name}" AS "{alias}"
370+ "#,
371+ real_name = column.real_name,
372+ alias = column.name,
373+ )
374+ })
375+ .collect();
396376
397− Ok(())
398−}
377+ db.run(&format!(
378+ r#"
379+ CREATE OR REPLACE VIEW {schema}."{view_name}" AS
380+ SELECT {columns}
381+ FROM "{table_name}"
382+ "#,
383+ schema = schema,
384+ table_name = table.real_name,
385+ view_name = table.name,
386+ columns = select_columns.join(","),
387+ ))
388+ .with_context(|| format!("failed to create view for table {}", table.name))?;
399389
400−fn 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)?;
390+ Ok(())
391+ }
407392
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."));
393+ pub fn remove(&mut self) -> anyhow::Result<()> {
394+ // Remove migration schemas and views
395+ if let Some(current_migration) = &self.state.current_migration {
396+ self.db.run(&format!(
397+ "DROP SCHEMA IF EXISTS {} CASCADE",
398+ schema_name_for_migration(current_migration)
399+ ))?;
417400 }
418− State::Idle => {
419− println!("No migration is in progress");
420− return Ok(());
401+
402+ if let Status::InProgress { migrations } = &self.state.status {
403+ let target_migration = migrations.last().unwrap().name.to_string();
404+ self.db.run(&format!(
405+ "DROP SCHEMA IF EXISTS {} CASCADE",
406+ schema_name_for_migration(&target_migration)
407+ ))?;
421408 }
422− };
423−
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))?;
429−
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;
409+
410+ // Remove all tables
411+ let schema = Schema::new();
412+ for table in schema.get_tables(&mut self.db)? {
413+ self.db.run(&format!(
414+ r#"
415+ DROP TABLE IF EXISTS "{}" CASCADE
416+ "#,
417+ table.real_name
418+ ))?;
438419 }
439420
440− print!("Aborting '{}' ", migration.name);
421+ // Reset state
422+ self.state.clear(&mut self.db)?;
441423
442− for (action_index, action) in migration.actions.iter().enumerate().rev() {
443− // Skip actions which shouldn't be aborted
424+ println!("Reshape and all data has been removed");
425+
426+ Ok(())
427+ }
428+
429+ pub fn abort(&mut self) -> anyhow::Result<()> {
430+ let (remaining_migrations, last_migration_index, last_action_index) = match self
431+ .state
432+ .status
433+ .clone()
434+ {
435+ Status::InProgress { migrations } | Status::Applying { migrations } => {
436+ // Set to the Aborting state. Once this is done, the migration has to
437+ // be fully aborted and can't be completed.
438+ self.state.aborting(migrations.clone(), 0, 0);
439+ self.state.save(&mut self.db)?;
440+
441+ (migrations, usize::MAX, usize::MAX)
442+ }
443+ Status::Aborting {
444+ migrations,
445+ last_migration_index,
446+ last_action_index,
447+ } => (migrations, last_migration_index, last_action_index),
448+ Status::Completing { .. } => {
449+ return Err(anyhow!("Migration completion has already been started. Please run `reshape complete` again to finish it."));
450+ }
451+ Status::Idle => {
452+ println!("No migration is in progress");
453+ return Ok(());
454+ }
455+ };
456+
457+ // Remove new migration's schema
458+ let target_migration = remaining_migrations.last().unwrap().name.to_string();
459+ let schema_name = schema_name_for_migration(&target_migration);
460+ self.db
461+ .run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
462+ .with_context(|| format!("failed to drop schema {}", schema_name))?;
463+
464+ // Abort all pending migrations
465+ self.abort_migrations(
466+ &remaining_migrations,
467+ last_migration_index,
468+ last_action_index,
469+ )?;
470+
471+ helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
472+
473+ self.state.status = state::Status::Idle;
474+ self.state
475+ .save(&mut self.db)
476+ .context("failed to save state")?;
477+
478+ Ok(())
479+ }
480+
481+ fn abort_migrations(
482+ &mut self,
483+ migrations: &[Migration],
484+ upper_migration_index: usize,
485+ upper_action_index: usize,
486+ ) -> anyhow::Result<()> {
487+ // Abort all migrations in reverse order
488+ for (migration_index, migration) in migrations.iter().enumerate().rev() {
489+ // Skip migrations which shouldn't be aborted
444490 // 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 {
491+ // the migration was never applied in the first place.
492+ if migration_index >= upper_migration_index {
447493 continue;
448494 }
449495
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()))?;
456−
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− }
496+ print!("Aborting '{}' ", migration.name);
462497
463− println!("{}", "done".green());
464− }
498+ for (action_index, action) in migration.actions.iter().enumerate().rev() {
499+ // Skip actions which shouldn't be aborted
500+ // The reason can be that they have already been aborted or that
501+ // the action was never applied in the first place.
502+ if migration_index == upper_migration_index - 1
503+ && action_index >= upper_action_index
504+ {
505+ continue;
506+ }
465507
466− helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
508+ let ctx = MigrationContext::new(migration_index, action_index);
509+ action
510+ .abort(&ctx, &mut self.db)
511+ .with_context(|| format!("failed to abort migration {}", migration.name))
512+ .with_context(|| format!("failed to abort action: {}", action.describe()))?;
513+
514+ // Update state with which migrations and actions have been aborted.
515+ // We don't need to run this in a transaction as aborts are idempotent.
516+ self.state
517+ .aborting(migrations.to_vec(), migration_index, action_index);
518+ self.state
519+ .save(&mut self.db)
520+ .context("failed to save state")?;
521+ }
467522
468− *state = State::Idle;
469− state.save(db).context("failed to save state")?;
523+ println!("{}", "done".green());
524+ }
525+ Ok(())
526+ }
527+}
470528
471− Ok(())
529+pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
530+ migrations
531+ .last()
532+ .map(|migration| schema_name_for_migration(&migration.name))
472533}
473534
474−fn create_schema_for_migration(
475− db: &mut DbConn,
476− migration_name: &str,
477− schema: &Schema,
478−) -> anyhow::Result<()> {
479− // Create schema for migration
535+pub fn schema_query_for_migration(migration_name: &str) -> String {
480536 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− })?;
488−
489− // Create views inside schema
490− for table in schema.get_tables(db)? {
491− create_view_for_table(db, &table, &schema_name)?;
492− }
493−
494− Ok(())
537+ format!("SET search_path TO {}", schema_name)
495538}
496539
497−fn 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();
511−
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))?;
524−
525− Ok(())
540+fn schema_name_for_migration(migration_name: &str) -> String {
541+ format!("migration_{}", migration_name)
526542}