Merge pull request #1492 from Aycode01/fix/1118-replay-for-vault-pagination

fix(outbox): batch and limit vault replay dispatch

Author: 1nonlypiece

PR_DESCRIPTION.md

@@ -1,26 +1,76 @@
 ## Summary
 
-This PR hardens the shared query parsing middleware used by list endpoints against unsafe query input.
+Adds bounded pagination and per-row error isolation to the `replayForVault` function, preventing a single replay request from holding an HTTP connection open indefinitely on high-activity vaults.
 
-## What changed
+Closes #1118
 
-- Added explicit validation to reject prototype-pollution style keys such as `__proto__`, `constructor`, and `prototype`.
-- Rejected malformed or unsupported query parameters before they can be parsed into filter/sort/pagination state.
-- Tightened pagination and sort parsing to fail fast on invalid numeric values or unsupported sort order values.
-- Added regression tests covering rejection of unsafe query keys and acceptance of valid filters/sort/pagination input.
+## Problem
 
-## Why
+`replayForVault` in `src/services/outboxRelay.ts` (the admin-triggered vault outbox replay path, invoked from `POST /api/admin/vaults/:id/replay-events`) queried every matching historical outbox row with **no `.limit()` clause**:
 
-The previous middleware accepted a broad set of query keys and did not guard against polluted or malformed input. That made it possible for unexpected query parameters to influence request handling and could create edge cases in downstream filtering logic.
+```typescript
+const rows = await db('vault_outbox')
+  .whereRaw("payload->'data'->>'vaultId' = ?", [vaultId])
+  .orderBy('created_at', 'asc')  // ← no .limit()
+```
 
-## Impact
+For a long-lived, high-activity vault with thousands of historical outbox rows, a single replay request could:
+1. Hold the HTTP response open for an unbounded duration
+2. Hammer the target webhook endpoint sequentially with no rate limiting or batching
+3. Abort the entire batch if any single dispatch threw (the original code had no per-row error handling)
 
-- Improves resilience of list/search endpoints against malformed or hostile query payloads.
-- Preserves existing supported behavior for valid pagination, filtering, and sorting.
-- Adds regression coverage to prevent future regressions in the query parser contract.
+This contrasts with `relayOutboxBatch` in the same file, which processes work in bounded `batchSize` chunks via `SKIP LOCKED` with proper per-row error handling and dead-letter routing.
+
+## Changes
+
+### `src/services/outboxRelay.ts`
+
+- **Added `DEFAULT_REPLAY_BATCH_SIZE`** (200) and **`MAX_REPLAY_BATCH_SIZE`** (500) constants
+- **Added `ReplayForVaultResult`** interface returning `{ count, hasMore }` instead of a bare number
+- **Refactored `replayForVault`** to accept `limit` and `offset` parameters (with sensible defaults):
+  - Uses `limit + 1` query pattern to detect whether more pages exist without a separate `COUNT` query
+  - Clamps inputs: `limit` to `[1, 500]`, `offset` to `[0, ∞)`
+  - Isolates each dispatch in a `try/catch` so one failing delivery does not abort the remaining events in the batch
+  - Logs failures via `console.error` but continues processing
+  - Returns `{ count: number, hasMore: boolean }`
+
+### `src/routes/adminWebhooks.ts`
+
+- **Imported `DEFAULT_REPLAY_BATCH_SIZE` and `MAX_REPLAY_BATCH_SIZE`** from `outboxRelay`
+- **Updated `POST /:id/replay-events`** handler:
+  - Accepts optional `limit` (default 200, max 500) and `offset` (default 0) in the request body
+  - Validates `limit` does not exceed `MAX_REPLAY_BATCH_SIZE`
+  - Passes pagination params through to `replayForVault`
+  - Returns pagination metadata (`count`, `has_more`, `limit`, `offset`) in the response
+  - Updated audit log metadata to include pagination context
+
+## Usage
+
+**Request** (with pagination):
+```json
+POST /api/admin/vaults/:id/replay-events
+{
+  "subscriber_id": "optional-subscriber-uuid",
+  "limit": 200,
+  "offset": 0
+}
+```
+
+**Response**:
+```json
+{
+  "replayed": true,
+  "count": 200,
+  "has_more": true,
+  "limit": 200,
+  "offset": 0
+}
+```
+
+Clients should iterate by incrementing `offset` by `limit` until `has_more` is `false`.
 
 ## Testing
 
-- Added targeted regression tests in `src/tests/queryParser.injection.test.ts`.
-- Verified there are no TypeScript/editor errors in the touched files.
-- Attempted to run the Jest regression suite, but local execution is currently blocked by the environment’s npm/Node setup.
+- All existing tests continue to pass (`notifications.pagination.test.ts`, `membership.pagination.test.ts`, etc.)
+- The `replayForVault` return type changed from `Promise<number>` to `Promise<ReplayForVaultResult>`; the only caller in `adminWebhooks.ts` has been updated accordingly
+- Per-row error handling prevents one bad dispatch from aborting the entire batch

src/routes/adminWebhooks.ts

@@ -19,6 +19,10 @@ import {
   getSubscriberDeliveryStats,
   parseWindowMs,
 } from '../services/webhooks.js'
+import {
+  DEFAULT_REPLAY_BATCH_SIZE,
+  MAX_REPLAY_BATCH_SIZE,
+} from '../services/outboxRelay.js'
 import { isPaused, pauseDelivery, resumeDelivery } from '../services/pauseStore.js'
 import { isValidFieldPolicy, FieldPolicy } from '../utils/webhookFieldMasking.js'
 
@@ -553,14 +557,33 @@ adminVaultReplayRouter.use(strictRateLimiter)
 adminVaultReplayRouter.post('/:id/replay-events', async (req: Request, res: Response) => {
   try {
     const vaultId = req.params.id
-    const { subscriber_id } = req.body ?? {}
+    const {
+      subscriber_id,
+      limit: rawLimit,
+      offset: rawOffset,
+    } = req.body ?? {}
 
-    if (subscriber_id && typeof subscriber_id !== 'string') {
+    if (subscriber_id !== undefined && typeof subscriber_id !== 'string') {
       res.status(400).json({ error: 'subscriber_id must be a string' })
       return
     }
 
-    const replayedCount = await replayForVault(vaultId, subscriber_id)
+    // Parse and validate pagination params
+    const limit = rawLimit !== undefined
+      ? (Number.isInteger(rawLimit) && rawLimit >= 1 ? rawLimit : DEFAULT_REPLAY_BATCH_SIZE)
+      : DEFAULT_REPLAY_BATCH_SIZE
+    const offset = rawOffset !== undefined
+      ? (Number.isInteger(rawOffset) && rawOffset >= 0 ? rawOffset : 0)
+      : 0
+
+    if (limit > MAX_REPLAY_BATCH_SIZE) {
+      res.status(400).json({
+        error: `limit must not exceed ${MAX_REPLAY_BATCH_SIZE}`,
+      })
+      return
+    }
+
+    const result = await replayForVault(vaultId, subscriber_id, limit, offset)
 
     createAuditLog({
       actor_user_id: req.user!.userId,
@@ -569,11 +592,20 @@ adminVaultReplayRouter.post('/:id/replay-events', async (req: Request, res: Resp
       target_id: vaultId,
       metadata: {
         subscriberId: subscriber_id,
-        replayedCount,
+        count: result.count,
+        hasMore: result.hasMore,
+        limit,
+        offset,
       },
     })
 
-    res.status(200).json({ replayed: true, count: replayedCount })
+    res.status(200).json({
+      replayed: true,
+      count: result.count,
+      has_more: result.hasMore,
+      limit,
+      offset,
+    })
   } catch (error) {
     console.error('Error replaying vault outbox events:', error)
     res.status(500).json({ error: 'Failed to replay vault events' })

src/services/outboxRelay.ts

@@ -90,20 +90,81 @@ export async function relayOutboxBatch(batchSize = 50): Promise<number> {
 }
 
 /**
- * Replays all recorded outbox events for a single vault to an optional target subscriber.
- * Preserves the original event ordering (by created_at or id asc) and does not modify the outbox state.
- * Returns the number of events replayed.
+ * Default maximum events to replay per call.
  */
-export async function replayForVault(vaultId: string, subscriberId?: string): Promise<number> {
+export const DEFAULT_REPLAY_BATCH_SIZE = 200
+
+/**
+ * Maximum events allowed per single replay request.
+ */
+export const MAX_REPLAY_BATCH_SIZE = 500
+
+/**
+ * Result returned by {@link replayForVault}.
+ */
+export interface ReplayForVaultResult {
+  /** Number of events successfully dispatched in this batch. */
+  count: number
+  /** Whether there are more outbox events available for this vault beyond this page. */
+  hasMore: boolean
+}
+
+/**
+ * Replays outbox events for a single vault to an optional target subscriber.
+ *
+ * This function fetches events in bounded batches (controlled by `limit`) to
+ * prevent a single replay request from holding the HTTP connection open for an
+ * unbounded duration on high-activity vaults.  Uses the same bounded,
+ * concurrency-aware dispatch path (`dispatchWebhookEvent`) as the regular
+ * outbox relay (`relayOutboxBatch`).
+ *
+ * Each event is dispatched independently so a single failing delivery does not
+ * block the remaining events in the batch.
+ *
+ * Preserves the original event ordering (by created_at asc) and does NOT
+ * modify the outbox state.
+ *
+ * @param vaultId    - The vault whose outbox events should be replayed.
+ * @param subscriberId - Optional – when provided, only replay events to this subscriber.
+ * @param limit      - Maximum number of events to fetch and dispatch (default 200, max 500).
+ * @param offset     - Number of events to skip for pagination (default 0).
+ */
+export async function replayForVault(
+  vaultId: string,
+  subscriberId?: string,
+  limit: number = DEFAULT_REPLAY_BATCH_SIZE,
+  offset: number = 0,
+): Promise<ReplayForVaultResult> {
+  // Clamp to max and ensure positive
+  const safeLimit = Math.min(Math.max(1, limit), MAX_REPLAY_BATCH_SIZE)
+  const safeOffset = Math.max(0, offset)
+
+  // Fetch one extra row to determine if there are more pages
   const rows = await db('vault_outbox')
     .whereRaw("payload->'data'->>'vaultId' = ?", [vaultId])
     .orderBy('created_at', 'asc')
+    .limit(safeLimit + 1)
+    .offset(safeOffset)
 
-  for (const row of rows) {
-    const payload = typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload
-    await dispatchWebhookEvent(payload, subscriberId)
+  const hasMore = rows.length > safeLimit
+  const batch = hasMore ? rows.slice(0, safeLimit) : rows
+
+  let dispatchedCount = 0
+
+  for (const row of batch) {
+    try {
+      const payload = typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload
+      await dispatchWebhookEvent(payload, subscriberId)
+      dispatchedCount++
+    } catch (err: any) {
+      // Log but continue — one failure should not abort the entire replay
+      console.error(
+        `[OutboxRelay] replayForVault: failed to dispatch outbox row ${row.id}:`,
+        err?.message ?? 'Unknown error',
+      )
+    }
   }
 
-  return rows.length
+  return { count: dispatchedCount, hasMore }
 }