feat: per-vault outbox event replay endpoint

docs/webhooks.md

@@ -124,6 +124,25 @@ Response (404):
 { "error": "Dead letter not found or already replayed" }
 ```
 
+#### POST `/api/admin/vaults/:id/replay-events`
+
+Replays all recorded outbox lifecycle events for a single vault to an optional target subscriber. Preserves original event ordering and event IDs/idempotency keys.
+
+Request Body (Optional):
+```json
+{
+  "subscriber_id": "90b1e428-2f19-4b2b-8a71-3cb56667104b"
+}
+```
+
+Response (200):
+```json
+{
+  "replayed": true,
+  "count": 3
+}
+```
+
 ## Test-Ping Endpoint
 
 `POST /api/webhooks/:id/test` lets subscribers self-verify their delivery URL and HMAC wiring before real vault events start flowing.

src/app-bootstrap.ts

@@ -19,7 +19,7 @@ import { orgAnalyticsRouter } from './routes/orgAnalytics.js'
 import { orgMembersRouter } from './routes/orgMembers.js'
 import { adminRouter } from './routes/admin.js'
 import { adminVerifiersRouter } from './routes/adminVerifiers.js'
-import { adminWebhooksRouter } from './routes/adminWebhooks.js'
+import { adminWebhooksRouter, adminVaultReplayRouter } from './routes/adminWebhooks.js'
 import { verificationsRouter } from './routes/verifications.js'
 import { apiKeysRouter } from './routes/apiKeys.js'
 import { notificationsRouter } from './routes/notifications.js'
@@ -74,6 +74,7 @@ export function bootstrapApp(options: BootstrapOptions = {}) {
   app.use('/api/admin', adminRouter)
   app.use('/api/admin/verifiers', adminVerifiersRouter)
   app.use('/api/admin/webhooks', adminWebhooksRouter)
+  app.use('/api/admin/vaults', adminVaultReplayRouter)
   app.use('/api/verifications', verificationsRouter)
   app.use('/api/api-keys', apiKeysRouter)
   app.use('/api/notifications', notificationsRouter)

src/routes/adminWebhooks.ts

@@ -4,6 +4,8 @@
 import { UserRole } from '../types/user.js'
 import { createAuditLog } from '../lib/audit-logs.js'
 import { db } from '../db/knex.js'
+import { replayForVault } from '../services/outboxRelay.js'
+import { strictRateLimiter } from '../middleware/rateLimiter.js'
 import {
   replayDeadLetter,
   upsertSubscriber,
@@ -421,3 +423,40 @@
     }
   },
 )
+
+export const adminVaultReplayRouter = Router()
+
+adminVaultReplayRouter.use(authenticate)
+adminVaultReplayRouter.use(requireAdmin)
+adminVaultReplayRouter.use(strictRateLimiter)
+
+adminVaultReplayRouter.post('/:id/replay-events', async (req: Request, res: Response) => {
+  try {
+    const vaultId = req.params.id
+    const { subscriber_id } = req.body ?? {}
+
+    if (subscriber_id && typeof subscriber_id !== 'string') {
+      res.status(400).json({ error: 'subscriber_id must be a string' })
+      return
+    }
+
+    const replayedCount = await replayForVault(vaultId, subscriber_id)
+
+    createAuditLog({
+      actor_user_id: req.user!.userId,
+      action: 'vault.outbox.replay',
+      target_type: 'vault',
+      target_id: vaultId,
+      metadata: {
+        subscriberId: subscriber_id,
+        replayedCount,
+      },
+    })
+
+    res.status(200).json({ replayed: true, count: replayedCount })
+  } catch (error) {
+    console.error('Error replaying vault outbox events:', error)
+    res.status(500).json({ error: 'Failed to replay vault events' })
+  }
+})
+

src/services/outboxRelay.ts

@@ -88,3 +88,22 @@ export async function relayOutboxBatch(batchSize = 50): Promise<number> {
     return rows.length
   })
 }
+
+/**
+ * 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.
+ */
+export async function replayForVault(vaultId: string, subscriberId?: string): Promise<number> {
+  const rows = await db('vault_outbox')
+    .whereRaw("payload->'data'->>'vaultId' = ?", [vaultId])
+    .orderBy('created_at', 'asc')
+
+  for (const row of rows) {
+    const payload = typeof row.payload === 'string' ? JSON.parse(row.payload) : row.payload
+    await dispatchWebhookEvent(payload, subscriberId)
+  }
+
+  return rows.length
+}
+

src/services/webhooks.ts

@@ -713,8 +713,20 @@ const deliverOnce = async (
  */
 export const dispatchWebhookEvent = async (
   payload: WebhookDeliveryPayload,
+  targetSubscriberId?: string,
 ): Promise<WebhookDeliveryResult[]> => {
-  const eligible = await repo.findByEvent(payload.organizationId, payload.eventType)
+  let eligible: WebhookSubscriber[]
+  if (targetSubscriberId) {
+    const sub = await repo.findById(targetSubscriberId)
+    const matchesEvent = sub && (sub.events.length === 0 || sub.events.includes(payload.eventType))
+    if (!sub || sub.organizationId !== payload.organizationId || !sub.active || !matchesEvent) {
+      eligible = []
+    } else {
+      eligible = [sub]
+    }
+  } else {
+    eligible = await repo.findByEvent(payload.organizationId, payload.eventType)
+  }
   const config = getCircuitBreakerConfig()
 
   // Load org allowlist once for the whole dispatch batch (defense in depth)