8b889b7
#![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 {
Command::Serve => serve(),
Command::Migrate => migrate(),
Command::Openapi { check } => openapi(check),
}
}
#[tokio::main]
async fn serve() -> anyhow::Result<()> {
use axum::Router;
use config::scoped::{ApihostConfig, ScopedConfig};
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};
use utoipa_swagger_ui::SwaggerUi;
use webapi::{openapi::api_doc, router};
export::telemetry::init_tracing();
let cfg = ApihostConfig::load().context("loading configuration")?;
if !cfg.apihost.enabled() {
info!("apihost is disabled by config; exiting");
return Ok(());
}
info!("connecting to database / redis / storage / email");
let state = export::AppState::with_config(cfg.clone().into_app_config())
.await
.context("building AppState")?;
info!("dependencies ready");
if let Err(err) = export::telemetry::init_meter_provider() {
warn!(error = %err, "failed to init meter provider; metrics disabled");
}
export::metrics::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()));
if cfg.rocketmq.enabled {
let conn = queue::RocketMqConn::init(&cfg.rocketmq)
.await
.map_err(|err| anyhow::anyhow!("failed to start the rocketmq producer: {err}"))?;
tokio::spawn(queue::relay::run(
state.database.primary().clone(),
queue::Publisher::new(conn),
CancellationToken::new(),
));
info!("event outbox relay started");
} else {
info!("rocketmq is disabled; chat message events are not published");
}
if cfg.push.enabled {
for _ in 0..cfg.push.workers.max(1) {
let worker = push::PushWorker::new(state.clone(), CancellationToken::new())
.map_err(|err| anyhow::anyhow!("failed to start push worker: {err}"))?;
tokio::spawn(worker.run());
}
info!(
workers = cfg.push.workers.max(1),
"web push workers started"
);
}
let api_router: Router = router(state)
.await
.map_err(|err| anyhow::anyhow!("failed to build router: {err}"))?;
let doc = api_doc();
let api_router =
api_router.merge(SwaggerUi::new("/api/docs").url("/api/docs/openapi.json", doc));
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 listener = TcpListener::bind(addr)
.await
.with_context(|| format!("binding {addr}"))?;
info!(%addr, "apihost listening");
axum::serve(listener, api_router.into_make_service())
.with_graceful_shutdown(export::telemetry::shutdown_signal())
.await
.context("axum server error")?;
export::telemetry::shutdown_telemetry();
info!("apihost shut down");
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")?;
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(())
}