@@ -8,8 +8,6 @@ use version::version;
88pub struct State {
99 pub version: String,
1010 pub status: Status,
11− pub current_migration: Option<String>,
12− pub migrations: Vec<Migration>,
1311}
1412
1513#[derive(Serialize, Deserialize, Clone, Debug)]
@@ -72,25 +70,25 @@ impl State {
7270
7371 let default = Self::default();
7472 self.status = default.status;
75− self.current_migration = default.current_migration;
76− self.migrations = default.migrations;
7773
7874 Ok(())
7975 }
8076
8177 // Complete will change the status from Completing to Idle
82− pub fn complete(&mut self) -> anyhow::Result<()> {
78+ pub fn complete(&mut self, db: &mut impl Conn) -> anyhow::Result<()> {
8379 let current_status = std::mem::replace(&mut self.status, Status::Idle);
84−
8580 match current_status {
86− Status::Completing { mut migrations, .. } => {
87− let target_migration = migrations.last().unwrap().name.to_string();
88− self.migrations.append(&mut migrations);
89− self.current_migration = Some(target_migration);
81+ Status::Completing { migrations, .. } => {
82+ // Add migrations and update state in a transaction to ensure atomicity
83+ let mut transaction = db.transaction()?;
84+ save_migrations(&mut transaction, &migrations)?;
85+ self.save(&mut transaction)?;
86+ transaction.commit()?;
9087 }
9188 _ => {
9289 // Move old status back
9390 self.status = current_status;
91+
9492 return Err(anyhow!(
9593 "couldn't update state to be completed, not in Completing state"
9694 ));
@@ -138,35 +136,24 @@ impl State {
138136 }
139137 }
140138
141− pub fn get_remaining_migrations(
142− &self,
143− new_migrations: impl IntoIterator<Item = Migration>,
144− ) -> anyhow::Result<Vec<Migration>> {
145− let mut new_iter = new_migrations.into_iter();
146−
147− // Ensure the new migration match up with the existing ones
148− for pair in self.migrations.iter().zip(new_iter.by_ref()) {
149− let (existing, ref new) = pair;
150− if existing != new {
151− return Err(anyhow!(
152− "existing migration {} does not match new migration {}",
153− existing.name,
154− new.name
155− ));
156− }
157− }
158−
159− let items: Vec<Migration> = new_iter.collect();
160−
161− // Return the remaining migrations
162− Ok(items)
163− }
164−
165139 fn ensure_schema_and_table(db: &mut impl Conn) {
166140 db.run("CREATE SCHEMA IF NOT EXISTS reshape").unwrap();
167141
168142 db.run("CREATE TABLE IF NOT EXISTS reshape.data (key TEXT PRIMARY KEY, value JSONB)")
169143 .unwrap();
144+
145+ db.run(
146+ "
147+ CREATE TABLE IF NOT EXISTS reshape.migrations (
148+ index INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
149+ name TEXT NOT NULL,
150+ description TEXT,
151+ actions JSONB NOT NULL,
152+ completed_at TIMESTAMP DEFAULT NOW()
153+ )
154+ ",
155+ )
156+ .unwrap();
170157 }
171158}
172159
@@ -175,8 +162,107 @@ impl Default for State {
175162 State {
176163 version: version!().to_string(),
177164 status: Status::Idle,
178− current_migration: None,
179− migrations: vec![],
180165 }
181166 }
182167}
168+
169+pub fn current_migration(db: &mut dyn Conn) -> anyhow::Result<Option<String>> {
170+ let name: Option<String> = db
171+ .query(
172+ "
173+ SELECT name
174+ FROM reshape.migrations
175+ ORDER BY index DESC
176+ LIMIT 1
177+ ",
178+ )?
179+ .first()
180+ .map(|row| row.get("name"));
181+ Ok(name)
182+}
183+
184+pub fn remaining_migrations(
185+ db: &mut impl Conn,
186+ new_migrations: impl IntoIterator<Item = Migration>,
187+) -> anyhow::Result<Vec<Migration>> {
188+ let mut new_iter = new_migrations.into_iter();
189+
190+ // Ensure the new migrations match up with the existing ones
191+ let mut highest_index: Option<i32> = None;
192+ loop {
193+ let migrations = get_migrations(db, highest_index)?;
194+ if migrations.is_empty() {
195+ break;
196+ }
197+
198+ for (index, existing) in migrations {
199+ highest_index = Some(index);
200+
201+ let new = match new_iter.next() {
202+ Some(migration) => migration,
203+ None => {
204+ return Err(anyhow!(
205+ "existing migration {} doesn't exist in local migrations",
206+ existing
207+ ))
208+ }
209+ };
210+
211+ if existing != new.name {
212+ return Err(anyhow!(
213+ "existing migration {} does not match new migration {}",
214+ existing,
215+ new.name
216+ ));
217+ }
218+ }
219+ }
220+
221+ // Return the remaining migrations
222+ let items: Vec<Migration> = new_iter.collect();
223+ Ok(items)
224+}
225+
226+fn get_migrations(
227+ db: &mut impl Conn,
228+ index_larger_than: Option<i32>,
229+) -> anyhow::Result<Vec<(i32, String)>> {
230+ let rows = if let Some(index_larger_than) = index_larger_than {
231+ db.query_with_params(
232+ "
233+ SELECT index, name
234+ FROM reshape.migrations
235+ WHERE index > $1
236+ ORDER BY index ASC
237+ LIMIT 100
238+ ",
239+ &[&index_larger_than],
240+ )?
241+ } else {
242+ db.query(
243+ "
244+ SELECT index, name
245+ FROM reshape.migrations
246+ LIMIT 100
247+ ",
248+ )?
249+ };
250+
251+ let migrations = rows
252+ .iter()
253+ .map(|row| (row.get("index"), row.get("name")))
254+ .collect();
255+ Ok(migrations)
256+}
257+
258+fn save_migrations(db: &mut impl Conn, migrations: &[Migration]) -> anyhow::Result<()> {
259+ for migration in migrations {
260+ let encoded_actions = serde_json::to_value(&migration.actions)?;
261+ db.query_with_params(
262+ "INSERT INTO reshape.migrations(name, description, actions) VALUES ($1, $2, $3)",
263+ &[&migration.name, &migration.description, &encoded_actions],
264+ )?;
265+ }
266+
267+ Ok(())
268+}