@@ -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
@@ -15,11 +15,10 @@ pub mod migrations;
1515mod schema;
1616mod state;
1717
18−pub use crate::state::{State, Status};
18+pub use crate::state::State;
1919
2020pub struct Reshape {
21− pub state: State,
22− db: DbConn,
21+ db: DbLocker,
2322}
2423
2524impl Reshape {
@@ -31,6 +30,7 @@ impl Reshape {
3130 pub fn new_with_options(
3231 host: &str,
3332 port: u16,
33+ database: &str,
3434 username: &str,
3535 password: &str,
3636 ) -> anyhow::Result<Reshape> {
@@ -39,501 +39,488 @@ impl Reshape {
3939 .host(host)
4040 .port(port)
4141 .user(username)
42+ .dbname(database)
4243 .password(password);
4344
4445 Self::new_with_config(&config)
4546 }
4647
4748 fn new_with_config(config: &Config) -> anyhow::Result<Reshape> {
48− let mut db = DbConn::connect(config)?;
49− let state = State::load(&mut db);
50−
51− Ok(Reshape { db, state })
49+ let db = DbLocker::connect(config)?;
50+ Ok(Reshape { db })
5251 }
5352
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);
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+ }
5962
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− }
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+ }
6569
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− }
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+ }
7276
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− }
77+ pub fn remove(&mut self) -> anyhow::Result<()> {
78+ self.db.lock(|db| {
79+ let mut state = State::load(db)?;
8380
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− ));
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+ ))?;
9487 }
95− }
96−
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)?;
103−
104− println!("Applying {} migrations\n", remaining_migrations.len());
10588
106− helpers::set_up_helpers(&mut self.db, ¤t_migration)
107− .context("failed to set up helpers")?;
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+ }
10896
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(());
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+ }
113107
114− for (migration_index, migration) in remaining_migrations.iter().enumerate() {
115− println!("Migrating '{}':", migration.name);
116− last_migration_index = migration_index;
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+ }
117117
118− for (action_index, action) in migration.actions.iter().enumerate() {
119− last_action_index = action_index;
118+ // Reset state
119+ state.clear(db)?;
120120
121− let description = action.describe();
122− print!(" + {} ", description);
121+ println!("Reshape and all data has been removed");
123122
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));
123+ Ok(())
124+ })
125+ }
126+}
128127
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− }
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+}
137133
138− println!();
139− }
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+}
140138
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");
139+fn schema_name_for_migration(migration_name: &str) -> String {
140+ format!("migration_{}", migration_name)
141+}
144142
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− );
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+ }
154155
155− // Abort will only
156− self.abort()?;
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+ }
157162
158− return Err(err);
159− }
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+ }
160172
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− );
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+ ));
194183 }
195−
196− Ok(())
197184 }
198185
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(());
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;
224225 }
225− };
226−
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")?;
235226 }
236227
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;
241− }
242−
243− println!("Completing '{}':", migration.name);
228+ println!();
229+ }
244230
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− }
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");
252234
253− let description = action.describe();
254− print!(" + {} ", description);
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+ );
255244
256− let ctx = MigrationContext::new(migration_index, action_index);
245+ // Abort will only
246+ abort(db, state)?;
257247
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− );
248+ return Err(err);
249+ }
265250
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− };
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+}
309269
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")?;
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`."))
315288 }
316− }
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+ };
317297
318− println!();
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+ }
306+
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;
319311 }
320312
321− // Remove helpers which are no longer in use
322− helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
313+ println!("Completing '{}':", migration.name);
323314
324− self.state
325− .complete(&mut self.db)
326− .context("failed to update state as completed")?;
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+ }
327320
328− Ok(())
329− }
321+ let description = action.describe();
322+ print!(" + {} ", description);
330323
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− }
324+ let ctx =
325+ MigrationContext::new(migration_index, action_index, state::current_migration(db)?);
351326
352− Ok(())
353− }
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+ );
354334
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();
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);
355+ }
356+ };
373357
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))?;
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")?;
386372
387− Ok(())
388− }
373+ true
374+ } else {
375+ false
376+ }
377+ };
389378
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− ))?;
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+ }
397385 }
398386
399− if let Status::InProgress { migrations } = &self.state.status {
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− }
387+ println!();
388+ }
406389
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− }
390+ // Remove helpers which are no longer in use
391+ helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
417392
418− // Reset state
419− self.state.clear(&mut self.db)?;
393+ state
394+ .complete(db)
395+ .context("failed to update state as completed")?;
420396
421− println!("Reshape and all data has been removed");
397+ Ok(())
398+}
422399
423− Ok(())
424− }
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)?;
425407
426− pub fn abort(&mut self) -> anyhow::Result<()> {
427− let (remaining_migrations, last_migration_index, last_action_index) = match self
428− .state
429− .status
430− .clone()
431− {
432− Status::InProgress { migrations } | Status::Applying { migrations } => {
433− // Set to the Aborting state. Once this is done, the migration has to
434− // be fully aborted and can't be completed.
435− self.state.aborting(migrations.clone(), 0, 0);
436− self.state.save(&mut self.db)?;
437−
438− (migrations, usize::MAX, usize::MAX)
439− }
440− Status::Aborting {
441− migrations,
442− last_migration_index,
443− last_action_index,
444− } => (migrations, last_migration_index, last_action_index),
445− Status::Completing { .. } => {
446− return Err(anyhow!("Migration completion has already been started. Please run `reshape complete` again to finish it."));
447− }
448− Status::Idle => {
449− println!("No migration is in progress");
450− return Ok(());
451− }
452− };
453−
454− // Remove new migration's schema
455− let target_migration = remaining_migrations.last().unwrap().name.to_string();
456− let schema_name = schema_name_for_migration(&target_migration);
457− self.db
458− .run(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema_name,))
459− .with_context(|| format!("failed to drop schema {}", schema_name))?;
460−
461− // Abort all pending migrations
462− self.abort_migrations(
463− &remaining_migrations,
408+ (migrations, usize::MAX, usize::MAX)
409+ }
410+ State::Aborting {
411+ migrations,
464412 last_migration_index,
465413 last_action_index,
466− )?;
467−
468− helpers::tear_down_helpers(&mut self.db).context("failed to tear down helpers")?;
469−
470− self.state.status = state::Status::Idle;
471− self.state
472− .save(&mut self.db)
473− .context("failed to save state")?;
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+ };
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;
438+ }
474439
475− Ok(())
476− }
440+ print!("Aborting '{}' ", migration.name);
477441
478− fn abort_migrations(
479− &mut self,
480− migrations: &[Migration],
481− upper_migration_index: usize,
482− upper_action_index: usize,
483− ) -> anyhow::Result<()> {
484− // Abort all migrations in reverse order
485− for (migration_index, migration) in migrations.iter().enumerate().rev() {
486− // Skip migrations which shouldn't be aborted
442+ for (action_index, action) in migration.actions.iter().enumerate().rev() {
443+ // Skip actions which shouldn't be aborted
487444 // The reason can be that they have already been aborted or that
488− // the migration was never applied in the first place.
489− if migration_index >= upper_migration_index {
445+ // the action was never applied in the first place.
446+ if migration_index == last_migration_index - 1 && action_index >= last_action_index {
490447 continue;
491448 }
492449
493− print!("Aborting '{}' ", migration.name);
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+ }
494462
495− for (action_index, action) in migration.actions.iter().enumerate().rev() {
496− // Skip actions which shouldn't be aborted
497− // The reason can be that they have already been aborted or that
498− // the action was never applied in the first place.
499− if migration_index == upper_migration_index - 1
500− && action_index >= upper_action_index
501− {
502− continue;
503− }
463+ println!("{}", "done".green());
464+ }
504465
505− let ctx = MigrationContext::new(migration_index, action_index);
506− action
507− .abort(&ctx, &mut self.db)
508− .with_context(|| format!("failed to abort migration {}", migration.name))
509− .with_context(|| format!("failed to abort action: {}", action.describe()))?;
510−
511− // Update state with which migrations and actions have been aborted.
512− // We don't need to run this in a transaction as aborts are idempotent.
513− self.state
514− .aborting(migrations.to_vec(), migration_index, action_index);
515− self.state
516− .save(&mut self.db)
517− .context("failed to save state")?;
518− }
466+ helpers::tear_down_helpers(db).context("failed to tear down helpers")?;
519467
520− println!("{}", "done".green());
521− }
522− Ok(())
523− }
524−}
468+ *state = State::Idle;
469+ state.save(db).context("failed to save state")?;
525470
526−pub fn latest_schema_from_migrations(migrations: &[Migration]) -> Option<String> {
527− migrations
528− .last()
529− .map(|migration| schema_name_for_migration(&migration.name))
471+ Ok(())
530472}
531473
532−pub fn schema_query_for_migration(migration_name: &str) -> String {
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
533480 let schema_name = schema_name_for_migration(migration_name);
534− format!("SET search_path TO {}", schema_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(())
535495}
536496
537−fn schema_name_for_migration(migration_name: &str) -> String {
538− format!("migration_{}", migration_name)
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(())
539526}