Genarrative/server-rs/crates/api-server/src/agc_analytics.rs

//! 客户端埋点只在数据库提交后确认；不使用普通路由 tracking outbox。
use axum::{
    Json, Router,
    extract::{DefaultBodyLimit, Extension, Query, State},
    http::StatusCode,
    middleware,
    routing::{get, post},
};
use serde_json::Value;
use shared_contracts::{admin::AdminAgcTrackingEventListQuery, agc_analytics::AgcAnalyticsBatch};
use spacetime_client::SpacetimeClientError;

use crate::{
    admin::{AuthenticatedAdmin, require_admin_auth},
    api_response::json_success_body,
    auth::{AuthenticatedAccessToken, require_bearer_auth},
    http_error::AppError,
    request_context::RequestContext,
    state::AppState,
};

const MAX_REQUEST_BYTES: usize = 2 * 1024 * 1024;

pub fn router(state: AppState) -> Router<AppState> {
    Router::new()
        .route(
            "/api/agc/analytics/batches",
            post(upload_batch)
                .route_layer(middleware::from_fn_with_state(
                    state.clone(),
                    require_bearer_auth,
                ))
                .layer(DefaultBodyLimit::max(MAX_REQUEST_BYTES)),
        )
        .route(
            "/admin/api/agc/tracking-events",
            get(list_events).route_layer(middleware::from_fn_with_state(state, require_admin_auth)),
        )
}

async fn upload_batch(
    State(state): State<AppState>,
    Extension(context): Extension<RequestContext>,
    Extension(auth): Extension<AuthenticatedAccessToken>,
    payload: Result<Json<Value>, axum::extract::rejection::JsonRejection>,
) -> Result<Json<Value>, AppError> {
    let Json(raw) = payload.map_err(|error| {
        let status = if error.status() == StatusCode::PAYLOAD_TOO_LARGE {
            StatusCode::PAYLOAD_TOO_LARGE
        } else {
            StatusCode::BAD_REQUEST
        };
        AppError::from_status(status).with_message("客户端埋点请求格式无效")
    })?;
    // serde 也能将位置数组解成 struct；上传合同只接受命名字段的 JSON 对象。
    if !raw.is_object()
        || !raw
            .get("events")
            .and_then(Value::as_array)
            .is_some_and(|events| events.iter().all(Value::is_object))
    {
        return Err(AppError::from_status(StatusCode::BAD_REQUEST)
            .with_message("客户端埋点必须使用事件对象"));
    }
    let batch: AgcAnalyticsBatch = serde_json::from_value(raw).map_err(|_| {
        AppError::from_status(StatusCode::BAD_REQUEST).with_message("客户端埋点字段不符合合同")
    })?;
    validate_subject_and_origin(
        &batch,
        auth.claims().user_id(),
        &state.config.agc_analytics_origin,
    )?;
    module_runtime::agc_analytics::validate_agc_analytics_batch(&batch).map_err(|error| {
        let status = if error == "events_too_large" {
            StatusCode::PAYLOAD_TOO_LARGE
        } else {
            StatusCode::BAD_REQUEST
        };
        AppError::from_status(status).with_message("客户端埋点批次格式不符合合同")
    })?;
    let acknowledgement = state
        .spacetime_client()
        .upload_agc_analytics_batch(batch)
        .await
        .map_err(map_database_error)?;
    Ok(json_success_body(Some(&context), acknowledgement))
}

async fn list_events(
    State(state): State<AppState>,
    Extension(context): Extension<RequestContext>,
    Extension(_admin): Extension<AuthenticatedAdmin>,
    Query(query): Query<AdminAgcTrackingEventListQuery>,
) -> Result<Json<Value>, AppError> {
    module_runtime::agc_analytics::validate_agc_tracking_query(&query).map_err(|_| {
        AppError::from_status(StatusCode::BAD_REQUEST).with_message("客户端埋点查询参数无效")
    })?;
    let payload = state
        .spacetime_client()
        .list_agc_tracking_events(query)
        .await
        .map_err(map_database_error)?;
    Ok(json_success_body(Some(&context), payload))
}

fn validate_subject_and_origin(
    batch: &AgcAnalyticsBatch,
    user_id: &str,
    expected_origin: &str,
) -> Result<(), AppError> {
    // 公开地址必须来自部署配置；客户端 body 和 Host 头均不是配置来源。
    let origin = url::Url::parse(expected_origin).ok().filter(|url| {
        let loopback = url.host_str().is_some_and(|host| {
            host == "localhost"
                || host
                    .trim_matches(['[', ']'])
                    .parse::<std::net::IpAddr>()
                    .is_ok_and(|ip| ip.is_loopback())
        });
        (url.scheme() == "https" || (url.scheme() == "http" && loopback))
            && url.host_str().is_some()
            && url.username().is_empty()
            && url.password().is_none()
            && url.origin().ascii_serialization() == expected_origin
    });
    if origin.is_none() {
        return Err(AppError::from_status(StatusCode::SERVICE_UNAVAILABLE)
            .with_message("客户端埋点接收地址尚未配置"));
    }
    if batch.user_id != user_id
        || batch.destination_origin != expected_origin
        || batch
            .events
            .iter()
            .any(|event| event.user_id.as_deref() != Some(user_id))
    {
        return Err(
            AppError::from_status(StatusCode::FORBIDDEN).with_message("客户端埋点身份或平台不匹配")
        );
    }
    Ok(())
}

fn map_database_error(error: SpacetimeClientError) -> AppError {
    match error {
        SpacetimeClientError::Procedure(message) if message == "agc_event_conflict" => {
            AppError::from_status(StatusCode::CONFLICT).with_message("客户端埋点事件 ID 内容冲突")
        }
        SpacetimeClientError::Procedure(message)
            if matches!(message.as_str(), "invalid_agc_query" | "invalid_agc_cursor") =>
        {
            AppError::from_status(StatusCode::BAD_REQUEST).with_message("客户端埋点查询参数无效")
        }
        // 不回显数据库或上传内容，超时/未知提交结果不产生确认。
        _ => AppError::from_status(StatusCode::SERVICE_UNAVAILABLE)
            .with_message("客户端埋点数据服务暂不可用"),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn batch() -> AgcAnalyticsBatch {
        AgcAnalyticsBatch {
            schema_version: 1,
            batch_id: uuid::Uuid::new_v4().to_string(),
            destination_origin: "https://dev.genarrative.world".into(),
            user_id: "user-a".into(),
            events: vec![],
        }
    }

    #[test]
    fn agc_analytics_subject_and_deployment_are_both_required() {
        let batch = batch();
        assert!(validate_subject_and_origin(&batch, "user-a", &batch.destination_origin).is_ok());
        for (user, origin, expected) in [
            (
                "user-b",
                "https://dev.genarrative.world",
                StatusCode::FORBIDDEN,
            ),
            (
                "user-a",
                "https://www.genarrative.world",
                StatusCode::FORBIDDEN,
            ),
            ("user-a", "", StatusCode::SERVICE_UNAVAILABLE),
            (
                "user-a",
                "https://dev.genarrative.world/path",
                StatusCode::SERVICE_UNAVAILABLE,
            ),
            (
                "user-a",
                "http://dev.genarrative.world",
                StatusCode::SERVICE_UNAVAILABLE,
            ),
        ] {
            assert_eq!(
                validate_subject_and_origin(&batch, user, origin)
                    .unwrap_err()
                    .status_code(),
                expected
            );
        }
        let mut local = batch;
        local.destination_origin = "http://127.0.0.1:8082".into();
        assert!(validate_subject_and_origin(&local, "user-a", &local.destination_origin).is_ok());
    }

    #[test]
    fn agc_analytics_database_failure_never_becomes_acknowledgement() {
        for (message, status) in [
            ("agc_event_conflict", StatusCode::CONFLICT),
            ("invalid_agc_cursor", StatusCode::BAD_REQUEST),
            ("unknown", StatusCode::SERVICE_UNAVAILABLE),
        ] {
            assert_eq!(
                map_database_error(SpacetimeClientError::Procedure(message.into())).status_code(),
                status
            );
        }
    }
}