diff --git a/backend/Cargo.lock b/backend/Cargo.lock index e27a9af0e..930450747 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1491,6 +1491,7 @@ dependencies = [ "secrecy", "serde", "serde_json", + "sha2 0.11.0", "thiserror", "tokio", "tokio-tungstenite 0.29.0", diff --git a/backend/crates/gateway-admin/Cargo.toml b/backend/crates/gateway-admin/Cargo.toml index 7e69fc6ba..0ea8235fe 100644 --- a/backend/crates/gateway-admin/Cargo.toml +++ b/backend/crates/gateway-admin/Cargo.toml @@ -36,7 +36,7 @@ tokio = { version = "1.52", features = ["rt", "time", "macros", "sync"] } [dev-dependencies] sha2 = "0.11" tempfile = "3" -tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt"] } +tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt", "test-util"] } [lints] workspace = true diff --git a/backend/crates/gateway-admin/src/lib.rs b/backend/crates/gateway-admin/src/lib.rs index 065b1f325..e78d3052b 100644 --- a/backend/crates/gateway-admin/src/lib.rs +++ b/backend/crates/gateway-admin/src/lib.rs @@ -27,8 +27,9 @@ pub use use_case::key_usage::KeyUsageService; pub use use_case::{ account_groups::AccountGroupService, accounts::AccountsService, auth::AuthService, backup::BackupService, client_distribution::ClientDistributionService, - client_keys::ClientKeyService, observability::ObservabilityService, openai::OpenAiService, - proxies::ProxiesService, settings::SettingsService, system::SystemService, xai::XaiService, + client_keys::ClientKeyService, import_tasks::ImportTasksService, + observability::ObservabilityService, openai::OpenAiService, proxies::ProxiesService, + settings::SettingsService, system::SystemService, xai::XaiService, }; use model::{AdminError, AdminErrorKind}; @@ -195,9 +196,15 @@ pub struct AdminServices { openai: Arc, xai: Arc, backups: Arc, + import_tasks: Arc, } impl AdminServices { + #[must_use] + pub fn import_tasks(&self) -> &dyn ImportTasksService { + self.import_tasks.as_ref() + } + #[must_use] pub fn key_usage(&self) -> &dyn KeyUsageService { self.key_usage.as_ref() @@ -276,7 +283,7 @@ impl AdminBundle { self.services.clone() } - /// 取出 Backup Worker 贡献；只能调用一次，与其它 Bundle 的贡献一并交给 Host。 + /// 取出 Admin Worker 贡献；只能调用一次，与其它 Bundle 的贡献一并交给 Host。 pub fn take_worker_contributions(&mut self) -> Vec { std::mem::take(&mut self.worker_contributions) } @@ -361,6 +368,21 @@ pub async fn initialize( store.client_keys(), store.observability(), )); + let openai = Arc::new(DefaultOpenAiService::new( + openai, + store.accounts(), + store.proxies(), + snapshot.clone(), + )); + let xai = Arc::new(DefaultXaiService::new( + xai, + store.accounts(), + store.proxies(), + snapshot.clone(), + )); + let import_tasks = + use_case::import_tasks::DefaultImportTasksService::new(openai.clone(), xai.clone()); + let import_task = use_case::import_tasks::ImportTaskWorker(import_tasks.clone()); let services = AdminServices { key_usage, proxies: Arc::new(use_case::proxies::DefaultProxiesService::new( @@ -392,21 +414,25 @@ pub async fn initialize( snapshot.clone(), )), system: Arc::new(DefaultSystemService::new(system)), - openai: Arc::new(DefaultOpenAiService::new( - openai, - store.accounts(), - store.proxies(), - snapshot.clone(), - )), - xai: Arc::new(DefaultXaiService::new( - xai, - store.accounts(), - store.proxies(), - snapshot.clone(), - )), + openai, + xai, + import_tasks, backups, }; - let worker_contributions = backup_worker_contribution(backup_task)?; + let mut worker_contributions = backup_worker_contribution(backup_task)?; + let id = WorkerId::try_new(WorkerKind::AccountImport, "admin") + .map_err(|_| AdminError::internal("导入 Worker ID 不合法"))?; + let restart = DaemonRestartPolicy::try_new(Duration::from_secs(1), Duration::from_secs(60)) + .map_err(|_| AdminError::internal("导入 Worker 重启策略不合法"))?; + let registration = WorkerRegistration::try_new( + id, + WorkerRunnable::Daemon { + restart, + task: Box::new(import_task), + }, + ) + .map_err(|_| AdminError::internal("导入 Worker 注册信息不合法"))?; + worker_contributions.push(WorkerContribution::Registration(registration)); Ok(AdminBundle { services, worker_contributions, diff --git a/backend/crates/gateway-admin/src/model/import_tasks.rs b/backend/crates/gateway-admin/src/model/import_tasks.rs new file mode 100644 index 000000000..0a81b256f --- /dev/null +++ b/backend/crates/gateway-admin/src/model/import_tasks.rs @@ -0,0 +1,84 @@ +//! 进程内账号导入任务；只保存执行所需输入与可公开的结果。 + +use chrono::{DateTime, Utc}; +use gateway_core::{account::ProviderAccountId, routing::ProviderKind}; +use uuid::Uuid; + +use super::{MutationContext, provider_credentials::ImportCredentials}; + +pub const MAX_IMPORT_TASK_ITEMS: usize = 200; + +#[derive(Debug)] +pub struct ImportTaskInput { + pub provider: ProviderKind, + pub command: ImportCredentials, +} + +#[derive(Debug)] +pub struct SubmitImportTask { + pub submission_id: Uuid, + /// API 对完整输入计算摘要；用于区分丢失响应后的重试与修改后的新提交。 + pub fingerprint: [u8; 32], + pub context: MutationContext, + pub items: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ImportItemStatus { + Pending, + Running, + Succeeded, + Failed, + Unknown, + Skipped, +} + +impl ImportItemStatus { + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Pending => "pending", + Self::Running => "running", + Self::Succeeded => "succeeded", + Self::Failed => "failed", + Self::Unknown => "unknown", + Self::Skipped => "skipped", + } + } +} + +#[derive(Debug, Clone)] +pub struct ImportTaskItem { + pub index: usize, + pub provider: ProviderKind, + pub status: ImportItemStatus, + pub account_ids: Vec, + pub message: Option, +} + +#[derive(Debug, Clone, Default)] +pub struct ImportTaskCounts { + pub pending: usize, + pub running: usize, + pub succeeded: usize, + pub failed: usize, + pub unknown: usize, + pub skipped: usize, + pub imported_accounts: usize, +} + +#[derive(Debug, Clone)] +pub struct ImportTaskSummary { + pub task_id: Uuid, + pub created_at: DateTime, + pub finished_at: Option>, + pub stop_requested: bool, + pub total: usize, + pub counts: ImportTaskCounts, +} + +#[derive(Debug, Clone)] +pub struct ImportTaskDetail { + pub summary: ImportTaskSummary, + pub items: Vec, +} diff --git a/backend/crates/gateway-admin/src/model/mod.rs b/backend/crates/gateway-admin/src/model/mod.rs index 581d874d4..c83730a8d 100644 --- a/backend/crates/gateway-admin/src/model/mod.rs +++ b/backend/crates/gateway-admin/src/model/mod.rs @@ -8,6 +8,7 @@ pub mod auth; pub mod backup; pub mod client_distribution; pub mod client_keys; +pub mod import_tasks; pub mod key_usage; pub mod observability; pub mod provider_credentials; diff --git a/backend/crates/gateway-admin/src/use_case/import_tasks.rs b/backend/crates/gateway-admin/src/use_case/import_tasks.rs new file mode 100644 index 000000000..68703c593 --- /dev/null +++ b/backend/crates/gateway-admin/src/use_case/import_tasks.rs @@ -0,0 +1,372 @@ +//! 导入队列由 Host 管理生命周期，HTTP 只负责接受输入和读取快照。 + +use std::{ + collections::VecDeque, + panic::AssertUnwindSafe, + sync::{Arc, Mutex, MutexGuard}, + time::Duration, +}; + +use chrono::Utc; +use futures::{FutureExt as _, StreamExt as _, future::BoxFuture, stream::FuturesUnordered}; +use gateway_core::{ + lifecycle::CancellationToken, + task::{DaemonTask, WorkerTaskError}, +}; +use tokio::{sync::Notify, time::Instant}; +use uuid::Uuid; + +use super::{openai::OpenAiService, xai::XaiService}; +use crate::model::{ + AdminError, AdminErrorKind, MutationActor, MutationContext, + import_tasks::{ + ImportItemStatus, ImportTaskCounts, ImportTaskDetail, ImportTaskInput, ImportTaskItem, + ImportTaskSummary, MAX_IMPORT_TASK_ITEMS, SubmitImportTask, + }, +}; + +const CONCURRENCY: usize = 3; +const MAX_ACTIVE_TASKS: usize = 8; +const MAX_RETAINED_TASKS: usize = 100; +const RETENTION: Duration = Duration::from_secs(3600); + +pub trait ImportTasksService: Send + Sync { + fn submit(&self, command: SubmitImportTask) -> Result; + fn list(&self, context: &MutationContext) -> Vec; + fn detail( + &self, + context: &MutationContext, + task_id: Uuid, + ) -> Result; + fn stop( + &self, + context: &MutationContext, + task_id: Uuid, + ) -> Result; +} + +struct Entry { + task_id: Uuid, + submission_id: Uuid, + fingerprint: [u8; 32], + owner: MutationActor, + created_at: chrono::DateTime, + finished_at: Option>, + finished_instant: Option, + stop_requested: bool, + items: Vec, + inputs: Vec>, +} + +impl Entry { + fn summary(&self) -> ImportTaskSummary { + let mut counts = ImportTaskCounts::default(); + for item in &self.items { + match item.status { + ImportItemStatus::Pending => counts.pending += 1, + ImportItemStatus::Running => counts.running += 1, + ImportItemStatus::Succeeded => counts.succeeded += 1, + ImportItemStatus::Failed => counts.failed += 1, + ImportItemStatus::Unknown => counts.unknown += 1, + ImportItemStatus::Skipped => counts.skipped += 1, + } + counts.imported_accounts += item.account_ids.len(); + } + ImportTaskSummary { + task_id: self.task_id, + created_at: self.created_at, + finished_at: self.finished_at, + stop_requested: self.stop_requested, + total: self.items.len(), + counts, + } + } + + fn detail(&self) -> ImportTaskDetail { + ImportTaskDetail { + summary: self.summary(), + items: self.items.clone(), + } + } + + fn finish_if_done(&mut self) { + if self.finished_at.is_none() + && self.items.iter().all(|item| { + !matches!( + item.status, + ImportItemStatus::Pending | ImportItemStatus::Running + ) + }) + { + self.finished_at = Some(Utc::now()); + self.finished_instant = Some(Instant::now()); + // 终态仅保留结果，连已清空的输入槽位分配也一并释放。 + self.inputs = Vec::new(); + } + } + + fn stop(&mut self) { + if self.finished_at.is_some() { + return; + } + self.stop_requested = true; + for (item, input) in self.items.iter_mut().zip(&mut self.inputs) { + if item.status == ImportItemStatus::Pending { + item.status = ImportItemStatus::Skipped; + *input = None; + } + } + self.finish_if_done(); + } +} + +#[derive(Default)] +struct Registry { + tasks: VecDeque, + shutting_down: bool, +} + +pub(crate) struct DefaultImportTasksService { + registry: Mutex, + notify: Notify, + openai: Arc, + xai: Arc, +} + +impl DefaultImportTasksService { + pub(crate) fn new(openai: Arc, xai: Arc) -> Arc { + Arc::new(Self { + registry: Mutex::default(), + notify: Notify::new(), + openai, + xai, + }) + } + + fn registry(&self) -> MutexGuard<'_, Registry> { + let mut registry = self + .registry + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + registry.tasks.retain(|entry| { + entry + .finished_instant + .is_none_or(|at| at.elapsed() < RETENTION) + }); + registry + } + + fn take_next(&self) -> Option<(Uuid, usize, ImportTaskInput)> { + let mut registry = self.registry(); + if registry.shutting_down { + return None; + } + let task_index = registry + .tasks + .iter() + .position(|entry| entry.inputs.iter().any(Option::is_some))?; + // 轮转任务，避免大批次独占后续全部执行机会。 + let mut entry = registry.tasks.remove(task_index)?; + let index = entry.inputs.iter().position(Option::is_some)?; + let input = entry.inputs[index].take()?; + entry.items[index].status = ImportItemStatus::Running; + let id = entry.task_id; + registry.tasks.push_back(entry); + Some((id, index, input)) + } + + async fn execute(&self, id: Uuid, index: usize, input: ImportTaskInput) { + let operation = async { + match input.provider.as_str() { + "openai" => self.openai.import_document(input.command).await, + "xai" => self.xai.import_document(input.command).await, + _ => Err(AdminError::invalid("不支持的导入平台")), + } + }; + let result = AssertUnwindSafe(operation).catch_unwind().await; + let mut registry = self.registry(); + let Some(entry) = registry.tasks.iter_mut().find(|entry| entry.task_id == id) else { + return; + }; + let item = &mut entry.items[index]; + match result { + Ok(Ok(result)) => { + item.status = ImportItemStatus::Succeeded; + item.account_ids = result.credential_ids; + } + Ok(Err(error)) => { + // 存储或发布异常也可能发生在提交之后；不自动重放可能轮换 RT 的操作。 + item.status = if matches!( + error.kind(), + AdminErrorKind::UpstreamResultUnknown + | AdminErrorKind::Internal + | AdminErrorKind::Unavailable + ) { + ImportItemStatus::Unknown + } else { + ImportItemStatus::Failed + }; + item.message = Some(error.message().to_owned()); + } + Err(_) => { + item.status = ImportItemStatus::Unknown; + item.message = Some("执行中断，请先核对账号列表再决定是否重新导入".to_owned()); + } + } + entry.finish_if_done(); + } +} + +impl ImportTasksService for DefaultImportTasksService { + fn submit(&self, command: SubmitImportTask) -> Result { + if command.items.is_empty() + || command.items.len() > MAX_IMPORT_TASK_ITEMS + || command.items.iter().any(|item| { + !matches!(item.provider.as_str(), "openai" | "xai") + || item.command.context.actor != command.context.actor + }) + { + return Err(AdminError::invalid("导入任务需包含 1–200 条有效输入")); + } + let mut registry = self.registry(); + if let Some(existing) = registry.tasks.iter().find(|entry| { + entry.owner == command.context.actor && entry.submission_id == command.submission_id + }) { + return if existing.fingerprint == command.fingerprint { + Ok(existing.summary()) + } else { + Err(AdminError::new( + AdminErrorKind::Conflict, + "提交标识已用于其他导入内容", + )) + }; + } + if registry.shutting_down { + return Err(AdminError::new( + AdminErrorKind::Unavailable, + "服务正在关闭，暂不接受导入任务", + )); + } + if registry + .tasks + .iter() + .filter(|entry| entry.finished_at.is_none()) + .count() + >= MAX_ACTIVE_TASKS + || registry.tasks.len() >= MAX_RETAINED_TASKS + { + return Err(AdminError::new( + AdminErrorKind::RateLimited, + "导入任务已满，请稍后再试", + )); + } + let items = command + .items + .iter() + .enumerate() + .map(|(index, input)| ImportTaskItem { + index: index + 1, + provider: input.provider.clone(), + status: ImportItemStatus::Pending, + account_ids: Vec::new(), + message: None, + }) + .collect(); + let entry = Entry { + task_id: Uuid::now_v7(), + submission_id: command.submission_id, + fingerprint: command.fingerprint, + owner: command.context.actor, + created_at: Utc::now(), + finished_at: None, + finished_instant: None, + stop_requested: false, + items, + inputs: command.items.into_iter().map(Some).collect(), + }; + let summary = entry.summary(); + registry.tasks.push_back(entry); + drop(registry); + self.notify.notify_one(); + Ok(summary) + } + + fn list(&self, context: &MutationContext) -> Vec { + let mut tasks: Vec<_> = self + .registry() + .tasks + .iter() + .filter(|entry| entry.owner == context.actor) + .map(Entry::summary) + .collect(); + tasks.sort_by_key(|entry| std::cmp::Reverse(entry.task_id)); + tasks + } + + fn detail( + &self, + context: &MutationContext, + task_id: Uuid, + ) -> Result { + self.registry() + .tasks + .iter() + .find(|entry| entry.task_id == task_id && entry.owner == context.actor) + .map(Entry::detail) + .ok_or_else(|| AdminError::not_found("导入任务不存在或已过期")) + } + + fn stop( + &self, + context: &MutationContext, + task_id: Uuid, + ) -> Result { + let mut registry = self.registry(); + let entry = registry + .tasks + .iter_mut() + .find(|entry| entry.task_id == task_id && entry.owner == context.actor) + .ok_or_else(|| AdminError::not_found("导入任务不存在或已过期"))?; + entry.stop(); + Ok(entry.detail()) + } +} + +pub(crate) struct ImportTaskWorker(pub Arc); + +impl DaemonTask for ImportTaskWorker { + fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + Box::pin(async move { + let mut active = FuturesUnordered::new(); + // 固定一个取消 future，避免循环反复注册中立 CancellationToken 的 waiter。 + let cancelled = cancellation.cancelled(); + tokio::pin!(cancelled); + let mut cleanup = tokio::time::interval(Duration::from_secs(60)); + loop { + if cancellation.is_cancelled() { + let mut registry = self.0.registry(); + registry.shutting_down = true; + for entry in &mut registry.tasks { + entry.stop(); + } + break; + } + while active.len() < CONCURRENCY { + let Some((id, index, input)) = self.0.take_next() else { + break; + }; + active.push(self.0.execute(id, index, input)); + } + tokio::select! { + () = &mut cancelled => {}, + _ = active.next(), if !active.is_empty() => {}, + () = self.0.notify.notified() => {}, + _ = cleanup.tick() => { drop(self.0.registry()); }, + } + } + // Host 提供总关闭预算；此处让已发出的交换与入库尽量完成。 + while active.next().await.is_some() {} + Ok(()) + }) + } +} diff --git a/backend/crates/gateway-admin/src/use_case/mod.rs b/backend/crates/gateway-admin/src/use_case/mod.rs index ad83960e4..c1b2a5cd2 100644 --- a/backend/crates/gateway-admin/src/use_case/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/mod.rs @@ -6,6 +6,7 @@ pub mod auth; pub mod backup; pub mod client_distribution; pub mod client_keys; +pub mod import_tasks; pub mod key_usage; pub mod observability; pub mod openai; diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index 8dbaae34f..f36c15d10 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -67,6 +67,7 @@ pub(super) struct FakeProviderAdmin { kind: ProviderKind, events: EventLog, failure: Mutex>, + import_gate: Mutex>>, quota_failure: Mutex>, pending: Arc>>, retry_authorization_after_abort: Mutex, @@ -90,6 +91,7 @@ impl FakeProviderAdmin { kind: ProviderKind::new(kind).expect("provider kind"), events, failure: Mutex::new(None), + import_gate: Mutex::new(None), quota_failure: Mutex::new(None), pending: Arc::new(Mutex::new(None)), retry_authorization_after_abort: Mutex::new(false), @@ -108,6 +110,12 @@ impl FakeProviderAdmin { }) } + pub(super) fn block_imports(&self) -> Arc { + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + *self.import_gate.lock().expect("import gate") = Some(gate.clone()); + gate + } + pub(super) fn fail_next(&self, kind: ProviderAdminErrorKind) { *self.failure.lock().expect("provider failure") = Some(ProviderAdminError::new(kind)); } @@ -333,6 +341,10 @@ impl ProviderAdmin for FakeProviderAdmin { _command: PrepareCredentialImport, ) -> Result { self.record("provider.prepare_import"); + let gate = self.import_gate.lock().expect("import gate").clone(); + if let Some(gate) = gate { + gate.acquire().await.expect("import permit").forget(); + } self.require_available()?; let account_ids = self .import_account_ids diff --git a/backend/crates/gateway-admin/tests/use_case/import_tasks.rs b/backend/crates/gateway-admin/tests/use_case/import_tasks.rs new file mode 100644 index 000000000..41d45798d --- /dev/null +++ b/backend/crates/gateway-admin/tests/use_case/import_tasks.rs @@ -0,0 +1,370 @@ +use std::{sync::Arc, time::Duration}; + +use gateway_admin::{ + AdminBundle, AdminServices, + model::{ + AdminErrorKind, MutationActor, MutationContext, + import_tasks::{ImportTaskInput, SubmitImportTask}, + provider_credentials::ImportCredentials, + }, + ports::provider::ProviderAdminErrorKind, +}; +use gateway_core::{ + lifecycle::CancellationToken, + routing::ProviderKind, + task::{WorkerContribution, WorkerKind, WorkerRunnable}, +}; +use uuid::Uuid; + +use super::{ + AdminHarness, + accounts::{ + EventLog, FakeAccountStore, FakeProviderAdmin, context, document, events, recorded, + }, +}; + +fn command(id: Uuid, count: usize) -> SubmitImportTask { + let context = context("task-submit"); + SubmitImportTask { + submission_id: id, + fingerprint: [1; 32], + context: context.clone(), + items: (0..count) + .map(|_| ImportTaskInput { + provider: ProviderKind::new("openai").unwrap(), + command: ImportCredentials { + outbound_proxy_id: None, + settings: None, + context: context.clone(), + document: document(), + }, + }) + .collect(), + } +} + +async fn harness() -> (AdminBundle, Arc, EventLog) { + let log = events(); + let provider = FakeProviderAdmin::new("openai", log.clone()); + let bundle = AdminHarness::new() + .provider(provider.clone()) + .accounts(FakeAccountStore::new("openai", log.clone())) + .build_bundle() + .await; + (bundle, provider, log) +} + +fn start(bundle: &mut AdminBundle) -> (CancellationToken, tokio::task::JoinHandle<()>) { + let registration = bundle + .take_worker_contributions() + .into_iter() + .find_map(|contribution| match contribution { + WorkerContribution::Registration(registration) + if registration.id.kind() == WorkerKind::AccountImport => + { + Some(registration) + } + _ => None, + }) + .expect("import registration"); + let WorkerRunnable::Daemon { task, .. } = registration.runnable else { + panic!("import daemon") + }; + let cancellation = CancellationToken::new(); + let token = cancellation.clone(); + let handle = tokio::spawn(async move { + task.run(token).await.expect("import daemon exits cleanly"); + }); + (cancellation, handle) +} + +async fn wait_started(log: &EventLog, count: usize) { + tokio::time::timeout(Duration::from_secs(5), async { + while recorded(log) + .iter() + .filter(|event| **event == "provider.prepare_import") + .count() + < count + { + tokio::task::yield_now().await; + } + }) + .await + .expect("imports started"); +} + +async fn wait_finished(services: &AdminServices, id: Uuid) { + tokio::time::timeout(Duration::from_secs(5), async { + while services + .import_tasks() + .detail(&context("poll"), id) + .unwrap() + .summary + .finished_at + .is_none() + { + tokio::task::yield_now().await; + } + }) + .await + .expect("imports finished"); +} + +#[tokio::test] +async fn retry_should_return_the_same_task_but_reject_changed_content() { + let (bundle, _, _) = harness().await; + let services = bundle.services(); + let id = Uuid::now_v7(); + let first = services.import_tasks().submit(command(id, 2)).unwrap(); + let retry = services.import_tasks().submit(command(id, 2)).unwrap(); + assert_eq!(first.task_id, retry.task_id); + assert_eq!( + services + .import_tasks() + .list(&context("reopened-page")) + .len(), + 1 + ); + let mut changed = command(id, 2); + changed.fingerprint = [2; 32]; + assert_eq!( + services.import_tasks().submit(changed).unwrap_err().kind(), + AdminErrorKind::Conflict + ); +} + +#[tokio::test] +async fn task_reads_and_stop_should_be_isolated_by_owner() { + let (bundle, _, _) = harness().await; + let services = bundle.services(); + let task = services + .import_tasks() + .submit(command(Uuid::now_v7(), 1)) + .unwrap(); + let other = MutationContext { + actor: MutationActor::AdminSession { + admin_user_id: "another-admin".to_owned(), + }, + request_id: "other-session".to_owned(), + }; + assert!(services.import_tasks().list(&other).is_empty()); + assert_eq!( + services + .import_tasks() + .detail(&other, task.task_id) + .unwrap_err() + .kind(), + AdminErrorKind::NotFound + ); + assert_eq!( + services + .import_tasks() + .stop(&other, task.task_id) + .unwrap_err() + .kind(), + AdminErrorKind::NotFound + ); +} + +#[tokio::test] +async fn stopping_should_skip_pending_items_and_allow_three_in_flight_to_commit() { + let (mut bundle, provider, log) = harness().await; + let gate = provider.block_imports(); + let services = bundle.services(); + let task = services + .import_tasks() + .submit(command(Uuid::now_v7(), 8)) + .unwrap(); + let (cancel, worker) = start(&mut bundle); + wait_started(&log, 3).await; + let stopped = services + .import_tasks() + .stop(&context("stop"), task.task_id) + .unwrap(); + assert_eq!(stopped.summary.counts.running, 3); + assert_eq!(stopped.summary.counts.skipped, 5); + gate.add_permits(3); + wait_finished(&services, task.task_id).await; + let finished = services + .import_tasks() + .detail(&context("new-page"), task.task_id) + .unwrap(); + assert_eq!(finished.summary.counts.imported_accounts, 3); + assert_eq!( + recorded(&log) + .iter() + .filter(|event| **event == "provider.prepare_import") + .count(), + 3 + ); + cancel.cancel(); + worker.await.unwrap(); +} + +#[tokio::test] +async fn concurrency_limit_should_be_shared_across_tasks() { + let (mut bundle, provider, log) = harness().await; + let gate = provider.block_imports(); + let services = bundle.services(); + let first = services + .import_tasks() + .submit(command(Uuid::now_v7(), 6)) + .unwrap(); + let second = services + .import_tasks() + .submit(command(Uuid::now_v7(), 6)) + .unwrap(); + let (cancel, worker) = start(&mut bundle); + wait_started(&log, 3).await; + let summaries = services.import_tasks().list(&context("poll")); + assert_eq!( + summaries + .iter() + .map(|task| task.counts.running) + .sum::(), + 3 + ); + assert!(summaries.iter().all(|task| task.counts.running > 0)); + gate.add_permits(12); + wait_finished(&services, first.task_id).await; + wait_finished(&services, second.task_id).await; + cancel.cancel(); + worker.await.unwrap(); +} + +#[tokio::test] +async fn failed_and_unknown_items_should_not_abort_the_batch_or_retry_credentials() { + for (kind, failed, unknown) in [ + (ProviderAdminErrorKind::Invalid, 1, 0), + (ProviderAdminErrorKind::Ambiguous, 0, 1), + ] { + let (mut bundle, provider, log) = harness().await; + provider.fail_next(kind); + let services = bundle.services(); + let task = services + .import_tasks() + .submit(command(Uuid::now_v7(), 4)) + .unwrap(); + let (cancel, worker) = start(&mut bundle); + wait_finished(&services, task.task_id).await; + let result = services + .import_tasks() + .detail(&context("poll"), task.task_id) + .unwrap(); + assert_eq!( + ( + result.summary.counts.succeeded, + result.summary.counts.failed, + result.summary.counts.unknown + ), + (3, failed, unknown) + ); + assert_eq!( + recorded(&log) + .iter() + .filter(|event| **event == "provider.prepare_import") + .count(), + 4 + ); + cancel.cancel(); + worker.await.unwrap(); + } +} + +#[tokio::test] +async fn shutdown_should_drain_running_items_and_reject_new_work() { + let (mut bundle, provider, log) = harness().await; + let gate = provider.block_imports(); + let services = bundle.services(); + let task = services + .import_tasks() + .submit(command(Uuid::now_v7(), 7)) + .unwrap(); + let (cancel, worker) = start(&mut bundle); + wait_started(&log, 3).await; + cancel.cancel(); + tokio::task::yield_now().await; + assert_eq!( + services + .import_tasks() + .submit(command(Uuid::now_v7(), 1)) + .unwrap_err() + .kind(), + AdminErrorKind::Unavailable + ); + assert!(!worker.is_finished()); + gate.add_permits(3); + worker.await.unwrap(); + let result = services + .import_tasks() + .detail(&context("poll"), task.task_id) + .unwrap(); + assert_eq!( + ( + result.summary.counts.succeeded, + result.summary.counts.skipped + ), + (3, 4) + ); +} + +#[tokio::test(start_paused = true)] +async fn completed_records_should_expire_and_new_process_should_start_empty() { + let (bundle, _, _) = harness().await; + let services = bundle.services(); + let task = services + .import_tasks() + .submit(command(Uuid::now_v7(), 2)) + .unwrap(); + services + .import_tasks() + .stop(&context("stop"), task.task_id) + .unwrap(); + tokio::time::advance(Duration::from_secs(3601)).await; + assert!(services.import_tasks().list(&context("poll")).is_empty()); + assert_eq!( + services + .import_tasks() + .detail(&context("poll"), task.task_id) + .unwrap_err() + .kind(), + AdminErrorKind::NotFound + ); + let (fresh, _, _) = harness().await; + assert!( + fresh + .services() + .import_tasks() + .list(&context("poll")) + .is_empty() + ); +} + +#[tokio::test] +async fn queue_should_reject_excess_work_before_starting_any_imports() { + let (bundle, _, log) = harness().await; + let services = bundle.services(); + assert_eq!( + services + .import_tasks() + .submit(command(Uuid::now_v7(), 201)) + .unwrap_err() + .kind(), + AdminErrorKind::Invalid + ); + for _ in 0..8 { + services + .import_tasks() + .submit(command(Uuid::now_v7(), 1)) + .unwrap(); + } + assert_eq!( + services + .import_tasks() + .submit(command(Uuid::now_v7(), 1)) + .unwrap_err() + .kind(), + AdminErrorKind::RateLimited + ); + assert!(recorded(&log).is_empty()); +} diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index c28eeba35..4c726f9db 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -4,6 +4,7 @@ mod auth; mod auth_key; mod backup; mod client_keys; +mod import_tasks; mod observability; mod openai; mod proxies; @@ -213,6 +214,10 @@ impl AdminHarness { } pub(super) async fn build(self) -> AdminServices { + self.build_bundle().await.services() + } + + pub(super) async fn build_bundle(self) -> gateway_admin::AdminBundle { gateway_admin::initialize( AdminConfig { session_ttl_minutes: self.session_ttl_minutes, @@ -247,7 +252,6 @@ impl AdminHarness { ) .await .expect("initialize admin test harness") - .services() } } diff --git a/backend/crates/gateway-api/Cargo.toml b/backend/crates/gateway-api/Cargo.toml index 76348eb3d..ac13c5b55 100644 --- a/backend/crates/gateway-api/Cargo.toml +++ b/backend/crates/gateway-api/Cargo.toml @@ -24,6 +24,7 @@ gateway-protocol = { path = "../gateway-protocol" } secrecy = "0.10" serde = { version = "1.0.228", features = ["derive"] } serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] } +sha2 = "0.11.0" thiserror = "2.0.18" tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt", "sync", "time"] } tower-http = { version = "0.7.0", features = ["cors", "fs", "request-id", "timeout", "trace"] } diff --git a/backend/crates/gateway-api/src/admin/accounts/credentials.rs b/backend/crates/gateway-api/src/admin/accounts/credentials.rs index 382f9adec..f65a1a526 100644 --- a/backend/crates/gateway-api/src/admin/accounts/credentials.rs +++ b/backend/crates/gateway-api/src/admin/accounts/credentials.rs @@ -31,7 +31,7 @@ impl AccountProvider { } /// 导入统一设置，复用编辑账号的备注、调度和分组约束。 -#[derive(Debug, Clone, Deserialize)] +#[derive(Debug, Clone, Deserialize, Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct AccountImportSettingsRequest { pub notes: Option, @@ -67,7 +67,7 @@ impl AccountImportSettingsRequest { } /// Provider-owned 账号导入请求；公共 API 不解释 `data` 内部字段。 -#[derive(Deserialize)] +#[derive(Deserialize, Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct AccountImportRequest { pub outbound_proxy_id: Option, diff --git a/backend/crates/gateway-api/src/admin/accounts/handlers.rs b/backend/crates/gateway-api/src/admin/accounts/handlers.rs index aa50d8751..9440d2318 100644 --- a/backend/crates/gateway-api/src/admin/accounts/handlers.rs +++ b/backend/crates/gateway-api/src/admin/accounts/handlers.rs @@ -10,6 +10,7 @@ where S: SessionState + Clone + Send + Sync + 'static, { Router::new() + .merge(super::import_tasks::router::()) .route("/api/admin/accounts", get(list_accounts::)) .route("/api/admin/accounts/detail", get(account_detail::)) .route("/api/admin/accounts/export", get(export_accounts::)) diff --git a/backend/crates/gateway-api/src/admin/accounts/import_tasks.rs b/backend/crates/gateway-api/src/admin/accounts/import_tasks.rs new file mode 100644 index 000000000..c81894960 --- /dev/null +++ b/backend/crates/gateway-api/src/admin/accounts/import_tasks.rs @@ -0,0 +1,240 @@ +//! 后台导入任务的有界请求与安全结果投影。 + +use axum::extract::DefaultBodyLimit; +use gateway_admin::model::import_tasks::{ + ImportTaskDetail, ImportTaskInput, ImportTaskSummary, MAX_IMPORT_TASK_ITEMS, SubmitImportTask, +}; +use sha2::{Digest as _, Sha256}; + +use super::*; +use crate::auth::SessionState; + +pub(super) fn router() -> Router +where + S: SessionState + Clone + Send + Sync + 'static, +{ + Router::new() + .route( + "/api/admin/accounts/import-tasks", + get(list::).post(submit::), + ) + .route("/api/admin/accounts/import-tasks/detail", get(detail::)) + .route("/api/admin/accounts/import-tasks/stop", post(stop::)) + .layer(DefaultBodyLimit::max(4 * 1024 * 1024)) +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct SubmitRequest { + submission_id: String, + items: Vec, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct TaskId { + task_id: String, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct SummaryView { + task_id: String, + created_at: String, + finished_at: Option, + stop_requested: bool, + total: usize, + counts: CountsView, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct CountsView { + pending: usize, + running: usize, + succeeded: usize, + failed: usize, + unknown: usize, + skipped: usize, + imported_accounts: usize, +} + +impl From for SummaryView { + fn from(task: ImportTaskSummary) -> Self { + let counts = task.counts; + Self { + task_id: task.task_id.to_string(), + created_at: task.created_at.to_rfc3339(), + finished_at: task.finished_at.map(|at| at.to_rfc3339()), + stop_requested: task.stop_requested, + total: task.total, + counts: CountsView { + pending: counts.pending, + running: counts.running, + succeeded: counts.succeeded, + failed: counts.failed, + unknown: counts.unknown, + skipped: counts.skipped, + imported_accounts: counts.imported_accounts, + }, + } + } +} + +#[derive(Serialize)] +struct DetailView { + #[serde(flatten)] + summary: SummaryView, + items: Vec, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ItemView { + index: usize, + provider: String, + status: &'static str, + account_ids: Vec, + message: Option, +} + +impl From for DetailView { + fn from(task: ImportTaskDetail) -> Self { + Self { + summary: task.summary.into(), + items: task + .items + .into_iter() + .map(|item| ItemView { + index: item.index, + provider: item.provider.to_string(), + status: item.status.as_str(), + account_ids: item + .account_ids + .into_iter() + .map(|id| id.to_string()) + .collect(), + message: item.message, + }) + .collect(), + } + } +} + +#[derive(Serialize)] +struct ListView { + items: Vec, +} + +async fn submit( + auth: AdminAuth, + State(state): State, + AdminJson(request): AdminJson, +) -> Result +where + S: SessionState + Send + Sync, +{ + let submission_id = Uuid::parse_str(&request.submission_id) + .map_err(|_| map_wire_error(WireValidationError::new("submissionId")))?; + if submission_id.is_nil() { + return Err(map_wire_error(WireValidationError::new("submissionId"))); + } + if request.items.is_empty() || request.items.len() > MAX_IMPORT_TASK_ITEMS { + return Err(map_wire_error(WireValidationError::new("items"))); + } + let encoded = serde_json::to_vec(&request.items) + .map_err(|_| map_wire_error(WireValidationError::new("items")))?; + let fingerprint = Sha256::digest(&encoded).into(); + drop(encoded); + let context = auth.context().mutation_context(); + let items = request + .items + .into_iter() + .map(|item| { + let (provider, command) = item.into_command(context.clone())?; + let provider = ProviderKind::new(match provider { + AccountProvider::OpenAi => "openai", + AccountProvider::Xai => "xai", + }) + .map_err(|_| WireValidationError::new("provider"))?; + Ok(ImportTaskInput { provider, command }) + }) + .collect::, WireValidationError>>() + .map_err(map_wire_error)?; + let task = state + .admin_services() + .import_tasks() + .submit(SubmitImportTask { + submission_id, + fingerprint, + context, + items, + }) + .map_err(map_service_error)?; + Ok(AdminResponse::new( + StatusCode::ACCEPTED, + AdminEnvelope::ok(SummaryView::from(task)), + )) +} + +async fn list(auth: AdminAuth, State(state): State) -> impl IntoResponse +where + S: SessionState + Send + Sync, +{ + let items = state + .admin_services() + .import_tasks() + .list(&auth.context().mutation_context()) + .into_iter() + .map(SummaryView::from) + .collect(); + AdminResponse::new(StatusCode::OK, AdminEnvelope::ok(ListView { items })) +} + +async fn detail( + auth: AdminAuth, + State(state): State, + AdminQuery(query): AdminQuery, +) -> Result +where + S: SessionState + Send + Sync, +{ + let task = state + .admin_services() + .import_tasks() + .detail( + &auth.context().mutation_context(), + parse_task_id(&query.task_id)?, + ) + .map_err(map_service_error)?; + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(DetailView::from(task)), + )) +} + +async fn stop( + auth: AdminAuth, + State(state): State, + AdminJson(request): AdminJson, +) -> Result +where + S: SessionState + Send + Sync, +{ + let task = state + .admin_services() + .import_tasks() + .stop( + &auth.context().mutation_context(), + parse_task_id(&request.task_id)?, + ) + .map_err(map_service_error)?; + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(DetailView::from(task)), + )) +} + +fn parse_task_id(value: &str) -> Result { + Uuid::parse_str(value).map_err(|_| map_wire_error(WireValidationError::new("taskId"))) +} diff --git a/backend/crates/gateway-api/src/admin/accounts/mod.rs b/backend/crates/gateway-api/src/admin/accounts/mod.rs index 27825afaf..5681138f0 100644 --- a/backend/crates/gateway-api/src/admin/accounts/mod.rs +++ b/backend/crates/gateway-api/src/admin/accounts/mod.rs @@ -66,6 +66,7 @@ const MAX_AVATAR_VERSION_BYTES: usize = 32; mod credentials; mod handlers; +mod import_tasks; mod presenter; mod wire; diff --git a/backend/crates/gateway-api/tests/admin/accounts/import_tasks.rs b/backend/crates/gateway-api/tests/admin/accounts/import_tasks.rs new file mode 100644 index 000000000..9ff84594e --- /dev/null +++ b/backend/crates/gateway-api/tests/admin/accounts/import_tasks.rs @@ -0,0 +1,166 @@ +use axum::{ + body::{Body, to_bytes}, + http::{Request, StatusCode, header}, +}; +use gateway_api::admin; +use serde_json::{Value, json}; +use tower::ServiceExt as _; + +use super::super::{AdminTestFixture, AdminTestState}; + +async fn send( + fixture: &AdminTestFixture, + method: &str, + uri: &str, + body: Value, + authenticated: bool, +) -> (StatusCode, Value) { + let mut request = Request::builder() + .method(method) + .uri(uri) + .header("x-request-id", "import-task-contract") + .header(header::CONTENT_TYPE, "application/json"); + if authenticated { + request = request.header(header::COOKIE, "cpr_session=valid-session"); + } + let response = admin::router::() + .with_state(fixture.state()) + .oneshot(request.body(Body::from(body.to_string())).unwrap()) + .await + .unwrap(); + assert_eq!(response.headers()[header::CACHE_CONTROL], "no-store"); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 8 * 1024 * 1024) + .await + .unwrap(); + let value: Value = serde_json::from_slice(&bytes).unwrap(); + assert!(!value.to_string().contains("synthetic-sensitive-token")); + (status, value) +} + +fn input() -> Value { + json!({"submissionId": "01994fcb-a333-7333-8000-000000000001", "items": [{"provider": "openai", "data": {"refreshToken": "synthetic-sensitive-token"}}]}) +} + +#[tokio::test] +async fn import_tasks_should_accept_list_restore_stop_and_deduplicate_without_echoing_credentials() +{ + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let (status, accepted) = send( + &fixture, + "POST", + "/api/admin/accounts/import-tasks", + input(), + true, + ) + .await; + assert_eq!(status, StatusCode::ACCEPTED); + let id = accepted["data"]["taskId"].as_str().unwrap(); + let (_, retried) = send( + &fixture, + "POST", + "/api/admin/accounts/import-tasks", + input(), + true, + ) + .await; + assert_eq!(retried["data"]["taskId"], id); + let (_, list) = send( + &fixture, + "GET", + "/api/admin/accounts/import-tasks", + Value::Null, + true, + ) + .await; + assert_eq!(list["data"]["items"].as_array().unwrap().len(), 1); + let (_, detail) = send( + &fixture, + "GET", + &format!("/api/admin/accounts/import-tasks/detail?taskId={id}"), + Value::Null, + true, + ) + .await; + assert_eq!(detail["data"]["items"][0]["status"], "pending"); + let (_, stopped) = send( + &fixture, + "POST", + "/api/admin/accounts/import-tasks/stop", + json!({"taskId": id}), + true, + ) + .await; + assert_eq!(stopped["data"]["counts"]["skipped"], 1); + assert!(stopped["data"]["finishedAt"].is_string()); + let mut changed = input(); + changed["items"][0]["data"]["refreshToken"] = json!("changed-synthetic-input"); + assert_eq!( + send( + &fixture, + "POST", + "/api/admin/accounts/import-tasks", + changed, + true + ) + .await + .0, + StatusCode::CONFLICT + ); +} + +#[tokio::test] +async fn import_tasks_should_require_admin_and_enforce_input_bounds() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + for (method, uri, body) in [ + ("POST", "/api/admin/accounts/import-tasks", input()), + ("GET", "/api/admin/accounts/import-tasks", Value::Null), + ( + "GET", + "/api/admin/accounts/import-tasks/detail?taskId=01994fcb-a333-7333-8000-000000000001", + Value::Null, + ), + ( + "POST", + "/api/admin/accounts/import-tasks/stop", + json!({"taskId": "01994fcb-a333-7333-8000-000000000001"}), + ), + ] { + assert_eq!( + send(&fixture, method, uri, body, false).await.0, + StatusCode::UNAUTHORIZED + ); + } + for count in [0, 201] { + let mut body = input(); + body["items"] = json!(vec![body["items"][0].clone(); count]); + assert_eq!( + send( + &fixture, + "POST", + "/api/admin/accounts/import-tasks", + body, + true + ) + .await + .0, + StatusCode::BAD_REQUEST + ); + } + let mut body = input(); + body["items"][0]["data"]["refreshToken"] = json!("x".repeat(4 * 1024 * 1024)); + assert_eq!( + send( + &fixture, + "POST", + "/api/admin/accounts/import-tasks", + body, + true + ) + .await + .0, + StatusCode::PAYLOAD_TOO_LARGE + ); +} diff --git a/backend/crates/gateway-api/tests/admin/accounts/mod.rs b/backend/crates/gateway-api/tests/admin/accounts/mod.rs index d9bbb19b7..2d956e27c 100644 --- a/backend/crates/gateway-api/tests/admin/accounts/mod.rs +++ b/backend/crates/gateway-api/tests/admin/accounts/mod.rs @@ -1,4 +1,5 @@ mod handlers; +mod import_tasks; mod presenter; mod personal_info { diff --git a/backend/crates/gateway-api/tests/architecture.rs b/backend/crates/gateway-api/tests/architecture.rs index f1ab43767..8eece02db 100644 --- a/backend/crates/gateway-api/tests/architecture.rs +++ b/backend/crates/gateway-api/tests/architecture.rs @@ -36,6 +36,7 @@ fn source_tree_should_match_frozen_machine_manifest() { "src/admin/account_groups.rs", "src/admin/accounts/credentials.rs", "src/admin/accounts/handlers.rs", + "src/admin/accounts/import_tasks.rs", "src/admin/accounts/mod.rs", "src/admin/accounts/presenter.rs", "src/admin/accounts/wire.rs", @@ -95,6 +96,7 @@ fn test_tree_should_match_frozen_rust_mirror() { let mut expected = vec![ "tests/admin/account_groups.rs", "tests/admin/accounts/handlers.rs", + "tests/admin/accounts/import_tasks.rs", "tests/admin/accounts/mod.rs", "tests/admin/accounts/presenter.rs", "tests/admin/auth.rs", diff --git a/backend/crates/gateway-core/src/task/mod.rs b/backend/crates/gateway-core/src/task/mod.rs index 876ed99e1..166fee18c 100644 --- a/backend/crates/gateway-core/src/task/mod.rs +++ b/backend/crates/gateway-core/src/task/mod.rs @@ -22,10 +22,11 @@ pub enum WorkerKind { Retention, OpsFlush, Backup, + AccountImport, } impl WorkerKind { - pub const ALL: [Self; 8] = [ + pub const ALL: [Self; 9] = [ Self::OAuthRefresh, Self::QuotaCatalogHealth, Self::RuntimeSnapshotReconciliation, @@ -34,6 +35,7 @@ impl WorkerKind { Self::Retention, Self::OpsFlush, Self::Backup, + Self::AccountImport, ]; #[must_use] @@ -47,6 +49,7 @@ impl WorkerKind { Self::Retention => "retention", Self::OpsFlush => "ops_flush", Self::Backup => "backup", + Self::AccountImport => "account_import", } } } diff --git a/backend/crates/gateway-host/tests/workers/mod.rs b/backend/crates/gateway-host/tests/workers/mod.rs index 4bee4d0a7..1a9f87ddd 100644 --- a/backend/crates/gateway-host/tests/workers/mod.rs +++ b/backend/crates/gateway-host/tests/workers/mod.rs @@ -17,7 +17,7 @@ use gateway_core::task::{ use gateway_host::workers::{WorkerStartError, WorkerSupervisor}; use tokio::sync::Notify; -const ACTIVE_KINDS: [WorkerKind; 7] = [ +const ACTIVE_KINDS: [WorkerKind; 8] = [ WorkerKind::OAuthRefresh, WorkerKind::QuotaCatalogHealth, WorkerKind::RuntimeSnapshotReconciliation, @@ -25,6 +25,7 @@ const ACTIVE_KINDS: [WorkerKind; 7] = [ WorkerKind::StaleModelRequestRecovery, WorkerKind::Retention, WorkerKind::Backup, + WorkerKind::AccountImport, ]; #[derive(Clone)] diff --git a/docs/api.md b/docs/api.md index b14de16d2..df4699cdc 100644 --- a/docs/api.md +++ b/docs/api.md @@ -319,6 +319,10 @@ Token 明细、费用明细、用时/首字与状态。Token 和费用复用现 | `GET` | `/api/admin/accounts/detail` | `accountId` | 查询账号详情、额度和本地用量 | | `GET` | `/api/admin/accounts/export` | `accountIds`、`confirm=export_sensitive_accounts` | 显式导出最多 200 个账号的敏感 Provider 文档 | | `POST` | `/api/admin/accounts/import` | `{ provider, data, settings?, outboundProxyId? }` | 导入或按上游身份更新账号，可同时应用调度、分组设置与默认代理 | +| `POST` | `/api/admin/accounts/import-tasks` | `{ submissionId, items: [{ provider, data, settings?, outboundProxyId? }] }` | 接受后台导入，返回 HTTP 202 和任务摘要 | +| `GET` | `/api/admin/accounts/import-tasks` | 无 | 当前管理员仍保留的任务，按创建时间倒序 | +| `GET` | `/api/admin/accounts/import-tasks/detail` | `taskId` | 任务摘要和逐条结果，不含原始凭据 | +| `POST` | `/api/admin/accounts/import-tasks/stop` | `{ taskId }` | 跳过未开始的条目，已开始的条目继续完成 | | `POST` | `/api/admin/accounts/refresh` | `{ accountId }` | 手工刷新 OAuth credential（`idToken` / `accessToken` / `refreshToken`），不刷新额度 | | `POST` | `/api/admin/accounts/recover` | `{ accountId }` | 管理员显式清除该账号的本地错误/额度/cooldown 事实并重新启用，不访问上游 | | `POST` | `/api/admin/accounts/rotate` | OpenAI rotation 字段 | 更新指定 OpenAI 账号的 OAuth token 或 API Key 上游设置 | @@ -456,6 +460,27 @@ OAuth 等待回调期间不持有保护；提交仍拒绝已删除、连接配 - `error`、`providerErrorCode`、`providerErrorType`、`upstreamStatus`、`upstreamContentType` 和 `upstreamBody` 是实际捕获的原始诊断字段；缺失时为 `null`，不会由本地猜测或翻译。 +### 后台导入任务 + +管理端通过后台任务导入账号，关闭页面不会取消执行。`submissionId` 为客户端生成的 UUID；同一管理员在任务记录 +保留期间使用相同标识和相同输入重新提交，会返回已有任务，内容改变则返回 409。修改输入须使用新标识。 + +每个任务接受 1–200 个 `items`，请求体上限为 4 MiB。每个条目沿用下节的 Provider 文档与设置合同，独立调用 +凭据准备、校验、事务提交与快照发布流程。批量 AT / RT 由前端按非空输入顺序拆成单账号条目，因此可逐条统计； +JSON 文件按 Provider 文档分项，不拆解内部代理引用或改变 Provider 对文档的原子性与部分成功语义。 +一个文档可导入多个账号，条目成功数与入库账号数可能不同。 + +摘要字段为 `taskId`、`createdAt`、`finishedAt`（未结束为 null）、`stopRequested`、`total`、`counts`。 +`counts` 包含 `pending`、`running`、`succeeded`、`failed`、`unknown`、`skipped` 和 `importedAccounts`。 +详情追加 `items: [{ index, provider, status, accountIds, message }]`，`index` 从 1 开始；`status` 对应上述前六类状态。 +列表返回 `{ items: [摘要] }`。结果未知时保留 `unknown`，先核对账号目录再决定是否重新导入；服务端不自动重试凭据交换。 +停止请求可重复调用，已结束的任务保持原结果；未知、已过期或其他管理员的任务 ID 返回 404。 + +任务仅保存在单实例进程内，不新增数据库表，也不使用 Redis 保存任务。所有后台导入共享 3 个执行槽位； +最多接受 8 个未结束任务、保留 100 个任务，达到上限返回 429。成功、失败或跳过后立即释放对应输入， +终态结果保留 1 小时后自动清理。服务重启会丢失任务与未执行输入，已提交的账号不受影响； +重新打开页面从服务端恢复当前管理员的任务列表，前端不持久化任务 ID 或凭据。 + ### 账号导入与 OAuth 导入的 `data` 必须是 JSON object，Admin API 请求上限为 64 MiB；Provider 可以收紧限制， diff --git a/docs/architecture.md b/docs/architecture.md index 4c49a337e..0e17c1bae 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -208,6 +208,16 @@ xAI Provider 负责 Codex custom 工具与 Grok function 工具的双向转换 `call_id` 配对；超限或转换失败终止流。默认 `store: false` 的续接由现有会话 owner 重放完整历史； 原生续接按上游约束处理 `instructions` 与 `previous_response_id`，不把协议差异交给 Core。 +### 后台账号导入 + +Admin 拥有进程内导入任务、管理员归属、条目状态与有界队列，以 `account_import` Daemon 贡献给 Host。 +API 只校验任务信封、生成完整输入摘要并投影安全结果，不解释 Provider 文档；每个执行条目复用已有 +OpenAI / xAI 导入用例。并发槽位由单个 Worker 统一管理，条目状态是进度统计的唯一来源。 +停止只跳过待执行输入；Host 关闭时停止接收新任务并在关闭预算内等待已开始条目完成。 +完成或跳过即释放原始输入，终态记录有时限和数量上限，不写 PostgreSQL 或 Redis。 +前端通过当前管理员的任务列表恢复视图，轮询与任务执行互不拥有生命周期。接口、限额与重启语义见 +[后台导入任务](api.md#后台导入任务)。 + ### 账号出站代理 出站代理属于账号配置，由各 Provider 在推理、OAuth 服务端交换/刷新、额度、目录和资料等请求中统一使用。 diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 514a4c0b5..2d1f9ea21 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -3,6 +3,7 @@ export * from './modules/account-groups' // 账号管理 export * from './modules/accounts' + // API Keys 管理 export * from './modules/api-keys' diff --git a/frontend/src/api/modules/accounts.ts b/frontend/src/api/modules/accounts.ts index 52983b8f9..4f2a0c40b 100644 --- a/frontend/src/api/modules/accounts.ts +++ b/frontend/src/api/modules/accounts.ts @@ -301,6 +301,29 @@ export interface AccountImportResponse { accountIds: string[] } +export type ImportItemStatus = 'pending' | 'running' | 'succeeded' | 'failed' | 'unknown' | 'skipped' + +export interface AccountImportTask { + taskId: string + createdAt: string + finishedAt: string | null + stopRequested: boolean + total: number + counts: Record & { importedAccounts: number } +} + +export interface AccountImportTaskItem { + index: number + provider: string + status: ImportItemStatus + accountIds: string[] + message: string | null +} + +export interface AccountImportTaskDetail extends AccountImportTask { + items: AccountImportTaskItem[] +} + export interface AccountOAuthCompleteResponse { accountId: string } @@ -391,6 +414,15 @@ interface AccountImportParam { data: unknown } +interface AccountImportTaskIdParam { + taskId: string +} + +interface CreateAccountImportTaskParam { + submissionId: string + items: AccountImportParam[] +} + interface AccountOAuthStartParam { outboundProxyUrl?: string outboundProxyId?: string @@ -531,6 +563,39 @@ export function importAccounts(data: AccountImportParam, options: RequestOptions }) } +export function createAccountImportTask(data: CreateAccountImportTaskParam) { + return request({ + url: '/api/admin/accounts/import-tasks', + method: 'POST', + data, + }) +} + +export function getAccountImportTasks(options: RequestOptions = {}) { + return request<{ items: AccountImportTask[] }>({ + url: '/api/admin/accounts/import-tasks', + method: 'GET', + ...options, + }) +} + +export function getAccountImportTask(data: AccountImportTaskIdParam, options: RequestOptions = {}) { + return request({ + url: '/api/admin/accounts/import-tasks/detail', + method: 'GET', + params: data, + ...options, + }) +} + +export function stopAccountImportTask(data: AccountImportTaskIdParam) { + return request({ + url: '/api/admin/accounts/import-tasks/stop', + method: 'POST', + data, + }) +} + export function updateAccount(data: AccountUpdateParam) { return request({ url: '/api/admin/accounts/update', diff --git a/frontend/src/utils/uuid.ts b/frontend/src/utils/uuid.ts new file mode 100644 index 000000000..6022c0cde --- /dev/null +++ b/frontend/src/utils/uuid.ts @@ -0,0 +1,8 @@ +export function generateRequestId() { + // 普通 HTTP 管理端没有 randomUUID；getRandomValues 仍可生成密码学安全的 UUIDv4。 + const bytes = globalThis.crypto.getRandomValues(new Uint8Array(16)) + bytes[6] = (bytes[6] & 0x0F) | 0x40 + bytes[8] = (bytes[8] & 0x3F) | 0x80 + const hex = Array.from(bytes, byte => byte.toString(16).padStart(2, '0')).join('') + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}` +} diff --git a/frontend/src/views/accounts/components/AccountFilters.vue b/frontend/src/views/accounts/components/AccountFilters.vue index 091fd99b6..641e65003 100644 --- a/frontend/src/views/accounts/components/AccountFilters.vue +++ b/frontend/src/views/accounts/components/AccountFilters.vue @@ -1,6 +1,6 @@ + + diff --git a/frontend/src/views/accounts/components/AccountImportTasks/index.vue b/frontend/src/views/accounts/components/AccountImportTasks/index.vue new file mode 100644 index 000000000..009234cad --- /dev/null +++ b/frontend/src/views/accounts/components/AccountImportTasks/index.vue @@ -0,0 +1,98 @@ + + + diff --git a/frontend/src/views/accounts/components/AccountImportTasks/presenter.ts b/frontend/src/views/accounts/components/AccountImportTasks/presenter.ts new file mode 100644 index 000000000..267733f38 --- /dev/null +++ b/frontend/src/views/accounts/components/AccountImportTasks/presenter.ts @@ -0,0 +1,42 @@ +import type { AccountImportTask, AccountImportTaskItem, ImportItemStatus } from '@/api' +import dayjs from 'dayjs' + +export const itemStates: Record = { + pending: { label: '等待中', color: 'bg-cp-fill-alter text-cp-text-secondary', fill: 'bg-cp-fill-secondary' }, + running: { label: '处理中', color: 'bg-cp-primary-container text-cp-primary-on-container', fill: 'bg-cp-primary' }, + succeeded: { label: '已入库', color: 'bg-cp-success-container text-cp-success-on-container', fill: 'bg-cp-success' }, + failed: { label: '失败', color: 'bg-cp-error-container text-cp-error-on-container', fill: 'bg-cp-error' }, + unknown: { label: '待核对', color: 'bg-cp-warning-container text-cp-warning-on-container', fill: 'bg-cp-warning' }, + skipped: { label: '未执行', color: 'bg-cp-fill-tertiary text-cp-text-secondary', fill: 'bg-cp-text-quaternary' }, +} + +export const outcomeOrder: ImportItemStatus[] = ['succeeded', 'failed', 'unknown', 'running', 'pending', 'skipped'] + +export function taskLabel(task: AccountImportTask) { + if (!task.finishedAt) + return task.stopRequested ? '正在停止' : task.counts.running > 0 ? '导入中' : '排队中' + if (task.stopRequested) + return '已停止' + return task.counts.failed + task.counts.unknown > 0 ? '已结束 · 有待处理项' : '全部完成' +} + +export function taskTime(value: string) { + return dayjs(value).format('MM-DD HH:mm:ss') +} + +export function processed(task: AccountImportTask) { + return task.total - task.counts.pending - task.counts.running +} + +export function itemDescription(item: AccountImportTaskItem) { + if (item.message) + return item.message.replace(/[。.]\s*$/u, '') + switch (item.status) { + case 'succeeded': return `已保存 ${item.accountIds.length} 个账号` + case 'running': return '正在校验凭据并入库' + case 'pending': return '等待执行' + case 'skipped': return '本条未执行' + case 'failed': return '本条导入未成功' + case 'unknown': return '请检查账号列表，确认是否已入库' + } +} diff --git a/frontend/src/views/accounts/composables/useAccountImportTasks.ts b/frontend/src/views/accounts/composables/useAccountImportTasks.ts new file mode 100644 index 000000000..b73c73d6b --- /dev/null +++ b/frontend/src/views/accounts/composables/useAccountImportTasks.ts @@ -0,0 +1,112 @@ +import type { AccountImportTask, AccountImportTaskDetail } from '@/api' +import { computed, onMounted, onScopeDispose, shallowRef, watch } from 'vue' +import { getAccountImportTask, getAccountImportTasks, stopAccountImportTask } from '@/api' +import { useAsyncAction } from '@/composables/useAsyncAction' +import { errorMessage } from '@/utils/async' + +export function useAccountImportTasks(options: { reload: () => Promise }) { + const open = shallowRef(false) + const tasks = shallowRef([]) + const selectedId = shallowRef('') + const detail = shallowRef(null) + const error = shallowRef('') + const loading = shallowRef(false) + const stopAction = useAsyncAction() + const activeCount = computed(() => tasks.value.filter(task => !task.finishedAt).length) + let timer: ReturnType | undefined + let controller: AbortController | undefined + let disposed = false + let initialized = false + + async function refresh(background = false) { + clearTimeout(timer) + controller?.abort() + const current = new AbortController() + controller = current + if (!background) + loading.value = true + try { + const result = await getAccountImportTasks({ + silent: true, + signal: current.signal, + }) + if (current.signal.aborted) + return + const previous = new Map(tasks.value.map(task => [task.taskId, task.counts.importedAccounts])) + // 首次查询只建立进度基线，历史入库结果不应重复触发页面的初始加载。 + const changed = initialized && result.items.some(task => task.counts.importedAccounts > (previous.get(task.taskId) ?? 0)) + initialized = true + tasks.value = result.items + if (changed) + void options.reload().catch(() => undefined) + if (!tasks.value.some(task => task.taskId === selectedId.value)) { + detail.value = null + selectedId.value = tasks.value[0]?.taskId ?? '' + } + if (open.value && selectedId.value) { + const selected = selectedId.value + const next = await getAccountImportTask({ + taskId: selected, + }, { + silent: true, + signal: current.signal, + }) + if (!current.signal.aborted && selectedId.value === selected) + detail.value = next + } + if (current.signal.aborted) + return + error.value = '' + } + catch (cause) { + if (!current.signal.aborted) + error.value = errorMessage(cause, '进度读取失败，请重试') + } + finally { + if (!current.signal.aborted && !disposed) { + loading.value = false + // 面板打开时也查询终态列表，以发现其他页面提交的任务和过期记录。 + if (open.value || activeCount.value > 0) + timer = setTimeout(() => void refresh(true), error.value ? 5000 : 1500) + } + } + } + + function select(taskId: string) { + selectedId.value = taskId + detail.value = null + void refresh() + } + + function created(task: AccountImportTask) { + tasks.value = [task, ...tasks.value.filter(entry => entry.taskId !== task.taskId)] + selectedId.value = task.taskId + detail.value = null + if (open.value) + void refresh() + else + open.value = true + } + + async function stop() { + const taskId = selectedId.value + if (!taskId) + return + await stopAction.run(async () => { + await stopAccountImportTask({ + taskId, + }) + await refresh() + }) + } + + watch(open, () => void refresh()) + onMounted(() => void refresh()) + onScopeDispose(() => { + disposed = true + controller?.abort() + clearTimeout(timer) + }) + + return { open, tasks, detail, selectedId, error, loading, activeCount, stopping: stopAction.loading, select, created, refresh, stop } +} diff --git a/frontend/src/views/accounts/composables/useAccountMutations.ts b/frontend/src/views/accounts/composables/useAccountMutations.ts index c85c092dc..24f430a50 100644 --- a/frontend/src/views/accounts/composables/useAccountMutations.ts +++ b/frontend/src/views/accounts/composables/useAccountMutations.ts @@ -1,5 +1,5 @@ import type { Ref } from 'vue' -import type { getAccounts } from '@/api' +import type { AccountImportTask, getAccounts } from '@/api' import type { RequestOptions } from '@/api/request' import dayjs from 'dayjs' import { ref, watch } from 'vue' @@ -23,6 +23,7 @@ type AccountRow = Awaited>['items'][number] export function useAccountMutations(options: { accounts: Ref selectedIds: Ref> + onImportTaskCreated: (task: AccountImportTask) => void reload: () => Promise replaceAccount: (account: AccountRow) => Promise }) { @@ -30,6 +31,7 @@ export function useAccountMutations(options: { const { downloadJson } = useDownload() const onboarding = useAccountOnboarding({ reload: loadAccounts, + onImportTaskCreated: options.onImportTaskCreated, }) const selectedAccountsById = new Map() const showDeleteModal = ref(false) diff --git a/frontend/src/views/accounts/composables/useAccountOnboarding.ts b/frontend/src/views/accounts/composables/useAccountOnboarding.ts index 8c8b3d743..29b329bec 100644 --- a/frontend/src/views/accounts/composables/useAccountOnboarding.ts +++ b/frontend/src/views/accounts/composables/useAccountOnboarding.ts @@ -1,16 +1,17 @@ -import type { getAccounts } from '@/api' +import type { AccountImportTask, getAccounts } from '@/api' import { computed, ref, shallowRef, watch } from 'vue' import { completeAccountOAuth, + createAccountImportTask, importAccounts, startAccountOAuth, } from '@/api' import { toast } from '@/components/base/BaseToast' import { useAsyncAction } from '@/composables/useAsyncAction' -import { errorMessage } from '@/utils/async' import { isRecord } from '@/utils/object' import { formatProviderLabel, isSupportedProvider } from '@/utils/providers' +import { generateRequestId } from '@/utils/uuid' import { accountImportSettings, accountProxyError, emptyAccountCreateForm } from '../components/AccountCreateModal/model' import { apiKeyAccountError, emptyApiKeyAccountForm } from '../utils/upstreamApiKey' @@ -28,6 +29,7 @@ const MAX_TOKEN_IMPORT_COUNT = 200 export function useAccountOnboarding(options: { reload: () => Promise + onImportTaskCreated: (task: AccountImportTask) => void }) { const createModalOpen = shallowRef(false) const reauthorizingAccount = shallowRef(null) @@ -36,6 +38,10 @@ export function useAccountOnboarding(options: { const creatingAccount = creatingAccountAction.loading const authorizingOAuth = authorizingOAuthAction.loading const createForm = ref(emptyAccountCreateForm()) + let submissionId: string | undefined + watch(createForm, () => { + submissionId = undefined + }, { deep: true, flush: 'sync' }) const showCreateModal = computed({ get: () => createModalOpen.value, @@ -61,10 +67,44 @@ export function useAccountOnboarding(options: { const proxyError = accountProxyError(createForm.value) if (proxyError) throw new Error(proxyError) - const message = createForm.value.provider === 'batch' - ? await importMixedAccountDocument() - : await importAccountDocument() - await finishCreate(message) + const mode = createForm.value.mode + if (mode === 'oauth') + throw new Error('请选择凭据导入方式') + if (mode === 'api_key') { + const provider = requireImportProvider(createForm.value.provider) + if (provider !== 'openai') + throw new Error('当前平台不支持 API Key 账号') + const form = createForm.value.apiKey + const error = apiKeyAccountError(form) + if (error) + throw new Error(error) + await importAccounts({ + provider, + settings: accountImportSettings(createForm.value), + outboundProxyId: createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined, + data: { provider, authentication_kind: 'api_key', name: form.name.trim(), base_url: form.base_url.trim(), api_key: form.apiKey, transport: form.transport }, + }) + await finishCreate('API Key 账号已添加') + return + } + const documents = createForm.value.provider === 'batch' + ? parseMixedImportDocuments(parseImportJson(createForm.value.importTexts.json)) + : accountImportDocuments(requireImportProvider(createForm.value.provider), mode, createForm.value.importTexts[mode]) + if (documents.length > MAX_TOKEN_IMPORT_COUNT) + throw new Error(`单次最多导入 ${MAX_TOKEN_IMPORT_COUNT} 个条目`) + submissionId ??= generateRequestId() + const task = await createAccountImportTask({ + submissionId, + items: documents.map(entry => ({ + provider: entry.provider, + data: entry.document, + settings: accountImportSettings(createForm.value), + outboundProxyId: createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined, + })), + }) + showCreateModal.value = false + options.onImportTaskCreated(task) + toast.success('导入任务已创建') }, ) } @@ -161,75 +201,6 @@ export function useAccountOnboarding(options: { } } - async function importAccountDocument() { - const provider = requireImportProvider(createForm.value.provider) - const mode = createForm.value.mode - if (mode === 'oauth') - throw new Error('请选择凭据导入方式') - if (mode === 'api_key') { - if (provider !== 'openai') - throw new Error('当前平台不支持 API Key 账号') - const form = createForm.value.apiKey - const error = apiKeyAccountError(form) - if (error) - throw new Error(error) - await importAccounts({ - provider, - settings: accountImportSettings(createForm.value), - outboundProxyId: createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined, - data: { provider, authentication_kind: 'api_key', name: form.name.trim(), base_url: form.base_url.trim(), api_key: form.apiKey, transport: form.transport }, - }) - return 'API Key 账号已添加' - } - const documents = accountImportDocuments( - provider, - mode, - createForm.value.importTexts[mode], - ) - let importedCount = 0 - for (const entry of documents) { - const result = await importAccounts({ - provider, - settings: accountImportSettings(createForm.value), - outboundProxyId: createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined, - data: entry.document, - }) - importedCount += result.importedCount - } - return `${formatProviderLabel(provider)} 账号已导入 ${importedCount} 个` - } - - async function importMixedAccountDocument() { - const documents = parseMixedImportDocuments(parseImportJson(createForm.value.importTexts.json)) - let importedCount = 0 - const failures: string[] = [] - - for (const entry of documents) { - try { - const result = await importAccounts({ - provider: entry.provider, - settings: accountImportSettings(createForm.value), - outboundProxyId: createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined, - data: entry.document, - }, { silent: true }) - importedCount += result.importedCount - } - catch (error) { - const message = errorMessage(error, '导入失败') - failures.push(`${formatProviderLabel(entry.provider, 'OpenAI')}：${message}`) - } - } - - if (importedCount === 0) { - throw new Error(failures.length > 0 ? `批量导入失败：${failures.join('、')}` : '批量文件没有可导入的账号') - } - - if (failures.length > 0) { - return `已导入 ${importedCount} 个账号，${failures.length} 个文档失败：${failures.join('、')}` - } - return `已导入 ${importedCount} 个账号` - } - async function finishCreate(message: string) { showCreateModal.value = false await options.reload() @@ -299,10 +270,7 @@ function accountImportDocuments( value: string, ): MixedImportDocument[] { if (provider === 'openai' && isOpenAiTokenImportMode(mode)) { - return [{ - provider, - document: parseOpenAiTokenImport(value, mode), - }] + return parseOpenAiTokenImport(value, mode).map(document => ({ provider, document })) } return providerImportDocuments(parseImportJson(value), provider) } @@ -320,9 +288,7 @@ function parseOpenAiTokenImport(value: string, mode: OpenAiTokenImportMode) { throw new Error(`单次最多导入 ${MAX_TOKEN_IMPORT_COUNT} 个 ${label}`) const credentialKey = mode === 'access_token' ? 'accessToken' : 'refreshToken' - return { - accounts: tokens.map(token => ({ [credentialKey]: token })), - } + return tokens.map(token => ({ accounts: [{ [credentialKey]: token }] })) } function isOpenAiTokenImportMode(value: string): value is OpenAiTokenImportMode { diff --git a/frontend/src/views/accounts/composables/useAccountResetCredits.ts b/frontend/src/views/accounts/composables/useAccountResetCredits.ts index 9acd9a5c9..a2830bdc7 100644 --- a/frontend/src/views/accounts/composables/useAccountResetCredits.ts +++ b/frontend/src/views/accounts/composables/useAccountResetCredits.ts @@ -1,14 +1,15 @@ import type { Account, AccountResetCredit } from '@/api' import { computed, shallowReactive, shallowRef, watch } from 'vue' - import { consumeAccountResetCredit, getAccountResetCredits, refreshAccountQuota, } from '@/api' + import { ApiError } from '@/api/request' import { toast } from '@/components/base/BaseToast' import { errorMessage } from '@/utils/async' +import { generateRequestId } from '@/utils/uuid' interface PendingResetCreditOperation { accountId: string @@ -172,7 +173,7 @@ export function useAccountResetCredits(options: { accountId: target.accountId, creditId: credit?.id, credit, - redeemRequestId: generateRedeemRequestId(), + redeemRequestId: generateRequestId(), hasTransportFailure: false, } : null) @@ -277,15 +278,6 @@ export function useAccountResetCredits(options: { } } -function generateRedeemRequestId() { - // 普通 HTTP 管理端没有 randomUUID；getRandomValues 仍可生成密码学安全的 UUIDv4。 - const bytes = globalThis.crypto.getRandomValues(new Uint8Array(16)) - bytes[6] = (bytes[6] & 0x0F) | 0x40 - bytes[8] = (bytes[8] & 0x3F) | 0x80 - const hex = Array.from(bytes, byte => byte.toString(16).padStart(2, '0')).join('') - return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}` -} - function isAmbiguousConsumeError(error: unknown) { if (!(error instanceof ApiError)) return false diff --git a/frontend/src/views/accounts/index.vue b/frontend/src/views/accounts/index.vue index 31b56d94d..0e7f6d1b1 100644 --- a/frontend/src/views/accounts/index.vue +++ b/frontend/src/views/accounts/index.vue @@ -20,6 +20,7 @@ import AccountCreateModal from './components/AccountCreateModal/index.vue' import AccountEditModal from './components/AccountEditModal.vue' import AccountFilters from './components/AccountFilters.vue' import AccountIdentityCell from './components/AccountIdentityCell.vue' +import AccountImportTasks from './components/AccountImportTasks/index.vue' import AccountOverviewCards from './components/AccountOverviewCards.vue' import AccountPlanBadge from './components/AccountPlanBadge.vue' import AccountQuotaPanel from './components/AccountQuotaPanel/index.vue' @@ -30,6 +31,7 @@ import AccountUsagePanel from './components/AccountUsagePanel.vue' import { useAccountBatchEditor } from './composables/useAccountBatchEditor' import { useAccountConnectionTest } from './composables/useAccountConnectionTest' import { useAccountEditor } from './composables/useAccountEditor' +import { useAccountImportTasks } from './composables/useAccountImportTasks' import { useAccountMutations } from './composables/useAccountMutations' import { useAccountsQuery } from './composables/useAccountsQuery' import { useAccountsTable } from './composables/useAccountsTable' @@ -61,6 +63,20 @@ const { loadGroups, } = useAccountGroupCatalog() +const importTasks = useAccountImportTasks({ + reload: () => Promise.all([loadAccounts(), loadGroups()]), +}) +const { + open: showImportTasks, + tasks: recentImportTasks, + selectedId: importTaskId, + detail: importTaskDetail, + loading: loadingImportTasks, + stopping: stoppingImportTask, + error: importTaskError, + activeCount: activeImportCount, +} = importTasks + const { showCreateModal, showDeleteModal, @@ -88,6 +104,7 @@ const { handleRefresh, handleRefreshQuota, } = useAccountMutations({ + onImportTaskCreated: importTasks.created, accounts, selectedIds, reload: () => Promise.all([loadAccounts(), loadGroups()]), @@ -195,6 +212,9 @@ const { :selected-count="selectedIds.size" :batch-deleting="batchDeleting" :exporting-accounts="exportingAccounts" + :has-import-tasks="recentImportTasks.length > 0" + :active-import-count="activeImportCount" + @import-tasks="showImportTasks = true" @delete-selected="showDeleteModal = true" @export-selected="handleExportAccounts" @create="openCreateAccount" @@ -353,6 +373,20 @@ const { @test="handleTestConnection()" /> + +