cel-memory-duckdb 0.1.0

//! Embedded SQL migrations bundled via `include_str!`.

use duckdb::{params, Connection};

use crate::error::DuckdbMemoryError;

struct Migration {
    version: i64,
    name: &'static str,
    sql: &'static str,
}

const MIGRATIONS: &[Migration] = &[Migration {
    version: 1,
    name: "001_initial.sql",
    sql: include_str!("../migrations/001_initial.sql"),
}];

/// Highest schema version this build can apply.
#[allow(dead_code)]
pub const LATEST_VERSION: i64 = 1;

/// Run all pending migrations against the connection.
pub fn run(conn: &Connection) -> Result<(), DuckdbMemoryError> {
    conn.execute(
        "CREATE TABLE IF NOT EXISTS _cel_memory_schema_version (
            version BIGINT PRIMARY KEY,
            applied_at TIMESTAMPTZ NOT NULL
        )",
        [],
    )?;

    let current: i64 = conn
        .query_row(
            "SELECT COALESCE(MAX(version), 0) FROM _cel_memory_schema_version",
            [],
            |row| row.get(0),
        )
        .unwrap_or(0);

    for m in MIGRATIONS {
        if m.version <= current {
            continue;
        }
        tracing::info!(version = m.version, name = m.name, "applying migration");
        conn.execute_batch("BEGIN TRANSACTION").ok();
        let result = (|| {
            conn.execute_batch(m.sql)
                .map_err(|source| DuckdbMemoryError::Migration {
                    name: m.name.to_string(),
                    source,
                })?;
            conn.execute(
                "INSERT INTO _cel_memory_schema_version(version, applied_at) VALUES (?, ?)",
                params![m.version, chrono::Utc::now()],
            )?;
            Ok::<(), DuckdbMemoryError>(())
        })();
        if result.is_ok() {
            conn.execute_batch("COMMIT").ok();
        } else {
            conn.execute_batch("ROLLBACK").ok();
            result?;
        }
    }
    Ok(())
}