diff --git a/backend/internal/repository/account_repo.go b/backend/internal/repository/account_repo.go index 6d26d474c6..8eb819aeab 100644 --- a/backend/internal/repository/account_repo.go +++ b/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 { diff --git a/backend/internal/repository/account_repo_auto_pause_test.go b/backend/internal/repository/account_repo_auto_pause_test.go new file mode 100644 index 0000000000..0eb48a296d --- /dev/null +++ b/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()) +}