Merge pull request #4205 from jianjianai/codex/fix-scheduler-auto-pause-event

修复账号到期暂停触发全量调度重建

backend/internal/repository/account_repo.go

@@ -1554,7 +1554,7 @@ func (r *accountRepository) SetSchedulable(ctx context.Context, id int64, schedu
 }
 
 func (r *accountRepository) AutoPauseExpiredAccounts(ctx context.Context, now time.Time) (int64, error) {
-	result, err := r.sql.ExecContext(ctx, `
+	rows, err := r.sql.QueryContext(ctx, `
 		UPDATE accounts
 		SET schedulable = FALSE,
 			updated_at = NOW()
@@ -1563,20 +1563,35 @@ func (r *accountRepository) AutoPauseExpiredAccounts(ctx context.Context, now ti
 			AND auto_pause_on_expired = TRUE
 			AND expires_at IS NOT NULL
 			AND expires_at <= $1
+		RETURNING id
 	`, now)
 	if err != nil {
 		return 0, err
 	}
-	rows, err := result.RowsAffected()
-	if err != nil {
+	defer func() {
+		_ = rows.Close()
+	}()
+
+	accountIDs := make([]int64, 0)
+	for rows.Next() {
+		var accountID int64
+		if err := rows.Scan(&accountID); err != nil {
+			return 0, err
+		}
+		accountIDs = append(accountIDs, accountID)
+	}
+	if err := rows.Err(); err != nil {
 		return 0, err
 	}
-	if rows > 0 {
-		if err := enqueueSchedulerOutbox(ctx, r.sql, service.SchedulerOutboxEventFullRebuild, nil, nil, nil); err != nil {
-			logger.LegacyPrintf("repository.account", "[SchedulerOutbox] enqueue auto pause rebuild failed: err=%v", err)
+
+	if len(accountIDs) > 0 {
+		// 只刷新本次暂停的账号及其所属分组，避免少量账号到期触发所有调度桶重建。
+		payload := map[string]any{"account_ids": accountIDs}
+		if err := enqueueSchedulerOutbox(ctx, r.sql, service.SchedulerOutboxEventAccountBulkChanged, nil, nil, payload); err != nil {
+			logger.LegacyPrintf("repository.account", "[SchedulerOutbox] enqueue auto pause account changes failed: err=%v", err)
 		}
 	}
-	return rows, nil
+	return int64(len(accountIDs)), nil
 }
 
 func (r *accountRepository) UpdateExtra(ctx context.Context, id int64, updates map[string]any) error {

backend/internal/repository/account_repo_auto_pause_test.go

@@ -0,0 +1,72 @@
+package repository
+
+import (
+	"context"
+	"database/sql/driver"
+	"encoding/json"
+	"reflect"
+	"regexp"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/stretchr/testify/require"
+)
+
+type accountIDsPayloadMatcher struct {
+	want []int64
+}
+
+func (m accountIDsPayloadMatcher) Match(value driver.Value) bool {
+	raw, ok := value.([]byte)
+	if !ok {
+		return false
+	}
+	var payload struct {
+		AccountIDs []int64 `json:"account_ids"`
+	}
+	if err := json.Unmarshal(raw, &payload); err != nil {
+		return false
+	}
+	return reflect.DeepEqual(m.want, payload.AccountIDs)
+}
+
+func TestAutoPauseExpiredAccountsEnqueuesAffectedAccounts(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+
+	now := time.Now()
+	mock.ExpectQuery(`(?s)UPDATE accounts.*RETURNING id`).
+		WithArgs(now).
+		WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(int64(11)).AddRow(int64(29)))
+	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload)")).
+		WithArgs(service.SchedulerOutboxEventAccountBulkChanged, nil, nil, accountIDsPayloadMatcher{want: []int64{11, 29}}).
+		WillReturnResult(sqlmock.NewResult(1, 1))
+
+	repo := newAccountRepositoryWithSQL(nil, db, nil)
+	updated, err := repo.AutoPauseExpiredAccounts(context.Background(), now)
+
+	require.NoError(t, err)
+	require.EqualValues(t, 2, updated)
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestAutoPauseExpiredAccountsSkipsOutboxWithoutChanges(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+
+	now := time.Now()
+	mock.ExpectQuery(`(?s)UPDATE accounts.*RETURNING id`).
+		WithArgs(now).
+		WillReturnRows(sqlmock.NewRows([]string{"id"}))
+
+	repo := newAccountRepositoryWithSQL(nil, db, nil)
+	updated, err := repo.AutoPauseExpiredAccounts(context.Background(), now)
+
+	require.NoError(t, err)
+	require.Zero(t, updated)
+	require.NoError(t, mock.ExpectationsWereMet())
+}