geph5-client 0.3.0

use anyctx::AnyCtx;
use event_listener::Event;
use sqlx::{Row, pool::PoolOptions};
use sqlx::{SqlitePool, sqlite::SqliteConnectOptions};
use std::str::FromStr;
use stdcode::StdcodeSerializeExt;

use crate::client::{Config, CtxField};

pub static DATABASE: CtxField<SqlitePool> = |ctx| {
    let db_path = ctx
        .init()
        .cache
        .as_ref()
        .map(|s| s.to_string_lossy().to_string())
        .unwrap_or_else(|| {
            dirs::config_dir()
                .map(|config_dir| {
                    config_dir
                        .join(format!(
                            "geph5-persist-{}.db",
                            hex::encode(blake3::hash(&ctx.init().credentials.stdcode()).as_bytes())
                        ))
                        .to_string_lossy()
                        .to_string()
                })
                .unwrap_or_else(|| {
                    tracing::error!(
                        "No cache path configured and dirs::config_dir() unavailable; falling back \
                         to in-memory database. Set the `cache` field in geph5-client's config to \
                         avoid hitting broker rate limits."
                    );
                    ":memory:".into()
                })
        });
    tracing::debug!("INITIALIZING DATABASE");
    let options = SqliteConnectOptions::from_str(&db_path)
        .unwrap()
        .create_if_missing(true)
        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
        .synchronous(sqlx::sqlite::SqliteSynchronous::Normal);

    // `connect_lazy_with` builds the pool synchronously (no `block_on`, so this
    // works from inside the tokio runtime); the schema is created on the single
    // connection's first use via `after_connect`.
    // (min/max == 1 prevents SQLITE_BUSY unless we have concurrent apps.)
    PoolOptions::new()
        .min_connections(1)
        .max_connections(1)
        .after_connect(|conn, _meta| {
            Box::pin(async move {
                sqlx::query(
                    "CREATE TABLE IF NOT EXISTS misc (
                        key TEXT PRIMARY KEY,
                        value BLOB NOT NULL
                    );",
                )
                .execute(&mut *conn)
                .await?;
                Ok(())
            })
        })
        .connect_lazy_with(options)
};

static EVENT: CtxField<Event> = |_| Event::new();

pub async fn db_write(ctx: &AnyCtx<Config>, key: &str, value: &[u8]) -> Result<(), sqlx::Error> {
    sqlx::query("INSERT INTO misc (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value")
        .bind(key)
        .bind(value)
        .execute(ctx.get(DATABASE))
        .await?;
    ctx.get(EVENT).notify(usize::MAX);
    Ok(())
}

pub async fn db_read(ctx: &AnyCtx<Config>, key: &str) -> Result<Option<Vec<u8>>, sqlx::Error> {
    let result = sqlx::query("SELECT value FROM misc WHERE key = ?")
        .bind(key)
        .fetch_optional(ctx.get(DATABASE))
        .await?
        .map(|row| row.get("value"));
    Ok(result)
}

pub async fn db_read_or_wait(ctx: &AnyCtx<Config>, key: &str) -> Result<Vec<u8>, sqlx::Error> {
    loop {
        let event = ctx.get(EVENT).listen();
        let result = db_read(ctx, key).await?;
        match result {
            Some(val) => return Ok(val),
            None => event.await,
        }
    }
}

pub async fn db_remove(ctx: &AnyCtx<Config>, key: &str) -> Result<(), sqlx::Error> {
    sqlx::query("DELETE FROM misc WHERE key = ?")
        .bind(key)
        .execute(ctx.get(DATABASE))
        .await?;
    ctx.get(EVENT).notify(usize::MAX);
    Ok(())
}