1use crate::ids::WorkspaceId;
2use std::fs;
3use std::ops::{Deref, DerefMut};
4use std::path::{Path, PathBuf};
5use std::str::FromStr;
6use std::sync::Arc;
7use std::time::Duration;
8
9use anyhow::{Context, Result};
10use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
11use sqlx::{Connection as _, Sqlite, SqliteConnection, SqlitePool, Transaction};
12use tokio::sync::{Mutex, OwnedMutexGuard};
13
14use crate::ids::new_id;
15use crate::workspaces::ensure_default_workspace;
16
17mod backup;
18mod changes;
19mod field_versions;
20mod inspection;
21mod rows;
22
23pub use backup::{
24 backup_database, default_backup_path, default_sqlite_backup_path, restore_database_file,
25 shm_path, wal_path,
26};
27pub(crate) use backup::{backup_database_with_connection, create_restore_safety_backup};
28pub(crate) use changes::{IdentifiedChange, insert_change, insert_change_with_identity};
29pub(crate) use field_versions::{
30 conflict_exists, entity_conflict_exists, entity_field_version, field_version,
31 set_entity_field_version, set_field_version,
32};
33pub use inspection::{DatabaseInspection, InspectedDatabase};
34pub(crate) use rows::{
35 recurrence_occurrence_from_row, recurrence_pause_interval_from_row, recurrence_series_from_row,
36 recurrence_series_label_from_row, task_from_row,
37};
38
39static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
40const FILE_DATABASE_CONNECTIONS: u32 = 5;
41
42#[derive(Clone)]
43pub struct Database {
44 pool: SqlitePool,
45 writer: Arc<Mutex<()>>,
46 path: PathBuf,
47 file_identity: Option<PathBuf>,
48 _inspection_dir: Option<Arc<tempfile::TempDir>>,
49}
50
51pub(crate) struct WriterConnection {
52 connection: sqlx::pool::PoolConnection<Sqlite>,
53 _guard: OwnedMutexGuard<()>,
54}
55
56impl Deref for WriterConnection {
57 type Target = SqliteConnection;
58
59 fn deref(&self) -> &Self::Target {
60 &self.connection
61 }
62}
63
64impl DerefMut for WriterConnection {
65 fn deref_mut(&mut self) -> &mut Self::Target {
66 &mut self.connection
67 }
68}
69
70impl Database {
71 pub async fn open(path: &Path) -> Result<Self> {
72 let connection_input = path.to_string_lossy();
73 let options = SqliteConnectOptions::from_str(&connection_input)?;
74 let storage = database_storage(&connection_input, &options);
75 let pool = open_db(path).await?;
76 let file_identity = match storage {
77 DatabaseStorage::InMemory => None,
78 DatabaseStorage::File => {
79 Some(fs::canonicalize(options.get_filename()).with_context(|| {
80 format!(
81 "could not resolve database path {}",
82 options.get_filename().display()
83 )
84 })?)
85 }
86 };
87 Ok(Self {
88 pool,
89 writer: Arc::new(Mutex::new(())),
90 path: path.to_path_buf(),
91 file_identity,
92 _inspection_dir: None,
93 })
94 }
95
96 pub fn path(&self) -> &Path {
97 &self.path
98 }
99
100 #[doc(hidden)]
101 pub fn file_identity(&self) -> Option<&Path> {
102 self.file_identity.as_deref()
103 }
104
105 pub async fn meta(&self, key: &str) -> Result<Option<String>> {
106 let mut conn = self.acquire_reader().await?;
107 get_meta(&mut conn, key).await
108 }
109
110 pub async fn conflict_exists(
111 &self,
112 workspace_id: &WorkspaceId,
113 task_id: &crate::ids::TaskId,
114 field: &str,
115 ) -> Result<bool> {
116 let mut conn = self.acquire_reader().await?;
117 conflict_exists(&mut conn, workspace_id, task_id, field).await
118 }
119
120 pub(crate) async fn acquire_reader(&self) -> Result<sqlx::pool::PoolConnection<Sqlite>> {
121 Ok(self.pool.acquire().await?)
122 }
123
124 pub(crate) async fn acquire_writer(&self) -> Result<WriterConnection> {
125 let guard = self.writer.clone().lock_owned().await;
126 let connection = self.pool.acquire().await?;
127 Ok(WriterConnection {
128 connection,
129 _guard: guard,
130 })
131 }
132}
133
134pub(crate) async fn open_db(path: &Path) -> Result<SqlitePool> {
135 let connection_input = path.to_string_lossy();
136 let mut options = SqliteConnectOptions::from_str(&connection_input)?;
137 let storage = database_storage(&connection_input, &options);
138 let existed_before_open = storage == DatabaseStorage::File && path.exists();
139 if storage == DatabaseStorage::File
140 && let Some(parent) = path.parent()
141 {
142 fs::create_dir_all(parent)
143 .with_context(|| format!("could not create {}", parent.display()))?;
144 }
145 options = options
146 .create_if_missing(true)
147 .foreign_keys(true)
148 .busy_timeout(Duration::from_secs(5));
149 let pool_options = match storage {
150 DatabaseStorage::InMemory => SqlitePoolOptions::new()
151 .min_connections(1)
152 .max_connections(1)
153 .idle_timeout(None)
154 .max_lifetime(None),
155 DatabaseStorage::File => {
156 options = options.journal_mode(SqliteJournalMode::Wal);
157 SqlitePoolOptions::new().max_connections(FILE_DATABASE_CONNECTIONS)
158 }
159 };
160 let pool = pool_options
161 .connect_with(options)
162 .await
163 .with_context(|| format!("could not open {}", path.display()))?;
164 backup::backup_before_pending_migrations(path, existed_before_open, &pool).await?;
165 MIGRATOR.run(&pool).await?;
166 initialize_meta(&pool).await?;
167 let mut conn = pool.acquire().await?;
168 crate::sync::protocol::replica_protocol(&mut conn).await?;
169 ensure_default_workspace(&mut conn).await?;
170 let mut tx = begin_immediate(&mut conn).await?;
171 crate::epic_membership::recover(&mut tx, false).await?;
172 tx.commit().await?;
173 Ok(pool)
174}
175
176#[derive(Clone, Copy, Debug, Eq, PartialEq)]
177enum DatabaseStorage {
178 InMemory,
179 File,
180}
181
182fn database_storage(connection_input: &str, options: &SqliteConnectOptions) -> DatabaseStorage {
183 let connection_input = connection_input
184 .trim_start_matches("sqlite://")
185 .trim_start_matches("sqlite:");
186 let mut database_and_params = connection_input.splitn(2, '?');
187 let database = database_and_params.next().unwrap_or_default();
188 let uses_memory_mode = database_and_params.next().is_some_and(|params| {
189 url::form_urlencoded::parse(params.as_bytes())
190 .any(|(key, value)| key == "mode" && value == "memory")
191 });
192 let filename = options.get_filename();
193
194 if database == ":memory:"
195 || database == "file::memory:"
196 || filename == Path::new(":memory:")
197 || filename == Path::new("file::memory:")
198 || uses_memory_mode
199 {
200 DatabaseStorage::InMemory
201 } else {
202 DatabaseStorage::File
203 }
204}
205
206async fn initialize_meta(pool: &SqlitePool) -> Result<()> {
207 let mut conn = pool.acquire().await?;
208 insert_meta_if_missing(&mut conn, "client_id", &new_id()).await?;
209 insert_meta_if_missing(&mut conn, "sync_cursor", "0").await?;
210 insert_meta_if_missing(&mut conn, "local_seq", "0").await?;
211 Ok(())
212}
213
214pub(crate) async fn current_schema_version(conn: &mut SqliteConnection) -> Result<i64> {
215 let version: Option<i64> = sqlx::query_scalar("SELECT MAX(version) FROM _sqlx_migrations")
216 .fetch_one(conn)
217 .await?;
218 Ok(version.unwrap_or(0))
219}
220
221pub(crate) async fn get_meta(conn: &mut SqliteConnection, key: &str) -> Result<Option<String>> {
222 Ok(
223 sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?", key)
224 .fetch_optional(&mut *conn)
225 .await?,
226 )
227}
228
229pub(crate) async fn set_meta(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
230 sqlx::query!(
231 "INSERT INTO meta(key, value) VALUES (?, ?)
232 ON CONFLICT(key) DO UPDATE SET value = excluded.value",
233 key,
234 value,
235 )
236 .execute(&mut *conn)
237 .await?;
238 Ok(())
239}
240
241pub(crate) async fn begin_immediate(
242 conn: &mut SqliteConnection,
243) -> sqlx::Result<Transaction<'_, Sqlite>> {
244 conn.begin_with("BEGIN IMMEDIATE").await
245}
246
247async fn insert_meta_if_missing(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
248 sqlx::query!(
249 "INSERT OR IGNORE INTO meta(key, value) VALUES (?, ?)",
250 key,
251 value
252 )
253 .execute(&mut *conn)
254 .await?;
255 Ok(())
256}
257
258#[cfg(test)]
259mod tests;
260
261#[cfg(test)]
262mod payload_tests;