653bd9f
#![recursion_limit = "256"]
mod cli;
mod event;
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::Event { mode } => event_run(mode),
}
}
#[tokio::main]
async fn event_run(mode: cli::EventMode) -> anyhow::Result<()> {
event::run(mode).await
}
#[tokio::main]
async fn serve() -> anyhow::Result<()> {
use actix_web::middleware::from_fn;
use actix_web::{App, HttpServer, web};
use config::AppConfig;
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 _telemetry_guard = telemetry.guard();
let cfg = AppConfig::load().context("loading configuration")?;
if !cfg.apihost.enabled() {
info!("apihost is disabled by config; exiting");
return Ok(());
}
push::protocol::set_push_enabled(cfg.push.enabled);
let cancel = CancellationToken::new();
let app_config = cfg.clone();
let session_config = app_config.session.clone();
let database = std::sync::Arc::new(
db::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 mq_runtime = queue::QueueRuntime::connect("serve")
.map_err(|err| anyhow::anyhow!("failed to create the rocketmq runtime: {err}"))?;
let conn = queue::RocketMqConn::init(&cfg.rocketmq, &mq_runtime)
.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 state = bootstrap::AppState::with_identity(
app_config,
database,
Some(mail),
Some(auth_events.clone()),
)
.await
.context("building AppState")?;
let session_store = auth::session::RedisStore::new(state.cache.clone());
let deps = webapi::bootstrap::Deps::new(state.clone(), session_store.clone());
chat::share::fanout::start(deps.chat.clone(), cancel.clone());
info!("dependencies ready");
let internal_rpc = start_internal_rpc(&cfg, &deps, &cancel)?;
spawn_queue_depth_reporter(
state.redis.clone(),
&[
("push.jobs", push::protocol::QUEUE_TOPIC.to_string()),
("push.jobs.dlq", push::protocol::DLQ_TOPIC.to_string()),
],
);
spawn_dependency_health_sampler(state.clone(), conn.clone());
let mut workers = tokio::task::JoinSet::new();
workers.spawn(chat::worker::run_background_tasks(
deps.chat.clone(),
cancel.clone(),
));
workers.spawn(deletion::run_worker(
state.clone(),
deps.accounts.clone(),
cancel.clone(),
));
let relay = events::outbox::OutboxRelay::start(
Some(auth_events),
Some(queue::Publisher::new(conn.clone())),
vec![("apihost", state.database.primary().clone())],
);
let relay_cancel = cancel.clone();
let (relay_stop, relay_stop_rx) = tokio::sync::oneshot::channel::<()>();
let relay_task = tokio::spawn(async move {
let mut relay = relay;
tokio::select! {
result = relay.wait() => {
match result {
Ok(()) => tracing::error!("event outbox relay stopped unexpectedly"),
Err(error) => tracing::error!(%error, "event outbox relay failed"),
}
relay_cancel.cancel();
true
}
_ = relay_stop_rx => {
if let Err(error) = relay.shutdown().await {
warn!(%error, "shutting down the event outbox relay failed");
}
false
}
}
});
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 app_deps = deps.clone();
let server = HttpServer::new(move || {
let state = app_state.clone();
let deps = app_deps.clone();
let app = App::new()
.app_data(web::Data::new(state.clone()))
.app_data(web::Data::new(deps.accounts.clone()))
.app_data(
web::JsonConfig::default()
.limit(webapi::MAX_REQUEST_BODY_BYTES)
.error_handler(api::http::json_error),
)
.app_data(web::PayloadConfig::new(webapi::MAX_REQUEST_BODY_BYTES));
app.wrap(from_fn(webapi::middleware::revalidate::revalidate))
.wrap(auth::session::SessionMiddleware::new(
session_config.clone(),
session_store.clone(),
))
.wrap(webapi::build_cors(&state.config.web))
.wrap(from_fn(webapi::middleware::csrf::csrf))
.wrap(oltp::actix::HttpTelemetry::new("gitdata"))
.wrap(from_fn(api::middleware::trace_request))
.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::routes(&deps))
.default_service(web::to(webapi::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 Some(task) = internal_rpc {
let _ = task.await;
}
let _ = relay_stop.send(());
let relay_failed = relay_task.await.unwrap_or(true);
for task in push_tasks {
let _ = task.await;
}
drain_workers(&mut workers, &cancel).await;
conn.shutdown().await;
mq_runtime.shutdown().await;
api_result?;
if relay_failed {
return Err(anyhow::anyhow!("event outbox relay stopped unexpectedly"));
}
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 _telemetry_guard = telemetry.guard();
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 mq_runtime = queue::QueueRuntime::connect("mail")
.map_err(|err| anyhow!("failed to create the rocketmq runtime: {err}"))?;
let conn = queue::RocketMqConn::init(&rocketmq, &mq_runtime)
.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, &mq_runtime, 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;
mq_runtime.shutdown().await;
result.context("mail service failed")
}
#[tokio::main]
async fn migrate() -> anyhow::Result<()> {
use config::migrate::MigrateConfig;
use models::migrate::sync_schema;
use sea_orm::Database;
use tracing::info;
const DEFAULT_LOG_FILTER: &str = "info,app=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 app -- 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)))
}
/// 启动平台侧内部 RPC 服务端（Identity + Chat/Issue/Pull/Space 授权）。
///
/// 未配置 `internal.rpc.listen_address` 时跳过：serve 仍可单独运行。
fn start_internal_rpc(
cfg: &config::AppConfig,
deps: &webapi::bootstrap::Deps,
cancel: &tokio_util::sync::CancellationToken,
) -> anyhow::Result<Option<tokio::task::JoinHandle<()>>> {
let listen = cfg.internal.rpc.listen_address.trim();
if listen.is_empty() {
return Ok(None);
}
if cfg.internal.rpc.rpc_key.trim().is_empty() {
return Err(anyhow!(
"internal.rpc.rpc_key must be configured when internal.rpc.listen_address is set"
));
}
let server_config = internal::ServerConfig::new(listen, cfg.internal.rpc.rpc_key.clone());
let router = internal::server::router(&server_config).map_err(|error| anyhow!("{error}"))?;
let gate = server_config.gate();
let router = auth::rpc::register(router, &deps.auth, gate.clone());
let router = chat::rpc::register(router, &deps.chat, gate.clone());
let router = space::rpc::register(router, &deps.space, gate);
let address = server_config.listen_address().to_string();
let shutdown = cancel.clone();
tracing::info!(%address, "internal rpc listening");
Ok(Some(tokio::spawn(async move {
let stop = async move { shutdown.cancelled().await };
if let Err(error) = internal::server::serve(router, &address, stop).await {
tracing::error!(%error, "internal rpc server stopped");
}
})))
}
const WORKER_SHUTDOWN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
/// 等待后台 worker 收尾；超时则中止，未完成的删除行由租约过期后重新领取。
async fn drain_workers(
workers: &mut tokio::task::JoinSet<()>,
cancel: &tokio_util::sync::CancellationToken,
) {
let drained = tokio::time::timeout(WORKER_SHUTDOWN_TIMEOUT, async {
while let Some(result) = workers.join_next().await {
match result {
Ok(()) => {
if !cancel.is_cancelled() {
tracing::warn!("a background worker stopped unexpectedly");
}
}
Err(error) if error.is_panic() => {
tracing::error!(%error, "background worker panicked");
}
Err(error) => {
tracing::error!(%error, "background worker failed");
}
}
}
})
.await;
if drained.is_err() {
tracing::warn!("background workers did not stop in time; aborting them");
workers.shutdown().await;
}
}
const QUEUE_DEPTH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);
const HEALTH_SAMPLE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10);
const HEALTH_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
/// 周期性采样 RocketMQ/S3 健康：`/readyz` 只读缓存，避免每次探活都新建连接。
fn spawn_dependency_health_sampler(state: bootstrap::AppState, conn: queue::RocketMqConn) {
tokio::spawn(async move {
let mut ticker = tokio::time::interval(HEALTH_SAMPLE_INTERVAL);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let auth_topic = state
.config
.rocketmq
.topic(events::EventKind::AuthLoginCreate.topic());
loop {
ticker.tick().await;
let rocketmq_up = tokio::time::timeout(
std::time::Duration::from_secs(6),
conn.health(&[&auth_topic, "TBW102"]),
)
.await
.is_ok_and(|result| result.is_ok());
state.health.mark_rocketmq(rocketmq_up);
oltp::metrics::health_dependency("rocketmq", rocketmq_up);
let s3_up = matches!(
tokio::time::timeout(HEALTH_PROBE_TIMEOUT, state.storage.probe()).await,
Ok(Ok(()))
);
state.health.mark_s3(s3_up);
oltp::metrics::health_dependency("s3", s3_up);
}
});
}
/// 周期性上报队列与死信队列的积压深度；进程退出时随运行时一起结束。
fn spawn_queue_depth_reporter(redis: cache::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");
}
}
}
}
});
}