OXYGENHUB
◆ v0.9.24 oxygenhub / crates / hub / src / db / mod.rs
7.1 KiB · Rust · last changed A phone as the second factor: Keystore keys approve sign-ins by the number the browser shows, and the sudo step asks a stale session again before sensitive changes (closes #356) 3 days ago source blame history raw
d23464c0 Andy McCall ↶ 1
//! SQLite through `sqlx` (hub spec §3.1, §17.2): WAL, foreign keys on,
3b1bc27f Andy McCall 2
//! migrations embedded and applied at start; or PostgreSQL, built with the
3
//! `postgres` feature (#211). One module per table group.
d23464c0 Andy McCall ↶ 4
19bb31ff Andy McCall ↶ 5
pub mod accounts;
25ab8c73 Andy McCall ↶ 6
pub mod activity;
a22e4aec Andy McCall ↶ 7
pub mod approvals;
1c52d940 Andy McCall ↶ 8
pub mod attachments;
4e2cad00 Andy McCall ↶ 9
#[cfg(feature = "postgres")]
10
pub mod copy;
42fa9554 Andy McCall ↶ 11
pub mod flow;
b64e2537 Andy McCall ↶ 12
pub mod hooks;
18e381ea Andy McCall ↶ 13
pub mod issues;
14
pub mod notify;
e1ec9fbf Andy McCall ↶ 15
pub mod projects;
de505df5 Andy McCall ↶ 16
pub mod proposals;
aeed63c0 Andy McCall ↶ 17
pub mod push;
0ee5c3fb Andy McCall ↶ 18
pub mod push_protection;
dd327d76 Andy McCall ↶ 19
pub mod queue;
3f4b0ea6 Andy McCall ↶ 20
pub mod releases;
d23464c0 Andy McCall ↶ 21
pub mod repos;
8501fd58 Andy McCall ↶ 22
pub mod runner_updates;
b6b7e4cf Andy McCall ↶ 23
pub mod runners;
0024ccb1 Andy McCall ↶ 24
pub mod sessions;
8dcaeeb8 Andy McCall ↶ 25
pub mod terraform;
965b950f Andy McCall ↶ 26
pub mod tlog;
d23464c0 Andy McCall ↶ 27
3b1bc27f Andy McCall 28
#[cfg(not(feature = "postgres"))]
29
pub use sqlite::{begin_write, open, Db};
d23464c0 Andy McCall ↶ 30
3b1bc27f Andy McCall 31
#[cfg(feature = "postgres")]
32
pub use postgres::{begin_write, open, Db};
d23464c0 Andy McCall ↶ 33
3b1bc27f Andy McCall 34
pub type Pool = sqlx::Pool<Db>;
eae4a974 Andy McCall ↶ 35
3b1bc27f Andy McCall 36
#[cfg(not(feature = "postgres"))]
37
mod sqlite {
38
    use super::Pool;
39
    use crate::config::Config;
40
    use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous};
41
42
    pub type Db = sqlx::Sqlite;
43
44
    /// How long a statement waits for SQLite's write lock. Most writers are
45
    /// background clients (runners appending logs, webhook deliveries), and a
46
    /// commit can stall for seconds while a build saturates the host's disk, so
47
    /// waiting beats failing.
48
    const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
49
50
    /// Open (creating if needed) and migrate `<data>/hub.db`.
51
    pub async fn open(config: &Config) -> anyhow::Result<Pool> {
52
        let options = SqliteConnectOptions::new()
53
            .filename(config.db_path())
54
            .create_if_missing(true)
55
            .journal_mode(SqliteJournalMode::Wal)
56
            .synchronous(SqliteSynchronous::Normal)
57
            .foreign_keys(true)
58
            .busy_timeout(BUSY_TIMEOUT);
59
        migrate(options.clone()).await?;
60
        let pool = SqlitePoolOptions::new().max_connections(8).connect_with(options).await?;
61
        Ok(pool)
62
    }
63
64
    /// Apply the migrations on a connection of their own with foreign keys
65
    /// off, as SQLite's way to change a table (rebuild it, drop the old one,
66
    /// rename) needs: with them on, dropping a table its children reference
67
    /// deletes the children too. Every reference must hold afterwards.
68
    async fn migrate(options: SqliteConnectOptions) -> anyhow::Result<()> {
69
        use sqlx::Connection;
70
        let mut conn = sqlx::SqliteConnection::connect_with(&options.foreign_keys(false)).await?;
71
        sqlx::migrate!("./migrations").run(&mut conn).await?;
72
        let broken: Vec<(String,)> = sqlx::query_as("SELECT \"table\" FROM pragma_foreign_key_check").fetch_all(&mut conn).await?;
73
        if let Some((table,)) = broken.first() {
74
            anyhow::bail!("after the migrations, {} rows break a foreign key (the first in {table})", broken.len());
75
        }
76
        conn.close().await?;
77
        Ok(())
78
    }
79
80
    /// A write transaction that takes SQLite's write lock at `BEGIN`
81
    /// (`BEGIN IMMEDIATE`). A deferred transaction that reads and then writes
82
    /// fails at once with "database is locked" when another writer holds the
83
    /// lock, whatever the busy timeout; an immediate one waits its turn.
84
    pub async fn begin_write(pool: &Pool) -> sqlx::Result<sqlx::Transaction<'static, Db>> {
85
        pool.begin_with("BEGIN IMMEDIATE").await
86
    }
d23464c0 Andy McCall ↶ 87
}
3b1bc27f Andy McCall 88
89
/// PostgreSQL (#211), for a hub on more than one host or a managed database.
90
/// The same queries as SQLite's; the schema is `migrations-pg`, which starts
91
/// as SQLite's 0001-0072 in one and takes each later change written for it.
92
#[cfg(feature = "postgres")]
93
mod postgres {
94
    use super::Pool;
95
    use crate::config::Config;
96
    use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
97
    use sqlx::Executor;
d23464c0 Andy McCall ↶ 98
3b1bc27f Andy McCall 99
    pub type Db = sqlx::Postgres;
100
101
    /// Connect to `[database] url`, use its schema, and migrate.
102
    pub async fn open(config: &Config) -> anyhow::Result<Pool> {
103
        if config.database.url.is_empty() {
104
            anyhow::bail!("this hub is built for PostgreSQL: set [database] url (or OXYHUB_DATABASE_URL)");
105
        }
106
        let options: PgConnectOptions = config.database.url.parse()?;
107
        let schema = schema_name(config);
108
        if let Some(schema) = &schema {
109
            let pool = PgPoolOptions::new().max_connections(1).connect_with(options.clone()).await?;
110
            pool.execute(format!("CREATE SCHEMA IF NOT EXISTS \"{schema}\"").as_str()).await?;
111
            pool.close().await;
112
        }
113
        let path = match &schema {
114
            // Plain (letters, digits, `_`), so it needs no quoting here.
115
            Some(s) => format!("{s},public"),
116
            None => "public".to_string(),
117
        };
118
        let options = options.options([("search_path", path.as_str())]);
119
        let pool = PgPoolOptions::new().max_connections(16).connect_with(options).await?;
120
        extension(&pool).await?;
121
        sqlx::migrate!("./migrations-pg").run(&pool).await?;
122
        Ok(pool)
745d225d Andy McCall ↶ 123
    }
3b1bc27f Andy McCall 124
125
    /// `citext`, for names compared without case, made once in `public` so
126
    /// every schema sees it. Two hubs starting at once may race to make it.
127
    async fn extension(pool: &Pool) -> anyhow::Result<()> {
128
        let made = pool.execute("CREATE EXTENSION IF NOT EXISTS citext WITH SCHEMA public").await;
129
        if let Err(e) = made {
130
            let there: bool =
131
                sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'citext')").fetch_one(pool).await?;
132
            if !there {
133
                return Err(e.into());
134
            }
135
        }
136
        Ok(())
137
    }
138
139
    /// The schema to use: none for `public`; `*` names it after the data
140
    /// directory, so hubs (and tests) can share one database.
141
    fn schema_name(config: &Config) -> Option<String> {
142
        let s = match config.database.schema.as_str() {
143
            "" | "public" => return None,
144
            "*" => {
145
                let dir = config.data.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default();
146
                format!("hub_{dir}")
147
            }
148
            s => s.to_string(),
149
        };
150
        Some(s.chars().map(|c| if c.is_ascii_alphanumeric() { c.to_ascii_lowercase() } else { '_' }).collect())
151
    }
152
153
    /// One writer at a time, as SQLite's `BEGIN IMMEDIATE` gives: callers
154
    /// read and then write (the next number, a check before an insert) and
155
    /// rely on nothing changing between. A transaction-scoped advisory lock.
156
    pub async fn begin_write(pool: &Pool) -> sqlx::Result<sqlx::Transaction<'static, Db>> {
157
        let mut tx = pool.begin().await?;
158
        sqlx::query("SELECT pg_advisory_xact_lock(7300211)").execute(&mut *tx).await?;
159
        Ok(tx)
160
    }
745d225d Andy McCall ↶ 161
}
162
3b1bc27f Andy McCall 163
/// SQL for the text at `key` of the JSON object in column `col` (null when
164
/// the column is empty or has no such key).
165
pub fn json_text(col: &str, key: &str) -> String {
166
    if cfg!(feature = "postgres") {
167
        format!("(CAST(NULLIF({col}, '') AS json) ->> '{key}')")
168
    } else {
169
        format!("json_extract({col}, '$.{key}')")
170
    }
171
}
172
d23464c0 Andy McCall ↶ 173
/// Milliseconds since the epoch, the hub's timestamp unit (as `oxy`'s).
174
pub fn now_ms() -> i64 {
175
    std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0)
941abdd6 Andy McCall ↶ 176
}