rustploy 0.1.15

//! Recuperação no boot: aborta deploys interrompidos, reconcilia status com o
//! Docker e restaura as rotas do ingress.

use crate::{
    engine,
    engine::{DockerClient, containers},
    features::ingress::{IngressController, TlsManager},
    platform::event_bus::EventBus,
    platform::secrets::SecretsManager,
    store::Db,
};
use shared::{DeployState, Service, ServiceStatus};
use std::{path::PathBuf, sync::Arc};
use tracing::{info, warn};

/// Recuperação no boot: devolve os deployments que estavam só na fila (para
/// re-enfileirar), aborta os que tinham começado (limpando stagings) e restaura
/// rotas e certificados dos serviços no ar.
#[allow(clippy::too_many_arguments)]
pub async fn recover(
    db: Arc<Db>,
    docker: Arc<DockerClient>,
    ingress: Arc<IngressController>,
    bus: Arc<EventBus>,
    secrets: Arc<SecretsManager>,
    tls: Arc<TlsManager>,
    db_path: PathBuf,
    drain_secs: u64,
    registry_internal_token: Option<Arc<str>>,
) -> Vec<String> {
    // deployment_ids em `Pending` (estavam na fila, nunca começaram) — devolvidos
    // ao chamador para re-enfileirar na `DeployQueue` (que ainda não existe neste
    // ponto do boot), preservando a fila entre restarts.
    let mut to_requeue: Vec<String> = Vec::new();

    let pending = match crate::store::deployments::get_non_terminal(&db).await {
        Ok(v) => v,
        Err(e) => {
            warn!(error = %e, "failed to query non-terminal deployments");
            return to_requeue;
        }
    };

    // Restore ingress routes for running services and re-provision missing TLS certs
    restore_routes(&db, &docker, &ingress, &tls).await;

    if pending.is_empty() {
        info!("no deployments to recover");
        return to_requeue;
    }

    info!(count = pending.len(), "recovering deployments");

    for dep in pending {
        let svc = match crate::store::services::get(&db, &dep.service_id).await {
            Ok(Some(s)) => s,
            _ => {
                warn!(
                    deployment_id = dep.id,
                    "service not found during recovery, marking failed"
                );
                let _ = crate::store::deployments::transition(
                    &db,
                    &dep.id,
                    &dep.state,
                    DeployState::Failed,
                    Some("service not found during recovery".into()),
                )
                .await;
                continue;
            }
        };

        match &dep.state {
            // Estava na fila, nunca começou: re-enfileira em vez de falhar,
            // preservando a fila entre restarts. Mantém o serviço em Queued.
            DeployState::Pending => {
                info!(deployment_id = dep.id, "re-enfileirando deploy pendente");
                let _ = crate::store::services::update_status(
                    &db,
                    &svc.id,
                    &ServiceStatus::Queued,
                    None,
                )
                .await;
                to_requeue.push(dep.id.clone());
            }

            // Pre-swap states (já começaram): safe to abort
            DeployState::PreDeployCheck
            | DeployState::ResolvingDeps
            | DeployState::PullingImage
            | DeployState::CloningRepo
            | DeployState::BuildingImage
            | DeployState::ComposingUp
            | DeployState::Staging
            | DeployState::HealthcheckPolling => {
                info!(
                    deployment_id = dep.id,
                    state = dep.state.label(),
                    "aborting pre-swap deployment"
                );
                // Remove staging containers if they exist (todas as réplicas;
                // nome atual ou legado, sempre conferindo o dono)
                let dep_short = engine::networks::id_short(&dep.id);
                for i in 0..svc.spec.replicas.max(1) {
                    if let Ok(Some(id)) = containers::find_staging(
                        &docker.inner,
                        &svc.id,
                        &svc.spec.name,
                        dep_short,
                        i,
                    )
                    .await
                    {
                        let _ = containers::remove(&docker.inner, &id).await;
                    }
                }
                let _ = crate::store::deployments::transition(
                    &db,
                    &dep.id,
                    &dep.state,
                    DeployState::Failed,
                    Some("daemon restarted during deploy".into()),
                )
                .await;
                let _ = crate::store::services::update_status(
                    &db,
                    &svc.id,
                    &ServiceStatus::Error("deploy interrupted by restart".into()),
                    None,
                )
                .await;
            }

            // Swap in progress: inspect actual containers to decide
            DeployState::SwappingIn | DeployState::Draining => {
                info!(
                    deployment_id = dep.id,
                    state = dep.state.label(),
                    "resuming swap-in-progress deployment"
                );
                let executor = Arc::new(crate::features::deploy::executor::DeployExecutor {
                    db: db.clone(),
                    docker: docker.clone(),
                    ingress: ingress.clone(),
                    bus: bus.clone(),
                    secrets: secrets.clone(),
                    tls: tls.clone(),
                    db_path: db_path.clone(),
                    drain_secs,
                    registry_internal_token: registry_internal_token.clone(),
                });
                let dep_id = dep.id.clone();
                tokio::spawn(async move { executor.run(dep_id).await });
            }

            // Promoting: complete the rename
            DeployState::Promoting => {
                info!(deployment_id = dep.id, "completing promotion");
                let executor = Arc::new(crate::features::deploy::executor::DeployExecutor {
                    db: db.clone(),
                    docker: docker.clone(),
                    ingress: ingress.clone(),
                    bus: bus.clone(),
                    secrets: secrets.clone(),
                    tls: tls.clone(),
                    db_path: db_path.clone(),
                    drain_secs,
                    registry_internal_token: registry_internal_token.clone(),
                });
                let dep_id = dep.id.clone();
                tokio::spawn(async move { executor.run(dep_id).await });
            }

            // Rolling back: complete the rollback
            DeployState::RollingBack => {
                info!(deployment_id = dep.id, "completing rollback");
                let executor = Arc::new(crate::features::deploy::executor::DeployExecutor {
                    db: db.clone(),
                    docker: docker.clone(),
                    ingress: ingress.clone(),
                    bus: bus.clone(),
                    secrets: secrets.clone(),
                    tls: tls.clone(),
                    db_path: db_path.clone(),
                    drain_secs,
                    registry_internal_token: registry_internal_token.clone(),
                });
                let dep_id = dep.id.clone();
                tokio::spawn(async move { executor.run(dep_id).await });
            }

            DeployState::Live
            | DeployState::Stopped
            | DeployState::Failed
            | DeployState::Pruning => {}
        }
    }

    to_requeue
}

/// Reconciles every service's DB status against actual Docker container state.
/// Marks stopped containers as Stopped, running containers as Running, and
/// keeps ingress routes in sync. Safe to call at any time; non-destructive.
pub async fn reconcile(
    db: &Db,
    docker: &DockerClient,
    ingress: &IngressController,
    tls: &Arc<TlsManager>,
) {
    let services = match crate::store::services::list_all(db).await {
        Ok(v) => v,
        Err(e) => {
            warn!(error = %e, "reconcile: falha ao listar serviços");
            return;
        }
    };

    for svc in services {
        // Servidor de banco compartilhado: re-liga às redes autorizadas (a
        // recriação do container, ou o restart do dockerd, desfaz). Barato
        // quando já está tudo conectado.
        if svc.spec.shared.is_some() {
            crate::features::deploy::shared_net::sync_server(db, docker, &svc).await;
        }
        let replicas = svc.spec.replicas.max(1);
        let net = match crate::store::projects::network_name(db, &svc.spec.project_id).await {
            Ok(n) => n,
            Err(e) => {
                warn!(service_id = %svc.id, error = %e, "recovery: falha ao ler a rede do projeto");
                continue;
            }
        };

        let mut ips = containers::live_replica_ips(&docker.inner, &svc.id, &net, replicas).await;

        if ips.is_empty()
            && let Some(ip) = compose_ingress_ip(docker, &svc, &net).await
        {
            ips.push(ip);
        }

        let is_running = !ips.is_empty();
        let was_running = matches!(svc.status, ServiceStatus::Running | ServiceStatus::Degraded);

        match (is_running, was_running) {
            (true, false) => {
                info!(
                    service = svc.spec.name,
                    "reconcile: container encontrado, marcando Running"
                );
                let _ = crate::store::services::update_status(
                    db,
                    &svc.id,
                    &ServiceStatus::Running,
                    None,
                )
                .await;
                reconcile_routes(&svc, ips, ingress, tls).await;
            }
            (false, true) => {
                info!(
                    service = svc.spec.name,
                    "reconcile: container ausente, marcando Stopped"
                );
                let _ = crate::store::services::update_status(
                    db,
                    &svc.id,
                    &ServiceStatus::Stopped,
                    None,
                )
                .await;
                ingress.remove_domains(&svc.spec);
                if let Some(host_port) = svc.spec.host_port {
                    ingress.remove_port_route(host_port);
                }
            }
            (true, true) => {
                // Container e DB ambos Running: garante que as rotas estão registradas.
                reconcile_routes(&svc, ips, ingress, tls).await;
            }
            (false, false) => {}
        }
    }
}

/// IP do container que atende o ingress numa stack Compose, ou `None` quando a
/// stack não está no ar.
///
/// Os dois caminhos de restauração de rota (o reconcile periódico e o
/// `restore_routes` do boot) precisam escolher o **mesmo** container que o
/// deploy escolheu — senão reiniciar o daemon move o domínio para outro
/// container da stack, e o serviço passa a responder 502 sem ninguém ter
/// mexido em nada.
async fn compose_ingress_ip(docker: &DockerClient, svc: &Service, net: &str) -> Option<String> {
    let ingress_service = match &svc.spec.source {
        shared::ServiceSource::Compose(c) => c.ingress_service.as_deref(),
        _ => None,
    };
    let project = svc.compose_project_name();
    let cid = containers::find_compose_ingress_container(
        &docker.inner,
        &project,
        ingress_service,
        &svc.spec.ingress_container_ports(),
    )
    .await
    .ok()
    .flatten()?;
    containers::get_container_ip(&docker.inner, &cid, net)
        .await
        .ok()
}

async fn reconcile_routes(
    svc: &Service,
    ips: Vec<String>,
    ingress: &IngressController,
    tls: &Arc<TlsManager>,
) {
    ingress.register_domains(&svc.spec, &ips, &svc.id);
    for route in svc.spec.domain_routes().into_iter().filter(|r| r.tls) {
        let tls = tls.clone();
        let domain = route.domain.clone();
        tokio::spawn(async move {
            if let Err(e) = tls.ensure_cert(&domain).await {
                warn!(domain = %domain, error = %e, "reconcile: falha ao garantir cert TLS");
            }
        });
    }
    if let Some(host_port) = svc.spec.host_port {
        let backends: Vec<String> = ips
            .iter()
            .map(|ip| format!("{ip}:{}", svc.spec.port))
            .collect();
        ingress.upsert_port_route(host_port, backends);
    }
}

/// Recria no ingress as rotas de domínio e de porta dos serviços no ar
/// (réplicas live, ou o container da stack Compose) e reabre a porta externa no
/// firewall.
async fn restore_routes(
    db: &Db,
    docker: &DockerClient,
    ingress: &IngressController,
    tls: &Arc<TlsManager>,
) {
    let services = match crate::store::services::get_running(db).await {
        Ok(v) => v,
        Err(e) => {
            warn!(error = %e, "failed to restore routes");
            return;
        }
    };

    info!(
        count = services.len(),
        "restoring ingress routes for running services"
    );

    for svc in services {
        let replicas = svc.spec.replicas.max(1);
        let net = match crate::store::projects::network_name(db, &svc.spec.project_id).await {
            Ok(n) => n,
            Err(e) => {
                warn!(service_id = %svc.id, error = %e, "recovery: falha ao ler a rede do projeto");
                continue;
            }
        };

        // Coleta IPs de todas as réplicas live (Git/Registry)
        let mut ips = containers::live_replica_ips(&docker.inner, &svc.id, &net, replicas).await;

        // Fallback para Compose: o container da stack que atende o ingress.
        if ips.is_empty()
            && let Some(ip) = compose_ingress_ip(docker, &svc, &net).await
        {
            ips.push(ip);
        }

        if ips.is_empty() {
            warn!(
                service = svc.spec.name,
                "no live containers found, skipping route restore"
            );
            continue;
        }

        if !svc.spec.domain_routes().is_empty() {
            ingress.register_domains(&svc.spec, &ips, &svc.id);
            info!(service = svc.spec.name, ?ips, "routes restored");

            for route in svc.spec.domain_routes().into_iter().filter(|r| r.tls) {
                info!(service = svc.spec.name, domain = %route.domain, "TLS: disparando ensure_cert no restart para serviço running");
                let tls = tls.clone();
                let domain = route.domain.clone();
                tokio::spawn(async move {
                    if let Err(e) = tls.ensure_cert(&domain).await {
                        warn!(domain = %domain, error = %e, "TLS: falha ao re-provisionar certificado no restart");
                    }
                });
            }
        }

        if let Some(host_port) = svc.spec.host_port {
            let backends: Vec<String> = ips
                .iter()
                .map(|ip| format!("{ip}:{}", svc.spec.port))
                .collect();
            ingress.upsert_port_route(host_port, backends.clone());
            info!(
                service = svc.spec.name,
                host_port,
                ?backends,
                "port routes restored"
            );
            // Auto-cura: se o admin resetou o ufw (ou o SO foi reinstalado), a
            // regra da porta volta sozinha junto com a rota.
            crate::features::ingress::firewall::ensure_allowed_bg(host_port);
        }
    }
}