fix(accounts): run bulk imports as background tasks (#119)

* fix(openai): sync account plans with quota updates

Co-authored-by: Codex <noreply@openai.com>

* fix(accounts): run bulk imports as background tasks

Co-authored-by: Codex <noreply@openai.com>

* test(store): preserve settings in account plan refresh fixture

Co-authored-by: Codex <noreply@openai.com>

---------

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-09-16 18:24:34 +08:00
committed by GitHub
co-authored by Codex Zyy
parent 866f71e3fb
commit d162f9b7f6
34 changed files with 1896 additions and 117 deletions
+1
View File
@@ -1491,6 +1491,7 @@ dependencies = [
"secrecy",
"serde",
"serde_json",
"sha2 0.11.0",
"thiserror",
"tokio",
"tokio-tungstenite 0.29.0",
+1 -1
View File
@@ -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
+42 -16
View File
@@ -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<dyn OpenAiService>,
xai: Arc<dyn XaiService>,
backups: Arc<dyn BackupService>,
import_tasks: Arc<dyn ImportTasksService>,
}
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<WorkerContribution> {
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,
@@ -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<ImportTaskInput>,
}
#[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<ProviderAccountId>,
pub message: Option<String>,
}
#[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<Utc>,
pub finished_at: Option<DateTime<Utc>>,
pub stop_requested: bool,
pub total: usize,
pub counts: ImportTaskCounts,
}
#[derive(Debug, Clone)]
pub struct ImportTaskDetail {
pub summary: ImportTaskSummary,
pub items: Vec<ImportTaskItem>,
}
@@ -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;
@@ -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<ImportTaskSummary, AdminError>;
fn list(&self, context: &MutationContext) -> Vec<ImportTaskSummary>;
fn detail(
&self,
context: &MutationContext,
task_id: Uuid,
) -> Result<ImportTaskDetail, AdminError>;
fn stop(
&self,
context: &MutationContext,
task_id: Uuid,
) -> Result<ImportTaskDetail, AdminError>;
}
struct Entry {
task_id: Uuid,
submission_id: Uuid,
fingerprint: [u8; 32],
owner: MutationActor,
created_at: chrono::DateTime<Utc>,
finished_at: Option<chrono::DateTime<Utc>>,
finished_instant: Option<Instant>,
stop_requested: bool,
items: Vec<ImportTaskItem>,
inputs: Vec<Option<ImportTaskInput>>,
}
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<Entry>,
shutting_down: bool,
}
pub(crate) struct DefaultImportTasksService {
registry: Mutex<Registry>,
notify: Notify,
openai: Arc<dyn OpenAiService>,
xai: Arc<dyn XaiService>,
}
impl DefaultImportTasksService {
pub(crate) fn new(openai: Arc<dyn OpenAiService>, xai: Arc<dyn XaiService>) -> Arc<Self> {
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<ImportTaskSummary, AdminError> {
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<ImportTaskSummary> {
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<ImportTaskDetail, AdminError> {
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<ImportTaskDetail, AdminError> {
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<DefaultImportTasksService>);
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(())
})
}
}
@@ -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;
@@ -67,6 +67,7 @@ pub(super) struct FakeProviderAdmin {
kind: ProviderKind,
events: EventLog,
failure: Mutex<Option<ProviderAdminError>>,
import_gate: Mutex<Option<Arc<tokio::sync::Semaphore>>>,
quota_failure: Mutex<Option<ProviderAdminErrorKind>>,
pending: Arc<Mutex<Option<PendingAuthorizationMutation>>>,
retry_authorization_after_abort: Mutex<bool>,
@@ -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<tokio::sync::Semaphore> {
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<PreparedCredentialImport, ProviderAdminError> {
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
@@ -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<FakeProviderAdmin>, 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::<usize>(),
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());
}
@@ -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()
}
}
+1
View File
@@ -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"] }
@@ -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<String>,
@@ -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<String>,
@@ -10,6 +10,7 @@ where
S: SessionState + Clone + Send + Sync + 'static,
{
Router::new()
.merge(super::import_tasks::router::<S>())
.route("/api/admin/accounts", get(list_accounts::<S>))
.route("/api/admin/accounts/detail", get(account_detail::<S>))
.route("/api/admin/accounts/export", get(export_accounts::<S>))
@@ -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<S>() -> Router<S>
where
S: SessionState + Clone + Send + Sync + 'static,
{
Router::new()
.route(
"/api/admin/accounts/import-tasks",
get(list::<S>).post(submit::<S>),
)
.route("/api/admin/accounts/import-tasks/detail", get(detail::<S>))
.route("/api/admin/accounts/import-tasks/stop", post(stop::<S>))
.layer(DefaultBodyLimit::max(4 * 1024 * 1024))
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct SubmitRequest {
submission_id: String,
items: Vec<AccountImportRequest>,
}
#[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<String>,
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<ImportTaskSummary> 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<ItemView>,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct ItemView {
index: usize,
provider: String,
status: &'static str,
account_ids: Vec<String>,
message: Option<String>,
}
impl From<ImportTaskDetail> 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<SummaryView>,
}
async fn submit<S>(
auth: AdminAuth,
State(state): State<S>,
AdminJson(request): AdminJson<SubmitRequest>,
) -> Result<impl IntoResponse, AdminError>
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::<Result<Vec<_>, 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<S>(auth: AdminAuth, State(state): State<S>) -> 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<S>(
auth: AdminAuth,
State(state): State<S>,
AdminQuery(query): AdminQuery<TaskId>,
) -> Result<impl IntoResponse, AdminError>
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<S>(
auth: AdminAuth,
State(state): State<S>,
AdminJson(request): AdminJson<TaskId>,
) -> Result<impl IntoResponse, AdminError>
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, AdminError> {
Uuid::parse_str(value).map_err(|_| map_wire_error(WireValidationError::new("taskId")))
}
@@ -66,6 +66,7 @@ const MAX_AVATAR_VERSION_BYTES: usize = 32;
mod credentials;
mod handlers;
mod import_tasks;
mod presenter;
mod wire;
@@ -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::<AdminTestState>()
.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
);
}
@@ -1,4 +1,5 @@
mod handlers;
mod import_tasks;
mod presenter;
mod personal_info {
@@ -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",
+4 -1
View File
@@ -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",
}
}
}
@@ -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)]
+25
View File
@@ -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 可以收紧限制,
+10
View File
@@ -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 服务端交换/刷新、额度、目录和资料等请求中统一使用。
+1
View File
@@ -3,6 +3,7 @@ export * from './modules/account-groups'
// 账号管理
export * from './modules/accounts'
// API Keys 管理
export * from './modules/api-keys'
+65
View File
@@ -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<ImportItemStatus, number> & { 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<AccountImportTask>({
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<AccountImportTaskDetail>({
url: '/api/admin/accounts/import-tasks/detail',
method: 'GET',
params: data,
...options,
})
}
export function stopAccountImportTask(data: AccountImportTaskIdParam) {
return request<AccountImportTaskDetail>({
url: '/api/admin/accounts/import-tasks/stop',
method: 'POST',
data,
})
}
export function updateAccount(data: AccountUpdateParam) {
return request<AccountUpdateResponse>({
url: '/api/admin/accounts/update',
+8
View File
@@ -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)}`
}
@@ -1,6 +1,6 @@
<script setup lang="ts">
import type { AccountGroup } from '@/api'
import { Download, Pencil, Search, Trash2, Upload } from '@lucide/vue'
import { Download, ListTodo, Pencil, Search, Trash2, Upload } from '@lucide/vue'
import { computed } from 'vue'
import BaseButton from '@/components/base/BaseButton.vue'
@@ -10,6 +10,8 @@ import ProviderFilterSegmented from '@/components/ProviderFilterSegmented.vue'
import { accountStatusFilterOptions } from '../constants'
const props = defineProps<{
hasImportTasks: boolean
activeImportCount: number
selectedCount: number
batchDeleting: boolean
exportingAccounts: boolean
@@ -20,6 +22,7 @@ const props = defineProps<{
const emit = defineEmits<{
deleteSelected: []
exportSelected: []
importTasks: []
create: []
editSelected: []
}>()
@@ -116,6 +119,10 @@ const groupOptions = computed(() => [
class="col-span-2 flex min-w-0 items-center justify-end gap-2"
:class="selectedCount > 0 ? 'w-full xl:w-auto' : 'justify-self-end'"
>
<BaseButton v-if="hasImportTasks" variant="secondary" class="whitespace-nowrap" @click="emit('importTasks')">
<ListTodo class="size-4" />
导入任务 <span v-if="activeImportCount" class="font-mono text-cp-link">{{ activeImportCount }}</span>
</BaseButton>
<slot name="actions" />
<BaseButton
variant="primary"
@@ -0,0 +1,130 @@
<script setup lang="ts">
import type { AccountImportTaskDetail } from '@/api'
import { ArrowUpRight, Check, CircleAlert, Square, X } from '@lucide/vue'
import { computed, shallowRef, watch } from 'vue'
import BaseButton from '@/components/base/BaseButton.vue'
import BaseEmpty from '@/components/base/BaseEmpty.vue'
import BaseScrollbar from '@/components/base/BaseScrollbar.vue'
import BaseSegmented from '@/components/base/BaseSegmented.vue'
import ProviderIconGroup from '@/components/ProviderIconGroup.vue'
import { itemDescription, itemStates, outcomeOrder, processed, taskLabel, taskTime } from './presenter'
const props = defineProps<{ task: AccountImportTaskDetail, stopping: boolean }>()
const emit = defineEmits<{ stop: [], viewAccounts: [] }>()
const filter = shallowRef('all')
const filterOptions = [
{ label: '全部', value: 'all' },
{ label: '需处理', value: 'attention' },
]
const hasAttention = computed(() => props.task.counts.failed + props.task.counts.unknown > 0)
const visibleItems = computed(() => filter.value === 'attention'
? props.task.items.filter(item => item.status === 'failed' || item.status === 'unknown')
: props.task.items)
watch(() => props.task.taskId, () => {
filter.value = hasAttention.value ? 'attention' : 'all'
}, { immediate: true })
watch(hasAttention, (value) => {
if (value) {
filter.value = 'attention'
}
})
</script>
<template>
<section class="min-w-0 space-y-5" aria-label="导入任务详情">
<div class="flex flex-wrap items-start justify-between gap-3">
<div>
<h3 class="text-base font-bold text-cp-text">
{{ taskLabel(task) }}
</h3>
<p class="mt-1 text-xs text-cp-text-secondary">
{{ taskTime(task.createdAt) }} 创建
</p>
</div>
<BaseButton
v-if="!task.finishedAt"
size="sm"
:loading="stopping"
:disabled="task.stopRequested || task.counts.pending === 0"
@click="emit('stop')"
>
<template #icon>
<Square class="size-3 text-cp-error" />
</template>
{{ task.stopRequested || task.counts.pending === 0 ? '等待当前条目结束' : '停止未开始条目' }}
</BaseButton>
<BaseButton v-else size="sm" variant="soft" @click="emit('viewAccounts')">
查看账号 <ArrowUpRight class="size-3.5" />
</BaseButton>
</div>
<div>
<div class="mb-2 flex items-baseline justify-between gap-3 text-xs">
<span class="text-cp-text-secondary">条目进度 <strong class="font-mono font-medium text-cp-text">{{ processed(task) }} / {{ task.total }}</strong></span>
<span class="font-semibold text-cp-text">已入库 {{ task.counts.importedAccounts }} 个账号</span>
</div>
<div
class="flex h-2 overflow-hidden rounded-full bg-cp-fill-alter"
role="progressbar"
aria-label="条目处理进度"
:aria-valuenow="processed(task)"
:aria-valuemax="task.total"
:aria-valuemin="0"
>
<div v-for="status in outcomeOrder" :key="status" :class="itemStates[status].fill" :style="{ width: `${task.counts[status] / task.total * 100}%` }" />
</div>
<div class="mt-3 flex flex-wrap gap-x-4 gap-y-2 text-xs text-cp-text-secondary">
<span v-for="status in outcomeOrder" :key="status" class="inline-flex items-center gap-1.5">
<span class="size-1.5 rounded-full" :class="itemStates[status].fill" aria-hidden="true" />
{{ itemStates[status].label }} <span class="font-mono text-cp-text">{{ task.counts[status] }}</span>
</span>
</div>
</div>
<div>
<div v-if="hasAttention" class="mb-3 flex flex-wrap items-center gap-x-4 gap-y-3">
<p v-if="task.counts.unknown" class="flex min-w-0 items-start gap-2 text-xs leading-relaxed text-cp-warning-text">
<CircleAlert class="mt-0.5 size-3.5 shrink-0" />
待核对条目可能已入库,请先查看账号,再决定是否重新导入
</p>
<BaseSegmented
v-model="filter"
class="ml-auto shrink-0"
label="筛选导入明细"
:options="filterOptions"
size="sm"
/>
</div>
<div>
<BaseScrollbar :key="`${task.taskId}-${filter}`" max-height="min(18rem, 32dvh)">
<BaseEmpty v-if="!visibleItems.length" size="sm" title="暂无需要处理的条目" surface="none" />
<ol v-else class="m-0 list-none space-y-1 p-0" aria-label="导入条目结果">
<li
v-for="item in visibleItems"
:key="item.index"
class="grid grid-cols-[1.75rem_minmax(0,1fr)_4.5rem] items-center gap-x-3 gap-y-1 rounded-cp-sm px-3 py-2.5 odd:bg-cp-fill-quaternary sm:grid-cols-[2rem_2rem_minmax(0,1fr)_4.5rem]"
>
<span class="font-mono text-xs text-cp-text-secondary">{{ String(item.index).padStart(2, '0') }}</span>
<ProviderIconGroup :provider="item.provider" size="sm" />
<p class="col-span-2 col-start-2 row-start-2 min-w-0 text-xs leading-relaxed wrap-anywhere text-cp-text-secondary sm:col-span-1 sm:col-start-3 sm:row-start-1">
{{ itemDescription(item) }}
</p>
<span class="col-start-3 row-start-1 inline-flex items-center justify-self-end sm:col-start-4" :title="itemStates[item.status].label">
<template v-if="item.status === 'succeeded' || item.status === 'failed'">
<Check v-if="item.status === 'succeeded'" class="size-4 text-cp-success-text" aria-hidden="true" />
<X v-else class="size-4 text-cp-error-text" aria-hidden="true" />
<span class="sr-only">{{ itemStates[item.status].label }}</span>
</template>
<span v-else class="rounded-cp-sm px-2 py-0.5 text-[11px] font-semibold" :class="itemStates[item.status].color">
{{ itemStates[item.status].label }}
</span>
</span>
</li>
</ol>
</BaseScrollbar>
</div>
</div>
</section>
</template>
@@ -0,0 +1,98 @@
<script setup lang="ts">
import type { AccountImportTask, AccountImportTaskDetail } from '@/api'
import { CircleAlert, ListTodo } from '@lucide/vue'
import { computed, shallowRef, useId, watch } from 'vue'
import BaseButton from '@/components/base/BaseButton.vue'
import BaseEmpty from '@/components/base/BaseEmpty.vue'
import BaseIconButton from '@/components/base/BaseIconButton.vue'
import BaseModal from '@/components/base/BaseModal/index.vue'
import BasePopover from '@/components/base/BasePopover.vue'
import BaseSelect from '@/components/base/BaseSelect.vue'
import { taskLabel, taskTime } from './presenter'
import TaskDetail from './TaskDetail.vue'
const props = defineProps<{
tasks: AccountImportTask[]
selectedId: string
detail: AccountImportTaskDetail | null
loading: boolean
stopping: boolean
error: string
}>()
const emit = defineEmits<{ select: [id: string], refresh: [], stop: [], viewAccounts: [] }>()
const open = defineModel<boolean>({ required: true })
const retentionOpen = shallowRef(false)
const retentionId = useId()
const taskOptions = computed(() => props.tasks.map(task => ({
value: task.taskId,
label: taskTime(task.createdAt),
description: `${task.total} 个条目 · ${taskLabel(task)}`,
})))
watch(open, () => {
retentionOpen.value = false
})
function handleRetentionKeydown(event: KeyboardEvent) {
if (event.key === 'Escape' && retentionOpen.value) {
// 先关闭说明浮层,避免同一次按键也关闭任务弹窗。
event.preventDefault()
event.stopPropagation()
retentionOpen.value = false
}
}
</script>
<template>
<BaseModal v-model="open" title="导入任务" description="关闭页面后继续导入,再次打开即可查看进度" size="lg">
<div v-if="error" role="alert" class="mb-4 flex items-center justify-between gap-3 rounded-cp bg-cp-warning-container p-3 text-xs text-cp-warning-on-container">
<span>进度暂未更新:{{ error.replace(/[。.]\s*$/u, '') }},恢复连接后将自动刷新</span>
<BaseButton size="sm" :loading="loading" @click="emit('refresh')">
重试
</BaseButton>
</div>
<BaseEmpty v-if="!tasks.length" class="min-h-[min(20rem,50dvh)] content-center" :icon="ListTodo" :title="loading ? '正在读取导入任务' : '暂无导入任务'" description="新建一次账号导入,即可在这里查看逐条入库结果" />
<div v-else class="min-h-[min(20rem,50dvh)] min-w-0">
<div v-if="tasks.length > 1" class="mb-5">
<BaseSelect
:model-value="selectedId"
:options="taskOptions"
size="sm"
aria-label="切换导入任务"
class="w-full min-w-0 sm:max-w-96"
@update:model-value="emit('select', $event)"
/>
</div>
<TaskDetail v-if="detail" :task="detail" :stopping="stopping" @stop="emit('stop')" @view-accounts="emit('viewAccounts')" />
<BaseEmpty v-else class="min-h-[min(16rem,40dvh)] content-center" :title="loading ? '正在读取条目结果' : '请选择一个任务'" surface="none" />
</div>
<template #footer>
<div class="flex w-full flex-wrap items-center justify-between gap-3">
<BasePopover v-model="retentionOpen" trigger="hover-click" placement="top-start" :hover-delay="240">
<template #trigger>
<BaseIconButton
label="任务保留说明"
size="sm"
:aria-expanded="retentionOpen"
:aria-describedby="retentionOpen ? retentionId : undefined"
@keydown="handleRetentionKeydown"
>
<CircleAlert class="size-4" />
</BaseIconButton>
</template>
<div :id="retentionId" role="tooltip" class="grid w-72 max-w-[calc(100vw-2rem)] gap-2 p-4 text-cp-xs leading-relaxed text-cp-text-secondary">
<p class="m-0">
结果保留 1 小时
</p>
<p class="m-0">
服务重启后任务记录消失,已入库账号保留
</p>
</div>
</BasePopover>
<BaseButton @click="open = false">
关闭
</BaseButton>
</div>
</template>
</BaseModal>
</template>
@@ -0,0 +1,42 @@
import type { AccountImportTask, AccountImportTaskItem, ImportItemStatus } from '@/api'
import dayjs from 'dayjs'
export const itemStates: Record<ImportItemStatus, { label: string, color: string, fill: string }> = {
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 '请检查账号列表,确认是否已入库'
}
}
@@ -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<unknown> }) {
const open = shallowRef(false)
const tasks = shallowRef<AccountImportTask[]>([])
const selectedId = shallowRef('')
const detail = shallowRef<AccountImportTaskDetail | null>(null)
const error = shallowRef('')
const loading = shallowRef(false)
const stopAction = useAsyncAction()
const activeCount = computed(() => tasks.value.filter(task => !task.finishedAt).length)
let timer: ReturnType<typeof setTimeout> | 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 }
}
@@ -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<ReturnType<typeof getAccounts>>['items'][number]
export function useAccountMutations(options: {
accounts: Ref<AccountRow[]>
selectedIds: Ref<Set<string>>
onImportTaskCreated: (task: AccountImportTask) => void
reload: () => Promise<unknown>
replaceAccount: (account: AccountRow) => Promise<boolean>
}) {
@@ -30,6 +31,7 @@ export function useAccountMutations(options: {
const { downloadJson } = useDownload()
const onboarding = useAccountOnboarding({
reload: loadAccounts,
onImportTaskCreated: options.onImportTaskCreated,
})
const selectedAccountsById = new Map<string, AccountRow>()
const showDeleteModal = ref(false)
@@ -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<unknown>
onImportTaskCreated: (task: AccountImportTask) => void
}) {
const createModalOpen = shallowRef(false)
const reauthorizingAccount = shallowRef<AccountRow | null>(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 {
@@ -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
+34
View File
@@ -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()"
/>
<AccountImportTasks
v-model="showImportTasks"
:tasks="recentImportTasks"
:selected-id="importTaskId"
:detail="importTaskDetail"
:loading="loadingImportTasks"
:stopping="stoppingImportTask"
:error="importTaskError"
@select="importTasks.select"
@refresh="importTasks.refresh"
@stop="importTasks.stop"
@view-accounts="showImportTasks = false; loadAccounts()"
/>
<AccountCreateModal
v-model="showCreateModal"
v-model:form="createForm"