#[cfg(test)] use std::sync::Mutex; #[cfg(test)] use std::sync::atomic::AtomicUsize; use std::{ collections::BTreeMap, error::Error, fmt, sync::{ Arc, atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}, }, }; use axum::extract::FromRef; use module_ai::{AiTaskService, InMemoryAiTaskStore}; use module_auth::{ AuthUserService, InMemoryAuthStore, PasswordEntryService, PhoneAuthService, RefreshSessionService, WechatAuthService, WechatAuthStateService, }; use platform_auth::{ AccessTokenClaims, AccessTokenClaimsInput, AuthProvider, BindingStatus, JwtConfig, JwtError, RefreshCookieConfig, RefreshCookieError, RefreshCookieSameSite, SmsAuthConfig, SmsAuthProvider, SmsAuthProviderKind, SmsProviderError, WechatProvider, sign_access_token, verify_access_token, }; use platform_llm::{LlmClient, LlmConfig, LlmError, LlmProvider, OpenAiChatTokenBudgetField}; use platform_matting::{MattingClient, MattingConfig}; use platform_oss::{OssClient, OssConfig, OssError}; use platform_wechat::{WechatClient, WechatConfig, pay::WechatPayClient}; #[cfg(test)] use spacetime_client::ExternalGenerationJobRecord; use spacetime_client::{ EditorGenerationModelPricingRecord, EditorGenerationPricingConfigRecord, EditorGenerationPricingConfigUpsertRecordInput, EditorGenerationPricingTierRecord, SpacetimeClient, SpacetimeClientConfig, SpacetimeClientError, SpacetimeClientHealthSnapshot, }; use time::OffsetDateTime; use tokio::sync::{Mutex as AsyncMutex, Semaphore, broadcast}; use tracing::{info, warn}; use crate::config::AppConfig; use crate::editor_generation_config::{ EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS, EditorGenerationModelPricing, EditorGenerationPricingConfig, EditorGenerationPricingError, EditorGenerationPricingStore, EditorGenerationPricingUnit, }; use crate::tracking_outbox::TrackingOutbox; use crate::wallet_refund_outbox::{ProfileWalletRefundOutboxWorker, WalletRefundOutbox}; use crate::wechat::pay::{build_wechat_pay_config, map_wechat_pay_init_error}; use crate::wechat::provider::build_wechat_provider; use crate::work_author::{ ORPHAN_WORK_AUTHOR_DISPLAY_NAME, ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE, ORPHAN_WORK_OWNER_USER_ID, }; const ADMIN_ROLE: &str = "admin"; const EDITOR_AGENT_LLM_MAX_RETRIES: u32 = 1; const EDITOR_AGENT_LLM_MAX_RETRY_BACKOFF_MS: u64 = 60_000; pub(crate) const CHARACTER_ANIMATION_OSS_MAX_CONCURRENCY: usize = 8; // P=8：父侧成功图片读取/解码槽。配 N=16 是内存与出口吞吐的折中， // 极端完整 body 内存按 (N + P) × 32 MiB 评估（见调度方案 §9.2）。 pub(crate) const BGFILTER_IMAGE_VALIDATION_MAX_CONCURRENCY: usize = 8; pub type HttpRequestPermitPool = Semaphore; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum HttpRequestPermitPoolKind { Default, Admin, } impl HttpRequestPermitPoolKind { pub fn as_str(self) -> &'static str { match self { Self::Default => "default", Self::Admin => "admin", } } } #[derive(Clone, Debug)] pub struct HttpRequestPermitPools { default: Option>, admin: Option>, } impl HttpRequestPermitPools { fn from_config(config: &AppConfig) -> Self { Self { default: config .max_concurrent_requests .map(HttpRequestPermitPool::new) .map(Arc::new), admin: config .admin_max_concurrent_requests .map(HttpRequestPermitPool::new) .map(Arc::new), } } pub fn pool( &self, kind: HttpRequestPermitPoolKind, ) -> Option<(HttpRequestPermitPoolKind, Arc)> { let selected = match kind { HttpRequestPermitPoolKind::Default => self.default.clone(), HttpRequestPermitPoolKind::Admin => self.admin.clone(), }; selected.map(|pool| (kind, pool)).or_else(|| { self.default .clone() .map(|pool| (HttpRequestPermitPoolKind::Default, pool)) }) } } #[derive(Clone, Debug)] pub struct BackpressureState { permit_pools: HttpRequestPermitPools, } impl BackpressureState { pub fn request_permit_pool( &self, kind: HttpRequestPermitPoolKind, ) -> Option<(HttpRequestPermitPoolKind, Arc)> { self.permit_pools.pool(kind) } } #[derive(Clone)] pub struct AppState(Arc); #[cfg(test)] #[derive(Clone)] struct TestExternalBackgroundRemovalEnqueue { expected_owner_user_id: String, expected_source_image_src: String, expected_idempotency_key: String, job: ExternalGenerationJobRecord, } impl fmt::Debug for AppState { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_tuple("AppState").field(self.0.as_ref()).finish() } } impl std::ops::Deref for AppState { type Target = AppStateInner; fn deref(&self) -> &Self::Target { &self.0 } } impl FromRef for BackpressureState { fn from_ref(state: &AppState) -> Self { Self { permit_pools: state.http_request_permit_pools(), } } } #[derive(Clone, Debug)] #[cfg(any())] pub struct PuzzleApiState { root_state: AppState, spacetime_client: SpacetimeClient, puzzle_gallery_cache: PuzzleGalleryCache, oss_client: Option, auth_user_service: AuthUserService, llm_client: Option, creative_agent_gpt5_client: Option, creation_agent_llm_web_search_enabled: bool, vector_engine_image_request_timeout_ms: u64, } #[cfg(any())] impl PuzzleApiState { pub fn root_state(&self) -> &AppState { &self.root_state } pub fn spacetime_client(&self) -> &SpacetimeClient { &self.spacetime_client } pub fn puzzle_gallery_cache(&self) -> &PuzzleGalleryCache { &self.puzzle_gallery_cache } pub fn oss_client(&self) -> Option<&OssClient> { self.oss_client.as_ref() } pub fn auth_user_service(&self) -> &AuthUserService { &self.auth_user_service } pub fn llm_client(&self) -> Option<&LlmClient> { self.llm_client.as_ref() } pub fn creative_agent_gpt5_client(&self) -> Option<&LlmClient> { self.creative_agent_gpt5_client.as_ref() } pub fn creation_agent_llm_web_search_enabled(&self) -> bool { self.creation_agent_llm_web_search_enabled } pub fn vector_engine_image_request_timeout_ms(&self) -> u64 { self.vector_engine_image_request_timeout_ms } pub fn vector_engine_base_url(&self) -> &str { self.root_state.config.vector_engine_base_url.as_str() } pub fn vector_engine_api_key(&self) -> Option<&str> { self.root_state.config.vector_engine_api_key.as_deref() } } #[cfg(any())] impl FromRef for PuzzleApiState { fn from_ref(state: &AppState) -> Self { // 中文注释：拼图路由只暴露本能力需要的依赖快照，避免 handler 直接看见完整 AppState。 Self { root_state: state.clone(), spacetime_client: state.spacetime_client.clone(), puzzle_gallery_cache: state.puzzle_gallery_cache.clone(), oss_client: state.oss_client.clone(), auth_user_service: state.auth_user_service.clone(), llm_client: state.llm_client.clone(), creative_agent_gpt5_client: state.creative_agent_gpt5_client.clone(), creation_agent_llm_web_search_enabled: state .config .creation_agent_llm_web_search_enabled, vector_engine_image_request_timeout_ms: state .config .vector_engine_image_request_timeout_ms, } } } // Axum/Hyper 会在路由树和连接 service 上频繁 clone state；AppState 外层必须保持浅拷贝。 pub struct AppStateInner { // 配置会在后续中间件、路由和平台适配接入时逐步消费。 #[allow(dead_code)] pub config: AppConfig, ready: AtomicBool, /// 本进程是否已至少一次连通 BgFilter worker（收到任意 HTTP 响应即算）。 /// 连接失败重试用它区分冷启动窗口（开机/首连，宽限退避）与运行中途故障（快速收口）。 bgfilter_worker_reached: AtomicBool, http_request_permit_pools: HttpRequestPermitPools, auth_jwt_config: JwtConfig, admin_runtime: Option, refresh_cookie_config: RefreshCookieConfig, #[cfg(any())] test_creation_entry_config: Arc>>, #[cfg(test)] test_feature_gate_config: Arc>>>, #[cfg(test)] test_spacetime_health: Arc>>, #[cfg(test)] test_editor_generation_enqueue_attempts: AtomicUsize, #[cfg(test)] test_fail_editor_generation_enqueue: AtomicBool, #[cfg(test)] test_external_background_removal_enqueue: Arc>>, oss_client: Option, #[cfg_attr(test, allow(dead_code))] auth_store: InMemoryAuthStore, /// 当前进程工作集所基于的正式认证投影版本；跨节点写入使用它做 CAS。 #[cfg_attr(test, allow(dead_code))] auth_projection_version: AtomicI64, /// 最近一次确认写入正式投影时对应的工作集 revision；不一致表示有待重试的本地变更。 #[cfg_attr(test, allow(dead_code))] auth_projection_synced_revision: AtomicU64, #[cfg_attr(test, allow(dead_code))] auth_projection_sync_lock: AsyncMutex<()>, password_entry_service: PasswordEntryService, refresh_session_service: RefreshSessionService, auth_user_service: AuthUserService, phone_auth_service: PhoneAuthService, wechat_auth_state_service: WechatAuthStateService, wechat_auth_service: WechatAuthService, wechat_provider: WechatProvider, wechat_client: WechatClient, wechat_pay_client: WechatPayClient, #[cfg_attr(not(test), allow(dead_code))] ai_task_service: AiTaskService, spacetime_client: SpacetimeClient, #[cfg(any())] puzzle_gallery_cache: PuzzleGalleryCache, tracking_outbox: Option>, wallet_refund_outbox: Option>, profile_wallet_refund_outbox_worker: Arc, editor_generation_pricing_store: EditorGenerationPricingStore, llm_client: Option, vector_engine_llm_client: Option, matting_client: Option, bgfilter_provider_http_client: reqwest::Client, bgfilter_worker_http_client: reqwest::Client, bgfilter_image_validation_limiter: Arc, character_animation_oss_http_client: reqwest::Client, character_animation_oss_io_limiter: Arc, editor_oss_http_client: reqwest::Client, #[cfg(any())] creative_agent_executor: Arc, // Phase 1 任务 E 的 creative session facade 暂存在 api-server。 // creative_agent_* 表由任务 D 收口后，这里只保留读写 facade。 #[cfg(any())] creative_agent_sessions: Arc>>, profile_recharge_order_updates: broadcast::Sender, #[cfg(any())] // 测试环境允许在未启动 SpacetimeDB 时，用内存快照兜底当前 runtime story 回归链。 test_runtime_snapshot_store: Arc>>, } impl fmt::Debug for AppStateInner { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { // AppState 聚合多个仍可独立 Debug 的平台 client。这里使用封闭摘要，禁止 Debug // 递归下钻到 JWT、管理员口令、OSS、微信、LLM 等运行时凭据。 f.debug_struct("AppStateInner") .field("config", &self.config) .field("ready", &self.ready.load(Ordering::Relaxed)) .field( "bgfilter_worker_reached", &self.bgfilter_worker_reached.load(Ordering::Relaxed), ) .field("admin_runtime_enabled", &self.admin_runtime.is_some()) .field("oss_client_enabled", &self.oss_client.is_some()) .field("spacetime_client", &self.spacetime_client) .field("tracking_outbox_enabled", &self.tracking_outbox.is_some()) .field( "wallet_refund_outbox_enabled", &self.wallet_refund_outbox.is_some(), ) .field("llm_client_enabled", &self.llm_client.is_some()) .field( "vector_engine_llm_client_enabled", &self.vector_engine_llm_client.is_some(), ) .field("matting_client_enabled", &self.matting_client.is_some()) .finish_non_exhaustive() } } #[derive(Clone, Debug)] #[cfg(any())] struct CreativeAgentSessionRuntimeRecord { owner_user_id: String, snapshot: CreativeAgentSessionSnapshot, } // 后台管理员运行态独立于普通玩家登录体系，只从环境变量构造。 #[derive(Clone, Debug)] pub struct AdminRuntime { username: Arc, password: Arc, subject: Arc, display_name: Arc, token_ttl_seconds: u64, jwt_config: JwtConfig, } #[derive(Clone, Debug)] pub struct AdminClaims { pub subject: String, pub username: String, pub display_name: String, pub roles: Vec, pub token_version: u64, pub issued_at: OffsetDateTime, pub expires_at: OffsetDateTime, } #[derive(Clone, Debug)] pub struct AdminSession { pub subject: String, pub username: String, pub display_name: String, pub roles: Vec, pub account_role: String, pub tab_permissions: Vec, pub action_permissions: Vec, pub issued_at: OffsetDateTime, pub expires_at: OffsetDateTime, } #[derive(Debug)] pub enum AppStateInitError { Jwt(JwtError), RefreshCookie(RefreshCookieError), AuthStore(String), DependencyUnavailable(String), SmsProvider(SmsProviderError), WechatPay(String), Oss(OssError), Llm(LlmError), } fn editor_generation_pricing_to_records( config: &EditorGenerationPricingConfig, ) -> Result, EditorGenerationPricingError> { config.validate()?; Ok(config .models .iter() .map(|(model, pricing)| EditorGenerationModelPricingRecord { model: model.clone(), unit: match pricing.unit { EditorGenerationPricingUnit::PerGeneration => "perGeneration", EditorGenerationPricingUnit::PerSecond => "perSecond", } .to_string(), price: pricing.price, prices: pricing .prices .iter() .map(|(key, price)| EditorGenerationPricingTierRecord { key: key.clone(), price: *price, }) .collect(), }) .collect()) } fn editor_generation_pricing_from_record( record: EditorGenerationPricingConfigRecord, legacy_fallback: &EditorGenerationPricingConfig, ) -> Result { let mut models = BTreeMap::new(); for pricing in record.models { let unit = match pricing.unit.as_str() { "perGeneration" => EditorGenerationPricingUnit::PerGeneration, "perSecond" => EditorGenerationPricingUnit::PerSecond, other => { return Err(EditorGenerationPricingError::Invalid(format!( "SpacetimeDB 模型定价配置 models.{}.unit 不合法：{other}", pricing.model ))); } }; let mut prices = BTreeMap::new(); for tier in pricing.prices { if prices.insert(tier.key.clone(), tier.price).is_some() { return Err(EditorGenerationPricingError::Invalid(format!( "SpacetimeDB 模型定价配置 models.{} 包含重复档位 {}", pricing.model, tier.key ))); } } let model = pricing.model; if models .insert( model.clone(), EditorGenerationModelPricing { unit, price: pricing.price, prices, }, ) .is_some() { return Err(EditorGenerationPricingError::Invalid(format!( "SpacetimeDB 模型定价配置包含重复模型 {model}" ))); } } if !models.contains_key(EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS) { let pricing = legacy_fallback .models .get(EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS) .cloned() .ok_or_else(|| { EditorGenerationPricingError::Invalid(format!( "本地模型定价配置缺少模型 {EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS}" )) })?; models.insert(EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS.to_string(), pricing); } let config = EditorGenerationPricingConfig { models }; config.validate()?; Ok(config) } fn editor_generation_pricing_upsert_input( config: &AppConfig, admin_user_id: String, models: Vec, updated_at_micros: i64, ) -> EditorGenerationPricingConfigUpsertRecordInput { EditorGenerationPricingConfigUpsertRecordInput { admin_user_id, models, updated_at_micros, bootstrap_secret: config .spacetime_runtime_service_bootstrap_secret .clone() .unwrap_or_default(), } } impl AppState { #[cfg(test)] pub fn new(config: AppConfig) -> Result { Self::new_with_empty_auth_store(config) } pub fn new_with_empty_auth_store(config: AppConfig) -> Result { // 中文注释：api-server 不再把本地 auth-store.json 当作用户认证真相源，启动恢复只允许来自 SpacetimeDB。 Self::new_with_auth_store(config, InMemoryAuthStore::default(), 0) } fn new_with_auth_store( config: AppConfig, auth_store: InMemoryAuthStore, auth_projection_version: i64, ) -> Result { let auth_jwt_config = JwtConfig::new( config.jwt_issuer.clone(), config.jwt_secret.clone(), config.jwt_access_token_ttl_seconds, )?; let admin_runtime = build_admin_runtime(&config, &auth_jwt_config)?; let refresh_cookie_same_site = RefreshCookieSameSite::parse(&config.refresh_cookie_same_site).ok_or( RefreshCookieError::InvalidConfig("refresh cookie SameSite 取值非法"), )?; let refresh_cookie_config = RefreshCookieConfig::new( config.refresh_cookie_name.clone(), config.refresh_cookie_path.clone(), config.refresh_cookie_secure, refresh_cookie_same_site, config.refresh_session_ttl_days, )?; let oss_client = build_oss_client(&config)?; let sms_provider = SmsAuthProvider::new(SmsAuthConfig::new( SmsAuthProviderKind::parse(&config.sms_auth_provider).ok_or_else(|| { SmsProviderError::InvalidConfig("短信 provider 配置非法".to_string()) })?, config.sms_endpoint.clone(), config.sms_access_key_id.clone(), config.sms_access_key_secret.clone(), config.sms_sign_name.clone(), config.sms_template_code.clone(), config.sms_template_param_key.clone(), config.sms_country_code.clone(), config.sms_scheme_name.clone(), config.sms_code_length, config.sms_code_type, config.sms_valid_time_seconds, config.sms_interval_seconds, config.sms_duplicate_policy, config.sms_case_auth_policy, config.sms_return_verify_code, config.sms_mock_verify_code.clone(), )?)?; let password_entry_service = PasswordEntryService::new(auth_store.clone()); let auth_user_service = AuthUserService::new(auth_store.clone()); auth_user_service .ensure_orphan_work_owner_user( ORPHAN_WORK_OWNER_USER_ID, ORPHAN_WORK_OWNER_USER_ID, ORPHAN_WORK_AUTHOR_DISPLAY_NAME, ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE, ) .map_err(|error| AppStateInitError::AuthStore(error.to_string()))?; let phone_auth_service = PhoneAuthService::new_with_verify_code_salt( auth_store.clone(), sms_provider, config.jwt_secret.clone(), ); let wechat_auth_state_service = WechatAuthStateService::new(auth_store.clone(), config.wechat_state_ttl_minutes); let wechat_auth_service = WechatAuthService::new(auth_store.clone()); let wechat_provider = build_wechat_provider(&config); let wechat_client = build_wechat_client(&config); let wechat_pay_client = WechatPayClient::from_config(&build_wechat_pay_config(&config)) .map_err(map_wechat_pay_init_error)?; let refresh_session_service = RefreshSessionService::new(auth_store.clone(), config.refresh_session_ttl_days); // AI 编排服务当前先挂接内存态 store，后续再按 task table / procedure 接到 SpacetimeDB 真相源。 let ai_task_service = AiTaskService::new(InMemoryAiTaskStore::default()); let spacetime_client = SpacetimeClient::new(spacetime_client_config_for_process(&config)); let tracking_outbox = TrackingOutbox::from_config(&config, spacetime_client.clone()); let wallet_refund_outbox = WalletRefundOutbox::from_config(&config, spacetime_client.clone()); let profile_wallet_refund_outbox_worker = ProfileWalletRefundOutboxWorker::from_config(&config, spacetime_client.clone()); let editor_generation_pricing_store = EditorGenerationPricingStore::load( config.editor_generation_pricing_override_path.clone(), ) .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?; let llm_client = build_llm_client(&config)?; let vector_engine_llm_client = build_vector_engine_llm_client(&config)?; let matting_client = build_matting_client(&config)?; let bgfilter_provider_http_client = build_bgfilter_provider_http_client(&config)?; let bgfilter_worker_http_client = build_bgfilter_worker_http_client(&config)?; let bgfilter_image_validation_concurrency = if config.process_role.runs_bgfilter_worker() { // 子 worker 已由 provider N 限流；图片校验槽与 N 对齐，避免引入第二个隐藏吞吐上限。 config.bgfilter_worker_concurrency.max(1) } else { // 父 API / external-generation worker 固定限制解码并发，避免动画响应同时进入 blocking pool。 BGFILTER_IMAGE_VALIDATION_MAX_CONCURRENCY }; let bgfilter_image_validation_limiter = Arc::new(Semaphore::new( bgfilter_image_validation_concurrency.min(Semaphore::MAX_PERMITS), )); let character_animation_oss_http_client = build_character_animation_oss_http_client()?; let character_animation_oss_io_limiter = Arc::new(Semaphore::new(CHARACTER_ANIMATION_OSS_MAX_CONCURRENCY)); let editor_oss_http_client = build_editor_oss_http_client()?; let http_request_permit_pools = HttpRequestPermitPools::from_config(&config); let (profile_recharge_order_updates, _) = broadcast::channel(128); // `ensure_orphan_work_owner_user` 只为公开作品作者回退提供进程内占位账号， // 不属于正式认证投影；将当前工作集 revision 作为已同步起点，首次认证请求会 // 先按正式投影刷新并自然丢弃该占位账号，避免把它误当成待提交认证变更。 let initial_auth_store_revision = auth_store.revision(); Ok(Self(Arc::new(AppStateInner { config, ready: AtomicBool::new(true), bgfilter_worker_reached: AtomicBool::new(false), http_request_permit_pools, auth_jwt_config, admin_runtime, refresh_cookie_config, #[cfg(any())] test_creation_entry_config: Arc::new(Mutex::new(Some( crate::creation_entry_config::test_creation_entry_config_response(), ))), #[cfg(test)] test_feature_gate_config: Arc::new(Mutex::new(Some(vec![]))), #[cfg(test)] test_spacetime_health: Arc::new(Mutex::new(Some( SpacetimeClientHealthSnapshot::healthy_for_test(), ))), #[cfg(test)] test_editor_generation_enqueue_attempts: AtomicUsize::new(0), #[cfg(test)] test_fail_editor_generation_enqueue: AtomicBool::new(false), #[cfg(test)] test_external_background_removal_enqueue: Arc::new(Mutex::new(None)), oss_client, auth_store, auth_projection_version: AtomicI64::new(auth_projection_version), auth_projection_synced_revision: AtomicU64::new(initial_auth_store_revision), auth_projection_sync_lock: AsyncMutex::new(()), password_entry_service, refresh_session_service, auth_user_service, phone_auth_service, wechat_auth_state_service, wechat_auth_service, wechat_provider, wechat_client, wechat_pay_client, ai_task_service, spacetime_client, #[cfg(any())] puzzle_gallery_cache: PuzzleGalleryCache::new(), tracking_outbox, wallet_refund_outbox, profile_wallet_refund_outbox_worker, editor_generation_pricing_store, llm_client, vector_engine_llm_client, matting_client, bgfilter_provider_http_client, bgfilter_worker_http_client, bgfilter_image_validation_limiter, character_animation_oss_http_client, character_animation_oss_io_limiter, editor_oss_http_client, #[cfg(any())] creative_agent_executor: Arc::new(MockLangChainRustAgentExecutor), #[cfg(any())] creative_agent_sessions: Arc::new(Mutex::new(HashMap::new())), profile_recharge_order_updates, #[cfg(any())] test_runtime_snapshot_store: Arc::new(Mutex::new(HashMap::new())), }))) } pub fn auth_jwt_config(&self) -> &JwtConfig { &self.auth_jwt_config } pub fn admin_runtime(&self) -> Option<&AdminRuntime> { self.admin_runtime.as_ref() } pub fn refresh_cookie_config(&self) -> &RefreshCookieConfig { &self.refresh_cookie_config } pub fn http_request_permit_pools(&self) -> HttpRequestPermitPools { self.http_request_permit_pools.clone() } pub(crate) async fn editor_generation_pricing( &self, ) -> Result { #[cfg(test)] { return self.editor_generation_pricing_store.snapshot(); } #[cfg(not(test))] match self .spacetime_client .get_editor_generation_pricing_config() .await { Ok(Some(record)) => { let legacy_fallback = self.editor_generation_pricing_store.snapshot()?; let pricing = editor_generation_pricing_from_record(record, &legacy_fallback)?; self.editor_generation_pricing_store .replace(pricing.clone())?; Ok(pricing) } Ok(None) => self.seed_editor_generation_pricing_config().await, Err(error) => { warn!( error = %error, "读取 SpacetimeDB 模型定价配置失败，使用本地缓存兜底" ); self.editor_generation_pricing_store.snapshot() } } } pub(crate) async fn save_editor_generation_pricing( &self, admin_user_id: String, next: EditorGenerationPricingConfig, ) -> Result { #[cfg(test)] { let _ = admin_user_id; return self.editor_generation_pricing_store.replace(next); } #[cfg(not(test))] { let models = editor_generation_pricing_to_records(&next)?; let record = self .spacetime_client .upsert_editor_generation_pricing_config(editor_generation_pricing_upsert_input( &self.config, admin_user_id, models, crate::editor_project::current_utc_micros(), )) .await .map_err(|error| EditorGenerationPricingError::Persistence(error.to_string()))?; let pricing = editor_generation_pricing_from_record(record, &next)?; self.editor_generation_pricing_store .replace(pricing.clone())?; Ok(pricing) } } pub(crate) async fn ensure_editor_generation_runtime_service_identity( &self, ) -> Result<(), EditorGenerationPricingError> { self.seed_editor_generation_pricing_config().await?; Ok(()) } #[cfg_attr(test, allow(dead_code))] async fn seed_editor_generation_pricing_config( &self, ) -> Result { let fallback = self.editor_generation_pricing_store.snapshot()?; let models = editor_generation_pricing_to_records(&fallback)?; let record = self .spacetime_client .initialize_editor_generation_pricing_config_if_missing( editor_generation_pricing_upsert_input( &self.config, "system:editor-generation-pricing".to_string(), models, crate::editor_project::current_utc_micros(), ), ) .await .map_err(|error| EditorGenerationPricingError::Persistence(error.to_string()))?; let pricing = editor_generation_pricing_from_record(record, &fallback)?; self.editor_generation_pricing_store .replace(pricing.clone())?; Ok(pricing) } pub fn is_ready(&self) -> bool { self.ready.load(Ordering::Acquire) } pub fn mark_not_ready(&self) { self.ready.store(false, Ordering::Release); } pub async fn spacetime_health_check(&self) -> SpacetimeClientHealthSnapshot { #[cfg(test)] if let Some(snapshot) = self .test_spacetime_health .lock() .expect("test spacetime health should lock") .clone() { return snapshot; } self.spacetime_client .health_check(self.config.spacetime_health_check_timeout) .await } #[cfg(test)] pub(crate) fn set_test_spacetime_health(&self, snapshot: SpacetimeClientHealthSnapshot) { *self .test_spacetime_health .lock() .expect("test spacetime health should lock") = Some(snapshot); } #[cfg(test)] pub(crate) fn fail_test_editor_generation_enqueue(&self) { self.test_fail_editor_generation_enqueue .store(true, Ordering::Release); } #[cfg(test)] pub(crate) fn record_test_editor_generation_enqueue_attempt(&self) -> bool { self.test_editor_generation_enqueue_attempts .fetch_add(1, Ordering::AcqRel); self.test_fail_editor_generation_enqueue .load(Ordering::Acquire) } #[cfg(test)] pub(crate) fn test_editor_generation_enqueue_attempts(&self) -> usize { self.test_editor_generation_enqueue_attempts .load(Ordering::Acquire) } #[cfg(test)] pub(crate) fn set_test_external_background_removal_enqueue( &self, expected_owner_user_id: impl Into, expected_source_image_src: impl Into, expected_idempotency_key: impl Into, job: ExternalGenerationJobRecord, ) { *self .test_external_background_removal_enqueue .lock() .expect("test external background removal enqueue should lock") = Some(TestExternalBackgroundRemovalEnqueue { expected_owner_user_id: expected_owner_user_id.into(), expected_source_image_src: expected_source_image_src.into(), expected_idempotency_key: expected_idempotency_key.into(), job, }); } #[cfg(test)] pub(crate) fn intercept_test_external_background_removal_enqueue( &self, owner_user_id: &str, source_image_src: &str, idempotency_key: &str, ) -> Option { let fixture = self .test_external_background_removal_enqueue .lock() .expect("test external background removal enqueue should lock") .clone()?; assert_eq!(owner_user_id, fixture.expected_owner_user_id); assert_eq!(source_image_src, fixture.expected_source_image_src); assert_eq!(idempotency_key, fixture.expected_idempotency_key); self.test_editor_generation_enqueue_attempts .fetch_add(1, Ordering::AcqRel); Some(fixture.job) } #[cfg(any())] pub async fn upsert_creation_entry_type_config( &self, input: module_runtime::CreationEntryTypeAdminUpsertInput, ) -> Result { match self .spacetime_client .upsert_creation_entry_type_config(input) .await { Ok(config) => { #[cfg(test)] self.cache_test_creation_entry_config(config.clone()); Ok(config) } #[cfg(test)] Err(_) => Ok(self.read_test_creation_entry_config()), #[cfg(not(test))] Err(error) => Err(error), } } /// 通过 SpacetimeDB 保存创作入口页多公告配置，并同步测试缓存。 #[cfg(any())] pub async fn upsert_creation_entry_event_banners_config( &self, input: module_runtime::CreationEntryEventBannersAdminUpsertInput, ) -> Result { #[cfg(test)] let test_event_banners_json = input.event_banners_json.clone(); match self .spacetime_client .upsert_creation_entry_event_banners_config(input) .await { Ok(config) => { #[cfg(test)] self.cache_test_creation_entry_config(config.clone()); Ok(config) } #[cfg(test)] Err(_) => { let mut config = self.read_test_creation_entry_config(); if let Ok(banners) = module_runtime::decode_creation_entry_event_banner_snapshots( test_event_banners_json.as_str(), ) { config.event_banners = banners .into_iter() .map(module_runtime::build_creation_entry_event_banner_response) .collect(); if let Some(first_banner) = config.event_banners.first().cloned() { config.event_banner = first_banner; } self.cache_test_creation_entry_config(config.clone()); } Ok(config) } #[cfg(not(test))] Err(error) => Err(error), } } /// 通过 SpacetimeDB 保存公开作品互动配置，并同步测试缓存。 #[cfg(any())] pub async fn upsert_public_work_interaction_config( &self, input: module_runtime::PublicWorkInteractionConfigAdminUpsertInput, ) -> Result { #[cfg(test)] let test_interactions_json = input.public_work_interactions_json.clone(); match self .spacetime_client .upsert_public_work_interaction_config(input) .await { Ok(config) => { #[cfg(test)] self.cache_test_creation_entry_config(config.clone()); Ok(config) } #[cfg(test)] Err(_) => { let mut config = self.read_test_creation_entry_config(); if let Ok(interactions) = module_runtime::decode_public_work_interaction_config_snapshots( test_interactions_json.as_str(), ) { config.public_work_interactions = interactions .into_iter() .map(module_runtime::build_public_work_interaction_config_response) .collect(); self.cache_test_creation_entry_config(config.clone()); } Ok(config) } #[cfg(not(test))] Err(error) => Err(error), } } #[cfg(any())] pub async fn get_creation_entry_config( &self, ) -> Result { match self.spacetime_client.get_creation_entry_config().await { Ok(config) => { #[cfg(test)] self.cache_test_creation_entry_config(config.clone()); Ok(config) } #[cfg(debug_assertions)] Err(error) if is_missing_creation_entry_config_procedure(&error) => { warn!( error = %error, "本地 SpacetimeDB 缺少创作入口配置 procedure，使用后端默认入口配置兜底" ); Ok(crate::creation_entry_config::default_creation_entry_config_response()) } #[cfg(test)] Err(_) => Ok(self.read_test_creation_entry_config()), #[cfg(not(test))] Err(error) => Err(error), } } pub async fn get_feature_gate_config( &self, ) -> Result, SpacetimeClientError> { match self.spacetime_client.get_feature_gate_config().await { Ok(config) => { #[cfg(test)] self.cache_test_feature_gate_config(config.clone()); Ok(config) } #[cfg(debug_assertions)] Err(error) if is_missing_feature_gate_config_procedure(&error) => { warn!( error = %error, "本地 SpacetimeDB 缺少灰度配置 procedure，使用空灰度配置兜底" ); Ok(vec![]) } #[cfg(test)] Err(_) => Ok(self.read_test_feature_gate_config()), #[cfg(not(test))] Err(error) => Err(error), } } pub async fn upsert_feature_gate_config( &self, input: module_runtime::FeatureGateConfigAdminUpsertInput, ) -> Result, SpacetimeClientError> { #[cfg(test)] let test_input = input.clone(); match self .spacetime_client .upsert_feature_gate_config(input) .await { Ok(config) => { #[cfg(test)] self.cache_test_feature_gate_config(config.clone()); Ok(config) } #[cfg(test)] Err(_) => { let normalized = module_runtime::normalize_feature_gate_admin_upsert_input(test_input) .map_err(|error| SpacetimeClientError::Runtime(error.to_string()))?; let mut config = self.read_test_feature_gate_config(); let now_micros = crate::editor_project::current_utc_micros(); let record = module_runtime::FeatureGateConfigSnapshot { gate_key: normalized.gate_key.clone(), enabled: normalized.enabled, rollout_percent: normalized.rollout_percent, allow_user_ids: normalized.allow_user_ids, allow_user_tags: normalized.allow_user_tags, deny_user_ids: normalized.deny_user_ids, description: normalized.description, updated_at_micros: now_micros, }; if let Some(existing) = config .iter_mut() .find(|item| item.gate_key == normalized.gate_key) { *existing = record; } else { config.push(record); } config.sort_by(|left, right| left.gate_key.cmp(&right.gate_key)); self.cache_test_feature_gate_config(config.clone()); Ok(config) } #[cfg(not(test))] Err(error) => Err(error), } } #[cfg(any())] pub async fn get_creation_entry_config_for_user( &self, user_id: Option<&str>, ) -> Result { let config = self.get_creation_entry_config().await?; let gates = self.get_feature_gate_config().await?; let user_context = self .feature_gate_user_context( user_id, creation_entry_feature_gates_require_user_tags(&config, &gates), ) .await; Ok( module_runtime::apply_feature_gates_to_creation_entry_config( config, &gates, &user_context, ), ) } pub async fn is_image_editor_agent_sidebar_enabled_for_user( &self, user_id: Option<&str>, ) -> Result { if !self.config.image_editor_agent_sidebar_enabled { return Ok(false); } let gates = self.get_feature_gate_config().await?; let gate = gates .iter() .find(|item| item.gate_key == module_runtime::IMAGE_EDITOR_AGENT_SIDEBAR_GATE_KEY); let user_context = self .feature_gate_user_context( user_id, gate.map(feature_gate_requires_user_tags).unwrap_or(false), ) .await; Ok(module_runtime::is_feature_gate_allowed(gate, &user_context)) } #[cfg(any())] pub async fn list_admin_work_visibility( &self, admin_user_id: String, ) -> Result, SpacetimeClientError> { self.spacetime_client .admin_list_work_visibility(admin_user_id) .await } #[cfg(any())] pub async fn update_admin_work_visibility( &self, admin_user_id: String, source_type: String, profile_id: String, visible: bool, ) -> Result { self.spacetime_client .admin_update_work_visibility(admin_user_id, source_type, profile_id, visible) .await } #[cfg(any())] pub async fn is_creation_entry_route_enabled_for_user( &self, creation_type_id: &str, user_id: Option<&str>, ) -> Result { let config = self.get_creation_entry_config_for_user(user_id).await?; Ok(config .creation_types .iter() .find(|item| item.id == creation_type_id) .map(|item| item.open) .unwrap_or(true)) } async fn feature_gate_user_context( &self, user_id: Option<&str>, include_user_tags: bool, ) -> module_runtime::FeatureGateUserContext { let user_id = user_id .map(str::trim) .filter(|value| !value.is_empty()) .map(str::to_string); let user_tags = if include_user_tags { match user_id.as_ref() { Some(user_id) => match self.spacetime_client.get_user_tags(user_id.clone()).await { Ok(tags) => tags, Err(error) => { warn!( user_id = %user_id, error = %error, "读取灰度用户标签失败，按无标签继续判定" ); vec![] } }, None => vec![], } } else { vec![] }; module_runtime::FeatureGateUserContext { user_id, user_tags } } #[cfg(any())] pub async fn is_public_work_interaction_enabled( &self, source_type: &str, action: crate::creation_entry_config::PublicWorkInteractionAction, ) -> Result { let config = self.get_creation_entry_config().await?; Ok(config .public_work_interactions .iter() .find(|item| item.source_type == source_type) .map(|item| match action { crate::creation_entry_config::PublicWorkInteractionAction::Like => { item.like_enabled } crate::creation_entry_config::PublicWorkInteractionAction::Remix => { item.remix_enabled } }) .unwrap_or(true)) } #[cfg(any())] pub(crate) fn set_test_public_work_interaction_enabled( &self, source_type: impl AsRef, action: crate::creation_entry_config::PublicWorkInteractionAction, enabled: bool, ) { let source_type = source_type.as_ref(); let mut config = self.read_test_creation_entry_config(); if let Some(item) = config .public_work_interactions .iter_mut() .find(|item| item.source_type == source_type) { match action { crate::creation_entry_config::PublicWorkInteractionAction::Like => { item.like_enabled = enabled; } crate::creation_entry_config::PublicWorkInteractionAction::Remix => { item.remix_enabled = enabled; } } } self.cache_test_creation_entry_config(config); } #[cfg(any())] pub(crate) fn set_test_creation_entry_route_enabled( &self, creation_type_id: impl AsRef, enabled: bool, ) { let creation_type_id = creation_type_id.as_ref(); let mut config = self.read_test_creation_entry_config(); if let Some(item) = config .creation_types .iter_mut() .find(|item| item.id == creation_type_id) { item.open = enabled; } else { config.creation_types.push( shared_contracts::creation_entry_config::CreationEntryTypeResponse { id: creation_type_id.to_string(), title: creation_type_id.to_string(), subtitle: String::new(), badge: String::new(), image_src: format!("/creation-type-references/{creation_type_id}.webp"), visible: enabled, open: enabled, sort_order: i32::try_from(config.creation_types.len()).unwrap_or(i32::MAX), category_id: module_runtime::DEFAULT_CREATION_ENTRY_CATEGORY_ID.to_string(), category_label: module_runtime::DEFAULT_CREATION_ENTRY_CATEGORY_LABEL .to_string(), category_sort_order: 0, updated_at_micros: 0, unified_creation_spec: shared_contracts::creation_entry_config::build_phase1_unified_creation_spec( creation_type_id, ), }, ); } self.cache_test_creation_entry_config(config); } #[cfg(test)] pub(crate) fn set_test_feature_gate_config( &self, config: Vec, ) { self.cache_test_feature_gate_config(config); } pub fn oss_client(&self) -> Option<&OssClient> { self.oss_client.as_ref() } pub fn password_entry_service(&self) -> &PasswordEntryService { &self.password_entry_service } pub async fn sync_auth_store_tables_to_spacetime(&self) -> Result<(), SpacetimeClientError> { #[cfg(test)] { self.auth_projection_synced_revision .store(self.auth_store.revision(), Ordering::Release); return Ok(()); } #[cfg(not(test))] let _sync_guard = self.auth_projection_sync_lock.lock().await; #[cfg(not(test))] for attempt in 0..3 { let base_updated_at_micros = self.auth_projection_version.load(Ordering::Acquire); let now_updated_at_micros = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000).map_err( |_| SpacetimeClientError::Runtime("认证状态更新时间超出 i64 范围".to_string()), )?; let updated_at_micros = if now_updated_at_micros > base_updated_at_micros { now_updated_at_micros } else { base_updated_at_micros.checked_add(1).ok_or_else(|| { SpacetimeClientError::Runtime("认证状态版本超出 i64 范围".to_string()) })? }; let (mut projection, attempted_revision) = self .auth_store .export_projection_view_with_revision(updated_at_micros) .map_err(SpacetimeClientError::Runtime)?; projection.base_updated_at_micros = base_updated_at_micros; // 当前仍由 module-auth 的进程内工作集执行业务规则；这里只用 typed projection 同步正式认证表。 match self .spacetime_client .sync_auth_store_projection(projection) .await { Ok(_) => { self.auth_projection_version .store(updated_at_micros, Ordering::Release); if self.auth_store.revision() == attempted_revision { self.auth_projection_synced_revision .store(attempted_revision, Ordering::Release); return Ok(()); } warn!( attempt, "认证投影同步期间工作集发生变化，将继续同步最新工作集" ); continue; } Err(error) => { warn!( error = %error, "认证投影同步 SpacetimeDB 正式表失败，当前认证流程中止" ); // 当前请求已经失败；只要同步尝试期间没有新的本地变更，恢复为 // 数据库快照，避免一次 CAS 冲突把本节点永久留在“待同步”状态。 if self.auth_store.revision() != attempted_revision { warn!( "认证投影同步失败期间工作集发生并发变化，跳过自动恢复以避免覆盖未提交变更" ); } else if let Ok(current_projection) = self .spacetime_client .export_auth_store_projection_from_tables() .await { match self.auth_store.refresh_from_projection_view_if_revision( current_projection.clone(), attempted_revision, ) { Ok(true) => { self.auth_projection_version .store(current_projection.updated_at_micros, Ordering::Release); self.auth_projection_synced_revision .store(self.auth_store.revision(), Ordering::Release); } Ok(false) => { warn!( "认证投影同步冲突期间工作集发生并发变化，跳过自动恢复以避免覆盖未提交变更" ); } Err(refresh_error) => { warn!( error = %refresh_error, "认证投影同步冲突后恢复进程内工作集失败" ); } } } return Err(error); } } } #[cfg(not(test))] Err(SpacetimeClientError::Runtime( "认证工作集在同步期间持续发生变化，未能完成投影同步".to_string(), )) } /// 在认证主链路执行前，从正式投影刷新一次本地工作集，避免请求落到另一节点后 /// 因本机工作集滞后而必须依赖粘性会话才能成功。 pub async fn refresh_auth_store_from_spacetime(&self) -> Result<(), SpacetimeClientError> { #[cfg(test)] return Ok(()); #[cfg(not(test))] { // 上一次业务操作可能已经改了工作集，但在返回响应前遇到数据库暂时不可用。 // 先重试提交这份待同步变更，避免只读请求把节点永久卡在 pending 状态。 if self.auth_projection_synced_revision.load(Ordering::Acquire) != self.auth_store.revision() { self.sync_auth_store_tables_to_spacetime().await?; } let _sync_guard = self.auth_projection_sync_lock.lock().await; let expected_revision = self.auth_store.revision(); if self.auth_projection_synced_revision.load(Ordering::Acquire) != expected_revision { return Err(SpacetimeClientError::Runtime( "认证工作集存在待同步变更，跳过只读刷新".to_string(), )); } let projection = self .spacetime_client .export_auth_store_projection_from_tables() .await?; let updated_at_micros = projection.updated_at_micros; let refreshed = self .auth_store .refresh_from_projection_view_if_revision(projection, expected_revision) .map_err(SpacetimeClientError::Runtime)?; if !refreshed { return Err(SpacetimeClientError::Runtime( "认证工作集刷新期间发生并发变更".to_string(), )); } self.auth_projection_version .store(updated_at_micros, Ordering::Release); self.auth_projection_synced_revision .store(self.auth_store.revision(), Ordering::Release); Ok(()) } } pub async fn try_restore_auth_store_from_spacetime( config: AppConfig, ) -> Result { let spacetime_client = SpacetimeClient::new(spacetime_client_config_for_startup_restore(&config)); initialize_editor_generation_runtime_service_identity_for_startup( &config, &spacetime_client, ) .await?; let mut spacetime_restore_available = false; let mut restore_errors = Vec::new(); match spacetime_client .export_auth_store_projection_from_tables() .await { Ok(projection) => { spacetime_restore_available = true; if let Some(candidate) = auth_store_candidate_from_projection_view( projection, AuthStoreRestoreSource::SpacetimeTables, )? { let state = Self::new_with_auth_store( config, candidate.auth_store, candidate.updated_at_micros.unwrap_or_default(), )?; info!( source = candidate.source.as_str(), updated_at_micros = candidate.updated_at_micros, "已恢复认证投影" ); return Ok(state); } } Err(error) => { warn!(error = %error, "从 SpacetimeDB 表恢复认证投影失败"); restore_errors.push(error.to_string()); } } if !spacetime_restore_available { return Err(AppStateInitError::DependencyUnavailable(format!( "SpacetimeDB 认证投影恢复不可用：{}", restore_errors.join("; ") ))); } Self::new_with_empty_auth_store(config) } pub fn refresh_session_service(&self) -> &RefreshSessionService { &self.refresh_session_service } pub fn auth_user_service(&self) -> &AuthUserService { &self.auth_user_service } pub fn phone_auth_service(&self) -> &PhoneAuthService { &self.phone_auth_service } pub fn wechat_auth_state_service(&self) -> &WechatAuthStateService { &self.wechat_auth_state_service } pub fn wechat_auth_service(&self) -> &WechatAuthService { &self.wechat_auth_service } pub fn wechat_provider(&self) -> &WechatProvider { &self.wechat_provider } pub fn wechat_client(&self) -> &WechatClient { &self.wechat_client } pub fn wechat_pay_client(&self) -> &WechatPayClient { &self.wechat_pay_client } pub fn wechat_pay_refund_reconciliation_enabled(&self) -> bool { self.config.wechat_pay_refund_reconciliation_enabled } #[cfg_attr(not(test), allow(dead_code))] pub fn ai_task_service(&self) -> &AiTaskService { &self.ai_task_service } pub fn spacetime_client(&self) -> &SpacetimeClient { &self.spacetime_client } #[cfg(any())] pub fn puzzle_gallery_cache(&self) -> &PuzzleGalleryCache { &self.puzzle_gallery_cache } pub fn tracking_outbox(&self) -> Option> { self.tracking_outbox.clone() } pub fn wallet_refund_outbox(&self) -> Option> { self.wallet_refund_outbox.clone() } pub fn profile_wallet_refund_outbox_worker(&self) -> Arc { self.profile_wallet_refund_outbox_worker.clone() } pub fn llm_client(&self) -> Option<&LlmClient> { self.llm_client.as_ref() } pub fn vector_engine_llm_client(&self) -> Option<&LlmClient> { self.vector_engine_llm_client.as_ref() } pub fn matting_client(&self) -> Option<&MattingClient> { self.matting_client.as_ref() } pub fn bgfilter_provider_http_client(&self) -> &reqwest::Client { &self.bgfilter_provider_http_client } pub fn bgfilter_worker_http_client(&self) -> &reqwest::Client { &self.bgfilter_worker_http_client } pub fn bgfilter_worker_reached(&self) -> bool { self.bgfilter_worker_reached.load(Ordering::Relaxed) } pub fn mark_bgfilter_worker_reached(&self) { self.bgfilter_worker_reached.store(true, Ordering::Relaxed); } pub fn bgfilter_image_validation_limiter(&self) -> Arc { self.bgfilter_image_validation_limiter.clone() } pub fn character_animation_oss_http_client(&self) -> &reqwest::Client { &self.character_animation_oss_http_client } pub fn character_animation_oss_io_limiter(&self) -> Arc { self.character_animation_oss_io_limiter.clone() } pub fn editor_oss_http_client(&self) -> &reqwest::Client { &self.editor_oss_http_client } #[cfg(any())] pub fn creative_agent_executor(&self) -> Arc { self.creative_agent_executor.clone() } pub fn subscribe_profile_recharge_order_updates( &self, ) -> tokio::sync::broadcast::Receiver { self.profile_recharge_order_updates.subscribe() } pub fn publish_profile_recharge_order_update(&self, order_id: impl Into) { let _ = self.profile_recharge_order_updates.send(order_id.into()); } #[cfg(any())] pub fn get_creative_agent_session( &self, session_id: &str, owner_user_id: &str, ) -> Option { self.creative_agent_sessions .lock() .expect("creative agent session store should lock") .get(session_id) .filter(|record| record.owner_user_id == owner_user_id) .map(|record| record.snapshot.clone()) } #[cfg(any())] pub fn put_creative_agent_session( &self, owner_user_id: String, session: CreativeAgentSessionSnapshot, ) { self.creative_agent_sessions .lock() .expect("creative agent session store should lock") .insert( session.session_id.clone(), CreativeAgentSessionRuntimeRecord { owner_user_id, snapshot: session, }, ); } #[cfg(any())] pub async fn get_runtime_snapshot_record( &self, user_id: String, ) -> Result, SpacetimeClientError> { match self .spacetime_client .get_runtime_snapshot(user_id.clone()) .await { Ok(record) => { #[cfg(test)] if let Some(snapshot) = record.as_ref() { self.cache_test_runtime_snapshot(snapshot.clone()); } Ok(record) } #[cfg(test)] Err(_) => Ok(self.read_test_runtime_snapshot(user_id.as_str())), #[cfg(not(test))] Err(error) => Err(error), } } #[cfg(any())] pub async fn put_runtime_snapshot_record( &self, user_id: String, saved_at_micros: i64, bottom_tab: String, game_state: Value, current_story: Option, updated_at_micros: i64, ) -> Result { match self .spacetime_client .put_runtime_snapshot( user_id.clone(), saved_at_micros, bottom_tab.clone(), game_state.clone(), current_story.clone(), updated_at_micros, ) .await { Ok(record) => { #[cfg(test)] self.cache_test_runtime_snapshot(record.clone()); Ok(record) } #[cfg(test)] Err(_) => { let snapshot = self.build_test_runtime_snapshot_record( user_id, saved_at_micros, bottom_tab, game_state, current_story, updated_at_micros, )?; self.cache_test_runtime_snapshot(snapshot.clone()); Ok(snapshot) } #[cfg(not(test))] Err(error) => Err(error), } } #[cfg(any())] pub async fn delete_runtime_snapshot_record( &self, user_id: String, ) -> Result { match self .spacetime_client .delete_runtime_snapshot(user_id.clone()) .await { Ok(deleted) => { #[cfg(test)] if deleted { self.remove_test_runtime_snapshot(user_id.as_str()); } Ok(deleted) } #[cfg(test)] Err(_) => Ok(self .remove_test_runtime_snapshot(user_id.as_str()) .is_some()), #[cfg(not(test))] Err(error) => Err(error), } } } fn feature_gate_requires_user_tags(gate: &module_runtime::FeatureGateConfigSnapshot) -> bool { gate.enabled && !gate.allow_user_tags.is_empty() } #[cfg(any())] fn creation_entry_feature_gates_require_user_tags( config: &CreationEntryConfigResponse, gates: &[module_runtime::FeatureGateConfigSnapshot], ) -> bool { let gate_keys = config .creation_types .iter() .map(|entry| module_runtime::creation_entry_feature_gate_key(&entry.id)) .collect::>(); gates .iter() .any(|gate| gate_keys.contains(&gate.gate_key) && feature_gate_requires_user_tags(gate)) } #[cfg(test)] impl AppState { pub(crate) fn test_auth_projection_is_synced(&self) -> bool { self.auth_projection_synced_revision.load(Ordering::Acquire) == self.auth_store.revision() } pub(crate) fn seed_test_refresh_session_for_user( &self, user: &module_auth::AuthUser, seed: &str, ) -> String { let session = self .refresh_session_service() .create_session( module_auth::CreateRefreshSessionInput { user_id: user.id.clone(), refresh_token_hash: platform_auth::hash_refresh_session_token(&format!( "test-refresh-token-{seed}" )), issued_by_provider: module_auth::AuthLoginMethod::Password, client_info: module_auth::RefreshSessionClientInfo { client_type: "web_browser".to_string(), client_runtime: "test".to_string(), client_platform: "test".to_string(), client_instance_id: Some(seed.to_string()), device_fingerprint: Some(format!("test-device-{seed}")), device_display_name: "Test Browser".to_string(), mini_program_app_id: None, mini_program_env: None, user_agent: Some("GenarrativeApiServerTest/1.0".to_string()), ip: Some("127.0.0.1".to_string()), }, }, OffsetDateTime::now_utc(), ) .expect("test refresh session should create"); session.session.session_id } pub(crate) fn seed_test_refresh_session_for_user_id( &self, user_id: &str, seed: &str, ) -> String { let user = self .auth_user_service() .get_user_by_id(user_id) .expect("test user lookup should succeed") .expect("test user should exist"); self.seed_test_refresh_session_for_user(&user, seed) } #[cfg(any())] fn cache_test_creation_entry_config(&self, config: CreationEntryConfigResponse) { *self .test_creation_entry_config .lock() .expect("test creation entry config should lock") = Some(config); } #[cfg(any())] fn read_test_creation_entry_config(&self) -> CreationEntryConfigResponse { self.test_creation_entry_config .lock() .expect("test creation entry config should lock") .clone() .unwrap_or_else(crate::creation_entry_config::test_creation_entry_config_response) } #[cfg(test)] fn cache_test_feature_gate_config( &self, config: Vec, ) { *self .test_feature_gate_config .lock() .expect("test feature gate config should lock") = Some(config); } #[cfg(test)] fn read_test_feature_gate_config(&self) -> Vec { self.test_feature_gate_config .lock() .expect("test feature gate config should lock") .clone() .unwrap_or_default() } pub(crate) async fn seed_test_phone_user_with_password( &self, phone_number: &str, password: &str, ) -> module_auth::AuthUser { let now = OffsetDateTime::now_utc(); self.phone_auth_service() .send_code( module_auth::SendPhoneCodeInput { country_code: None, pure_phone_number: phone_number.to_string(), scene: module_auth::PhoneAuthScene::Login, }, now, ) .await .expect("test phone code should send"); let user = self .phone_auth_service() .login( module_auth::PhoneLoginInput { country_code: None, pure_phone_number: phone_number.to_string(), verify_code: "123456".to_string(), }, now + time::Duration::seconds(1), ) .await .expect("test phone login should create user") .user; let changed = self .password_entry_service() .change_password(module_auth::ChangePasswordInput { user_id: user.id.clone(), current_password: None, new_password: password.to_string(), }) .await .expect("test password should set"); changed.user } #[cfg(any())] fn cache_test_runtime_snapshot(&self, record: RuntimeSnapshotRecord) { self.test_runtime_snapshot_store .lock() .expect("test runtime snapshot store should lock") .insert(record.user_id.clone(), record); } #[cfg(any())] fn read_test_runtime_snapshot(&self, user_id: &str) -> Option { self.test_runtime_snapshot_store .lock() .expect("test runtime snapshot store should lock") .get(user_id) .cloned() } #[cfg(any())] fn remove_test_runtime_snapshot(&self, user_id: &str) -> Option { self.test_runtime_snapshot_store .lock() .expect("test runtime snapshot store should lock") .remove(user_id) } #[cfg(any())] fn build_test_runtime_snapshot_record( &self, user_id: String, saved_at_micros: i64, bottom_tab: String, game_state: Value, current_story: Option, updated_at_micros: i64, ) -> Result { let previous = self.read_test_runtime_snapshot(user_id.as_str()); let game_state_json = serde_json::to_string(&game_state).map_err(|error| { SpacetimeClientError::Runtime(format!("测试快照 game_state 序列化失败: {error}")) })?; let current_story_json = current_story .as_ref() .map(serde_json::to_string) .transpose() .map_err(|error| { SpacetimeClientError::Runtime(format!("测试快照 current_story 序列化失败: {error}")) })?; Ok(RuntimeSnapshotRecord { user_id, version: SAVE_SNAPSHOT_VERSION, saved_at: format_utc_micros(saved_at_micros), saved_at_micros, bottom_tab, game_state, current_story, game_state_json, current_story_json, created_at_micros: previous .as_ref() .map(|record| record.created_at_micros) .unwrap_or(updated_at_micros), updated_at_micros, }) } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum AuthStoreRestoreSource { SpacetimeTables, } impl AuthStoreRestoreSource { fn as_str(self) -> &'static str { match self { Self::SpacetimeTables => "spacetime_tables", } } } #[derive(Debug)] struct AuthStoreRestoreCandidate { source: AuthStoreRestoreSource, updated_at_micros: Option, auth_store: InMemoryAuthStore, } fn auth_store_candidate_from_projection_view( projection: module_auth::AuthStoreProjectionView, source: AuthStoreRestoreSource, ) -> Result, AppStateInitError> { if projection.users.is_empty() && projection.identities.is_empty() && projection.refresh_sessions.is_empty() && projection.phone_codes.is_empty() && projection.wechat_states.is_empty() && projection.updated_at_micros == 0 { return Ok(None); } let updated_at_micros = Some(projection.updated_at_micros); let auth_store = InMemoryAuthStore::from_projection_view(projection) .map_err(AppStateInitError::AuthStore)?; Ok(Some(AuthStoreRestoreCandidate { source, updated_at_micros, auth_store, })) } fn spacetime_client_config_for_process(config: &AppConfig) -> SpacetimeClientConfig { let runs_http = config.process_role.runs_http(); SpacetimeClientConfig { server_url: config.spacetime_server_url.clone(), database: config.spacetime_database.clone(), token: config.spacetime_token.clone(), pool_size: if runs_http { config.spacetime_pool_size } else { 1 }, procedure_timeout: config.spacetime_procedure_timeout, subscribe_cached_read_models: runs_http, } } fn spacetime_client_config_for_startup_restore(config: &AppConfig) -> SpacetimeClientConfig { SpacetimeClientConfig { server_url: config.spacetime_server_url.clone(), database: config.spacetime_database.clone(), token: config.spacetime_token.clone(), pool_size: 1, procedure_timeout: config.spacetime_procedure_timeout, subscribe_cached_read_models: false, } } async fn initialize_editor_generation_runtime_service_identity_for_startup( config: &AppConfig, spacetime_client: &SpacetimeClient, ) -> Result<(), AppStateInitError> { let pricing_store = EditorGenerationPricingStore::load(config.editor_generation_pricing_override_path.clone()) .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?; let fallback = pricing_store .snapshot() .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?; let models = editor_generation_pricing_to_records(&fallback) .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?; spacetime_client .initialize_editor_generation_pricing_config_if_missing( editor_generation_pricing_upsert_input( config, "system:editor-generation-pricing".to_string(), models, crate::editor_project::current_utc_micros(), ), ) .await .map_err(|error| { AppStateInitError::DependencyUnavailable(format!("初始化模型定价服务身份失败：{error}")) })?; Ok(()) } impl fmt::Display for AppStateInitError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Jwt(error) => write!(f, "{error}"), Self::RefreshCookie(error) => write!(f, "{error}"), Self::AuthStore(error) | Self::DependencyUnavailable(error) | Self::WechatPay(error) => { write!(f, "{error}") } Self::SmsProvider(error) => write!(f, "{error}"), Self::Oss(error) => write!(f, "{error}"), Self::Llm(error) => write!(f, "{error}"), } } } impl Error for AppStateInitError {} impl From for AppStateInitError { fn from(value: JwtError) -> Self { Self::Jwt(value) } } impl From for AppStateInitError { fn from(value: RefreshCookieError) -> Self { Self::RefreshCookie(value) } } impl From for AppStateInitError { fn from(value: SmsProviderError) -> Self { Self::SmsProvider(value) } } impl From for AppStateInitError { fn from(value: OssError) -> Self { Self::Oss(value) } } impl From for AppStateInitError { fn from(value: LlmError) -> Self { Self::Llm(value) } } impl AdminRuntime { pub fn is_enabled(&self) -> bool { !self.username.trim().is_empty() && !self.password.trim().is_empty() } pub fn username(&self) -> &str { &self.username } pub fn password(&self) -> &str { &self.password } pub fn subject(&self) -> &str { &self.subject } pub fn display_name(&self) -> &str { &self.display_name } pub fn build_claims(&self, now: OffsetDateTime) -> Result { self.build_account_claims( self.subject.to_string(), self.username.to_string(), self.display_name.to_string(), vec![ADMIN_ROLE.to_string(), "owner".to_string()], 1, now, ) } pub fn build_account_claims( &self, subject: String, username: String, display_name: String, roles: Vec, token_version: u64, now: OffsetDateTime, ) -> Result { let expires_at = now .checked_add(time::Duration::seconds( i64::try_from(self.token_ttl_seconds) .map_err(|_| "后台 token TTL 超出 i64 上限".to_string())?, )) .ok_or_else(|| "后台 token 过期时间计算溢出".to_string())?; Ok(AdminClaims { subject, username, display_name, roles, token_version, issued_at: now, expires_at, }) } pub fn sign_token(&self, claims: &AdminClaims) -> Result { let jwt_claims = AccessTokenClaims::from_input( AccessTokenClaimsInput { user_id: claims.subject.clone(), session_id: format!("admin-session:{}:{}", claims.subject, claims.token_version), provider: AuthProvider::Password, roles: claims.roles.clone(), token_version: claims.token_version, phone_verified: false, binding_status: BindingStatus::Active, display_name: Some(claims.display_name.clone()), }, &self.jwt_config, claims.issued_at, ) .map_err(|error| error.to_string())?; sign_access_token(&jwt_claims, &self.jwt_config).map_err(|error| error.to_string()) } pub fn verify_token(&self, token: &str) -> Result { verify_access_token(token, &self.jwt_config).map_err(|error| error.to_string()) } pub fn validate_claims(&self, claims: &AccessTokenClaims) -> Result { if claims.user_id() != self.subject.as_ref() { return Err("后台管理员主体不匹配".to_string()); } if !claims.roles.iter().any(|role| role == ADMIN_ROLE) { return Err("当前令牌不是管理员令牌".to_string()); } let issued_at = OffsetDateTime::from_unix_timestamp(claims.iat as i64) .map_err(|_| "后台令牌签发时间无效".to_string())?; let expires_at = OffsetDateTime::from_unix_timestamp(claims.exp as i64) .map_err(|_| "后台令牌过期时间无效".to_string())?; Ok(AdminSession { subject: claims.user_id().to_string(), username: self.username.to_string(), display_name: self.display_name.to_string(), roles: claims.roles.clone(), account_role: "owner".to_string(), tab_permissions: Vec::new(), action_permissions: Vec::new(), issued_at, expires_at, }) } pub fn validate_account_claims( &self, claims: &AccessTokenClaims, expected_subject: &str, expected_token_version: u64, username: String, display_name: String, tab_permissions: Vec, action_permissions: Vec, ) -> Result { if claims.user_id() != expected_subject { return Err("后台管理员主体不匹配".to_string()); } if !claims.roles.iter().any(|role| role == ADMIN_ROLE) { return Err("当前令牌不是管理员令牌".to_string()); } if claims.token_version() != expected_token_version { return Err("后台登录状态已失效".to_string()); } let issued_at = OffsetDateTime::from_unix_timestamp(claims.iat as i64) .map_err(|_| "后台令牌签发时间无效".to_string())?; let expires_at = OffsetDateTime::from_unix_timestamp(claims.exp as i64) .map_err(|_| "后台令牌过期时间无效".to_string())?; Ok(AdminSession { subject: claims.user_id().to_string(), username, display_name, roles: claims.roles.clone(), account_role: "member".to_string(), tab_permissions, action_permissions, issued_at, expires_at, }) } pub fn is_owner_subject(&self, subject: &str) -> bool { subject == self.subject.as_ref() } pub fn build_session(&self, claims: &AdminClaims) -> AdminSession { AdminSession { subject: claims.subject.clone(), username: claims.username.clone(), display_name: claims.display_name.clone(), roles: claims.roles.clone(), account_role: "owner".to_string(), tab_permissions: Vec::new(), action_permissions: Vec::new(), issued_at: claims.issued_at, expires_at: claims.expires_at, } } } fn build_oss_client(config: &AppConfig) -> Result, AppStateInitError> { let oss_fields = [ ("ALIYUN_OSS_BUCKET", config.oss_bucket.as_deref()), ("ALIYUN_OSS_ENDPOINT", config.oss_endpoint.as_deref()), ( "ALIYUN_OSS_ACCESS_KEY_ID", config.oss_access_key_id.as_deref(), ), ( "ALIYUN_OSS_ACCESS_KEY_SECRET", config.oss_access_key_secret.as_deref(), ), ]; let has_any_oss_field = oss_fields .iter() .any(|(_, value)| value.is_some_and(|value| !value.trim().is_empty())); if !has_any_oss_field { return Ok(None); } let missing_fields = oss_fields .iter() .filter_map(|(name, value)| match value { Some(value) if !value.trim().is_empty() => None, _ => Some(*name), }) .collect::>(); if !missing_fields.is_empty() { warn!( missing_fields = %missing_fields.join(","), "OSS 环境变量配置不完整，跳过 OSS 客户端初始化" ); 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_read_expire_seconds, config.oss_post_expire_seconds, config.oss_post_max_size_bytes, config.oss_success_action_status, )?; Ok(Some(OssClient::new(oss_config))) } fn build_matting_client(config: &AppConfig) -> Result, AppStateInitError> { if !config.aliyun_matting_enabled { return Ok(None); } let (Some(access_key_id), Some(access_key_secret)) = ( config .aliyun_matting_access_key_id .as_deref() .map(str::trim) .filter(|value| !value.is_empty()), config .aliyun_matting_access_key_secret .as_deref() .map(str::trim) .filter(|value| !value.is_empty()), ) else { warn!("阿里云抠图 AccessKey 未配置，跳过抠图客户端初始化"); return Ok(None); }; let matting_config = MattingConfig::with_timeout( config.aliyun_matting_endpoint.clone(), access_key_id.to_string(), access_key_secret.to_string(), config.aliyun_matting_request_timeout_ms, ) .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))?; MattingClient::new(matting_config) .map(Some) .map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string())) } fn build_bgfilter_provider_http_client( config: &AppConfig, ) -> Result { reqwest::Client::builder() .redirect(reqwest::redirect::Policy::none()) .timeout(std::time::Duration::from_millis( config.bgfilter_provider_attempt_timeout_ms().max(1), )) .connect_timeout(std::time::Duration::from_secs(30)) .pool_idle_timeout(std::time::Duration::from_secs(300)) .pool_max_idle_per_host(64) .tcp_keepalive(std::time::Duration::from_secs(60)) .build() .map_err(|error| { AppStateInitError::DependencyUnavailable(format!( "构建 BgFilter provider HTTP 客户端失败：{error}" )) }) } fn build_bgfilter_worker_http_client( config: &AppConfig, ) -> Result { reqwest::Client::builder() .redirect(reqwest::redirect::Policy::none()) .connect_timeout(std::time::Duration::from_millis( config.bgfilter_worker_connect_timeout_ms.max(1), )) .pool_idle_timeout(std::time::Duration::from_secs(300)) .pool_max_idle_per_host(128) .tcp_keepalive(std::time::Duration::from_secs(60)) .build() .map_err(|error| { AppStateInitError::DependencyUnavailable(format!( "构建 BgFilter 内部 worker HTTP 客户端失败：{error}" )) }) } fn build_character_animation_oss_http_client() -> Result { reqwest::Client::builder() .connect_timeout(std::time::Duration::from_secs(30)) .timeout(std::time::Duration::from_secs(60)) .pool_idle_timeout(std::time::Duration::from_secs(300)) .pool_max_idle_per_host(8) .tcp_keepalive(std::time::Duration::from_secs(60)) .build() .map_err(|error| { AppStateInitError::DependencyUnavailable(format!( "初始化角色动画 OSS HTTP Client 失败：{error}" )) }) } // 中文注释：编辑器图片的 OSS 读写此前每次调用都 `reqwest::Client::new()`，既没有任何 // 超时也没有连接复用——一个半开或黑洞的连接可以无限期挂住，同时占着 HTTP 准入许可和 // 已经读入的最多 32 MiB 图片缓冲。这里收口成进程级共享客户端，给整段请求（含 body // 流式读写）一个绝对上界，作为所有调用方的兜底。读写共用一个客户端，两个方向都受同一份 // EDITOR_REFERENCE_IMAGE_MAX_SIZE_BYTES 约束，合用也能让 GET / PUT / HEAD 复用连接池。 // // connect 取 10s：同区域 OSS 建连是百毫秒级，30s 只会让不可达端点多占 20s 槽位。 // total 取 120s：按较慢的写方向定尺寸——32 MiB 上传在 60s 内要求持续约 4.4 Mbps， // 留一倍余量避免误伤正常流量。读方向不因此变松：完美像素的 GET 另受 30s 处理预算约束， // 客户端超时只是兜底。 fn build_editor_oss_http_client() -> Result { reqwest::Client::builder() .connect_timeout(std::time::Duration::from_secs(10)) .timeout(std::time::Duration::from_secs(120)) .pool_idle_timeout(std::time::Duration::from_secs(300)) .pool_max_idle_per_host(8) .tcp_keepalive(std::time::Duration::from_secs(60)) .build() .map_err(|error| { AppStateInitError::DependencyUnavailable(format!( "初始化编辑器 OSS HTTP 客户端失败：{error}" )) }) } fn build_wechat_client(config: &AppConfig) -> WechatClient { WechatClient::new(WechatConfig { app_id: config.wechat_mini_program_app_id.clone(), app_secret: config.wechat_mini_program_app_secret.clone(), stable_access_token_endpoint: config.wechat_stable_access_token_endpoint.clone(), virtual_payment_query_order_endpoint: config .wechat_mini_program_virtual_payment_query_order_endpoint .clone(), virtual_payment_notify_provide_goods_endpoint: config .wechat_mini_program_virtual_payment_notify_provide_goods_endpoint .clone(), }) } fn build_llm_client(config: &AppConfig) -> Result, AppStateInitError> { let Some(api_key) = config .llm_api_key .as_ref() .map(|value| value.trim()) .filter(|value| !value.is_empty()) else { return Ok(None); }; let llm_config = LlmConfig::new( config.llm_provider, config.llm_base_url.clone(), api_key.to_string(), config.llm_model.clone(), config.llm_request_timeout_ms, config.llm_max_retries, config.llm_retry_backoff_ms, )?; Ok(Some(LlmClient::new(llm_config)?)) } fn build_vector_engine_llm_client( config: &AppConfig, ) -> Result, AppStateInitError> { // 中文注释：Apimart 已于 2026-06 弃用，LLM 文本调用统一迁移到 VectorEngine。 let Some(api_key) = config .vector_engine_api_key .as_ref() .map(|value| value.trim()) .filter(|value| !value.is_empty()) else { return Ok(None); }; let base_url = if config.vector_engine_base_url.ends_with("/v1") { config.vector_engine_base_url.clone() } else { format!("{}/v1", config.vector_engine_base_url.trim_end_matches('/')) }; let llm_config = LlmConfig::new( LlmProvider::OpenAiCompatible, base_url, api_key.to_string(), platform_llm::EDITOR_AGENT_GPT5_MODEL.to_string(), config.llm_request_timeout_ms, config.llm_max_retries.min(EDITOR_AGENT_LLM_MAX_RETRIES), config .llm_retry_backoff_ms .min(EDITOR_AGENT_LLM_MAX_RETRY_BACKOFF_MS), )? .with_openai_chat_token_budget_field(OpenAiChatTokenBudgetField::MaxCompletionTokens); Ok(Some(LlmClient::new(llm_config)?)) } // 只有在用户名和密码都已配置时才启用后台，避免半配置状态暴露伪入口。 fn build_admin_runtime( config: &AppConfig, base_jwt_config: &JwtConfig, ) -> Result, AppStateInitError> { let Some(username) = config .admin_username .as_ref() .map(|value| value.trim()) .filter(|value| !value.is_empty()) else { return Ok(None); }; let Some(password) = config .admin_password .as_ref() .map(|value| value.trim()) .filter(|value| !value.is_empty()) else { return Ok(None); }; let jwt_config = JwtConfig::new( base_jwt_config.issuer().to_string(), config.jwt_secret.clone(), config.admin_token_ttl_seconds, )?; Ok(Some(AdminRuntime { username: Arc::::from(username), password: Arc::::from(password), subject: Arc::::from(format!("admin:{username}")), display_name: Arc::::from(format!("管理员 {username}")), token_ttl_seconds: config.admin_token_ttl_seconds, jwt_config, })) } #[cfg(any())] fn is_missing_creation_entry_config_procedure(error: &SpacetimeClientError) -> bool { match error { SpacetimeClientError::Procedure(message) => message.contains("No such procedure"), _ => false, } } #[cfg(debug_assertions)] fn is_missing_feature_gate_config_procedure(error: &SpacetimeClientError) -> bool { match error { SpacetimeClientError::Procedure(message) => message.contains("No such procedure"), _ => false, } } #[cfg(test)] mod tests { use module_ai::{AiTaskKind, generate_ai_task_id}; use spacetime_client::SpacetimeClientStage; use super::*; #[test] fn debug_summaries_redact_all_runtime_credentials() { const SENSITIVE_KEY_LURE: &str = "ISSUE_148_DEBUG_SECRET_LURE"; let secret = || Some(SENSITIVE_KEY_LURE.to_string()); let config = AppConfig { bgfilter_internal_token: secret(), editor_bgfilter_token: secret(), aliyun_matting_access_key_id: secret(), aliyun_matting_access_key_secret: secret(), admin_username: Some("debug-admin".to_string()), admin_password: secret(), internal_api_secret: secret(), jwt_secret: SENSITIVE_KEY_LURE.to_string(), sms_access_key_id: secret(), sms_access_key_secret: secret(), sms_mock_verify_code: SENSITIVE_KEY_LURE.to_string(), wechat_app_secret: secret(), wechat_mini_program_app_secret: secret(), wechat_pay_private_key_pem: secret(), wechat_pay_private_key_path: Some(std::path::PathBuf::from(SENSITIVE_KEY_LURE)), wechat_pay_platform_public_key_pem: secret(), wechat_pay_platform_public_key_path: Some(std::path::PathBuf::from(SENSITIVE_KEY_LURE)), wechat_pay_api_v3_key: secret(), wechat_mini_program_virtual_payment_offer_id: secret(), wechat_mini_program_virtual_payment_app_key: secret(), wechat_mini_program_virtual_payment_sandbox_app_key: secret(), wechat_mini_program_message_token: secret(), wechat_mini_program_message_encoding_aes_key: secret(), oss_bucket: Some("debug-bucket".to_string()), oss_endpoint: Some("oss.example.invalid".to_string()), oss_access_key_id: secret(), oss_access_key_secret: secret(), spacetime_server_url: format!("https://spacetime.invalid/?token={SENSITIVE_KEY_LURE}"), spacetime_database: format!("debug-{SENSITIVE_KEY_LURE}"), spacetime_token: secret(), spacetime_runtime_service_bootstrap_secret: secret(), llm_base_url: "https://llm.example.invalid".to_string(), llm_api_key: secret(), llm_model: "debug-model".to_string(), dashscope_api_key: secret(), vector_engine_base_url: "https://vector.example.invalid".to_string(), vector_engine_api_key: secret(), // SFX V2 新增的 ElevenLabs 凭据同样落在这条泄漏面上：base_url 可能带 userinfo // 或查询凭据，与 spacetime_server_url 同等对待。 elevenlabs_base_url: format!("https://elevenlabs.invalid/?key={SENSITIVE_KEY_LURE}"), elevenlabs_api_key: secret(), hyper3d_api_key: secret(), volcengine_speech_api_key: secret(), volcengine_speech_app_id: secret(), volcengine_speech_access_key: secret(), ark_character_video_api_key: secret(), ..AppConfig::default() }; let spacetime_config = spacetime_client_config_for_process(&config); let spacetime_client = SpacetimeClient::new(spacetime_config.clone()); let state = AppState::new(config.clone()).expect("state should build"); let config_debug = format!("{config:?}"); let expected_config_debug = format!( "AppConfig {{ bind_port: {:?}, listen_backlog: {:?}, worker_threads: {:?}, process_role: {:?}, external_generation_mode: {:?}, external_generation_worker_concurrency: {:?}, max_concurrent_requests: {:?}, admin_max_concurrent_requests: {:?}, spacetime_pool_size: {:?}, sms_auth_enabled: {:?}, wechat_auth_enabled: {:?}, wechat_pay_enabled: {:?}, aliyun_matting_enabled: {:?}, tracking_outbox_enabled: {:?}, wallet_refund_outbox_enabled: {:?}, otel_enabled: {:?}, credentials: \"\", .. }}", config.bind_port, config.listen_backlog, config.worker_threads, config.process_role, config.external_generation_mode, config.external_generation_worker_concurrency, config.max_concurrent_requests, config.admin_max_concurrent_requests, config.spacetime_pool_size, config.sms_auth_enabled, config.wechat_auth_enabled, config.wechat_pay_enabled, config.aliyun_matting_enabled, config.tracking_outbox_enabled, config.wallet_refund_outbox_enabled, config.otel_enabled, ); assert_eq!( config_debug, expected_config_debug, "AppConfig Debug 只能输出显式允许的枚举、数值、布尔值和脱敏占位；新增自由字符串必须默认缺席" ); let outputs = [ ("AppConfig", config_debug.clone()), ("SpacetimeClientConfig", format!("{spacetime_config:?}")), ("SpacetimeClient", format!("{spacetime_client:?}")), ("AppStateInner", format!("{:?}", state.0.as_ref())), ("AppState", format!("{state:?}")), ]; for (type_name, output) in outputs { assert!( !output.contains(SENSITIVE_KEY_LURE), "{type_name} Debug leaked the credential lure: {output}" ); } assert!(config_debug.contains("process_role")); assert!(config_debug.contains("")); assert!(format!("{spacetime_config:?}").contains("pool_size")); assert!(format!("{state:?}").contains("ready: true")); } #[test] fn app_state_reuses_character_animation_oss_client_and_eight_permits() { let state = AppState::new(AppConfig::default()).expect("state should build"); assert!(std::ptr::eq( state.character_animation_oss_http_client(), state.character_animation_oss_http_client(), )); assert_eq!( state .character_animation_oss_io_limiter() .available_permits(), CHARACTER_ANIMATION_OSS_MAX_CONCURRENCY ); } #[test] fn app_state_reuses_editor_oss_client() { let state = AppState::new(AppConfig::default()).expect("state should build"); assert!(std::ptr::eq( state.editor_oss_http_client(), state.editor_oss_http_client(), )); } #[test] fn bgfilter_image_validation_limiter_is_bounded_per_process_role() { let parent = AppState::new(AppConfig::default()).expect("parent state should build"); assert_eq!( parent .bgfilter_image_validation_limiter() .available_permits(), BGFILTER_IMAGE_VALIDATION_MAX_CONCURRENCY ); let worker = AppState::new(AppConfig { process_role: crate::config::ProcessRole::BgfilterWorker, bgfilter_worker_concurrency: 6, ..AppConfig::default() }) .expect("worker state should build"); assert_eq!( worker .bgfilter_image_validation_limiter() .available_permits(), 6 ); } #[test] fn editor_generation_pricing_typed_record_round_trips() { let expected = crate::editor_generation_config::parse_editor_generation_pricing_json( crate::editor_generation_config::EDITOR_GENERATION_PRICING_DEFAULT_JSON, "test default pricing", ) .expect("default pricing should parse"); let record = EditorGenerationPricingConfigRecord { config_id: "global".to_string(), models: editor_generation_pricing_to_records(&expected) .expect("pricing should map to records"), updated_by_admin_user_id: Some("admin:test".to_string()), updated_at: "2026-07-10T00:00:00Z".to_string(), updated_at_micros: 1, }; let actual = editor_generation_pricing_from_record(record, &expected) .expect("typed pricing record should map back"); assert_eq!(actual, expected); } #[test] fn editor_generation_pricing_typed_record_backfills_legacy_sfx_model() { let fallback = crate::editor_generation_config::parse_editor_generation_pricing_json( crate::editor_generation_config::EDITOR_GENERATION_PRICING_DEFAULT_JSON, "test default pricing", ) .expect("default pricing should parse"); let mut models = editor_generation_pricing_to_records(&fallback).expect("pricing should map to records"); models.retain(|pricing| pricing.model != EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS); let record = EditorGenerationPricingConfigRecord { config_id: "global".to_string(), models, updated_by_admin_user_id: Some("admin:test".to_string()), updated_at: "2026-08-07T00:00:00Z".to_string(), updated_at_micros: 1, }; let actual = editor_generation_pricing_from_record(record, &fallback) .expect("legacy pricing should receive only the new SFX model fallback"); assert_eq!( actual.models.get(EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS), fallback.models.get(EDITOR_SOUND_EFFECT_MODEL_ELEVENLABS) ); } #[test] fn editor_generation_pricing_upsert_input_uses_runtime_service_bootstrap_secret() { let mut config = AppConfig::default(); let without_secret = editor_generation_pricing_upsert_input( &config, "admin:test".to_string(), Vec::new(), 123, ); assert_eq!(without_secret.bootstrap_secret, ""); config.spacetime_runtime_service_bootstrap_secret = Some("11".repeat(32)); let with_secret = editor_generation_pricing_upsert_input( &config, "admin:test".to_string(), Vec::new(), 123, ); assert_eq!(with_secret.bootstrap_secret, "11".repeat(32)); assert_eq!(with_secret.admin_user_id, "admin:test"); assert_eq!(with_secret.updated_at_micros, 123); } #[test] fn editor_generation_pricing_typed_record_rejects_duplicate_model() { let config = crate::editor_generation_config::parse_editor_generation_pricing_json( crate::editor_generation_config::EDITOR_GENERATION_PRICING_DEFAULT_JSON, "test default pricing", ) .expect("default pricing should parse"); let mut models = editor_generation_pricing_to_records(&config).expect("pricing should map to records"); models.push(models[0].clone()); let record = EditorGenerationPricingConfigRecord { config_id: "global".to_string(), models, updated_by_admin_user_id: None, updated_at: "2026-07-10T00:00:00Z".to_string(), updated_at_micros: 1, }; let error = editor_generation_pricing_from_record(record, &config) .expect_err("duplicate model should fail"); assert!(error.to_string().contains("重复模型")); } #[cfg(any())] #[test] fn detects_missing_creation_entry_config_procedure_for_debug_fallback() { assert!(is_missing_creation_entry_config_procedure( &SpacetimeClientError::Procedure( "No such procedure: get_creation_entry_config".to_string(), ), )); assert!(is_missing_creation_entry_config_procedure( &SpacetimeClientError::Procedure("No such procedure".to_string()), )); assert!(!is_missing_creation_entry_config_procedure( &SpacetimeClientError::Timeout(SpacetimeClientStage::ProcedureResult), )); } #[test] fn detects_missing_feature_gate_config_procedure_for_debug_fallback() { assert!(is_missing_feature_gate_config_procedure( &SpacetimeClientError::Procedure( "No such procedure: get_feature_gate_config".to_string(), ), )); assert!(!is_missing_feature_gate_config_procedure( &SpacetimeClientError::Timeout(SpacetimeClientStage::ProcedureResult), )); } #[test] fn feature_gate_user_tags_are_only_required_for_enabled_tag_allowlist() { let mut gate = test_feature_gate("image-editor:agent-sidebar"); assert!(!feature_gate_requires_user_tags(&gate)); gate.allow_user_tags = vec!["beta".to_string()]; assert!(feature_gate_requires_user_tags(&gate)); gate.enabled = false; assert!(!feature_gate_requires_user_tags(&gate)); } #[cfg(any())] #[test] fn creation_entry_tag_lookup_only_needs_matching_enabled_tag_gate() { let config = crate::creation_entry_config::test_creation_entry_config_response(); let mut unrelated_gate = test_feature_gate("image-editor:agent-sidebar"); unrelated_gate.allow_user_tags = vec!["beta".to_string()]; assert!(!creation_entry_feature_gates_require_user_tags( &config, &[unrelated_gate], )); let mut disabled_gate = test_feature_gate("creation-entry:puzzle"); disabled_gate.enabled = false; disabled_gate.allow_user_tags = vec!["beta".to_string()]; assert!(!creation_entry_feature_gates_require_user_tags( &config, &[disabled_gate], )); let mut matching_gate = test_feature_gate("creation-entry:puzzle"); matching_gate.allow_user_tags = vec!["beta".to_string()]; assert!(creation_entry_feature_gates_require_user_tags( &config, &[matching_gate], )); } #[test] fn app_state_exposes_usable_ai_task_service() { let state = AppState::new(AppConfig::default()).expect("state should build"); let task_id = generate_ai_task_id(1_713_680_000_000_000); let created = state .ai_task_service() .create_task(module_ai::AiTaskCreateInput { task_id: task_id.clone(), task_kind: AiTaskKind::StoryGeneration, owner_user_id: "user_001".to_string(), request_label: "营地开场".to_string(), source_module: "story".to_string(), source_entity_id: Some("storysess_001".to_string()), request_payload_json: Some("{\"scene\":\"camp\"}".to_string()), stages: AiTaskKind::StoryGeneration.default_stage_blueprints(), created_at_micros: 1_713_680_000_000_000, }) .expect("ai task should create"); assert_eq!(created.task_id, task_id); assert_eq!(created.task_kind, AiTaskKind::StoryGeneration); assert_eq!(created.stages.len(), 4); } #[test] fn app_state_skips_llm_client_when_api_key_missing() { let state = AppState::new(AppConfig::default()).expect("state should build"); assert!(state.llm_client().is_none()); assert!(state.vector_engine_llm_client().is_none()); } #[test] fn app_state_skips_oss_client_when_oss_config_is_partial() { let mut config = AppConfig::default(); config.oss_bucket = Some("genarrative-assets".to_string()); config.oss_endpoint = Some("oss-cn-hangzhou.aliyuncs.com".to_string()); let state = AppState::new(config).expect("state should build with partial oss config"); assert!(state.oss_client().is_none()); } #[test] fn app_state_builds_editor_agent_llm_client_from_vector_engine_settings() { let mut config = AppConfig::default(); config.llm_api_key = None; config.llm_max_retries = 2; config.llm_retry_backoff_ms = 120_000; config.vector_engine_base_url = "https://api.vectorengine.test".to_string(); config.vector_engine_api_key = Some("ve-key".to_string()); let state = AppState::new(config).expect("state should build"); let client = state .vector_engine_llm_client() .expect("editor agent LLM client should exist"); assert_eq!( client.config().model(), platform_llm::EDITOR_AGENT_GPT5_MODEL ); assert_eq!( client.config().chat_completions_url(), "https://api.vectorengine.test/v1/chat/completions" ); assert!(!client.config().official_fallback()); assert_eq!( client.config().openai_chat_token_budget_field(), OpenAiChatTokenBudgetField::MaxCompletionTokens ); assert_eq!(client.config().max_retries(), 1); assert_eq!(client.config().retry_backoff_ms(), 60_000); } fn test_feature_gate(gate_key: &str) -> module_runtime::FeatureGateConfigSnapshot { module_runtime::FeatureGateConfigSnapshot { gate_key: gate_key.to_string(), enabled: true, rollout_percent: 0, allow_user_ids: vec![], allow_user_tags: vec![], deny_user_ids: vec![], description: String::new(), updated_at_micros: 1, } } #[cfg(any())] #[test] fn puzzle_api_state_exposes_puzzle_dependency_snapshot() { let mut config = AppConfig::default(); config.creation_agent_llm_web_search_enabled = false; config.vector_engine_image_request_timeout_ms = 987_654; let state = AppState::new(config).expect("state should build"); let puzzle_state: PuzzleApiState = FromRef::from_ref(&state); assert!(!puzzle_state.creation_agent_llm_web_search_enabled()); assert_eq!( puzzle_state.vector_engine_image_request_timeout_ms(), 987_654 ); assert!(puzzle_state.llm_client().is_none()); assert!(puzzle_state.creative_agent_gpt5_client().is_none()); assert!(puzzle_state.oss_client().is_none()); } }