Merge pull request #7091 from SavitarC/feat/subscription-bulk-actions

feat: 支持订阅管理批量操作

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"

backend/internal/handler/admin/subscription_bulk_action_test.go

@@ -0,0 +1,212 @@
+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 &copy, nil
+}
+
+func (r *bulkActionHandlerSubscriptionRepo) GetByIDForUpdate(ctx context.Context, id int64) (*service.UserSubscription, error) {
+	return r.GetByID(ctx, id)
+}
+
+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())
+		})
+	}
+}

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) {

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)

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)
+			})
+		}
+	}
+}

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

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())
+}

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
+}

backend/internal/service/subscription_bulk_action_test.go

@@ -0,0 +1,230 @@
+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 &copy, 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) GetByIDForUpdate(ctx context.Context, id int64) (*UserSubscription, error) {
+	return r.GetByID(ctx, id)
+}
+
+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 &copy, 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)
+		})
+	}
+}

backend/internal/service/subscription_bulk_action_transaction_test.go

@@ -0,0 +1,140 @@
+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] = &copy
+		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 &copy, nil
+}
+
+func (r *transactionalBulkSubscriptionRepo) GetByIDForUpdate(ctx context.Context, id int64) (*UserSubscription, error) {
+	return r.GetByID(ctx, id)
+}
+
+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())
+		})
+	}
+}

backend/internal/service/subscription_renewal_lock_test.go

@@ -115,6 +115,22 @@ func TestAssignOrExtendSubscriptionSerializedRenewalsAccumulateDays(t *testing.T
 	require.Equal(t, initialExpiry.AddDate(0, 0, 14), second.ExpiresAt)
 }
 
+func TestExtendSubscriptionUsesLockedCurrentRow(t *testing.T) {
+	now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC)
+	initialExpiry := now.AddDate(0, 0, 10)
+	repo := &lockingRenewalRepo{current: UserSubscription{
+		ID: 37, UserID: 41, GroupID: 43, ExpiresAt: initialExpiry, Status: SubscriptionStatusActive,
+	}}
+	svc := NewSubscriptionService(nil, repo, nil, nil, nil)
+	svc.now = func() time.Time { return now }
+
+	updated, err := svc.ExtendSubscription(context.Background(), 7, 5)
+
+	require.NoError(t, err)
+	require.Equal(t, 1, repo.lockReads)
+	require.Equal(t, initialExpiry.AddDate(0, 0, 5), updated.ExpiresAt)
+}
+
 func TestAssignSubscriptionDoesNotReactivateRowSuspendedAfterStaleRead(t *testing.T) {
 	now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC)
 	windowStart := now.Add(-24 * time.Hour)

backend/internal/service/subscription_service.go

@@ -652,55 +652,72 @@ func (s *SubscriptionService) RestoreSubscription(ctx context.Context, subscript
 
 // ExtendSubscription 调整订阅时长（正数延长，负数缩短）
 func (s *SubscriptionService) ExtendSubscription(ctx context.Context, subscriptionID int64, days int) (*UserSubscription, error) {
-	sub, err := s.userSubRepo.GetByID(ctx, subscriptionID)
+	err := s.withSubscriptionUpdateTx(ctx, func(txCtx context.Context) error {
+		// Lock the row before reading its expiry. Without this lock, concurrent
+		// adjustments can both calculate from the same stale expiry and lose one
+		// of the updates when they write the absolute timestamp.
+		sub, err := s.userSubRepo.GetByIDForUpdate(txCtx, subscriptionID)
+		if err != nil {
+			return ErrSubscriptionNotFound
+		}
+
+		// 限制调整天数范围
+		if days > MaxValidityDays {
+			days = MaxValidityDays
+		}
+		if days < -MaxValidityDays {
+			days = -MaxValidityDays
+		}
+
+		now := time.Now()
+		if s.now != nil {
+			now = s.now()
+		}
+		isExpired := !sub.ExpiresAt.After(now)
+
+		// 如果订阅已过期，不允许负向调整
+		if isExpired && days < 0 {
+			return infraerrors.BadRequest("CANNOT_SHORTEN_EXPIRED", "cannot shorten an expired subscription")
+		}
+
+		// 计算新的过期时间
+		var newExpiresAt time.Time
+		if isExpired {
+			// 已过期：从当前时间开始增加天数
+			newExpiresAt = now.AddDate(0, 0, days)
+		} else {
+			// 未过期：从原过期时间增加/减少天数
+			newExpiresAt = sub.ExpiresAt.AddDate(0, 0, days)
+		}
+
+		if newExpiresAt.After(MaxExpiresAt) {
+			newExpiresAt = MaxExpiresAt
+		}
+
+		// 检查新的过期时间必须大于当前时间
+		if !newExpiresAt.After(now) {
+			return ErrAdjustWouldExpire
+		}
+
+		if err := s.userSubRepo.ExtendExpiry(txCtx, subscriptionID, newExpiresAt); err != nil {
+			return err
+		}
+
+		// 如果订阅已过期，恢复为active状态
+		if sub.Status == SubscriptionStatusExpired {
+			if err := s.userSubRepo.UpdateStatus(txCtx, subscriptionID, SubscriptionStatusActive); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
 	if err != nil {
-		return nil, ErrSubscriptionNotFound
-	}
-
-	// 限制调整天数范围
-	if days > MaxValidityDays {
-		days = MaxValidityDays
-	}
-	if days < -MaxValidityDays {
-		days = -MaxValidityDays
-	}
-
-	now := time.Now()
-	isExpired := !sub.ExpiresAt.After(now)
-
-	// 如果订阅已过期，不允许负向调整
-	if isExpired && days < 0 {
-		return nil, infraerrors.BadRequest("CANNOT_SHORTEN_EXPIRED", "cannot shorten an expired subscription")
-	}
-
-	// 计算新的过期时间
-	var newExpiresAt time.Time
-	if isExpired {
-		// 已过期：从当前时间开始增加天数
-		newExpiresAt = now.AddDate(0, 0, days)
-	} else {
-		// 未过期：从原过期时间增加/减少天数
-		newExpiresAt = sub.ExpiresAt.AddDate(0, 0, days)
-	}
-
-	if newExpiresAt.After(MaxExpiresAt) {
-		newExpiresAt = MaxExpiresAt
-	}
-
-	// 检查新的过期时间必须大于当前时间
-	if !newExpiresAt.After(now) {
-		return nil, ErrAdjustWouldExpire
-	}
-
-	if err := s.userSubRepo.ExtendExpiry(ctx, subscriptionID, newExpiresAt); err != nil {
 		return nil, err
 	}
 
-	// 如果订阅已过期，恢复为active状态
-	if sub.Status == SubscriptionStatusExpired {
-		if err := s.userSubRepo.UpdateStatus(ctx, subscriptionID, SubscriptionStatusActive); err != nil {
-			return nil, err
-		}
+	sub, err := s.userSubRepo.GetByID(ctx, subscriptionID)
+	if err != nil {
+		return nil, err
 	}
 
 	// 失效订阅缓存
@@ -714,7 +731,7 @@ func (s *SubscriptionService) ExtendSubscription(ctx context.Context, subscripti
 		}()
 	}
 
-	return s.userSubRepo.GetByID(ctx, subscriptionID)
+	return sub, nil
 }
 
 // GetByID 根据ID获取订阅