icydb-cli 0.275.0

//! Deployed-only source-migration command execution and rendering.

use candid::Decode;
use icydb::db::{SchemaMigrationPhase, SchemaMigrationStatusPage};
use icydb::{Error, db::SchemaMigrationCommand as RuntimeMigrationCommand};

use crate::{
    cli::{CanisterTarget, ConfirmedMigrationTarget, SchemaMigrationCommand},
    endpoint::{SCHEMA_MIGRATE_ENDPOINT, SCHEMA_MIGRATION_ENDPOINT},
    icp::require_created_canister,
    observability::{call_query, call_update, endpoint_result_error, render_hex_lower},
};

// Public status omits cleanup's private cursor, so identical pages cannot prove
// a stall. Bound paid calls per invocation instead; retained cleanup is resumable.
const MAX_ABORT_UPDATE_CALLS: usize = 1_024;

pub(super) fn run_schema_migration_command(command: SchemaMigrationCommand) -> Result<(), String> {
    let status = match &command {
        SchemaMigrationCommand::Status(target) => load_status(target),
        SchemaMigrationCommand::Advance(target) => {
            let status = load_status(target)?;
            advance(target, &status)
        }
        SchemaMigrationCommand::Run(target) => run_to_terminal(target),
        SchemaMigrationCommand::Abort(args) => run_confirmed_abort(args),
        SchemaMigrationCommand::Adopt(args) => run_confirmed(args, "adopt", adopt),
    }?;
    print_status(&status);

    // A successful endpoint call is not proof that the requested operation
    // succeeded. Preserve its status and findings before reporting CLI failure.
    let phase = status.phase();
    let (operation, succeeded) = match command {
        SchemaMigrationCommand::Status(_) | SchemaMigrationCommand::Adopt(_) => return Ok(()),
        SchemaMigrationCommand::Advance(_) => (
            "advance",
            !matches!(
                phase,
                SchemaMigrationPhase::Rejected | SchemaMigrationPhase::Aborted
            ),
        ),
        SchemaMigrationCommand::Run(_) => ("run", phase == SchemaMigrationPhase::Applied),
        SchemaMigrationCommand::Abort(_) => ("abort", phase == SchemaMigrationPhase::Aborted),
    };
    if !succeeded {
        return Err(format!(
            "schema migration {operation} did not succeed: phase {phase:?}"
        ));
    }

    Ok(())
}

fn run_confirmed_abort(
    args: &ConfirmedMigrationTarget,
) -> Result<SchemaMigrationStatusPage, String> {
    if !args.confirmed() {
        return Err("schema migration abort requires explicit --yes confirmation".to_string());
    }
    let target = args.target();
    let mut status = load_status(target)?;
    let database = status.database_identity();
    let plan = status
        .plan_digest()
        .ok_or_else(|| "deployed schema has no migration plan to abort".to_string())?;
    for completed_calls in 1..=MAX_ABORT_UPDATE_CALLS {
        let next = abort(target, &status)?;
        if next.database_identity() != database || next.plan_digest() != Some(plan) {
            return Err(
                "deployed migration database or plan identity changed while aborting".into(),
            );
        }
        if matches!(
            next.phase(),
            SchemaMigrationPhase::Aborted | SchemaMigrationPhase::Applied
        ) {
            return Ok(next);
        }
        status = next;
        if completed_calls % 128 == 0 {
            println!(
                "schema migration abort: {completed_calls} update calls completed; phase {:?}",
                status.phase()
            );
        }
    }
    print_status(&status);
    Err(format!(
        "schema migration abort reached its limit of {MAX_ABORT_UPDATE_CALLS} update calls; inspect migration status, then rerun abort --yes to continue cleanup"
    ))
}

fn run_confirmed(
    args: &ConfirmedMigrationTarget,
    operation: &str,
    run: impl FnOnce(
        &CanisterTarget,
        &SchemaMigrationStatusPage,
    ) -> Result<SchemaMigrationStatusPage, String>,
) -> Result<SchemaMigrationStatusPage, String> {
    if !args.confirmed() {
        return Err(format!(
            "schema migration {operation} requires explicit --yes confirmation"
        ));
    }
    let status = load_status(args.target())?;
    run(args.target(), &status)
}

fn run_to_terminal(target: &CanisterTarget) -> Result<SchemaMigrationStatusPage, String> {
    let mut status = load_status(target)?;
    let database = status.database_identity();
    let plan = status
        .plan_digest()
        .ok_or_else(|| "deployed schema has no migration plan to run".to_string())?;
    loop {
        if terminal(status.phase()) {
            return Ok(status);
        }
        let next = advance(target, &status)?;
        if next.database_identity() != database || next.plan_digest() != Some(plan) {
            return Err(
                "deployed migration database or plan identity changed while running".into(),
            );
        }
        if next == status {
            return Err("deployed migration made no bounded progress".into());
        }
        status = next;
    }
}

const fn terminal(phase: SchemaMigrationPhase) -> bool {
    matches!(
        phase,
        SchemaMigrationPhase::Applied
            | SchemaMigrationPhase::Rejected
            | SchemaMigrationPhase::Aborted
    )
}

fn load_status(target: &CanisterTarget) -> Result<SchemaMigrationStatusPage, String> {
    require_created_canister(target.environment(), target.canister_name())?;
    let candid = call_query(
        target.environment(),
        target.canister_name(),
        SCHEMA_MIGRATION_ENDPOINT.method(),
        "(record { cursor = null })",
    )?;
    decode_response(
        &candid,
        "schema migration status",
        target,
        SCHEMA_MIGRATION_ENDPOINT.method(),
    )
}

fn advance(
    target: &CanisterTarget,
    status: &SchemaMigrationStatusPage,
) -> Result<SchemaMigrationStatusPage, String> {
    let plan = status
        .plan_digest()
        .ok_or_else(|| "deployed schema has no migration plan to advance".to_string())?;
    let command = RuntimeMigrationCommand::Advance {
        expected_database: status.database_identity(),
        expected_head: status.accepted_head().clone(),
        expected_plan: plan,
        acknowledged_finding_page: None,
    };
    call_command(target, &command)
}

fn abort(
    target: &CanisterTarget,
    status: &SchemaMigrationStatusPage,
) -> Result<SchemaMigrationStatusPage, String> {
    let plan = status
        .plan_digest()
        .ok_or_else(|| "deployed schema has no migration plan to abort".to_string())?;
    call_command(
        target,
        &RuntimeMigrationCommand::Abort {
            expected_database: status.database_identity(),
            expected_head: status.accepted_head().clone(),
            expected_plan: plan,
        },
    )
}

fn adopt(
    target: &CanisterTarget,
    status: &SchemaMigrationStatusPage,
) -> Result<SchemaMigrationStatusPage, String> {
    call_command(
        target,
        &RuntimeMigrationCommand::Adopt {
            expected_database: status.database_identity(),
            expected_head: status.accepted_head().clone(),
        },
    )
}

fn call_command(
    target: &CanisterTarget,
    command: &RuntimeMigrationCommand,
) -> Result<SchemaMigrationStatusPage, String> {
    let candid = call_update(
        target.environment(),
        target.canister_name(),
        SCHEMA_MIGRATE_ENDPOINT.method(),
        &command_candid(command),
    )?;
    decode_response(
        &candid,
        "schema migration",
        target,
        SCHEMA_MIGRATE_ENDPOINT.method(),
    )
}

fn decode_response(
    candid: &[u8],
    label: &str,
    target: &CanisterTarget,
    method: &str,
) -> Result<SchemaMigrationStatusPage, String> {
    let response = Decode!(candid, Result<SchemaMigrationStatusPage, Error>)
        .map_err(|error| error.to_string())?;
    response.map_err(|error| endpoint_result_error(label, target, method, error))
}

fn command_candid(command: &RuntimeMigrationCommand) -> String {
    match command {
        RuntimeMigrationCommand::Adopt {
            expected_database,
            expected_head,
        } => format!(
            "(variant {{ Adopt = record {{ expected_database = {}; expected_head = {} }} }})",
            blob(expected_database.to_bytes()),
            head_candid(expected_head),
        ),
        RuntimeMigrationCommand::Advance {
            expected_database,
            expected_head,
            expected_plan,
            acknowledged_finding_page,
        } => format!(
            "(variant {{ Advance = record {{ expected_database = {}; expected_head = {}; expected_plan = {}; acknowledged_finding_page = {} }} }})",
            blob(expected_database.to_bytes()),
            head_candid(expected_head),
            blob(expected_plan.to_bytes()),
            acknowledged_finding_page.map_or_else(
                || "(null : opt nat64)".to_string(),
                |value| format!("opt ({value} : nat64)")
            ),
        ),
        RuntimeMigrationCommand::Abort {
            expected_database,
            expected_head,
            expected_plan,
        } => format!(
            "(variant {{ Abort = record {{ expected_database = {}; expected_head = {}; expected_plan = {} }} }})",
            blob(expected_database.to_bytes()),
            head_candid(expected_head),
            blob(expected_plan.to_bytes()),
        ),
    }
}

fn head_candid(head: &icydb::db::ExpectedAcceptedHead) -> String {
    match head {
        icydb::db::ExpectedAcceptedHead::Empty => "variant { Empty }".to_string(),
        icydb::db::ExpectedAcceptedHead::Exact {
            revision,
            fingerprint,
        } => format!(
            "variant {{ Exact = record {{ revision = ({revision} : nat64); fingerprint = {} }} }}",
            blob(fingerprint.to_bytes()),
        ),
    }
}

fn render_accepted_head(head: &icydb::db::ExpectedAcceptedHead) -> String {
    match head {
        icydb::db::ExpectedAcceptedHead::Empty => "variant { Empty }".to_string(),
        icydb::db::ExpectedAcceptedHead::Exact {
            revision,
            fingerprint,
        } => format!(
            "variant {{ Exact = record {{ revision = {revision}; fingerprint = {} }} }}",
            blob(fingerprint.to_bytes()),
        ),
    }
}

fn blob(bytes: [u8; 32]) -> String {
    const HEX: &[u8; 16] = b"0123456789abcdef";
    let mut value = String::from("blob \"");
    for byte in bytes {
        value.push('\\');
        value.push(char::from(HEX[usize::from(byte >> 4)]));
        value.push(char::from(HEX[usize::from(byte & 0x0f)]));
    }
    value.push('"');
    value
}

fn print_status(status: &SchemaMigrationStatusPage) {
    println!("IcyDB schema migration");
    println!("  phase: {:?}", status.phase());
    println!(
        "  database: {}",
        render_hex_lower(&status.database_identity().to_bytes())
    );
    println!(
        "  accepted head: {}",
        render_accepted_head(status.accepted_head())
    );
    println!(
        "  plan: {}",
        status.plan_digest().map_or_else(
            || "None".to_string(),
            |digest| render_hex_lower(&digest.to_bytes()),
        )
    );
    println!("  rows validated: {}", status.rows_validated());
    println!("  rows rewritten: {}", status.rows_rewritten());
    println!("  indexes rebuilt: {}", status.indexes_rebuilt());
    println!("  findings: {}", status.findings().len());
    for finding in status.findings() {
        println!(
            "    {:?}: entity {} key {}",
            finding.kind(),
            finding.entity_tag(),
            render_hex_lower(finding.primary_key()),
        );
    }
}

#[cfg(test)]
mod tests {
    use super::{blob, command_candid};
    use candid::Decode;
    use icydb::db::{ExpectedAcceptedHead, SchemaMigrationCommand, TargetDatabaseIdentity};
    use icydb_schema::{ExpectedSchemaFingerprint, SchemaMigrationPlanDigest};
    use std::{path::Path, process::Command};

    #[test]
    fn deployed_migration_command_candid_is_exact_and_local_source_free() {
        let candid = command_candid(&SchemaMigrationCommand::Adopt {
            expected_database: TargetDatabaseIdentity::from_bytes([1; 32]),
            expected_head: ExpectedAcceptedHead::Empty,
        });
        assert!(candid.contains("variant { Adopt"));
        assert!(candid.contains(&blob([1; 32])));
        assert!(candid.contains("variant { Empty }"));
    }
    #[test]
    fn migration_commands_decode_without_canister_type_metadata() {
        let didc = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../.tools/ic/bin/didc");
        for revision in [1, u64::MAX] {
            let head = ExpectedAcceptedHead::Exact {
                revision,
                fingerprint: ExpectedSchemaFingerprint::from_bytes([3; 32]),
            };
            let database = TargetDatabaseIdentity::from_bytes([1; 32]);
            let plan = SchemaMigrationPlanDigest::from_bytes([2; 32]);
            for command in [
                SchemaMigrationCommand::Adopt {
                    expected_database: database,
                    expected_head: head.clone(),
                },
                SchemaMigrationCommand::Abort {
                    expected_database: database,
                    expected_head: head.clone(),
                    expected_plan: plan,
                },
                SchemaMigrationCommand::Advance {
                    expected_database: database,
                    expected_head: head.clone(),
                    expected_plan: plan,
                    acknowledged_finding_page: None,
                },
                SchemaMigrationCommand::Advance {
                    expected_database: database,
                    expected_head: head.clone(),
                    expected_plan: plan,
                    acknowledged_finding_page: Some(u64::MAX),
                },
            ] {
                // No DID, service metadata or network supplies types to this encoder.
                let output = Command::new(&didc)
                    .args(["encode", &command_candid(&command)])
                    .output()
                    .expect("selected common didc should execute");
                assert!(
                    output.status.success(),
                    "Candid arguments should encode: {output:?}"
                );
                let text = String::from_utf8(output.stdout).unwrap();
                let bytes = crate::icp::test_support::hex_response_bytes(&text).unwrap();
                assert_eq!(Decode!(&bytes, SchemaMigrationCommand).unwrap(), command);
            }
        }
    }
}