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},
    config::AppConfig,
    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)?;
    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,
    config: &AppConfig,
) -> Result<(), AppError> {
    // 复用部署渠道，不使用客户端 body 或 Host / Forwarded 头推断服务器所属平台。
    let expected_origin = match config.client_download_channel.as_str() {
        "dev" => "https://dev.genarrative.world",
        "release" => "https://www.genarrative.world",
        _ => {
            return Err(AppError::from_status(StatusCode::SERVICE_UNAVAILABLE)
                .with_message("客户端埋点接收渠道无效"));
        }
    };
    // 本地开发允许回环地址及可变端口，兼容端口漂移和容器映射；线上部署不启用此例外。
    let local_origin = config.client_download_channel == "dev"
        && matches!(
            config.environment.as_str(),
            "development" | "test" | "container"
        )
        && url::Url::parse(&batch.destination_origin).is_ok_and(|url| {
            matches!(url.scheme(), "http" | "https")
                && url.host_str().is_some_and(|host| {
                    host == "localhost"
                        || host
                            .trim_matches(['[', ']'])
                            .parse::<std::net::IpAddr>()
                            .is_ok_and(|ip| ip.is_loopback())
                })
                && url.username().is_empty()
                && url.password().is_none()
                && url.origin().ascii_serialization() == batch.destination_origin
        });
    if batch.user_id != user_id
        || (batch.destination_origin != expected_origin && !local_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() {
        for (channel, origin, other_origin) in [
            (
                "dev",
                "https://dev.genarrative.world",
                "https://www.genarrative.world",
            ),
            (
                "release",
                "https://www.genarrative.world",
                "https://dev.genarrative.world",
            ),
        ] {
            let config = AppConfig {
                environment: "production".into(),
                client_download_channel: channel.into(),
                ..AppConfig::default()
            };
            let mut batch = batch();
            batch.destination_origin = origin.into();
            assert!(validate_subject_and_origin(&batch, "user-a", &config).is_ok());
            assert_eq!(
                validate_subject_and_origin(&batch, "user-b", &config)
                    .unwrap_err()
                    .status_code(),
                StatusCode::FORBIDDEN
            );
            for rejected in [
                other_origin,
                "http://127.0.0.1:8082",
                "https://custom.example",
            ] {
                batch.destination_origin = rejected.into();
                assert_eq!(
                    validate_subject_and_origin(&batch, "user-a", &config)
                        .unwrap_err()
                        .status_code(),
                    StatusCode::FORBIDDEN,
                    "{channel}: {rejected}"
                );
            }
        }
        for channel in ["", "qa-2026", "dev-win"] {
            let config = AppConfig {
                client_download_channel: channel.into(),
                ..AppConfig::default()
            };
            assert_eq!(
                validate_subject_and_origin(&batch(), "user-a", &config)
                    .unwrap_err()
                    .status_code(),
                StatusCode::SERVICE_UNAVAILABLE
            );
        }
    }

    #[test]
    fn agc_analytics_loopback_supports_local_ports_only_in_dev_environments() {
        let mut batch = batch();
        for environment in ["development", "test", "container", "production", "staging"] {
            for channel in ["dev", "release"] {
                let config = AppConfig {
                    environment: environment.into(),
                    client_download_channel: channel.into(),
                    ..AppConfig::default()
                };
                let allowed =
                    channel == "dev" && matches!(environment, "development" | "test" | "container");
                for origin in [
                    "http://127.0.0.1:8082",
                    "http://localhost:18080",
                    "http://[::1]:19001",
                    "https://localhost:8443",
                ] {
                    batch.destination_origin = origin.into();
                    assert_eq!(
                        validate_subject_and_origin(&batch, "user-a", &config).is_ok(),
                        allowed,
                        "{environment}/{channel}: {origin}"
                    );
                }
            }
        }
    }

    #[test]
    fn agc_analytics_local_exception_rejects_noncanonical_and_nonloopback_origins() {
        let config = AppConfig::default();
        let mut batch = batch();
        for origin in [
            "http://192.168.1.2:8082",
            "http://localhost.example:8082",
            "http://0.0.0.0:8082",
            "http://localhost:8082/",
            "http://localhost:8082/api",
            "http://localhost:8082?x=1",
            "http://localhost:8082#fragment",
            "http://user:password@localhost:8082",
            "ftp://localhost:8082",
            "https://dev.genarrative.world/",
            "http://dev.genarrative.world",
            "",
        ] {
            batch.destination_origin = origin.into();
            assert_eq!(
                validate_subject_and_origin(&batch, "user-a", &config)
                    .unwrap_err()
                    .status_code(),
                StatusCode::FORBIDDEN,
                "{origin}"
            );
        }
    }

    #[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
            );
        }
    }
}