Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990//! SQLite pool initialization and projection-cursor persistence.//!//! Two durable stores back the server://! 1. Hydrant's fjall store (`HYDRANT_DATABASE_PATH`) for raw ATProto records.//! 2. This SQLite projection database (`DATABASE_URL`) for app-specific derived//! views. The pool here owns the second.
use std::str::FromStr;
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};use sqlx::{Sqlite, SqlitePool, Transaction};
use super::config::ServerConfig;
/// Open the SQLite pool, enable WAL mode, run migrations, and return the pool.////// WAL mode is set on the connection options (not in a migration) because SQLite/// silently ignores `journal_mode` changes inside the transaction sqlx uses to/// run migrations.pub async fn init_db(cfg: &ServerConfig) -> anyhow::Result<SqlitePool> { ensure_data_dir(&cfg.database_url)?;
let options = SqliteConnectOptions::from_str(&cfg.database_url)? .create_if_missing(true) .journal_mode(SqliteJournalMode::Wal);
let pool = SqlitePoolOptions::new() .max_connections(5) .connect_with(options) .await?;
sqlx::migrate!("./migrations").run(&pool).await?; tracing::info!("sqlite migrations applied"); Ok(pool)}
fn ensure_data_dir(db_url: &str) -> anyhow::Result<()> { if let Some(path) = db_url.strip_prefix("sqlite:") && let Some(parent) = std::path::Path::new(path).parent() { std::fs::create_dir_all(parent)?; } Ok(())}
struct CursorRow { value: String,}
/// Load the last persisted projection cursor (the Hydrant event id), if any.pub async fn load_cursor(db: &SqlitePool) -> anyhow::Result<Option<u64>> { let row: Option<CursorRow> = sqlx::query_as!( CursorRow, "SELECT value FROM projection_state WHERE key = ?", "cursor" ) .fetch_optional(db) .await?; Ok(row.and_then(|r| r.value.parse().ok()))}
/// Persist the projection cursor outside a projection transaction.////// Used for events that are acknowledged but not projected (e.g. account status/// events), so the durable cursor advances past them and a reconnect/resume does/// not replay them indefinitely.pub async fn advance_cursor(db: &SqlitePool, cursor: u64) -> anyhow::Result<()> { let mut tx = db.begin().await?; save_cursor(&mut tx, cursor).await?; tx.commit().await?; Ok(())}
/// Persist the projection cursor inside an in-flight projection transaction.pub async fn save_cursor(tx: &mut Transaction<'_, Sqlite>, cursor: u64) -> anyhow::Result<()> { let now = chrono::Utc::now() .timestamp_nanos_opt() .expect("current timestamp must fit in i64 nanoseconds"); sqlx::query!( "INSERT INTO projection_state (key, value, updated_at) VALUES ('cursor', ?, ?) \ ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at \ WHERE CAST(excluded.value AS INTEGER) > CAST(projection_state.value AS INTEGER)", cursor.to_string(), now ) .execute(&mut **tx) .await?; Ok(())}