//! `app event` 子命令：事件系统（RocketMQ 消费 + socket.io 通道 + 内部 RPC）。
//!
//! 只加载 `event` 段（socket/actor/redis/consumer/rpc/platform），不加载数据库、
//! S3 或任何业务域配置；身份与权限全部经内部 RPC 问 `serve`。

use std::sync::Arc;
use std::time::Duration;

use crate::cli::EventMode;
use anyhow::{Context, anyhow};
use config::event::EventConfig;
use events::gateway::event_service::EventService;
use events::gateway::platform::PlatformClient;
use events::gateway::socket_runtime::{self, EventActor};
use events::gateway::subscriptions::Subscriptions;
use socket::{SocketHub, SocketOptions};
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;

pub async fn run(mode: EventMode) -> anyhow::Result<()> {
    let telemetry = oltp::init(&oltp::TelemetryConfig::from_env("gitdata-event"));

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

    let platform = PlatformClient::new(&cfg.platform).map_err(|error| anyhow!("{error}"))?;
    let nonce_store = internal::NonceStore::connect(&cfg.redis, "event")
        .await
        .map_err(|error| anyhow!("{error}"))?;
    let private_pem = cfg
        .rpc
        .identity_private_key
        .resolve()
        .map_err(|error| anyhow!("resolving event identity private key: {error}"))?;
    let mac_key = cfg
        .rpc
        .platform_mac_key
        .resolve()
        .map_err(|error| anyhow!("resolving platform mac key: {error}"))?;
    let verifier = internal::EnvelopeVerifier::new(
        &cfg.rpc.identity_key_id,
        &private_pem,
        &cfg.rpc.platform_mac_key_id,
        mac_key.as_bytes(),
        cfg.rpc.clock_skew_secs,
        cfg.rpc.nonce_ttl_secs,
        nonce_store,
    )
    .map_err(|error| anyhow!("{error}"))?;

    let mut options = actor::ActorHostOptions::from(&cfg.actor);
    if let Some(node_id) = cfg
        .actor
        .node_id
        .as_deref()
        .filter(|value| !value.is_empty())
    {
        options.node_id = node_id.to_string();
    }
    let host = actor::ActorHost::builder(options, cfg.redis.clone())
        .register(|_| EventActor)
        .start()
        .await
        .map_err(|error| anyhow!("starting the actor host: {error}"))?;

    let subscriptions = Subscriptions::connect(&cfg.redis)
        .await
        .map_err(|error| anyhow!("connecting the subscription store: {error}"))?;
    let push_store = host.push_store();
    let backend = socket_runtime::backend(
        host.clone(),
        platform.clone(),
        subscriptions.clone(),
        push_store,
    );
    let socket_options = SocketOptions::from(&cfg.socket);
    let (hub, events_rx) = SocketHub::new(socket_options, backend)
        .map_err(|error| anyhow!("building the socket hub: {error}"))?;
    let service = EventService::new(
        host.clone(),
        platform.clone(),
        subscriptions.clone(),
        Arc::new(verifier),
        hub.clone(),
    );

    let cancel = CancellationToken::new();
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    let server_config = internal::ServerConfig::new(
        cfg.rpc.listen_address.clone(),
        cfg.rpc.rpc_key.clone(),
        vec![
            "internal.v1.EventIngressService",
            "internal.v1.SubscriptionService",
        ],
    );
    let router = internal::server::router(&server_config)
        .map_err(|error| anyhow!("building the internal rpc server: {error}"))?;
    let gate = server_config.gate();
    let router = events::rpc::register_ingress(router, &service, gate.clone());
    let router = events::rpc::register(router, &service, gate);
    let rpc_listen = server_config.listen_address().to_string();
    let mut rpc_shutdown = shutdown_rx.clone();
    let rpc_task = tokio::spawn(async move {
        let stop = async move {
            while !*rpc_shutdown.borrow() {
                if rpc_shutdown.changed().await.is_err() {
                    break;
                }
            }
        };
        if let Err(error) = internal::server::serve(router, &rpc_listen, stop).await {
            tracing::error!(%error, "event internal rpc server stopped");
        }
    });

    // 信号：oltp 统一安装（同进程只装一次）。
    let signal = oltp::shutdown::install();
    let signal_tx = shutdown_tx.clone();
    let signal_cancel = cancel.clone();
    let signal_task = tokio::spawn(async move {
        signal.wait().await;
        signal_tx.send_replace(true);
        signal_cancel.cancel();
    });

    tracing::info!(
        mode = ?mode,
        socket = %cfg.socket.listen_address,
        rpc = %cfg.rpc.listen_address,
        "event system starting"
    );

    let socket_future = socket_runtime::run_socket_node(
        hub.clone(),
        events_rx,
        host.clone(),
        platform.clone(),
        subscriptions.clone(),
        &cfg.socket.listen_address,
        shutdown_rx.clone(),
    );
    let consumer_future = socket_runtime::run_consumer(
        &cfg.consumer,
        host.clone(),
        platform.clone(),
        subscriptions.clone(),
        shutdown_rx.clone(),
    );

    let result = match mode {
        EventMode::Single => {
            let (socket_result, consumer_result) = tokio::join!(socket_future, consumer_future);
            socket_result.and(consumer_result)
        }
        EventMode::Actor => socket_future.await,
        EventMode::Consumer => consumer_future.await,
    };

    shutdown_tx.send_replace(true);
    signal_task.abort();
    let _ = rpc_task.await;
    host.shutdown().await;
    tokio::time::sleep(Duration::from_millis(10)).await;
    if let Err(error) = telemetry.shutdown() {
        tracing::warn!(%error, "flushing telemetry failed");
    }
    result.map_err(|error| anyhow!("event system failed: {error}"))
}