//! 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; 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; #[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 `/hub.db`. pub async fn open(config: &Config) -> anyhow::Result { let options = SqliteConnectOptions::new() .filename(config.db_path()) .create_if_missing(true) .journal_mode(SqliteJournalMode::Wal) .synchronous(SqliteSynchronous::Normal) .foreign_keys(true) .busy_timeout(BUSY_TIMEOUT); migrate(options.clone()).await?; let pool = SqlitePoolOptions::new().max_connections(8).connect_with(options).await?; Ok(pool) } /// 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. async fn migrate(options: SqliteConnectOptions) -> anyhow::Result<()> { use sqlx::Connection; let mut conn = sqlx::SqliteConnection::connect_with(&options.foreign_keys(false)).await?; sqlx::migrate!("./migrations").run(&mut conn).await?; let broken: Vec<(String,)> = sqlx::query_as("SELECT \"table\" FROM pragma_foreign_key_check").fetch_all(&mut conn).await?; if let Some((table,)) = broken.first() { anyhow::bail!("after the migrations, {} rows break a foreign key (the first in {table})", broken.len()); } conn.close().await?; Ok(()) } /// 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. pub async fn begin_write(pool: &Pool) -> sqlx::Result> { pool.begin_with("BEGIN IMMEDIATE").await } } /// 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 { 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 { 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> { 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). pub fn json_text(col: &str, key: &str) -> String { if cfg!(feature = "postgres") { format!("(CAST(NULLIF({col}, '') AS json) ->> '{key}')") } else { format!("json_extract({col}, '$.{key}')") } } /// Milliseconds since the epoch, the hub's timestamp unit (as `oxy`'s). pub fn now_ms() -> i64 { std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as i64).unwrap_or(0) }