Coverage / crates/hub/src/db/mod.rs 94.6%

//! SQLite through `sqlx` (hub spec §3.1, §17.2): WAL, foreign keys on,

//! migrations embedded and applied at start; or PostgreSQL, built with the

//! `postgres` feature (#211). One module per table group.

pub mod accounts;

pub mod activity;

#[cfg(feature = "postgres")]

pub mod copy;

pub mod flow;

pub mod hooks;

pub mod issues;

pub mod notify;

pub mod projects;

pub mod proposals;

pub mod push_protection;

pub mod queue;

pub mod releases;

pub mod repos;

pub mod runner_updates;

pub mod runners;

pub mod sessions;

pub mod terraform;

pub mod tlog;

#[cfg(not(feature = "postgres"))]

pub use sqlite::{begin_write, open, Db};

#[cfg(feature = "postgres")]

pub use postgres::{begin_write, open, Db};

pub type Pool = sqlx::Pool<Db>;

#[cfg(not(feature = "postgres"))]

mod sqlite {

    use super::Pool;

    use crate::config::Config;

    use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous};

    pub type Db = sqlx::Sqlite;

    /// How long a statement waits for SQLite's write lock. Most writers are

    /// background clients (runners appending logs, webhook deliveries), and a

    /// commit can stall for seconds while a build saturates the host's disk, so

    /// waiting beats failing.

    const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

    /// Open (creating if needed) and migrate `<data>/hub.db`.

    /// Apply the migrations on a connection of their own with foreign keys

    /// off, as SQLite's way to change a table (rebuild it, drop the old one,

    /// rename) needs: with them on, dropping a table its children reference

    /// deletes the children too. Every reference must hold afterwards.

        use sqlx::Connection;

    /// A write transaction that takes SQLite's write lock at `BEGIN`

    /// (`BEGIN IMMEDIATE`). A deferred transaction that reads and then writes

    /// fails at once with "database is locked" when another writer holds the

    /// lock, whatever the busy timeout; an immediate one waits its turn.

}

/// PostgreSQL (#211), for a hub on more than one host or a managed database.

/// The same queries as SQLite's; the schema is `migrations-pg`, which starts

/// as SQLite's 0001-0072 in one and takes each later change written for it.

#[cfg(feature = "postgres")]

mod postgres {

    use super::Pool;

    use crate::config::Config;

    use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

    use sqlx::Executor;

    pub type Db = sqlx::Postgres;

    /// Connect to `[database] url`, use its schema, and migrate.

    pub async fn open(config: &Config) -> anyhow::Result<Pool> {

        if config.database.url.is_empty() {

            anyhow::bail!("this hub is built for PostgreSQL: set [database] url (or OXYHUB_DATABASE_URL)");

        }

        let options: PgConnectOptions = config.database.url.parse()?;

        let schema = schema_name(config);

        if let Some(schema) = &schema {

            let pool = PgPoolOptions::new().max_connections(1).connect_with(options.clone()).await?;

            pool.execute(format!("CREATE SCHEMA IF NOT EXISTS \"{schema}\"").as_str()).await?;

            pool.close().await;

        }

        let path = match &schema {

            // Plain (letters, digits, `_`), so it needs no quoting here.

            Some(s) => format!("{s},public"),

            None => "public".to_string(),

        };

        let options = options.options([("search_path", path.as_str())]);

        let pool = PgPoolOptions::new().max_connections(16).connect_with(options).await?;

        extension(&pool).await?;

        sqlx::migrate!("./migrations-pg").run(&pool).await?;

        Ok(pool)

    }

    /// `citext`, for names compared without case, made once in `public` so

    /// every schema sees it. Two hubs starting at once may race to make it.

    async fn extension(pool: &Pool) -> anyhow::Result<()> {

        let made = pool.execute("CREATE EXTENSION IF NOT EXISTS citext WITH SCHEMA public").await;

        if let Err(e) = made {

            let there: bool =

                sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'citext')").fetch_one(pool).await?;

            if !there {

                return Err(e.into());

            }

        }

        Ok(())

    }

    /// The schema to use: none for `public`; `*` names it after the data

    /// directory, so hubs (and tests) can share one database.

    fn schema_name(config: &Config) -> Option<String> {

        let s = match config.database.schema.as_str() {

            "" | "public" => return None,

            "*" => {

                let dir = config.data.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default();

                format!("hub_{dir}")

            }

            s => s.to_string(),

        };

        Some(s.chars().map(|c| if c.is_ascii_alphanumeric() { c.to_ascii_lowercase() } else { '_' }).collect())

    }

    /// One writer at a time, as SQLite's `BEGIN IMMEDIATE` gives: callers

    /// read and then write (the next number, a check before an insert) and

    /// rely on nothing changing between. A transaction-scoped advisory lock.

    pub async fn begin_write(pool: &Pool) -> sqlx::Result<sqlx::Transaction<'static, Db>> {

        let mut tx = pool.begin().await?;

        sqlx::query("SELECT pg_advisory_xact_lock(7300211)").execute(&mut *tx).await?;

        Ok(tx)

    }

}

/// SQL for the text at `key` of the JSON object in column `col` (null when

/// the column is empty or has no such key).

    } else {

    }

/// Milliseconds since the epoch, the hub's timestamp unit (as `oxy`'s).