#![recursion_limit = "256"]

mod cli;

use std::net::SocketAddr;
use std::path::PathBuf;

use anyhow::{Context, anyhow};
use clap::Parser;

use crate::cli::{Cli, Command};

fn main() -> anyhow::Result<()> {
    match Cli::parse().command.unwrap_or(Command::Serve) {
        Command::Serve => serve(),
        Command::Mail => mail(),
        Command::Migrate => migrate(),
        Command::Openapi { check } => openapi(check),
        Command::IdentityOpenapi => identity_openapi(),
    }
}

#[tokio::main]
async fn serve() -> anyhow::Result<()> {
    use actix_web::middleware::from_fn;
    use actix_web::{App, HttpServer, web};
    use config::scoped::{ApihostConfig, ScopedConfig};
    use tokio_util::sync::CancellationToken;
    use tracing::{info, warn};
    use utoipa_swagger_ui::SwaggerUi;
    use webapi::openapi::api_doc;

    let telemetry = oltp::init(&oltp::TelemetryConfig::from_env("gitdata"));

    let cfg = ApihostConfig::load().context("loading configuration")?;
    if !cfg.apihost.enabled() {
        info!("apihost is disabled by config; exiting");
        return Ok(());
    }

    let cancel = CancellationToken::new();
    let app_config = cfg.clone().into_app_config();
    let session_config = app_config.session.clone();
    let session_store = auth::session::RedisStore::connect(&app_config.redis)
        .await
        .context("connecting the session store")?;
    let database = std::sync::Arc::new(
        export::AppDatabase::connect(app_config.database.clone())
            .await
            .context("connecting to the shared database")?,
    );
    let bind_addr = cfg.apihost.bind_addr().to_string();
    let port = cfg.apihost.port();
    let addr: SocketAddr = format!("{bind_addr}:{port}")
        .parse()
        .with_context(|| format!("parsing bind address {bind_addr}:{port}"))?;
    let conn = queue::RocketMqConn::init(&cfg.rocketmq)
        .await
        .map_err(|err| anyhow::anyhow!("failed to start the rocketmq producer: {err}"))?;

    let auth_events = auth::events::EventHandler::new(conn.clone());
    let email_pool: config::email::EmailPoolConfig = config::loader::figment()
        .ok()
        .and_then(|figment| {
            figment
                .extract_inner::<config::email::EmailPoolConfig>("email")
                .ok()
        })
        .unwrap_or_default();
    let mail =
        auth::email::MailPublisher::new(queue::Publisher::new(conn.clone()), email_pool.mail_topic);
    let mut state = export::AppState::with_identity(
        app_config,
        database,
        None,
        Some(mail),
        Some(auth_events.clone()),
    )
    .await
    .context("building AppState")?;
    let auth_service = auth::AuthService::new(state.clone());
    state.kenauth = Some(std::sync::Arc::new(auth_service));
    if let Some(client) = state.kenauth.clone() {
        export::identity_client::install(client);
    }
    info!("dependencies ready");

    spawn_queue_depth_reporter(
        state.redis.clone(),
        &[
            ("push.jobs", export::push::QUEUE_TOPIC.to_string()),
            ("push.jobs.dlq", export::push::DLQ_TOPIC.to_string()),
        ],
    );

    tokio::spawn(chat::worker::run_background_tasks(state.clone()));
    tokio::spawn(deletion::run_worker(state.clone()));

    let relay = events::outbox::OutboxRelay::start(
        Some(auth_events),
        Some(queue::Publisher::new(conn.clone())),
        vec![("apihost", state.database.primary().clone())],
    );
    info!("event outbox relay started");

    let mut push_tasks = Vec::new();
    if cfg.push.enabled {
        for _ in 0..cfg.push.workers.max(1) {
            let worker = push::PushWorker::new(state.clone(), cancel.clone())
                .map_err(|err| anyhow::anyhow!("failed to start push worker: {err}"))?;
            push_tasks.push(tokio::spawn(worker.run()));
        }
        info!(
            workers = cfg.push.workers.max(1),
            "web push workers started"
        );
    }

    let doc = api_doc();
    let app_state = state.clone();

    let server = HttpServer::new(move || {
        let state = app_state.clone();
        let app = App::new()
            .app_data(web::Data::new(state.clone()))
            .app_data(
                web::JsonConfig::default()
                    .limit(webapi::MAX_REQUEST_BODY_BYTES)
                    .error_handler(export::http::json_error),
            )
            .app_data(web::PayloadConfig::new(webapi::MAX_REQUEST_BODY_BYTES));
        app.wrap(auth::session::SessionMiddleware::new(
            session_config.clone(),
            session_store.clone(),
        ))
        .wrap(auth::envelope::EnvelopeContext)
        .wrap(webapi::build_cors(&state.config.web))
        .wrap(from_fn(export::middleware::trace_request))
        .wrap(oltp::actix::HttpTelemetry::new("gitdata"))
        .wrap(from_fn(auth::share::session::revalidate::revalidate))
        .wrap(from_fn(auth::share::session::csrf::csrf))
        .service(web::resource("/api/docs").route(web::get().to(|| async {
            actix_web::HttpResponse::Found()
                .insert_header((actix_web::http::header::LOCATION, "/api/docs/"))
                .finish()
        })))
        .service(SwaggerUi::new("/api/docs/{_:.*}").url("/api/docs/openapi.json", doc.clone()))
        .service(webapi::scope(state))
        .default_service(web::to(auth::assets::serve))
    })
    .disable_signals()
    .bind(addr)
    .with_context(|| format!("binding {addr}"))?
    .run();
    info!(%addr, "apihost listening");

    let handle = server.handle();
    let api_cancel = cancel.clone();
    tokio::spawn(async move {
        api_cancel.cancelled().await;
        handle.stop(true).await;
    });
    let api = async move { server.await.context("actix server error") };

    let shutdown = oltp::shutdown::install();
    let signal = async {
        tokio::select! {
            _ = shutdown.wait() => cancel.cancel(),
            _ = cancel.cancelled() => {},
        }
    };
    let (api_result, ()) = tokio::join!(api, signal);
    if let Err(error) = relay.shutdown().await {
        warn!(%error, "shutting down the event outbox relay failed");
    }
    for task in push_tasks {
        let _ = task.await;
    }
    conn.shutdown().await;
    api_result?;

    if let Err(error) = telemetry.shutdown() {
        warn!(%error, "flushing telemetry failed");
    }

    info!("apihost shut down");
    Ok(())
}

#[tokio::main]
async fn mail() -> anyhow::Result<()> {
    use tokio_util::sync::CancellationToken;

    let telemetry = oltp::init(&oltp::TelemetryConfig::from_env("gitdata-mail"));

    let figment = config::loader::figment().context("loading configuration")?;
    let rocketmq: config::rocketmq::RocketMqConfig = figment
        .extract_inner("rocketmq")
        .context("loading rocketmq configuration")?;
    let smtp: config::smtp::SmtpConfig = figment
        .extract_inner("smtp")
        .context("loading smtp configuration")?;
    let email: config::email::EmailPoolConfig = figment
        .extract_inner("email")
        .context("loading email configuration")?;
    rocketmq.is_valid().map_err(anyhow::Error::msg)?;
    smtp.is_valid().map_err(anyhow::Error::msg)?;
    email.is_valid().map_err(anyhow::Error::msg)?;

    let conn = queue::RocketMqConn::init(&rocketmq)
        .await
        .map_err(|err| anyhow!("failed to start the rocketmq producer: {err}"))?;
    let events = auth::events::EventHandler::new(conn.clone());
    let consumer = auth::email::EmailConsumer::new(smtp, email, Some(events))
        .context("building the mail consumer")?;

    let cancel = CancellationToken::new();
    let run_cancel = cancel.clone();
    let run = consumer.run(rocketmq, run_cancel);
    let shutdown = oltp::shutdown::install();
    let signal = async {
        tokio::select! {
            _ = shutdown.wait() => cancel.cancel(),
            _ = cancel.cancelled() => {},
        }
    };
    let (result, ()) = tokio::join!(run, signal);
    conn.shutdown().await;
    result.context("mail service failed")?;
    if let Err(error) = telemetry.shutdown() {
        tracing::warn!(%error, "flushing telemetry failed");
    }
    Ok(())
}

#[tokio::main]
async fn migrate() -> anyhow::Result<()> {
    use config::scoped::{MigrateConfig, ScopedConfig};
    use models::migrate::sync_schema;
    use sea_orm::Database;
    use tracing::info;

    const DEFAULT_LOG_FILTER: &str = "info,kenapi=debug";

    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(DEFAULT_LOG_FILTER)),
        )
        .init();

    let cfg = MigrateConfig::load().context("loading configuration")?;

    let opts = cfg
        .database
        .primary_connect_options()
        .map_err(|e| anyhow::anyhow!(e))?;
    let db = Database::connect(opts)
        .await
        .context("connecting to postgres")?;

    info!("running schema migration");
    sync_schema(&db).await.context("running schema sync")?;
    if let Some(auth_config) = auth_config()? {
        let identity_db = auth::db::DbConn::new(db.clone(), Vec::new());
        auth::migrate::sync_schema(&identity_db)
            .await
            .context("synchronizing identity schema")?;
        drop(auth_config);
    }
    info!("schema migration completed");

    Ok(())
}

fn openapi(check: bool) -> anyhow::Result<()> {
    fn generate<F>(generate: F) -> anyhow::Result<String>
    where
        F: FnOnce() -> anyhow::Result<String> + Send + 'static,
    {
        std::thread::Builder::new()
            .stack_size(256 * 1024 * 1024)
            .spawn(generate)
            .context("spawn generator thread")?
            .join()
            .map_err(|_| anyhow!("openapi generation panicked (stack overflow?)"))?
    }

    fn check_file(path: &std::path::Path, json: &str) -> anyhow::Result<()> {
        let existing =
            std::fs::read_to_string(path).with_context(|| format!("reading {}", path.display()))?;
        if existing.trim_end() != json.trim_end() {
            eprintln!(
                "{} is out of date; run `cargo run -p kenapi -- openapi` to regenerate",
                path.display()
            );
            std::process::exit(1);
        }
        Ok(())
    }

    let web = generate(|| {
        let doc = webapi::openapi::api_doc();
        doc.to_pretty_json().context("serializing web openapi doc")
    })?;

    let crate_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    let root = crate_dir
        .ancestors()
        .nth(1)
        .ok_or_else(|| anyhow!("workspace root not found above {}", crate_dir.display()))?;
    let web_path = root.join("openapi.json");

    if check {
        check_file(&web_path, &web)?;
        println!("openapi.json is up to date");
        return Ok(());
    }

    std::fs::write(&web_path, web).with_context(|| format!("writing {}", web_path.display()))?;
    println!("wrote {}", web_path.display());
    Ok(())
}

fn auth_config() -> anyhow::Result<Option<auth::config::Config>> {
    let figment = config::loader::figment().context("loading configuration")?;
    let enabled = figment
        .extract_inner::<bool>("auth.enabled")
        .context("loading auth.enabled")?;
    if !enabled {
        return Ok(None);
    }
    let value: serde_json::Value = figment.extract().context("loading configuration")?;
    Ok(Some(auth::config::Config::from_value(value)))
}

fn identity_openapi() -> anyhow::Result<()> {
    use utoipa::OpenApi;
    let doc = auth::openapi::ApiDoc::openapi();
    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../lib/auth/OpenApi.Auth.json");
    std::fs::write(&path, doc.to_pretty_json()?).context("writing identity OpenAPI")?;
    println!("wrote {}", path.display());
    Ok(())
}

const QUEUE_DEPTH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);

/// 周期性上报队列与死信队列的积压深度；进程退出时随运行时一起结束。
fn spawn_queue_depth_reporter(redis: export::AppRedis, queues: &[(&'static str, String)]) {
    if queues.is_empty() {
        return;
    }
    let queues = queues.to_vec();
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(QUEUE_DEPTH_INTERVAL);
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            ticker.tick().await;
            for (topic, key) in &queues {
                match redis.llen(key).await {
                    Ok(depth) => oltp::metrics::queue_depth(topic, depth),
                    Err(error) => {
                        tracing::debug!(%error, %key, "queue depth sample failed");
                    }
                }
            }
        }
    });
}