turso_orm_migration/
manager.rs

1//! The schema manager handed to migrations, modeled by [`SchemaManager`].
2//!
3//! The manager is a thin view over the transaction a migration runs in. It
4//! exists so that `up` and `down` receive one argument that can both run
5//! DDL and inspect the catalog, while keeping the transaction itself
6//! reachable through [`SchemaManager::get_connection`] for data migrations.
7//! It never commits or rolls back; the migrator owns the transaction's
8//! lifetime.
9//!
10//! Catalog lookups go through `sqlite_schema` and `pragma_table_info`
11//! because they are the only portable way to ask SQLite what exists; the
12//! same `has_table` helper serves the migrator's own check for the
13//! bookkeeping table.
14
15use turso_orm::sql::{
16    AlterTable, Build, CreateIndex, CreateTable, DropIndex, DropTable, Statement,
17};
18use turso_orm::{ConnectionTrait, DbErr, Transaction};
19
20/// Runs DDL inside the migration's transaction and inspects the catalog.
21#[derive(Debug)]
22pub struct SchemaManager<'c> {
23    /// The transaction the migration runs in.
24    conn: &'c Transaction,
25}
26
27impl<'c> SchemaManager<'c> {
28    /// Wraps a transaction.
29    pub fn new(conn: &'c Transaction) -> Self {
30        Self { conn }
31    }
32
33    /// The transaction the migration runs in, for data migrations and raw statements.
34    pub fn get_connection(&self) -> &'c Transaction {
35        self.conn
36    }
37
38    /// Executes any DDL or DML statement builder.
39    ///
40    /// # Errors
41    ///
42    /// Returns [`DbErr::Driver`] when the statement fails.
43    pub async fn exec_stmt(&self, stmt: impl Build) -> Result<(), DbErr> {
44        self.conn.execute(stmt.to_statement()).await?;
45        Ok(())
46    }
47
48    /// Executes a `CREATE TABLE`.
49    ///
50    /// # Errors
51    ///
52    /// Returns [`DbErr::Driver`] when the statement fails.
53    pub async fn create_table(&self, stmt: CreateTable) -> Result<(), DbErr> {
54        self.exec_stmt(stmt).await
55    }
56
57    /// Executes an `ALTER TABLE`.
58    ///
59    /// # Errors
60    ///
61    /// Returns [`DbErr::Driver`] when the statement fails.
62    pub async fn alter_table(&self, stmt: AlterTable) -> Result<(), DbErr> {
63        self.exec_stmt(stmt).await
64    }
65
66    /// Executes a `DROP TABLE`.
67    ///
68    /// # Errors
69    ///
70    /// Returns [`DbErr::Driver`] when the statement fails.
71    pub async fn drop_table(&self, stmt: DropTable) -> Result<(), DbErr> {
72        self.exec_stmt(stmt).await
73    }
74
75    /// Executes a `CREATE INDEX`.
76    ///
77    /// # Errors
78    ///
79    /// Returns [`DbErr::Driver`] when the statement fails.
80    pub async fn create_index(&self, stmt: CreateIndex) -> Result<(), DbErr> {
81        self.exec_stmt(stmt).await
82    }
83
84    /// Executes a `DROP INDEX`.
85    ///
86    /// # Errors
87    ///
88    /// Returns [`DbErr::Driver`] when the statement fails.
89    pub async fn drop_index(&self, stmt: DropIndex) -> Result<(), DbErr> {
90        self.exec_stmt(stmt).await
91    }
92
93    /// Whether a table named `table` exists.
94    ///
95    /// # Errors
96    ///
97    /// Returns [`DbErr::Migration`] when the catalog query returns no row;
98    /// [`DbErr::Driver`] when the query fails.
99    pub async fn has_table(&self, table: &str) -> Result<bool, DbErr> {
100        has_table(self.conn, table).await
101    }
102
103    /// Whether `table` has a column named `column`.
104    ///
105    /// # Errors
106    ///
107    /// Returns [`DbErr::Migration`] when the catalog query returns no row;
108    /// [`DbErr::Driver`] when the query fails.
109    pub async fn has_column(&self, table: &str, column: &str) -> Result<bool, DbErr> {
110        count_positive(
111            self.conn,
112            Statement::from_sql_and_values(
113                "SELECT COUNT(*) AS n FROM pragma_table_info(?) WHERE name = ?",
114                [table, column],
115            ),
116        )
117        .await
118    }
119
120    /// Whether an index named `index` exists.
121    ///
122    /// # Errors
123    ///
124    /// Returns [`DbErr::Migration`] when the catalog query returns no row;
125    /// [`DbErr::Driver`] when the query fails.
126    pub async fn has_index(&self, index: &str) -> Result<bool, DbErr> {
127        count_positive(
128            self.conn,
129            Statement::from_sql_and_values(
130                "SELECT COUNT(*) AS n FROM sqlite_schema WHERE type = 'index' AND name = ?",
131                [index],
132            ),
133        )
134        .await
135    }
136}
137
138/// Whether a table named `table` exists, on any connection.
139///
140/// Shared with the migrator, which needs the check before any transaction
141/// is open.
142///
143/// # Errors
144///
145/// Returns [`DbErr::Migration`] when the catalog query returns no row;
146/// [`DbErr::Driver`] when the query fails.
147pub(crate) async fn has_table<C: ConnectionTrait>(conn: &C, table: &str) -> Result<bool, DbErr> {
148    count_positive(
149        conn,
150        Statement::from_sql_and_values(
151            "SELECT COUNT(*) AS n FROM sqlite_schema WHERE type = 'table' AND name = ?",
152            [table],
153        ),
154    )
155    .await
156}
157
158/// Runs a `SELECT COUNT(*) AS n ...` statement and reports whether the count is positive.
159///
160/// # Errors
161///
162/// Returns [`DbErr::Migration`] when the query returns no row, which a
163/// `COUNT(*)` never should; [`DbErr::Driver`] when the query fails or `n`
164/// cannot be decoded.
165async fn count_positive<C: ConnectionTrait>(conn: &C, stmt: Statement) -> Result<bool, DbErr> {
166    let row = conn
167        .query_one(stmt)
168        .await?
169        .ok_or_else(|| DbErr::Migration("catalog query returned no row".into()))?;
170    Ok(row.get::<i64>("n")? > 0)
171}
turso_orm_migration/manager.rs

turso_orm_migration/
manager.rs
