//! Full boot sequence: config, store, git checks, gitconfig and hooks, routers, SSH task.
//!
//! Three servers run until one of them stops or a shutdown signal arrives:
//! - public HTTP on `config.listen` (UI and smart-HTTP),
//! - raw user-content origin on `config.user_content_listen` (raw.rs, explicit routes only),
//! - internal hook API on the unix socket `data_dir/internal.sock`,
//! - SSH on `config.ssh_listen` (forge_ssh, which has no shutdown hook, so it is aborted).

use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use forge_core::{Config, Store, secret};
use forge_git::GitEnv;
use forge_ssh::SshConfig;
use tokio::net::{TcpListener, UnixListener};
use tokio::signal::unix::{SignalKind, signal};
use tokio::sync::Notify;
use tokio::sync::watch;
use topcoat::cookie::RouterBuilderCookieExt;
use topcoat::router::tower::TowerLayer;
use topcoat::router::{Compression, Router, RouterBuilderDiscoverExt, RouterService};

use crate::internal;
use crate::raw;
use crate::ssh_backend::SshBackend;
use crate::state::AppState;

/// Minimum git version forge relies on (receive-pack and stateless-rpc behaviour).
pub(crate) const MIN_GIT: (u32, u32, u32) = (2, 38, 0);

/// Boots every component and serves until shutdown.
pub async fn serve(config: Config) -> forge_core::Result<()> {
    let listen = config.listen;
    let user_content_listen = config.user_content_listen;
    let ssh_listen = config.ssh_listen;

    std::fs::create_dir_all(&config.data_dir)?;
    // The data dir holds the instance key and the internal socket. Keep it private to this user.
    std::fs::set_permissions(&config.data_dir, std::fs::Permissions::from_mode(0o700))?;
    let data_dir = config.data_dir.canonicalize()?;

    let key = Arc::new(secret::load_or_create_key(&data_dir.join("forge.key"))?);
    let store = Store::open(&config.database).await?;
    store.set_secret_key(*key);

    let git = Arc::new(GitEnv {
        git_bin: find_git(),
        data_dir: data_dir.clone(),
    });
    forge_git::version::require_min_version(&git, MIN_GIT).await?;
    let version = forge_git::version::git_version(&git).await?;
    forge_git::setup::write_managed_gitconfig(&data_dir)?;
    let forge_bin = std::env::current_exe()?.canonicalize()?;
    forge_git::setup::write_hook_shims(&data_dir, &forge_bin)?;

    // Hash the running binary once, before any runner can ask for it. Hashing reads the whole
    // executable, which takes tens of seconds for a debug build, so it runs off the boot path and
    // the listeners bind first. Any caller that needs the hash waits on the same `OnceLock`.
    tokio::task::spawn_blocking(|| {
        crate::self_update::running();
    });

    let internal_token: Arc<str> = Arc::from(random_hex());
    let admin_token: Arc<str> = Arc::from(random_hex());
    crate::admin_client::write_admin_token(&data_dir, &admin_token)?;
    let internal_socket = data_dir.join("internal.sock");
    // sun_path holds at most 108 bytes including the NUL terminator.
    if internal_socket.as_os_str().len() >= 108 {
        return Err(forge_core::Error::Config(format!(
            "internal socket path is too long for a unix socket (max 107 bytes): {}; choose a shorter data_dir",
            internal_socket.display()
        )));
    }
    match std::fs::remove_file(&internal_socket) {
        Ok(()) => {}
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => return Err(e.into()),
    }
    let unix = UnixListener::bind(&internal_socket)?;
    std::fs::set_permissions(&internal_socket, std::fs::Permissions::from_mode(0o600))?;

    let mailer: Arc<dyn crate::mail::Mailer> = match config.mail.transport {
        forge_core::config::MailTransport::Smtp => {
            Arc::new(crate::mail::SmtpMailer::new(store.clone()))
        }
        forge_core::config::MailTransport::Log => {
            Arc::new(crate::mail::LogMailer::new(data_dir.join("mail-log")))
        }
    };
    let state = AppState {
        config: Arc::new(config),
        store,
        git,
        key,
        internal_token,
        admin_token,
        jobs: Arc::new(crate::backup::Jobs::default()),
        internal_socket: internal_socket.clone(),
        throttle: Arc::new(crate::throttle::Throttle::default()),
        webhook_waker: Arc::new(Notify::new()),
        mirror_waker: Arc::new(Notify::new()),
        blobs: crate::blobs::BlobStore::open(&data_dir.join("blobs"))?,
        replication: Arc::new(crate::replication::ReplicationQueue::default()),
        ci_waker: Arc::new(Notify::new()),
        mailer,
        mail_waker: Arc::new(Notify::new()),
        notify_hub: Arc::new(crate::notify_hub::NotifyHub::new()),
        ci_live: Arc::new(crate::ci::CiLive::default()),
    };
    if let Err(e) = crate::signatures::rebuild_allowed_signers(&state).await {
        tracing::error!(error = %e, "building the SSH allowed-signers file failed");
    }
    if let Err(e) = crate::signatures::rebuild_gpg_keyring(&state).await {
        tracing::error!(error = %e, "rebuilding the GPG keyring failed");
    }
    crate::webhooks::worker::spawn(state.clone());
    crate::mirrors::worker::spawn(state.clone());
    crate::advisories::worker::spawn(state.clone());
    crate::mail::worker::spawn(state.clone());
    if state.config.backup.enabled() {
        crate::backup::scheduler::spawn(state.clone());
    }
    if state.config.replication.enabled() {
        crate::replication::worker::spawn(state.clone());
    }
    crate::ci::reaper::spawn(state.clone());

    // Topcoat's builder-wide compression is off because it cannot filter by content type or path.
    // `ResponseCompression` compresses text responses only and skips git, LFS, raw and archive paths.
    let public: Router = Router::builder()
        .discover()
        .app_context(state.clone())
        .cookies()
        .layer(crate::pages::settings_profile::avatar_body_limit())
        .layer(crate::pages::repo_ctx::RepoMovedLayer)
        .layer(TowerLayer::new(crate::compression::ResponseCompression))
        .compression(Compression::off())
        .build();
    let internal_router = internal::router(state.clone());
    let raw_router = raw::router(state.clone());

    let tcp = TcpListener::bind(listen).await?;
    let raw_tcp = TcpListener::bind(user_content_listen).await.map_err(|e| {
        forge_core::Error::Config(format!(
            "cannot bind the user-content listener {user_content_listen}: {e}"
        ))
    })?;

    tracing::info!(
        listen = %listen,
        user_content = %user_content_listen,
        ssh = %ssh_listen,
        data_dir = %data_dir.display(),
        git = %format_version(version),
        "forge starting"
    );

    let (stop_tx, stop_rx) = watch::channel(false);

    // Open notification streams never end on their own, so they are closed with the rest.
    let hub = Arc::clone(&state.notify_hub);
    let hub_stop = stop_rx.clone();
    tokio::spawn(async move {
        wait_stop(hub_stop).await;
        hub.close();
    });

    let ssh_config = SshConfig {
        listen: ssh_listen,
        host_key_path: data_dir.join("ssh_host_ed25519_key"),
        auth_timeout: Duration::from_secs(30),
        inactivity_timeout: Duration::from_secs(10 * 60),
    };
    let ssh_backend = Arc::new(SshBackend {
        state: state.clone(),
    });
    let ssh_stop = stop_tx.clone();
    let mut ssh_task = tokio::spawn(async move {
        let result = forge_ssh::run(ssh_config, ssh_backend).await;
        // An SSH failure brings the whole process down.
        ssh_stop.send_replace(true);
        result
    });

    let signal_stop = stop_tx.clone();
    tokio::spawn(async move {
        tokio::select! {
            () = shutdown_signal() => tracing::info!("shutdown signal received"),
            () = crate::self_update::restart_requested() => tracing::info!("self-update: restarting"),
        }
        signal_stop.send_replace(true);
    });

    let http_stop = stop_tx.clone();
    let http_rx = stop_rx.clone();
    let http = async move {
        let result =
            topcoat::serve_until(tcp, RouterService::new(public), wait_stop(http_rx)).await;
        http_stop.send_replace(true);
        result
    };

    let internal_stop = stop_tx.clone();
    let internal_rx = stop_rx.clone();
    let internal_server = async move {
        let result = topcoat::serve_until(
            unix,
            RouterService::new(internal_router),
            wait_stop(internal_rx),
        )
        .await;
        internal_stop.send_replace(true);
        result
    };

    let raw_stop = stop_tx.clone();
    let raw_rx = stop_rx.clone();
    let raw_server = async move {
        let result =
            topcoat::serve_until(raw_tcp, RouterService::new(raw_router), wait_stop(raw_rx)).await;
        raw_stop.send_replace(true);
        result
    };

    let (http_result, internal_result, raw_result) =
        tokio::join!(http, internal_server, raw_server);

    // forge_ssh::run has no shutdown hook. If it is still running it is cancelled here.
    ssh_task.abort();
    let ssh_result: std::io::Result<()> = match (&mut ssh_task).await {
        Ok(result) => result.map_err(|e| std::io::Error::other(e.to_string())),
        Err(e) if e.is_cancelled() => Ok(()),
        Err(e) => Err(std::io::Error::other(e.to_string())),
    };

    let _ = std::fs::remove_file(&internal_socket);
    tracing::info!("forge stopped");

    http_result?;
    internal_result?;
    raw_result?;
    ssh_result?;
    Ok(())
}

/// Resolves when the shared stop flag is set.
async fn wait_stop(mut rx: watch::Receiver<bool>) {
    let _ = rx.wait_for(|stopped| *stopped).await;
}

/// Resolves on ctrl_c or SIGTERM.
async fn shutdown_signal() {
    let ctrl_c = async {
        if let Err(e) = tokio::signal::ctrl_c().await {
            tracing::warn!(error = %e, "ctrl_c handler unavailable");
            std::future::pending::<()>().await;
        }
    };
    let terminate = async {
        match signal(SignalKind::terminate()) {
            Ok(mut term) => {
                term.recv().await;
            }
            Err(e) => {
                tracing::warn!(error = %e, "SIGTERM handler unavailable");
                std::future::pending::<()>().await;
            }
        }
    };
    tokio::select! {
        () = ctrl_c => {}
        () = terminate => {}
    }
}

/// Finds the git binary on PATH, falling back to /usr/bin/git.
pub(crate) fn find_git() -> PathBuf {
    std::env::var_os("PATH")
        .and_then(|paths| {
            std::env::split_paths(&paths)
                .map(|dir| dir.join("git"))
                .find(|candidate| candidate.is_file())
        })
        .and_then(|candidate| candidate.canonicalize().ok())
        .unwrap_or_else(|| PathBuf::from("/usr/bin/git"))
}

/// 32 random bytes as lowercase hex. Used as the per-boot internal API token.
fn random_hex() -> String {
    let mut buf = [0u8; 32];
    rand::fill(&mut buf);
    hex::encode(buf)
}

fn format_version((major, minor, patch): (u32, u32, u32)) -> String {
    format!("{major}.{minor}.{patch}")
}