rustploy 0.1.0

//! Ponto de entrada do `rustployd`: resolve config e diretórios, sobe banco,
//! Docker, ingress, API e registry, faz a recuperação de boot e lança os loops
//! de fundo (fila de deploy, watchdog, métricas, logs, backup de env).

mod api;
mod db;
mod deploy;
mod docker;
mod env_backup;
mod env_switch;
mod event_bus;
mod firewall;
mod git_providers;
mod health;
mod ingress;
mod jobs;
mod logs;
mod maintenance;
mod migration;
mod metrics;
mod ports;
mod registry;
mod secrets;
mod shared_db;
mod watchdog;

use mimalloc::MiMalloc;

#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;

use anyhow::Result;
use api::AppState;
use event_bus::EventBus;
use ingress::{IngressController, TlsManager};
use shared::{Event, RustployConfig};
use std::{path::PathBuf, sync::Arc};
use tracing::{error, info, warn};
use tracing_subscriber::EnvFilter;

#[tokio::main]
async fn main() -> Result<()> {
    let config = RustployConfig::global();
    init_logging(&config.daemon.log_level);

    // rustls pulls in both `ring` and `aws-lc-rs` providers transitively, so it
    // cannot pick a process-level CryptoProvider on its own and panics on the
    // first TLS use (ACME). Install one explicitly before any TLS work.
    rustls::crypto::ring::default_provider()
        .install_default()
        .expect("failed to install rustls ring CryptoProvider");

    info!(version = env!("CARGO_PKG_VERSION"), "rustployd starting");

    // Database — resolve path with fallback
    let db_path = resolve_data_path(&config.daemon.db_path);
    let db = Arc::new(db::connect(&db_path).await?);
    info!("database connected");

    // Directório de backups de env vars: config ou <db_path>/env_backups/
    let backup_dir = config
        .env_backup
        .dir
        .as_deref()
        .map(PathBuf::from)
        .unwrap_or_else(|| db_path.join("env_backups"));

    // Docker
    let docker = Arc::new(docker::DockerClient::connect(&config.docker.socket_path)?);
    if let Err(e) = docker.ping().await {
        error!(error = %e, "docker engine unreachable");
        std::process::exit(1);
    }
    info!("docker engine connected");

    let bus = Arc::new(EventBus::new());
    let ingress = Arc::new(IngressController::new());

    let master_key = resolve_master_key_path(&config.secrets.master_key_path);
    let secrets = Arc::new(secrets::SecretsManager::new(&master_key, db.clone())?);

    // TLS / ACME — email do banco tem precedência sobre o config
    let acme_config = {
        let mut acme = config.ingress.acme.clone();
        if let Ok(Some(email)) =
            db::daemon_settings::get(&db, db::daemon_settings::KEY_ACME_EMAIL).await
        {
            if !email.trim().is_empty() {
                info!(email = %email, "ACME: usando email do banco de dados");
                acme.email = Some(email);
                acme.enabled = true;
            }
        }
        acme
    };
    let certs_dir = resolve_data_path(&config.daemon.db_path).join("certs");
    let tls = Arc::new(
        TlsManager::new(certs_dir, acme_config.clone()).expect("failed to initialize TLS manager"),
    );

    // Token interno usado pelo deploy executor pra puxar imagens do próprio
    // registry embutido, sem ação manual do usuário (Fase 3 do plano do
    // registry). Regenerado a cada boot (ver db/registry_tokens.rs) — não
    // pode ser fatal: um erro aqui não deve derrubar o boot do daemon inteiro.
    let registry_internal_token = if config.registry.enabled {
        match registry::internal_token::ensure(&db).await {
            Ok(t) => Some(t),
            Err(e) => {
                tracing::warn!(error = %e, "registry: falha ao gerar token interno, pull de imagens do registry embutido vai falhar até o próximo restart");
                None
            }
        }
    } else {
        None
    };

    // Grava o nome da stack Compose dos serviços que ainda não têm, com o que já
    // está no Docker (nada muda de nome). Antes da recovery, que já lê o gravado.
    match docker::containers::list_compose_projects(&docker.inner).await {
        Ok(live) => {
            if let Err(e) = db::services::backfill_compose_projects(&db, &live).await {
                error!(error = %e, "falha ao gravar o nome das stacks Compose");
            }
        }
        Err(e) => {
            warn!(error = %e, "não listei as stacks Compose; nomes seguem pela fórmula legada")
        }
    }

    // Recovery — devolve os deploys que estavam na fila (Pending) para
    // re-enfileirar depois que o AppState/DeployQueue existir.
    let deploys_to_requeue = deploy::recovery::recover(
        db.clone(),
        docker.clone(),
        ingress.clone(),
        bus.clone(),
        secrets.clone(),
        tls.clone(),
        db_path.clone(),
        config.deploy.drain_secs,
        registry_internal_token.clone(),
    )
    .await;

    // Env var backup loop
    {
        let db2 = db.clone();
        let bdir = backup_dir.clone();
        let interval_secs = config.env_backup.interval_secs;
        tokio::spawn(async move {
            env_backup::backup_loop(db2, bdir, interval_secs).await;
        });
    }

    // Metrics background task
    {
        let docker_inner = docker.inner.clone();
        let db2 = db.clone();
        let bus2 = bus.clone();
        let interval = config.metrics.interval_secs;
        tokio::spawn(async move {
            metrics::collect_loop(Arc::new(docker_inner), db2, bus2, interval).await;
        });
    }

    // Container log streaming task
    {
        let docker_inner = Arc::new(docker.inner.clone());
        let db2 = db.clone();
        let bus2 = bus.clone();
        tokio::spawn(async move {
            logs::stream_loop(docker_inner, db2, bus2).await;
        });
    }

    bus.publish(Event::DaemonReady {
        version: env!("CARGO_PKG_VERSION").to_string(),
    });

    // Storage do registry OCI embutido, criado ANTES do AppState: o listener
    // HTTP do registry, o handler RegistryGc e o job diário de GC precisam
    // compartilhar a MESMA instância (mesma trava de commit).
    let registry_storage = if config.registry.enabled {
        let storage_dir = config
            .registry
            .storage_dir
            .as_deref()
            .map(PathBuf::from)
            .unwrap_or_else(|| db_path.join("registry"));
        match registry::storage::RegistryStorage::new(storage_dir) {
            Ok(s) => Some(Arc::new(s)),
            Err(e) => {
                tracing::error!(error = %e, "registry: falha ao preparar storage_dir, registry desabilitado");
                None
            }
        }
    } else {
        None
    };

    // Domínio público do registry — banco tem precedência sobre o config
    // (mesmo padrão do e-mail ACME acima). Só registra rota/cert se o storage
    // subiu de verdade (não adianta expor um listener que nunca vai existir).
    if registry_storage.is_some() {
        let mut registry_domain = config.registry.domain.clone();
        if let Ok(Some(d)) =
            db::daemon_settings::get(&db, db::daemon_settings::KEY_REGISTRY_DOMAIN).await
        {
            if !d.trim().is_empty() {
                registry_domain = Some(d);
            }
        }
        if let Some(domain) = registry_domain {
            let port = config.registry.port;
            ingress.upsert_route(&domain, vec![format!("127.0.0.1:{port}")], "rp-registry");
            let tls2 = tls.clone();
            let d = domain.clone();
            tokio::spawn(async move {
                if let Err(e) = tls2.ensure_cert(&d).await {
                    warn!(domain = %d, error = %e, "registry: falha ao provisionar certificado");
                } else {
                    info!(domain = %d, "registry: certificado provisionado");
                }
            });
        }
    }

    let state = AppState::new(
        db,
        docker,
        ingress.clone(),
        bus,
        secrets,
        tls.clone(),
        db_path,
        backup_dir,
        config.deploy.drain_secs,
        config.api.clone(),
        registry_storage.clone(),
        registry_internal_token.clone(),
    );

    // Execuções de job que ficaram sem fim num restart (rodando, ou esperando a
    // vez numa fila que só existia em memória): viram "interrompidas". Antes do
    // agendador, que só dispara um job sem execução em andamento.
    match jobs::recover_interrupted(&state.db).await {
        Ok(0) => {}
        Ok(n) => info!(
            n,
            "execuções de job interrompidas pelo restart foram fechadas"
        ),
        Err(e) => warn!(error = %e, "falha ao fechar execuções de job interrompidas"),
    }

    // Worker da fila global de deploys (um por vez). Spawnado depois do
    // AppState; re-enfileira os deploys que estavam na fila antes do restart
    // (na ordem de criação devolvida pelo recovery).
    for dep_id in deploys_to_requeue {
        state.deploy_queue.enqueue(dep_id);
    }
    {
        let state2 = state.clone();
        tokio::spawn(async move {
            crate::deploy::queue::run_worker(state2).await;
        });
    }

    // GC diário do registry
    {
        let db2 = state.db.clone();
        let gc_storage = registry_storage.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(24 * 3600));
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            loop {
                interval.tick().await;
                if let Some(storage) = &gc_storage {
                    match registry::gc::run(&db2, storage).await {
                        Ok(r) if r.blobs_removed > 0 => tracing::info!(
                            blobs = r.blobs_removed,
                            bytes = r.bytes_freed,
                            "registry: GC diário liberou espaço"
                        ),
                        Err(e) => tracing::warn!(error = %e, "registry: falha no GC diário"),
                        _ => {}
                    }
                }
            }
        });
    }

    // Migração deixada "Running" por um daemon que caiu: não retoma, vira Failed.
    if let Ok(n) = db::migration::fail_orphans(&state.db).await
        && n > 0
    {
        tracing::warn!(n, "migrações interrompidas pelo restart marcadas como Failed");
    }

    // Watchdog: detecta containers parados/removidos, tenta restart e redeploy
    {
        let state2 = state.clone();
        tokio::spawn(async move {
            watchdog::watchdog_loop(state2).await;
        });
    }

    // Reconciliation loop: sincroniza status DB ↔ Docker a cada 30 segundos
    {
        let state2 = state.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
            loop {
                interval.tick().await;
                deploy::recovery::reconcile(
                    &state2.db,
                    &state2.docker,
                    &state2.ingress,
                    &state2.tls,
                )
                .await;
            }
        });
    }

    // Scheduler de jobs one-shot: verifica a cada 30s se algum job agendado venceu
    {
        let state2 = state.clone();
        tokio::spawn(async move {
            jobs::scheduler::scheduler_loop(state2).await;
        });
    }

    // Scheduler da limpeza automática de Docker (containers/imagens/volumes/
    // redes/cache de build não usados) — verifica a cada 60s se está devida
    // (ver docs/plano-limpeza-automatica-docker.md).
    {
        let state2 = state.clone();
        tokio::spawn(async move {
            maintenance::scheduler::scheduler_loop(state2).await;
        });
    }

    // Loop de renovação de certificados TLS (a cada 12 horas)
    {
        let tls_renew = tls.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(12 * 3600));
            loop {
                interval.tick().await;
                match tls_renew.renew_expiring().await {
                    Ok(renewed) if !renewed.is_empty() => {
                        info!(domains = ?renewed, "TLS: certificados renovados");
                    }
                    Err(e) => {
                        warn!(error = %e, "TLS: erro na renovação periódica");
                    }
                    _ => {}
                }
            }
        });
    }

    // Ingress Proxy: roteamento de domínios e portas.
    // O listener HTTPS sobe sempre — certs chegam via ACME dinamicamente.
    {
        let routes = ingress.table_handle();
        let http_port = config.ingress.http_port;
        let https_port = config.ingress.https_port;
        let tls = tls.clone();
        tokio::spawn(async move {
            ingress::proxy::start_proxy(routes, http_port, https_port, Some(tls)).await;
        });
    }

    // API HTTP/JSON + SSE — canal administrativo remoto; é o que o GUI
    // (glacier-ui/Luau) e clientes externos consomem. Bind loopback + HTTP puro
    // por padrão (atrás do ingress); se `api.domain` estiver setado, o próprio
    // listener termina TLS na sua porta com cert automático via ACME.
    //
    // Serve TAMBÉM as rotas públicas sem Bearer (`api::public_routes`): o
    // webhook de CI/CD e o callback OAuth do Gitea, que antes tinham um listener
    // dedicado na porta 8788 (ver docs/plano-unificacao-webhook-api.md). Ou
    // seja: com a API desabilitada (`[api] enabled = false`), não há webhook.
    if config.api.enabled {
        let state2 = state.clone();
        let api_cfg = config.api.clone();
        let tls2 = tls.clone();
        tokio::spawn(async move {
            api::http_api::run(state2, api_cfg, Some(tls2)).await;
        });
    }

    // Registry OCI embutido — Fase 1: loopback only, sem auth, sem ingress
    // (ver docs/plano-registry-embutido.md). Bind falho apenas loga e desabilita
    // o listener, como os demais listeners opcionais acima.
    if let Some(storage) = registry_storage.clone() {
        let db2 = state.db.clone();
        let port = config.registry.port;
        tokio::spawn(async move {
            registry::http::run(db2, storage, port).await;
        });
    }

    // Todo listener do daemon roda em `tokio::spawn` acima, então sem isto a
    // `main` retornaria e o processo morreria no boot: este pending existe só
    // para segurar o processo. Encerramento é o sinal do systemd (SIGTERM).
    std::future::pending::<()>().await;
    Ok(())
}

fn init_logging(level: &str) {
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(level));
    tracing_subscriber::fmt()
        .with_env_filter(filter)
        .without_time()
        .compact()
        .init();
}

fn fallback_dir() -> PathBuf {
    shared::fallback_data_dir()
}

/// Tries to use `configured` as the data directory.
/// Falls back to `~/.local/share/rustploy/db` if the path is not writable.
fn resolve_data_path(configured: &str) -> PathBuf {
    let path = PathBuf::from(configured);
    if can_write_dir(&path) {
        return path;
    }
    let fallback = fallback_dir().join("db");
    warn!(
        primary = %path.display(),
        fallback = %fallback.display(),
        "db path not writable, using fallback"
    );
    let _ = std::fs::create_dir_all(&fallback);
    fallback
}

/// Tries to use `configured` as the master key path.
/// Falls back to `~/.local/share/rustploy/master.key` if the directory is
/// not writable (e.g. `/etc/rustploy/` requires root).
fn resolve_master_key_path(configured: &str) -> PathBuf {
    let path = PathBuf::from(configured);
    // If the key already exists and is readable, use it as-is.
    if path.exists() {
        return path;
    }
    // Otherwise we need to be able to create it — check parent writability.
    let parent = path.parent().unwrap_or(&path);
    if can_write_dir(parent) {
        return path;
    }
    let fallback = fallback_dir().join("master.key");
    warn!(
        primary = %path.display(),
        fallback = %fallback.display(),
        "master key directory not writable, using fallback"
    );
    let _ = std::fs::create_dir_all(fallback.parent().unwrap());
    fallback
}

/// Returns true only when `dir` (or its path) is both creatable and writable.
/// Unlike a bare `create_dir_all` check, this actually probes write access
/// even when the directory already exists (e.g. created by a previous root run).
fn can_write_dir(dir: &std::path::Path) -> bool {
    if std::fs::create_dir_all(dir).is_err() {
        return false;
    }
    let probe = dir.join(".rustploy_write_probe");
    if std::fs::write(&probe, b"").is_err() {
        return false;
    }
    let _ = std::fs::remove_file(probe);
    true
}