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

//! AGC 项目定时快照上传的服务端入口。
//!
//! 客户端只提交"某个项目里发生变化的文件内容"和"当前清单"，对象键、bucket 与
//! 存储凭据都由服务端决定。写入固定在内部前缀下，客户端直传票据不覆盖该前缀。

use crate::{
    api_response::json_success_body, auth::AuthenticatedAccessToken, http_error::AppError,
    request_context::RequestContext, state::AppState,
};
use axum::{
    Json,
    body::Bytes,
    extract::{Extension, Query, State},
    http::StatusCode,
};
use platform_oss::{
    OssDeleteObjectRequest, OssGetObjectRequest, OssInternalPutObjectRequest, OssObjectAccess,
    agc_project_snapshot_file_object_key, agc_project_snapshot_manifest_object_key,
};
use serde_json::Value;
use shared_contracts::agc_project_snapshots::{
    AGC_PROJECT_SNAPSHOT_MAX_FILE_BYTES, AGC_PROJECT_SNAPSHOT_MAX_MANIFEST_FILES,
    AGC_PROJECT_SNAPSHOT_MAX_PROJECT_BYTES, AGC_PROJECT_SNAPSHOT_SCHEMA_VERSION,
    AgcProjectSnapshotFileUploadQuery, AgcProjectSnapshotFileUploadResponse,
    AgcProjectSnapshotManifestRequest, AgcProjectSnapshotManifestResponse,
    agc_project_snapshot_checksum, validate_agc_project_snapshot_checksum,
    validate_agc_project_snapshot_project_id, validate_agc_project_snapshot_relative_path,
};
use std::{
    collections::{HashMap, HashSet},
    sync::{Mutex, OnceLock},
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use tracing::{debug, info, warn};

/// 单文件请求体上限比文件上限留一点余量，超限请求由 body limit 直接拒绝。
pub(crate) const MAX_FILE_REQUEST_BODY_BYTES: usize =
    AGC_PROJECT_SNAPSHOT_MAX_FILE_BYTES as usize + 1024;
/// 清单请求体上限：条目数本身有上限，这里再给一个字节级兜底。
pub(crate) const MAX_MANIFEST_REQUEST_BODY_BYTES: usize = 32 * 1024 * 1024;
/// 清单里单条路径的字段长度上限（与契约校验口径一致）。
const MAX_MANIFEST_PATH_CHARS: usize = 1024;
/// 单个用户每小时允许的文件上传次数。进程内计数，用于抑制异常客户端，不是计费级配额。
const MAX_FILE_UPLOADS_PER_USER_PER_HOUR: usize = 3_000;
/// 单个用户每小时允许的清单写入次数。
const MAX_MANIFEST_UPLOADS_PER_USER_PER_HOUR: usize = 120;
/// 同一项目两次清单写入的最小间隔，避免异常客户端高频覆盖清单。
const MIN_MANIFEST_INTERVAL_MS: u64 = 5_000;
/// 单次清单写入最多回收多少个不再被引用的对象，避免一次请求做过量删除。
const MAX_OBJECTS_RECLAIMED_PER_MANIFEST: usize = 2_000;

/// 写入一次增量同步里的单个文件。
///
/// 对象键由字节数和内容摘要共同决定，因此"对象已存在且长度一致"就是内容已存在的
/// 充分判据；探测失败按未存在处理，PUT 本身是幂等的。
pub async fn upload_project_snapshot_file(
    State(state): State<AppState>,
    Extension(ctx): Extension<RequestContext>,
    Extension(auth): Extension<AuthenticatedAccessToken>,
    Query(query): Query<AgcProjectSnapshotFileUploadQuery>,
    body: Bytes,
) -> Result<Json<Value>, AppError> {
    consume_user_upload_quota(auth.claims().user_id(), ProjectSnapshotUploadKind::File)?;
    let size_bytes = validate_uploaded_file(&query, &body)?;
    let digest = query
        .checksum
        .strip_prefix("fnv1a64:")
        .unwrap_or_default()
        .to_string();
    let object_key = agc_project_snapshot_file_object_key(
        auth.claims().user_id(),
        &query.project_id,
        size_bytes,
        &digest,
        &query.relative_path,
    )
    .map_err(|error| bad_request(error.to_string()))?;

    let oss = project_snapshot_oss(&state)?;
    let skipped = match oss
        .head_internal_object(state.editor_oss_http_client(), &object_key)
        .await
    {
        Ok(Some(existing)) => existing.content_length == size_bytes,
        // 确定不存在、或探测失败时都继续写入：PUT 幂等，宁可多传一次也不漏传。
        Ok(None) | Err(_) => false,
    };
    if !skipped {
        oss.put_internal_object(
            state.editor_oss_http_client(),
            OssInternalPutObjectRequest {
                object_key: object_key.clone(),
                content_type: Some("application/octet-stream".to_string()),
                access: OssObjectAccess::Private,
                metadata: Default::default(),
                body: body.to_vec(),
            },
        )
        .await
        .map_err(|_| {
            AppError::from_status(StatusCode::BAD_GATEWAY).with_message("项目快照文件上传失败")
        })?;
    }

    Ok(json_success_body(
        Some(&ctx),
        AgcProjectSnapshotFileUploadResponse {
            project_id: query.project_id,
            relative_path: query.relative_path,
            object_key,
            skipped,
            checksum: query.checksum,
            size_bytes,
        },
    ))
}

fn validate_uploaded_file(
    query: &AgcProjectSnapshotFileUploadQuery,
    body: &[u8],
) -> Result<u64, AppError> {
    validate_agc_project_snapshot_project_id(&query.project_id).map_err(bad_request)?;
    validate_agc_project_snapshot_relative_path(&query.relative_path).map_err(bad_request)?;
    validate_agc_project_snapshot_checksum(&query.checksum).map_err(bad_request)?;
    let size_bytes =
        u64::try_from(body.len()).map_err(|_| bad_request("项目快照文件长度超出可支持范围"))?;
    if size_bytes > AGC_PROJECT_SNAPSHOT_MAX_FILE_BYTES {
        return Err(bad_request("项目快照文件超过单文件上限"));
    }
    if size_bytes != query.size_bytes {
        return Err(bad_request("项目快照文件长度与声明不一致"));
    }
    if !agc_project_snapshot_checksum(body).eq_ignore_ascii_case(&query.checksum) {
        return Err(bad_request("项目快照文件内容与声明摘要不一致"));
    }
    Ok(size_bytes)
}

/// 覆盖写入该项目的远端清单。删除文件只在这里消失，本期不删除远端对象。
pub async fn upload_project_snapshot_manifest(
    State(state): State<AppState>,
    Extension(ctx): Extension<RequestContext>,
    Extension(auth): Extension<AuthenticatedAccessToken>,
    Json(payload): Json<AgcProjectSnapshotManifestRequest>,
) -> Result<Json<Value>, AppError> {
    consume_user_upload_quota(auth.claims().user_id(), ProjectSnapshotUploadKind::Manifest)?;
    validate_manifest(&payload)?;
    let user_id = auth.claims().user_id().to_string();
    let object_key = agc_project_snapshot_manifest_object_key(&user_id, &payload.project_id)
        .map_err(|error| bad_request(error.to_string()))?;
    // 上一版清单同时承担两个职责：项目级写入频率闸门，以及本轮远端对象回收的引用基线。
    // 读不到或解析失败时只跳过回收，绝不据此删除任何对象。
    let previous = read_project_snapshot_manifest(&state, &object_key).await;
    if let Some(previous) = previous.as_ref()
        && unix_millis_now().saturating_sub(previous.synced_at_ms) < MIN_MANIFEST_INTERVAL_MS
    {
        return Err(AppError::from_status(StatusCode::TOO_MANY_REQUESTS)
            .with_message("同一项目的项目快照写入过于频繁，请稍后重试")
            .with_header("retry-after", axum::http::HeaderValue::from_static("5")));
    }
    let body = serde_json::to_vec(&payload)
        .map_err(|error| internal(format!("序列化项目快照清单失败：{error}")))?;
    let total_bytes = payload
        .files
        .iter()
        .fold(0_u64, |total, file| total.saturating_add(file.size_bytes));
    let oss = project_snapshot_oss(&state)?;
    oss.put_internal_object(
        state.editor_oss_http_client(),
        OssInternalPutObjectRequest {
            object_key: object_key.clone(),
            content_type: Some("application/json".to_string()),
            access: OssObjectAccess::Private,
            metadata: Default::default(),
            body,
        },
    )
    .await
    .map_err(|_| {
        AppError::from_status(StatusCode::BAD_GATEWAY).with_message("项目快照清单上传失败")
    })?;
    // 清单写入成功之后再回收：任何时刻远端对象集合都是当前清单的超集，
    // 不会出现清单引用了刚被删掉的对象。
    reclaim_unreferenced_objects(&state, oss, &user_id, previous.as_ref(), &payload).await;

    Ok(json_success_body(
        Some(&ctx),
        AgcProjectSnapshotManifestResponse {
            project_id: payload.project_id,
            sync_revision: payload.sync_revision,
            object_key,
            file_count: u32::try_from(payload.files.len()).unwrap_or(u32::MAX),
            total_bytes,
        },
    ))
}

/// 单个用户在窗口内的上传计数。进程内计数，用于抑制异常客户端与失控重试；
/// 跨节点配额由"单项目累计体积上限 + 清单引用回收"保证。
#[derive(Clone, Copy, Default)]
struct ProjectSnapshotUserWindow {
    started_at: Option<Instant>,
    files: usize,
    manifests: usize,
}

#[derive(Clone, Copy)]
enum ProjectSnapshotUploadKind {
    File,
    Manifest,
}

static PROJECT_SNAPSHOT_USER_WINDOWS: OnceLock<Mutex<HashMap<String, ProjectSnapshotUserWindow>>> =
    OnceLock::new();

fn consume_user_upload_quota(
    user_id: &str,
    kind: ProjectSnapshotUploadKind,
) -> Result<(), AppError> {
    let windows = PROJECT_SNAPSHOT_USER_WINDOWS.get_or_init(|| Mutex::new(HashMap::new()));
    let mut windows = windows
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    let window = windows.entry(user_id.to_string()).or_default();
    let now = Instant::now();
    let expired = window
        .started_at
        .is_none_or(|started_at| now.duration_since(started_at) >= Duration::from_secs(3_600));
    if expired {
        *window = ProjectSnapshotUserWindow {
            started_at: Some(now),
            ..ProjectSnapshotUserWindow::default()
        };
    }
    let (counter, limit) = match kind {
        ProjectSnapshotUploadKind::File => (&mut window.files, MAX_FILE_UPLOADS_PER_USER_PER_HOUR),
        ProjectSnapshotUploadKind::Manifest => (
            &mut window.manifests,
            MAX_MANIFEST_UPLOADS_PER_USER_PER_HOUR,
        ),
    };
    if *counter >= limit {
        return Err(AppError::from_status(StatusCode::TOO_MANY_REQUESTS)
            .with_message("项目快照上传次数超出当前小时上限，请稍后重试")
            .with_header("retry-after", axum::http::HeaderValue::from_static("60")));
    }
    *counter += 1;
    Ok(())
}

fn unix_millis_now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
        .unwrap_or(0)
}

/// 读取该项目上一版清单。只在对象不存在时返回 `None`；读取或解析失败同样返回
/// `None`，调用方据此跳过回收（fail-closed：不确定就不删）。
async fn read_project_snapshot_manifest(
    state: &AppState,
    object_key: &str,
) -> Option<AgcProjectSnapshotManifestRequest> {
    let oss = project_snapshot_oss(state).ok()?;
    match oss
        .get_object(
            state.editor_oss_http_client(),
            OssGetObjectRequest {
                object_key: object_key.to_string(),
                max_bytes: MAX_MANIFEST_REQUEST_BODY_BYTES,
            },
        )
        .await
    {
        Ok(bytes) => serde_json::from_slice::<AgcProjectSnapshotManifestRequest>(&bytes)
            .map_err(|error| {
                warn!(object_key = %object_key, error = %error, "项目快照上一版清单解析失败，本轮跳过回收");
                error
            })
            .ok(),
        Err(error) => {
            debug!(object_key = %object_key, error = %error, "项目快照上一版清单不可读，本轮跳过回收");
            None
        }
    }
}

/// 回收不再被当前清单引用、且确实由上一版清单登记过的对象。
///
/// 只按上一版清单里的 `(路径, 字节数, 摘要)` 反推出确定的对象键，不做 LIST，
/// 因此不可能误删其它项目或其它功能的对象。任何单个删除失败都只记日志：
/// 清单已经写入成功，回收失败不影响本次同步的语义，下一次写入会再试。
async fn reclaim_unreferenced_objects(
    state: &AppState,
    oss: &platform_oss::OssClient,
    user_id: &str,
    previous: Option<&AgcProjectSnapshotManifestRequest>,
    next: &AgcProjectSnapshotManifestRequest,
) {
    let Some(previous) = previous else {
        return;
    };
    let retained = next
        .files
        .iter()
        .map(|file| {
            (
                file.relative_path.as_str(),
                file.size_bytes,
                file.checksum.as_str(),
            )
        })
        .collect::<HashSet<_>>();
    let mut reclaimed = 0_usize;
    let mut skipped = 0_usize;
    for file in &previous.files {
        if retained.contains(&(
            file.relative_path.as_str(),
            file.size_bytes,
            file.checksum.as_str(),
        )) {
            continue;
        }
        if reclaimed >= MAX_OBJECTS_RECLAIMED_PER_MANIFEST {
            // 剩下的留给下一次清单写入继续回收，不在这里无限删除。
            skipped += 1;
            continue;
        }
        let Some(digest) = file.checksum.strip_prefix("fnv1a64:") else {
            continue;
        };
        let Ok(object_key) = agc_project_snapshot_file_object_key(
            user_id,
            &previous.project_id,
            file.size_bytes,
            digest,
            &file.relative_path,
        ) else {
            continue;
        };
        match oss
            .delete_object(
                state.editor_oss_http_client(),
                OssDeleteObjectRequest {
                    object_key: object_key.clone(),
                },
            )
            .await
        {
            Ok(()) => reclaimed += 1,
            Err(error) => {
                warn!(object_key = %object_key, error = %error, "项目快照旧版本对象回收失败");
            }
        }
    }
    if reclaimed > 0 || skipped > 0 {
        info!(
            project_id = %previous.project_id,
            reclaimed,
            skipped,
            "项目快照旧版本对象回收完成"
        );
    }
}

pub(crate) fn validate_manifest(
    payload: &AgcProjectSnapshotManifestRequest,
) -> Result<(), AppError> {
    if payload.schema_version != AGC_PROJECT_SNAPSHOT_SCHEMA_VERSION {
        return Err(bad_request("项目快照清单版本不受支持"));
    }
    validate_agc_project_snapshot_project_id(&payload.project_id).map_err(bad_request)?;
    if payload.synced_at_ms == 0 {
        return Err(bad_request("项目快照清单缺少同步时刻"));
    }
    if payload.project_name.as_ref().is_some_and(|name| {
        name.trim().is_empty() || name.len() > 512 || name.chars().any(char::is_control)
    }) {
        return Err(bad_request(
            "项目快照名称必须是 1 到 512 字节且不含控制字符",
        ));
    }
    if payload.files.len() > AGC_PROJECT_SNAPSHOT_MAX_MANIFEST_FILES {
        return Err(bad_request("项目快照清单文件数量超过上限"));
    }
    let mut seen = HashSet::with_capacity(payload.files.len());
    let mut paths = HashMap::new();
    let mut total_bytes = 0_u64;
    for file in &payload.files {
        if file.relative_path.chars().count() > MAX_MANIFEST_PATH_CHARS {
            return Err(bad_request("项目快照清单路径过长"));
        }
        validate_agc_project_snapshot_relative_path(&file.relative_path).map_err(bad_request)?;
        validate_agc_project_snapshot_checksum(&file.checksum).map_err(bad_request)?;
        if file.size_bytes > AGC_PROJECT_SNAPSHOT_MAX_FILE_BYTES {
            return Err(bad_request("项目快照清单包含超过单文件上限的条目"));
        }
        if !seen.insert(file.relative_path.as_str()) {
            return Err(bad_request("项目快照清单包含重复路径"));
        }
        let mut path = String::new();
        let components = file.relative_path.split('/').collect::<Vec<_>>();
        for (index, component) in components.iter().enumerate() {
            if index > 0 {
                path.push('/');
            }
            path.push_str(component);
            let is_file = index + 1 == components.len();
            let identity = path.to_lowercase();
            if let Some((original, previous_is_file)) = paths.get(&identity) {
                if original != &path || *previous_is_file || is_file {
                    return Err(bad_request("项目快照清单包含大小写或文件目录冲突"));
                }
            } else {
                paths.insert(identity, (path.clone(), is_file));
            }
        }
        total_bytes = total_bytes.saturating_add(file.size_bytes);
    }
    // 清单描述的是项目当前全量文件，所以这里的累计体积就是该项目的常驻占用；
    // 单项目上限同时约束了远端对象回收之后的实际存储量。
    if total_bytes > AGC_PROJECT_SNAPSHOT_MAX_PROJECT_BYTES {
        return Err(AppError::from_status(StatusCode::PAYLOAD_TOO_LARGE)
            .with_message("项目快照累计体积超过单项目上限"));
    }
    Ok(())
}

pub(crate) fn project_snapshot_oss(state: &AppState) -> Result<&platform_oss::OssClient, AppError> {
    state.project_snapshot_oss_client().ok_or_else(|| {
        AppError::from_status(StatusCode::SERVICE_UNAVAILABLE)
            .with_message("AGC 项目快照 OSS 未配置")
    })
}

fn bad_request(m: impl Into<String>) -> AppError {
    AppError::from_status(StatusCode::BAD_REQUEST).with_message(m)
}

fn internal<E: ToString>(e: E) -> AppError {
    AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_message(e.to_string())
}

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

    #[test]
    fn project_snapshots_upload_accepts_empty_files_and_rejects_false_checksums() {
        let mut query = AgcProjectSnapshotFileUploadQuery {
            project_id: "project-1".into(),
            relative_path: "game/empty.txt".into(),
            checksum: "fnv1a64:cbf29ce484222325".into(),
            size_bytes: 0,
        };
        assert_eq!(validate_uploaded_file(&query, b"").unwrap(), 0);
        assert!(validate_uploaded_file(&query, b"x").is_err());
        query.size_bytes = 1;
        assert!(validate_uploaded_file(&query, b"x").is_err());
        query.checksum = agc_project_snapshot_checksum(b"x");
        assert_eq!(validate_uploaded_file(&query, b"x").unwrap(), 1);
    }

    fn manifest_file(relative_path: &str, size_bytes: u64) -> AgcProjectSnapshotManifestFile {
        AgcProjectSnapshotManifestFile {
            relative_path: relative_path.to_string(),
            size_bytes,
            checksum: "fnv1a64:0123456789abcdef".to_string(),
        }
    }

    fn manifest(files: Vec<AgcProjectSnapshotManifestFile>) -> AgcProjectSnapshotManifestRequest {
        AgcProjectSnapshotManifestRequest {
            schema_version: AGC_PROJECT_SNAPSHOT_SCHEMA_VERSION,
            project_id: "gameagent-1a2b3c4d".to_string(),
            sync_revision: 1,
            synced_at_ms: 1_700_000_000_000,
            files,
            project_name: None,
            pending_files: None,
        }
    }

    #[test]
    fn project_snapshot_manifest_accepts_a_bounded_payload() {
        assert!(
            validate_manifest(&manifest(vec![
                manifest_file("game/index.html", 1024),
                manifest_file("assets/Hero.png", 2048),
            ]))
            .is_ok()
        );
    }

    #[test]
    fn project_snapshot_manifest_rejects_duplicates_and_out_of_range_values() {
        assert!(
            validate_manifest(&manifest(vec![
                manifest_file("game/index.html", 1),
                manifest_file("game/index.html", 1),
            ]))
            .is_err(),
            "重复路径会让远端清单产生歧义"
        );

        let mut traversal = manifest(vec![manifest_file("game/index.html", 1)]);
        traversal.files[0].relative_path = "../outside.txt".to_string();
        assert!(validate_manifest(&traversal).is_err());

        let mut unsafe_project = manifest(vec![manifest_file("game/index.html", 1)]);
        unsafe_project.project_id = "../escape".to_string();
        assert!(validate_manifest(&unsafe_project).is_err());

        let mut bad_checksum = manifest(vec![manifest_file("game/index.html", 1)]);
        bad_checksum.files[0].checksum = "md5:not-hex".to_string();
        assert!(validate_manifest(&bad_checksum).is_err());

        let mut oversized = manifest(vec![manifest_file("game/index.html", 1)]);
        oversized.files[0].size_bytes = AGC_PROJECT_SNAPSHOT_MAX_FILE_BYTES + 1;
        assert!(validate_manifest(&oversized).is_err());

        let mut stale_schema = manifest(vec![manifest_file("game/index.html", 1)]);
        stale_schema.schema_version = AGC_PROJECT_SNAPSHOT_SCHEMA_VERSION + 1;
        assert!(validate_manifest(&stale_schema).is_err());

        let mut missing_timestamp = manifest(vec![manifest_file("game/index.html", 1)]);
        missing_timestamp.synced_at_ms = 0;
        assert!(validate_manifest(&missing_timestamp).is_err());
    }

    #[test]
    fn project_snapshot_manifest_rejects_too_many_entries() {
        let files = (0..=AGC_PROJECT_SNAPSHOT_MAX_MANIFEST_FILES)
            .map(|index| manifest_file(&format!("game/file-{index}.txt"), 1))
            .collect::<Vec<_>>();
        assert!(validate_manifest(&manifest(files)).is_err());
    }

    #[test]
    fn project_snapshot_manifest_rejects_a_project_over_the_total_cap() {
        // 单文件与条目数都在上限内，只有项目累计体积越界：这必须按 413 拒绝，
        // 否则远端回收之后仍会长期占住超大配额。
        let files = (0..40)
            .map(|index| {
                manifest_file(
                    &format!("assets/blob-{index}.bin"),
                    AGC_PROJECT_SNAPSHOT_MAX_FILE_BYTES,
                )
            })
            .collect::<Vec<_>>();
        let error = validate_manifest(&manifest(files)).expect_err("超额项目必须被拒绝");
        assert_eq!(error.status_code(), StatusCode::PAYLOAD_TOO_LARGE);
    }

    #[test]
    fn project_snapshot_user_quota_stops_after_the_hourly_file_limit() {
        let user = "user-quota-fixture";
        for _ in 0..MAX_FILE_UPLOADS_PER_USER_PER_HOUR {
            consume_user_upload_quota(user, ProjectSnapshotUploadKind::File)
                .expect("配额内的上传必须放行");
        }
        let error = consume_user_upload_quota(user, ProjectSnapshotUploadKind::File)
            .expect_err("超出小时上限必须被拒绝");
        assert_eq!(error.status_code(), StatusCode::TOO_MANY_REQUESTS);
        // 清单与文件是两条独立配额，不受文件配额影响。
        consume_user_upload_quota(user, ProjectSnapshotUploadKind::Manifest)
            .expect("清单配额独立计数");
    }
}