diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 2b339f3d96..4012e5d228 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -70,6 +70,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { concurrencyCache := repository.ProvideConcurrencyCache(redisClient, configConfig) schedulerCache := repository.ProvideSchedulerCache(redisClient, configConfig) accountRepository := repository.NewAccountRepository(client, db, schedulerCache) + adminAccountRepository := repository.NewAdminAccountRepository(client, db, schedulerCache) concurrencyService := service.ProvideConcurrencyService(concurrencyCache, accountRepository, configConfig) apiKeyService := service.ProvideAPIKeyService(apiKeyRepository, userRepository, groupRepository, userSubscriptionRepository, userGroupRateRepository, apiKeyCache, configConfig, billingCacheService, concurrencyService) apiKeyAuthCacheInvalidator := service.ProvideAPIKeyAuthCacheInvalidator(apiKeyService) @@ -175,7 +176,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { dashboardHandler := admin.NewDashboardHandler(dashboardService, dashboardAggregationService) proxyExitInfoProber := repository.NewProxyExitInfoProber(configConfig) proxyLatencyCache := repository.NewProxyLatencyCache(redisClient) - adminService := service.NewAdminService(userRepository, groupRepository, accountRepository, proxyRepository, apiKeyRepository, redeemCodeRepository, userGroupRateRepository, userRPMCache, billingCacheService, proxyExitInfoProber, proxyLatencyCache, apiKeyAuthCacheInvalidator, client, settingService, subscriptionService, userSubscriptionRepository, privacyClientFactory, openAIGatewayService) + adminService := service.NewAdminService(userRepository, groupRepository, adminAccountRepository, proxyRepository, apiKeyRepository, redeemCodeRepository, userGroupRateRepository, userRPMCache, billingCacheService, proxyExitInfoProber, proxyLatencyCache, apiKeyAuthCacheInvalidator, client, settingService, subscriptionService, userSubscriptionRepository, privacyClientFactory, openAIGatewayService) adminUserHandler := admin.NewUserHandler(adminService, concurrencyService, serviceUserPlatformQuotaRepository, billingCache) groupCapacityService := service.NewGroupCapacityService(accountRepository, groupRepository, concurrencyService, sessionLimitCache, rpmCache) groupHandler := admin.NewGroupHandler(adminService, dashboardService, groupCapacityService) diff --git a/backend/internal/handler/admin/account_handler.go b/backend/internal/handler/admin/account_handler.go index e4ed5b46b0..4e89b18bd2 100644 --- a/backend/internal/handler/admin/account_handler.go +++ b/backend/internal/handler/admin/account_handler.go @@ -860,6 +860,53 @@ func (h *AccountHandler) Create(c *gin.Context) { response.Success(c, result.Data) } +// Duplicate handles creating an independent account from an existing account's configuration. +// POST /api/v1/admin/accounts/:id/duplicate +func (h *AccountHandler) Duplicate(c *gin.Context) { + accountID, err := strconv.ParseInt(c.Param("id"), 10, 64) + if err != nil { + response.BadRequest(c, "Invalid account ID") + return + } + actorScope := adminActorScope(c) + + result, err := executeAdminIdempotent( + c, + "admin.accounts.duplicate", + struct { + AccountID int64 `json:"account_id"` + }{AccountID: accountID}, + service.DefaultWriteIdempotencyTTL(), + func(ctx context.Context) (any, error) { + account, execErr := h.adminService.DuplicateAccount(ctx, accountID, actorScope, c.GetHeader("Idempotency-Key")) + if execErr != nil { + return nil, execErr + } + return h.buildAccountResponseWithRuntime(ctx, account), nil + }, + ) + if err != nil { + reason := infraerrors.Reason(err) + if reason == infraerrors.Reason(service.ErrIdempotencyInProgress) || reason == infraerrors.Reason(service.ErrIdempotencyStoreUnavail) { + recovered, recoverErr := h.adminService.RecoverDuplicateAccount(c.Request.Context(), accountID, actorScope, c.GetHeader("Idempotency-Key")) + if recoverErr != nil { + slog.Warn("account_duplicate_recovery_failed", "account_id", accountID, "actor_scope", actorScope, "reason", reason, "error", recoverErr) + } else if recovered != nil { + c.Header("X-Idempotency-Recovered", "true") + response.Success(c, h.buildAccountResponseWithRuntime(c.Request.Context(), recovered)) + return + } + } + response.ErrorFrom(c, err) + return + } + + if result != nil && result.Replayed { + c.Header("X-Idempotency-Replayed", "true") + } + response.Success(c, result.Data) +} + // Update handles updating an account // PUT /api/v1/admin/accounts/:id func (h *AccountHandler) Update(c *gin.Context) { diff --git a/backend/internal/handler/admin/account_handler_duplicate_test.go b/backend/internal/handler/admin/account_handler_duplicate_test.go new file mode 100644 index 0000000000..c99e882b1c --- /dev/null +++ b/backend/internal/handler/admin/account_handler_duplicate_test.go @@ -0,0 +1,295 @@ +//go:build unit + +package admin + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +type duplicateAccountAdminServiceStub struct { + service.AdminService + account *service.Account + calls int + recoverCalls int + accountID int64 + actorScope string + operationKey string + recoverScope string + recoverKey string + recoverErr error + created bool +} + +type blockingDuplicateAdminServiceStub struct { + service.AdminService + account *service.Account + started chan struct{} + release chan struct{} + calls atomic.Int32 + recoverCalls atomic.Int32 + recoverErr error +} + +type failOnceMarkSucceededRepo struct { + *memoryIdempotencyRepoStub + failNext bool +} + +func (r *failOnceMarkSucceededRepo) MarkSucceeded(ctx context.Context, id int64, responseStatus int, responseBody string, expiresAt time.Time) error { + if r.failNext { + r.failNext = false + return errors.New("mark succeeded failed") + } + return r.memoryIdempotencyRepoStub.MarkSucceeded(ctx, id, responseStatus, responseBody, expiresAt) +} + +func (s *duplicateAccountAdminServiceStub) DuplicateAccount(_ context.Context, accountID int64, actorScope, operationKey string) (*service.Account, error) { + s.calls++ + s.accountID = accountID + s.actorScope = actorScope + s.operationKey = operationKey + s.created = true + return s.account, nil +} + +func (s *duplicateAccountAdminServiceStub) RecoverDuplicateAccount(_ context.Context, _ int64, actorScope, operationKey string) (*service.Account, error) { + s.recoverCalls++ + s.recoverScope = actorScope + s.recoverKey = operationKey + if s.recoverErr != nil { + return nil, s.recoverErr + } + if !s.created { + return nil, nil + } + return s.account, nil +} + +func (s *blockingDuplicateAdminServiceStub) DuplicateAccount(_ context.Context, _ int64, _, _ string) (*service.Account, error) { + s.calls.Add(1) + close(s.started) + <-s.release + return s.account, nil +} + +func (s *blockingDuplicateAdminServiceStub) RecoverDuplicateAccount(_ context.Context, _ int64, _, _ string) (*service.Account, error) { + s.recoverCalls.Add(1) + return nil, s.recoverErr +} + +func setupDuplicateAccountRouter(t *testing.T, svc service.AdminService) *gin.Engine { + t.Helper() + previousCoordinator := service.DefaultIdempotencyCoordinator() + service.SetDefaultIdempotencyCoordinator(nil) + t.Cleanup(func() { service.SetDefaultIdempotencyCoordinator(previousCoordinator) }) + + gin.SetMode(gin.TestMode) + router := gin.New() + router.Use(func(c *gin.Context) { + c.Set(string(middleware2.ContextKeyUser), middleware2.AuthSubject{UserID: 77}) + c.Next() + }) + handler := NewAccountHandler(svc, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) + router.POST("/api/v1/admin/accounts/:id/duplicate", handler.Duplicate) + return router +} + +func TestDuplicateAccountHandlerRedactsCredentials(t *testing.T) { + svc := &duplicateAccountAdminServiceStub{ + account: &service.Account{ + ID: 43, + Name: "primary (Copy)", + Platform: service.PlatformAnthropic, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: false, + Credentials: map[string]any{"api_key": "top-secret-key"}, + }, + } + router := setupDuplicateAccountRouter(t, svc) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil) + + router.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code) + require.Equal(t, 1, svc.calls) + require.Contains(t, recorder.Body.String(), `"name":"primary (Copy)"`) + require.NotContains(t, recorder.Body.String(), "top-secret-key") + var responseBody struct { + Data struct { + Credentials map[string]any `json:"credentials"` + Schedulable bool `json:"schedulable"` + } `json:"data"` + } + require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &responseBody)) + require.Empty(t, responseBody.Data.Credentials) + require.False(t, responseBody.Data.Schedulable) +} + +func TestDuplicateAccountHandlerRejectsInvalidID(t *testing.T) { + svc := &duplicateAccountAdminServiceStub{} + router := setupDuplicateAccountRouter(t, svc) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/not-a-number/duplicate", nil) + + router.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusBadRequest, recorder.Code) + require.Zero(t, svc.calls) +} + +func TestDuplicateAccountHandlerReplaysSameIdempotencyKey(t *testing.T) { + svc := &duplicateAccountAdminServiceStub{ + account: &service.Account{ + ID: 43, + Name: "primary (Copy)", + Platform: service.PlatformAnthropic, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: false, + }, + } + router := setupDuplicateAccountRouter(t, svc) + repo := newMemoryIdempotencyRepoStub() + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(repo, service.DefaultIdempotencyConfig())) + + call := func() *httptest.ResponseRecorder { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil) + request.Header.Set("Idempotency-Key", "duplicate-account-42") + router.ServeHTTP(recorder, request) + return recorder + } + + first := call() + second := call() + + require.Equal(t, http.StatusOK, first.Code) + require.Equal(t, http.StatusOK, second.Code) + require.Equal(t, 1, svc.calls) + require.Equal(t, int64(42), svc.accountID) + require.Equal(t, "admin:77", svc.actorScope) + require.Equal(t, "duplicate-account-42", svc.operationKey) + require.Equal(t, "true", second.Header().Get("X-Idempotency-Replayed")) +} + +func TestDuplicateAccountHandlerRecoversAfterMarkSucceededFailure(t *testing.T) { + svc := &duplicateAccountAdminServiceStub{ + account: &service.Account{ + ID: 43, + Name: "primary (Copy)", + Platform: service.PlatformAnthropic, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: false, + }, + } + router := setupDuplicateAccountRouter(t, svc) + repo := &failOnceMarkSucceededRepo{memoryIdempotencyRepoStub: newMemoryIdempotencyRepoStub(), failNext: true} + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(repo, service.DefaultIdempotencyConfig())) + + call := func() *httptest.ResponseRecorder { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil) + request.Header.Set("Idempotency-Key", "duplicate-account-42-recovery") + router.ServeHTTP(recorder, request) + return recorder + } + + first := call() + second := call() + + require.Equal(t, http.StatusOK, first.Code) + require.Equal(t, http.StatusOK, second.Code) + require.Equal(t, "true", first.Header().Get("X-Idempotency-Recovered")) + require.Equal(t, "true", second.Header().Get("X-Idempotency-Recovered")) + require.Equal(t, 1, svc.calls, "ambiguous retries must not repeat the create side effect") + require.Equal(t, 2, svc.recoverCalls) + require.Equal(t, "admin:77", svc.recoverScope) + require.Equal(t, "duplicate-account-42-recovery", svc.recoverKey) + require.Contains(t, second.Body.String(), `"id":43`) +} + +func TestDuplicateAccountHandlerPreservesIdempotencyErrorWhenRecoveryLookupFails(t *testing.T) { + svc := &duplicateAccountAdminServiceStub{ + account: &service.Account{ + ID: 43, + Name: "primary (Copy)", + Platform: service.PlatformAnthropic, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: false, + }, + recoverErr: errors.New("recovery database unavailable"), + } + router := setupDuplicateAccountRouter(t, svc) + repo := &failOnceMarkSucceededRepo{memoryIdempotencyRepoStub: newMemoryIdempotencyRepoStub(), failNext: true} + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(repo, service.DefaultIdempotencyConfig())) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil) + request.Header.Set("Idempotency-Key", "duplicate-account-42-recovery-error") + + router.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusServiceUnavailable, recorder.Code) + require.Contains(t, recorder.Body.String(), "IDEMPOTENCY_STORE_UNAVAILABLE") + require.Equal(t, 1, svc.calls) + require.Equal(t, 1, svc.recoverCalls) + require.Equal(t, "admin:77", svc.recoverScope) +} + +func TestDuplicateAccountHandlerDoesNotReexecuteWhileOriginalIsProcessing(t *testing.T) { + svc := &blockingDuplicateAdminServiceStub{ + account: &service.Account{ + ID: 43, + Name: "primary (Copy)", + Platform: service.PlatformAnthropic, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: false, + }, + started: make(chan struct{}), + release: make(chan struct{}), + recoverErr: errors.New("recovery database unavailable"), + } + router := setupDuplicateAccountRouter(t, svc) + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(newMemoryIdempotencyRepoStub(), service.DefaultIdempotencyConfig())) + + call := func() *httptest.ResponseRecorder { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/admin/accounts/42/duplicate", nil) + request.Header.Set("Idempotency-Key", "duplicate-account-42-active") + router.ServeHTTP(recorder, request) + return recorder + } + firstDone := make(chan *httptest.ResponseRecorder, 1) + go func() { firstDone <- call() }() + <-svc.started + + second := call() + require.Equal(t, http.StatusConflict, second.Code) + require.Contains(t, second.Body.String(), "IDEMPOTENCY_IN_PROGRESS") + require.Equal(t, int32(1), svc.calls.Load()) + require.Equal(t, int32(1), svc.recoverCalls.Load()) + + close(svc.release) + select { + case first := <-firstDone: + require.Equal(t, http.StatusOK, first.Code) + case <-time.After(time.Second): + t.Fatal("original duplicate request did not finish") + } +} diff --git a/backend/internal/handler/admin/admin_service_stub_test.go b/backend/internal/handler/admin/admin_service_stub_test.go index 5e9c4d517e..d93ee7111c 100644 --- a/backend/internal/handler/admin/admin_service_stub_test.go +++ b/backend/internal/handler/admin/admin_service_stub_test.go @@ -418,6 +418,15 @@ func (s *stubAdminService) CreateAccount(ctx context.Context, input *service.Cre return &account, nil } +func (s *stubAdminService) DuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*service.Account, error) { + account := service.Account{ID: 301, Name: "account (Copy)", Status: service.StatusActive, Schedulable: false} + return &account, nil +} + +func (s *stubAdminService) RecoverDuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*service.Account, error) { + return nil, nil +} + func (s *stubAdminService) UpdateAccount(ctx context.Context, id int64, input *service.UpdateAccountInput) (*service.Account, error) { s.updateAccountCalls++ if s.updateAccountErr != nil { diff --git a/backend/internal/handler/admin/idempotency_helper.go b/backend/internal/handler/admin/idempotency_helper.go index aa8eeaaf79..1894faeaae 100644 --- a/backend/internal/handler/admin/idempotency_helper.go +++ b/backend/internal/handler/admin/idempotency_helper.go @@ -37,14 +37,9 @@ func executeAdminIdempotent( return &service.IdempotencyExecuteResult{Data: data}, nil } - actorScope := "admin:0" - if subject, ok := middleware2.GetAuthSubjectFromContext(c); ok { - actorScope = "admin:" + strconv.FormatInt(subject.UserID, 10) - } - return coordinator.Execute(c.Request.Context(), service.IdempotencyExecuteOptions{ Scope: scope, - ActorScope: actorScope, + ActorScope: adminActorScope(c), Method: c.Request.Method, Route: c.FullPath(), IdempotencyKey: c.GetHeader("Idempotency-Key"), @@ -54,6 +49,14 @@ func executeAdminIdempotent( }, execute) } +func adminActorScope(c *gin.Context) string { + actorScope := "admin:0" + if subject, ok := middleware2.GetAuthSubjectFromContext(c); ok { + actorScope = "admin:" + strconv.FormatInt(subject.UserID, 10) + } + return actorScope +} + func executeAdminIdempotentJSON( c *gin.Context, scope string, diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index e85bf46fc1..9af99223f7 100644 --- a/backend/internal/repository/account_repo.go +++ b/backend/internal/repository/account_repo.go @@ -73,6 +73,12 @@ func NewAccountRepository(client *dbent.Client, sqlDB *sql.DB, schedulerCache se return newAccountRepositoryWithSQL(client, sqlDB, schedulerCache) } +// NewAdminAccountRepository exposes the account repository's atomic duplication capability +// as an explicit dependency of the admin service. +func NewAdminAccountRepository(client *dbent.Client, sqlDB *sql.DB, schedulerCache service.SchedulerCache) service.AdminAccountRepository { + return newAccountRepositoryWithSQL(client, sqlDB, schedulerCache) +} + // newAccountRepositoryWithSQL 是内部构造函数，支持依赖注入 SQL 执行器。 // 这种设计便于单元测试时注入 mock 对象。 func newAccountRepositoryWithSQL(client *dbent.Client, sqlq sqlExecutor, schedulerCache service.SchedulerCache) *accountRepository { @@ -80,11 +86,21 @@ func newAccountRepositoryWithSQL(client *dbent.Client, sqlq sqlExecutor, schedul } func (r *accountRepository) Create(ctx context.Context, account *service.Account) error { + if err := createAccountRecord(ctx, r.client, account); err != nil { + return err + } + if err := enqueueSchedulerOutbox(ctx, r.sql, service.SchedulerOutboxEventAccountChanged, &account.ID, nil, buildSchedulerGroupPayload(account.GroupIDs)); err != nil { + logger.LegacyPrintf("repository.account", "[SchedulerOutbox] enqueue account create failed: account=%d err=%v", account.ID, err) + } + return nil +} + +func createAccountRecord(ctx context.Context, client *dbent.Client, account *service.Account) error { if account == nil { return service.ErrAccountNilInput } - builder := r.client.Account.Create(). + builder := client.Account.Create(). SetName(account.Name). SetNillableNotes(account.Notes). SetPlatform(account.Platform). @@ -146,8 +162,58 @@ func (r *accountRepository) Create(ctx context.Context, account *service.Account account.ID = created.ID account.CreatedAt = created.CreatedAt account.UpdatedAt = created.UpdatedAt - if err := enqueueSchedulerOutbox(ctx, r.sql, service.SchedulerOutboxEventAccountChanged, &account.ID, nil, buildSchedulerGroupPayload(account.GroupIDs)); err != nil { - logger.LegacyPrintf("repository.account", "[SchedulerOutbox] enqueue account create failed: account=%d err=%v", account.ID, err) + return nil +} + +// CreateWithAccountGroups atomically persists an account, its exact per-group priorities, +// and the scheduler outbox event used to publish the new routing snapshot. +func (r *accountRepository) CreateWithAccountGroups(ctx context.Context, account *service.Account, groups []service.AccountGroup) error { + if account == nil { + return service.ErrAccountNilInput + } + tx, err := r.client.Tx(ctx) + if err != nil && !errors.Is(err, dbent.ErrTxStarted) { + return err + } + + var txClient *dbent.Client + if err == nil { + defer func() { _ = tx.Rollback() }() + txClient = tx.Client() + } else { + // Reuse a caller-owned transaction when this repository is already transactional. + txClient = r.client + } + + if err := createAccountRecord(ctx, txClient, account); err != nil { + return err + } + groupIDs := make([]int64, 0, len(groups)) + if len(groups) > 0 { + builders := make([]*dbent.AccountGroupCreate, 0, len(groups)) + for i := range groups { + groups[i].AccountID = account.ID + groupIDs = append(groupIDs, groups[i].GroupID) + builders = append(builders, txClient.AccountGroup.Create(). + SetAccountID(account.ID). + SetGroupID(groups[i].GroupID). + SetPriority(groups[i].Priority), + ) + } + if _, err := txClient.AccountGroup.CreateBulk(builders...).Save(ctx); err != nil { + return err + } + } + account.GroupIDs = groupIDs + account.AccountGroups = append([]service.AccountGroup(nil), groups...) + if err := enqueueSchedulerOutbox(ctx, txClient, service.SchedulerOutboxEventAccountChanged, &account.ID, nil, buildSchedulerGroupPayload(groupIDs)); err != nil { + return err + } + + if tx != nil { + if err := tx.Commit(); err != nil { + return err + } } return nil } diff --git a/backend/internal/repository/account_repo_duplicate_integration_test.go b/backend/internal/repository/account_repo_duplicate_integration_test.go new file mode 100644 index 0000000000..f36ee1d45e --- /dev/null +++ b/backend/internal/repository/account_repo_duplicate_integration_test.go @@ -0,0 +1,73 @@ +//go:build integration + +package repository + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +func TestCreateWithAccountGroupsPersistsPausedCopyAtomically(t *testing.T) { + ctx := context.Background() + client := testEntClient(t) + repo := newAccountRepositoryWithSQL(client, integrationDB, nil) + suffix := time.Now().UnixNano() + + group, err := client.Group.Create(). + SetName(fmt.Sprintf("duplicate-atomic-%d", suffix)). + SetPlatform(service.PlatformAnthropic). + Save(ctx) + require.NoError(t, err) + + success := &service.Account{ + Name: fmt.Sprintf("duplicate-success-%d", suffix), + Platform: service.PlatformAnthropic, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: false, + Credentials: map[string]any{"api_key": "secret"}, + Extra: map[string]any{}, + } + require.NoError(t, repo.CreateWithAccountGroups(ctx, success, []service.AccountGroup{{GroupID: group.ID, Priority: 37}})) + t.Cleanup(func() { + _, _ = integrationDB.ExecContext(context.Background(), "DELETE FROM scheduler_outbox WHERE account_id = $1", success.ID) + _, _ = integrationDB.ExecContext(context.Background(), "DELETE FROM account_groups WHERE account_id = $1", success.ID) + _, _ = integrationDB.ExecContext(context.Background(), "DELETE FROM accounts WHERE id = $1", success.ID) + _, _ = integrationDB.ExecContext(context.Background(), "DELETE FROM groups WHERE id = $1", group.ID) + }) + + var schedulable bool + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT schedulable FROM accounts WHERE id = $1", success.ID).Scan(&schedulable)) + require.False(t, schedulable) + var priority int + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT priority FROM account_groups WHERE account_id = $1 AND group_id = $2", success.ID, group.ID).Scan(&priority)) + require.Equal(t, 37, priority) + var outboxCount int + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE account_id = $1", success.ID).Scan(&outboxCount)) + require.Equal(t, 1, outboxCount) + + failure := &service.Account{ + Name: fmt.Sprintf("duplicate-failure-%d", suffix), + Platform: service.PlatformAnthropic, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: false, + Credentials: map[string]any{"api_key": "secret"}, + Extra: map[string]any{}, + } + err = repo.CreateWithAccountGroups(ctx, failure, []service.AccountGroup{{GroupID: int64(^uint64(0) >> 1), Priority: 1}}) + require.Error(t, err) + + var accountCount, groupCount, failedOutboxCount int + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM accounts WHERE name = $1", failure.Name).Scan(&accountCount)) + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM account_groups WHERE account_id = $1", failure.ID).Scan(&groupCount)) + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE account_id = $1", failure.ID).Scan(&failedOutboxCount)) + require.Zero(t, accountCount) + require.Zero(t, groupCount) + require.Zero(t, failedOutboxCount) +} diff --git a/backend/internal/repository/wire.go b/backend/internal/repository/wire.go index ec5078eac8..0ee91d146b 100644 --- a/backend/internal/repository/wire.go +++ b/backend/internal/repository/wire.go @@ -68,6 +68,7 @@ var ProviderSet = wire.NewSet( NewAPIKeyRepository, NewGroupRepository, NewAccountRepository, + NewAdminAccountRepository, NewScheduledTestPlanRepository, // 定时测试计划仓储 NewScheduledTestResultRepository, // 定时测试结果仓储 NewProxyRepository, diff --git a/backend/internal/server/api_contract_test.go b/backend/internal/server/api_contract_test.go index a5e3fde155..d42c5c934f 100644 --- a/backend/internal/server/api_contract_test.go +++ b/backend/internal/server/api_contract_test.go @@ -1721,6 +1721,10 @@ func (s *stubAccountRepo) Create(ctx context.Context, account *service.Account) return errors.New("not implemented") } +func (s *stubAccountRepo) CreateWithAccountGroups(ctx context.Context, account *service.Account, groups []service.AccountGroup) error { + return errors.New("not implemented") +} + func (s *stubAccountRepo) GetByID(ctx context.Context, id int64) (*service.Account, error) { return nil, service.ErrAccountNotFound } diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go index 5132022d4a..6bf67bcbb1 100644 --- a/backend/internal/server/routes/admin.go +++ b/backend/internal/server/routes/admin.go @@ -297,6 +297,7 @@ func registerAccountRoutes(admin *gin.RouterGroup, h *handler.Handlers) { accounts.GET("", h.Admin.Account.List) accounts.GET("/:id", h.Admin.Account.GetByID) accounts.POST("", h.Admin.Account.Create) + accounts.POST("/:id/duplicate", h.Admin.Account.Duplicate) accounts.POST("/check-mixed-channel", h.Admin.Account.CheckMixedChannel) accounts.POST("/import/codex-session", h.Admin.Account.ImportCodexSession) accounts.POST("/sync/crs", h.Admin.Account.SyncFromCRS) diff --git a/backend/internal/service/account_service.go b/backend/internal/service/account_service.go index 5956684f98..109ef95680 100644 --- a/backend/internal/service/account_service.go +++ b/backend/internal/service/account_service.go @@ -90,6 +90,19 @@ type AccountRepository interface { ListShadowsByParent(ctx context.Context, parentID int64) ([]*Account, error) } +type AccountDuplicateRepository interface { + // CreateWithAccountGroups atomically persists an account, its exact group priorities, + // and the scheduler outbox event for the new routing snapshot. + CreateWithAccountGroups(ctx context.Context, account *Account, groups []AccountGroup) error +} + +// AdminAccountRepository makes the account-duplication write capability an explicit +// construction dependency without forcing read-only gateway test doubles to implement it. +type AdminAccountRepository interface { + AccountRepository + AccountDuplicateRepository +} + // AccountBulkUpdate describes the fields that can be updated in a bulk operation. // Nil pointers mean "do not change". type AccountBulkUpdate struct { diff --git a/backend/internal/service/admin_account.go b/backend/internal/service/admin_account.go index 8cb6d8e63b..f4c3375650 100644 --- a/backend/internal/service/admin_account.go +++ b/backend/internal/service/admin_account.go @@ -2,6 +2,8 @@ package service import ( "context" + "crypto/sha256" + "encoding/json" "errors" "fmt" "log/slog" @@ -60,6 +62,275 @@ func (s *adminServiceImpl) GetAccountsByIDs(ctx context.Context, ids []int64) ([ return accounts, nil } +const maxAccountNameRunes = 100 +const duplicateAccountOperationIDExtraKey = "duplicate_operation_id" + +func duplicateAccountName(sourceName string) string { + const suffix = " (Copy)" + nameRunes := []rune(strings.TrimSpace(sourceName)) + maxBaseRunes := maxAccountNameRunes - len([]rune(suffix)) + if len(nameRunes) > maxBaseRunes { + nameRunes = nameRunes[:maxBaseRunes] + } + return string(nameRunes) + suffix +} + +func cloneAccountJSONMap(value map[string]any) (map[string]any, error) { + if value == nil { + return nil, nil + } + payload, err := json.Marshal(value) + if err != nil { + return nil, err + } + cloned := make(map[string]any, len(value)) + if err := json.Unmarshal(payload, &cloned); err != nil { + return nil, err + } + return cloned, nil +} + +var duplicateAccountDiscardedExtraKeys = map[string]struct{}{ + // A retry identity belongs to the operation that created one copy, not to later copies. + duplicateAccountOperationIDExtraKey: {}, + // External sync identity belongs to one local account only. + "crs_account_id": {}, + "crs_kind": {}, + "crs_synced_at": {}, + // Local quota usage and derived window timestamps must start fresh. + "quota_used": {}, + "quota_daily_used": {}, + "quota_weekly_used": {}, + "quota_daily_start": {}, + "quota_weekly_start": {}, + "quota_daily_reset_at": {}, + "quota_weekly_reset_at": {}, + // Provider observations, capability probes, and transient scheduling state. + "model_rate_limits": {}, + "session_window_utilization": {}, + "passive_usage_7d_utilization": {}, + "passive_usage_7d_reset": {}, + "passive_usage_7d_oi_utilization": {}, + "passive_usage_7d_oi_reset": {}, + "passive_usage_sampled_at": {}, + "grok_usage_snapshot": {}, + "grok_billing_snapshot": {}, + "openai_responses_supported": {}, + "openai_compact_supported": {}, + "openai_compact_checked_at": {}, + "openai_compact_last_status": {}, + "openai_compact_last_error": {}, + "antigravity_credits_overages": {}, + "antigravity_force_token_refresh": {}, + "antigravity_force_token_refresh_at": {}, + "antigravity_force_token_refresh_reason": {}, + "drive_storage_limit": {}, + "drive_storage_usage": {}, + "drive_tier_updated_at": {}, + "codex_primary_used_percent": {}, + "codex_primary_reset_after_seconds": {}, + "codex_primary_window_minutes": {}, + "codex_secondary_used_percent": {}, + "codex_secondary_reset_after_seconds": {}, + "codex_secondary_window_minutes": {}, + "codex_primary_over_secondary_percent": {}, + "codex_usage_updated_at": {}, + "codex_5h_used_percent": {}, + "codex_5h_reset_after_seconds": {}, + "codex_5h_window_minutes": {}, + "codex_5h_reset_at": {}, + "codex_7d_used_percent": {}, + "codex_7d_reset_after_seconds": {}, + "codex_7d_window_minutes": {}, + "codex_7d_reset_at": {}, +} + +func duplicateAccountExtra(value map[string]any) (map[string]any, error) { + cloned, err := cloneAccountJSONMap(value) + if err != nil { + return nil, err + } + for key := range duplicateAccountDiscardedExtraKeys { + delete(cloned, key) + } + return cloned, nil +} + +func canDuplicateAccountType(accountType string) bool { + switch accountType { + case AccountTypeAPIKey, AccountTypeUpstream, AccountTypeBedrock, AccountTypeServiceAccount: + return true + default: + return false + } +} + +func duplicateAccountGroups(source *Account) ([]AccountGroup, []int64) { + if len(source.AccountGroups) > 0 { + groups := make([]AccountGroup, 0, len(source.AccountGroups)) + groupIDs := make([]int64, 0, len(source.AccountGroups)) + for _, sourceGroup := range source.AccountGroups { + groups = append(groups, AccountGroup{GroupID: sourceGroup.GroupID, Priority: sourceGroup.Priority}) + groupIDs = append(groupIDs, sourceGroup.GroupID) + } + return groups, groupIDs + } + + groups := make([]AccountGroup, 0, len(source.GroupIDs)) + groupIDs := append([]int64(nil), source.GroupIDs...) + for i, groupID := range groupIDs { + groups = append(groups, AccountGroup{GroupID: groupID, Priority: i + 1}) + } + return groups, groupIDs +} + +func duplicateAccountOperationID(sourceID int64, actorScope, operationKey string) string { + operationKey = strings.TrimSpace(operationKey) + if operationKey == "" { + return "" + } + actorScope = strings.TrimSpace(actorScope) + if actorScope == "" { + actorScope = "admin:0" + } + payload := "admin.accounts.duplicate\x00" + actorScope + "\x00" + strconv.FormatInt(sourceID, 10) + "\x00" + operationKey + digest := sha256.Sum256([]byte(payload)) + return fmt.Sprintf("%x", digest) +} + +func (s *adminServiceImpl) findDuplicateByOperationID(ctx context.Context, operationID string) (*Account, error) { + if operationID == "" { + return nil, nil + } + accounts, err := s.accountRepo.FindByExtraField(ctx, duplicateAccountOperationIDExtraKey, operationID) + if err != nil { + return nil, fmt.Errorf("find duplicate account operation: %w", err) + } + if len(accounts) == 0 { + return nil, nil + } + account := accounts[0] + return &account, nil +} + +// RecoverDuplicateAccount performs a read-only lookup for an already committed duplicate. +// It is used when the idempotency coordinator cannot confirm whether response persistence +// succeeded, and deliberately never repeats the create side effect. +func (s *adminServiceImpl) RecoverDuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error) { + return s.findDuplicateByOperationID(ctx, duplicateAccountOperationID(id, actorScope, operationKey)) +} + +func cloneAccountValuePointer[T any](value *T) *T { + if value == nil { + return nil + } + cloned := *value + return &cloned +} + +// DuplicateAccount creates a paused account from source configuration without carrying first-class +// runtime state. Credentials and extra configuration are deep-copied so normalization of the new +// account cannot mutate the in-memory source. Linked credential shadows are excluded because they +// intentionally do not own credentials and must be created through CreateShadow. +func (s *adminServiceImpl) DuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error) { + operationID := duplicateAccountOperationID(id, actorScope, operationKey) + existing, err := s.RecoverDuplicateAccount(ctx, id, actorScope, operationKey) + if err != nil { + return nil, err + } + if existing != nil { + return existing, nil + } + + source, err := s.accountRepo.GetByID(ctx, id) + if err != nil { + return nil, err + } + if source.IsCredentialShadow() { + return nil, infraerrors.BadRequest( + "ACCOUNT_DUPLICATE_SHADOW_UNSUPPORTED", + "linked credential shadow accounts cannot be duplicated; duplicate the parent account instead", + ) + } + if !canDuplicateAccountType(source.Type) { + return nil, infraerrors.BadRequest( + "ACCOUNT_DUPLICATE_CREDENTIAL_TYPE_UNSUPPORTED", + "accounts with rotating or unsupported credential types cannot be duplicated", + ) + } + + credentials, err := cloneAccountJSONMap(source.Credentials) + if err != nil { + return nil, fmt.Errorf("clone account credentials: %w", err) + } + extra, err := duplicateAccountExtra(source.Extra) + if err != nil { + return nil, fmt.Errorf("clone account extra configuration: %w", err) + } + if operationID != "" { + if extra == nil { + extra = make(map[string]any, 1) + } + extra[duplicateAccountOperationIDExtraKey] = operationID + } + + var expiresAt *int64 + if source.ExpiresAt != nil { + unix := source.ExpiresAt.Unix() + expiresAt = &unix + } + autoPauseOnExpired := source.AutoPauseOnExpired + groups, groupIDs := duplicateAccountGroups(source) + proxyID := source.ProxyID + if source.ProxyFallbackOriginID != nil { + // Proxy fallback is transient runtime state; duplicate the configured origin. + proxyID = source.ProxyFallbackOriginID + } + input := &CreateAccountInput{ + Name: duplicateAccountName(source.Name), + Notes: cloneAccountValuePointer(source.Notes), + Platform: source.Platform, + Type: source.Type, + Credentials: credentials, + Extra: extra, + ProxyID: cloneAccountValuePointer(proxyID), + Concurrency: source.Concurrency, + Priority: source.Priority, + RateMultiplier: cloneAccountValuePointer(source.RateMultiplier), + LoadFactor: cloneAccountValuePointer(source.LoadFactor), + GroupIDs: groupIDs, + ExpiresAt: expiresAt, + AutoPauseOnExpired: &autoPauseOnExpired, + SkipDefaultGroupBind: true, + SkipMixedChannelCheck: true, + } + accountExtra, err := normalizeOpenAILongContextBillingExtra(input.Platform, input.Extra) + if err != nil { + return nil, fmt.Errorf("normalize duplicate account extra: %w", err) + } + if err := NormalizeHeaderOverrideCredentials(input.Credentials); err != nil { + return nil, err + } + duplicate, err := buildAccountForCreate(input, accountExtra) + if err != nil { + return nil, err + } + // A copied credential must be reviewed before it can share live traffic with its source. + duplicate.Schedulable = false + if s.accountDuplicateRepo == nil { + return nil, errors.New("account duplicate repository is not configured") + } + if err := s.accountDuplicateRepo.CreateWithAccountGroups(ctx, duplicate, groups); err != nil { + return nil, fmt.Errorf("create duplicate account: %w", err) + } + for i := range groups { + groups[i].AccountID = duplicate.ID + } + duplicate.AccountGroups = groups + duplicate.GroupIDs = groupIDs + return duplicate, nil +} + func normalizeAccountConcurrency(platform, accountType string, concurrency int) int { if platform == PlatformGrok && accountType == AccountTypeOAuth { if concurrency <= 0 { @@ -122,40 +393,7 @@ func normalizeOpenAILongContextBillingUpdateExtra(account *Account, input *Updat return normalized, nil } -func (s *adminServiceImpl) CreateAccount(ctx context.Context, input *CreateAccountInput) (*Account, error) { - accountExtra, err := normalizeOpenAILongContextBillingExtra(input.Platform, input.Extra) - if err != nil { - return nil, err - } - - // 绑定分组 - groupIDs := input.GroupIDs - // 如果没有指定分组,自动绑定对应平台的默认分组 - if len(groupIDs) == 0 && !input.SkipDefaultGroupBind { - defaultGroupName := input.Platform + "-default" - groups, err := s.groupRepo.ListActiveByPlatform(ctx, input.Platform) - if err == nil { - for _, g := range groups { - if g.Name == defaultGroupName { - groupIDs = []int64{g.ID} - break - } - } - } - } - - // 检查混合渠道风险（除非用户已确认） - if len(groupIDs) > 0 && !input.SkipMixedChannelCheck { - if err := s.checkMixedChannelRisk(ctx, 0, input.Platform, groupIDs); err != nil { - return nil, err - } - } - - // 校验并规范化请求头覆写配置（header 名小写化、格式检查） - if err := NormalizeHeaderOverrideCredentials(input.Credentials); err != nil { - return nil, err - } - +func buildAccountForCreate(input *CreateAccountInput, accountExtra map[string]any) (*Account, error) { account := &Account{ Name: input.Name, Notes: normalizeAccountNotes(input.Notes), @@ -198,6 +436,47 @@ func (s *adminServiceImpl) CreateAccount(ctx context.Context, input *CreateAccou } account.LoadFactor = input.LoadFactor } + return account, nil +} + +func (s *adminServiceImpl) CreateAccount(ctx context.Context, input *CreateAccountInput) (*Account, error) { + accountExtra, err := normalizeOpenAILongContextBillingExtra(input.Platform, input.Extra) + if err != nil { + return nil, err + } + + // 绑定分组 + groupIDs := input.GroupIDs + // 如果没有指定分组,自动绑定对应平台的默认分组 + if len(groupIDs) == 0 && !input.SkipDefaultGroupBind { + defaultGroupName := input.Platform + "-default" + groups, err := s.groupRepo.ListActiveByPlatform(ctx, input.Platform) + if err == nil { + for _, g := range groups { + if g.Name == defaultGroupName { + groupIDs = []int64{g.ID} + break + } + } + } + } + + // 检查混合渠道风险（除非用户已确认） + if len(groupIDs) > 0 && !input.SkipMixedChannelCheck { + if err := s.checkMixedChannelRisk(ctx, 0, input.Platform, groupIDs); err != nil { + return nil, err + } + } + + // 校验并规范化请求头覆写配置（header 名小写化、格式检查） + if err := NormalizeHeaderOverrideCredentials(input.Credentials); err != nil { + return nil, err + } + + account, err := buildAccountForCreate(input, accountExtra) + if err != nil { + return nil, err + } if err := s.accountRepo.Create(ctx, account); err != nil { return nil, err } diff --git a/backend/internal/service/admin_service.go b/backend/internal/service/admin_service.go index 2a7125f51f..964566a80c 100644 --- a/backend/internal/service/admin_service.go +++ b/backend/internal/service/admin_service.go @@ -67,6 +67,12 @@ type AdminService interface { GetAccount(ctx context.Context, id int64) (*Account, error) GetAccountsByIDs(ctx context.Context, ids []int64) ([]*Account, error) CreateAccount(ctx context.Context, input *CreateAccountInput) (*Account, error) + // DuplicateAccount creates an independent account from an existing account's configuration. + // First-class runtime columns are intentionally reset by the normal account creation path. + DuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error) + // RecoverDuplicateAccount returns a previously committed duplicate for an ambiguous retry. + // It never creates an account. + RecoverDuplicateAccount(ctx context.Context, id int64, actorScope, operationKey string) (*Account, error) UpdateAccount(ctx context.Context, id int64, input *UpdateAccountInput) (*Account, error) // UpdateAccountExtra 仅对 Extra 做 JSONB 增量合并（key 级覆盖），不会影响其它字段或运行态键。 // 用于刷新流程持久化 account_uuid / org_uuid 等少量键，避免被全量快照覆盖。 @@ -577,6 +583,7 @@ type adminServiceImpl struct { userRepo UserRepository groupRepo GroupRepository accountRepo AccountRepository + accountDuplicateRepo AccountDuplicateRepository proxyRepo ProxyRepository apiKeyRepo APIKeyRepository redeemCodeRepo RedeemCodeRepository @@ -602,7 +609,7 @@ type userGroupRateBatchReader interface { func NewAdminService( userRepo UserRepository, groupRepo GroupRepository, - accountRepo AccountRepository, + accountRepo AdminAccountRepository, proxyRepo ProxyRepository, apiKeyRepo APIKeyRepository, redeemCodeRepo RedeemCodeRepository, @@ -623,6 +630,7 @@ func NewAdminService( userRepo: userRepo, groupRepo: groupRepo, accountRepo: accountRepo, + accountDuplicateRepo: accountRepo, proxyRepo: proxyRepo, apiKeyRepo: apiKeyRepo, redeemCodeRepo: redeemCodeRepo, diff --git a/backend/internal/service/admin_service_duplicate_account_test.go b/backend/internal/service/admin_service_duplicate_account_test.go new file mode 100644 index 0000000000..f89d887035 --- /dev/null +++ b/backend/internal/service/admin_service_duplicate_account_test.go @@ -0,0 +1,322 @@ +//go:build unit + +package service + +import ( + "context" + "errors" + "net/http" + "strings" + "testing" + "time" + "unicode/utf8" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/stretchr/testify/require" +) + +type duplicateAccountRepoStub struct { + *sparkShadowRepoStub + atomicCreateErr error + accountGroupsOf map[int64][]AccountGroup +} + +func newDuplicateAccountRepoStub() *duplicateAccountRepoStub { + return &duplicateAccountRepoStub{ + sparkShadowRepoStub: newSparkShadowRepoStub(), + accountGroupsOf: make(map[int64][]AccountGroup), + } +} + +func (s *duplicateAccountRepoStub) CreateWithAccountGroups(ctx context.Context, account *Account, groups []AccountGroup) error { + if s.atomicCreateErr != nil { + return s.atomicCreateErr + } + groupIDs := make([]int64, 0, len(groups)) + for _, group := range groups { + groupIDs = append(groupIDs, group.GroupID) + } + account.GroupIDs = groupIDs + if err := s.Create(ctx, account); err != nil { + return err + } + clonedGroups := make([]AccountGroup, len(groups)) + copy(clonedGroups, groups) + for i := range clonedGroups { + clonedGroups[i].AccountID = account.ID + } + account.AccountGroups = clonedGroups + s.accountGroupsOf[account.ID] = clonedGroups + if len(groupIDs) > 0 { + s.groupsOf[account.ID] = append([]int64(nil), groupIDs...) + } + stored := *account + s.accounts[account.ID] = &stored + s.mockAccountRepoForGemini.accountsByID[account.ID] = &stored + return nil +} + +func (s *duplicateAccountRepoStub) FindByExtraField(_ context.Context, key string, value any) ([]Account, error) { + wanted, ok := value.(string) + if !ok { + return nil, nil + } + var matches []Account + for _, account := range s.accounts { + if actual, ok := account.Extra[key].(string); ok && actual == wanted { + matches = append(matches, *account) + } + } + return matches, nil +} + +func TestDuplicateAccountCopiesConfigurationAndResetsRuntimeState(t *testing.T) { + ctx := context.Background() + repo := newDuplicateAccountRepoStub() + svc := &adminServiceImpl{accountRepo: repo, accountDuplicateRepo: repo} + + notes := "keep this note" + proxyID := int64(17) + originalProxyID := int64(11) + rateMultiplier := 1.25 + loadFactor := 9 + expiresAt := time.Date(2027, time.March, 4, 5, 6, 7, 0, time.UTC) + rateLimitedAt := time.Now().Add(-time.Minute) + rateLimitResetAt := time.Now().Add(time.Hour) + overloadUntil := time.Now().Add(2 * time.Hour) + tempUnschedulableUntil := time.Now().Add(3 * time.Hour) + sessionWindowStart := time.Now().Add(-2 * time.Hour) + sessionWindowEnd := time.Now().Add(2 * time.Hour) + + source := &Account{ + Name: "primary", + Notes: ¬es, + Platform: PlatformAnthropic, + Type: AccountTypeAPIKey, + ProxyID: &proxyID, + ProxyFallbackOriginID: &originalProxyID, + Concurrency: 6, + Priority: 40, + RateMultiplier: &rateMultiplier, + LoadFactor: &loadFactor, + Status: StatusError, + Schedulable: true, + ErrorMessage: "upstream unavailable", + ExpiresAt: &expiresAt, + AutoPauseOnExpired: false, + Credentials: map[string]any{ + "api_key": "secret", + "nested": map[string]any{"token": "source-token"}, + }, + Extra: map[string]any{ + "config": map[string]any{"region": "us-east-1"}, + "items": []any{map[string]any{"enabled": true}}, + "quota_limit": 1000, + "quota_used": 450, + "quota_daily_used": 25, + "quota_daily_start": "2026-07-15T00:00:00Z", + "model_rate_limits": map[string]any{"gpt-5": "2099-01-01T00:00:00Z"}, + "codex_5h_used_percent": 80, + "codex_cli_only": true, + "grok_usage_snapshot": map[string]any{"status_code": 429}, + "openai_responses_supported": false, + "openai_compact_checked_at": "2026-07-15T00:00:00Z", + "session_window_utilization": 0.8, + "passive_usage_sampled_at": "2026-07-15T00:00:00Z", + "antigravity_force_token_refresh": true, + "antigravity_credits_overages": map[string]any{"enabled": true}, + "crs_account_id": "remote-42", + "crs_kind": "openai-api-key", + "crs_synced_at": "2026-07-15T00:00:00Z", + }, + GroupIDs: []int64{7, 3}, + AccountGroups: []AccountGroup{{GroupID: 7, Priority: 50}, {GroupID: 3, Priority: 7}}, + RateLimitedAt: &rateLimitedAt, + RateLimitResetAt: &rateLimitResetAt, + OverloadUntil: &overloadUntil, + TempUnschedulableUntil: &tempUnschedulableUntil, + TempUnschedulableReason: "maintenance", + SessionWindowStart: &sessionWindowStart, + SessionWindowEnd: &sessionWindowEnd, + SessionWindowStatus: "active", + } + require.NoError(t, repo.Create(ctx, source)) + + duplicate, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "") + + require.NoError(t, err) + require.NotEqual(t, source.ID, duplicate.ID) + require.Equal(t, "primary (Copy)", duplicate.Name) + require.Equal(t, source.Platform, duplicate.Platform) + require.Equal(t, source.Type, duplicate.Type) + require.Equal(t, source.Concurrency, duplicate.Concurrency) + require.Equal(t, source.Priority, duplicate.Priority) + require.Equal(t, source.AutoPauseOnExpired, duplicate.AutoPauseOnExpired) + require.Equal(t, source.GroupIDs, duplicate.GroupIDs) + require.Equal(t, source.Credentials, duplicate.Credentials) + require.Equal(t, map[string]any{ + "config": map[string]any{"region": "us-east-1"}, + "items": []any{map[string]any{"enabled": true}}, + "quota_limit": float64(1000), + "codex_cli_only": true, + }, duplicate.Extra) + require.NotNil(t, duplicate.ExpiresAt) + require.True(t, source.ExpiresAt.Equal(*duplicate.ExpiresAt)) + require.Equal(t, source.Notes, duplicate.Notes) + require.Equal(t, source.ProxyFallbackOriginID, duplicate.ProxyID) + require.Equal(t, source.RateMultiplier, duplicate.RateMultiplier) + require.Equal(t, source.LoadFactor, duplicate.LoadFactor) + require.Equal(t, source.GroupIDs, repo.groupsOf[duplicate.ID]) + require.Equal(t, []AccountGroup{ + {AccountID: duplicate.ID, GroupID: 7, Priority: 50}, + {AccountID: duplicate.ID, GroupID: 3, Priority: 7}, + }, repo.accountGroupsOf[duplicate.ID]) + + require.Equal(t, StatusActive, duplicate.Status) + require.False(t, duplicate.Schedulable) + require.Empty(t, duplicate.ErrorMessage) + require.Nil(t, duplicate.LastUsedAt) + require.Nil(t, duplicate.RateLimitedAt) + require.Nil(t, duplicate.RateLimitResetAt) + require.Nil(t, duplicate.OverloadUntil) + require.Nil(t, duplicate.TempUnschedulableUntil) + require.Empty(t, duplicate.TempUnschedulableReason) + require.Nil(t, duplicate.SessionWindowStart) + require.Nil(t, duplicate.SessionWindowEnd) + require.Empty(t, duplicate.SessionWindowStatus) + + duplicate.Credentials["nested"].(map[string]any)["token"] = "changed" + duplicate.Extra["config"].(map[string]any)["region"] = "changed" + duplicate.Extra["items"].([]any)[0].(map[string]any)["enabled"] = false + storedSource, getErr := repo.GetByID(ctx, source.ID) + require.NoError(t, getErr) + require.Equal(t, "source-token", storedSource.Credentials["nested"].(map[string]any)["token"]) + require.Equal(t, "us-east-1", storedSource.Extra["config"].(map[string]any)["region"]) + require.Equal(t, true, storedSource.Extra["items"].([]any)[0].(map[string]any)["enabled"]) + require.Equal(t, "remote-42", storedSource.Extra["crs_account_id"]) +} + +func TestDuplicateAccountRejectsCredentialShadow(t *testing.T) { + ctx := context.Background() + repo := newDuplicateAccountRepoStub() + svc := &adminServiceImpl{accountRepo: repo, accountDuplicateRepo: repo} + parentID := int64(99) + shadow := &Account{ + Name: "shadow", + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + ParentAccountID: &parentID, + QuotaDimension: QuotaDimensionSpark, + } + require.NoError(t, repo.Create(ctx, shadow)) + + _, err := svc.DuplicateAccount(ctx, shadow.ID, "admin:1", "") + + require.Error(t, err) + require.Equal(t, http.StatusBadRequest, infraerrors.Code(err)) + require.Equal(t, "ACCOUNT_DUPLICATE_SHADOW_UNSUPPORTED", infraerrors.Reason(err)) + require.Len(t, repo.accounts, 1) +} + +func TestDuplicateAccountRejectsRotatingOrUnknownCredentialTypes(t *testing.T) { + for _, accountType := range []string{AccountTypeOAuth, AccountTypeSetupToken, "legacy-cookie"} { + t.Run(accountType, func(t *testing.T) { + ctx := context.Background() + repo := newDuplicateAccountRepoStub() + svc := &adminServiceImpl{accountRepo: repo, accountDuplicateRepo: repo} + source := &Account{ + Name: "rotating-credential-account", + Platform: PlatformOpenAI, + Type: accountType, + Credentials: map[string]any{"refresh_token": "shared-token"}, + } + require.NoError(t, repo.Create(ctx, source)) + + _, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "") + + require.Error(t, err) + require.Equal(t, http.StatusBadRequest, infraerrors.Code(err)) + require.Equal(t, "ACCOUNT_DUPLICATE_CREDENTIAL_TYPE_UNSUPPORTED", infraerrors.Reason(err)) + require.Len(t, repo.accounts, 1) + }) + } +} + +func TestDuplicateAccountPreservesUngroupedState(t *testing.T) { + ctx := context.Background() + repo := newDuplicateAccountRepoStub() + svc := &adminServiceImpl{accountRepo: repo, accountDuplicateRepo: repo} + source := &Account{ + Name: "ungrouped", + Platform: PlatformAnthropic, + Type: AccountTypeAPIKey, + Credentials: map[string]any{"api_key": "secret"}, + GroupIDs: nil, + } + require.NoError(t, repo.Create(ctx, source)) + + duplicate, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "") + + require.NoError(t, err) + require.Empty(t, duplicate.GroupIDs) + require.NotContains(t, repo.groupsOf, duplicate.ID) +} + +func TestDuplicateAccountAtomicCreateFailureLeavesNoOrphan(t *testing.T) { + ctx := context.Background() + repo := newDuplicateAccountRepoStub() + svc := &adminServiceImpl{accountRepo: repo, accountDuplicateRepo: repo} + source := &Account{ + Name: "source", + Platform: PlatformAnthropic, + Type: AccountTypeAPIKey, + Credentials: map[string]any{"api_key": "secret"}, + GroupIDs: []int64{7}, + AccountGroups: []AccountGroup{{GroupID: 7, Priority: 25}}, + } + require.NoError(t, repo.Create(ctx, source)) + repo.atomicCreateErr = errors.New("group binding failed") + + _, err := svc.DuplicateAccount(ctx, source.ID, "admin:1", "") + + require.ErrorContains(t, err, "group binding failed") + require.Len(t, repo.accounts, 1) +} + +func TestDuplicateAccountNamePreservesSuffixWithinSchemaLimit(t *testing.T) { + name := duplicateAccountName(strings.Repeat("界", 100)) + + require.Equal(t, 100, utf8.RuneCountInString(name)) + require.True(t, strings.HasSuffix(name, " (Copy)")) +} + +func TestDuplicateAccountReturnsExistingCopyForSameOperationKey(t *testing.T) { + ctx := context.Background() + repo := newDuplicateAccountRepoStub() + svc := &adminServiceImpl{accountRepo: repo, accountDuplicateRepo: repo} + source := &Account{ + Name: "source", + Platform: PlatformAnthropic, + Type: AccountTypeAPIKey, + Credentials: map[string]any{"api_key": "secret"}, + } + require.NoError(t, repo.Create(ctx, source)) + + first, err := svc.DuplicateAccount(ctx, source.ID, "admin:7", "stable-operation-key") + require.NoError(t, err) + second, err := svc.DuplicateAccount(ctx, source.ID, "admin:7", "stable-operation-key") + require.NoError(t, err) + recovered, err := svc.RecoverDuplicateAccount(ctx, source.ID, "admin:7", "stable-operation-key") + require.NoError(t, err) + otherAdminRecovery, err := svc.RecoverDuplicateAccount(ctx, source.ID, "admin:8", "stable-operation-key") + require.NoError(t, err) + otherAdminCopy, err := svc.DuplicateAccount(ctx, source.ID, "admin:8", "stable-operation-key") + require.NoError(t, err) + + require.Equal(t, first.ID, second.ID) + require.Equal(t, first.ID, recovered.ID) + require.Nil(t, otherAdminRecovery, "durable recovery identity must remain scoped to the initiating admin") + require.NotEqual(t, first.ID, otherAdminCopy.ID) + require.Len(t, repo.accounts, 3) + require.NotEmpty(t, first.Extra[duplicateAccountOperationIDExtraKey]) +} diff --git a/frontend/src/api/__tests__/admin.accounts.duplicate.spec.ts b/frontend/src/api/__tests__/admin.accounts.duplicate.spec.ts new file mode 100644 index 0000000000..f6bc22c266 --- /dev/null +++ b/frontend/src/api/__tests__/admin.accounts.duplicate.spec.ts @@ -0,0 +1,59 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { post } = vi.hoisted(() => ({ + post: vi.fn() +})) + +vi.mock('@/api/client', () => ({ + apiClient: { post } +})) + +import { duplicate } from '@/api/admin/accounts' + +describe('admin account duplicate API', () => { + beforeEach(() => { + sessionStorage.clear() + post.mockReset() + post.mockResolvedValue({ data: { id: 43, name: 'primary (Copy)' } }) + vi.spyOn(globalThis.crypto, 'randomUUID').mockReturnValue('11111111-1111-4111-8111-111111111111') + }) + + it('sends a stable idempotency key with the duplicate request', async () => { + const account = await duplicate(42) + + expect(post).toHaveBeenCalledWith('/admin/accounts/42/duplicate', undefined, { + headers: { + 'Idempotency-Key': 'account-duplicate-42-11111111-1111-4111-8111-111111111111' + } + }) + expect(account).toEqual({ id: 43, name: 'primary (Copy)' }) + }) + + it('reuses the operation key after an ambiguous failed request', async () => { + post.mockRejectedValueOnce(new Error('network timeout')) + await expect(duplicate(99)).rejects.toThrow('network timeout') + + post.mockResolvedValueOnce({ data: { id: 100, name: 'retry (Copy)' } }) + await duplicate(99) + + expect(post).toHaveBeenCalledTimes(2) + const firstHeaders = post.mock.calls[0][2].headers + const secondHeaders = post.mock.calls[1][2].headers + expect(secondHeaders).toEqual(firstHeaders) + }) + + it('reuses the operation key after a page reload', async () => { + post.mockRejectedValueOnce(new Error('network timeout')) + await expect(duplicate(77)).rejects.toThrow('network timeout') + const firstHeaders = post.mock.calls[0][2].headers + + vi.resetModules() + post.mockResolvedValueOnce({ data: { id: 78, name: 'reload (Copy)' } }) + const { duplicate: duplicateAfterReload } = await import('@/api/admin/accounts') + await duplicateAfterReload(77) + + expect(post).toHaveBeenCalledTimes(2) + expect(post.mock.calls[1][2].headers).toEqual(firstHeaders) + expect(sessionStorage.length).toBe(0) + }) +}) diff --git a/frontend/src/api/admin/accounts.ts b/frontend/src/api/admin/accounts.ts index 2f9625b430..64a47a50fb 100644 --- a/frontend/src/api/admin/accounts.ts +++ b/frontend/src/api/admin/accounts.ts @@ -138,6 +138,50 @@ export async function create(accountData: CreateAccountRequest): Promise() + +function duplicateOperationStorageKey(id: number): string { + return `sub2api:admin:account-duplicate:${id}` +} + +function getStoredDuplicateOperationKey(id: number): string | null { + try { + return globalThis.sessionStorage?.getItem(duplicateOperationStorageKey(id)) ?? null + } catch { + return null + } +} + +function storeDuplicateOperationKey(id: number, key: string | null): void { + try { + if (key) globalThis.sessionStorage?.setItem(duplicateOperationStorageKey(id), key) + else globalThis.sessionStorage?.removeItem(duplicateOperationStorageKey(id)) + } catch { + // In-memory retry protection still works when browser storage is unavailable. + } +} + +export async function duplicate(id: number): Promise { + let idempotencyKey = duplicateOperationKeys.get(id) ?? getStoredDuplicateOperationKey(id) + if (!idempotencyKey) { + const requestID = globalThis.crypto?.randomUUID?.() ?? `${Date.now()}-${Math.random().toString(36).slice(2)}` + idempotencyKey = `account-duplicate-${id}-${requestID}` + } + duplicateOperationKeys.set(id, idempotencyKey) + storeDuplicateOperationKey(id, idempotencyKey) + const { data } = await apiClient.post(`/admin/accounts/${id}/duplicate`, undefined, { + headers: { 'Idempotency-Key': idempotencyKey } + }) + duplicateOperationKeys.delete(id) + storeDuplicateOperationKey(id, null) + return data +} + /** * Update account * @param id - Account ID @@ -809,6 +853,7 @@ export const accountsAPI = { listWithEtag, getById, create, + duplicate, update, checkMixedChannelRisk, delete: deleteAccount, diff --git a/frontend/src/components/admin/account/AccountActionMenu.vue b/frontend/src/components/admin/account/AccountActionMenu.vue index 9bb5e8891e..14922a37b7 100644 --- a/frontend/src/components/admin/account/AccountActionMenu.vue +++ b/frontend/src/components/admin/account/AccountActionMenu.vue @@ -22,6 +22,10 @@ {{ t('admin.scheduledTests.schedule') }} +