tenant.openagents/omega

1pub mod kvp;
2pub mod query;
3
4// Re-export
5pub use anyhow;
6use anyhow::Context as _;
7pub use gpui;
8use gpui::{App, AppContext, Global};
9pub use indoc::indoc;
10pub use inventory;
11pub use paths::database_dir;
12pub use sqlez;
13pub use sqlez_macros;
14pub use uuid;
15
16pub use release_channel::RELEASE_CHANNEL;
17use release_channel::ReleaseChannel;
18use sqlez::domain::Migrator;
19use sqlez::thread_safe_connection::ThreadSafeConnection;
20use sqlez_macros::sql;
21use std::fs::create_dir_all;
22use std::future::Future;
23use std::path::{Path, PathBuf};
24use std::sync::atomic::AtomicBool;
25use std::sync::{LazyLock, atomic::Ordering};
26use util::{ResultExt, maybe};
27use zed_env_vars::ZED_STATELESS;
28
29/// A migration registered via `static_connection!` and collected at link time.
30pub struct DomainMigration {
31    pub name: &'static str,
32    pub migrations: &'static [&'static str],
33    pub dependencies: &'static [&'static str],
34    pub should_allow_migration_change: fn(usize, &str, &str) -> bool,
35}
36
37inventory::collect!(DomainMigration);
38
39/// The shared database connection backing all domain-specific DB wrappers.
40/// Set as a GPUI global per-App. Falls back to a shared LazyLock if not set.
41pub struct AppDatabase(pub ThreadSafeConnection);
42
43impl Global for AppDatabase {}
44
45/// Migrator that runs all inventory-registered domain migrations.
46pub struct AppMigrator;
47
48impl Migrator for AppMigrator {
49    fn migrate(connection: &sqlez::connection::Connection) -> anyhow::Result<()> {
50        let registrations: Vec<&DomainMigration> = inventory::iter::<DomainMigration>().collect();
51        let sorted = topological_sort(&registrations);
52        for reg in &sorted {
53            let mut should_allow = reg.should_allow_migration_change;
54            connection.migrate(reg.name, reg.migrations, &mut should_allow)?;
55        }
56        Ok(())
57    }
58}
59
60impl AppDatabase {
61    /// Opens the production database and runs all inventory-registered
62    /// migrations in dependency order.
63    pub fn new() -> Self {
64        let db_dir = database_dir();
65        let connection = gpui::block_on(open_db::<AppMigrator>(db_dir, *RELEASE_CHANNEL));
66        Self(connection)
67    }
68
69    /// Creates a new in-memory database with a unique name and runs all
70    /// inventory-registered migrations in dependency order.
71    #[cfg(any(test, feature = "test-support"))]
72    pub fn test_new() -> Self {
73        let name = format!("test-db-{}", uuid::Uuid::new_v4());
74        let connection = gpui::block_on(open_test_db::<AppMigrator>(&name));
75        Self(connection)
76    }
77
78    /// Returns the per-App connection if set, otherwise falls back to
79    /// the shared LazyLock.
80    pub fn global(cx: &App) -> &ThreadSafeConnection {
81        #[allow(unreachable_code)]
82        if let Some(db) = cx.try_global::<Self>() {
83            return &db.0;
84        } else {
85            #[cfg(any(feature = "test-support", test))]
86            return &TEST_APP_DATABASE.0;
87
88            panic!("database not initialized")
89        }
90    }
91}
92
93fn topological_sort<'a>(registrations: &[&'a DomainMigration]) -> Vec<&'a DomainMigration> {
94    let mut sorted: Vec<&DomainMigration> = Vec::new();
95    let mut visited: std::collections::HashSet<&str> = std::collections::HashSet::new();
96
97    fn visit<'a>(
98        name: &str,
99        registrations: &[&'a DomainMigration],
100        sorted: &mut Vec<&'a DomainMigration>,
101        visited: &mut std::collections::HashSet<&'a str>,
102    ) {
103        if visited.contains(name) {
104            return;
105        }
106        if let Some(reg) = registrations.iter().find(|r| r.name == name) {
107            for dep in reg.dependencies {
108                visit(dep, registrations, sorted, visited);
109            }
110            visited.insert(reg.name);
111            sorted.push(reg);
112        }
113    }
114
115    for reg in registrations {
116        visit(reg.name, registrations, &mut sorted, &mut visited);
117    }
118    sorted
119}
120
121/// Shared fallback `AppDatabase` used when no per-App global is set.
122#[cfg(any(test, feature = "test-support"))]
123static TEST_APP_DATABASE: LazyLock<AppDatabase> = LazyLock::new(AppDatabase::test_new);
124
125const CONNECTION_INITIALIZE_QUERY: &str = sql!(
126    PRAGMA foreign_keys=TRUE;
127);
128
129const DB_INITIALIZE_QUERY: &str = sql!(
130    PRAGMA journal_mode=WAL;
131    PRAGMA busy_timeout=500;
132    PRAGMA case_sensitive_like=TRUE;
133    PRAGMA synchronous=NORMAL;
134);
135
136const FALLBACK_DB_NAME: &str = "FALLBACK_MEMORY_DB";
137
138const DB_FILE_NAME: &str = "db.sqlite";
139
140pub static ALL_FILE_DB_FAILED: LazyLock<AtomicBool> = LazyLock::new(|| AtomicBool::new(false));
141
142/// A type that can be used as a database scope for path construction.
143pub trait DbScope {
144    fn scope_name(&self) -> &str;
145}
146
147impl DbScope for ReleaseChannel {
148    fn scope_name(&self) -> &str {
149        self.dev_name()
150    }
151}
152
153/// A database scope shared across all release channels.
154pub struct GlobalDbScope;
155
156impl DbScope for GlobalDbScope {
157    fn scope_name(&self) -> &str {
158        "global"
159    }
160}
161
162/// Returns the path to the `AppDatabase` SQLite file for the given scope
163/// under `db_dir`.
164pub fn db_path(db_dir: &Path, scope: impl DbScope) -> PathBuf {
165    db_dir
166        .join(format!("0-{}", scope.scope_name()))
167        .join(DB_FILE_NAME)
168}
169
170/// Open or create a database at the given directory path.
171/// This will retry a couple times if there are failures. If opening fails once, the db directory
172/// is moved to a backup folder and a new one is created. If that fails, a shared in memory db is created.
173/// In either case, static variables are set so that the user can be notified.
174pub async fn open_db<M: Migrator + 'static>(
175    db_dir: &Path,
176    scope: impl DbScope,
177) -> ThreadSafeConnection {
178    if *ZED_STATELESS {
179        return open_fallback_db::<M>().await;
180    }
181
182    let db_path = db_path(db_dir, scope);
183
184    let connection = maybe!(async {
185        if let Some(parent) = db_path.parent() {
186            create_dir_all(parent)
187                .context("Could not create db directory")
188                .log_err()?;
189        }
190        open_main_db::<M>(&db_path).await
191    })
192    .await;
193
194    if let Some(connection) = connection {
195        return connection;
196    }
197
198    // Set another static ref so that we can escalate the notification
199    ALL_FILE_DB_FAILED.store(true, Ordering::Release);
200
201    // If still failed, create an in memory db with a known name
202    open_fallback_db::<M>().await
203}
204
205async fn open_main_db<M: Migrator>(db_path: &Path) -> Option<ThreadSafeConnection> {
206    log::trace!("Opening database {}", db_path.display());
207    ThreadSafeConnection::builder::<M>(db_path.to_string_lossy().as_ref(), true)
208        .with_db_initialization_query(DB_INITIALIZE_QUERY)
209        .with_connection_initialize_query(CONNECTION_INITIALIZE_QUERY)
210        .build()
211        .await
212        .log_err()
213}
214
215async fn open_fallback_db<M: Migrator>() -> ThreadSafeConnection {
216    log::warn!("Opening fallback in-memory database");
217    ThreadSafeConnection::builder::<M>(FALLBACK_DB_NAME, false)
218        .with_db_initialization_query(DB_INITIALIZE_QUERY)
219        .with_connection_initialize_query(CONNECTION_INITIALIZE_QUERY)
220        .build()
221        .await
222        .expect(
223            "Fallback in memory database failed. Likely initialization queries or migrations have fundamental errors",
224        )
225}
226
227#[cfg(any(test, feature = "test-support"))]
228pub async fn open_test_db<M: Migrator>(db_name: &str) -> ThreadSafeConnection {
229    use sqlez::thread_safe_connection::locking_queue;
230
231    ThreadSafeConnection::builder::<M>(db_name, false)
232        .with_db_initialization_query(DB_INITIALIZE_QUERY)
233        .with_connection_initialize_query(CONNECTION_INITIALIZE_QUERY)
234        // Serialize queued writes via a mutex and run them synchronously
235        .with_write_queue_constructor(locking_queue())
236        .build()
237        .await
238        .unwrap()
239}
240
241/// Implements a basic DB wrapper for a given domain
242///
243/// Arguments:
244/// - type of connection wrapper
245/// - dependencies, whose migrations should be run prior to this domain's migrations
246#[macro_export]
247macro_rules! static_connection {
248    ($t:ident, [ $($d:ty),* ]) => {
249        impl ::std::ops::Deref for $t {
250            type Target = $crate::sqlez::thread_safe_connection::ThreadSafeConnection;
251
252            fn deref(&self) -> &Self::Target {
253                &self.0
254            }
255        }
256
257        impl ::std::clone::Clone for $t {
258            fn clone(&self) -> Self {
259                $t(self.0.clone())
260            }
261        }
262
263        impl $t {
264            /// Returns an instance backed by the per-App database if set,
265            /// or the shared fallback connection otherwise.
266            pub fn global(cx: &$crate::gpui::App) -> Self {
267                $t($crate::AppDatabase::global(cx).clone())
268            }
269
270            #[cfg(any(test, feature = "test-support"))]
271            pub async fn open_test_db(name: &'static str) -> Self {
272                $t($crate::open_test_db::<$t>(name).await)
273            }
274        }
275
276        $crate::inventory::submit! {
277            $crate::DomainMigration {
278                name: <$t as $crate::sqlez::domain::Domain>::NAME,
279                migrations: <$t as $crate::sqlez::domain::Domain>::MIGRATIONS,
280                dependencies: &[$(<$d as $crate::sqlez::domain::Domain>::NAME),*],
281                should_allow_migration_change: <$t as $crate::sqlez::domain::Domain>::should_allow_migration_change,
282            }
283        }
284    }
285}
286
287pub fn write_and_log<F>(cx: &App, db_write: impl FnOnce() -> F + Send + 'static)
288where
289    F: Future<Output = anyhow::Result<()>> + Send,
290{
291    cx.background_spawn(async move { db_write().await.log_err() })
292        .detach()
293}
294
295#[cfg(test)]
296mod tests {
297    use std::thread;
298
299    use sqlez::domain::Domain;
300    use sqlez_macros::sql;
301
302    use crate::open_db;
303
304    // Test bad migration panics
305    #[gpui::test]
306    #[should_panic]
307    async fn test_bad_migration_panics() {
308        enum BadDB {}
309
310        impl Domain for BadDB {
311            const NAME: &str = "db_tests";
312            const MIGRATIONS: &[&str] = &[
313                sql!(CREATE TABLE test(value);),
314                // failure because test already exists
315                sql!(CREATE TABLE test(value);),
316            ];
317        }
318
319        let tempdir = tempfile::Builder::new()
320            .prefix("DbTests")
321            .tempdir()
322            .unwrap();
323        let _bad_db = open_db::<BadDB>(tempdir.path(), release_channel::ReleaseChannel::Dev).await;
324    }
325
326    /// Test that DB exists but corrupted (causing recreate)
327    #[gpui::test]
328    async fn test_db_corruption(cx: &mut gpui::TestAppContext) {
329        cx.executor().allow_parking();
330
331        enum CorruptedDB {}
332
333        impl Domain for CorruptedDB {
334            const NAME: &str = "db_tests";
335            const MIGRATIONS: &[&str] = &[sql!(CREATE TABLE test(value);)];
336        }
337
338        enum GoodDB {}
339
340        impl Domain for GoodDB {
341            const NAME: &str = "db_tests"; //Notice same name
342            const MIGRATIONS: &[&str] = &[sql!(CREATE TABLE test2(value);)];
343        }
344
345        let tempdir = tempfile::Builder::new()
346            .prefix("DbTests")
347            .tempdir()
348            .unwrap();
349        {
350            let corrupt_db =
351                open_db::<CorruptedDB>(tempdir.path(), release_channel::ReleaseChannel::Dev).await;
352            assert!(corrupt_db.persistent());
353        }
354
355        let good_db = open_db::<GoodDB>(tempdir.path(), release_channel::ReleaseChannel::Dev).await;
356        assert!(
357            good_db.select_row::<usize>("SELECT * FROM test2").unwrap()()
358                .unwrap()
359                .is_none()
360        );
361    }
362
363    /// Test that DB exists but corrupted (causing recreate)
364    #[gpui::test(iterations = 30)]
365    async fn test_simultaneous_db_corruption(cx: &mut gpui::TestAppContext) {
366        cx.executor().allow_parking();
367
368        enum CorruptedDB {}
369
370        impl Domain for CorruptedDB {
371            const NAME: &str = "db_tests";
372
373            const MIGRATIONS: &[&str] = &[sql!(CREATE TABLE test(value);)];
374        }
375
376        enum GoodDB {}
377
378        impl Domain for GoodDB {
379            const NAME: &str = "db_tests"; //Notice same name
380            const MIGRATIONS: &[&str] = &[sql!(CREATE TABLE test2(value);)]; // But different migration
381        }
382
383        let tempdir = tempfile::Builder::new()
384            .prefix("DbTests")
385            .tempdir()
386            .unwrap();
387        {
388            // Setup the bad database
389            let corrupt_db =
390                open_db::<CorruptedDB>(tempdir.path(), release_channel::ReleaseChannel::Dev).await;
391            assert!(corrupt_db.persistent());
392        }
393
394        // Try to connect to it a bunch of times at once
395        let mut guards = vec![];
396        for _ in 0..10 {
397            let tmp_path = tempdir.path().to_path_buf();
398            let guard = thread::spawn(move || {
399                let good_db = gpui::block_on(open_db::<GoodDB>(
400                    tmp_path.as_path(),
401                    release_channel::ReleaseChannel::Dev,
402                ));
403                assert!(
404                    good_db.select_row::<usize>("SELECT * FROM test2").unwrap()()
405                        .unwrap()
406                        .is_none()
407                );
408            });
409
410            guards.push(guard);
411        }
412
413        for guard in guards.into_iter() {
414            assert!(guard.join().is_ok());
415        }
416    }
417}
418