449 lines
15 KiB
Rust
449 lines
15 KiB
Rust
mod auth;
|
|
mod config;
|
|
mod error;
|
|
mod recording_routes;
|
|
mod routes;
|
|
mod speech;
|
|
mod summary;
|
|
mod summary_worker;
|
|
mod transcription_worker;
|
|
|
|
use std::{path::PathBuf, sync::Arc, time::Duration};
|
|
|
|
use axum::{
|
|
Json, Router,
|
|
extract::{DefaultBodyLimit, MatchedPath},
|
|
http::{HeaderName, HeaderValue, Request, StatusCode},
|
|
response::Response,
|
|
routing::get,
|
|
};
|
|
use config::{Config, LogFormat};
|
|
use sqlx::{PgPool, postgres::PgPoolOptions};
|
|
use tower::ServiceBuilder;
|
|
use tower_http::{
|
|
classify::ServerErrorsFailureClass,
|
|
cors::CorsLayer,
|
|
request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer},
|
|
trace::TraceLayer,
|
|
};
|
|
use tracing::Span;
|
|
use tracing_subscriber::EnvFilter;
|
|
use utoipa_scalar::{Scalar, Servable};
|
|
|
|
#[derive(Clone)]
|
|
pub struct AppState {
|
|
pub pool: Option<PgPool>,
|
|
pub audio_storage_dir: PathBuf,
|
|
pub speech: speech::SpeechService,
|
|
pub summary: summary::SummaryService,
|
|
pub auth: auth::AuthService,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub(crate) struct LogContext {
|
|
pub(crate) environment: String,
|
|
pub(crate) app_version: String,
|
|
pub(crate) git_sha: String,
|
|
}
|
|
|
|
impl From<&Config> for LogContext {
|
|
fn from(config: &Config) -> Self {
|
|
Self {
|
|
environment: config.app_env.clone(),
|
|
app_version: config.app_version.clone(),
|
|
git_sha: config.git_sha.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn should_log_http_request_started(path: &str) -> bool {
|
|
path != "/health"
|
|
}
|
|
|
|
fn should_log_http_response(is_health_check: bool, status: StatusCode) -> bool {
|
|
!is_health_check || !status.is_success()
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug)]
|
|
struct HealthCheckResponse;
|
|
|
|
async fn mark_health_check_response(
|
|
request: axum::extract::Request,
|
|
next: axum::middleware::Next,
|
|
) -> Response {
|
|
let is_health_check = request.uri().path() == "/health";
|
|
let mut response = next.run(request).await;
|
|
if is_health_check {
|
|
response.extensions_mut().insert(HealthCheckResponse);
|
|
}
|
|
response
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
|
dotenvy::dotenv().ok();
|
|
let config = Config::from_env().map_err(std::io::Error::other)?;
|
|
init_tracing(&config);
|
|
let log_context = LogContext::from(&config);
|
|
let pool = connect_database(&config, &log_context).await?;
|
|
let speech = speech::SpeechService::new(config.speech.clone());
|
|
let auth = auth::AuthService::new(config.auth.clone());
|
|
let state = AppState {
|
|
pool,
|
|
audio_storage_dir: config.audio_storage_dir.clone(),
|
|
speech: speech.clone(),
|
|
summary: summary::SummaryService::new(config.summary.clone()),
|
|
auth,
|
|
};
|
|
if let Some(pool) = state.pool.clone() {
|
|
if speech.configured() {
|
|
transcription_worker::spawn(pool.clone(), speech.clone(), log_context.clone());
|
|
}
|
|
summary_worker::spawn(pool, state.summary.clone(), log_context.clone());
|
|
}
|
|
let app = build_app(
|
|
state,
|
|
config.cors_allowed_origin.as_deref(),
|
|
log_context.clone(),
|
|
)?;
|
|
let address = std::net::SocketAddr::from((config.host, config.port));
|
|
let listener = tokio::net::TcpListener::bind(address).await?;
|
|
|
|
tracing::info!(
|
|
service = "api",
|
|
environment = %log_context.environment,
|
|
app_version = %log_context.app_version,
|
|
git_sha = %log_context.git_sha,
|
|
event = "api_started",
|
|
%address,
|
|
scalar = %format!("http://{address}/scalar"),
|
|
provider = ?speech.provider_name(),
|
|
"API server is listening"
|
|
);
|
|
axum::serve(listener, app)
|
|
.with_graceful_shutdown(shutdown_signal(log_context))
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
fn init_tracing(config: &Config) {
|
|
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into());
|
|
match config.log_format {
|
|
LogFormat::Json => tracing_subscriber::fmt()
|
|
.json()
|
|
.flatten_event(true)
|
|
.with_ansi(false)
|
|
.with_current_span(true)
|
|
.with_span_list(true)
|
|
.with_env_filter(filter)
|
|
.init(),
|
|
LogFormat::Pretty => tracing_subscriber::fmt()
|
|
.with_env_filter(filter)
|
|
.pretty()
|
|
.init(),
|
|
}
|
|
}
|
|
|
|
fn build_app(
|
|
state: AppState,
|
|
cors_allowed_origin: Option<&str>,
|
|
log_context: LogContext,
|
|
) -> Result<Router, String> {
|
|
let (api_router, openapi) = routes::router();
|
|
let openapi_json = openapi.clone();
|
|
let mut app = api_router
|
|
.route(
|
|
"/openapi.json",
|
|
get(move || {
|
|
let openapi = openapi_json.clone();
|
|
async move { Json(openapi) }
|
|
}),
|
|
)
|
|
.merge(Scalar::with_url("/scalar", openapi))
|
|
.with_state(Arc::new(state))
|
|
.layer(DefaultBodyLimit::max(20 * 1024 * 1024));
|
|
|
|
if let Some(origin) = cors_allowed_origin {
|
|
let allowed_origin = HeaderValue::from_str(origin)
|
|
.map_err(|_| "CORS_ALLOWED_ORIGIN must be a valid HTTP header value".to_owned())?;
|
|
app = app.layer(
|
|
CorsLayer::new()
|
|
.allow_origin(allowed_origin)
|
|
.expose_headers([HeaderName::from_static("x-request-id")]),
|
|
);
|
|
}
|
|
|
|
let request_span_context = log_context.clone();
|
|
let request_started_context = log_context.clone();
|
|
let request_finished_context = log_context.clone();
|
|
let request_failed_context = log_context;
|
|
Ok(app.layer(
|
|
ServiceBuilder::new()
|
|
.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
|
|
.layer(
|
|
TraceLayer::new_for_http()
|
|
.make_span_with(move |request: &Request<_>| {
|
|
make_http_request_span(request, &request_span_context)
|
|
})
|
|
.on_request(move |request: &Request<_>, span: &Span| {
|
|
if !should_log_http_request_started(request.uri().path()) {
|
|
return;
|
|
}
|
|
tracing::info!(
|
|
parent: span,
|
|
service = "api",
|
|
environment = %request_started_context.environment,
|
|
app_version = %request_started_context.app_version,
|
|
git_sha = %request_started_context.git_sha,
|
|
event = "http_request_started",
|
|
request_id = request_id_from_headers(request),
|
|
"HTTP request started"
|
|
);
|
|
})
|
|
.on_response(move |response: &Response, latency: Duration, span: &Span| {
|
|
let status = response.status();
|
|
let is_health_check =
|
|
response.extensions().get::<HealthCheckResponse>().is_some();
|
|
if !should_log_http_response(is_health_check, status) {
|
|
return;
|
|
}
|
|
let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX);
|
|
let request_id = response
|
|
.headers()
|
|
.get("x-request-id")
|
|
.and_then(|value| value.to_str().ok())
|
|
.unwrap_or("unknown");
|
|
if status.is_server_error() {
|
|
tracing::error!(
|
|
parent: span,
|
|
service = "api",
|
|
environment = %request_finished_context.environment,
|
|
app_version = %request_finished_context.app_version,
|
|
git_sha = %request_finished_context.git_sha,
|
|
event = "http_request_finished",
|
|
status = status.as_u16(),
|
|
latency_ms,
|
|
request_id,
|
|
error_class = "server_error",
|
|
"HTTP request finished"
|
|
);
|
|
} else {
|
|
tracing::info!(
|
|
parent: span,
|
|
service = "api",
|
|
environment = %request_finished_context.environment,
|
|
app_version = %request_finished_context.app_version,
|
|
git_sha = %request_finished_context.git_sha,
|
|
event = "http_request_finished",
|
|
status = status.as_u16(),
|
|
latency_ms,
|
|
request_id,
|
|
"HTTP request finished"
|
|
);
|
|
}
|
|
})
|
|
.on_failure(
|
|
move |failure: ServerErrorsFailureClass, latency: Duration, span: &Span| {
|
|
let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX);
|
|
tracing::error!(
|
|
parent: span,
|
|
service = "api",
|
|
environment = %request_failed_context.environment,
|
|
app_version = %request_failed_context.app_version,
|
|
git_sha = %request_failed_context.git_sha,
|
|
event = "http_request_failed",
|
|
latency_ms,
|
|
error_class = ?failure,
|
|
"HTTP request failed"
|
|
);
|
|
},
|
|
),
|
|
)
|
|
.layer(axum::middleware::from_fn(mark_health_check_response))
|
|
.layer(PropagateRequestIdLayer::x_request_id()),
|
|
))
|
|
}
|
|
|
|
fn make_http_request_span<B>(request: &Request<B>, context: &LogContext) -> Span {
|
|
let route = request
|
|
.extensions()
|
|
.get::<MatchedPath>()
|
|
.map(MatchedPath::as_str)
|
|
.unwrap_or_else(|| request.uri().path());
|
|
let request_id = request_id_from_headers(request);
|
|
|
|
tracing::info_span!(
|
|
"http_request",
|
|
service = "api",
|
|
environment = %context.environment,
|
|
app_version = %context.app_version,
|
|
git_sha = %context.git_sha,
|
|
method = %request.method(),
|
|
route,
|
|
path = %request.uri().path(),
|
|
request_id,
|
|
)
|
|
}
|
|
|
|
fn request_id_from_headers<B>(request: &Request<B>) -> &str {
|
|
request
|
|
.headers()
|
|
.get("x-request-id")
|
|
.and_then(|value| value.to_str().ok())
|
|
.unwrap_or("invalid")
|
|
}
|
|
|
|
async fn connect_database(
|
|
config: &Config,
|
|
log_context: &LogContext,
|
|
) -> Result<Option<PgPool>, sqlx::Error> {
|
|
let Some(database_url) = &config.database_url else {
|
|
tracing::warn!(
|
|
service = "api",
|
|
environment = %log_context.environment,
|
|
app_version = %log_context.app_version,
|
|
git_sha = %log_context.git_sha,
|
|
event = "database_not_configured",
|
|
"DATABASE_URL is not set; data endpoints will return HTTP 503"
|
|
);
|
|
return Ok(None);
|
|
};
|
|
|
|
let pool = PgPoolOptions::new()
|
|
.max_connections(config.database_max_connections)
|
|
.connect(database_url)
|
|
.await?;
|
|
tracing::info!(
|
|
service = "api",
|
|
environment = %log_context.environment,
|
|
app_version = %log_context.app_version,
|
|
git_sha = %log_context.git_sha,
|
|
event = "database_connected",
|
|
"connected to PostgreSQL"
|
|
);
|
|
Ok(Some(pool))
|
|
}
|
|
|
|
async fn shutdown_signal(log_context: LogContext) {
|
|
let _ = tokio::signal::ctrl_c().await;
|
|
tracing::info!(
|
|
service = "api",
|
|
environment = %log_context.environment,
|
|
app_version = %log_context.app_version,
|
|
git_sha = %log_context.git_sha,
|
|
event = "api_shutdown",
|
|
"shutdown signal received"
|
|
);
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use axum::{
|
|
body::Body,
|
|
http::{Request, StatusCode},
|
|
};
|
|
use tower::ServiceExt;
|
|
use uuid::Uuid;
|
|
|
|
use super::{
|
|
AppState, HealthCheckResponse, LogContext, build_app, should_log_http_request_started,
|
|
should_log_http_response,
|
|
};
|
|
use crate::{
|
|
auth::AuthService, config::AuthConfig, speech::SpeechService, summary::SummaryService,
|
|
};
|
|
|
|
fn app() -> axum::Router {
|
|
build_app(
|
|
AppState {
|
|
pool: None,
|
|
audio_storage_dir: std::path::PathBuf::from("./data/audio"),
|
|
speech: SpeechService::new(None),
|
|
summary: SummaryService::new(None),
|
|
auth: AuthService::new(AuthConfig {
|
|
wechat: None,
|
|
session_ttl_days: 30,
|
|
allow_development_user_header: false,
|
|
}),
|
|
},
|
|
None,
|
|
LogContext {
|
|
environment: "test".to_owned(),
|
|
app_version: "test".to_owned(),
|
|
git_sha: "test".to_owned(),
|
|
},
|
|
)
|
|
.unwrap()
|
|
}
|
|
|
|
#[test]
|
|
fn successful_health_request_is_not_logged() {
|
|
assert!(!should_log_http_request_started("/health"));
|
|
assert!(!should_log_http_response(true, StatusCode::OK));
|
|
}
|
|
|
|
#[test]
|
|
fn failed_health_request_is_logged() {
|
|
assert!(should_log_http_response(
|
|
true,
|
|
StatusCode::INTERNAL_SERVER_ERROR
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn ordinary_successful_request_is_logged() {
|
|
assert!(should_log_http_request_started("/api/v1/profiles"));
|
|
assert!(should_log_http_response(false, StatusCode::OK));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn health_response_is_marked_for_log_filtering() {
|
|
let response = app()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri("/health")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert!(response.extensions().get::<HealthCheckResponse>().is_some());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn health_generates_a_uuid_request_id() {
|
|
let response = app()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri("/health")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let request_id = response
|
|
.headers()
|
|
.get("x-request-id")
|
|
.unwrap()
|
|
.to_str()
|
|
.unwrap();
|
|
assert!(Uuid::parse_str(request_id).is_ok());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn health_propagates_a_client_request_id() {
|
|
let response = app()
|
|
.oneshot(
|
|
Request::builder()
|
|
.uri("/health")
|
|
.header("x-request-id", "test-request-id")
|
|
.body(Body::empty())
|
|
.unwrap(),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.headers()["x-request-id"], "test-request-id");
|
|
}
|
|
}
|