typedpg_cli 0.5.0

//! The migration commands behind `cargo typedpg migrate`, shared with the
//! TypeScript CLI (`typedpg migrate`, in `typedpg_ts`), which configures
//! them from `typedpg.config.json` instead of `Cargo.toml`.

use std::fs;
use std::path::{Path, PathBuf};

use chrono::Utc;
use typedpg::migrate::{self, MigrationSource, MigrationsConfig};

#[derive(clap::Subcommand)]
pub enum MigrateAction {
    /// Apply all pending migrations
    Up,
    /// Show migration status
    Status,
    /// Revert a migration (defaults to the last applied)
    Down {
        /// Name of migration to revert (defaults to last applied)
        name: Option<String>,
        /// Force revert even without a .down.sql file
        #[arg(long)]
        force: bool,
    },
    /// Create a new empty migration file
    Create {
        /// Migration name (e.g., "add_users_table")
        name: String,
    },
}

/// Write an empty `<timestamp>_<name>.sql` and its `.down.sql` in `dir`;
/// returns the two paths.
///
/// Neither file is overwritten: a migration written by hand, or one created
/// a moment ago, is never silently emptied.
pub fn create_migration(dir: &Path, name: &str) -> std::io::Result<(PathBuf, PathBuf)> {
    let timestamp = Utc::now().format("%Y%m%d%H%M%S");
    fs::create_dir_all(dir)?;
    let up_file = dir.join(format!("{timestamp}_{name}.sql"));
    let down_file = dir.join(format!("{timestamp}_{name}.down.sql"));
    write_new(&up_file)?;
    if let Err(e) = write_new(&down_file) {
        // Nothing is left behind: a migration whose down file could not be
        // written is a half-created one, and the runner would apply it.
        let _ = fs::remove_file(&up_file);
        return Err(e);
    }
    Ok((up_file, down_file))
}

/// Write an empty file, failing when it already exists.
fn write_new(path: &Path) -> std::io::Result<()> {
    fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(path)
        .map(|_| ())
        .map_err(|e| {
            if e.kind() == std::io::ErrorKind::AlreadyExists {
                std::io::Error::new(
                    e.kind(),
                    format!(
                        "refusing to overwrite {}: a file with that name already exists (the \
                         migration's name carries a second-resolution timestamp — retry in a \
                         moment, or use a different name)",
                        path.display()
                    ),
                )
            } else {
                e
            }
        })
}

/// Connect honoring the URL's `sslmode` as libpq does: `disable` is plain,
/// `require` is TLS or nothing, and `prefer` (the default) tries TLS and
/// falls back to a plain connection when that attempt fails — for instance
/// on a server that does not offer TLS, or when the TLS handshake itself
/// fails.
///
/// libpq does not verify the certificate for `require` and `prefer`: it only
/// checks that the server presents one. Verifying would make a server with
/// a self-signed certificate unreachable — and `prefer` would then fall
/// back to a plain connection, sending the password and the migration SQL
/// in the clear where libpq keeps the connection encrypted.
///
/// The modes that do verify (`verify-ca`, `verify-full`) are not supported:
/// the connection library behind this CLI has no way to ask for them.
pub async fn connect(
    database_url: &str,
) -> Result<tokio_postgres::Client, Box<dyn std::error::Error>> {
    use tokio_postgres::config::SslMode;

    check_sslmode(database_url)?;
    let config: tokio_postgres::Config = database_url
        .parse()
        .map_err(|e| format!("invalid DATABASE_URL: {e}"))?;
    match config.get_ssl_mode() {
        SslMode::Disable => connect_plain(&config).await,
        SslMode::Require => connect_tls(&config).await,
        _ => match connect_tls(&config).await {
            Ok(client) => Ok(client),
            Err(tls_err) => {
                eprintln!(
                    "typedpg: the TLS handshake failed ({}); falling back to an unencrypted \
                     connection — the password and the migration SQL will travel in clear",
                    tls_err
                );
                let mut plain = config.clone();
                plain.ssl_mode(SslMode::Disable);
                connect_plain(&plain)
                    .await
                    .map_err(|e| format!("{e} (after a TLS attempt failed: {tls_err})").into())
            }
        },
    }
}

/// The `sslmode` the URL asks for, when it is one this CLI cannot honor.
///
/// The connection library takes `disable`, `prefer` and `require`; the modes
/// that verify the server's certificate (`verify-ca`, `verify-full`) are
/// rejected by it with a message that says nothing about which parameter was
/// wrong. The CLI says it instead — and says what it can do.
fn check_sslmode(database_url: &str) -> Result<(), Box<dyn std::error::Error>> {
    let Some(value) = parameter(database_url, "sslmode") else {
        return Ok(());
    };
    match value.as_str() {
        "disable" | "prefer" | "require" | "allow" => Ok(()),
        other => Err(format!(
            "unsupported sslmode `{other}`: this CLI connects with `disable`, `prefer` or \
             `require`, and never verifies the server's certificate"
        )
        .into()),
    }
}

/// The value of the `name` parameter in a libpq connection string or URL.
fn parameter(url: &str, name: &str) -> Option<String> {
    let value = if let Some(query) = url.split_once('?').map(|(_, q)| q) {
        query
            .split('&')
            .find_map(|pair| pair.strip_prefix(&format!("{name}=")))
    } else {
        url.split_whitespace()
            .find_map(|word| word.strip_prefix(&format!("{name}=")))
    }?;
    Some(value.trim_matches('\'').to_owned())
}

async fn connect_tls(
    config: &tokio_postgres::Config,
) -> Result<tokio_postgres::Client, Box<dyn std::error::Error>> {
    // libpq's `require` and `prefer`: the connection is encrypted, but the
    // certificate is not verified.
    let mut builder = native_tls::TlsConnector::builder();
    builder.danger_accept_invalid_certs(true);
    builder.danger_accept_invalid_hostnames(true);
    let tls_connector = builder.build()?;
    let tls = postgres_native_tls::MakeTlsConnector::new(tls_connector);
    let (client, connection) = config.connect(tls).await?;
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            eprintln!("Database connection error: {e}");
        }
    });
    Ok(client)
}

async fn connect_plain(
    config: &tokio_postgres::Config,
) -> Result<tokio_postgres::Client, Box<dyn std::error::Error>> {
    let (client, connection) = config.connect(tokio_postgres::NoTls).await?;
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            eprintln!("Database connection error: {e}");
        }
    });
    Ok(client)
}

/// Run `action` (any but `Create`, which needs no connection) with the
/// runner settings `config`.
pub async fn run_migrate_action(
    action: MigrateAction,
    client: &mut tokio_postgres::Client,
    source: &MigrationSource,
    config: &MigrationsConfig,
) -> Result<(), Box<dyn std::error::Error>> {
    match action {
        MigrateAction::Up => {
            let applied = migrate::run(client, source, config).await?;
            if applied.is_empty() {
                println!("No pending migrations");
            } else {
                for name in &applied {
                    println!("Applying {name}... done");
                }
                println!("Applied {} migration(s)", applied.len());
            }
        }
        MigrateAction::Status => {
            let statuses = migrate::status(client, source, config).await?;
            if statuses.is_empty() {
                println!("No migrations found");
            } else {
                for s in &statuses {
                    if s.applied {
                        let ts = s
                            .applied_at
                            .map(|t| t.format("%Y-%m-%d %H:%M:%S UTC").to_string())
                            .unwrap_or_default();
                        let drift = if s.drifted { " [MODIFIED]" } else { "" };
                        println!("  \u{2713} {:<30} (applied {ts}){drift}", s.name);
                    } else {
                        println!("  \u{00b7} {:<30} (pending)", s.name);
                    }
                }
            }
        }
        MigrateAction::Down { name, force } => {
            let revert_name = match name {
                Some(n) => n,
                None => {
                    // Find the last applied migration
                    let statuses = migrate::status(client, source, config).await?;
                    statuses
                        .iter()
                        .rev()
                        .find(|s| s.applied)
                        .map(|s| s.name.clone())
                        .ok_or("No applied migrations to revert")?
                }
            };
            migrate::revert(client, source, &revert_name, force, config).await?;
            println!("Reverting {revert_name}... done");
        }
        MigrateAction::Create { .. } => unreachable!(),
    }

    Ok(())
}