turso_orm_migration/migrator.rs
1//! Applying and reverting migrations, modeled by [`MigratorTrait`].
2//!
3//! The migrator keeps a bookkeeping table, `turso_migrations` unless
4//! [`MigratorTrait::migration_table_name`] says otherwise, with one row per
5//! applied migration. Every operation starts by making sure that table
6//! exists and reading it, then walks the declared migrations in order (or in
7//! reverse for `down`) and skips the ones whose state already matches.
8//!
9//! Each migration runs inside its own `BEGIN IMMEDIATE` transaction, and
10//! the bookkeeping insert or delete is issued on that same transaction
11//! before the commit. Taking the write lock up front avoids a busy error
12//! mid-migration, and bundling the version row with the schema change means
13//! a failure leaves neither a partial schema nor a misleading version row.
14
15use async_trait::async_trait;
16use turso_orm::sql::{ColumnDef, Expr, Order, Query, Table};
17use turso_orm::{ConnectionTrait, Database, DbErr, Statement, TransactionMode, TransactionTrait};
18
19use crate::MigrationTrait;
20use crate::manager::{SchemaManager, has_table};
21
22/// The default name of the bookkeeping table.
23const DEFAULT_TABLE: &str = "turso_migrations";
24
25/// The status of one declared migration.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub struct MigrationStatus {
28 /// The migration name.
29 pub name: String,
30 /// Whether the migration has been applied.
31 pub applied: bool,
32}
33
34/// Lists migrations and applies or reverts them in order.
35///
36/// Implement [`migrations`](Self::migrations) only; every other method has
37/// a default built on it.
38#[async_trait]
39pub trait MigratorTrait: Send {
40 /// Every migration, oldest first.
41 fn migrations() -> Vec<Box<dyn MigrationTrait>>;
42
43 /// The name of the bookkeeping table, `turso_migrations` by default.
44 ///
45 /// Override it to run several migrators against one database, or to
46 /// keep the table name of a schema that was migrated by another tool.
47 fn migration_table_name() -> &'static str {
48 DEFAULT_TABLE
49 }
50
51 /// Creates the bookkeeping table if it does not exist yet.
52 ///
53 /// # Errors
54 ///
55 /// Returns [`DbErr::Driver`] when the statement fails.
56 async fn install(db: &Database) -> Result<(), DbErr> {
57 let stmt = Table::create()
58 .table(Self::migration_table_name())
59 .if_not_exists()
60 .col(ColumnDef::text("version").primary_key().not_null())
61 .col(ColumnDef::integer("applied_at").not_null());
62 db.execute(turso_orm::Build::to_statement(&stmt)).await?;
63 Ok(())
64 }
65
66 /// The names of the applied migrations, in application order.
67 ///
68 /// Returns an empty list when the bookkeeping table does not exist, so
69 /// that status can be queried on a database that was never migrated.
70 ///
71 /// # Errors
72 ///
73 /// Returns [`DbErr::Migration`] when the catalog query returns no row;
74 /// [`DbErr::Driver`] when a query fails or a version cannot be decoded.
75 async fn get_applied_migrations(db: &Database) -> Result<Vec<String>, DbErr> {
76 if !has_table(db, Self::migration_table_name()).await? {
77 return Ok(Vec::new());
78 }
79 // The timestamp has second resolution, so the name breaks ties
80 // between migrations applied within the same second.
81 let stmt = Query::select()
82 .column("version")
83 .from(Self::migration_table_name())
84 .order_by("applied_at", Order::Asc)
85 .order_by("version", Order::Asc);
86 let rows = db.query_all(turso_orm::Build::to_statement(&stmt)).await?;
87 rows.iter()
88 .map(|r| r.get::<String>("version").map_err(DbErr::from))
89 .collect()
90 }
91
92 /// The status of every declared migration, in declaration order.
93 ///
94 /// # Errors
95 ///
96 /// Returns the errors of [`get_applied_migrations`](Self::get_applied_migrations).
97 async fn status(db: &Database) -> Result<Vec<MigrationStatus>, DbErr> {
98 let applied = Self::get_applied_migrations(db).await?;
99 Ok(Self::migrations()
100 .iter()
101 .map(|m| MigrationStatus {
102 name: m.name().to_owned(),
103 applied: applied.iter().any(|a| a == m.name()),
104 })
105 .collect())
106 }
107
108 /// Applies the pending migrations, all of them or the first `steps`.
109 ///
110 /// Each migration and its version row are committed together; on the
111 /// first failure the transaction is dropped and rolled back, and the
112 /// error is returned without touching later migrations.
113 ///
114 /// # Errors
115 ///
116 /// Returns [`DbErr::Driver`] when a statement or the transaction fails;
117 /// any error the migration's `up` returns.
118 async fn up(db: &Database, steps: Option<u32>) -> Result<(), DbErr> {
119 Self::install(db).await?;
120 let applied = Self::get_applied_migrations(db).await?;
121 let mut remaining = steps.map_or(usize::MAX, |s| s as usize);
122 for migration in Self::migrations() {
123 if remaining == 0 {
124 break;
125 }
126 if applied.iter().any(|a| a == migration.name()) {
127 continue;
128 }
129 tracing::info!(name = migration.name(), "applying migration");
130 // `IMMEDIATE` takes the write lock now rather than at the first
131 // write, so the migration cannot hit a busy error halfway.
132 let txn = db.begin_with_mode(TransactionMode::Immediate).await?;
133 {
134 let manager = SchemaManager::new(&txn);
135 migration.up(&manager).await?;
136 }
137 let now = std::time::SystemTime::now()
138 .duration_since(std::time::UNIX_EPOCH)
139 .map(|d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
140 .unwrap_or_default();
141 let insert = Query::insert()
142 .into_table(Self::migration_table_name())
143 .columns(["version", "applied_at"])
144 .values([Expr::val(migration.name()), Expr::val(now)]);
145 txn.execute(turso_orm::Build::to_statement(&insert)).await?;
146 txn.commit().await?;
147 remaining -= 1;
148 }
149 Ok(())
150 }
151
152 /// Reverts the applied migrations, newest first, all of them or `steps`.
153 ///
154 /// Each migration's `down` and the deletion of its version row are
155 /// committed together, mirroring [`up`](Self::up).
156 ///
157 /// # Errors
158 ///
159 /// Returns [`DbErr::Driver`] when a statement or the transaction fails;
160 /// any error the migration's `down` returns, including the default
161 /// [`DbErr::Migration`] of an irreversible migration.
162 async fn down(db: &Database, steps: Option<u32>) -> Result<(), DbErr> {
163 Self::install(db).await?;
164 let applied = Self::get_applied_migrations(db).await?;
165 let mut remaining = steps.map_or(usize::MAX, |s| s as usize);
166 for migration in Self::migrations().into_iter().rev() {
167 if remaining == 0 {
168 break;
169 }
170 if !applied.iter().any(|a| a == migration.name()) {
171 continue;
172 }
173 tracing::info!(name = migration.name(), "reverting migration");
174 let txn = db.begin_with_mode(TransactionMode::Immediate).await?;
175 {
176 let manager = SchemaManager::new(&txn);
177 migration.down(&manager).await?;
178 }
179 let delete = Query::delete()
180 .from_table(Self::migration_table_name())
181 .and_where(Expr::col("version").eq(Expr::val(migration.name())));
182 txn.execute(turso_orm::Build::to_statement(&delete)).await?;
183 txn.commit().await?;
184 remaining -= 1;
185 }
186 Ok(())
187 }
188
189 /// Drops every user table, including the bookkeeping table, then applies all migrations.
190 ///
191 /// Internal `sqlite_*` and `__turso_*` tables are left alone.
192 ///
193 /// # Errors
194 ///
195 /// Returns [`DbErr::Driver`] when a query, a drop or the transaction
196 /// fails; the errors of [`up`](Self::up).
197 async fn fresh(db: &Database) -> Result<(), DbErr> {
198 let rows = db
199 .query_all(Statement::from_string(
200 "SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE '__turso_%'",
201 ))
202 .await?;
203 let txn = db.begin_with_mode(TransactionMode::Immediate).await?;
204 // `PRAGMA foreign_keys` has no effect inside a transaction, so a
205 // table referenced by another one may refuse to drop first. Drop in
206 // rounds, retrying the tables that failed, until nothing is left or a
207 // round makes no progress — then the last error is the real one.
208 let mut pending: Vec<String> = rows
209 .iter()
210 .map(|row| row.get::<String>("name"))
211 .collect::<Result<_, _>>()?;
212 while !pending.is_empty() {
213 let before = pending.len();
214 let mut failed = Vec::new();
215 let mut last_error = None;
216 for name in pending {
217 let drop = Table::drop().table(name.clone()).if_exists();
218 if let Err(err) = txn.execute(turso_orm::Build::to_statement(&drop)).await {
219 failed.push(name);
220 last_error = Some(err);
221 }
222 }
223 if failed.len() == before
224 && let Some(err) = last_error
225 {
226 return Err(err.into());
227 }
228 pending = failed;
229 }
230 txn.commit().await?;
231 Self::up(db, None).await
232 }
233
234 /// Reverts every migration, then applies every migration.
235 ///
236 /// # Errors
237 ///
238 /// Returns the errors of [`down`](Self::down) and [`up`](Self::up).
239 async fn refresh(db: &Database) -> Result<(), DbErr> {
240 Self::down(db, None).await?;
241 Self::up(db, None).await
242 }
243
244 /// Reverts every migration.
245 ///
246 /// # Errors
247 ///
248 /// Returns the errors of [`down`](Self::down).
249 async fn reset(db: &Database) -> Result<(), DbErr> {
250 Self::down(db, None).await
251 }
252}