kdletters
4951b71d71
Project CI / AI game creator shell Rust crates (push) Successful in 2m46s
Project CI / AI game creator shell Rust smoke (push) Successful in 3m37s
Project CI / AI game creator shell Rust lane 2/2 (push) Failing after 6m5s
Project CI / AI game creator shell Rust lane 1/2 (push) Failing after 6m40s
Project CI / Repository checks (push) Successful in 5m37s
Project CI / Frontend tests (push) Successful in 6m21s
Project CI / Backend tests (push) Successful in 9m56s
Project CI / Native shell tests (push) Successful in 10m57s
Project CI / AI game creator shell web tests (push) Successful in 5m29s
项目快照对象键升级为 agc/project-snapshots/v2/{channel}/{user}/{project}/，渠道取部署配置 GENARRATIVE_AGC_PROJECT_SNAPSHOT_CHANNEL（缺省沿用客户端下载渠道），非法渠道在写入处失败关闭
后台新增渠道列表接口，项目工程列表与下载接受 channel，游标绑定渠道并拒绝跨渠道复用
后台项目工程用户列改为素材查询口径：昵称 + 陶泥号 + 用户详情入口，由 api-server 解析作者信息
后台项目工程改为游标分页：每页 20/50/100、上一页/下一页与当前页提示，翻页失败保留当前页
部署环境示例补充快照渠道配置，并同步运维、技术方案、里程碑与决策记录文档
1083 lines
40 KiB
Rust
1083 lines
40 KiB
Rust
//! 后台按项目读取私有快照，并在完整性验证后导出原始工程目录。
|
|
|
|
use std::{
|
|
collections::BTreeSet,
|
|
future::Future,
|
|
io::Write,
|
|
sync::{Arc, OnceLock},
|
|
time::Duration,
|
|
};
|
|
|
|
use axum::{
|
|
Json,
|
|
body::{Body, Bytes},
|
|
extract::{Extension, Path, Query, State},
|
|
http::{StatusCode, header},
|
|
response::Response,
|
|
};
|
|
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
|
|
use platform_oss::{
|
|
OssClient, OssError, OssGetObjectRequest, agc_project_snapshot_file_object_key,
|
|
agc_project_snapshot_manifest_object_key, project_snapshots::SnapshotDirectoryPage,
|
|
validate_agc_project_snapshot_channel,
|
|
};
|
|
use serde::{Deserialize, Serialize};
|
|
use serde_json::Value;
|
|
use shared_contracts::{
|
|
admin::{
|
|
AdminProjectSnapshotChannelsResponse, AdminProjectSnapshotDownloadQuery,
|
|
AdminProjectSnapshotItem, AdminProjectSnapshotStatus, AdminProjectSnapshotsQuery,
|
|
AdminProjectSnapshotsResponse,
|
|
},
|
|
agc_project_snapshots::{
|
|
AgcProjectSnapshotManifestFile, AgcProjectSnapshotManifestRequest,
|
|
agc_project_snapshot_checksum, validate_agc_project_snapshot_project_id,
|
|
},
|
|
};
|
|
use tempfile::NamedTempFile;
|
|
use tokio::{
|
|
io::AsyncReadExt,
|
|
sync::{OwnedSemaphorePermit, Semaphore},
|
|
};
|
|
use zip::{ZipWriter, write::SimpleFileOptions};
|
|
|
|
use crate::{
|
|
admin::AuthenticatedAdmin,
|
|
api_response::json_success_body,
|
|
http_error::AppError,
|
|
project_snapshots::{MAX_MANIFEST_REQUEST_BODY_BYTES, project_snapshot_oss, validate_manifest},
|
|
request_context::RequestContext,
|
|
state::AppState,
|
|
work_author::resolve_work_author_by_user_id,
|
|
};
|
|
|
|
const MAX_DIRECTORY_REQUESTS: usize = 100;
|
|
const MAX_SCANNED_PROJECTS: usize = 100;
|
|
static DOWNLOAD_PERMITS: OnceLock<Arc<Semaphore>> = OnceLock::new();
|
|
|
|
/// 游标只含已验证的两层目录标识，不接受任意 OSS 前缀或对象键。
|
|
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
|
|
#[serde(deny_unknown_fields)]
|
|
struct SnapshotCursor {
|
|
channel: String,
|
|
user: Option<String>,
|
|
project: Option<String>,
|
|
finished_user: bool,
|
|
last_user: bool,
|
|
}
|
|
|
|
impl SnapshotCursor {
|
|
fn decode(value: Option<&str>, channel: &str) -> Result<Self, AppError> {
|
|
let Some(value) = value.filter(|value| !value.is_empty()) else {
|
|
return Ok(Self {
|
|
channel: channel.to_string(),
|
|
finished_user: true,
|
|
..Self::default()
|
|
});
|
|
};
|
|
if value.len() > 1024 {
|
|
return Err(bad_request("项目列表游标无效"));
|
|
}
|
|
let bytes = URL_SAFE_NO_PAD
|
|
.decode(value)
|
|
.map_err(|_| bad_request("项目列表游标无效"))?;
|
|
let cursor: Self =
|
|
serde_json::from_slice(&bytes).map_err(|_| bad_request("项目列表游标无效"))?;
|
|
// 游标只在同一渠道内有效：跨渠道复用会让目录推进落在别的渠道上。
|
|
if cursor.channel != channel {
|
|
return Err(bad_request("项目列表游标与渠道不一致"));
|
|
}
|
|
for segment in [&cursor.user, &cursor.project].into_iter().flatten() {
|
|
validate_agc_project_snapshot_project_id(segment)
|
|
.map_err(|_| bad_request("项目列表游标无效"))?;
|
|
}
|
|
if cursor.user.is_none() || (cursor.finished_user && cursor.project.is_some()) {
|
|
return Err(bad_request("项目列表游标无效"));
|
|
}
|
|
Ok(cursor)
|
|
}
|
|
|
|
fn encode(&self) -> String {
|
|
URL_SAFE_NO_PAD.encode(serde_json::to_vec(self).expect("fixed cursor serializes"))
|
|
}
|
|
}
|
|
|
|
trait SnapshotStore: Sync {
|
|
fn directories(
|
|
&self,
|
|
user: Option<&str>,
|
|
after: Option<&str>,
|
|
limit: usize,
|
|
) -> impl Future<Output = Result<SnapshotDirectoryPage, AppError>> + Send;
|
|
fn manifest(
|
|
&self,
|
|
user: &str,
|
|
project: &str,
|
|
) -> impl Future<Output = Result<Option<AgcProjectSnapshotManifestRequest>, AppError>> + Send;
|
|
fn file(
|
|
&self,
|
|
user: &str,
|
|
project: &str,
|
|
file: &AgcProjectSnapshotManifestFile,
|
|
) -> impl Future<Output = Result<Vec<u8>, AppError>> + Send;
|
|
}
|
|
|
|
struct OssSnapshotStore<'a> {
|
|
oss: &'a OssClient,
|
|
client: &'a reqwest::Client,
|
|
channel: String,
|
|
}
|
|
|
|
impl SnapshotStore for OssSnapshotStore<'_> {
|
|
async fn directories(
|
|
&self,
|
|
user: Option<&str>,
|
|
after: Option<&str>,
|
|
limit: usize,
|
|
) -> Result<SnapshotDirectoryPage, AppError> {
|
|
self.oss
|
|
.list_project_snapshot_directories(self.client, &self.channel, user, after, limit)
|
|
.await
|
|
.map_err(|_| upstream("读取项目工程目录失败"))
|
|
}
|
|
|
|
async fn manifest(
|
|
&self,
|
|
user: &str,
|
|
project: &str,
|
|
) -> Result<Option<AgcProjectSnapshotManifestRequest>, AppError> {
|
|
let object_key = agc_project_snapshot_manifest_object_key(&self.channel, user, project)
|
|
.map_err(|_| bad_request("项目工程身份无效"))?;
|
|
let bytes = match self
|
|
.oss
|
|
.get_object(
|
|
self.client,
|
|
OssGetObjectRequest {
|
|
object_key,
|
|
max_bytes: MAX_MANIFEST_REQUEST_BODY_BYTES,
|
|
},
|
|
)
|
|
.await
|
|
{
|
|
Ok(bytes) => bytes,
|
|
Err(OssError::ObjectNotFound(_)) => return Ok(None),
|
|
Err(_) => return Err(upstream("读取项目工程清单失败")),
|
|
};
|
|
let manifest: AgcProjectSnapshotManifestRequest =
|
|
serde_json::from_slice(&bytes).map_err(|_| upstream("项目工程清单格式无效"))?;
|
|
validate_manifest(&manifest).map_err(|_| upstream("项目工程清单未通过安全校验"))?;
|
|
if manifest.project_id != project {
|
|
return Err(upstream("项目工程清单身份不一致"));
|
|
}
|
|
Ok(Some(manifest))
|
|
}
|
|
|
|
async fn file(
|
|
&self,
|
|
user: &str,
|
|
project: &str,
|
|
file: &AgcProjectSnapshotManifestFile,
|
|
) -> Result<Vec<u8>, AppError> {
|
|
let digest = file.checksum.strip_prefix("fnv1a64:").unwrap_or_default();
|
|
let object_key = agc_project_snapshot_file_object_key(
|
|
&self.channel,
|
|
user,
|
|
project,
|
|
file.size_bytes,
|
|
digest,
|
|
&file.relative_path,
|
|
)
|
|
.map_err(|_| upstream("项目工程文件路径无效"))?;
|
|
self.oss
|
|
.get_object(
|
|
self.client,
|
|
OssGetObjectRequest {
|
|
object_key,
|
|
max_bytes: file.size_bytes as usize,
|
|
},
|
|
)
|
|
.await
|
|
.map_err(|_| upstream("项目工程文件缺失或读取失败，请刷新后重试"))
|
|
}
|
|
}
|
|
|
|
fn item(
|
|
channel: &str,
|
|
user: String,
|
|
manifest: &AgcProjectSnapshotManifestRequest,
|
|
) -> AdminProjectSnapshotItem {
|
|
AdminProjectSnapshotItem {
|
|
user_id: user,
|
|
project_id: manifest.project_id.clone(),
|
|
project_name: manifest
|
|
.project_name
|
|
.clone()
|
|
.unwrap_or_else(|| manifest.project_id.clone()),
|
|
sync_revision: manifest.sync_revision,
|
|
synced_at_ms: manifest.synced_at_ms,
|
|
file_count: manifest.files.len() as u32,
|
|
total_bytes: manifest.files.iter().map(|file| file.size_bytes).sum(),
|
|
status: match manifest.pending_files {
|
|
Some(0) => AdminProjectSnapshotStatus::Ready,
|
|
Some(_) => AdminProjectSnapshotStatus::Partial,
|
|
None => AdminProjectSnapshotStatus::Unverified,
|
|
},
|
|
channel: channel.to_string(),
|
|
author_display_name: None,
|
|
author_public_user_code: None,
|
|
}
|
|
}
|
|
|
|
/// 后台列表的“用户”列与素材查询同口径：昵称 + 陶泥号，账号不可解析时沿用占位作者。
|
|
/// 查账号只影响展示，失败不回滚已经读到的清单统计。
|
|
fn attach_snapshot_authors(state: &AppState, items: &mut [AdminProjectSnapshotItem]) {
|
|
for entry in items.iter_mut() {
|
|
let author = resolve_work_author_by_user_id(state, &entry.user_id, None, None);
|
|
entry.author_display_name = Some(author.display_name);
|
|
entry.author_public_user_code = author.public_user_code;
|
|
}
|
|
}
|
|
|
|
async fn list_snapshots(
|
|
store: &impl SnapshotStore,
|
|
mut cursor: SnapshotCursor,
|
|
limit: usize,
|
|
) -> Result<AdminProjectSnapshotsResponse, AppError> {
|
|
let mut items = Vec::new();
|
|
let mut directory_requests = 0;
|
|
let mut scanned = 0;
|
|
while items.len() < limit
|
|
&& directory_requests < MAX_DIRECTORY_REQUESTS
|
|
&& scanned < MAX_SCANNED_PROJECTS
|
|
{
|
|
if cursor.finished_user {
|
|
let users = store.directories(None, cursor.user.as_deref(), 1).await?;
|
|
directory_requests += 1;
|
|
let Some(user) = users.directories.into_iter().next() else {
|
|
return Ok(AdminProjectSnapshotsResponse {
|
|
items,
|
|
next_cursor: None,
|
|
});
|
|
};
|
|
cursor = SnapshotCursor {
|
|
channel: cursor.channel.clone(),
|
|
user: Some(user),
|
|
project: None,
|
|
finished_user: false,
|
|
last_user: users.next_marker.is_none(),
|
|
};
|
|
if directory_requests >= MAX_DIRECTORY_REQUESTS {
|
|
break;
|
|
}
|
|
}
|
|
let user = cursor
|
|
.user
|
|
.as_deref()
|
|
.ok_or_else(|| bad_request("项目列表游标无效"))?;
|
|
let projects = store
|
|
.directories(
|
|
Some(user),
|
|
cursor.project.as_deref(),
|
|
(limit - items.len()).min(MAX_SCANNED_PROJECTS - scanned),
|
|
)
|
|
.await?;
|
|
directory_requests += 1;
|
|
for project in projects.directories {
|
|
scanned += 1;
|
|
if let Some(manifest) = store.manifest(user, &project).await? {
|
|
items.push(item(&cursor.channel, user.to_string(), &manifest));
|
|
}
|
|
cursor.project = Some(project);
|
|
}
|
|
if let Some(next) = projects.next_marker {
|
|
cursor.project = Some(next);
|
|
} else {
|
|
if cursor.last_user {
|
|
return Ok(AdminProjectSnapshotsResponse {
|
|
items,
|
|
next_cursor: None,
|
|
});
|
|
}
|
|
cursor.project = None;
|
|
cursor.finished_user = true;
|
|
}
|
|
}
|
|
Ok(AdminProjectSnapshotsResponse {
|
|
items,
|
|
next_cursor: Some(cursor.encode()),
|
|
})
|
|
}
|
|
|
|
pub async fn admin_list_project_snapshots(
|
|
State(state): State<AppState>,
|
|
Extension(ctx): Extension<RequestContext>,
|
|
Extension(_admin): Extension<AuthenticatedAdmin>,
|
|
Query(query): Query<AdminProjectSnapshotsQuery>,
|
|
) -> Result<Json<Value>, AppError> {
|
|
let channel = requested_snapshot_channel(&state, query.channel.as_deref())?;
|
|
let cursor = SnapshotCursor::decode(query.cursor.as_deref(), &channel)?;
|
|
let store = OssSnapshotStore {
|
|
oss: project_snapshot_oss(&state)?,
|
|
client: state.editor_oss_http_client(),
|
|
channel,
|
|
};
|
|
let mut response =
|
|
list_snapshots(&store, cursor, query.limit.unwrap_or(20).clamp(1, 100)).await?;
|
|
attach_snapshot_authors(&state, &mut response.items);
|
|
Ok(json_success_body(Some(&ctx), response))
|
|
}
|
|
|
|
/// 目标渠道：显式传入时必须合法（空串按未提供处理），缺省用本部署渠道。
|
|
/// 非法渠道失败关闭，不落到别的渠道。
|
|
fn requested_snapshot_channel(
|
|
state: &AppState,
|
|
requested: Option<&str>,
|
|
) -> Result<String, AppError> {
|
|
let requested = requested.map(str::trim).filter(|value| !value.is_empty());
|
|
match requested {
|
|
Some(channel) => validate_agc_project_snapshot_channel(channel)
|
|
.map_err(|_| bad_request("项目工程渠道无效")),
|
|
None => crate::project_snapshots::project_snapshot_channel(state),
|
|
}
|
|
}
|
|
|
|
/// 后台渠道列表：本部署渠道与远端已存在渠道的并集。
|
|
/// 远端目录里不符合渠道命名的条目直接跳过，不作为可查询渠道暴露。
|
|
pub async fn admin_list_project_snapshot_channels(
|
|
State(state): State<AppState>,
|
|
Extension(ctx): Extension<RequestContext>,
|
|
Extension(_admin): Extension<AuthenticatedAdmin>,
|
|
) -> Result<Json<Value>, AppError> {
|
|
let default_channel = crate::project_snapshots::project_snapshot_channel(&state)?;
|
|
let oss = project_snapshot_oss(&state)?;
|
|
let client = state.editor_oss_http_client();
|
|
let mut channels = BTreeSet::from([default_channel.clone()]);
|
|
let mut after: Option<String> = None;
|
|
for _ in 0..MAX_DIRECTORY_REQUESTS {
|
|
let page = oss
|
|
.list_project_snapshot_channels(client, after.as_deref(), 100)
|
|
.await
|
|
.map_err(|_| upstream("读取项目工程渠道失败"))?;
|
|
channels.extend(
|
|
page.directories
|
|
.into_iter()
|
|
.filter(|name| validate_agc_project_snapshot_channel(name).is_ok()),
|
|
);
|
|
match page.next_marker {
|
|
Some(next) => after = Some(next),
|
|
None => break,
|
|
}
|
|
}
|
|
Ok(json_success_body(
|
|
Some(&ctx),
|
|
AdminProjectSnapshotChannelsResponse {
|
|
default_channel,
|
|
channels: channels.into_iter().collect(),
|
|
},
|
|
))
|
|
}
|
|
|
|
fn verify_file(file: &AgcProjectSnapshotManifestFile, bytes: &[u8]) -> Result<(), AppError> {
|
|
if bytes.len() as u64 != file.size_bytes
|
|
|| !agc_project_snapshot_checksum(bytes).eq_ignore_ascii_case(&file.checksum)
|
|
{
|
|
return Err(upstream("项目工程文件长度或摘要不一致，无法导出"));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn archive_blocking<T: Send + 'static>(
|
|
permit: Option<Arc<OwnedSemaphorePermit>>,
|
|
work: impl FnOnce() -> Result<T, AppError> + Send + 'static,
|
|
) -> Result<T, AppError> {
|
|
tokio::task::spawn_blocking(move || {
|
|
// 取消等待不会停止 blocking 线程；额度必须由实际工作持有到退出。
|
|
let _permit = permit;
|
|
work()
|
|
})
|
|
.await
|
|
.map_err(|_| internal())?
|
|
}
|
|
|
|
async fn build_archive_guarded(
|
|
store: &impl SnapshotStore,
|
|
user: &str,
|
|
manifest: &AgcProjectSnapshotManifestRequest,
|
|
permit: Option<Arc<OwnedSemaphorePermit>>,
|
|
) -> Result<NamedTempFile, AppError> {
|
|
let tempfile = archive_blocking(permit.clone(), || {
|
|
tempfile::Builder::new()
|
|
.prefix("agc-project-")
|
|
.suffix(".zip")
|
|
.tempfile()
|
|
.map_err(|_| internal())
|
|
})
|
|
.await?;
|
|
write_archive(store, user, manifest, tempfile, permit).await
|
|
}
|
|
|
|
#[cfg(test)]
|
|
async fn build_archive(
|
|
store: &impl SnapshotStore,
|
|
user: &str,
|
|
manifest: &AgcProjectSnapshotManifestRequest,
|
|
) -> Result<NamedTempFile, AppError> {
|
|
build_archive_guarded(store, user, manifest, None).await
|
|
}
|
|
|
|
async fn write_archive(
|
|
store: &impl SnapshotStore,
|
|
user: &str,
|
|
manifest: &AgcProjectSnapshotManifestRequest,
|
|
tempfile: NamedTempFile,
|
|
permit: Option<Arc<OwnedSemaphorePermit>>,
|
|
) -> Result<NamedTempFile, AppError> {
|
|
validate_manifest(manifest)?;
|
|
if manifest.pending_files.is_some_and(|pending| pending > 0) {
|
|
return Err(AppError::from_status(StatusCode::CONFLICT)
|
|
.with_message("项目尚未完整上传，请等待客户端同步完成"));
|
|
}
|
|
// ZipWriter 直接拥有临时文件；任何 await 被取消或后台写入失败都会释放该文件。
|
|
let mut archive = ZipWriter::new(tempfile);
|
|
for file in &manifest.files {
|
|
let bytes = store.file(user, &manifest.project_id, file).await?;
|
|
verify_file(file, &bytes)?;
|
|
let path = file.relative_path.clone();
|
|
archive = archive_blocking(permit.clone(), move || {
|
|
archive
|
|
.start_file(
|
|
path,
|
|
SimpleFileOptions::default()
|
|
.compression_method(zip::CompressionMethod::Deflated)
|
|
.unix_permissions(0o644),
|
|
)
|
|
.map_err(|_| internal())?;
|
|
archive.write_all(&bytes).map_err(|_| internal())?;
|
|
Ok::<_, AppError>(archive)
|
|
})
|
|
.await?;
|
|
}
|
|
archive_blocking(permit, move || archive.finish().map_err(|_| internal())).await
|
|
}
|
|
|
|
fn archive_filename(manifest: &AgcProjectSnapshotManifestRequest) -> String {
|
|
let name = manifest
|
|
.project_name
|
|
.as_deref()
|
|
.unwrap_or(&manifest.project_id);
|
|
let safe: String = name
|
|
.chars()
|
|
.map(|c| {
|
|
if c.is_control() || "<>:\"/\\|?*".contains(c) {
|
|
'_'
|
|
} else {
|
|
c
|
|
}
|
|
})
|
|
.take(80)
|
|
.collect();
|
|
let safe = safe.trim_matches([' ', '.']);
|
|
format!(
|
|
"{}-r{}.zip",
|
|
if safe.is_empty() { "project" } else { safe },
|
|
manifest.sync_revision
|
|
)
|
|
}
|
|
|
|
fn archive_response(
|
|
archive: NamedTempFile,
|
|
permit: Arc<OwnedSemaphorePermit>,
|
|
filename: &str,
|
|
) -> Result<Response, AppError> {
|
|
let size = archive.as_file().metadata().map_err(|_| internal())?.len();
|
|
let reader = archive.reopen().map_err(|_| internal())?;
|
|
let stream = async_stream::stream! {
|
|
let _archive = archive;
|
|
let _permit = permit;
|
|
let mut reader = tokio::fs::File::from_std(reader);
|
|
let mut buffer = vec![0_u8; 64 * 1024];
|
|
loop {
|
|
let count = match reader.read(&mut buffer).await {
|
|
Ok(count) => count,
|
|
Err(error) => { yield Err(error); break; }
|
|
};
|
|
if count == 0 { break; }
|
|
yield Ok::<_, std::io::Error>(Bytes::copy_from_slice(&buffer[..count]));
|
|
}
|
|
};
|
|
let body = Body::from_stream(stream);
|
|
Response::builder()
|
|
.header(header::CONTENT_TYPE, "application/zip")
|
|
.header(header::CONTENT_LENGTH, size.to_string())
|
|
.header(header::CACHE_CONTROL, "private, no-store")
|
|
.header(
|
|
header::CONTENT_DISPOSITION,
|
|
format!(
|
|
"attachment; filename=\"project.zip\"; filename*=UTF-8''{}",
|
|
urlencoding::encode(filename)
|
|
),
|
|
)
|
|
.body(body)
|
|
.map_err(|_| internal())
|
|
}
|
|
|
|
pub async fn admin_download_project_snapshot(
|
|
State(state): State<AppState>,
|
|
Extension(_admin): Extension<AuthenticatedAdmin>,
|
|
Path((user, project)): Path<(String, String)>,
|
|
Query(query): Query<AdminProjectSnapshotDownloadQuery>,
|
|
) -> Result<Response, AppError> {
|
|
validate_agc_project_snapshot_project_id(&user).map_err(bad_request)?;
|
|
validate_agc_project_snapshot_project_id(&project).map_err(bad_request)?;
|
|
let permit = Arc::new(
|
|
DOWNLOAD_PERMITS
|
|
.get_or_init(|| Arc::new(Semaphore::new(2)))
|
|
.clone()
|
|
.try_acquire_owned()
|
|
.map_err(|_| {
|
|
AppError::from_status(StatusCode::TOO_MANY_REQUESTS)
|
|
.with_message("工程下载任务已满，请稍后重试")
|
|
})?,
|
|
);
|
|
let store = OssSnapshotStore {
|
|
oss: project_snapshot_oss(&state)?,
|
|
client: state.editor_oss_http_client(),
|
|
channel: requested_snapshot_channel(&state, query.channel.as_deref())?,
|
|
};
|
|
let manifest = store.manifest(&user, &project).await?.ok_or_else(|| {
|
|
AppError::from_status(StatusCode::NOT_FOUND).with_message("项目工程清单不存在")
|
|
})?;
|
|
let filename = archive_filename(&manifest);
|
|
let archive = tokio::time::timeout(
|
|
Duration::from_secs(600),
|
|
build_archive_guarded(&store, &user, &manifest, Some(permit.clone())),
|
|
)
|
|
.await
|
|
.map_err(|_| {
|
|
AppError::from_status(StatusCode::GATEWAY_TIMEOUT).with_message("项目工程打包超时")
|
|
})??;
|
|
archive_response(archive, permit, &filename)
|
|
}
|
|
|
|
fn bad_request(message: impl Into<String>) -> AppError {
|
|
AppError::from_status(StatusCode::BAD_REQUEST).with_message(message)
|
|
}
|
|
fn upstream(message: &str) -> AppError {
|
|
AppError::from_status(StatusCode::BAD_GATEWAY).with_message(message)
|
|
}
|
|
fn internal() -> AppError {
|
|
AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_message("项目工程归档失败")
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use http_body_util::BodyExt;
|
|
use shared_contracts::agc_project_snapshots::AGC_PROJECT_SNAPSHOT_SCHEMA_VERSION;
|
|
use std::{collections::BTreeMap, io::Read, sync::Mutex};
|
|
|
|
#[derive(Default)]
|
|
struct Store {
|
|
manifests: BTreeMap<(String, String), AgcProjectSnapshotManifestRequest>,
|
|
files: BTreeMap<String, Vec<u8>>,
|
|
list_calls: Mutex<Vec<(Option<String>, Option<String>, usize)>>,
|
|
stall: bool,
|
|
}
|
|
|
|
impl SnapshotStore for Store {
|
|
async fn directories(
|
|
&self,
|
|
user: Option<&str>,
|
|
after: Option<&str>,
|
|
limit: usize,
|
|
) -> Result<SnapshotDirectoryPage, AppError> {
|
|
self.list_calls.lock().unwrap().push((
|
|
user.map(str::to_string),
|
|
after.map(str::to_string),
|
|
limit,
|
|
));
|
|
let directories = self
|
|
.manifests
|
|
.keys()
|
|
.filter_map(|(owner, project)| match user {
|
|
Some(user) if user == owner => Some(project.clone()),
|
|
Some(_) => None,
|
|
None => Some(owner.clone()),
|
|
})
|
|
.filter(|key| after.is_none_or(|after| key.as_str() > after))
|
|
.collect::<std::collections::BTreeSet<_>>()
|
|
.into_iter()
|
|
.collect::<Vec<_>>();
|
|
let has_more = directories.len() > limit;
|
|
let directories = directories.into_iter().take(limit).collect::<Vec<_>>();
|
|
let next_marker = has_more.then(|| directories.last().unwrap().clone());
|
|
Ok(SnapshotDirectoryPage {
|
|
directories,
|
|
next_marker,
|
|
})
|
|
}
|
|
async fn manifest(
|
|
&self,
|
|
user: &str,
|
|
project: &str,
|
|
) -> Result<Option<AgcProjectSnapshotManifestRequest>, AppError> {
|
|
Ok(self
|
|
.manifests
|
|
.get(&(user.to_string(), project.to_string()))
|
|
.cloned())
|
|
}
|
|
async fn file(
|
|
&self,
|
|
_user: &str,
|
|
_project: &str,
|
|
file: &AgcProjectSnapshotManifestFile,
|
|
) -> Result<Vec<u8>, AppError> {
|
|
if self.stall {
|
|
std::future::pending::<()>().await;
|
|
}
|
|
self.files
|
|
.get(&file.relative_path)
|
|
.cloned()
|
|
.ok_or_else(|| upstream("fixture missing"))
|
|
}
|
|
}
|
|
|
|
fn manifest(project: &str, files: &[(&str, &[u8])]) -> AgcProjectSnapshotManifestRequest {
|
|
AgcProjectSnapshotManifestRequest {
|
|
schema_version: AGC_PROJECT_SNAPSHOT_SCHEMA_VERSION,
|
|
project_id: project.to_string(),
|
|
project_name: Some("我的工程".to_string()),
|
|
sync_revision: 7,
|
|
synced_at_ms: 1_700_000_000_000,
|
|
pending_files: Some(0),
|
|
files: files
|
|
.iter()
|
|
.map(|(path, bytes)| AgcProjectSnapshotManifestFile {
|
|
relative_path: path.to_string(),
|
|
size_bytes: bytes.len() as u64,
|
|
checksum: agc_project_snapshot_checksum(bytes),
|
|
})
|
|
.collect(),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn project_snapshots_status_preserves_unknown_and_partial() {
|
|
let mut manifest = manifest(
|
|
"project-1",
|
|
&[("game/empty", b""), ("game/main.js", b"123")],
|
|
);
|
|
let projection = item("dev", "user-1".to_string(), &manifest);
|
|
assert_eq!(projection.status, AdminProjectSnapshotStatus::Ready);
|
|
assert_eq!((projection.file_count, projection.total_bytes), (2, 3));
|
|
assert_eq!(projection.channel, "dev");
|
|
manifest.pending_files = Some(2);
|
|
assert_eq!(
|
|
item("release", "user-1".into(), &manifest).status,
|
|
AdminProjectSnapshotStatus::Partial
|
|
);
|
|
let mut legacy = serde_json::to_value(manifest).unwrap();
|
|
legacy.as_object_mut().unwrap().remove("pendingFiles");
|
|
legacy.as_object_mut().unwrap().remove("projectName");
|
|
let manifest = serde_json::from_value(legacy).unwrap();
|
|
let projection = item("dev", "user-1".into(), &manifest);
|
|
assert_eq!(projection.status, AdminProjectSnapshotStatus::Unverified);
|
|
assert_eq!(projection.project_name, "project-1");
|
|
}
|
|
|
|
/// 后台“用户”列与素材查询同口径：能查到账号时给昵称与陶泥号，查不到时给占位作者。
|
|
#[test]
|
|
fn project_snapshots_items_resolve_author_profile_for_admin_list() {
|
|
use crate::{
|
|
config::AppConfig,
|
|
work_author::{ORPHAN_WORK_AUTHOR_DISPLAY_NAME, ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE},
|
|
};
|
|
|
|
let state = AppState::new(AppConfig::default()).expect("state should build");
|
|
state
|
|
.auth_user_service()
|
|
.ensure_orphan_work_owner_user("user-1", "user-1", "陶泥用户", "SY-00000007")
|
|
.expect("fixture user should be inserted");
|
|
|
|
let mut items = vec![
|
|
item("dev", "user-1".to_string(), &manifest("project-1", &[])),
|
|
item(
|
|
"dev",
|
|
"user-missing".to_string(),
|
|
&manifest("project-2", &[]),
|
|
),
|
|
];
|
|
attach_snapshot_authors(&state, &mut items);
|
|
|
|
assert_eq!(items[0].author_display_name.as_deref(), Some("陶泥用户"));
|
|
assert_eq!(
|
|
items[0].author_public_user_code.as_deref(),
|
|
Some("SY-00000007")
|
|
);
|
|
assert_eq!(
|
|
items[1].author_display_name.as_deref(),
|
|
Some(ORPHAN_WORK_AUTHOR_DISPLAY_NAME)
|
|
);
|
|
assert_eq!(
|
|
items[1].author_public_user_code.as_deref(),
|
|
Some(ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE)
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn project_snapshots_pagination_continues_directories_without_rescan() {
|
|
let mut store = Store::default();
|
|
for (user, project) in [
|
|
("user-a", "project-a"),
|
|
("user-a", "project-b"),
|
|
("user-b", "project-c"),
|
|
] {
|
|
store
|
|
.manifests
|
|
.insert((user.into(), project.into()), manifest(project, &[]));
|
|
}
|
|
let first = list_snapshots(&store, SnapshotCursor::decode(None, "dev").unwrap(), 1)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(first.items[0].project_id, "project-a");
|
|
assert_eq!(first.items[0].channel, "dev");
|
|
let second = list_snapshots(
|
|
&store,
|
|
SnapshotCursor::decode(first.next_cursor.as_deref(), "dev").unwrap(),
|
|
1,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(second.items[0].project_id, "project-b");
|
|
let third = list_snapshots(
|
|
&store,
|
|
SnapshotCursor::decode(second.next_cursor.as_deref(), "dev").unwrap(),
|
|
1,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(third.items[0].project_id, "project-c");
|
|
assert!(third.next_cursor.is_none());
|
|
let calls = store.list_calls.lock().unwrap();
|
|
assert_eq!(calls.len(), 5);
|
|
assert_eq!(
|
|
calls[2],
|
|
(Some("user-a".into()), Some("project-a".into()), 1)
|
|
);
|
|
assert_eq!(calls[3], (None, Some("user-a".into()), 1));
|
|
assert!(SnapshotCursor::decode(Some("not-json"), "dev").is_err());
|
|
// 游标不能跨渠道复用，否则目录推进会落在别的渠道上。
|
|
let next = first.next_cursor.as_deref().expect("first page cursor");
|
|
assert!(SnapshotCursor::decode(Some(next), "release").is_err());
|
|
let escaped = SnapshotCursor {
|
|
channel: "dev".into(),
|
|
user: Some("../user".into()),
|
|
..SnapshotCursor::default()
|
|
}
|
|
.encode();
|
|
assert!(SnapshotCursor::decode(Some(&escaped), "dev").is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn project_snapshots_zip_restores_paths_and_empty_files() {
|
|
let files: [(&str, &[u8]); 3] = [
|
|
("game/src/main.js", b"export const ready=true"),
|
|
("assets/图像.txt", b"asset"),
|
|
("game/empty.txt", b""),
|
|
];
|
|
let store = Store {
|
|
files: files
|
|
.iter()
|
|
.map(|(path, bytes)| (path.to_string(), bytes.to_vec()))
|
|
.collect(),
|
|
..Store::default()
|
|
};
|
|
let mut manifest = manifest("project-1", &files);
|
|
// 历史清单可以导出已存文件，但状态不能冒充完整。
|
|
manifest.pending_files = None;
|
|
let archive = build_archive(&store, "user-1", &manifest).await.unwrap();
|
|
let mut zip = zip::ZipArchive::new(archive.reopen().unwrap()).unwrap();
|
|
assert_eq!(zip.len(), 3);
|
|
for (path, expected) in files {
|
|
let mut actual = Vec::new();
|
|
zip.by_name(path).unwrap().read_to_end(&mut actual).unwrap();
|
|
assert_eq!(actual, expected);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn project_snapshots_zip_fails_closed_and_cleans_tempfiles() {
|
|
let files: [(&str, &[u8]); 1] = [("game/main.js", b"good")];
|
|
let mut manifest = manifest("project-1", &files);
|
|
for (data, pending) in [
|
|
(None, Some(0)),
|
|
(Some(b"bad!".to_vec()), Some(0)),
|
|
(Some(b"good".to_vec()), Some(1)),
|
|
] {
|
|
manifest.pending_files = pending;
|
|
let store = Store {
|
|
files: data
|
|
.map(|bytes| BTreeMap::from([("game/main.js".into(), bytes)]))
|
|
.unwrap_or_default(),
|
|
..Store::default()
|
|
};
|
|
let tempfile = NamedTempFile::new().unwrap();
|
|
let path = tempfile.path().to_path_buf();
|
|
let error = write_archive(&store, "user-1", &manifest, tempfile, None)
|
|
.await
|
|
.unwrap_err();
|
|
assert_eq!(
|
|
error.status_code(),
|
|
if pending == Some(1) {
|
|
StatusCode::CONFLICT
|
|
} else {
|
|
StatusCode::BAD_GATEWAY
|
|
}
|
|
);
|
|
assert!(!path.exists());
|
|
}
|
|
manifest.pending_files = Some(0);
|
|
let tempfile = NamedTempFile::new().unwrap();
|
|
let path = tempfile.path().to_path_buf();
|
|
let stalled = Store {
|
|
stall: true,
|
|
..Store::default()
|
|
};
|
|
assert!(
|
|
tokio::time::timeout(
|
|
Duration::from_millis(10),
|
|
write_archive(&stalled, "user-1", &manifest, tempfile, None)
|
|
)
|
|
.await
|
|
.is_err()
|
|
);
|
|
assert!(!path.exists(), "打包取消必须清理临时文件");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn project_snapshots_response_releases_file_and_permit_on_finish_or_cancel() {
|
|
for cancel in [false, true] {
|
|
let mut tempfile = NamedTempFile::new().unwrap();
|
|
tempfile.write_all(&vec![42; 130_000]).unwrap();
|
|
let path = tempfile.path().to_path_buf();
|
|
let permits = Arc::new(Semaphore::new(1));
|
|
let response = archive_response(
|
|
tempfile,
|
|
Arc::new(permits.clone().acquire_owned().await.unwrap()),
|
|
"我的工程.zip",
|
|
)
|
|
.unwrap();
|
|
assert_eq!(response.headers()[header::CONTENT_TYPE], "application/zip");
|
|
assert!(
|
|
response.headers()[header::CONTENT_DISPOSITION]
|
|
.to_str()
|
|
.unwrap()
|
|
.contains("filename*=UTF-8''%")
|
|
);
|
|
assert_eq!(permits.available_permits(), 0);
|
|
let mut body = response.into_body();
|
|
if cancel {
|
|
assert!(body.frame().await.unwrap().is_ok());
|
|
drop(body);
|
|
} else {
|
|
assert_eq!(body.collect().await.unwrap().to_bytes().len(), 130_000);
|
|
}
|
|
assert!(!path.exists());
|
|
assert_eq!(permits.available_permits(), 1);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn project_snapshots_archive_rejects_ambiguous_paths_and_unsafe_names() {
|
|
for paths in [
|
|
vec!["../x"],
|
|
vec!["/absolute"],
|
|
vec!["game/CON.txt"],
|
|
vec!["game/name."],
|
|
vec!["game/name "],
|
|
vec!["game/a", "game/A"],
|
|
vec!["Game/a", "game/b"],
|
|
vec!["game/a", "game/a/b"],
|
|
] {
|
|
let files = paths
|
|
.iter()
|
|
.map(|path| (*path, b"".as_slice()))
|
|
.collect::<Vec<_>>();
|
|
assert!(
|
|
validate_manifest(&manifest("project-1", &files)).is_err(),
|
|
"{paths:?}"
|
|
);
|
|
}
|
|
let mut manifest = manifest("project-1", &[]);
|
|
manifest.project_name = Some("../我的\\工程\r\n.zip".into());
|
|
let name = archive_filename(&manifest);
|
|
assert!(!name.contains(['/', '\\', '\r', '\n']));
|
|
assert!(name.ends_with("-r7.zip"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn project_snapshots_cancelled_wait_keeps_permit_until_blocking_work_exits() {
|
|
let permits = Arc::new(Semaphore::new(1));
|
|
let permit = Arc::new(permits.clone().acquire_owned().await.unwrap());
|
|
let entered = Arc::new(tokio::sync::Notify::new());
|
|
let started = entered.clone();
|
|
let (release, wait) = std::sync::mpsc::channel();
|
|
let task = tokio::spawn(archive_blocking(Some(permit), move || {
|
|
started.notify_one();
|
|
wait.recv().unwrap();
|
|
Ok(())
|
|
}));
|
|
entered.notified().await;
|
|
task.abort();
|
|
assert!(task.await.unwrap_err().is_cancelled());
|
|
assert_eq!(
|
|
permits.available_permits(),
|
|
0,
|
|
"取消等待不得归还仍在压缩的额度"
|
|
);
|
|
release.send(()).unwrap();
|
|
let permit = tokio::time::timeout(Duration::from_secs(2), permits.clone().acquire_owned())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
drop(permit);
|
|
assert_eq!(permits.available_permits(), 1);
|
|
}
|
|
|
|
/// 显式运行的只读 OSS 验证；配置仅进入内存，禁止输出凭据和用户文件内容。
|
|
#[tokio::test]
|
|
#[ignore = "requires explicitly supplied private OSS environment files; read-only"]
|
|
async fn project_snapshots_live_readonly_list_and_archive() {
|
|
let paths = std::env::var_os("GENARRATIVE_PROJECT_SNAPSHOT_SMOKE_ENV_FILES")
|
|
.expect("显式提供只读 smoke 配置文件列表");
|
|
let mut settings = BTreeMap::new();
|
|
for path in std::env::split_paths(&paths) {
|
|
let source =
|
|
std::fs::read_to_string(path).unwrap_or_else(|_| panic!("无法读取 smoke 配置文件"));
|
|
// 其它服务配置可能使用不同转义语法；这里只解析当前验证实际消费的键。
|
|
let source = source
|
|
.trim_start_matches('\u{feff}')
|
|
.lines()
|
|
.filter(|line| {
|
|
let line = line
|
|
.trim_start()
|
|
.strip_prefix("export ")
|
|
.unwrap_or(line.trim_start());
|
|
line.split_once('=').is_some_and(|(key, _)| {
|
|
matches!(
|
|
key.trim(),
|
|
"GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_BUCKET"
|
|
| "GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_ENDPOINT"
|
|
| "GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_ACCESS_KEY_ID"
|
|
| "GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_ACCESS_KEY_SECRET"
|
|
| "ALIYUN_OSS_ACCESS_KEY_ID"
|
|
| "ALIYUN_OSS_ACCESS_KEY_SECRET"
|
|
)
|
|
})
|
|
})
|
|
.collect::<Vec<_>>()
|
|
.join("\n");
|
|
let entries = dotenvy::from_read_iter(source.as_bytes());
|
|
for entry in entries {
|
|
let (key, value) = entry.unwrap_or_else(|_| panic!("无法解析 smoke 配置行"));
|
|
settings.insert(key, value);
|
|
}
|
|
}
|
|
let resolve = |specific: &str, fallback: &str, default: &str| {
|
|
settings
|
|
.get(specific)
|
|
.or_else(|| settings.get(fallback))
|
|
.cloned()
|
|
.unwrap_or_else(|| default.to_string())
|
|
};
|
|
let config = platform_oss::OssConfig::new(
|
|
resolve("GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_BUCKET", "", "agc-dev"),
|
|
resolve(
|
|
"GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_ENDPOINT",
|
|
"",
|
|
"oss-rg-china-mainland.aliyuncs.com",
|
|
),
|
|
resolve(
|
|
"GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_ACCESS_KEY_ID",
|
|
"ALIYUN_OSS_ACCESS_KEY_ID",
|
|
"",
|
|
),
|
|
resolve(
|
|
"GENARRATIVE_AGC_PROJECT_SNAPSHOT_OSS_ACCESS_KEY_SECRET",
|
|
"ALIYUN_OSS_ACCESS_KEY_SECRET",
|
|
"",
|
|
),
|
|
600,
|
|
600,
|
|
64 * 1024 * 1024,
|
|
200,
|
|
)
|
|
.unwrap_or_else(|_| panic!("只读 smoke OSS 配置无效"));
|
|
let oss = OssClient::new(config);
|
|
let client = reqwest::Client::builder()
|
|
.timeout(Duration::from_secs(60))
|
|
.build()
|
|
.unwrap();
|
|
let channel = std::env::var("GENARRATIVE_AGC_PROJECT_SNAPSHOT_CHANNEL")
|
|
.unwrap_or_else(|_| "dev".to_string());
|
|
let store = OssSnapshotStore {
|
|
oss: &oss,
|
|
client: &client,
|
|
channel: channel.clone(),
|
|
};
|
|
let mut cursor = SnapshotCursor::decode(None, &channel).unwrap();
|
|
let mut projects = 0;
|
|
let mut total_files = 0;
|
|
let mut total_bytes = 0_u64;
|
|
for page_index in 0..100 {
|
|
let page = list_snapshots(&store, cursor, 1)
|
|
.await
|
|
.expect("真实目录读取失败");
|
|
for item in page.items {
|
|
let manifest = store
|
|
.manifest(&item.user_id, &item.project_id)
|
|
.await
|
|
.unwrap()
|
|
.expect("清单应存在");
|
|
assert_ne!(
|
|
item.status,
|
|
AdminProjectSnapshotStatus::Partial,
|
|
"只读 smoke 遇到未完成项目"
|
|
);
|
|
let archive = build_archive(&store, &item.user_id, &manifest)
|
|
.await
|
|
.expect("真实归档失败");
|
|
let path = archive.path().to_path_buf();
|
|
{
|
|
let mut zip = zip::ZipArchive::new(archive.reopen().unwrap()).unwrap();
|
|
assert_eq!(zip.len(), manifest.files.len());
|
|
for file in &manifest.files {
|
|
let mut bytes = Vec::new();
|
|
zip.by_name(&file.relative_path)
|
|
.expect("ZIP 应保留原路径")
|
|
.read_to_end(&mut bytes)
|
|
.unwrap();
|
|
verify_file(file, &bytes).expect("解压后大小和摘要应匹配");
|
|
}
|
|
}
|
|
drop(archive);
|
|
assert!(!path.exists());
|
|
projects += 1;
|
|
total_files += item.file_count;
|
|
total_bytes += item.total_bytes;
|
|
eprintln!(
|
|
"read-only snapshot {projects}: files={}, bytes={}, status={:?}",
|
|
item.file_count, item.total_bytes, item.status
|
|
);
|
|
}
|
|
let Some(next) = page.next_cursor else { break };
|
|
assert!(page_index < 99, "只读 smoke 已达到分页上限");
|
|
cursor = SnapshotCursor::decode(Some(&next), &channel).unwrap();
|
|
}
|
|
assert!(projects > 0, "真实 bucket 未发现项目清单");
|
|
eprintln!(
|
|
"read-only snapshot verification passed: projects={projects}, files={total_files}, bytes={total_bytes}"
|
|
);
|
|
}
|
|
}
|