mod assets;

use crate::cache::Cache;
use crate::config::Config;
use crate::config::app::AppRegistry;
use crate::config::auth::AuthConfig;
use crate::config::jwt::JwtConfig;
use crate::config::pow::PowConfig;
use crate::config::redis::RedisConfig;
use crate::config::rocketmq::RocketMqConfig;
use crate::config::server::ServerConfig;
use crate::config::session::SessionConfig;
use crate::config::web::WebConfig;
use crate::db::DbConn;
use crate::email::EmailServiceBuilder;
use crate::events::EventHandler;
use crate::queue::RocketMqConn;
use crate::server::AuthSvc;
use crate::session::{RedisStore, SessionMiddleware};
use actix_web::{App, HttpServer, web};
use anyhow::{Context, Result};
use std::net::{IpAddr, SocketAddr};

const APP: &str = "identity";
/// 邮件订阅者排空时的宽限时间。
const EMAIL_SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(10);
const GRPC_SHUTDOWN_GRACE_SECS: u64 = 10;

#[derive(Clone)]
struct AppState {
    service: AuthSvc,
    session: SessionConfig,
    store: RedisStore,
}

/// host 网络部署下绑非回环地址即把单密钥门禁的 gRPC 面暴露公网；
/// 显式设 KENOAUTH_ALLOW_PUBLIC_BIND=1 才放行。
fn ensure_public_bind_allowed(addr: SocketAddr, label: &str) -> Result<SocketAddr> {
    let loopback = match addr.ip() {
        IpAddr::V4(ip) => ip.is_loopback(),
        IpAddr::V6(ip) => ip.is_loopback(),
    };
    if loopback || std::env::var("KENOAUTH_ALLOW_PUBLIC_BIND").as_deref() == Ok("1") {
        return Ok(addr);
    }
    Err(anyhow::anyhow!(
        "{label} binds a non-loopback address ({addr}); the grpc plane is guarded by a single \
         shared rpc_key. Set KENOAUTH_ALLOW_PUBLIC_BIND=1 to confirm public exposure."
    ))
}

pub async fn start(config: &Config, db: DbConn) -> Result<Runtime> {
    let server: ServerConfig = config.section()?;
    crate::jwt::service::install(config.section()?).context("loading service identity")?;
    let auth_config: AuthConfig = config.section()?;
    let jwt_config: JwtConfig = config.section()?;
    let pow_config: PowConfig = config.section()?;
    let apps: AppRegistry = config.section_or_default()?;
    let redis: RedisConfig = config.section()?;
    let session_config: SessionConfig = config.section()?;
    let rocketmq: RocketMqConfig = config.section()?;
    let web: WebConfig = config.section_or_default()?;
    let email_pool: crate::config::email::EmailPoolConfig = config.section_or_default()?;

    auth_config
        .validate_auth()
        .map_err(anyhow::Error::msg)
        .context("invalid auth configuration")?;
    pow_config
        .validate()
        .map_err(anyhow::Error::msg)
        .context("invalid pow configuration")?;
    apps.validate()
        .map_err(anyhow::Error::msg)
        .context("invalid apps configuration")?;
    redis
        .validate()
        .map_err(anyhow::Error::msg)
        .context("invalid redis configuration")?;
    rocketmq
        .validate()
        .map_err(anyhow::Error::msg)
        .context("invalid rocketmq configuration")?;
    email_pool
        .is_valid()
        .map_err(anyhow::Error::msg)
        .context("invalid email configuration")?;
    web.is_valid()
        .map_err(anyhow::Error::msg)
        .context("invalid web configuration")?;
    crate::config::session::ensure_secure_cookie(
        session_config.secure,
        server.http_addr,
        std::env::var("KENOAUTH_ALLOW_INSECURE_COOKIE").as_deref() == Ok("1"),
    )
    .map_err(anyhow::Error::msg)
    .context("invalid session configuration")?;
    let cors = crate::envelope::Cors::new(&web);

    let cache = Cache::connect(&redis)
        .await
        .context("connecting to redis")?;
    let mq = RocketMqConn::init(&rocketmq)
        .await
        .context("connecting to rocketmq")?;
    let events = EventHandler::new(mq);

    let mut email_builder = EmailServiceBuilder::new(email_pool)
        .with_events(events.clone())
        .with_database(db.clone());
    if config.has("smtp") {
        let smtp: crate::config::smtp::SmtpConfig = config.section()?;
        email_builder = email_builder.with_transport(smtp);
    }
    let email_service = email_builder
        .register_default_templates()
        .await
        .build()
        .await
        .context("building the email service")?;
    let email_enabled = email_service.enabled();
    let email_queue_depth = email_service.depth();
    let email = Some(email_service);

    let session_max_age_secs = session_config.max_age.max(1) as u64;
    let service = AuthSvc::new(
        db,
        cache.clone(),
        email,
        Some(events),
        auth_config,
        pow_config,
        apps,
        jwt_config,
        session_max_age_secs,
    )
    .await?;
    let state = AppState {
        service: service.clone(),
        session: session_config,
        store: RedisStore::new(cache),
    };

    let grpc_addr = ensure_public_bind_allowed(server.grpc_addr, "grpc")?;
    let grpc_listener = tokio::net::TcpListener::bind(grpc_addr)
        .await
        .with_context(|| format!("binding grpc on {grpc_addr}"))?;
    let grpc = crate::server::grpc::router(service.clone())
        .await
        .context("building the grpc router")?;

    let http = HttpServer::new(move || {
        App::new()
            .wrap(SessionMiddleware::new(
                state.session.clone(),
                state.store.clone(),
            ))
            .wrap(crate::oltp::HttpTelemetry::new("identity"))
            .wrap(crate::envelope::EnvelopeContext)
            .wrap(cors.clone())
            .app_data(web::Data::new(state.service.clone()))
            .app_data(web::Data::new(web.clone()))
            .configure(crate::server::http::router::configure)
            .default_service(web::to(assets::serve))
    })
    .disable_signals()
    .bind(ensure_public_bind_allowed(server.http_addr, "http")?)
    .with_context(|| format!("binding http on {}", server.http_addr))?
    .run();

    let event_forwarding = service.verify_event_forwarding().await;
    tracing::info!(
        app = APP,
        config = %config.path().display(),
        http = %server.http_addr,
        grpc = %server.grpc_addr,
        assets = assets::count(),
        email = email_enabled,
        email_queue_depth,
        event_forwarding,
        "identity serving"
    );

    Ok(Runtime {
        service,
        http,
        grpc,
        grpc_listener,
    })
}

pub struct Runtime {
    service: AuthSvc,
    http: actix_web::dev::Server,
    grpc: crate::server::grpc::GrpcRouter,
    grpc_listener: tokio::net::TcpListener,
}

impl Runtime {
    pub fn service(&self) -> &AuthSvc {
        &self.service
    }

    pub async fn serve_until(self, shutdown: tokio_util::sync::CancellationToken) -> Result<()> {
        let Self {
            service,
            http,
            grpc,
            grpc_listener,
        } = self;
        let handle = http.handle();
        let grpc_cancel = shutdown.clone();
        let incoming = tokio_stream::wrappers::TcpListenerStream::new(grpc_listener);
        let grpc = grpc.serve_with_incoming_shutdown(incoming, async move {
            grpc_cancel.cancelled().await;
        });
        tokio::pin!(http, grpc);
        let mut grpc_done = false;
        let mut http_done = false;
        let result = tokio::select! {
            result = &mut http => { http_done = true; result.context("identity HTTP server stopped") },
            result = &mut grpc => { grpc_done = true; result.context("identity gRPC server stopped") },
            _ = shutdown.cancelled() => Ok(()),
        };
        shutdown.cancel();
        let http_shutdown = async {
            if !http_done {
                let stop = handle.stop(true);
                let (_, result) = tokio::join!(stop, &mut http);
                if let Err(error) = result {
                    tracing::warn!(%error, "identity HTTP shutdown failed");
                }
            }
        };
        let grpc_shutdown = async {
            if !grpc_done {
                match tokio::time::timeout(
                    std::time::Duration::from_secs(GRPC_SHUTDOWN_GRACE_SECS),
                    &mut grpc,
                )
                .await
                {
                    Ok(Ok(())) => {}
                    Ok(Err(error)) => tracing::warn!(%error, "identity gRPC shutdown failed"),
                    Err(_) => tracing::warn!("identity gRPC shutdown timed out"),
                }
            }
        };
        tokio::join!(http_shutdown, grpc_shutdown);
        if let Some(email) = service.email()
            && let Err(error) = email.shutdown(EMAIL_SHUTDOWN_GRACE).await
        {
            tracing::warn!(%error, "draining the email queue failed");
        }
        if let Some(events) = service.events()
            && let Err(error) = events.shutdown().await
        {
            tracing::warn!(%error, "shutting down identity producer failed");
        }
        result
    }
}