diff --git a/backend/crates/gateway-admin/Cargo.toml b/backend/crates/gateway-admin/Cargo.toml index 09ce322b3..6e96f617e 100644 --- a/backend/crates/gateway-admin/Cargo.toml +++ b/backend/crates/gateway-admin/Cargo.toml @@ -31,7 +31,7 @@ sha2 = "0.11" subtle = "2.6.1" thiserror = "2.0.18" tracing = "0.1.44" -uuid = { version = "1.23.3", features = ["v7"] } +uuid = { version = "1.23.3", features = ["v4", "v7"] } cron = "0.17" chrono-tz = "0.10" tokio = { version = "1.52", features = ["rt", "time", "macros", "sync"] } diff --git a/backend/crates/gateway-admin/src/lib.rs b/backend/crates/gateway-admin/src/lib.rs index 00ff49590..b7bca49bd 100644 --- a/backend/crates/gateway-admin/src/lib.rs +++ b/backend/crates/gateway-admin/src/lib.rs @@ -21,6 +21,7 @@ pub mod freeze_recovery; pub mod model; pub mod ports; pub mod public_service; +mod reset_credit_automation; mod use_case; pub use use_case::plugins::{PluginDistributionPorts, PluginManagementService, PluginsService}; @@ -423,6 +424,8 @@ async fn initialize_inner( let accounts = Arc::new(DefaultAccountsService::new( store.accounts(), store.account_runtime(), + store.settings(), + timezone, registry.clone(), snapshot.clone(), probe.clone(), @@ -553,6 +556,33 @@ async fn initialize_inner( .map_err(|_| AdminError::internal("导入 Worker 注册信息不合法"))?; worker_contributions.push(WorkerContribution::Registration(registration)); worker_contributions.extend(freeze_recovery_worker_contribution(freeze_recovery)?); + let id = WorkerId::try_new(WorkerKind::ResetCreditAutomation, "admin") + .map_err(|_| AdminError::internal("重置卡 Worker ID 不合法"))?; + let schedule = WorkerSchedule::try_new( + Duration::from_secs(5), + Duration::from_secs(5), + Duration::from_secs(60), + Duration::from_secs(90), + Duration::from_secs(30), + ) + .map_err(|_| AdminError::internal("重置卡 Worker 调度配置不合法"))?; + let lease = WorkerLeaseRequest::try_new(id.clone(), Duration::from_secs(90)) + .map_err(|_| AdminError::internal("重置卡 Worker 租约配置不合法"))?; + worker_contributions.push(WorkerContribution::Registration( + WorkerRegistration::try_new( + id, + WorkerRunnable::Scheduled { + schedule, + lease: Some(lease), + task: Box::new(reset_credit_automation::ResetCreditAutomationTask { + accounts, + store: store.accounts(), + settings: store.settings(), + }), + }, + ) + .map_err(|_| AdminError::internal("重置卡 Worker 注册失败"))?, + )); Ok(AdminBundle { services, worker_contributions, diff --git a/backend/crates/gateway-admin/src/model/mod.rs b/backend/crates/gateway-admin/src/model/mod.rs index dcd37e802..65213fd34 100644 --- a/backend/crates/gateway-admin/src/model/mod.rs +++ b/backend/crates/gateway-admin/src/model/mod.rs @@ -22,6 +22,7 @@ pub mod provider_credentials; pub mod proxies; pub mod quota_forecast; pub mod quota_forecast_sampling; +pub mod reset_credit_automation; pub mod retention; pub mod settings; pub mod system; diff --git a/backend/crates/gateway-admin/src/model/reset_credit_automation.rs b/backend/crates/gateway-admin/src/model/reset_credit_automation.rs new file mode 100644 index 000000000..e635710bf --- /dev/null +++ b/backend/crates/gateway-admin/src/model/reset_credit_automation.rs @@ -0,0 +1,45 @@ +//! 自动重置卡的持久计划与查询投影，不保存上游凭据或完整卡片 + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PlannedResetCredit { + pub id: String, + pub expires_at: DateTime, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PendingResetCredit { + pub credit: PlannedResetCredit, + pub redeem_request_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ResetCreditAutomationState { + pub next_discovery_at: DateTime, + pub credits: Vec, + pub pending: Option, + pub retry_at: Option>, + pub failures: u32, + pub last_result: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ResetCreditAutomation { + pub account_id: String, + pub enabled: bool, + /// 零表示尚未创建；每次持久状态写入都递增版本 + pub revision: u64, + pub next_run_at: DateTime, + pub state: ResetCreditAutomationState, +} + +#[derive(Debug, Clone)] +pub struct ResetCreditAutomationStatus { + pub enabled: bool, + pub lead_minutes: u32, + pub next_discovery_at: Option>, + pub next_consume_at: Option>, + pub last_result: Option, +} diff --git a/backend/crates/gateway-admin/src/model/settings.rs b/backend/crates/gateway-admin/src/model/settings.rs index abd2e52f0..48a9bbf6d 100644 --- a/backend/crates/gateway-admin/src/model/settings.rs +++ b/backend/crates/gateway-admin/src/model/settings.rs @@ -75,6 +75,7 @@ pub struct RuntimeSettingsValues { pub account_warmup_enabled: bool, pub account_warmup_schedule_time: String, pub account_warmup_model: Option, + pub reset_credit_lead_minutes: u32, } /// 完整运行设置事实 @@ -189,6 +190,9 @@ impl From for ReplaceRuntimeSettings { impl RuntimeSettingsValues { /// 校验共同事实，返回字段名供边界转换错误；SQL 类型转换仍由 Store 处理 pub fn validate(&self) -> Result<(), &'static str> { + if !(1..=1440).contains(&self.reset_credit_lead_minutes) { + return Err("reset_credit_lead_minutes"); + } self.request_location .validate() .map_err(|_| "request_location")?; diff --git a/backend/crates/gateway-admin/src/ports/mod.rs b/backend/crates/gateway-admin/src/ports/mod.rs index 22c605935..5f53b9662 100644 --- a/backend/crates/gateway-admin/src/ports/mod.rs +++ b/backend/crates/gateway-admin/src/ports/mod.rs @@ -11,6 +11,7 @@ pub mod plugins; pub mod pricing; pub mod provider; pub mod proxy; +pub mod reset_credit_automation; pub mod retention; pub mod store; pub mod system; diff --git a/backend/crates/gateway-admin/src/ports/reset_credit_automation.rs b/backend/crates/gateway-admin/src/ports/reset_credit_automation.rs new file mode 100644 index 000000000..9bdf1be71 --- /dev/null +++ b/backend/crates/gateway-admin/src/ports/reset_credit_automation.rs @@ -0,0 +1,29 @@ +//! 自动重置卡计划的持久化端口，配置变更沿用管理事务与审计 + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; + +use super::store::AdminStoreResult; +use crate::model::{MutationContext, reset_credit_automation::ResetCreditAutomation}; + +#[async_trait] +pub trait ResetCreditAutomationStore: Send + Sync { + async fn reset_credit_automation( + &self, + account_id: &str, + ) -> AdminStoreResult>; + + /// 只读取到期的本地计划，按到期时间稳定排序 + async fn due_reset_credit_automations( + &self, + now: DateTime, + limit: u32, + ) -> AdminStoreResult>; + + /// 按 record.revision 比较写入，成功返回新版本；配置变更传入审计上下文 + async fn save_reset_credit_automation( + &self, + record: &ResetCreditAutomation, + context: Option<&MutationContext>, + ) -> AdminStoreResult; +} diff --git a/backend/crates/gateway-admin/src/ports/store.rs b/backend/crates/gateway-admin/src/ports/store.rs index fcf5cc9aa..ca43ad57a 100644 --- a/backend/crates/gateway-admin/src/ports/store.rs +++ b/backend/crates/gateway-admin/src/ports/store.rs @@ -109,7 +109,9 @@ pub type AdminStoreResult = Result; /// 账号目录与公共账号写操作 #[async_trait] -pub trait AccountStore: Send + Sync { +pub trait AccountStore: + super::reset_credit_automation::ResetCreditAutomationStore + Send + Sync +{ /// 插件回调按授权 Provider/账号在数据库过滤后做稳定 cursor 分页 async fn list_plugin_accounts( &self, diff --git a/backend/crates/gateway-admin/src/reset_credit_automation.rs b/backend/crates/gateway-admin/src/reset_credit_automation.rs new file mode 100644 index 000000000..f8a45369c --- /dev/null +++ b/backend/crates/gateway-admin/src/reset_credit_automation.rs @@ -0,0 +1,125 @@ +//! 每日错峰发现与已知卡片的临期执行，周期只读取本地到期计划 + +use crate::{ + model::reset_credit_automation::{PlannedResetCredit, ResetCreditAutomation}, + ports::store::{AccountStore, SettingsStore}, + use_case::accounts::AccountsService, +}; +use chrono::{DateTime, Utc}; +use futures::StreamExt as _; +use gateway_core::{ + account::ProviderAccountId, + task::{ScheduledTask, WorkerCycleContext, WorkerTaskError}, + time::DeploymentTimeZone, +}; +use sha2::{Digest as _, Sha256}; +use std::{sync::Arc, time::Duration}; + +/// 当地次日随机发现时间；随机数由调用者提供，结果持久化后不会因重启重抽 +pub(crate) fn next_discovery( + now: DateTime, + timezone: DeploymentTimeZone, + random: u64, +) -> Option> { + let start = timezone.days_after(now, 1)?; + let end = timezone.days_after(now, 2)?; + let seconds = u64::try_from((end - start).num_seconds()).ok()?; + if seconds == 0 { + return None; + } + start.checked_add_signed(chrono::Duration::seconds( + i64::try_from(random % seconds).ok()?, + )) +} + +/// 提前窗口的前 10% 用于账号错峰，最多一分钟，其余时间保留给网络失败恢复 +pub(crate) fn consume_at( + account_id: &str, + credit: &PlannedResetCredit, + lead_minutes: u32, +) -> DateTime { + let lead_seconds = i64::from(lead_minutes) * 60; + let mut hash = Sha256::new(); + hash.update(account_id.as_bytes()); + hash.update([0]); + hash.update(credit.id.as_bytes()); + let digest = hash.finalize(); + let offset = + i64::from(u16::from_be_bytes([digest[0], digest[1]])) % (lead_seconds / 10).clamp(1, 60); + credit.expires_at - chrono::Duration::seconds(lead_seconds - offset) +} + +pub(crate) fn next_consumption( + record: &ResetCreditAutomation, + lead_minutes: u32, +) -> Option> { + record + .state + .credits + .iter() + .map(|credit| consume_at(&record.account_id, credit, lead_minutes)) + .min() +} + +pub(crate) fn reschedule( + record: &mut ResetCreditAutomation, + lead_minutes: u32, + now: DateTime, +) { + record.next_run_at = if let Some(retry) = record.state.retry_at { + retry + } else if record.state.pending.is_some() { + now + } else { + next_consumption(record, lead_minutes).map_or(record.state.next_discovery_at, |due| { + due.min(record.state.next_discovery_at) + }) + }; +} + +pub(crate) struct ResetCreditAutomationTask { + pub(crate) accounts: Arc, + pub(crate) store: Arc, + pub(crate) settings: Arc, +} + +impl ScheduledTask for ResetCreditAutomationTask { + fn run_cycle( + &self, + context: WorkerCycleContext, + ) -> futures::future::BoxFuture<'_, Result<(), WorkerTaskError>> { + Box::pin(async move { + let settings = self + .settings + .load_runtime_settings() + .await + .map_err(|_| WorkerTaskError::safe("重置卡设置读取失败"))?; + // 单轮有界，积压沿到期顺序继续；并发复用已有账号刷新配置 + let due = self + .store + .due_reset_credit_automations(Utc::now(), 100) + .await + .map_err(|_| WorkerTaskError::safe("重置卡计划读取失败"))?; + let work = futures::stream::iter(due) + .map(|account_id| async move { + let id = ProviderAccountId::new(account_id.clone()) + .map_err(|_| WorkerTaskError::safe("重置卡账号 ID 不合法"))?; + if let Err(error) = self.accounts.reconcile_reset_credit_automation(id).await { + tracing::warn!( + account_id, + error_kind = error.kind().as_str(), + "自动重置卡任务失败" + ); + } + Ok::<_, WorkerTaskError>(()) + }) + .buffer_unordered(settings.values.refresh_concurrency.max(1) as usize) + .collect::>(); + tokio::select! { + () = context.cancellation().cancelled() => {}, + _ = tokio::time::timeout(Duration::from_secs(30), work) => {}, + } + Ok(()) + }) + } +} diff --git a/backend/crates/gateway-admin/src/use_case/accounts.rs b/backend/crates/gateway-admin/src/use_case/accounts/mod.rs similarity index 92% rename from backend/crates/gateway-admin/src/use_case/accounts.rs rename to backend/crates/gateway-admin/src/use_case/accounts/mod.rs index 32376cc76..a88911696 100644 --- a/backend/crates/gateway-admin/src/use_case/accounts.rs +++ b/backend/crates/gateway-admin/src/use_case/accounts/mod.rs @@ -1,5 +1,7 @@ //! 统一账号目录与原生 Provider 分派 +mod reset_credit_automation; + use std::{ collections::{BTreeMap, BTreeSet}, sync::Arc, @@ -123,6 +125,29 @@ pub trait AccountsService: Send + Sync { Err(AdminError::invalid("当前 Provider 不支持账号头像")) } + async fn reset_credit_automation_status( + &self, + _account_id: ProviderAccountId, + ) -> Result + { + Err(AdminError::invalid("账号不支持自动重置卡")) + } + async fn set_reset_credit_automation( + &self, + _context: &MutationContext, + _account_id: ProviderAccountId, + _enabled: bool, + ) -> Result + { + Err(AdminError::invalid("账号不支持自动重置卡")) + } + async fn reconcile_reset_credit_automation( + &self, + _account_id: ProviderAccountId, + ) -> Result<(), AdminError> { + Err(AdminError::invalid("账号不支持自动重置卡")) + } + async fn reset_credits( &self, _context: &MutationContext, @@ -160,6 +185,8 @@ pub trait AccountsService: Send + Sync { pub(crate) struct DefaultAccountsService { accounts: Arc, account_runtime: Arc, + settings: Arc, + timezone: gateway_core::time::DeploymentTimeZone, providers: ProviderAdminRegistry, snapshot: Arc, probe: Arc, @@ -172,6 +199,8 @@ impl DefaultAccountsService { pub(crate) fn new( accounts: Arc, account_runtime: Arc, + settings: Arc, + timezone: gateway_core::time::DeploymentTimeZone, providers: ProviderAdminRegistry, snapshot: Arc, probe: Arc, @@ -179,6 +208,8 @@ impl DefaultAccountsService { Self { accounts, account_runtime, + settings, + timezone, providers, snapshot, probe, @@ -891,27 +922,48 @@ impl AccountsService for DefaultAccountsService { .map_err(|error| map_provider_error(error, "provider profile avatar")) } + async fn reset_credit_automation_status( + &self, + account_id: ProviderAccountId, + ) -> Result + { + self.automation_status(&account_id).await + } + async fn set_reset_credit_automation( + &self, + context: &MutationContext, + account_id: ProviderAccountId, + enabled: bool, + ) -> Result + { + self.configure_automation(context, &account_id, enabled) + .await + } + async fn reconcile_reset_credit_automation( + &self, + account_id: ProviderAccountId, + ) -> Result<(), AdminError> { + self.reconcile_automation(&account_id).await + } + async fn reset_credits( &self, context: &MutationContext, account_id: ProviderAccountId, ) -> Result { - let (_, provider) = self.provider_for_account(&account_id).await?; - match provider.reset_credits(&account_id).await { - Ok(credits) => Ok(credits), - Err(error) - if error.kind() - == crate::ports::provider::ProviderAdminErrorKind::CredentialRefreshRequired => - { - self.refresh(context, account_id.clone()).await?; - let (_, provider) = self.provider_for_account(&account_id).await?; - provider - .reset_credits(&account_id) - .await - .map_err(map_reset_credits_error_after_refresh) - } - Err(error) => Err(map_provider_error(error, "provider reset credits")), + let lock = self.reset_credit_lock(&account_id).await; + let _guard = lock.lock().await; + let credits = self + .reset_credits_upstream(context, account_id.clone()) + .await?; + if let Err(error) = self.remember_reset_credits(&account_id, &credits).await { + tracing::warn!( + account_id = account_id.as_str(), + error_kind = error.kind().as_str(), + "保存重置卡计划失败" + ); } + Ok(credits) } async fn consume_reset_credit( @@ -919,27 +971,26 @@ impl AccountsService for DefaultAccountsService { context: &MutationContext, command: ConsumeProviderResetCredit, ) -> Result { - let account_id = command.account_id.clone(); - // 覆盖 credential refresh + 同键重试的完整账号级临界区，避免 401 两次调用 - // 之间插入另一笔不可逆消费 - let lock = self.reset_credit_lock(&account_id).await; + let lock = self.reset_credit_lock(&command.account_id).await; let _guard = lock.lock().await; - let (_, provider) = self.provider_for_account(&account_id).await?; - match provider.consume_reset_credit(command.clone()).await { - Ok(result) => Ok(result), - Err(error) - if error.kind() - == crate::ports::provider::ProviderAdminErrorKind::CredentialRefreshRequired => - { - self.refresh(context, account_id.clone()).await?; - let (_, provider) = self.provider_for_account(&account_id).await?; - provider - .consume_reset_credit(command) - .await - .map_err(map_reset_credits_error_after_refresh) - } - Err(error) => Err(map_provider_error(error, "provider reset-credit consume")), + if self + .accounts + .reset_credit_automation(command.account_id.as_str()) + .await + .map_err(|e| map_store_error(e, "reset-credit automation"))? + .is_some_and(|record| { + record + .state + .pending + .is_some_and(|pending| pending.credit.expires_at > Utc::now()) + }) + { + return Err(AdminError::new( + crate::model::AdminErrorKind::Conflict, + "自动重置结果待确认，请稍后刷新", + )); } + self.consume_reset_credit_upstream(context, command).await } async fn models( diff --git a/backend/crates/gateway-admin/src/use_case/accounts/reset_credit_automation.rs b/backend/crates/gateway-admin/src/use_case/accounts/reset_credit_automation.rs new file mode 100644 index 000000000..8c62360f1 --- /dev/null +++ b/backend/crates/gateway-admin/src/use_case/accounts/reset_credit_automation.rs @@ -0,0 +1,422 @@ +//! 重置卡自动使用的账号锁、持久幂等键、查询缓存与重试编排 + +use super::*; + +impl DefaultAccountsService { + pub(super) async fn reset_credits_upstream( + &self, + context: &MutationContext, + account_id: ProviderAccountId, + ) -> Result { + let (_, provider) = self.provider_for_account(&account_id).await?; + match provider.reset_credits(&account_id).await { + Ok(credits) => Ok(credits), + Err(error) + if error.kind() + == crate::ports::provider::ProviderAdminErrorKind::CredentialRefreshRequired => + { + self.refresh(context, account_id.clone()).await?; + let (_, provider) = self.provider_for_account(&account_id).await?; + provider + .reset_credits(&account_id) + .await + .map_err(map_reset_credits_error_after_refresh) + } + Err(error) => Err(map_provider_error(error, "provider reset credits")), + } + } + + pub(super) async fn consume_reset_credit_upstream( + &self, + context: &MutationContext, + command: ConsumeProviderResetCredit, + ) -> Result { + let account_id = command.account_id.clone(); + let (_, provider) = self.provider_for_account(&account_id).await?; + match provider.consume_reset_credit(command.clone()).await { + Ok(result) => Ok(result), + Err(error) + if error.kind() + == crate::ports::provider::ProviderAdminErrorKind::CredentialRefreshRequired => + { + self.refresh(context, account_id.clone()).await?; + let (_, provider) = self.provider_for_account(&account_id).await?; + provider + .consume_reset_credit(command) + .await + .map_err(map_reset_credits_error_after_refresh) + } + Err(error) => Err(map_provider_error(error, "provider reset-credit consume")), + } + } +} + +use crate::{ + model::{MutationActor, reset_credit_automation::*}, + reset_credit_automation::{consume_at, next_consumption, next_discovery, reschedule}, +}; +use chrono::{DateTime, Utc}; +use rand_core::{OsRng, RngCore as _}; + +impl DefaultAccountsService { + async fn automation_lead_minutes(&self) -> Result { + Ok(self + .settings + .load_runtime_settings() + .await + .map_err(|e| map_store_error(e, "runtime settings"))? + .values + .reset_credit_lead_minutes) + } + + pub(super) async fn automation_status( + &self, + account_id: &ProviderAccountId, + ) -> Result { + self.require_reset_credit_automation(account_id).await?; + let lead_minutes = self.automation_lead_minutes().await?; + let record = self + .accounts + .reset_credit_automation(account_id.as_str()) + .await + .map_err(|e| map_store_error(e, "reset-credit automation"))?; + Ok(ResetCreditAutomationStatus { + enabled: record.as_ref().is_some_and(|r| r.enabled), + lead_minutes, + next_discovery_at: record + .as_ref() + .filter(|r| r.enabled) + .map(|r| r.state.next_discovery_at), + next_consume_at: record.as_ref().filter(|r| r.enabled).and_then(|r| { + next_consumption(r, lead_minutes) + .or_else(|| { + r.state + .pending + .as_ref() + .map(|pending| consume_at(&r.account_id, &pending.credit, lead_minutes)) + }) + .map(|due| r.state.retry_at.map_or(due, |retry| due.max(retry))) + }), + last_result: record.and_then(|r| { + if r.state + .pending + .as_ref() + .is_some_and(|pending| pending.credit.expires_at <= Utc::now()) + { + Some("expired_unknown".to_owned()) + } else { + r.state.last_result + } + }), + }) + } + + async fn require_reset_credit_automation( + &self, + account_id: &ProviderAccountId, + ) -> Result { + let (stored, provider) = self.provider_for_account(account_id).await?; + let capabilities = + provider.account_capabilities(account_id, &stored.account.authentication_kind); + if !capabilities.reset_credits || !capabilities.consume_reset_credit { + return Err(AdminError::invalid("当前账号不支持自动重置卡")); + } + Ok(stored.account.enabled) + } + + pub(super) async fn configure_automation( + &self, + context: &MutationContext, + account_id: &ProviderAccountId, + enabled: bool, + ) -> Result { + let lock = self.reset_credit_lock(account_id).await; + let _guard = lock.lock().await; + self.require_reset_credit_automation(account_id).await?; + let now = Utc::now(); + let mut record = self + .accounts + .reset_credit_automation(account_id.as_str()) + .await + .map_err(|e| map_store_error(e, "reset-credit automation"))? + .unwrap_or_else(|| ResetCreditAutomation { + account_id: account_id.as_str().to_owned(), + enabled: false, + revision: 0, + next_run_at: now, + state: ResetCreditAutomationState { + next_discovery_at: now, + credits: vec![], + pending: None, + retry_at: None, + failures: 0, + last_result: None, + }, + }); + if record.enabled != enabled { + record.enabled = enabled; + // 首次开启错开一分钟；重新开启保留已知到期时间及未决幂等键 + if enabled && record.revision == 0 { + record.state.next_discovery_at = + now + chrono::Duration::seconds(i64::from(OsRng.next_u32() % 60)); + } + let lead = self.automation_lead_minutes().await?; + reschedule(&mut record, lead, now); + self.save_automation(&mut record, Some(context)).await?; + let settings = self + .settings + .load_runtime_settings() + .await + .map_err(|e| map_store_error(e, "runtime settings"))?; + publish_committed(self.snapshot.as_ref(), settings.config_revision).await?; + } + self.automation_status(account_id).await + } + + async fn save_automation( + &self, + record: &mut ResetCreditAutomation, + context: Option<&MutationContext>, + ) -> Result<(), AdminError> { + record.revision = self + .accounts + .save_reset_credit_automation(record, context) + .await + .map_err(|e| map_store_error(e, "reset-credit automation"))?; + Ok(()) + } + + fn observe_reset_credits( + &self, + record: &mut ResetCreditAutomation, + credits: &ProviderResetCredits, + now: DateTime, + ) -> Result<(), AdminError> { + record.state.credits = credits + .credits + .iter() + .filter(|credit| credit.status.as_deref() == Some("available")) + .filter_map(|credit| { + credit + .expires_at + .filter(|expiry| *expiry > now) + .map(|expires_at| PlannedResetCredit { + id: credit.id.clone(), + expires_at, + }) + }) + .collect(); + record.state.credits.sort_by_key(|credit| credit.expires_at); + record.state.credits.dedup_by(|a, b| a.id == b.id); + if record.state.next_discovery_at <= now + || self + .timezone + .local(record.state.next_discovery_at) + .date_naive() + == self.timezone.local(now).date_naive() + { + record.state.next_discovery_at = + next_discovery(now, self.timezone, OsRng.next_u64()) + .ok_or_else(|| AdminError::internal("无法计算重置卡发现时间"))?; + } + if record.state.pending.is_none() { + record.state.retry_at = None; + record.state.failures = 0; + } + Ok(()) + } + + pub(super) async fn remember_reset_credits( + &self, + account_id: &ProviderAccountId, + credits: &ProviderResetCredits, + ) -> Result<(), AdminError> { + let Some(mut record) = self + .accounts + .reset_credit_automation(account_id.as_str()) + .await + .map_err(|e| map_store_error(e, "reset-credit automation"))? + else { + return Ok(()); + }; + if !record.enabled { + return Ok(()); + } + let now = Utc::now(); + self.observe_reset_credits(&mut record, credits, now)?; + reschedule(&mut record, self.automation_lead_minutes().await?, now); + self.save_automation(&mut record, None).await + } + + async fn automation_failed( + &self, + record: &mut ResetCreditAutomation, + lead: u32, + now: DateTime, + ) -> Result<(), AdminError> { + if record.state.pending.is_none() + && !next_consumption(record, lead).is_some_and(|due| due <= now) + { + record.state.next_discovery_at = + next_discovery(now, self.timezone, OsRng.next_u64()) + .ok_or_else(|| AdminError::internal("无法计算重置卡发现时间"))?; + record.state.retry_at = None; + record.state.last_result = Some("discovery_failed".to_owned()); + reschedule(record, lead, now); + return self.save_automation(record, None).await; + } + record.state.failures = record.state.failures.saturating_add(1); + let delay = (30_i64 * 2_i64.pow(record.state.failures.min(4) - 1)).min(300) + + i64::from(OsRng.next_u32() % 30); + record.state.retry_at = Some(now + chrono::Duration::seconds(delay)); + record.state.last_result = Some( + if record.state.pending.is_some() { + "pending" + } else { + "discovery_failed" + } + .to_owned(), + ); + reschedule(record, lead, now); + self.save_automation(record, None).await + } + + pub(super) async fn reconcile_automation( + &self, + account_id: &ProviderAccountId, + ) -> Result<(), AdminError> { + let lock = self.reset_credit_lock(account_id).await; + let _guard = lock.lock().await; + let Some(mut record) = self + .accounts + .reset_credit_automation(account_id.as_str()) + .await + .map_err(|e| map_store_error(e, "reset-credit automation"))? + else { + return Ok(()); + }; + if !record.enabled { + return Ok(()); + } + match self.require_reset_credit_automation(account_id).await { + Ok(true) => {} + Ok(false) => return Ok(()), + Err(error) => { + record.next_run_at = next_discovery(Utc::now(), self.timezone, OsRng.next_u64()) + .ok_or_else(|| AdminError::internal("无法计算重置卡发现时间"))?; + self.save_automation(&mut record, None).await?; + return Err(error); + } + } + let now = Utc::now(); + let lead = self.automation_lead_minutes().await?; + record + .state + .credits + .retain(|credit| credit.expires_at > now); + if record + .state + .pending + .as_ref() + .is_some_and(|pending| pending.credit.expires_at <= now) + { + record.state.pending = None; + record.state.retry_at = None; + record.state.last_result = Some("expired_unknown".to_owned()); + } + if record.state.retry_at.is_some_and(|retry| retry > now) { + reschedule(&mut record, lead, now); + return self.save_automation(&mut record, None).await; + } + let context = MutationContext { + actor: MutationActor::System, + request_id: "reset-credit-automation".to_owned(), + }; + let due_credit = next_consumption(&record, lead).is_some_and(|due| due <= now); + if record.state.pending.is_none() + && (due_credit + || record.state.next_discovery_at <= now + || record.state.retry_at.is_some()) + { + // 先保存下一次发现，查询被取消或进程重启也不会重复刷普通发现接口 + if record.state.next_discovery_at <= now { + record.state.next_discovery_at = + next_discovery(now, self.timezone, OsRng.next_u64()) + .ok_or_else(|| AdminError::internal("无法计算重置卡发现时间"))?; + } + if due_credit || record.state.retry_at.is_some() { + record.state.retry_at = Some(now + chrono::Duration::seconds(30)); + } + reschedule(&mut record, lead, now); + self.save_automation(&mut record, None).await?; + // 真正消费前重新核对卡片；每日发现结果也可直接承担本次核对 + match self + .reset_credits_upstream(&context, account_id.clone()) + .await + { + Ok(credits) => self.observe_reset_credits(&mut record, &credits, Utc::now())?, + Err(error) => { + self.automation_failed(&mut record, lead, Utc::now()) + .await?; + return Err(error); + } + } + if let Some(credit) = record + .state + .credits + .iter() + .find(|credit| consume_at(account_id.as_str(), credit, lead) <= Utc::now()) + .cloned() + { + record.state.pending = Some(PendingResetCredit { + credit, + redeem_request_id: uuid::Uuid::new_v4().to_string(), + }); + record.state.last_result = Some("pending".to_owned()); + record.state.retry_at = Some(Utc::now() + chrono::Duration::seconds(30)); + reschedule(&mut record, lead, Utc::now()); + // 必须在发送消费请求前落盘，进程退出或响应丢失继续使用相同幂等键 + self.save_automation(&mut record, None).await?; + } + } + let Some(pending) = record.state.pending.clone() else { + reschedule(&mut record, lead, Utc::now()); + return self.save_automation(&mut record, None).await; + }; + let command = ConsumeProviderResetCredit { + account_id: account_id.clone(), + credit_id: Some(pending.credit.id.clone()), + redeem_request_id: uuid::Uuid::parse_str(&pending.redeem_request_id) + .map_err(|_| AdminError::internal("重置卡幂等键不合法"))?, + }; + match self.consume_reset_credit_upstream(&context, command).await { + Ok(result) => { + let confirmed = matches!(result.code.as_str(), "reset" | "already_redeemed"); + record.state.pending = None; + record + .state + .credits + .retain(|credit| credit.id != pending.credit.id); + record.state.retry_at = None; + record.state.failures = 0; + record.state.last_result = Some(if confirmed { + result.code + } else { + "not_redeemed".to_owned() + }); + reschedule(&mut record, lead, Utc::now()); + // 已确认的结果先落盘，再刷新额度；观测失败不能重放已完成的消费 + self.save_automation(&mut record, None).await?; + if confirmed { + let _ = self.quota(account_id, true).await; + } + } + Err(error) => { + self.automation_failed(&mut record, lead, Utc::now()) + .await?; + return Err(error); + } + } + Ok(()) + } +} diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs similarity index 97% rename from backend/crates/gateway-admin/tests/use_case/accounts.rs rename to backend/crates/gateway-admin/tests/use_case/accounts/mod.rs index 1dc6decf1..02415471c 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs @@ -1,5 +1,7 @@ //! 账号管理用例测试及共享 Provider、存储替身 +mod reset_credit_automation; + use std::{ collections::BTreeMap, sync::{Arc, Mutex}, @@ -87,6 +89,11 @@ pub(super) struct FakeProviderAdmin { quota_refresh_account: Mutex, AccountRecord)>>, current_credential_revision: Mutex, reset_credit_commands: Mutex>, + reset_credits: Mutex>, + reset_credit_failure: Mutex>, + reset_credit_result: Mutex, + reset_credit_query_started: tokio::sync::Notify, + reset_credit_query_gate: Mutex>>, profile_result: Mutex>, subscription_result: Mutex, ProviderAdminErrorKind>>, personal_info_barrier: Mutex>>, @@ -112,6 +119,11 @@ impl FakeProviderAdmin { quota_refresh_account: Mutex::new(None), current_credential_revision: Mutex::new(revision(1)), reset_credit_commands: Mutex::new(Vec::new()), + reset_credits: Mutex::default(), + reset_credit_failure: Mutex::default(), + reset_credit_result: Mutex::new("reset".into()), + reset_credit_query_started: tokio::sync::Notify::new(), + reset_credit_query_gate: Mutex::new(None), profile_result: Mutex::new(Ok(empty_profile_statistics())), subscription_result: Mutex::new(Ok(None)), personal_info_barrier: Mutex::new(None), @@ -498,6 +510,28 @@ impl ProviderAdmin for FakeProviderAdmin { Ok(self.quota.lock().expect("provider quota").clone()) } + async fn reset_credits( + &self, + _: &ProviderAccountId, + ) -> Result + { + self.record("provider.reset_credits"); + self.require_available()?; + self.reset_credit_query_started.notify_one(); + let gate = self.reset_credit_query_gate.lock().unwrap().take(); + if let Some(gate) = gate { + let _ = gate.await; + } + + let credits = self.reset_credits.lock().unwrap().clone(); + Ok( + gateway_admin::model::provider_credentials::ProviderResetCredits { + available_count: credits.len() as u64, + credits, + }, + ) + } + async fn consume_reset_credit( &self, command: ConsumeProviderResetCredit, @@ -508,8 +542,11 @@ impl ProviderAdmin for FakeProviderAdmin { .expect("provider reset-credit commands") .push(command); self.require_available()?; + if let Some(kind) = self.reset_credit_failure.lock().unwrap().take() { + return Err(ProviderAdminError::new(kind)); + } Ok(ProviderResetCreditResult { - code: "reset".to_owned(), + code: self.reset_credit_result.lock().unwrap().clone(), credit: None, }) } @@ -577,6 +614,9 @@ pub(super) struct FakeAccountStore { quota_window_queries: Mutex>, quota_forecast_history: Mutex, update_commands: Mutex>, + automations: Mutex< + BTreeMap, + >, pub(super) lowered_limits: Mutex>, } @@ -599,6 +639,7 @@ impl FakeAccountStore { quota_forecast_history: Mutex::new(QuotaForecastHistory::default()), update_commands: Mutex::new(Vec::new()), lowered_limits: Mutex::new(Vec::new()), + automations: Mutex::default(), }) } @@ -1145,6 +1186,7 @@ impl SettingsStore for StaticSettingsStore { account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, }) } @@ -3667,3 +3709,53 @@ async fn account_capacity_should_batch_page_ids_and_distinguish_idle_from_unavai } } } + +#[async_trait] +impl gateway_admin::ports::reset_credit_automation::ResetCreditAutomationStore + for FakeAccountStore +{ + async fn reset_credit_automation( + &self, + account_id: &str, + ) -> AdminStoreResult< + Option, + > { + Ok(self.automations.lock().unwrap().get(account_id).cloned()) + } + async fn due_reset_credit_automations( + &self, + now: chrono::DateTime, + limit: u32, + ) -> AdminStoreResult> { + let records = self.automations.lock().unwrap(); + let mut due = records + .values() + .filter(|r| r.enabled && r.next_run_at <= now) + .collect::>(); + due.sort_by_key(|r| r.next_run_at); + Ok(due + .into_iter() + .take(limit as usize) + .map(|r| r.account_id.clone()) + .collect()) + } + async fn save_reset_credit_automation( + &self, + record: &gateway_admin::model::reset_credit_automation::ResetCreditAutomation, + _: Option<&MutationContext>, + ) -> AdminStoreResult { + let mut records = self.automations.lock().unwrap(); + if records.get(&record.account_id).map_or(0, |r| r.revision) != record.revision { + return Err(AdminStoreError::new( + AdminStoreErrorKind::Conflict, + "automation", + "stale revision", + )); + } + let mut next = record.clone(); + next.revision += 1; + let revision = next.revision; + records.insert(next.account_id.clone(), next); + Ok(revision) + } +} diff --git a/backend/crates/gateway-admin/tests/use_case/accounts/reset_credit_automation.rs b/backend/crates/gateway-admin/tests/use_case/accounts/reset_credit_automation.rs new file mode 100644 index 000000000..e84dfb578 --- /dev/null +++ b/backend/crates/gateway-admin/tests/use_case/accounts/reset_credit_automation.rs @@ -0,0 +1,497 @@ +//! 自动重置的持久幂等恢复、低频发现与账号独立错峰 + +use super::*; +use gateway_admin::{ + model::reset_credit_automation::*, + ports::reset_credit_automation::ResetCreditAutomationStore as _, +}; + +fn enable_capability(provider: &FakeProviderAdmin) { + let mut capabilities = provider.capabilities.lock().unwrap(); + capabilities.reset_credits = true; + capabilities.consume_reset_credit = true; +} + +fn plan(expires_at: chrono::DateTime) -> ResetCreditAutomation { + ResetCreditAutomation { + account_id: "acct_test".into(), + enabled: true, + revision: 1, + next_run_at: Utc::now(), + state: ResetCreditAutomationState { + next_discovery_at: Utc::now() + TimeDelta::days(1), + credits: vec![PlannedResetCredit { + id: "card_1".into(), + expires_at, + }], + pending: None, + retry_at: None, + failures: 0, + last_result: None, + }, + } +} + +fn card( + expires_at: chrono::DateTime, +) -> gateway_admin::model::provider_credentials::ProviderResetCredit { + gateway_admin::model::provider_credentials::ProviderResetCredit { + id: "card_1".into(), + status: Some("available".into()), + expires_at: Some(expires_at), + title: None, + reset_type: None, + } +} + +#[tokio::test] +async fn automation_restores_pending_idempotency_after_service_restart() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + let store = FakeAccountStore::new("openai", events.clone()); + let expiry = Utc::now() + TimeDelta::minutes(5); + provider.reset_credits.lock().unwrap().push(card(expiry)); + *provider.reset_credit_failure.lock().unwrap() = Some(ProviderAdminErrorKind::Ambiguous); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), plan(expiry)); + let account_id = ProviderAccountId::new("acct_test").unwrap(); + let services = accounts_service(provider.clone(), store.clone()).await; + services + .accounts() + .reconcile_reset_credit_automation(account_id.clone()) + .await + .expect_err("unknown consume result"); + let persisted = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap(); + let pending = persisted.state.pending.as_ref().unwrap(); + let command = provider.reset_credit_commands()[0].clone(); + assert_eq!( + pending.redeem_request_id, + command.redeem_request_id.to_string() + ); + assert_eq!(command.redeem_request_id.get_version_num(), 4); + assert_eq!(provider.quota_requests().len(), 0); + services + .accounts() + .reconcile_reset_credit_automation(account_id.clone()) + .await + .unwrap(); + assert_eq!( + provider.reset_credit_commands().len(), + 1, + "wait until persisted retry time" + ); + store + .automations + .lock() + .unwrap() + .get_mut("acct_test") + .unwrap() + .state + .retry_at = Some(Utc::now() - TimeDelta::seconds(1)); + drop(services); + let restarted = accounts_service(provider.clone(), store.clone()).await; + restarted + .accounts() + .reconcile_reset_credit_automation(account_id) + .await + .unwrap(); + assert_eq!(provider.reset_credit_commands(), [command.clone(), command]); + assert_eq!( + recorded(&events) + .iter() + .filter(|e| **e == "provider.reset_credits") + .count(), + 1 + ); + let record = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap(); + assert!(record.state.pending.is_none()); + assert!(record.state.credits.is_empty()); + assert_eq!(record.state.last_result.as_deref(), Some("reset")); + assert!(provider.quota_requests().iter().any(|q| q.refresh)); +} + +#[tokio::test] +async fn automation_only_queries_due_accounts_and_rechecks_credit_before_consume() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + let store = FakeAccountStore::new("openai", events.clone()); + let mut record = plan(Utc::now() + TimeDelta::hours(6)); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), record.clone()); + let services = accounts_service(provider.clone(), store.clone()).await; + let account_id = ProviderAccountId::new("acct_test").unwrap(); + services + .accounts() + .reconcile_reset_credit_automation(account_id.clone()) + .await + .unwrap(); + assert!(!recorded(&events).contains(&"provider.reset_credits")); + record.state.credits[0].expires_at = Utc::now() + TimeDelta::minutes(5); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), record); + services + .accounts() + .reconcile_reset_credit_automation(account_id) + .await + .unwrap(); + assert!(recorded(&events).contains(&"provider.reset_credits")); + assert!( + provider.reset_credit_commands().is_empty(), + "disappeared credit must never be consumed" + ); +} + +#[tokio::test] +async fn automation_defaults_off_and_disabled_pending_survives_reenable() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + let store = FakeAccountStore::new("openai", events.clone()); + let services = accounts_service(provider.clone(), store.clone()).await; + let account_id = ProviderAccountId::new("acct_test").unwrap(); + assert!( + !services + .accounts() + .reset_credit_automation_status(account_id.clone()) + .await + .unwrap() + .enabled + ); + services + .accounts() + .set_reset_credit_automation(&context("enable"), account_id.clone(), true) + .await + .unwrap(); + let first = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap(); + assert!(first.state.next_discovery_at >= Utc::now() - TimeDelta::seconds(1)); + let pending = PendingResetCredit { + credit: PlannedResetCredit { + id: "pending".into(), + expires_at: Utc::now() + TimeDelta::hours(1), + }, + redeem_request_id: uuid::Uuid::new_v4().to_string(), + }; + store + .automations + .lock() + .unwrap() + .get_mut("acct_test") + .unwrap() + .state + .pending = Some(pending.clone()); + services + .accounts() + .set_reset_credit_automation(&context("disable"), account_id.clone(), false) + .await + .unwrap(); + services + .accounts() + .reconcile_reset_credit_automation(account_id.clone()) + .await + .unwrap(); + assert!(provider.reset_credit_commands().is_empty()); + services + .accounts() + .set_reset_credit_automation(&context("reenable"), account_id, true) + .await + .unwrap(); + let next = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap(); + assert_eq!(next.state.pending, Some(pending)); + assert_eq!(next.state.next_discovery_at, first.state.next_discovery_at); +} + +#[tokio::test] +async fn automation_schedules_each_account_inside_the_lead_window_and_next_local_day() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + let store = FakeAccountStore::new("openai", events); + let now = Utc::now(); + let expiry = now + TimeDelta::days(2); + provider.reset_credits.lock().unwrap().push(card(expiry)); + let accounts = (0..100) + .map(|n| { + let mut account = account_record("openai"); + account.id = format!("acct_{n}"); + let mut record = plan(expiry); + record.account_id = account.id.clone(); + record.state.next_discovery_at = now - TimeDelta::seconds(1); + store + .automations + .lock() + .unwrap() + .insert(account.id.clone(), record); + account + }) + .collect::>(); + store.set_accounts(accounts.clone()); + let services = accounts_service(provider, store.clone()).await; + let timezone = gateway_core::time::DeploymentTimeZone::default(); + let start = timezone.days_after(now, 1).unwrap(); + let end = timezone.days_after(now, 2).unwrap(); + let mut times = std::collections::BTreeSet::new(); + for account in accounts { + let id = ProviderAccountId::new(account.id.clone()).unwrap(); + services + .accounts() + .reconcile_reset_credit_automation(id.clone()) + .await + .unwrap(); + let status = services + .accounts() + .reset_credit_automation_status(id) + .await + .unwrap(); + let discovery = status.next_discovery_at.unwrap(); + assert!(discovery >= start && discovery < end); + times.insert(status.next_consume_at.unwrap()); + } + assert!( + times.len() > 30, + "accounts must not all fire at the same instant" + ); + assert!(times.iter().all( + |due| *due >= expiry - TimeDelta::minutes(10) && *due < expiry - TimeDelta::minutes(9) + )); +} + +#[tokio::test] +async fn automation_discards_externally_used_credit_and_stops_on_already_redeemed() { + for already_redeemed_after_query in [false, true] { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + let store = FakeAccountStore::new("openai", events.clone()); + let expiry = Utc::now() + TimeDelta::minutes(5); + let mut remote_credit = card(expiry); + if !already_redeemed_after_query { + remote_credit.status = Some("redeemed".into()); + } + provider.reset_credits.lock().unwrap().push(remote_credit); + *provider.reset_credit_result.lock().unwrap() = "already_redeemed".into(); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), plan(expiry)); + let services = accounts_service(provider.clone(), store.clone()).await; + let account_id = ProviderAccountId::new("acct_test").unwrap(); + services + .accounts() + .reconcile_reset_credit_automation(account_id.clone()) + .await + .unwrap(); + services + .accounts() + .reconcile_reset_credit_automation(account_id) + .await + .unwrap(); + assert_eq!( + provider.reset_credit_commands().len(), + usize::from(already_redeemed_after_query) + ); + let stored = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap(); + assert!(stored.state.credits.is_empty()); + assert!(stored.state.pending.is_none()); + if already_redeemed_after_query { + assert_eq!( + stored.state.last_result.as_deref(), + Some("already_redeemed") + ); + } + } +} + +#[tokio::test] +async fn automation_daily_discovery_is_not_polled_again_after_success_or_failure() { + for failed in [false, true] { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + if failed { + provider.fail_next(ProviderAdminErrorKind::Unavailable); + } + let store = FakeAccountStore::new("openai", events.clone()); + let mut record = plan(Utc::now() + TimeDelta::days(3)); + record.state.credits.clear(); + record.state.next_discovery_at = Utc::now() - TimeDelta::seconds(1); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), record); + let services = accounts_service(provider.clone(), store.clone()).await; + let id = ProviderAccountId::new("acct_test").unwrap(); + let first = services + .accounts() + .reconcile_reset_credit_automation(id.clone()) + .await; + assert_eq!(first.is_err(), failed); + let scheduled = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap() + .state + .next_discovery_at; + services + .accounts() + .reconcile_reset_credit_automation(id) + .await + .unwrap(); + assert_eq!( + recorded(&events) + .iter() + .filter(|e| **e == "provider.reset_credits") + .count(), + 1 + ); + assert_eq!( + store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap() + .state + .next_discovery_at, + scheduled + ); + assert!(scheduled > Utc::now()); + } +} + +#[tokio::test] +async fn automation_cancelled_discovery_preserves_next_random_time_across_restart() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + let store = FakeAccountStore::new("openai", events.clone()); + let mut record = plan(Utc::now() + TimeDelta::days(3)); + record.state.credits.clear(); + record.state.next_discovery_at = Utc::now() - TimeDelta::seconds(1); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), record); + let (release, gate) = tokio::sync::oneshot::channel(); + *provider.reset_credit_query_gate.lock().unwrap() = Some(gate); + let services = accounts_service(provider.clone(), store.clone()).await; + let task = tokio::spawn(async move { + services + .accounts() + .reconcile_reset_credit_automation(ProviderAccountId::new("acct_test").unwrap()) + .await + }); + tokio::time::timeout( + std::time::Duration::from_secs(2), + provider.reset_credit_query_started.notified(), + ) + .await + .unwrap(); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + drop(release); + let before = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap(); + assert!(before.state.next_discovery_at > Utc::now()); + let restarted = accounts_service(provider.clone(), store.clone()).await; + restarted + .accounts() + .reconcile_reset_credit_automation(ProviderAccountId::new("acct_test").unwrap()) + .await + .unwrap(); + let after = store + .reset_credit_automation("acct_test") + .await + .unwrap() + .unwrap(); + assert_eq!( + before.state.next_discovery_at, + after.state.next_discovery_at + ); + assert_eq!( + recorded(&events) + .iter() + .filter(|e| **e == "provider.reset_credits") + .count(), + 1 + ); +} + +#[tokio::test] +async fn disabled_automation_only_blocks_manual_consume_until_pending_credit_expires() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + enable_capability(&provider); + let store = FakeAccountStore::new("openai", events); + let mut record = plan(Utc::now() + TimeDelta::minutes(5)); + record.enabled = false; + record.state.pending = Some(PendingResetCredit { + credit: record.state.credits[0].clone(), + redeem_request_id: uuid::Uuid::new_v4().to_string(), + }); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), record.clone()); + let services = accounts_service(provider.clone(), store.clone()).await; + let command = ConsumeProviderResetCredit { + account_id: ProviderAccountId::new("acct_test").unwrap(), + credit_id: Some("another_card".into()), + redeem_request_id: uuid::Uuid::new_v4(), + }; + services + .accounts() + .consume_reset_credit(&context("manual-with-pending"), command.clone()) + .await + .expect_err("unresolved automation is still active"); + assert!(provider.reset_credit_commands().is_empty()); + record.state.pending.as_mut().unwrap().credit.expires_at = Utc::now() - TimeDelta::seconds(1); + store + .automations + .lock() + .unwrap() + .insert("acct_test".into(), record); + services + .accounts() + .consume_reset_credit(&context("manual-with-pending"), command) + .await + .unwrap(); + assert_eq!(provider.reset_credit_commands().len(), 1); +} diff --git a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs index f8c2c5c16..4626161e8 100644 --- a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs +++ b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs @@ -62,6 +62,7 @@ fn runtime_settings(enabled: bool, probe_enabled: bool, adaptive: bool) -> Runti account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, } } diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index 483087187..4433d71ab 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -1162,3 +1162,31 @@ fn unavailable_system() -> SystemOperationError { ) } mod service_contract; + +#[async_trait] +impl gateway_admin::ports::reset_credit_automation::ResetCreditAutomationStore + for UnavailableStore +{ + async fn reset_credit_automation( + &self, + _: &str, + ) -> AdminStoreResult< + Option, + > { + Ok(None) + } + async fn due_reset_credit_automations( + &self, + _: chrono::DateTime, + _: u32, + ) -> AdminStoreResult> { + Ok(Vec::new()) + } + async fn save_reset_credit_automation( + &self, + _: &gateway_admin::model::reset_credit_automation::ResetCreditAutomation, + _: Option<&gateway_admin::model::MutationContext>, + ) -> AdminStoreResult { + Err(unavailable("reset-credit automation")) + } +} diff --git a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs index 086203c8e..87fcec6bc 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability/mod.rs @@ -1140,6 +1140,7 @@ impl SettingsStore for FixtureSettingsStore { account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, }) } diff --git a/backend/crates/gateway-admin/tests/use_case/settings.rs b/backend/crates/gateway-admin/tests/use_case/settings.rs index d4fde2f7e..a3b02d39d 100644 --- a/backend/crates/gateway-admin/tests/use_case/settings.rs +++ b/backend/crates/gateway-admin/tests/use_case/settings.rs @@ -355,6 +355,7 @@ async fn settings_should_reject_zero_refresh_margin_before_store_call() { account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, }, ) diff --git a/backend/crates/gateway-api/src/admin/accounts/handlers.rs b/backend/crates/gateway-api/src/admin/accounts/handlers.rs index c48819369..b4b391716 100644 --- a/backend/crates/gateway-api/src/admin/accounts/handlers.rs +++ b/backend/crates/gateway-api/src/admin/accounts/handlers.rs @@ -36,6 +36,11 @@ where "/api/admin/accounts/profile-avatar", get(account_profile_avatar::), ) + .route( + "/api/admin/accounts/reset-credits/automation", + get(account_reset_credit_automation::) + .post(set_account_reset_credit_automation::), + ) .route( "/api/admin/accounts/reset-credits", get(account_reset_credits::).post(consume_account_reset_credit::), @@ -625,3 +630,54 @@ where }); Ok(Sse::new(stream).keep_alive(KeepAlive::default())) } + +async fn account_reset_credit_automation( + _auth: AdminAuth, + State(state): State, + AdminQuery(query): AdminQuery, +) -> Result +where + S: SessionState + Send + Sync, +{ + let time = crate::time::TimePresenter::new(state.admin_services().timezone()); + let result = state + .admin_services() + .accounts() + .reset_credit_automation_status(query.into_id().map_err(map_wire_error)?) + .await + .map_err(map_service_error)?; + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(AccountResetCreditAutomationData::from((result, time))), + )) +} + +async fn set_account_reset_credit_automation( + auth: AdminAuth, + State(state): State, + AdminJson(request): AdminJson, +) -> Result +where + S: SessionState + Send + Sync, +{ + let time = crate::time::TimePresenter::new(state.admin_services().timezone()); + let account_id = AccountIdQuery { + account_id: request.account_id, + } + .into_id() + .map_err(map_wire_error)?; + let result = state + .admin_services() + .accounts() + .set_reset_credit_automation( + &auth.context().mutation_context(), + account_id, + request.enabled, + ) + .await + .map_err(map_service_error)?; + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(AccountResetCreditAutomationData::from((result, time))), + )) +} diff --git a/backend/crates/gateway-api/src/admin/accounts/wire.rs b/backend/crates/gateway-api/src/admin/accounts/wire.rs index 348f5e751..52cbd62d9 100644 --- a/backend/crates/gateway-api/src/admin/accounts/wire.rs +++ b/backend/crates/gateway-api/src/admin/accounts/wire.rs @@ -1192,3 +1192,49 @@ fn profile_activity_calendar( weeks, }) } + +/// 账号自动使用临期重置卡开关 +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct AccountResetCreditAutomationRequest { + pub account_id: String, + pub enabled: bool, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AccountResetCreditAutomationData { + pub enabled: bool, + pub lead_minutes: u32, + pub next_discovery_at_display: Option, + pub next_consume_at_display: Option, + pub last_result: Option, +} + +impl + From<( + gateway_admin::model::reset_credit_automation::ResetCreditAutomationStatus, + crate::time::TimePresenter, + )> for AccountResetCreditAutomationData +{ + fn from( + (status, time): ( + gateway_admin::model::reset_credit_automation::ResetCreditAutomationStatus, + crate::time::TimePresenter, + ), + ) -> Self { + Self { + enabled: status.enabled, + lead_minutes: status.lead_minutes, + next_discovery_at_display: status + .next_discovery_at + .as_ref() + .map(|value| time.datetime(value)), + next_consume_at_display: status + .next_consume_at + .as_ref() + .map(|value| time.datetime(value)), + last_result: status.last_result, + } + } +} diff --git a/backend/crates/gateway-api/src/admin/settings.rs b/backend/crates/gateway-api/src/admin/settings.rs index 4a69395f8..2219ce96e 100644 --- a/backend/crates/gateway-api/src/admin/settings.rs +++ b/backend/crates/gateway-api/src/admin/settings.rs @@ -66,6 +66,7 @@ pub struct RuntimeSettingsFields { pub account_warmup_enabled: bool, pub account_warmup_schedule_time: String, pub account_warmup_model: Option, + pub reset_credit_lead_minutes: u32, } /// 运行配置投影与设置页字段的聚合响应 @@ -785,6 +786,7 @@ impl RuntimeSettingsFields { account_warmup_enabled: self.account_warmup_enabled, account_warmup_schedule_time: self.account_warmup_schedule_time, account_warmup_model: self.account_warmup_model, + reset_credit_lead_minutes: self.reset_credit_lead_minutes, }; values.validate().map_err(settings_validation_error)?; Ok(values) @@ -825,6 +827,7 @@ impl From for RuntimeSett account_warmup_enabled: settings.account_warmup_enabled, account_warmup_schedule_time: settings.account_warmup_schedule_time, account_warmup_model: settings.account_warmup_model, + reset_credit_lead_minutes: settings.reset_credit_lead_minutes, } } } @@ -840,6 +843,7 @@ fn settings_validation_error(field: &'static str) -> WireValidationError { "account_auto_freeze_window_seconds" => "accountAutoFreezeWindowSeconds", "account_warmup_enabled" => "accountWarmupEnabled", "account_warmup_model" => "accountWarmupModel", + "reset_credit_lead_minutes" => "resetCreditLeadMinutes", "account_warmup_schedule_time" => "accountWarmupScheduleTime", "audit_retention_days" => "auditRetentionDays", "concurrency_wait_timeout_seconds" => "concurrencyWaitTimeoutSeconds", diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 353a3a9cc..f644af044 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -646,6 +646,7 @@ impl SettingsStore for MemorySettingsStore { account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, }; *settings = updated.clone(); @@ -1789,6 +1790,7 @@ fn test_runtime_settings() -> RuntimeSettings { account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, } } @@ -1845,3 +1847,29 @@ fn unsupported_provider() -> ProviderAdminError { fn unavailable_system() -> SystemOperationError { SystemOperationError::new(SystemOperationErrorKind::Internal, "unused test operation") } + +#[async_trait] +impl gateway_admin::ports::reset_credit_automation::ResetCreditAutomationStore for UnusedStore { + async fn reset_credit_automation( + &self, + _: &str, + ) -> AdminStoreResult< + Option, + > { + Ok(None) + } + async fn due_reset_credit_automations( + &self, + _: chrono::DateTime, + _: u32, + ) -> AdminStoreResult> { + Ok(Vec::new()) + } + async fn save_reset_credit_automation( + &self, + _: &gateway_admin::model::reset_credit_automation::ResetCreditAutomation, + _: Option<&gateway_admin::model::MutationContext>, + ) -> AdminStoreResult { + Err(unavailable("reset-credit automation")) + } +} diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index 66e53be26..afde7b1b9 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -74,7 +74,8 @@ fn update_body() -> Value { "accountAutoFreezeAdaptiveConcurrency": true, "accountWarmupEnabled": false, "accountWarmupScheduleTime": "08:00", - "accountWarmupModel": null + "accountWarmupModel": null, + "resetCreditLeadMinutes": 10 }) } @@ -408,6 +409,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, }; @@ -459,6 +461,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { "accountWarmupEnabled": false, "accountWarmupScheduleTime": "08:00", "accountWarmupModel": null, + "resetCreditLeadMinutes": 10, "updatedAt": "2026-08-02T10:30:00Z", "updatedAtDisplay": "2026-08-02 18:30:00" }) @@ -535,6 +538,7 @@ fn settings_request_and_response_fields_should_stay_in_lockstep() { account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, }; @@ -1617,3 +1621,24 @@ async fn account_affinity_and_rotation_budget_round_trip_and_reject_invalid_valu assert_eq!(response.status(), expected_status, "{field}"); } } + +#[test] +fn reset_credit_lead_minutes_requires_an_integer_within_one_day() { + for (value, valid) in [ + (0, false), + (1, true), + (10, true), + (1440, true), + (1441, false), + ] { + let mut body = update_body(); + body["resetCreditLeadMinutes"] = json!(value); + let request: UpdateRuntimeSettingsRequest = serde_json::from_value(body).unwrap(); + let result = request.validate(); + if valid { + assert!(result.is_ok()); + } else { + assert_eq!(result.unwrap_err().field(), "resetCreditLeadMinutes"); + } + } +} diff --git a/backend/crates/gateway-core/src/task.rs b/backend/crates/gateway-core/src/task.rs index 490337207..52d3f4469 100644 --- a/backend/crates/gateway-core/src/task.rs +++ b/backend/crates/gateway-core/src/task.rs @@ -26,10 +26,11 @@ pub enum WorkerKind { Backup, AccountImport, AccountFreezeRecovery, + ResetCreditAutomation, } impl WorkerKind { - pub const ALL: [Self; 11] = [ + pub const ALL: [Self; 12] = [ Self::OAuthRefresh, Self::QuotaCatalogHealth, Self::RuntimeSnapshotReconciliation, @@ -41,6 +42,7 @@ impl WorkerKind { Self::Backup, Self::AccountImport, Self::AccountFreezeRecovery, + Self::ResetCreditAutomation, ]; #[must_use] @@ -57,6 +59,7 @@ impl WorkerKind { Self::Backup => "backup", Self::AccountImport => "account_import", Self::AccountFreezeRecovery => "account_freeze_recovery", + Self::ResetCreditAutomation => "reset_credit_automation", } } } diff --git a/backend/crates/gateway-host/tests/workers.rs b/backend/crates/gateway-host/tests/workers.rs index bb8962f41..767503d7e 100644 --- a/backend/crates/gateway-host/tests/workers.rs +++ b/backend/crates/gateway-host/tests/workers.rs @@ -19,7 +19,7 @@ use gateway_core::task::{ use gateway_host::workers::{WorkerStartError, WorkerSupervisor}; use tokio::sync::Notify; -const ACTIVE_KINDS: [WorkerKind; 10] = [ +const ACTIVE_KINDS: [WorkerKind; 11] = [ WorkerKind::OAuthRefresh, WorkerKind::QuotaCatalogHealth, WorkerKind::RuntimeSnapshotReconciliation, @@ -30,6 +30,7 @@ const ACTIVE_KINDS: [WorkerKind; 10] = [ WorkerKind::Backup, WorkerKind::AccountImport, WorkerKind::AccountFreezeRecovery, + WorkerKind::ResetCreditAutomation, ]; #[derive(Clone)] diff --git a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs index 209cb0d89..5ae70a05d 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/services/settings.rs @@ -124,6 +124,7 @@ pub struct RuntimeSettings { pub account_warmup_enabled: bool, pub account_warmup_schedule_time: String, pub account_warmup_model: Option, + pub reset_credit_lead_minutes: u32, pub request_profiles: ProviderRequestProfiles, pub config_revision: Revision, pub model_mappings: ModelMappings, @@ -163,6 +164,7 @@ pub struct ReplaceRuntimeSettings { pub account_warmup_enabled: bool, pub account_warmup_schedule_time: String, pub account_warmup_model: Option, + pub reset_credit_lead_minutes: u32, /// 读取设置时的版本；与写入在同一事务内比较，防止覆盖并发更新 pub expected_revision: Revision, /// 只覆盖提交的 Provider；未提交项保留当前持久值 @@ -365,6 +367,7 @@ impl From for ReplaceRuntimeSettings { account_warmup_enabled: settings.account_warmup_enabled, account_warmup_schedule_time: settings.account_warmup_schedule_time, account_warmup_model: settings.account_warmup_model, + reset_credit_lead_minutes: settings.reset_credit_lead_minutes, expected_revision: settings.config_revision, request_profile_updates: settings .request_profiles diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/mod.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/mod.rs index 79bb1e4de..490365751 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/mod.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/mod.rs @@ -60,6 +60,7 @@ mod core_adapter; mod mapping; mod quota_forecast; mod repository; +mod reset_credit_automation; mod rows; mod runtime; diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/reset_credit_automation.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/reset_credit_automation.rs new file mode 100644 index 000000000..24990c2a0 --- /dev/null +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/reset_credit_automation.rs @@ -0,0 +1,134 @@ +//! 账号重置卡计划的 PostgreSQL CAS 与配置审计，不维护独立卡片库存 + +use super::*; +use gateway_admin::{ + model::{ + audit::MutationAuditOperation, + reset_credit_automation::{ResetCreditAutomation, ResetCreditAutomationState}, + }, + ports::reset_credit_automation::ResetCreditAutomationStore, +}; + +#[derive(serde::Serialize, serde::Deserialize)] +struct StoredResetCreditAutomation { + enabled: bool, + revision: u64, + state: ResetCreditAutomationState, +} + +#[async_trait] +impl ResetCreditAutomationStore for PgAdminAccountStore { + async fn reset_credit_automation( + &self, + account_id: &str, + ) -> AdminStoreResult> { + let row = sqlx::query("select reset_credit_automation_json, reset_credit_next_run_at from provider_accounts where id = $1") + .bind(account_id).fetch_optional(&self.pool).await + .map_err(|source| automation_error("load reset-credit automation", source))?; + let Some(row) = row else { + return Ok(None); + }; + let stored: Option> = row + .try_get("reset_credit_automation_json") + .map_err(|source| automation_error("decode reset-credit automation", source))?; + stored + .map(|stored| { + Ok(ResetCreditAutomation { + account_id: account_id.to_owned(), + enabled: stored.enabled, + revision: stored.revision, + next_run_at: row.try_get("reset_credit_next_run_at").map_err(|source| { + automation_error("decode reset-credit deadline", source) + })?, + state: stored.0.state, + }) + }) + .transpose() + } + + async fn due_reset_credit_automations( + &self, + now: DateTime, + limit: u32, + ) -> AdminStoreResult> { + sqlx::query_scalar("select id from provider_accounts where enabled and reset_credit_automation_json ->> 'enabled' = 'true' and reset_credit_next_run_at <= $1 order by reset_credit_next_run_at, id limit $2") + .bind(now).bind(i64::from(limit)).fetch_all(&self.pool).await + .map_err(|source| automation_error("list due reset-credit automation", source)) + } + + async fn save_reset_credit_automation( + &self, + record: &ResetCreditAutomation, + context: Option<&MutationContext>, + ) -> AdminStoreResult { + let mut tx = self + .pool + .begin() + .await + .map_err(|source| automation_error("begin reset-credit automation", source))?; + // 配置写入统一先锁全局版本，再锁账号计划，避免和运行设置重排相互等待 + let config_revision = if context.is_some() { + Some( + bump_config_revision_in_transaction(&mut tx) + .await + .map_err(|error| admin_store_error("reset-credit automation", error))?, + ) + } else { + None + }; + let expected = i64::try_from(record.revision).map_err(|_| invalid_automation_revision())?; + let next = expected + .checked_add(1) + .and_then(|value| u64::try_from(value).ok()) + .ok_or_else(invalid_automation_revision)?; + let stored = StoredResetCreditAutomation { + enabled: record.enabled, + revision: next, + state: record.state.clone(), + }; + let saved = sqlx::query("update provider_accounts set reset_credit_automation_json = $2, reset_credit_next_run_at = $3 where id = $1 and coalesce((reset_credit_automation_json ->> 'revision')::bigint, 0) = $4") + .bind(&record.account_id).bind(sqlx::types::Json(stored)).bind(record.next_run_at).bind(expected) + .execute(&mut *tx).await + .map_err(|source| automation_error("save reset-credit automation", source))?; + if saved.rows_affected() != 1 { + return Err(AdminStoreError::new( + AdminStoreErrorKind::Conflict, + "reset-credit automation", + "automation state changed", + )); + } + if let (Some(context), Some(revision)) = (context, config_revision) { + append_admin_audit_event_in_transaction( + &mut tx, + mutation_audit( + context, + MutationAuditOperation::ProviderAccountUpdate, + &record.account_id, + vec!["reset_credit_automation".to_owned()], + ), + revision, + ) + .await + .map_err(|error| admin_store_error("reset-credit automation", error))?; + } + tx.commit() + .await + .map_err(|source| automation_error("commit reset-credit automation", source))?; + Ok(next) + } +} + +fn invalid_automation_revision() -> AdminStoreError { + AdminStoreError::new( + AdminStoreErrorKind::Invalid, + "reset-credit automation", + "invalid revision", + ) +} + +fn automation_error(operation: &'static str, source: sqlx::Error) -> AdminStoreError { + admin_store_error( + "reset-credit automation", + postgres_unavailable(operation, source), + ) +} diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index 26ed073c2..9afcbc04a 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -159,7 +159,7 @@ pub(crate) async fn load_runtime_settings_from_pool(pool: &PgPool) -> StoreResul account_auto_freeze_window_seconds, account_auto_freeze_duration_seconds, account_auto_freeze_probe_enabled, account_auto_freeze_probe_model, account_auto_freeze_adaptive_concurrency, - account_warmup_enabled, account_warmup_schedule_time, account_warmup_model + account_warmup_enabled, account_warmup_schedule_time, account_warmup_model, reset_credit_lead_minutes from runtime_settings where id = 1", ) .fetch_optional(pool) @@ -337,7 +337,7 @@ pub(crate) async fn load_runtime_settings_in_transaction( account_auto_freeze_window_seconds, account_auto_freeze_duration_seconds, account_auto_freeze_probe_enabled, account_auto_freeze_probe_model, account_auto_freeze_adaptive_concurrency, - account_warmup_enabled, account_warmup_schedule_time, account_warmup_model + account_warmup_enabled, account_warmup_schedule_time, account_warmup_model, reset_credit_lead_minutes from runtime_settings where id = 1", ) .fetch_optional(&mut **transaction) @@ -356,6 +356,12 @@ pub(crate) async fn update_runtime_settings_in_transaction( update: &RuntimeSettingsUpdate, ) -> StoreResult { update.validate()?; + let previous_lead: i32 = sqlx::query_scalar( + "select reset_credit_lead_minutes from runtime_settings where id = 1 for update", + ) + .fetch_one(&mut **transaction) + .await + .map_err(|source| postgres_unavailable("load reset-credit lead time", source))?; let refresh_margin_seconds = i64::try_from(update.values.refresh_margin_seconds).map_err(|_| invalid_numeric())?; let request_profile_updates = update @@ -412,6 +418,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( max_account_rotations = $32, openai_session_affinity_ttl_hours = $33, codex_privacy_policy_json = $34, + reset_credit_lead_minutes = $35, updated_at = now() where id = 1 returning config_revision", @@ -461,6 +468,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( .bind(i64::from(update.values.max_account_rotations)) .bind(i64::from(update.values.openai_session_affinity_ttl_hours)) .bind(sqlx::types::Json(&update.values.codex_privacy_policy)) + .bind(i64::from(update.values.reset_credit_lead_minutes)) .fetch_optional(&mut **transaction) .await .map_err(|source| postgres_unavailable("update runtime settings in transaction", source))? @@ -469,6 +477,12 @@ pub(crate) async fn update_runtime_settings_in_transaction( entity: "runtime settings", id: "1".to_owned(), })?; + // 提前量变化只唤醒本地计划重算，不触发账号全量上游查询 + if i64::from(previous_lead) != i64::from(update.values.reset_credit_lead_minutes) { + sqlx::query("update provider_accounts set reset_credit_next_run_at = least(reset_credit_next_run_at, now()), reset_credit_automation_json = jsonb_set(reset_credit_automation_json, '{revision}', to_jsonb((reset_credit_automation_json ->> 'revision')::bigint + 1)) where reset_credit_automation_json ->> 'enabled' = 'true' ") + .execute(&mut **transaction).await + .map_err(|source| postgres_unavailable("reschedule reset-credit automation", source))?; + } Revision::new(u64::try_from(next).map_err(|_| invalid_numeric())?) } @@ -551,6 +565,7 @@ struct RuntimeSettingsRow { account_warmup_enabled: bool, account_warmup_schedule_time: String, account_warmup_model: Option, + reset_credit_lead_minutes: i32, } fn runtime_settings_from_row(row: RuntimeSettingsRow) -> StoreResult { @@ -618,6 +633,8 @@ fn runtime_settings_from_row(row: RuntimeSettingsRow) -> StoreResult record.revision); + assert!(replanned.next_run_at <= Utc::now()); + record = replanned; + + assert_eq!( + store + .due_reset_credit_automations(now + TimeDelta::seconds(1), 1) + .await + .unwrap(), + [record.account_id.clone()] + ); + repository + .update_provider_account(profile(&record.account_id, "updated profile")) + .await + .unwrap(); + repository + .set_provider_account_enabled(&record.account_id, false) + .await + .unwrap(); + assert_eq!( + store + .reset_credit_automation(&record.account_id) + .await + .unwrap() + .unwrap(), + record, + "account edits preserve the automation plan" + ); + assert!( + store + .due_reset_credit_automations(now + TimeDelta::seconds(1), 10) + .await + .unwrap() + .is_empty() + ); + repository + .set_provider_account_enabled(&record.account_id, true) + .await + .unwrap(); + let stale = record.clone(); + record.enabled = false; + record.revision = store + .save_reset_credit_automation(&record, None) + .await + .unwrap(); + assert_eq!( + store + .save_reset_credit_automation(&stale, None) + .await + .unwrap_err() + .kind(), + gateway_admin::ports::store::AdminStoreErrorKind::Conflict + ); + assert!( + store + .due_reset_credit_automations(now + TimeDelta::seconds(1), 10) + .await + .unwrap() + .is_empty() + ); + sqlx::query("delete from provider_accounts where id = $1") + .bind(&record.account_id) + .execute(&database.pool) + .await + .unwrap(); + assert!( + store + .reset_credit_automation(&record.account_id) + .await + .unwrap() + .is_none() + ); + database.close().await; +} diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index 107e69495..f3f170b64 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -48,6 +48,7 @@ pub(super) fn settings_with_margin(refresh_margin_seconds: u64) -> RuntimeSettin account_warmup_enabled: false, account_warmup_schedule_time: "08:00".to_owned(), account_warmup_model: None, + reset_credit_lead_minutes: 10, }, } } @@ -211,6 +212,7 @@ async fn privacy_steps_migration_preserves_rules_and_other_settings() { .await .unwrap(); expected_row["codex_privacy_policy_json"] = expected.clone(); + expected_row["reset_credit_lead_minutes"] = serde_json::json!(10); // 完整升级还会清理自动回切字段，其余运行设置必须原样保留 expected_row["smart_scheduling_json"] .as_object_mut() diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 131a12e66..6b75da534 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -31,3 +31,4 @@ f10058e2b4c974b6d2ffa11be78b085504a233205253948c03357ef6fe234bca 0030_analytics b49832e205405de7ef078f3b65c0c19f2c030b8984d42411598738b6b6c146bf 0031_proxy_and_privacy.sql a0e409402611dbc76bb71393b8eda7c717756c18974dbabca305fdcb4b115f31 0032_usage_and_scheduling.sql 6acd13f237cd163fdf9b52ec9c003ca75b723980e5db5c52c9776413cfe85c4b 0033_proxy_websocket_capacity.sql +0bcd3b00ad7038d9a62e955ff607032f072b08c9f529cf43f794a9820965bdbd 0034_reset_credit_automation.sql diff --git a/backend/migrations/0034_reset_credit_automation.sql b/backend/migrations/0034_reset_credit_automation.sql new file mode 100644 index 000000000..e12b04de5 --- /dev/null +++ b/backend/migrations/0034_reset_credit_automation.sql @@ -0,0 +1,22 @@ +-- 主动重置卡配置与计划随账号保存，到期索引只覆盖开启自动使用的账号 +ALTER TABLE runtime_settings + ADD COLUMN reset_credit_lead_minutes integer NOT NULL DEFAULT 10 + CHECK (reset_credit_lead_minutes BETWEEN 1 AND 1440); + +ALTER TABLE provider_accounts + ADD COLUMN reset_credit_automation_json jsonb, + ADD COLUMN reset_credit_next_run_at timestamptz, + ADD CONSTRAINT provider_accounts_reset_credit_automation_shape CHECK ( + (reset_credit_automation_json IS NULL AND reset_credit_next_run_at IS NULL) + OR coalesce(( + reset_credit_automation_json IS NOT NULL AND reset_credit_next_run_at IS NOT NULL + AND jsonb_typeof(reset_credit_automation_json) = 'object' + AND jsonb_typeof(reset_credit_automation_json -> 'enabled') = 'boolean' + AND jsonb_typeof(reset_credit_automation_json -> 'revision') = 'number' + AND jsonb_typeof(reset_credit_automation_json -> 'state') = 'object' + ), false) + ); + +CREATE INDEX provider_accounts_reset_credit_due_idx + ON provider_accounts (reset_credit_next_run_at, id) + WHERE enabled AND reset_credit_automation_json ->> 'enabled' = 'true'; diff --git a/docs/api.md b/docs/api.md index 13297bab4..e443a65e5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -713,6 +713,8 @@ config 返回 `{ name, plaintextKey }`，仅读取服务端会话绑定的当前 | `GET` | `/api/admin/accounts/profile-avatar` | `accountId`、可选 `version` | 返回支持头像能力的 Provider 字节流，需要管理员会话；`version` 仅区分浏览器缓存 | | `GET` | `/api/admin/accounts/reset-credits` | `accountId` | 查询支持该能力的 Provider 重置卡，不读取本地库存 | | `POST` | `/api/admin/accounts/reset-credits` | `{ accountId, creditId?, redeemRequestId }` | 使用 UUIDv4 幂等键消费一张 Provider 重置卡 | +| `GET` | `/api/admin/accounts/reset-credits/automation` | `accountId` | 读取自动使用开关与本地计划，不查询上游 | +| `POST` | `/api/admin/accounts/reset-credits/automation` | `{ accountId, enabled }` | 设置账号自动使用开关，默认关闭 | | `GET` | `/api/admin/accounts/models` | `accountId` | 优先读取该 Provider + 套餐的模型 cache，缺失时有限实时拉取 | | `GET` | `/api/admin/accounts/models/catalog` | `accountId` | 读取指定账号的完整 Codex 原生模型目录，返回 `{ modelCount, observedAt, catalog }` | | `POST` | `/api/admin/accounts/models/refresh` | `{ accountId }` | 强制拉取最新模型并覆盖 cache | @@ -1231,8 +1233,8 @@ revision 变化时丢弃结果 ### 主动额度重置卡 -`GET /api/admin/accounts/reset-credits?accountId=...` 每次都查询对应 Provider；后端不把卡片列表写入 -PostgreSQL 或 Redis。当前由 OpenAI Provider 为 OAuth 账号提供，账号视图分别以 `resetCredits` 和 +`GET /api/admin/accounts/reset-credits?accountId=...` 每次都查询对应 Provider，不用本地计划代替库存。 +当前由 OpenAI Provider 为 OAuth 账号提供，账号视图分别以 `resetCredits` 和 `consumeResetCredit` 表示查询和消费能力 查询响应： @@ -1272,6 +1274,19 @@ PostgreSQL 或 Redis。当前由 OpenAI Provider 为 OAuth 账号提供，账号 消费期间账号身份或凭据版本变化时返回结果未知，不将旧结果套用到新绑定账号 +自动使用通过 `/api/admin/accounts/reset-credits/automation` 按账号开启，默认关闭；账号本身停用时暂停执行。 +GET 和 POST 均返回 `enabled`、`leadMinutes`、`nextDiscoveryAtDisplay`、`nextConsumeAtDisplay`、`lastResult`， +时间展示沿用部署时区，无计划时为 `null`。运行设置 `resetCreditLeadMinutes` 默认 10，范围 1～1440 分钟 + +开启后在一分钟内错峰执行首次发现，此后按部署时区每天随机发现一次，普通发现失败也等待下一日。 +已知且有明确到期时间的可用卡片在提前窗口内错峰执行，消费前重新查询；已在其他地方使用或已到期的卡片 +会移出计划，查询失败不直接消费。上游返回 `already_redeemed` 后结束该次操作，不切换另一张卡重试。 +无卡片 ID 或到期时间时不自动消费；每天发现一次可能错过两次发现之间发放并到期的卡片 + +`lastResult` 的 `pending` 表示消费结果待确认；重试沿用已持久化的幂等键，重启或关闭再开启也不换键。 +`expired_unknown` 表示未确认操作的卡片已经到期，停止重试；`reset`、`already_redeemed` 表示对应终态， +`discovery_failed` 表示最近查询失败，`not_redeemed` 表示上游明确未完成消费。自动操作未决且对应卡片尚未到期时，手动消费返回 `409` + ## 6. 账号分组 分组是 Provider-neutral 的账号集合；一个组可包含任意 Provider 账号，一个账号也可属于多个组。 @@ -1453,6 +1468,7 @@ HTTP 返回 `429`，`error.code` 为 `key_daily_budget_exceeded` 或 `key_weekly | [历史保留](#历史保留) | `usageRetentionDays`、`opsEventRetentionDays`、`auditRetentionDays` | | [账号自动冻结](#账号自动冻结) | `accountAutoFreezeEnabled`、`accountAutoFreezeThreshold`、`accountAutoFreezeWindowSeconds`、`accountAutoFreezeDurationSeconds`、`accountAutoFreezeProbeEnabled`、`accountAutoFreezeProbeModel`、`accountAutoFreezeAdaptiveConcurrency` | | [账号预热](#账号预热) | `accountWarmupEnabled`、`accountWarmupScheduleTime`、`accountWarmupModel` | +| [主动额度重置卡](#主动额度重置卡) | `resetCreditLeadMinutes` | ### 历史保留 diff --git a/docs/architecture.md b/docs/architecture.md index 117a9aab8..513ce5367 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -847,7 +847,8 @@ PostgreSQL 周期对账才是正确性基础 | admission、lease、cooldown、circuit、会话亲和、continuation、失败轮次、OAuth pending、目录 cache | Redis | 可重建、可过期的协调状态 | | 控制面统一登录会话与登录限流桶 | Redis | AuthService 唯一拥有；保存 Admin / Key 身份、绑定 ID、绝对有效期和计数，不保存原始凭据 | | 日志、OAuth 恢复记录、在线更新状态、备份暂存 | `.runtime/` | 部署节点本地运行文件 | -| 重置卡库存与消费结果 | 对应 Provider 的上游 | 后端不建立本地卡库存；前端按账号在浏览器会话期间保留最近查询、未决消费幂等键与发送锁 | +| 重置卡库存与消费结果 | 对应 Provider 的上游 | 本地到期计划不能证明卡片仍可用；手动查询直连上游，自动消费前重新核对 | +| 重置卡自动使用配置、发现时间与未决幂等键 | PostgreSQL 账号记录 | Admin 拥有计划与重试编排；账号删除时一并清理，重启沿用已保存时间及幂等键 | | Provider 公开模型与官方发布资料 | Provider/runtime cache | 由官方目录或发布源刷新，与 PostgreSQL 中的用户身份选择分别管理 | | Windows 安装包临时直链 | Host 进程内短缓存 | 按需解析、严格校验、到期前丢弃；不写 PostgreSQL/Redis，也不代理包字节 | @@ -889,8 +890,14 @@ OpenAI 订阅周期属于按需个人信息，不是额度事实。Admin 账号 刷新信息时请求，关闭后取消等待，不维护两套请求状态 主动额度重置是 OpenAI Provider 的不可逆上游操作：列表查询和消费都直接使用当前 Desktop 请求画像； -卡片不写 PostgreSQL/Redis。消费请求携带调用方生成的 UUIDv4 幂等键，同一账号的消费在进程内串行； -发送结果不明确时必须复用原键。确认成功后管理端再显式刷新卡片与 quota，不能直接改本地重置时间 +完整卡片库存不落库。消费请求携带调用方生成的 UUIDv4 幂等键，同一账号的手动与自动消费共用进程内锁； +发送结果不明确时必须复用原键。确认成功后显式回读上游 quota，不能直接改本地重置时间 + +Admin 仅为开启自动使用的账号保存卡片 ID、到期时间、下一次发现时间和未决消费。每日发现按账号独立随机， +已知到期计划在提前窗口前段错峰；Worker 租约限制为单个执行者，每轮最多处理 100 个到期账号，并发复用 +凭据刷新配置。周期扫描只访问本地计划，消费前重新核对上游；外部已消费的卡片移出计划，已消费响应为终态。 +消费幂等键先持久化再发送，结果未确认时退避重试，到期后停止；确认结果先落库再刷新额度。 +计划写入通过版本比较拒绝过期覆盖，全局提前时间变更唤醒本地重算，关闭自动使用保留未决键供重新开启后恢复 ## 10. 观测与后台任务 diff --git a/frontend/src/api/modules/accounts.ts b/frontend/src/api/modules/accounts.ts index 575037290..48a00825d 100644 --- a/frontend/src/api/modules/accounts.ts +++ b/frontend/src/api/modules/accounts.ts @@ -708,3 +708,29 @@ export function getAccountDetail(data: AccountIdParam, options: RequestOptions = ...options, }) } + +export interface AccountResetCreditAutomation { + enabled: boolean + leadMinutes: number + nextDiscoveryAtDisplay: string | null + nextConsumeAtDisplay: string | null + lastResult: string | null +} + +export function getAccountResetCreditAutomation(data: AccountIdParam, options: RequestOptions = {}) { + return request({ + url: '/api/admin/accounts/reset-credits/automation', + method: 'GET', + params: data, + ...options, + }) +} + +export function setAccountResetCreditAutomation(data: AccountIdParam & { enabled: boolean }, options: RequestOptions = {}) { + return request({ + url: '/api/admin/accounts/reset-credits/automation', + method: 'POST', + data, + ...options, + }) +} diff --git a/frontend/src/api/modules/settings/index.ts b/frontend/src/api/modules/settings/index.ts index 354a3d2c3..1d1b109e4 100644 --- a/frontend/src/api/modules/settings/index.ts +++ b/frontend/src/api/modules/settings/index.ts @@ -76,6 +76,7 @@ export interface RuntimeSettings { accountWarmupEnabled: boolean accountWarmupScheduleTime: string accountWarmupModel: string | null + resetCreditLeadMinutes: number updatedAt: string updatedAtDisplay: string } diff --git a/frontend/src/views/accounts/components/account-quota-panel/Credits.vue b/frontend/src/views/accounts/components/account-quota-panel/Credits.vue index fe4c81078..251d901b8 100644 --- a/frontend/src/views/accounts/components/account-quota-panel/Credits.vue +++ b/frontend/src/views/accounts/components/account-quota-panel/Credits.vue @@ -30,7 +30,7 @@ const hasBalance = computed(() => !props.credits?.unlimited && balanceDisplay.va