70677ed
//! `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 _telemetry_guard = telemetry.guard();
let cfg = EventConfig::load().context("loading event configuration")?;
let mq_runtime = match mode {
EventMode::Actor => None,
EventMode::Single | EventMode::Consumer => Some(
queue::QueueRuntime::connect("event")
.map_err(|error| anyhow!("creating the rocketmq runtime: {error}"))?,
),
};
let platform = PlatformClient::new(&cfg.platform).map_err(|error| anyhow!("{error}"))?;
let conn = cache::CacheConn::init(&cfg.redis)
.await
.map_err(|error| anyhow!("connecting redis: {error}"))?;
let actor_conn = actor::RedisConnection::from_conn(conn.clone());
let nonce_store = internal::NonceStore::from_conn(conn);
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())
.connection(actor_conn.clone())
.register(|_| EventActor)
.start()
.await
.map_err(|error| anyhow!("starting the actor host: {error}"))?;
let subscriptions = Subscriptions::from_connection(actor_conn);
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());
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 = async {
match mq_runtime.as_ref() {
Some(runtime) => {
socket_runtime::run_consumer(
&cfg.consumer,
runtime,
host.clone(),
platform.clone(),
subscriptions.clone(),
shutdown_rx.clone(),
)
.await
}
None => std::future::pending().await,
}
};
let result = match mode {
EventMode::Single => {
let (socket_result, consumer_result) = tokio::join!(socket_future, consumer_future);
socket_result.and(consumer_result)
}
EventMode::Actor => {
drop(consumer_future);
socket_future.await
}
EventMode::Consumer => consumer_future.await,
};
shutdown_tx.send_replace(true);
signal_task.abort();
let _ = rpc_task.await;
host.shutdown().await;
if let Some(runtime) = mq_runtime {
runtime.shutdown().await;
}
tokio::time::sleep(Duration::from_millis(10)).await;
result.map_err(|error| anyhow!("event system failed: {error}"))
}