@@ -5,7 +5,7 @@ use crate::{
55
66use anyhow::{anyhow, Context};
77use colored::*;
8−use db::{Conn, DbConn};
8+use db::{Conn, DbConn, DbLocker};
99use postgres::Config;
1010use schema::Table;
1111
@@ -18,8 +18,7 @@ mod state;
1818pub use crate::state::State;
1919
2020pub struct Reshape {
21− pub state: State,
22− db: DbConn,
21+ db: DbLocker,
2322}
2423
2524impl Reshape {
@@ -45,494 +44,480 @@ impl Reshape {
4544 }
4645
4746 fn new_with_config(config: &Config) -> anyhow::Result<Reshape> {
48− let mut db = DbConn::connect(config)?;
49− let state = State::load(&mut db)?;
47+ let db = DbLocker::connect(config)?;
48+ Ok(Reshape { db })
49+ }
5050
51− Ok(Reshape { db, state })
51+ pub fn migrate(
52+ &mut self,
53+ migrations: impl IntoIterator<Item = Migration>,
54+ ) -> anyhow::Result<()> {
55+ self.db.lock(|db| {
56+ let mut state = State::load(db)?;
57+ migrate(db, &mut state, migrations)
58+ })
5259 }
5360
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)?;
61+ pub fn complete(&mut self) -> anyhow::Result<()> {
62+ self.db.lock(|db| {
63+ let mut state = State::load(db)?;
64+ complete(db, &mut state)
65+ })
66+ }
5967
60− // Make sure no migration is in progress
61− if let State::InProgress { .. } = &self.state {
62− println!("Migration already in progress, please complete using 'reshape complete'");
63− return Ok(());
64− }
68+ pub fn abort(&mut self) -> anyhow::Result<()> {
69+ self.db.lock(|db| {
70+ let mut state = State::load(db)?;
71+ abort(db, &mut state)
72+ })
73+ }
6574
66− if let State::Completing { .. } = &self.state {
67− println!(
68− "Migration already in progress and has started completion, please finish using 'reshape complete'"
69− );
70− return Ok(());
71− }
75+ pub fn remove(&mut self) -> anyhow::Result<()> {
76+ self.db.lock(|db| {
77+ let mut state = State::load(db)?;
7278
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 = state::current_migration(&mut self.db)?;
78− let remaining_migrations = state::remaining_migrations(&mut self.db, migrations)?;
79− if remaining_migrations.is_empty() {
80− println!("No migrations left to apply");
81− return Ok(());
82− }
79+ // Remove migration schemas and views
80+ if let Some(current_migration) = &state::current_migration(db)? {
81+ db.run(&format!(
82+ "DROP SCHEMA IF EXISTS {} CASCADE",
83+ schema_name_for_migration(current_migration)
84+ ))?;
85+ }
8386
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::Applying {
87− migrations: existing_migrations,
88− } = &self.state
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− ));
87+ if let State::InProgress { migrations } = &state {
88+ let target_migration = migrations.last().unwrap().name.to_string();
89+ db.run(&format!(
90+ "DROP SCHEMA IF EXISTS {} CASCADE",
91+ schema_name_for_migration(&target_migration)
92+ ))?;
9493 }
95− }
9694
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)?;
95+ // Remove all tables
96+ let schema = Schema::new();
97+ for table in schema.get_tables(db)? {
98+ db.run(&format!(
99+ r#"
100+ DROP TABLE IF EXISTS "{}" CASCADE
101+ "#,
102+ table.real_name
103+ ))?;
104+ }
103105
104− println!("Applying {} migrations\n", remaining_migrations.len());
106+ // Reset state
107+ state.clear(db)?;
105108
106− helpers::set_up_helpers(&mut self.db, ¤t_migration)
107− .context("failed to set up helpers")?;
109+ println!("Reshape and all data has been removed");
108110
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(());
111+ Ok(())
112+ })
113+ }
114+}
115+
116+pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
117+ migrations
118+ .last()
119+ .map(|migration| schema_name_for_migration(&migration.name))
120+}
113121
114− for (migration_index, migration) in remaining_migrations.iter().enumerate() {
115− println!("Migrating '{}':", migration.name);
116− last_migration_index = migration_index;
122+pub fn schema_query_for_migration(migration_name: &str) -> String {
123+ let schema_name = schema_name_for_migration(migration_name);
124+ format!("SET search_path TO {}", schema_name)
125+}
117126
118− for (action_index, action) in migration.actions.iter().enumerate() {
119− last_action_index = action_index;
127+fn schema_name_for_migration(migration_name: &str) -> String {
128+ format!("migration_{}", migration_name)
129+}
120130
121− let description = action.describe();
122− print!(" + {} ", description);
131+fn migrate(
132+ db: &mut DbConn,
133+ state: &mut State,
134+ migrations: impl IntoIterator<Item = Migration>,
135+) -> anyhow::Result<()> {
136+ // Make sure no migration is in progress
137+ if let State::InProgress { .. } = &state {
138+ println!("Migration already in progress, please complete using 'reshape complete'");
139+ return Ok(());
140+ }
123141
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));
142+ if let State::Completing { .. } = &state {
143+ println!(
144+ "Migration already in progress and has started completion, please finish using 'reshape complete'"
145+ );
146+ return Ok(());
147+ }
128148
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− }
149+ // Determine which migrations need to be applied by comparing the provided migrations
150+ // with the already applied ones stored in the state. This will throw an error if the
151+ // two sets of migrations don't agree, for example if a new migration has been added
152+ // in between two existing ones.
153+ let current_migration = state::current_migration(db)?;
154+ let remaining_migrations = state::remaining_migrations(db, migrations)?;
155+ if remaining_migrations.is_empty() {
156+ println!("No migrations left to apply");
157+ return Ok(());
158+ }
137159
138− println!();
160+ // If we have already started applying some migrations we need to ensure that
161+ // they are the same ones we want to apply now
162+ if let State::Applying {
163+ migrations: existing_migrations,
164+ } = &state
165+ {
166+ if existing_migrations != &remaining_migrations {
167+ return Err(anyhow!(
168+ "a previous migration seems to have failed without cleaning up. Please run `reshape abort` and then run migrate again."
169+ ));
139170 }
171+ }
140172
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");
173+ // Move to the "Applying" state which is necessary as we can't run the migrations
174+ // and state update as a single transaction. If a migration unexpectedly fails without
175+ // automatically aborting, this state saves us from dangling migrations. It forces the user
176+ // to either run migrate again (which works as all migrations are idempotent) or abort.
177+ state.applying(remaining_migrations.clone());
178+ state.save(db)?;
144179
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− );
180+ println!("Applying {} migrations\n", remaining_migrations.len());
154181
155− // Abort will only
156− self.abort()?;
182+ helpers::set_up_helpers(db, ¤t_migration).context("failed to set up helpers")?;
157183
158− return Err(err);
159− }
184+ let mut new_schema = Schema::new();
185+ let mut last_migration_index = usize::MAX;
186+ let mut last_action_index = usize::MAX;
187+ let mut result: anyhow::Result<()> = Ok(());
160188
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− }
189+ for (migration_index, migration) in remaining_migrations.iter().enumerate() {
190+ println!("Migrating '{}':", migration.name);
191+ last_migration_index = migration_index;
195192
196− Ok(())
197− }
193+ for (action_index, action) in migration.actions.iter().enumerate() {
194+ last_action_index = action_index;
198195
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.clone() {
202− State::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::Completing {
211− migrations,
212− current_migration_index,
213− current_action_index
214− } => (migrations, current_migration_index, current_action_index),
215− State::Aborting { .. } => {
216− return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape abort`."))
217− }
218− State::Applying { .. } => {
219− return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))
220− }
221− State::Idle => {
222− println!("No migration in progress");
223− return Ok(());
224− }
225− };
196+ let description = action.describe();
197+ print!(" + {} ", description);
226198
227− // Remove previous migration's schema
228− if let Some(current_migration) = &state::current_migration(&mut self.db)? {
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")?;
235− }
199+ let ctx = MigrationContext::new(migration_index, action_index);
200+ result = action
201+ .run(&ctx, db, &new_schema)
202+ .with_context(|| format!("failed to {}", description));
236203
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 {
240− continue;
204+ if result.is_ok() {
205+ action.update_schema(&ctx, &mut new_schema);
206+ println!("{}", "done".green());
207+ } else {
208+ println!("{}", "failed".red());
209+ break;
241210 }
211+ }
242212
243− println!("Completing '{}':", migration.name);
213+ println!();
214+ }
244215
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− }
216+ // If a migration failed, we abort all the migrations that were applied
217+ if let Err(err) = result {
218+ println!("A migration failed, aborting migrations that have already been applied");
252219
253− let description = action.describe();
254− print!(" + {} ", description);
220+ // Set to the Aborting state. This is to ensure that the failed
221+ // migration is fully aborted and nothing is left dangling.
222+ // If the abort is interrupted for any reason, the user can try again
223+ // by running `reshape abort`.
224+ state.aborting(
225+ remaining_migrations.clone(),
226+ last_migration_index + 1,
227+ last_action_index + 1,
228+ );
255229
256− let ctx = MigrationContext::new(migration_index, action_index);
230+ // Abort will only
231+ abort(db, state)?;
257232
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,
233+ return Err(err);
234+ }
235+
236+ // Create schema and views for migration
237+ let target_migration = remaining_migrations.last().unwrap().name.to_string();
238+ create_schema_for_migration(db, &target_migration, &new_schema)
239+ .with_context(|| format!("failed to create schema for migration {}", target_migration))?;
240+
241+ // Update state once migrations have been performed
242+ state.in_progress(remaining_migrations);
243+ state.save(db).context("failed to save in-progress state")?;
244+
245+ // If we started from a blank slate, we can finish the migration immediately
246+ if current_migration.is_none() {
247+ println!("Automatically completing migrations\n");
248+ complete(db, state).context("failed to automatically complete migrations")?;
249+
250+ println!("Migrations complete:");
251+ println!(
252+ " - Run '{}' from your application to use the latest schema",
253+ schema_query_for_migration(&target_migration)
254+ );
255+ } else {
256+ println!("Migrations have been applied and the new schema is ready for use:");
257+ println!(
258+ " - Run '{}' from your application to use the latest schema",
259+ schema_query_for_migration(&target_migration)
260+ );
261+ println!(
262+ " - Run 'reshape complete' once your application has been updated and the previous schema is no longer in use"
264263 );
264+ }
265265
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
307− }
308− };
266+ Ok(())
267+}
309268
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)
314− .context("failed to save state after completing action")?;
269+fn 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")?;
277+
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 abort`."))
315287 }
316− }
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+ };
317296
318− println!();
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+ }
305+
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;
319310 }
320311
321− // Remove helpers which are no longer in use
322− helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
312+ println!("Completing '{}':", migration.name);
323313
324− self.state
325− .complete(&mut self.db)
326− .context("failed to update state as completed")?;
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+ }
327319
328− Ok(())
329− }
320+ let description = action.describe();
321+ print!(" + {} ", description);
330322
331− fn create_schema_for_migration(
332− &mut self,
333− migration_name: &str,
334− schema: &Schema,
335− ) -> anyhow::Result<()> {
336− // Create schema for migration
337− let schema_name = schema_name_for_migration(migration_name);
338− self.db
339− .run(&format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name))
340− .with_context(|| {
341− format!(
342− "failed to create schema {} for migration {}",
343− schema_name, migration_name
344− )
345− })?;
346−
347− // Create views inside schema
348− for table in schema.get_tables(&mut self.db)? {
349− Self::create_view_for_table(&mut self.db, &table, &schema_name)?;
350− }
323+ let ctx = MigrationContext::new(migration_index, action_index);
351324
352− Ok(())
353− }
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+ );
354332
355− fn create_view_for_table(
356− db: &mut impl Conn,
357− table: &Table,
358− schema: &str,
359− ) -> anyhow::Result<()> {
360− let select_columns: Vec<String> = table
361− .columns
362− .iter()
363− .map(|column| {
364− format!(
365− r#"
366− "{real_name}" AS "{alias}"
367− "#,
368− real_name = column.real_name,
369− alias = column.name,
370− )
371− })
372− .collect();
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));
344+
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+ };
373355
374− db.run(&format!(
375− r#"
376− CREATE OR REPLACE VIEW {schema}."{view_name}" AS
377− SELECT {columns}
378− FROM "{table_name}"
379− "#,
380− schema = schema,
381− table_name = table.real_name,
382− view_name = table.name,
383− columns = select_columns.join(","),
384− ))
385− .with_context(|| format!("failed to create view for table {}", table.name))?;
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")?;
386370
387− Ok(())
388− }
371+ true
372+ } else {
373+ false
374+ }
375+ };
389376
390− pub fn remove(&mut self) -> anyhow::Result<()> {
391− // Remove migration schemas and views
392− if let Some(current_migration) = &state::current_migration(&mut self.db)? {
393− self.db.run(&format!(
394− "DROP SCHEMA IF EXISTS {} CASCADE",
395− schema_name_for_migration(current_migration)
396− ))?;
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+ }
397383 }
398384
399− if let State::InProgress { migrations } = &self.state {
400− let target_migration = migrations.last().unwrap().name.to_string();
401− self.db.run(&format!(
402− "DROP SCHEMA IF EXISTS {} CASCADE",
403− schema_name_for_migration(&target_migration)
404− ))?;
405− }
385+ println!();
386+ }
406387
407− // Remove all tables
408− let schema = Schema::new();
409− for table in schema.get_tables(&mut self.db)? {
410− self.db.run(&format!(
411− r#"
412− DROP TABLE IF EXISTS "{}" CASCADE
413− "#,
414− table.real_name
415− ))?;
416− }
388+ // Remove helpers which are no longer in use
389+ helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
417390
418− // Reset state
419− self.state.clear(&mut self.db)?;
391+ state
392+ .complete(db)
393+ .context("failed to update state as completed")?;
420394
421− println!("Reshape and all data has been removed");
395+ Ok(())
396+}
422397
423− Ok(())
424− }
398+fn 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)?;
425405
426− pub fn abort(&mut self) -> anyhow::Result<()> {
427− let (remaining_migrations, last_migration_index, last_action_index) = match self
428− .state
429− .clone()
430− {
431− State::InProgress { migrations } | State::Applying { migrations } => {
432− // Set to the Aborting state. Once this is done, the migration has to
433− // be fully aborted and can't be completed.
434− self.state.aborting(migrations.clone(), 0, 0);
435− self.state.save(&mut self.db)?;
436−
437− (migrations, usize::MAX, usize::MAX)
438− }
439− State::Aborting {
440− migrations,
441− last_migration_index,
442− last_action_index,
443− } => (migrations, last_migration_index, last_action_index),
444− State::Completing { .. } => {
445− return Err(anyhow!("Migration completion has already been started. Please run `reshape complete` again to finish it."));
446− }
447− State::Idle => {
448− println!("No migration is in progress");
449− return Ok(());
450− }
451− };
452−
453− // Remove new migration's schema
454− let target_migration = remaining_migrations.last().unwrap().name.to_string();
455− let schema_name = schema_name_for_migration(&target_migration);
456− self.db
457− .run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
458− .with_context(|| format!("failed to drop schema {}", schema_name))?;
459−
460− // Abort all pending migrations
461− self.abort_migrations(
462− &remaining_migrations,
406+ (migrations, usize::MAX, usize::MAX)
407+ }
408+ State::Aborting {
409+ migrations,
463410 last_migration_index,
464411 last_action_index,
465− )?;
466−
467− helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
468−
469− self.state = State::Idle;
470− self.state
471− .save(&mut self.db)
472− .context("failed to save state")?;
412+ } => (migrations, last_migration_index, last_action_index),
413+ State::Completing { .. } => {
414+ return Err(anyhow!("Migration completion has already been started. Please run `reshape complete` again to finish it."));
415+ }
416+ State::Idle => {
417+ println!("No migration is in progress");
418+ return Ok(());
419+ }
420+ };
421+
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))?;
427+
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+ }
473437
474− Ok(())
475− }
438+ print!("Aborting '{}' ", migration.name);
476439
477− fn abort_migrations(
478− &mut self,
479− migrations: &[Migration],
480− upper_migration_index: usize,
481− upper_action_index: usize,
482− ) -> anyhow::Result<()> {
483− // Abort all migrations in reverse order
484− for (migration_index, migration) in migrations.iter().enumerate().rev() {
485− // Skip migrations which shouldn't be aborted
440+ for (action_index, action) in migration.actions.iter().enumerate().rev() {
441+ // Skip actions which shouldn't be aborted
486442 // The reason can be that they have already been aborted or that
487− // the migration was never applied in the first place.
488− if migration_index >= upper_migration_index {
443+ // the action was never applied in the first place.
444+ if migration_index == last_migration_index - 1 && action_index >= last_action_index {
489445 continue;
490446 }
491447
492− print!("Aborting '{}' ", migration.name);
493−
494− for (action_index, action) in migration.actions.iter().enumerate().rev() {
495− // Skip actions which shouldn't be aborted
496− // The reason can be that they have already been aborted or that
497− // the action was never applied in the first place.
498− if migration_index == upper_migration_index - 1
499− && action_index >= upper_action_index
500− {
501− continue;
502− }
503−
504− let ctx = MigrationContext::new(migration_index, action_index);
505− action
506− .abort(&ctx, &mut self.db)
507− .with_context(|| format!("failed to abort migration {}", migration.name))
508− .with_context(|| format!("failed to abort action: {}", action.describe()))?;
509−
510− // Update state with which migrations and actions have been aborted.
511− // We don't need to run this in a transaction as aborts are idempotent.
512− self.state
513− .aborting(migrations.to_vec(), migration_index, action_index);
514− self.state
515− .save(&mut self.db)
516− .context("failed to save state")?;
517− }
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()))?;
518453
519− println!("{}", "done".green());
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")?;
520458 }
521− Ok(())
459+
460+ println!("{}", "done".green());
522461 }
523−}
524462
525−pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
526− migrations
527− .last()
528− .map(|migration| schema_name_for_migration(&migration.name))
463+ helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
464+
465+ *state = State::Idle;
466+ state.save(db).context("failed to save state")?;
467+
468+ Ok(())
529469}
530470
531−pub fn schema_query_for_migration(migration_name: &str) -> String {
471+fn create_schema_for_migration(
472+ db: &mut DbConn,
473+ migration_name: &str,
474+ schema: &Schema,
475+) -> anyhow::Result<()> {
476+ // Create schema for migration
532477 let schema_name = schema_name_for_migration(migration_name);
533− format!("SET search_path TO {}", schema_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+ })?;
485+
486+ // Create views inside schema
487+ for table in schema.get_tables(db)? {
488+ create_view_for_table(db, &table, &schema_name)?;
489+ }
490+
491+ Ok(())
534492}
535493
536−fn schema_name_for_migration(migration_name: &str) -> String {
537− format!("migration_{}", migration_name)
494+fn 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();
508+
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))?;
521+
522+ Ok(())
538523}