From 3d6c207723b40c2f727f0fec3e37dbee49217b71 Mon Sep 17 00:00:00 2001 From: SavitarC <71271200+SavitarC@users.noreply.github.com> Date: Sun, 13 Sep 2026 11:49:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E8=AE=A2=E9=98=85?= =?UTF-8?q?=E7=AE=A1=E7=90=86=E6=89=B9=E9=87=8F=E6=93=8D=E4=BD=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../handler/admin/idempotency_helper.go | 54 ++++- .../admin/subscription_bulk_action_test.go | 208 ++++++++++++++++ .../handler/admin/subscription_handler.go | 20 +- backend/internal/server/routes/admin.go | 1 + .../subscription_bulk_action_routes_test.go | 50 ++++ backend/internal/service/idempotency.go | 68 +++++- .../service/idempotency_detached_test.go | 119 +++++++++ .../service/subscription_bulk_action.go | 129 ++++++++++ .../service/subscription_bulk_action_test.go | 226 ++++++++++++++++++ ...bscription_bulk_action_transaction_test.go | 136 +++++++++++ .../admin.subscriptions.bulk.spec.ts | 29 +++ frontend/src/api/admin/subscriptions.ts | 46 +++- .../BulkSubscriptionActionDialog.vue | 225 +++++++++++++++++ .../BulkSubscriptionActionDialog.spec.ts | 222 +++++++++++++++++ .../bulkSubscriptionOperation.spec.ts | 41 ++++ .../subscription/bulkSubscriptionOperation.ts | 71 ++++++ .../src/i18n/locales/en/admin/channels.ts | 35 +++ .../src/i18n/locales/zh/admin/channels.ts | 35 +++ .../src/views/admin/SubscriptionsView.vue | 170 ++++++++++++- .../SubscriptionsView.bulkActions.spec.ts | 143 +++++++++++ 20 files changed, 2002 insertions(+), 26 deletions(-) create mode 100644 backend/internal/handler/admin/subscription_bulk_action_test.go create mode 100644 backend/internal/server/routes/subscription_bulk_action_routes_test.go create mode 100644 backend/internal/service/idempotency_detached_test.go create mode 100644 backend/internal/service/subscription_bulk_action.go create mode 100644 backend/internal/service/subscription_bulk_action_test.go create mode 100644 backend/internal/service/subscription_bulk_action_transaction_test.go create mode 100644 frontend/src/api/__tests__/admin.subscriptions.bulk.spec.ts create mode 100644 frontend/src/components/admin/subscription/BulkSubscriptionActionDialog.vue create mode 100644 frontend/src/components/admin/subscription/__tests__/BulkSubscriptionActionDialog.spec.ts create mode 100644 frontend/src/components/admin/subscription/__tests__/bulkSubscriptionOperation.spec.ts create mode 100644 frontend/src/components/admin/subscription/bulkSubscriptionOperation.ts create mode 100644 frontend/src/views/admin/__tests__/SubscriptionsView.bulkActions.spec.ts diff --git a/backend/internal/handler/admin/idempotency_helper.go b/backend/internal/handler/admin/idempotency_helper.go index 1894faeaae..8017e7211e 100644 --- a/backend/internal/handler/admin/idempotency_helper.go +++ b/backend/internal/handler/admin/idempotency_helper.go @@ -27,10 +27,27 @@ func executeAdminIdempotent( payload any, ttl time.Duration, execute func(context.Context) (any, error), +) (*service.IdempotencyExecuteResult, error) { + return executeAdminIdempotentWithTimeout(c, scope, payload, ttl, 0, execute) +} + +func executeAdminIdempotentWithTimeout( + c *gin.Context, + scope string, + payload any, + ttl time.Duration, + executionTimeout time.Duration, + execute func(context.Context) (any, error), ) (*service.IdempotencyExecuteResult, error) { coordinator := service.DefaultIdempotencyCoordinator() if coordinator == nil { - data, err := execute(c.Request.Context()) + ctx := c.Request.Context() + if executionTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(context.WithoutCancel(ctx), executionTimeout) + defer cancel() + } + data, err := execute(ctx) if err != nil { return nil, err } @@ -38,14 +55,15 @@ func executeAdminIdempotent( } return coordinator.Execute(c.Request.Context(), service.IdempotencyExecuteOptions{ - Scope: scope, - ActorScope: adminActorScope(c), - Method: c.Request.Method, - Route: c.FullPath(), - IdempotencyKey: c.GetHeader("Idempotency-Key"), - Payload: payload, - RequireKey: true, - TTL: ttl, + Scope: scope, + ActorScope: adminActorScope(c), + Method: c.Request.Method, + Route: c.FullPath(), + IdempotencyKey: c.GetHeader("Idempotency-Key"), + Payload: payload, + RequireKey: true, + TTL: ttl, + ExecutionTimeout: executionTimeout, }, execute) } @@ -64,7 +82,18 @@ func executeAdminIdempotentJSON( ttl time.Duration, execute func(context.Context) (any, error), ) { - executeAdminIdempotentJSONWithMode(c, scope, payload, ttl, idempotencyStoreUnavailableFailClose, execute) + executeAdminIdempotentJSONWithMode(c, scope, payload, ttl, 0, idempotencyStoreUnavailableFailClose, execute) +} + +func executeAdminIdempotentJSONWithTimeout( + c *gin.Context, + scope string, + payload any, + ttl time.Duration, + executionTimeout time.Duration, + execute func(context.Context) (any, error), +) { + executeAdminIdempotentJSONWithMode(c, scope, payload, ttl, executionTimeout, idempotencyStoreUnavailableFailClose, execute) } func executeAdminIdempotentJSONFailOpenOnStoreUnavailable( @@ -74,7 +103,7 @@ func executeAdminIdempotentJSONFailOpenOnStoreUnavailable( ttl time.Duration, execute func(context.Context) (any, error), ) { - executeAdminIdempotentJSONWithMode(c, scope, payload, ttl, idempotencyStoreUnavailableFailOpen, execute) + executeAdminIdempotentJSONWithMode(c, scope, payload, ttl, 0, idempotencyStoreUnavailableFailOpen, execute) } func executeAdminIdempotentJSONWithMode( @@ -82,10 +111,11 @@ func executeAdminIdempotentJSONWithMode( scope string, payload any, ttl time.Duration, + executionTimeout time.Duration, mode idempotencyStoreUnavailableMode, execute func(context.Context) (any, error), ) { - result, err := executeAdminIdempotent(c, scope, payload, ttl, execute) + result, err := executeAdminIdempotentWithTimeout(c, scope, payload, ttl, executionTimeout, execute) if err != nil { if infraerrors.Code(err) == infraerrors.Code(service.ErrIdempotencyStoreUnavail) { strategy := "fail_close" diff --git a/backend/internal/handler/admin/subscription_bulk_action_test.go b/backend/internal/handler/admin/subscription_bulk_action_test.go new file mode 100644 index 0000000000..eb1f08e293 --- /dev/null +++ b/backend/internal/handler/admin/subscription_bulk_action_test.go @@ -0,0 +1,208 @@ +package admin + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +type bulkActionHandlerSubscriptionRepo struct { + service.UserSubscriptionRepository + sub *service.UserSubscription + extendCalls int +} + +type cancellationAwareBulkActionRepo struct { + *bulkActionHandlerSubscriptionRepo + cancelRequest context.CancelFunc +} + +func (r *cancellationAwareBulkActionRepo) GetByID(ctx context.Context, id int64) (*service.UserSubscription, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return r.bulkActionHandlerSubscriptionRepo.GetByID(ctx, id) +} + +func (r *cancellationAwareBulkActionRepo) ExtendExpiry(ctx context.Context, id int64, expiry time.Time) error { + if err := ctx.Err(); err != nil { + return err + } + err := r.bulkActionHandlerSubscriptionRepo.ExtendExpiry(ctx, id, expiry) + r.cancelRequest() + return err +} + +type cancellationAwareAdminIdempotencyRepo struct { + *memoryIdempotencyRepoStub + saves int +} + +func (r *cancellationAwareAdminIdempotencyRepo) MarkSucceeded(ctx context.Context, id int64, status int, body string, expiresAt time.Time) error { + if err := ctx.Err(); err != nil { + return err + } + r.saves++ + return r.memoryIdempotencyRepoStub.MarkSucceeded(ctx, id, status, body, expiresAt) +} + +func (r *bulkActionHandlerSubscriptionRepo) GetByID(_ context.Context, id int64) (*service.UserSubscription, error) { + if r.sub == nil || r.sub.ID != id { + return nil, service.ErrSubscriptionNotFound + } + copy := *r.sub + return ©, nil +} + +func (r *bulkActionHandlerSubscriptionRepo) ExtendExpiry(_ context.Context, _ int64, expiry time.Time) error { + r.extendCalls++ + r.sub.ExpiresAt = expiry + return nil +} + +func bulkActionHandlerRequest(router *gin.Engine, path, body, key string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, path, bytes.NewBufferString(body)) + req.Header.Set("Content-Type", "application/json") + if key != "" { + req.Header.Set("Idempotency-Key", key) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec +} + +func TestSubscriptionBulkAction_ReplaysPartialResultWithoutRepeatingExtension(t *testing.T) { + gin.SetMode(gin.TestMode) + cfg := service.DefaultIdempotencyConfig() + cfg.ObserveOnly = false + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(newMemoryIdempotencyRepoStub(), cfg)) + t.Cleanup(func() { service.SetDefaultIdempotencyCoordinator(nil) }) + expiresAt := time.Now().AddDate(0, 0, 30) + repo := &bulkActionHandlerSubscriptionRepo{sub: &service.UserSubscription{ID: 1, UserID: 1, GroupID: 10, ExpiresAt: expiresAt}} + svc := service.NewSubscriptionService(nil, repo, nil, nil, nil) + t.Cleanup(svc.Stop) + h := NewSubscriptionHandler(svc) + router := gin.New() + path := "/api/v1/admin/subscriptions/bulk-action" + router.POST(path, h.BulkAction) + body := `{"subscription_ids":[1,404,1],"action":"extend","days":7}` + + missingKey := bulkActionHandlerRequest(router, path, body, "") + require.Equal(t, http.StatusBadRequest, missingKey.Code) + require.Zero(t, repo.extendCalls) + + first := bulkActionHandlerRequest(router, path, body, "bulk-extend-once") + require.Equal(t, http.StatusOK, first.Code, first.Body.String()) + var result struct { + Data service.BulkSubscriptionActionResult `json:"data"` + } + require.NoError(t, json.Unmarshal(first.Body.Bytes(), &result)) + require.Equal(t, 1, result.Data.SuccessCount) + require.Equal(t, 1, result.Data.FailedCount) + require.Len(t, result.Data.Results, 2) + + replayed := bulkActionHandlerRequest(router, path, body, "bulk-extend-once") + require.Equal(t, http.StatusOK, replayed.Code, replayed.Body.String()) + require.Equal(t, "true", replayed.Header().Get("X-Idempotency-Replayed")) + require.JSONEq(t, first.Body.String(), replayed.Body.String()) + require.Equal(t, 1, repo.extendCalls) + require.Equal(t, expiresAt.AddDate(0, 0, 7), repo.sub.ExpiresAt) + + conflict := bulkActionHandlerRequest(router, path, `{"subscription_ids":[1,404,1],"action":"extend","days":14}`, "bulk-extend-once") + require.Equal(t, http.StatusConflict, conflict.Code) + require.Equal(t, 1, repo.extendCalls) +} + +func TestSubscriptionBulkAction_ClientCancellationStillPersistsReplay(t *testing.T) { + gin.SetMode(gin.TestMode) + idempotencyRepo := &cancellationAwareAdminIdempotencyRepo{memoryIdempotencyRepoStub: newMemoryIdempotencyRepoStub()} + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(idempotencyRepo, service.DefaultIdempotencyConfig())) + t.Cleanup(func() { service.SetDefaultIdempotencyCoordinator(nil) }) + requestCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + expiresAt := time.Now().AddDate(0, 0, 30) + repo := &cancellationAwareBulkActionRepo{ + bulkActionHandlerSubscriptionRepo: &bulkActionHandlerSubscriptionRepo{sub: &service.UserSubscription{ID: 1, UserID: 1, GroupID: 10, ExpiresAt: expiresAt}}, + cancelRequest: cancel, + } + svc := service.NewSubscriptionService(nil, repo, nil, nil, nil) + t.Cleanup(svc.Stop) + router := gin.New() + path := "/api/v1/admin/subscriptions/bulk-action" + router.POST(path, NewSubscriptionHandler(svc).BulkAction) + body := `{"subscription_ids":[1],"action":"extend","days":7}` + request := httptest.NewRequest(http.MethodPost, path, bytes.NewBufferString(body)).WithContext(requestCtx) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Idempotency-Key", "canceled-request") + first := httptest.NewRecorder() + router.ServeHTTP(first, request) + require.ErrorIs(t, requestCtx.Err(), context.Canceled) + require.Equal(t, http.StatusOK, first.Code, first.Body.String()) + require.Equal(t, 1, idempotencyRepo.saves) + + replayed := bulkActionHandlerRequest(router, path, body, "canceled-request") + require.Equal(t, http.StatusOK, replayed.Code, replayed.Body.String()) + require.Equal(t, "true", replayed.Header().Get("X-Idempotency-Replayed")) + require.JSONEq(t, first.Body.String(), replayed.Body.String()) + require.Equal(t, 1, repo.extendCalls) + require.Equal(t, expiresAt.AddDate(0, 0, 7), repo.sub.ExpiresAt) +} + +func TestSubscriptionBulkAction_ValidatesBeforeIdempotencyAndExecution(t *testing.T) { + gin.SetMode(gin.TestMode) + service.SetDefaultIdempotencyCoordinator(service.NewIdempotencyCoordinator(storeUnavailableRepoStub{}, service.DefaultIdempotencyConfig())) + t.Cleanup(func() { service.SetDefaultIdempotencyCoordinator(nil) }) + router := gin.New() + path := "/api/v1/admin/subscriptions/bulk-action" + router.POST(path, NewSubscriptionHandler(nil).BulkAction) + tooManyIDs := make([]int64, 101) + for i := range tooManyIDs { + tooManyIDs[i] = 1 + } + tooManyBody, err := json.Marshal(service.BulkSubscriptionActionInput{SubscriptionIDs: tooManyIDs, Action: "revoke"}) + require.NoError(t, err) + for _, body := range []string{ + `{"subscription_ids":[1],"action":"delete"}`, + `{"subscription_ids":[1,0],"action":"revoke"}`, + `{"subscription_ids":[],"action":"restore"}`, + `{"subscription_ids":[1],"action":"extend","days":0}`, + `{"subscription_ids":[1],"action":"extend","days":36501}`, + `{"subscription_ids":[1],"action":"reset_quota"}`, + `{"subscription_ids":[1],"action":"reset_quota","daily":"yes"}`, + `{"subscription_ids":[1],"action":"revoke"`, + string(tooManyBody), + } { + t.Run(body, func(t *testing.T) { + response := bulkActionHandlerRequest(router, path, body, "bulk-invalid") + require.Equal(t, http.StatusBadRequest, response.Code, response.Body.String()) + }) + } + + // A valid request must still fail closed when replay protection is unavailable. + response := bulkActionHandlerRequest(router, path, `{"subscription_ids":[1],"action":"revoke"}`, "bulk-valid") + require.Equal(t, http.StatusServiceUnavailable, response.Code) +} + +func TestSubscriptionBulkAssign_RejectsInvalidUserIDsBeforeExecution(t *testing.T) { + gin.SetMode(gin.TestMode) + router := gin.New() + path := "/api/v1/admin/subscriptions/bulk-assign" + router.POST(path, NewSubscriptionHandler(nil).BulkAssign) + for _, ids := range [][]int64{{}, {1, 0}, {1, -1}, make([]int64, 101)} { + t.Run(fmt.Sprint(len(ids), ids), func(t *testing.T) { + body, err := json.Marshal(BulkAssignSubscriptionRequest{UserIDs: ids, GroupID: 1, ValidityDays: 30}) + require.NoError(t, err) + response := bulkActionHandlerRequest(router, path, string(body), "") + require.Equal(t, http.StatusBadRequest, response.Code, response.Body.String()) + }) + } +} diff --git a/backend/internal/handler/admin/subscription_handler.go b/backend/internal/handler/admin/subscription_handler.go index d370dad886..2174e18840 100644 --- a/backend/internal/handler/admin/subscription_handler.go +++ b/backend/internal/handler/admin/subscription_handler.go @@ -3,6 +3,7 @@ package admin import ( "context" "strconv" + "time" "github.com/Wei-Shaw/sub2api/internal/handler/dto" "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" @@ -48,7 +49,7 @@ type AssignSubscriptionRequest struct { // BulkAssignSubscriptionRequest represents bulk assign subscription request type BulkAssignSubscriptionRequest struct { - UserIDs []int64 `json:"user_ids" binding:"required,min=1"` + UserIDs []int64 `json:"user_ids" binding:"required,min=1,max=100,dive,gt=0"` GroupID int64 `json:"group_id" binding:"required"` ValidityDays int `json:"validity_days" binding:"omitempty,max=36500"` // max 100 years Notes string `json:"notes"` @@ -186,6 +187,23 @@ func (h *SubscriptionHandler) BulkAssign(c *gin.Context) { response.Success(c, dto.BulkAssignResultFromService(result)) } +// BulkAction applies one operation to selected subscriptions, returning each outcome. +// POST /api/v1/admin/subscriptions/bulk-action +func (h *SubscriptionHandler) BulkAction(c *gin.Context) { + var req service.BulkSubscriptionActionInput + if err := c.ShouldBindJSON(&req); err != nil { + response.BadRequest(c, "Invalid request: "+err.Error()) + return + } + if err := req.Validate(); err != nil { + response.ErrorFrom(c, err) + return + } + executeAdminIdempotentJSONWithTimeout(c, "admin.subscriptions.bulk-action", req, service.DefaultWriteIdempotencyTTL(), 2*time.Minute, func(ctx context.Context) (any, error) { + return h.subscriptionService.BulkSubscriptionAction(ctx, &req) + }) +} + // Extend handles adjusting a subscription (extend or shorten) // POST /api/v1/admin/subscriptions/:id/extend func (h *SubscriptionHandler) Extend(c *gin.Context) { diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go index c16ef695e0..405a5f343f 100644 --- a/backend/internal/server/routes/admin.go +++ b/backend/internal/server/routes/admin.go @@ -675,6 +675,7 @@ func registerSubscriptionRoutes(admin *gin.RouterGroup, h *handler.Handlers) { subscriptions.GET("/:id/progress", h.Admin.Subscription.GetProgress) subscriptions.POST("/assign", h.Admin.Subscription.Assign) subscriptions.POST("/bulk-assign", h.Admin.Subscription.BulkAssign) + subscriptions.POST("/bulk-action", h.Admin.Subscription.BulkAction) subscriptions.POST("/:id/extend", h.Admin.Subscription.Extend) subscriptions.POST("/:id/reset-quota", h.Admin.Subscription.ResetQuota) subscriptions.POST("/:id/revoke", h.Admin.Subscription.Revoke) diff --git a/backend/internal/server/routes/subscription_bulk_action_routes_test.go b/backend/internal/server/routes/subscription_bulk_action_routes_test.go new file mode 100644 index 0000000000..f758d48e0c --- /dev/null +++ b/backend/internal/server/routes/subscription_bulk_action_routes_test.go @@ -0,0 +1,50 @@ +package routes + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/handler" + adminhandler "github.com/Wei-Shaw/sub2api/internal/handler/admin" + servermiddleware "github.com/Wei-Shaw/sub2api/internal/server/middleware" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestSubscriptionBulkActionRoutesRequireAdminAuthentication(t *testing.T) { + gin.SetMode(gin.TestMode) + router := gin.New() + handlers := &handler.Handlers{Admin: &handler.AdminHandlers{Subscription: adminhandler.NewSubscriptionHandler(nil)}} + adminAuth := servermiddleware.AdminAuthMiddleware(func(c *gin.Context) { + if c.GetHeader("Authorization") == "" { + servermiddleware.AbortWithError(c, http.StatusUnauthorized, "UNAUTHORIZED", "Authorization required") + return + } + servermiddleware.AbortWithError(c, http.StatusForbidden, "FORBIDDEN", "Admin access required") + }) + auditLog := servermiddleware.AuditLogMiddleware(func(c *gin.Context) { c.Next() }) + stepUp := servermiddleware.StepUpAuthMiddleware(func(c *gin.Context) { c.Next() }) + RegisterAdminRoutes(router.Group("/api/v1"), handlers, adminAuth, auditLog, stepUp, nil, nil) + + for _, path := range []string{"/api/v1/admin/subscriptions/bulk-action", "/api/v1/admin/subscriptions/bulk-assign"} { + for _, tc := range []struct { + name string + auth string + wantStatus int + }{ + {name: "unauthenticated", wantStatus: http.StatusUnauthorized}, + {name: "non-admin", auth: "Bearer user-token", wantStatus: http.StatusForbidden}, + } { + t.Run(path+"/"+tc.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, path, nil) + if tc.auth != "" { + request.Header.Set("Authorization", tc.auth) + } + router.ServeHTTP(recorder, request) + require.Equal(t, tc.wantStatus, recorder.Code) + }) + } + } +} diff --git a/backend/internal/service/idempotency.go b/backend/internal/service/idempotency.go index 4241468789..bb261b6718 100644 --- a/backend/internal/service/idempotency.go +++ b/backend/internal/service/idempotency.go @@ -87,6 +87,9 @@ type IdempotencyExecuteOptions struct { Payload any TTL time.Duration RequireKey bool + // ExecutionTimeout opts into bounded execution independent of client cancellation, + // with lease renewal and a separate short window to persist the final result. + ExecutionTimeout time.Duration } type IdempotencyExecuteResult struct { @@ -202,6 +205,11 @@ func (c *IdempotencyCoordinator) Execute( opts IdempotencyExecuteOptions, execute func(context.Context) (any, error), ) (*IdempotencyExecuteResult, error) { + if opts.ExecutionTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(context.WithoutCancel(ctx), opts.ExecutionTimeout) + defer cancel() + } if execute == nil { return nil, infraerrors.InternalServer("IDEMPOTENCY_EXECUTOR_NIL", "idempotency executor is nil") } @@ -392,12 +400,26 @@ func (c *IdempotencyCoordinator) Execute( return nil, ErrIdempotencyStoreUnavail } + if opts.ExecutionTimeout > 0 { + stopRenewal := c.renewProcessingLease(ctx, record, ttl) + defer stopRenewal() + } + execStart := time.Now() defer func() { recordIdempotencyProcessingDuration(opts.Route, opts.Scope, time.Since(execStart), nil) }() data, execErr := execute(ctx) + persistCtx := ctx + if opts.ExecutionTimeout > 0 { + // The execution deadline can expire after some items have completed. + // Store those outcomes even then, using a fresh but bounded context. + var cancel context.CancelFunc + persistCtx, cancel = context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer cancel() + expiresAt = time.Now().Add(ttl) + } if execErr != nil { backoffUntil := time.Now().Add(c.cfg.FailedRetryBackoff) reason := infraerrors.Reason(execErr) @@ -408,7 +430,7 @@ func (c *IdempotencyCoordinator) Execute( logIdempotencyAudit(opts.Route, opts.Scope, keyHash, "processing->failed_retryable", false, map[string]string{ "reason": reason, }) - if markErr := c.repo.MarkFailedRetryable(ctx, record.ID, reason, backoffUntil, expiresAt); markErr != nil { + if markErr := c.repo.MarkFailedRetryable(persistCtx, record.ID, reason, backoffUntil, expiresAt); markErr != nil { RecordIdempotencyStoreUnavailable(opts.Route, opts.Scope, "mark_failed_retryable_error") logIdempotencyAudit(opts.Route, opts.Scope, keyHash, "processing->store_unavailable", false, map[string]string{ "operation": "mark_failed_retryable", @@ -425,7 +447,7 @@ func (c *IdempotencyCoordinator) Execute( }) return nil, ErrIdempotencyStoreUnavail.WithCause(marshalErr) } - if markErr := c.repo.MarkSucceeded(ctx, record.ID, 200, storedBody, expiresAt); markErr != nil { + if markErr := c.repo.MarkSucceeded(persistCtx, record.ID, 200, storedBody, expiresAt); markErr != nil { RecordIdempotencyStoreUnavailable(opts.Route, opts.Scope, "mark_succeeded_error") logIdempotencyAudit(opts.Route, opts.Scope, keyHash, "processing->store_unavailable", false, map[string]string{ "operation": "mark_succeeded", @@ -437,6 +459,48 @@ func (c *IdempotencyCoordinator) Execute( return &IdempotencyExecuteResult{Data: data}, nil } +// renewProcessingLease uses the same repository lease primitive as system +// operations. Renewal continues through slow batches and stops before Execute exits. +func (c *IdempotencyCoordinator) renewProcessingLease(ctx context.Context, record *IdempotencyRecord, ttl time.Duration) func() { + lease := c.cfg.ProcessingTimeout + if lease <= 0 { + lease = DefaultIdempotencyConfig().ProcessingTimeout + } + interval := min(lease/3, ttl/3) + if interval < time.Millisecond { + interval = time.Millisecond + } + renewCtx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-renewCtx.Done(): + return + case <-ticker.C: + now := time.Now() + callCtx, callCancel := context.WithTimeout(renewCtx, min(2*time.Second, interval)) + ok, err := c.repo.ExtendProcessingLock(callCtx, record.ID, record.RequestFingerprint, now.Add(lease), now.Add(ttl)) + callCancel() + if err != nil { + RecordIdempotencyStoreUnavailable("", record.Scope, "renew_processing_lock_error") + continue + } + if !ok { + return + } + } + } + }() + return func() { + cancel() + <-done + } +} + func (c *IdempotencyCoordinator) conflictWithRetryAfter(base *infraerrors.ApplicationError, lockedUntil *time.Time, now time.Time) error { if lockedUntil == nil { return base diff --git a/backend/internal/service/idempotency_detached_test.go b/backend/internal/service/idempotency_detached_test.go new file mode 100644 index 0000000000..5cd890cb2f --- /dev/null +++ b/backend/internal/service/idempotency_detached_test.go @@ -0,0 +1,119 @@ +package service + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type detachedIdempotencyRepo struct { + *inMemoryIdempotencyRepo + renewals atomic.Int32 + savedWithBound bool +} + +func (r *detachedIdempotencyRepo) MarkSucceeded(ctx context.Context, id int64, status int, body string, expiresAt time.Time) error { + if err := ctx.Err(); err != nil { + return err + } + deadline, bounded := ctx.Deadline() + r.savedWithBound = bounded && time.Until(deadline) <= 5*time.Second + return r.inMemoryIdempotencyRepo.MarkSucceeded(ctx, id, status, body, expiresAt) +} + +func (r *detachedIdempotencyRepo) ExtendProcessingLock(ctx context.Context, id int64, fingerprint string, lockedUntil, expiresAt time.Time) (bool, error) { + ok, err := r.inMemoryIdempotencyRepo.ExtendProcessingLock(ctx, id, fingerprint, lockedUntil, expiresAt) + if ok && err == nil { + r.renewals.Add(1) + } + return ok, err +} + +func TestIdempotencyCoordinator_DetachedExecutionPersistsAfterCancellation(t *testing.T) { + for _, deadlineExpires := range []bool{false, true} { + name := "client cancellation" + if deadlineExpires { + name = "execution deadline" + } + t.Run(name, func(t *testing.T) { + repo := &detachedIdempotencyRepo{inMemoryIdempotencyRepo: newInMemoryIdempotencyRepo()} + coordinator := NewIdempotencyCoordinator(repo, DefaultIdempotencyConfig()) + requestCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + opts := IdempotencyExecuteOptions{ + Scope: "admin.bulk", Method: "POST", Route: "/bulk", IdempotencyKey: "cancel-test", RequireKey: true, + Payload: map[string]int{"days": 7}, ExecutionTimeout: time.Second, + } + if deadlineExpires { + opts.ExecutionTimeout = 10 * time.Millisecond + } + executions := 0 + execute := func(ctx context.Context) (any, error) { + executions++ + if deadlineExpires { + <-ctx.Done() + } else { + cancel() + require.NoError(t, ctx.Err(), "client cancellation must not interrupt the accepted mutation") + } + return map[string]int{"success_count": 1, "failed_count": 1}, nil + } + result, err := coordinator.Execute(requestCtx, opts, execute) + require.NoError(t, err) + require.NotNil(t, result.Data) + require.True(t, repo.savedWithBound, "result persistence must use a live bounded context") + replay, err := coordinator.Execute(context.Background(), opts, execute) + require.NoError(t, err) + require.True(t, replay.Replayed) + require.Equal(t, 1, executions) + }) + } +} + +func TestIdempotencyCoordinator_DetachedExecutionRenewsLongBatchLease(t *testing.T) { + repo := &detachedIdempotencyRepo{inMemoryIdempotencyRepo: newInMemoryIdempotencyRepo()} + cfg := DefaultIdempotencyConfig() + cfg.ProcessingTimeout = 30 * time.Millisecond + cfg.DefaultTTL = 50 * time.Millisecond + coordinator := NewIdempotencyCoordinator(repo, cfg) + opts := IdempotencyExecuteOptions{ + Scope: "admin.bulk", Method: "POST", Route: "/bulk", IdempotencyKey: "long-batch", RequireKey: true, + Payload: map[string]int{"days": 7}, ExecutionTimeout: 2 * time.Second, + } + var executions atomic.Int32 + release := make(chan struct{}) + t.Cleanup(func() { + select { + case <-release: + default: + close(release) + } + }) + execute := func(ctx context.Context) (any, error) { + executions.Add(1) + select { + case <-release: + case <-ctx.Done(): + } + return map[string]int{"success_count": 1}, nil + } + completed := make(chan error, 1) + go func() { + _, err := coordinator.Execute(context.Background(), opts, execute) + completed <- err + }() + // Ten renewals exceed both the original processing lease and record TTL. + require.Eventually(t, func() bool { return repo.renewals.Load() >= 10 }, time.Second, 5*time.Millisecond) + _, err := coordinator.Execute(context.Background(), opts, execute) + require.ErrorIs(t, err, ErrIdempotencyInProgress) + require.Equal(t, int32(1), executions.Load()) + close(release) + require.NoError(t, <-completed) + replay, err := coordinator.Execute(context.Background(), opts, execute) + require.NoError(t, err) + require.True(t, replay.Replayed) + require.Equal(t, int32(1), executions.Load()) +} diff --git a/backend/internal/service/subscription_bulk_action.go b/backend/internal/service/subscription_bulk_action.go new file mode 100644 index 0000000000..f8be8f6302 --- /dev/null +++ b/backend/internal/service/subscription_bulk_action.go @@ -0,0 +1,129 @@ +package service + +import ( + "context" + "errors" + "log" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/Wei-Shaw/sub2api/internal/util/logredact" +) + +const MaxBulkSubscriptionActions = 100 + +// BulkSubscriptionActionInput applies one operation to a bounded set of subscriptions. +type BulkSubscriptionActionInput struct { + SubscriptionIDs []int64 `json:"subscription_ids"` + Action string `json:"action"` + Days int `json:"days,omitempty"` + Daily bool `json:"daily,omitempty"` + Weekly bool `json:"weekly,omitempty"` + Monthly bool `json:"monthly,omitempty"` +} + +// Validate checks the entire request before any subscription is changed. +func (input *BulkSubscriptionActionInput) Validate() error { + if input == nil { + return ErrSubscriptionNilInput + } + if len(input.SubscriptionIDs) == 0 || len(input.SubscriptionIDs) > MaxBulkSubscriptionActions { + return infraerrors.BadRequest("INVALID_BULK_SUBSCRIPTIONS", "subscription_ids must contain between 1 and 100 IDs") + } + for _, id := range input.SubscriptionIDs { + if id <= 0 { + return infraerrors.BadRequest("INVALID_SUBSCRIPTION_ID", "subscription IDs must be positive") + } + } + switch input.Action { + case "extend": + if input.Days == 0 || input.Days < -MaxValidityDays || input.Days > MaxValidityDays { + return infraerrors.BadRequest("INVALID_ADJUSTMENT_DAYS", "days must be nonzero and between -36500 and 36500") + } + case "reset_quota": + if !input.Daily && !input.Weekly && !input.Monthly { + return ErrInvalidInput + } + case "revoke", "restore": + default: + return infraerrors.BadRequest("INVALID_SUBSCRIPTION_ACTION", "action must be extend, reset_quota, revoke, or restore") + } + return nil +} + +type BulkSubscriptionActionItemResult struct { + SubscriptionID int64 `json:"subscription_id"` + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +type BulkSubscriptionActionResult struct { + SuccessCount int `json:"success_count"` + FailedCount int `json:"failed_count"` + Results []BulkSubscriptionActionItemResult `json:"results"` +} + +// BulkSubscriptionAction preserves input order and executes duplicate IDs only once. +// Individual failures remain in the result instead of returning a top-level error, +// allowing callers to retain the completed portion of a partially successful batch. +func (s *SubscriptionService) BulkSubscriptionAction(ctx context.Context, input *BulkSubscriptionActionInput) (*BulkSubscriptionActionResult, error) { + if err := input.Validate(); err != nil { + return nil, err + } + result := &BulkSubscriptionActionResult{ + Results: make([]BulkSubscriptionActionItemResult, 0, len(input.SubscriptionIDs)), + } + seen := make(map[int64]struct{}, len(input.SubscriptionIDs)) + for _, id := range input.SubscriptionIDs { + if _, exists := seen[id]; exists { + continue + } + seen[id] = struct{}{} + err := ctx.Err() + if err == nil { + var changed *UserSubscription + // Single-item services may read again or update status after their first + // write. Keep all those steps in one transaction, so a failed item can + // be retried without repeating a previously committed extension/reset. + err = s.withSubscriptionUpdateTx(ctx, func(txCtx context.Context) error { + var mutationErr error + switch input.Action { + case "extend": + changed, mutationErr = s.ExtendSubscription(txCtx, id, input.Days) + case "reset_quota": + changed, mutationErr = s.AdminResetQuota(txCtx, id, input.Daily, input.Weekly, input.Monthly) + case "revoke": + changed, mutationErr = s.userSubRepo.GetByID(txCtx, id) + if mutationErr == nil { + mutationErr = s.RevokeSubscription(txCtx, id) + } + case "restore": + changed, mutationErr = s.RestoreSubscription(txCtx, id) + } + return mutationErr + }) + if err == nil && changed != nil { + // Invalidate again after commit: concurrent readers could have filled + // a cache with the old row while the transaction was still open. + if cacheErr := s.invalidateSubscriptionCaches(changed.UserID, changed.GroupID); cacheErr != nil { + log.Printf("[SubscriptionBulkAction] committed action=%s subscription_id=%d cache_error=%s", input.Action, id, logredact.RedactText(cacheErr.Error())) + } + } + } + item := BulkSubscriptionActionItemResult{SubscriptionID: id, Success: err == nil} + if err != nil { + result.FailedCount++ + item.Error = infraerrors.Message(err) + if errors.Is(err, context.Canceled) { + item.Error = context.Canceled.Error() + } else if errors.Is(err, context.DeadlineExceeded) { + item.Error = context.DeadlineExceeded.Error() + } else if infraerrors.Code(err) >= 500 { + log.Printf("[SubscriptionBulkAction] action=%s subscription_id=%d error=%s", input.Action, id, logredact.RedactText(err.Error())) + } + } else { + result.SuccessCount++ + } + result.Results = append(result.Results, item) + } + return result, nil +} diff --git a/backend/internal/service/subscription_bulk_action_test.go b/backend/internal/service/subscription_bulk_action_test.go new file mode 100644 index 0000000000..4f43e64240 --- /dev/null +++ b/backend/internal/service/subscription_bulk_action_test.go @@ -0,0 +1,226 @@ +package service + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/stretchr/testify/require" +) + +type bulkActionSubscriptionRepo struct { + UserSubscriptionRepository + subscriptions map[int64]*UserSubscription + mutations []int64 + afterMutation func() +} + +func (r *bulkActionSubscriptionRepo) GetByIDIncludeDeleted(_ context.Context, id int64) (*UserSubscription, error) { + sub := r.subscriptions[id] + if sub == nil { + return nil, ErrSubscriptionNotFound + } + copy := *sub + return ©, nil +} + +func (r *bulkActionSubscriptionRepo) GetByID(ctx context.Context, id int64) (*UserSubscription, error) { + sub, err := r.GetByIDIncludeDeleted(ctx, id) + if err != nil { + return nil, err + } + if sub.DeletedAt != nil { + return nil, ErrSubscriptionNotFound + } + return sub, nil +} + +func (r *bulkActionSubscriptionRepo) mutated(id int64) { + r.mutations = append(r.mutations, id) + if r.afterMutation != nil { + r.afterMutation() + } +} + +func (r *bulkActionSubscriptionRepo) ExtendExpiry(_ context.Context, id int64, expiresAt time.Time) error { + r.subscriptions[id].ExpiresAt = expiresAt + r.mutated(id) + return nil +} + +func (r *bulkActionSubscriptionRepo) ResetUsageWindows(_ context.Context, id int64, daily, weekly, monthly bool, dailyStart, periodicStart time.Time) error { + sub := r.subscriptions[id] + if daily { + sub.DailyUsageUSD, sub.DailyWindowStart = 0, &dailyStart + } + if weekly { + sub.WeeklyUsageUSD, sub.WeeklyWindowStart = 0, &periodicStart + } + if monthly { + sub.MonthlyUsageUSD, sub.MonthlyWindowStart = 0, &periodicStart + } + r.mutated(id) + return nil +} + +func (r *bulkActionSubscriptionRepo) Delete(_ context.Context, id int64) error { + now := time.Now() + r.subscriptions[id].DeletedAt = &now + r.mutated(id) + return nil +} + +func (r *bulkActionSubscriptionRepo) ExistsActiveByUserIDAndGroupID(_ context.Context, userID, groupID int64) (bool, error) { + for _, sub := range r.subscriptions { + if sub.UserID == userID && sub.GroupID == groupID && sub.DeletedAt == nil { + return true, nil + } + } + return false, nil +} + +func (r *bulkActionSubscriptionRepo) Restore(_ context.Context, id int64, status string) (*UserSubscription, error) { + sub := r.subscriptions[id] + sub.DeletedAt, sub.Status = nil, status + r.mutated(id) + copy := *sub + return ©, nil +} + +func TestBulkSubscriptionAction_PartialSuccessAndDeduplication(t *testing.T) { + for _, action := range []string{"extend", "reset_quota", "revoke", "restore"} { + t.Run(action, func(t *testing.T) { + expiresAt := time.Now().AddDate(0, 0, 30) + repo := &bulkActionSubscriptionRepo{subscriptions: map[int64]*UserSubscription{}} + for _, id := range []int64{1, 2} { + sub := &UserSubscription{ID: id, UserID: id, GroupID: 10, Status: SubscriptionStatusActive, ExpiresAt: expiresAt, DailyUsageUSD: 2, WeeklyUsageUSD: 5, MonthlyUsageUSD: 8} + if action == "restore" { + deletedAt := time.Now().Add(-time.Hour) + sub.DeletedAt = &deletedAt + } + repo.subscriptions[id] = sub + } + svc := NewSubscriptionService(nil, repo, nil, nil, nil) + t.Cleanup(svc.Stop) + + result, err := svc.BulkSubscriptionAction(context.Background(), &BulkSubscriptionActionInput{ + SubscriptionIDs: []int64{2, 404, 1, 2, 404}, Action: action, Days: 7, Daily: true, Weekly: true, + }) + require.NoError(t, err) + require.Equal(t, 2, result.SuccessCount) + require.Equal(t, 1, result.FailedCount) + require.Equal(t, []BulkSubscriptionActionItemResult{ + {SubscriptionID: 2, Success: true}, + {SubscriptionID: 404, Error: "subscription not found"}, + {SubscriptionID: 1, Success: true}, + }, result.Results) + require.Equal(t, []int64{2, 1}, repo.mutations) + for _, sub := range repo.subscriptions { + switch action { + case "extend": + require.Equal(t, expiresAt.AddDate(0, 0, 7), sub.ExpiresAt) + case "reset_quota": + require.Zero(t, sub.DailyUsageUSD) + require.Zero(t, sub.WeeklyUsageUSD) + require.Equal(t, float64(8), sub.MonthlyUsageUSD) + case "revoke": + require.NotNil(t, sub.DeletedAt) + case "restore": + require.Nil(t, sub.DeletedAt) + require.Equal(t, SubscriptionStatusActive, sub.Status) + } + } + }) + } +} + +func TestBulkSubscriptionAction_ValidatesBeforeAnyRepositoryAccess(t *testing.T) { + tooMany := make([]int64, MaxBulkSubscriptionActions+1) + for i := range tooMany { + tooMany[i] = 1 + } + for name, input := range map[string]*BulkSubscriptionActionInput{ + "nil": nil, + "empty IDs": {Action: "revoke"}, + "too many IDs": {SubscriptionIDs: tooMany, Action: "revoke"}, + "invalid later ID": {SubscriptionIDs: []int64{1, 0}, Action: "revoke"}, + "negative ID": {SubscriptionIDs: []int64{1, -1}, Action: "revoke"}, + "unknown action": {SubscriptionIDs: []int64{1}, Action: "delete"}, + "missing action": {SubscriptionIDs: []int64{1}}, + "zero adjustment": {SubscriptionIDs: []int64{1}, Action: "extend"}, + "large adjustment": {SubscriptionIDs: []int64{1}, Action: "extend", Days: MaxValidityDays + 1}, + "small adjustment": {SubscriptionIDs: []int64{1}, Action: "extend", Days: -MaxValidityDays - 1}, + "no reset windows": {SubscriptionIDs: []int64{1}, Action: "reset_quota"}, + } { + t.Run(name, func(t *testing.T) { + // A nil repository would panic if validation allowed any execution. + svc := &SubscriptionService{} + result, err := svc.BulkSubscriptionAction(context.Background(), input) + require.Error(t, err) + require.Equal(t, 400, infraerrors.Code(err)) + require.Nil(t, result) + }) + } + for _, days := range []int{-MaxValidityDays, -1, 1, MaxValidityDays} { + input := BulkSubscriptionActionInput{SubscriptionIDs: []int64{1}, Action: "extend", Days: days} + require.NoError(t, input.Validate()) + } +} + +func TestBulkSubscriptionAction_CancellationPreservesCompletedResults(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + repo := &bulkActionSubscriptionRepo{ + subscriptions: map[int64]*UserSubscription{1: {ID: 1, UserID: 1, GroupID: 10}}, + afterMutation: cancel, + } + svc := NewSubscriptionService(nil, repo, nil, nil, nil) + t.Cleanup(svc.Stop) + result, err := svc.BulkSubscriptionAction(ctx, &BulkSubscriptionActionInput{SubscriptionIDs: []int64{1, 2, 3}, Action: "revoke"}) + require.NoError(t, err) + require.Equal(t, 1, result.SuccessCount) + require.Equal(t, 2, result.FailedCount) + require.Equal(t, []int64{1}, repo.mutations) + require.True(t, result.Results[0].Success) + for _, item := range result.Results[1:] { + require.False(t, item.Success) + require.Equal(t, context.Canceled.Error(), item.Error) + } +} + +type failingBulkActionSubscriptionRepo struct { + *bulkActionSubscriptionRepo + err error +} + +func (r failingBulkActionSubscriptionRepo) Delete(context.Context, int64) error { + return r.err +} + +func TestBulkSubscriptionAction_DoesNotExposeInternalErrors(t *testing.T) { + for _, tc := range []struct { + name string + err error + want string + }{ + {name: "internal failure", err: errors.New("postgres: internal connection details"), want: "internal error"}, + {name: "wrapped cancellation", err: fmt.Errorf("postgres: internal connection details: %w", context.Canceled), want: context.Canceled.Error()}, + {name: "wrapped deadline", err: fmt.Errorf("postgres: internal connection details: %w", context.DeadlineExceeded), want: context.DeadlineExceeded.Error()}, + } { + t.Run(tc.name, func(t *testing.T) { + repo := failingBulkActionSubscriptionRepo{ + bulkActionSubscriptionRepo: &bulkActionSubscriptionRepo{subscriptions: map[int64]*UserSubscription{1: {ID: 1}}}, + err: tc.err, + } + svc := NewSubscriptionService(nil, repo, nil, nil, nil) + t.Cleanup(svc.Stop) + result, err := svc.BulkSubscriptionAction(context.Background(), &BulkSubscriptionActionInput{SubscriptionIDs: []int64{1}, Action: "revoke"}) + require.NoError(t, err) + require.Equal(t, 1, result.FailedCount) + require.Equal(t, tc.want, result.Results[0].Error) + }) + } +} diff --git a/backend/internal/service/subscription_bulk_action_transaction_test.go b/backend/internal/service/subscription_bulk_action_transaction_test.go new file mode 100644 index 0000000000..f497c11be4 --- /dev/null +++ b/backend/internal/service/subscription_bulk_action_transaction_test.go @@ -0,0 +1,136 @@ +package service + +import ( + "context" + "errors" + "testing" + "time" + + "entgo.io/ent/dialect" + entsql "entgo.io/ent/dialect/sql" + "github.com/DATA-DOG/go-sqlmock" + dbent "github.com/Wei-Shaw/sub2api/ent" + "github.com/stretchr/testify/require" +) + +// This repository stages mutations against the Ent transaction from context and +// makes them visible only when the real Ent/sql transaction commits. +type transactionalBulkSubscriptionRepo struct { + UserSubscriptionRepository + committed UserSubscription + pending map[*dbent.Tx]*UserSubscription + reads map[*dbent.Tx]int + postReadFailure bool + statusFailure bool +} + +func (r *transactionalBulkSubscriptionRepo) GetByID(ctx context.Context, _ int64) (*UserSubscription, error) { + tx := dbent.TxFromContext(ctx) + if tx == nil { + return nil, errors.New("bulk mutation must use a transaction") + } + if r.pending[tx] == nil { + copy := r.committed + r.pending[tx] = © + tx.OnCommit(func(next dbent.Committer) dbent.Committer { + return dbent.CommitFunc(func(ctx context.Context, tx *dbent.Tx) error { + if err := next.Commit(ctx, tx); err != nil { + return err + } + r.committed = *r.pending[tx] + return nil + }) + }) + } + r.reads[tx]++ + if r.reads[tx] > 1 && r.postReadFailure { + return nil, errors.New("refresh failed after write") + } + copy := *r.pending[tx] + return ©, nil +} + +func (r *transactionalBulkSubscriptionRepo) ExtendExpiry(ctx context.Context, _ int64, expiry time.Time) error { + r.pending[dbent.TxFromContext(ctx)].ExpiresAt = expiry + return nil +} + +func (r *transactionalBulkSubscriptionRepo) UpdateStatus(ctx context.Context, _ int64, status string) error { + if r.statusFailure { + return errors.New("status update failed after expiry write") + } + r.pending[dbent.TxFromContext(ctx)].Status = status + return nil +} + +func (r *transactionalBulkSubscriptionRepo) ResetUsageWindows(ctx context.Context, _ int64, daily, weekly, monthly bool, dailyStart, periodicStart time.Time) error { + sub := r.pending[dbent.TxFromContext(ctx)] + if daily { + sub.DailyUsageUSD, sub.DailyWindowStart = 0, &dailyStart + } + if weekly { + sub.WeeklyUsageUSD, sub.WeeklyWindowStart = 0, &periodicStart + } + if monthly { + sub.MonthlyUsageUSD, sub.MonthlyWindowStart = 0, &periodicStart + } + return nil +} + +func TestBulkSubscriptionAction_RollsBackPostWriteFailureBeforeRetry(t *testing.T) { + for _, tc := range []struct { + name string + action string + postReadFailure bool + statusFailure bool + }{ + {name: "extend refresh failure", action: "extend", postReadFailure: true}, + {name: "extend status failure", action: "extend", statusFailure: true}, + {name: "quota refresh failure", action: "reset_quota", postReadFailure: true}, + } { + t.Run(tc.name, func(t *testing.T) { + sqlDB, mock, err := sqlmock.New() + require.NoError(t, err) + client := dbent.NewClient(dbent.Driver(entsql.OpenDB(dialect.Postgres, sqlDB))) + t.Cleanup(func() { _ = client.Close() }) + expiresAt := time.Now().AddDate(0, 0, 30) + status := SubscriptionStatusActive + if tc.statusFailure { + status = SubscriptionStatusExpired + } + repo := &transactionalBulkSubscriptionRepo{ + committed: UserSubscription{ID: 1, UserID: 10, GroupID: 20, Status: status, ExpiresAt: expiresAt, DailyUsageUSD: 7}, + pending: make(map[*dbent.Tx]*UserSubscription), + reads: make(map[*dbent.Tx]int), + postReadFailure: tc.postReadFailure, + statusFailure: tc.statusFailure, + } + svc := NewSubscriptionService(nil, repo, nil, client, nil) + t.Cleanup(svc.Stop) + input := &BulkSubscriptionActionInput{SubscriptionIDs: []int64{1}, Action: tc.action, Days: 7, Daily: true} + + mock.ExpectBegin() + mock.ExpectRollback() + result, err := svc.BulkSubscriptionAction(context.Background(), input) + require.NoError(t, err) + require.Equal(t, 1, result.FailedCount) + require.Equal(t, expiresAt, repo.committed.ExpiresAt, "failed operation must not commit its earlier expiry write") + require.Equal(t, status, repo.committed.Status) + require.Equal(t, float64(7), repo.committed.DailyUsageUSD) + + repo.postReadFailure, repo.statusFailure = false, false + mock.ExpectBegin() + mock.ExpectCommit() + result, err = svc.BulkSubscriptionAction(context.Background(), input) + require.NoError(t, err) + require.Equal(t, 1, result.SuccessCount) + if tc.action == "extend" { + require.Equal(t, expiresAt.AddDate(0, 0, 7), repo.committed.ExpiresAt, "retry must extend only once") + require.Equal(t, SubscriptionStatusActive, repo.committed.Status) + } else { + require.Zero(t, repo.committed.DailyUsageUSD) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/frontend/src/api/__tests__/admin.subscriptions.bulk.spec.ts b/frontend/src/api/__tests__/admin.subscriptions.bulk.spec.ts new file mode 100644 index 0000000000..c634f20812 --- /dev/null +++ b/frontend/src/api/__tests__/admin.subscriptions.bulk.spec.ts @@ -0,0 +1,29 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { subscriptionsAPI } from '../admin/subscriptions' + +const { post } = vi.hoisted(() => ({ post: vi.fn() })) +vi.mock('../client', () => ({ apiClient: { post } })) + +describe('admin subscription batch APIs', () => { + beforeEach(() => vi.clearAllMocks()) + + it('sends the explicit operation key and returns partial results', async () => { + const request = { subscription_ids: [3, 5], action: 'extend' as const, days: 7 } + const result = { success_count: 1, failed_count: 1, results: [ + { subscription_id: 3, success: true }, { subscription_id: 5, success: false, error: 'Not found' } + ] } + post.mockResolvedValue({ data: result }) + expect(await subscriptionsAPI.bulkAction(request, 'operation-123')).toEqual(result) + expect(post).toHaveBeenCalledWith('/admin/subscriptions/bulk-action', request, { + headers: { 'Idempotency-Key': 'operation-123' } + }) + }) + + it('returns the assignment summary and per-user failures', async () => { + const request = { user_ids: [11, 12], group_id: 3, validity_days: 30 } + const result = { success_count: 1, failed_count: 1, subscriptions: [], errors: ['User 12: conflict'], statuses: { 11: 'created', 12: 'failed' } } + post.mockResolvedValue({ data: result }) + expect(await subscriptionsAPI.bulkAssign(request)).toEqual(result) + expect(post).toHaveBeenCalledWith('/admin/subscriptions/bulk-assign', request) + }) +}) diff --git a/frontend/src/api/admin/subscriptions.ts b/frontend/src/api/admin/subscriptions.ts index 893dc13ab8..f43658ab40 100644 --- a/frontend/src/api/admin/subscriptions.ts +++ b/frontend/src/api/admin/subscriptions.ts @@ -13,6 +13,45 @@ import type { PaginatedResponse } from '@/types' +export type SubscriptionBulkAction = 'extend' | 'reset_quota' | 'revoke' | 'restore' + +export interface SubscriptionBulkActionRequest { + subscription_ids: number[] + action: SubscriptionBulkAction + days?: number + daily?: boolean + weekly?: boolean + monthly?: boolean +} + +export interface SubscriptionBulkActionResult { + success_count: number + failed_count: number + results: Array<{ subscription_id: number; success: boolean; error?: string }> +} + +export interface BulkAssignSubscriptionResult { + success_count: number + created_count: number + reused_count: number + failed_count: number + subscriptions: UserSubscription[] + errors: string[] + statuses?: Record +} + +export async function bulkAction( + request: SubscriptionBulkActionRequest, + idempotencyKey: string +): Promise { + const { data } = await apiClient.post( + '/admin/subscriptions/bulk-action', + request, + { headers: { 'Idempotency-Key': idempotencyKey } } + ) + return data +} + /** * List all subscriptions with pagination * @param page - Page number (default: 1) @@ -82,12 +121,12 @@ export async function assign(request: AssignSubscriptionRequest): Promise { - const { data } = await apiClient.post( +): Promise { + const { data } = await apiClient.post( '/admin/subscriptions/bulk-assign', request ) @@ -196,6 +235,7 @@ export const subscriptionsAPI = { getProgress, assign, bulkAssign, + bulkAction, extend, revoke, restore, diff --git a/frontend/src/components/admin/subscription/BulkSubscriptionActionDialog.vue b/frontend/src/components/admin/subscription/BulkSubscriptionActionDialog.vue new file mode 100644 index 0000000000..edd6a8e136 --- /dev/null +++ b/frontend/src/components/admin/subscription/BulkSubscriptionActionDialog.vue @@ -0,0 +1,225 @@ + + + diff --git a/frontend/src/components/admin/subscription/__tests__/BulkSubscriptionActionDialog.spec.ts b/frontend/src/components/admin/subscription/__tests__/BulkSubscriptionActionDialog.spec.ts new file mode 100644 index 0000000000..77cd30d012 --- /dev/null +++ b/frontend/src/components/admin/subscription/__tests__/BulkSubscriptionActionDialog.spec.ts @@ -0,0 +1,222 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { flushPromises, mount } from '@vue/test-utils' +import type { SubscriptionBulkAction, SubscriptionBulkActionResult } from '@/api/admin/subscriptions' +import type { UserSubscription } from '@/types' +import BulkSubscriptionActionDialog from '../BulkSubscriptionActionDialog.vue' + +const bulkAction = vi.hoisted(() => vi.fn()) +vi.mock('@/api/admin', () => ({ adminAPI: { subscriptions: { bulkAction } } })) +vi.mock('vue-i18n', () => ({ + useI18n: () => ({ + t: (key: string, params?: Record) => params ? `${key} ${JSON.stringify(params)}` : key + }) +})) + +function subscription(id: number): UserSubscription { + return { + id, + user_id: id, + group_id: 8, + status: 'active', + user: { email: `user${id}@example.com` }, + group: { name: 'Subscription group' } + } as UserSubscription +} + +const successResult: SubscriptionBulkActionResult = { + success_count: 2, + failed_count: 0, + results: [ + { subscription_id: 1, success: true }, + { subscription_id: 2, success: true } + ] +} + +function mountDialog(action: SubscriptionBulkAction = 'extend', subscriptions = [subscription(1), subscription(2)]) { + const wrapper = mount(BulkSubscriptionActionDialog, { + props: { show: true, action, subscriptions }, + global: { + stubs: { + BaseDialog: { + name: 'BaseDialog', + props: ['show', 'title', 'closeOnEscape', 'showCloseButton'], + emits: ['close'], + template: '

{{ title }}

' + } + } + } + }) + wrappers.push(wrapper) + return wrapper +} + +const wrappers: ReturnType[] = [] +let adminId = 100 + +beforeEach(() => { + bulkAction.mockReset() + bulkAction.mockResolvedValue(successResult) + localStorage.setItem('auth_user', JSON.stringify({ id: ++adminId })) + sessionStorage.clear() +}) + +afterEach(() => { + wrappers.splice(0).forEach(wrapper => wrapper.unmount()) +}) + +describe('BulkSubscriptionActionDialog', () => { + it.each([30, -7, 36500, -36500])('submits whole-day adjustments of %s and shows target identities', async days => { + const wrapper = mountDialog() + expect(wrapper.text()).toContain('user1@example.com') + expect(wrapper.text()).toContain('Subscription group') + await wrapper.get('input[type="number"]').setValue(days) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(bulkAction).toHaveBeenCalledWith({ subscription_ids: [1, 2], action: 'extend', days }, expect.any(String)) + expect(wrapper.emitted('completed')).toEqual([[successResult]]) + expect(wrapper.find('button[type="submit"]').exists()).toBe(false) + }) + + it.each([0, 1.5, 36501, -36501, ''])('rejects invalid day value %s', async days => { + const wrapper = mountDialog() + await wrapper.get('input[type="number"]').setValue(days) + await wrapper.get('form').trigger('submit') + expect(wrapper.text()).toContain('admin.subscriptions.bulk.invalidDays') + expect(bulkAction).not.toHaveBeenCalled() + }) + + it('defaults to all quota windows and validates at least one selected window', async () => { + const wrapper = mountDialog('reset_quota') + for (const checkbox of wrapper.findAll('input[type="checkbox"]')) { + expect((checkbox.element as HTMLInputElement).checked).toBe(true) + await checkbox.setValue(false) + } + await wrapper.get('form').trigger('submit') + expect(bulkAction).not.toHaveBeenCalled() + expect(wrapper.text()).toContain('admin.subscriptions.bulk.selectWindow') + + await wrapper.get('input[name="weekly"]').setValue(true) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(bulkAction).toHaveBeenCalledWith({ + subscription_ids: [1, 2], action: 'reset_quota', daily: false, weekly: true, monthly: false + }, expect.any(String)) + }) + + it.each(['revoke', 'restore'] as const)('submits %s without unrelated parameters', async action => { + const wrapper = mountDialog(action) + expect(wrapper.text()).toContain(`admin.subscriptions.bulk.${action}Hint`) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(bulkAction).toHaveBeenCalledWith({ subscription_ids: [1, 2], action }, expect.any(String)) + }) + + it.each([0, 101])('rejects a selection of %s subscriptions', async count => { + const wrapper = mountDialog('revoke', Array.from({ length: count }, (_, index) => subscription(index + 1))) + await wrapper.get('form').trigger('submit') + expect(bulkAction).not.toHaveBeenCalled() + expect(wrapper.find('[role="alert"]').exists()).toBe(true) + }) + + it('prevents duplicate submits, parameter edits, and closing while submitting', async () => { + let resolve!: (value: SubscriptionBulkActionResult) => void + bulkAction.mockReturnValue(new Promise(done => { resolve = done })) + const wrapper = mountDialog() + await wrapper.get('form').trigger('submit') + await wrapper.get('form').trigger('submit') + const dialog = wrapper.findComponent({ name: 'BaseDialog' }) + dialog.vm.$emit('close') + expect(bulkAction).toHaveBeenCalledTimes(1) + expect(wrapper.get('input[type="number"]').attributes('disabled')).toBeDefined() + expect(dialog.props('closeOnEscape')).toBe(false) + expect(dialog.props('showCloseButton')).toBe(false) + expect(wrapper.emitted('close')).toBeUndefined() + resolve(successResult) + await flushPromises() + expect(dialog.props('closeOnEscape')).toBe(true) + }) + + it('shows partial failures and keeps the original target list after the parent refreshes', async () => { + const partial: SubscriptionBulkActionResult = { + success_count: 1, failed_count: 1, + results: [{ subscription_id: 1, success: true }, { subscription_id: 2, success: false, error: 'Subscription is revoked' }] + } + bulkAction.mockResolvedValue(partial) + const wrapper = mountDialog('reset_quota') + await wrapper.get('form').trigger('submit') + await flushPromises() + await wrapper.setProps({ subscriptions: [subscription(2)] }) + expect(wrapper.text()).toContain('user1@example.com') + expect(wrapper.text()).toContain('user2@example.com') + expect(wrapper.text()).toContain('Subscription is revoked') + expect(wrapper.emitted('completed')).toEqual([[partial]]) + expect(wrapper.findAll('footer button')).toHaveLength(1) + await wrapper.get('form').trigger('submit') + expect(bulkAction).toHaveBeenCalledTimes(1) + await wrapper.get('footer button').trigger('click') + expect(wrapper.emitted('close')).toHaveLength(1) + }) + + it('retries an unknown outcome with the original payload and key, even when props change', async () => { + bulkAction.mockRejectedValueOnce({ status: 0, message: 'Connection lost' }) + const wrapper = mountDialog() + await wrapper.get('input[type="number"]').setValue(-7) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(wrapper.text()).toContain('Connection lost') + expect(wrapper.text()).toContain('admin.subscriptions.bulk.retryHint') + expect(wrapper.get('input[type="number"]').attributes('disabled')).toBeDefined() + expect(wrapper.findComponent({ name: 'BaseDialog' }).props('closeOnEscape')).toBe(true) + await wrapper.setProps({ action: 'revoke', subscriptions: [subscription(3)] }) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(bulkAction.mock.calls[1]).toEqual(bulkAction.mock.calls[0]) + expect(bulkAction.mock.calls[1]![0]).toEqual({ subscription_ids: [1, 2], action: 'extend', days: -7 }) + expect(wrapper.emitted('completed')).toEqual([[successResult]]) + }) + + it('retains a pending key across closing and reopening, then releases it on completion', async () => { + bulkAction.mockRejectedValueOnce({ status: 503, message: 'Unavailable' }) + const first = mountDialog() + await first.get('form').trigger('submit') + await flushPromises() + await first.get('footer button[type="button"]').trigger('click') + expect(first.emitted('close')).toHaveLength(1) + first.unmount() + + const retry = mountDialog('extend', [subscription(2), subscription(1)]) + await retry.get('form').trigger('submit') + await flushPromises() + expect(bulkAction.mock.calls[1]).toEqual(bulkAction.mock.calls[0]) + retry.unmount() + + const next = mountDialog() + await next.get('form').trigger('submit') + await flushPromises() + expect(bulkAction.mock.calls[2]![1]).not.toBe(bulkAction.mock.calls[0]![1]) + }) + + it('uses a new key after a definitive initial validation failure and allows correcting parameters', async () => { + bulkAction.mockRejectedValueOnce({ status: 400, message: 'Invalid adjustment' }) + const wrapper = mountDialog() + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(wrapper.get('input[type="number"]').attributes('disabled')).toBeUndefined() + await wrapper.get('input[type="number"]').setValue(7) + await wrapper.get('form').trigger('submit') + await flushPromises() + expect(bulkAction.mock.calls[1]![0].days).toBe(7) + expect(bulkAction.mock.calls[1]![1]).not.toBe(bulkAction.mock.calls[0]![1]) + }) + + it('keeps the pending key when a retry is rate limited after an unknown outcome', async () => { + bulkAction.mockRejectedValueOnce({ status: 0 }).mockRejectedValueOnce({ status: 429 }) + const wrapper = mountDialog() + for (let attempt = 0; attempt < 3; attempt++) { + await wrapper.get('form').trigger('submit') + await flushPromises() + } + expect(bulkAction.mock.calls[1]).toEqual(bulkAction.mock.calls[0]) + expect(bulkAction.mock.calls[2]).toEqual(bulkAction.mock.calls[0]) + }) +}) diff --git a/frontend/src/components/admin/subscription/__tests__/bulkSubscriptionOperation.spec.ts b/frontend/src/components/admin/subscription/__tests__/bulkSubscriptionOperation.spec.ts new file mode 100644 index 0000000000..8d57ee7958 --- /dev/null +++ b/frontend/src/components/admin/subscription/__tests__/bulkSubscriptionOperation.spec.ts @@ -0,0 +1,41 @@ +import { beforeEach, describe, expect, it } from 'vitest' +import { completeBulkSubscriptionOperation, prepareBulkSubscriptionOperation } from '../bulkSubscriptionOperation' + +let adminId = 1000 + +beforeEach(() => { + localStorage.setItem('auth_user', JSON.stringify({ id: ++adminId })) + sessionStorage.clear() +}) + +describe('bulk subscription operation identity', () => { + it('canonicalizes duplicate and reordered IDs for storage and the sent request', () => { + const first = prepareBulkSubscriptionOperation({ action: 'extend', subscription_ids: [3, 1, 3], days: 7 }) + const retry = prepareBulkSubscriptionOperation({ subscription_ids: [1, 3], days: 7, action: 'extend' }) + expect(retry.request).toEqual({ action: 'extend', subscription_ids: [1, 3], days: 7 }) + expect(retry.key).toBe(first.key) + expect(retry.outcomeUncertain).toBe(true) + }) + + it('isolates different parameters, actions, targets, and administrators', () => { + const request = { action: 'extend' as const, subscription_ids: [1], days: 7 } + const first = prepareBulkSubscriptionOperation(request) + const differentDays = prepareBulkSubscriptionOperation({ ...request, days: 14 }) + const differentAction = prepareBulkSubscriptionOperation({ action: 'revoke', subscription_ids: [1] }) + const differentTargets = prepareBulkSubscriptionOperation({ ...request, subscription_ids: [2] }) + localStorage.setItem('auth_user', JSON.stringify({ id: ++adminId })) + const differentAdmin = prepareBulkSubscriptionOperation(request) + expect(new Set([first, differentDays, differentAction, differentTargets, differentAdmin].map(operation => operation.key)).size).toBe(5) + }) + + it('reuses a stored pending key and clears both storage and memory after completion', () => { + const request = { action: 'reset_quota' as const, subscription_ids: [1], daily: true, weekly: false, monthly: false } + const scope = `sub2api:admin:subscription-bulk:${adminId}:${JSON.stringify({ subscription_ids: [1], action: 'reset_quota', daily: true, weekly: false, monthly: false })}` + sessionStorage.setItem(scope, 'saved-operation-key') + const resumed = prepareBulkSubscriptionOperation(request) + expect(resumed.key).toBe('saved-operation-key') + completeBulkSubscriptionOperation(resumed) + expect(sessionStorage.getItem(scope)).toBeNull() + expect(prepareBulkSubscriptionOperation(request).key).not.toBe(resumed.key) + }) +}) diff --git a/frontend/src/components/admin/subscription/bulkSubscriptionOperation.ts b/frontend/src/components/admin/subscription/bulkSubscriptionOperation.ts new file mode 100644 index 0000000000..f7a60396fd --- /dev/null +++ b/frontend/src/components/admin/subscription/bulkSubscriptionOperation.ts @@ -0,0 +1,71 @@ +import type { SubscriptionBulkActionRequest } from '@/api/admin/subscriptions' + +export interface BulkSubscriptionOperation { + request: SubscriptionBulkActionRequest + key: string + storageKey: string | null + outcomeUncertain: boolean +} + +const pendingKeys = new Map() + +function currentAdminId(): number | null { + try { + const user = JSON.parse(globalThis.localStorage?.getItem('auth_user') ?? 'null') as { id?: unknown } | null + const id = user?.id + return typeof id === 'number' && Number.isSafeInteger(id) && id > 0 ? id : null + } catch { + return null + } +} + +function readStoredKey(storageKey: string): string | null { + try { + return globalThis.sessionStorage?.getItem(storageKey) ?? null + } catch { + return null + } +} + +function storeKey(storageKey: string, key: string | null) { + try { + if (key) globalThis.sessionStorage?.setItem(storageKey, key) + else globalThis.sessionStorage?.removeItem(storageKey) + } catch { + // Keep same-session retries safe in memory when browser storage is unavailable. + } +} + +export function prepareBulkSubscriptionOperation(input: SubscriptionBulkActionRequest): BulkSubscriptionOperation { + // Send the same canonical payload used for storage, including ID order, so a + // retry from a differently sorted table still matches the backend fingerprint. + const request: SubscriptionBulkActionRequest = { + subscription_ids: [...new Set(input.subscription_ids)].sort((a, b) => a - b), + action: input.action + } + if (input.action === 'extend') request.days = input.days + if (input.action === 'reset_quota') { + request.daily = !!input.daily + request.weekly = !!input.weekly + request.monthly = !!input.monthly + } + const adminId = currentAdminId() + const storageKey = adminId ? `sub2api:admin:subscription-bulk:${adminId}:${JSON.stringify(request)}` : null + let key = storageKey ? pendingKeys.get(storageKey) ?? readStoredKey(storageKey) : null + const outcomeUncertain = !!key + if (!key) { + const requestId = globalThis.crypto?.randomUUID?.() ?? `${Date.now()}-${Math.random().toString(36).slice(2)}` + key = `subscription-bulk-${adminId ?? 'unknown'}-${requestId}` + } + if (storageKey) { + pendingKeys.set(storageKey, key) + storeKey(storageKey, key) + } + return { request, key, storageKey, outcomeUncertain } +} + +export function completeBulkSubscriptionOperation(operation: BulkSubscriptionOperation) { + if (!operation.storageKey) return + pendingKeys.delete(operation.storageKey) + storeKey(operation.storageKey, null) +} diff --git a/frontend/src/i18n/locales/en/admin/channels.ts b/frontend/src/i18n/locales/en/admin/channels.ts index 05ef5fdb87..04826aa0f5 100644 --- a/frontend/src/i18n/locales/en/admin/channels.ts +++ b/frontend/src/i18n/locales/en/admin/channels.ts @@ -658,6 +658,41 @@ export default { // Subscriptions subscriptions: { + batchAssign: { + enable: 'Assign to multiple users', + hint: 'Search and add up to 100 users to assign the same group and validity period.', + selected: '{count} users added', + removeUser: 'Remove {email}', + result: 'Assignment complete: {success} succeeded, {failed} failed', + retryHint: 'Successful users have been removed. Resolve any errors and submit the remaining users again.' + }, + bulk: { + extend: 'Bulk Adjust Expiration', + reset_quota: 'Bulk Reset Quota', + revoke: 'Bulk Revoke', + restore: 'Bulk Restore', + selected: '{count} subscriptions selected', + selectSubscription: 'Select subscription #{id}', + clearSelection: 'Clear Selection', + selectionHint: 'Select up to 100 subscriptions on this page. Changing pages or filters clears the selection. Each action only processes subscriptions with an applicable status.', + selectionLimit: 'You can process up to 100 subscriptions at a time', + selectionRequired: 'Select at least one subscription', + confirmTargets: 'This action will process the following {count} subscriptions', + groupFallback: 'Group #{id}', + extendHint: 'Enter a positive whole number to extend or a negative one to shorten, up to 36500 days. Expired subscriptions are extended from now and cannot be shortened. The new expiration must be in the future.', + invalidDays: 'Enter a nonzero whole number of days between -36500 and 36500', + resetWindows: 'Select quota windows to reset', + resetHint: 'Usage in the selected windows will be zeroed and restarted from today.', + selectWindow: 'Select at least one quota window', + revokeHint: 'These subscriptions will no longer be usable. You can restore them later from the revoked list.', + restoreHint: 'These subscriptions will be enabled again. Subscriptions whose original validity has ended will be restored as expired.', + confirm: 'Confirm Action', + retry: 'Retry Original Action', + retryHint: 'The outcome is not yet confirmed. Retrying continues the original operation to avoid duplicate changes. You can also close this dialog, then select the same subscriptions and settings to retry.', + requestFailed: 'The bulk request failed. Please retry.', + result: 'Completed: {success} succeeded, {failed} failed', + itemFailed: 'Action failed' + }, title: 'Subscription Management', description: 'Manage user subscriptions and quota limits', assignSubscription: 'Assign Subscription', diff --git a/frontend/src/i18n/locales/zh/admin/channels.ts b/frontend/src/i18n/locales/zh/admin/channels.ts index 4bf370b424..1e2e6dc2b9 100644 --- a/frontend/src/i18n/locales/zh/admin/channels.ts +++ b/frontend/src/i18n/locales/zh/admin/channels.ts @@ -658,6 +658,41 @@ export default { // Subscriptions Management subscriptions: { + batchAssign: { + enable: '批量分配订阅', + hint: '搜索并添加多个用户，统一分配所选分组和有效期。每次最多 100 人。', + selected: '已添加 {count} 位用户', + removeUser: '移除 {email}', + result: '分配完成：成功 {success} 人，失败 {failed} 人', + retryHint: '已移除成功项，可检查失败原因后重新提交剩余用户。' + }, + bulk: { + extend: '批量调整有效期', + reset_quota: '批量重置配额', + revoke: '批量撤销', + restore: '批量恢复', + selected: '已选择 {count} 条订阅', + selectSubscription: '选择订阅 #{id}', + clearSelection: '清空选择', + selectionHint: '当前页选择，翻页或筛选会清空，最多 100 条。各操作仅处理适用状态的订阅。', + selectionLimit: '每次最多操作 100 条订阅', + selectionRequired: '请至少选择一条订阅', + confirmTargets: '本次将处理以下 {count} 条订阅', + groupFallback: '分组 #{id}', + extendHint: '正整数延长，负整数缩短，最多调整 36500 天。已过期订阅从当前时间起延长，不能缩短；缩短后的到期时间必须在未来。', + invalidDays: '请输入 -36500 到 36500 之间的非零整数天数', + resetWindows: '选择要重置的配额窗口', + resetHint: '所选窗口的用量将归零，并从今天开始重新计算。', + selectWindow: '请至少选择一个配额窗口', + revokeHint: '撤销后这些订阅将无法继续使用，可稍后在已撤销列表中恢复。', + restoreHint: '恢复后将重新启用这些订阅。原有效期已结束的订阅将显示为已过期。', + confirm: '确认执行', + retry: '重试原操作', + retryHint: '暂未确认操作结果。重试会继续原操作，避免重复处理。也可关闭后重新选择相同订阅和参数重试。', + requestFailed: '批量操作请求失败，请重试', + result: '处理完成：成功 {success} 条，失败 {failed} 条', + itemFailed: '操作失败' + }, title: '订阅管理', description: '管理用户订阅和配额限制', assignSubscription: '分配订阅', diff --git a/frontend/src/views/admin/SubscriptionsView.vue b/frontend/src/views/admin/SubscriptionsView.vue index 840ee477c8..5bbc742445 100644 --- a/frontend/src/views/admin/SubscriptionsView.vue +++ b/frontend/src/views/admin/SubscriptionsView.vue @@ -165,6 +165,32 @@ +
+
+ + {{ t('admin.subscriptions.bulk.selected', { count: selectedCount }) }} + + + +
+

{{ t('admin.subscriptions.bulk.selectionHint') }}

+
@@ -173,10 +199,15 @@ :columns="columns" :data="subscriptions" :loading="loading" + row-key="id" + selectable + :selected-keys="selectedIds" + :selection-label="getSubscriptionSelectionLabel" :server-side-sort="true" default-sort-key="created_at" default-sort-order="desc" @sort="handleSort" + @update:selected-keys="handleSelectedKeysUpdate" > + +
+ +

{{ t('admin.subscriptions.batchAssign.hint') }}

{{ user.email }} #{{ user.id }}
+
+

+ {{ t('admin.subscriptions.batchAssign.selected', { count: assignUsers.length }) }} +

+
    +
  • + {{ user.email }} #{{ user.id }} + +
    • +
+
+

{{ t('admin.subscriptions.validityHint') }}

+
+

{{ t('admin.subscriptions.batchAssign.result', { success: batchAssignResult.success_count, failed: batchAssignResult.failed_count }) }}

+
    +
  • {{ error }}
    • +
+

{{ t('admin.subscriptions.batchAssign.retryHint') }}

+