GenarrativeAI/Genarrative
5
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

feat: add oss direct upload adapter

Browse Source
This commit is contained in:
kdletters
2026-04-21 14:36:34 +08:00
parent 39eb7a513c
commit 5675c40119
20 changed files with 1308 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server-rs/crates/api-server/Cargo.toml
View File

@@ -6,7 +6,9 @@ license.workspace = true
[dependencies]
axum = "0.8"
dotenvy = "0.15"
platform-auth = { path = "../platform-auth" }
platform-oss = { path = "../platform-oss" }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
shared-logging = { path = "../shared-logging" }

18
server-rs/crates/api-server/src/api_response.rs
View File

@@ -27,15 +27,15 @@ pub fn json_success_body<T>(request_context: Option<&RequestContext>, data: T) -
where
T: Serialize,
{
if let Some(context) = request_context {
if context.wants_envelope() {
return Json(json!({
"ok": true,
"data": data,
"error": null,
"meta": build_api_response_meta(Some(context)),
}));
}
if let Some(context) = request_context
&& context.wants_envelope()
{
return Json(json!({
"ok": true,
"data": data,
"error": null,
"meta": build_api_response_meta(Some(context)),
}));
}
Json(serde_json::to_value(data).unwrap_or(Value::Null))

14
server-rs/crates/api-server/src/app.rs
View File

@@ -1,8 +1,16 @@
use axum::{Router, body::Body, extract::Extension, http::Request, middleware, routing::get};
use axum::{
Router,
body::Body,
extract::Extension,
http::Request,
middleware,
routing::{get, post},
};
use tower_http::trace::{DefaultOnFailure, DefaultOnRequest, DefaultOnResponse, TraceLayer};
use tracing::{Level, info_span};
use crate::{
assets::create_direct_upload_ticket,
auth::{
attach_refresh_session_token, inspect_auth_claims, inspect_refresh_session_cookie,
require_bearer_auth,
@@ -37,6 +45,10 @@ pub fn build_router(state: AppState) -> Router {
attach_refresh_session_token,
)),
)
.route(
"/api/assets/direct-upload-tickets",
post(create_direct_upload_ticket),
)
// 错误归一化层放在 tracing 里侧，让 tracing 记录到最终对外返回的状态与错误体形态。
.layer(middleware::from_fn(normalize_error_response))
// 响应头回写放在错误归一化外侧，确保最终写回的是归一化后的最终响应。

206
server-rs/crates/api-server/src/assets.rs Normal file
View File

@@ -0,0 +1,206 @@
use std::collections::BTreeMap;
use axum::{
Json,
extract::{Extension, State},
http::StatusCode,
};
use platform_oss::{LegacyAssetPrefix, OssObjectAccess, OssPostObjectRequest};
use serde::Deserialize;
use serde_json::{Value, json};
use crate::{
api_response::json_success_body, http_error::AppError, request_context::RequestContext,
state::AppState,
};
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateDirectUploadTicketRequest {
pub legacy_prefix: String,
#[serde(default)]
pub path_segments: Vec<String>,
pub file_name: String,
#[serde(default)]
pub content_type: Option<String>,
#[serde(default)]
pub access: Option<OssObjectAccess>,
#[serde(default)]
pub metadata: BTreeMap<String, String>,
#[serde(default)]
pub max_size_bytes: Option<u64>,
#[serde(default)]
pub expire_seconds: Option<u64>,
#[serde(default)]
pub success_action_status: Option<u16>,
}
pub async fn create_direct_upload_ticket(
State(state): State<AppState>,
Extension(request_context): Extension<RequestContext>,
Json(payload): Json<CreateDirectUploadTicketRequest>,
) -> Result<Json<Value>, AppError> {
let oss_client = state.oss_client().ok_or_else(|| {
AppError::from_status(StatusCode::SERVICE_UNAVAILABLE).with_details(json!({
"provider": "aliyun-oss",
"reason": "OSS 未完成环境变量配置",
}))
})?;
let legacy_prefix = LegacyAssetPrefix::parse(&payload.legacy_prefix).ok_or_else(|| {
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
"field": "legacyPrefix",
"supported": platform_oss::LEGACY_PUBLIC_PREFIXES,
}))
})?;
let signed = oss_client
.sign_post_object(OssPostObjectRequest {
prefix: legacy_prefix,
path_segments: payload.path_segments,
file_name: payload.file_name,
content_type: payload.content_type,
access: payload.access.unwrap_or(OssObjectAccess::Public),
metadata: payload.metadata,
max_size_bytes: payload.max_size_bytes,
expire_seconds: payload.expire_seconds,
success_action_status: payload.success_action_status,
})
.map_err(|error| {
AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
"provider": "aliyun-oss",
"message": error.to_string(),
}))
})?;
Ok(json_success_body(
Some(&request_context),
json!({
"upload": signed,
}),
))
}
#[cfg(test)]
mod tests {
use axum::{
body::Body,
http::{Request, StatusCode},
};
use http_body_util::BodyExt;
use serde_json::{Value, json};
use tower::ServiceExt;
use crate::{app::build_router, config::AppConfig, state::AppState};
#[tokio::test]
async fn direct_upload_ticket_returns_service_unavailable_when_oss_missing() {
let app = build_router(AppState::new(AppConfig::default()).expect("state should build"));
let response = app
.oneshot(
Request::builder()
.method("POST")
.uri("/api/assets/direct-upload-tickets")
.header("content-type", "application/json")
.body(Body::from(
json!({
"legacyPrefix": "/generated-characters/*",
"pathSegments": ["hero", "visual", "asset-01"],
"fileName": "master.png"
})
.to_string(),
))
.expect("request should build"),
)
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
let body = response
.into_body()
.collect()
.await
.expect("body should collect")
.to_bytes();
let payload: Value =
serde_json::from_slice(&body).expect("response body should be valid json");
assert_eq!(
payload["error"]["code"],
Value::String("SERVICE_UNAVAILABLE".to_string())
);
assert_eq!(
payload["error"]["details"]["provider"],
Value::String("aliyun-oss".to_string())
);
}
#[tokio::test]
async fn direct_upload_ticket_returns_signed_payload_when_oss_configured() {
let config = AppConfig {
oss_bucket: Some("genarrative-assets".to_string()),
oss_endpoint: Some("oss-cn-shanghai.aliyuncs.com".to_string()),
oss_access_key_id: Some("test-access-key-id".to_string()),
oss_access_key_secret: Some("test-access-key-secret".to_string()),
oss_public_base_url: Some("https://cdn.genarrative.local".to_string()),
..AppConfig::default()
};
let app = build_router(AppState::new(config).expect("state should build"));
let response = app
.oneshot(
Request::builder()
.method("POST")
.uri("/api/assets/direct-upload-tickets")
.header("content-type", "application/json")
.header("x-request-id", "req-oss-ticket")
.header("x-genarrative-response-envelope", "1")
.body(Body::from(
json!({
"legacyPrefix": "/generated-characters/*",
"pathSegments": ["hero_001", "visual", "asset_01"],
"fileName": "master.png",
"contentType": "image/png",
"metadata": {
"asset-kind": "character-visual"
}
})
.to_string(),
))
.expect("request should build"),
)
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let body = response
.into_body()
.collect()
.await
.expect("body should collect")
.to_bytes();
let payload: Value =
serde_json::from_slice(&body).expect("response body should be valid json");
assert_eq!(payload["ok"], Value::Bool(true));
assert_eq!(
payload["data"]["upload"]["objectKey"],
Value::String("generated-characters/hero_001/visual/asset_01/master.png".to_string())
);
assert_eq!(
payload["data"]["upload"]["publicUrl"],
Value::String(
"https://cdn.genarrative.local/generated-characters/hero_001/visual/asset_01/master.png"
.to_string()
)
);
assert_eq!(
payload["data"]["upload"]["formFields"]["OSSAccessKeyId"],
Value::String("test-access-key-id".to_string())
);
}
}

98
server-rs/crates/api-server/src/config.rs
View File

@@ -14,6 +14,14 @@ pub struct AppConfig {
pub refresh_cookie_secure: bool,
pub refresh_cookie_same_site: String,
pub refresh_session_ttl_days: u32,
pub oss_bucket: Option<String>,
pub oss_endpoint: Option<String>,
pub oss_access_key_id: Option<String>,
pub oss_access_key_secret: Option<String>,
pub oss_public_base_url: Option<String>,
pub oss_post_expire_seconds: u64,
pub oss_post_max_size_bytes: u64,
pub oss_success_action_status: u16,
}
impl Default for AppConfig {
@@ -30,6 +38,14 @@ impl Default for AppConfig {
refresh_cookie_secure: false,
refresh_cookie_same_site: "Lax".to_string(),
refresh_session_ttl_days: 30,
oss_bucket: None,
oss_endpoint: None,
oss_access_key_id: None,
oss_access_key_secret: None,
oss_public_base_url: None,
oss_post_expire_seconds: 10 * 60,
oss_post_max_size_bytes: 20 * 1024 * 1024,
oss_success_action_status: 200,
}
}
}
@@ -38,22 +54,22 @@ impl AppConfig {
pub fn from_env() -> Self {
let mut config = Self::default();
if let Ok(bind_host) = env::var("GENARRATIVE_API_HOST") {
if !bind_host.trim().is_empty() {
config.bind_host = bind_host;
}
if let Ok(bind_host) = env::var("GENARRATIVE_API_HOST")
&& !bind_host.trim().is_empty()
{
config.bind_host = bind_host;
}
if let Ok(bind_port) = env::var("GENARRATIVE_API_PORT") {
if let Ok(parsed_port) = bind_port.parse::<u16>() {
config.bind_port = parsed_port;
}
if let Ok(bind_port) = env::var("GENARRATIVE_API_PORT")
&& let Ok(parsed_port) = bind_port.parse::<u16>()
{
config.bind_port = parsed_port;
}
if let Ok(log_filter) = env::var("GENARRATIVE_API_LOG") {
if !log_filter.trim().is_empty() {
config.log_filter = log_filter;
}
if let Ok(log_filter) = env::var("GENARRATIVE_API_LOG")
&& !log_filter.trim().is_empty()
{
config.log_filter = log_filter;
}
if let Some(jwt_issuer) =
@@ -99,6 +115,30 @@ impl AppConfig {
config.refresh_session_ttl_days = refresh_session_ttl_days;
}
config.oss_bucket = read_first_non_empty_env(&["ALIYUN_OSS_BUCKET"]);
config.oss_endpoint = read_first_non_empty_env(&["ALIYUN_OSS_ENDPOINT"]);
config.oss_access_key_id = read_first_non_empty_env(&["ALIYUN_OSS_ACCESS_KEY_ID"]);
config.oss_access_key_secret = read_first_non_empty_env(&["ALIYUN_OSS_ACCESS_KEY_SECRET"]);
config.oss_public_base_url = read_first_non_empty_env(&["ALIYUN_OSS_PUBLIC_BASE_URL"]);
if let Some(oss_post_expire_seconds) =
read_first_duration_seconds_env(&["ALIYUN_OSS_POST_EXPIRE_SECONDS"])
{
config.oss_post_expire_seconds = oss_post_expire_seconds;
}
if let Some(oss_post_max_size_bytes) =
read_first_positive_u64_env(&["ALIYUN_OSS_POST_MAX_SIZE_BYTES"])
{
config.oss_post_max_size_bytes = oss_post_max_size_bytes;
}
if let Some(oss_success_action_status) =
read_first_positive_u16_env(&["ALIYUN_OSS_SUCCESS_ACTION_STATUS"])
{
config.oss_success_action_status = oss_success_action_status;
}
config
}
@@ -144,6 +184,22 @@ fn read_first_positive_u32_env(keys: &[&str]) -> Option<u32> {
})
}
fn read_first_positive_u64_env(keys: &[&str]) -> Option<u64> {
keys.iter().find_map(|key| {
env::var(key)
.ok()
.and_then(|value| parse_positive_u64(&value))
})
}
fn read_first_positive_u16_env(keys: &[&str]) -> Option<u16> {
keys.iter().find_map(|key| {
env::var(key)
.ok()
.and_then(|value| parse_positive_u16(&value))
})
}
fn parse_duration_seconds(raw: &str) -> Option<u64> {
let raw = raw.trim();
if raw.is_empty() {
@@ -185,3 +241,21 @@ fn parse_positive_u32(raw: &str) -> Option<u32> {
Some(value)
}
fn parse_positive_u64(raw: &str) -> Option<u64> {
let value = raw.trim().parse::<u64>().ok()?;
if value == 0 {
return None;
}
Some(value)
}
fn parse_positive_u16(raw: &str) -> Option<u16> {
let value = raw.trim().parse::<u16>().ok()?;
if value == 0 {
return None;
}
Some(value)
}

6
server-rs/crates/api-server/src/http_error.rs
View File

@@ -39,6 +39,11 @@ impl AppError {
self.code
}
pub fn with_details(mut self, details: Value) -> Self {
self.details = Some(details);
self
}
pub fn into_response_with_context(self, request_context: Option<&RequestContext>) -> Response {
let status_code = self.status_code;
let payload = self.to_payload();
@@ -70,6 +75,7 @@ fn resolve_http_error(status_code: StatusCode) -> (&'static str, &'static str) {
StatusCode::CONFLICT => ("CONFLICT", "请求冲突"),
StatusCode::TOO_MANY_REQUESTS => ("TOO_MANY_REQUESTS", "请求过于频繁"),
StatusCode::BAD_GATEWAY => ("UPSTREAM_ERROR", "上游服务请求失败"),
StatusCode::SERVICE_UNAVAILABLE => ("SERVICE_UNAVAILABLE", "服务暂不可用"),
_ if status_code.is_client_error() => ("BAD_REQUEST", "请求参数不合法"),
_ => ("INTERNAL_SERVER_ERROR", "服务器内部错误"),
}

7
server-rs/crates/api-server/src/main.rs
View File

@@ -1,5 +1,6 @@
mod api_response;
mod app;
mod assets;
mod auth;
mod config;
mod error_middleware;
@@ -17,6 +18,10 @@ use crate::{app::build_router, config::AppConfig, state::AppState};
#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
// 运行本地开发与联调时，优先从仓库根目录的 .env / .env.local 加载变量，避免手工逐项导出 OSS 配置。
let _ = dotenvy::from_filename(".env");
let _ = dotenvy::from_filename(".env.local");
// 统一先从配置对象读取监听地址，避免后续把环境变量读取散落到入口和路由层。
let config = AppConfig::from_env();
init_tracing(&config.log_filter)?;
@@ -25,7 +30,7 @@ async fn main() -> Result<(), std::io::Error> {
let listener = TcpListener::bind(bind_address).await?;
let state = AppState::new(config)
.map_err(|error| std::io::Error::other(format!("初始化鉴权配置失败：{error}")))?;
.map_err(|error| std::io::Error::other(format!("初始化应用状态失败：{error}")))?;
let router = build_router(state);
info!(%bind_address, "api-server 已完成 tracing 初始化并开始监听");

24
server-rs/crates/api-server/src/response_headers.rs
View File

@@ -19,12 +19,12 @@ pub async fn propagate_request_id_header(request: Request, next: Next) -> Respon
let request_context = request.extensions().get::<RequestContext>().cloned();
let mut response = next.run(request).await;
if let Some(request_id) = request_id {
if let Ok(header_value) = HeaderValue::from_str(&request_id) {
response
.headers_mut()
.insert(HeaderName::from_static(X_REQUEST_ID_HEADER), header_value);
}
if let Some(request_id) = request_id
&& let Ok(header_value) = HeaderValue::from_str(&request_id)
{
response
.headers_mut()
.insert(HeaderName::from_static(X_REQUEST_ID_HEADER), header_value);
}
if let Ok(header_value) = HeaderValue::from_str(API_VERSION) {
@@ -37,12 +37,12 @@ pub async fn propagate_request_id_header(request: Request, next: Next) -> Respon
.insert(HeaderName::from_static(ROUTE_VERSION_HEADER), header_value);
}
if let Some(request_context) = request_context {
if let Ok(header_value) = HeaderValue::from_str(&request_context.elapsed().to_string()) {
response
.headers_mut()
.insert(HeaderName::from_static(RESPONSE_TIME_HEADER), header_value);
}
if let Some(request_context) = request_context
&& let Ok(header_value) = HeaderValue::from_str(&request_context.elapsed().to_string())
{
response
.headers_mut()
.insert(HeaderName::from_static(RESPONSE_TIME_HEADER), header_value);
}
response

41
server-rs/crates/api-server/src/state.rs
View File

@@ -3,6 +3,7 @@ use std::{error::Error, fmt};
use platform_auth::{
JwtConfig, JwtError, RefreshCookieConfig, RefreshCookieError, RefreshCookieSameSite,
};
use platform_oss::{OssClient, OssConfig, OssError};
use crate::config::AppConfig;
@@ -14,12 +15,14 @@ pub struct AppState {
pub config: AppConfig,
auth_jwt_config: JwtConfig,
refresh_cookie_config: RefreshCookieConfig,
oss_client: Option<OssClient>,
}
#[derive(Debug)]
pub enum AppStateInitError {
Jwt(JwtError),
RefreshCookie(RefreshCookieError),
Oss(OssError),
}
impl AppState {
@@ -40,11 +43,13 @@ impl AppState {
refresh_cookie_same_site,
config.refresh_session_ttl_days,
)?;
let oss_client = build_oss_client(&config)?;
Ok(Self {
config,
auth_jwt_config,
refresh_cookie_config,
oss_client,
})
}
@@ -55,6 +60,10 @@ impl AppState {
pub fn refresh_cookie_config(&self) -> &RefreshCookieConfig {
&self.refresh_cookie_config
}
pub fn oss_client(&self) -> Option<&OssClient> {
self.oss_client.as_ref()
}
}
impl fmt::Display for AppStateInitError {
@@ -62,6 +71,7 @@ impl fmt::Display for AppStateInitError {
match self {
Self::Jwt(error) => write!(f, "{error}"),
Self::RefreshCookie(error) => write!(f, "{error}"),
Self::Oss(error) => write!(f, "{error}"),
}
}
}
@@ -79,3 +89,34 @@ impl From<RefreshCookieError> for AppStateInitError {
Self::RefreshCookie(value)
}
}
impl From<OssError> for AppStateInitError {
fn from(value: OssError) -> Self {
Self::Oss(value)
}
}
fn build_oss_client(config: &AppConfig) -> Result<Option<OssClient>, AppStateInitError> {
let has_any_oss_field = config.oss_bucket.is_some()
|| config.oss_endpoint.is_some()
|| config.oss_access_key_id.is_some()
|| config.oss_access_key_secret.is_some()
|| config.oss_public_base_url.is_some();
if !has_any_oss_field {
return Ok(None);
}
let oss_config = OssConfig::new(
config.oss_bucket.clone().unwrap_or_default(),
config.oss_endpoint.clone().unwrap_or_default(),
config.oss_access_key_id.clone().unwrap_or_default(),
config.oss_access_key_secret.clone().unwrap_or_default(),
config.oss_public_base_url.clone(),
config.oss_post_expire_seconds,
config.oss_post_max_size_bytes,
config.oss_success_action_status,
)?;
Ok(Some(OssClient::new(oss_config)))
}