arcature-cli 2026.1.1

//! Advisory-lock-guarded migration delegation (Phase 4 spec §14, §20, §25).
//!
//! `arc db migrate` (and `rollback`/`fresh`/`reset`/`refresh`) shells out to
//! the application-owned SeaORM migrator binary (`cargo run -p migration -- …`),
//! but first takes a PostgreSQL session-level advisory lock to serialize
//! concurrent migration runners. This is the smallest PG-specific guard
//! that closes the concurrent-migration gap verified in Commit 5: SeaORM
//! Migration 2.0.1's `Migrator::up` does not take an advisory lock, so two
//! concurrent runners can both read the same pending list and execute the
//! same DDL before either commits (Phase 4 spec §25).
//!
//! The advisory lock is session-level (`pg_advisory_lock`), so the command
//! keeps its owning SQLx connection checked out for the entire child-process
//! lifetime and then closes that exact session. Pool checkout must never be
//! used as the ownership boundary: another pool operation may use a different
//! backend session. If the CLI process is killed, PostgreSQL releases the lock
//! when that backend connection terminates.

use std::env;

use arcature_db::{Db, DbConfig};

use crate::error::{CommandError, DbCommandError};
use crate::process::{ProcessSpec, run};
use crate::project::discover;

/// The Arcature migration advisory lock key. A fixed constant so all
/// Arcature instances — and all processes — use the same lock, serializing
/// concurrent migration runners across the entire system.
///
/// `0x41524354` = "ARCT" in ASCII (A=0x41, R=0x52, C=0x43, T=0x54).
const ADVISORY_LOCK_KEY: i64 = 0x4152_4354;

/// Apply pending migrations (`arc db migrate`).
pub(crate) fn run_migrate() -> Result<(), CommandError> {
    run_with_lock(&["up"])
}

/// Roll back applied migrations (`arc db rollback [--steps N]`).
pub(crate) fn run_rollback(steps: Option<u32>) -> Result<(), CommandError> {
    let args: Vec<String> = match steps {
        Some(n) => vec!["down".into(), "--num".into(), n.to_string()],
        None => vec!["down".into()],
    };
    run_with_lock_owned(&args)
}

/// Drop all tables and reapply all migrations (`arc db fresh --force`).
pub(crate) fn run_fresh() -> Result<(), CommandError> {
    run_with_lock(&["fresh"])
}

/// Roll back all migrations (`arc db reset --force`).
pub(crate) fn run_reset() -> Result<(), CommandError> {
    run_with_lock(&["reset"])
}

/// Roll back all, then reapply all (`arc db refresh --force`).
pub(crate) fn run_refresh() -> Result<(), CommandError> {
    run_with_lock(&["refresh"])
}

/// Run a migration subcommand with the advisory lock held. The lock is
/// acquired before the child process spawns and released when the `Db` is
/// closed after the child completes (or fails).
fn run_with_lock(args: &[&str]) -> Result<(), CommandError> {
    let owned: Vec<String> = args.iter().map(|s| (*s).to_string()).collect();
    run_with_lock_owned(&owned)
}

/// Run a migration subcommand with owned string arguments (needed for
/// `rollback --steps N` where N is a runtime number).
fn run_with_lock_owned(args: &[String]) -> Result<(), CommandError> {
    let database_url = env::var("DATABASE_URL").map_err(|_| DbCommandError::MissingDatabaseUrl)?;
    let project = discover()?;

    // Create a single-threaded Tokio runtime for the async DB work. The child
    // process runs synchronously while the locked connection remains checked
    // out, so its session-level lock cannot be confused with another pool
    // checkout.
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|e| DbCommandError::Runtime(e.to_string()))?;

    // Connect and acquire the advisory lock on one checked-out connection.
    // Holding this `PoolConnection` is essential: session-level locks belong
    // to a PostgreSQL backend session, not to a `PgPool`.
    let (db, mut connection) = runtime.block_on(async {
        let db = Db::connect(DbConfig::new(&database_url)?.application_name("arcature-db-migrate"))
            .await?;

        let mut connection = db
            .sqlx()
            .acquire()
            .await
            .map_err(DbCommandError::AdvisoryLock)?;

        arcature_db::sqlx::query("SELECT pg_advisory_lock($1)")
            .bind(ADVISORY_LOCK_KEY)
            .execute(&mut *connection)
            .await
            .map_err(DbCommandError::AdvisoryLock)?;

        Ok::<_, DbCommandError>((db, connection))
    })?;

    // Spawn the migrator child process. DATABASE_URL is inherited from the
    // environment. The child runs in the project root so `cargo run -p
    // migration` finds the workspace.
    let mut spec_args: Vec<String> =
        vec!["run".into(), "-p".into(), "migration".into(), "--".into()];
    spec_args.extend(args.iter().cloned());

    let spec = ProcessSpec::new("cargo", project.root().to_path_buf());
    let spec = spec_args
        .iter()
        .fold(spec, |spec, arg| spec.arg(arg.clone()));

    let result = run(&spec);

    // Close the exact lock-owning session, even when the child fails. Merely
    // returning it to the pool would be unsafe: its session-level advisory
    // lock would survive and could later be reused by an unrelated checkout.
    // Closing the pool afterwards is a defensive cleanup of the remaining
    // connection resources.
    let cleanup = runtime.block_on(async {
        let unlock = arcature_db::sqlx::query_scalar::<_, bool>("SELECT pg_advisory_unlock($1)")
            .bind(ADVISORY_LOCK_KEY)
            .fetch_one(&mut *connection)
            .await
            .map_err(DbCommandError::AdvisoryLock)
            .and_then(|released| {
                released
                    .then_some(())
                    .ok_or(DbCommandError::AdvisoryLockNotOwned)
            });
        let close = connection
            .close()
            .await
            .map_err(DbCommandError::AdvisoryLock);
        db.close().await;
        unlock?;
        close
    });

    cleanup?;

    // Report the child's result. A process failure surfaces as
    // CommandError::Process (the migrator exited non-zero, preserving the
    // error per Phase 4 spec §14).
    result?;

    Ok(())
}