feat: add shared drain pipeline for batching and retry

apps/playground/app/config/tests.config.ts

@@ -265,6 +265,44 @@ export const testConfig = {
         },
       ],
     } as TestSection,
+    {
+      id: 'pipeline',
+      label: 'Pipeline',
+      icon: 'i-lucide-layers',
+      title: 'Drain Pipeline (Batching + Retry)',
+      description: 'Events are buffered and sent in batches (size: 5, interval: 2s). Watch the terminal for "[evlog/pipeline] Flushing batch of N events" messages.',
+      layout: 'cards',
+      tests: [
+        {
+          id: 'pipeline-single',
+          label: '1 Request',
+          description: 'Single event - buffered until batch size (5) or interval (2s) is reached',
+          endpoint: '/api/test/success',
+          method: 'GET',
+          badge: {
+            label: 'Buffered',
+            color: 'blue',
+          },
+          toastOnSuccess: {
+            title: 'Event buffered',
+            description: 'Check terminal - will flush after 2s or when 5 events accumulate',
+          },
+        },
+        {
+          id: 'pipeline-batch',
+          label: 'Fire 10 Requests',
+          description: 'Fires 10 requests in parallel - should produce 2 batches of 5 events',
+          badge: {
+            label: '2 batches',
+            color: 'green',
+          },
+          toastOnSuccess: {
+            title: '10 requests sent',
+            description: 'Check terminal - should see 2 batches of 5 events',
+          },
+        },
+      ],
+    } as TestSection,
     {
       id: 'drains',
       label: 'Drains',

apps/playground/app/pages/index.vue

@@ -39,6 +39,12 @@ async function handleBatchRequest() {
   )
 }
 
+async function handlePipelineBatch() {
+  await Promise.all(
+    Array.from({ length: 10 }, () => $fetch('/api/test/success')),
+  )
+}
+
 // Get custom onClick for specific tests
 function getOnClick(testId: string) {
   if (testId === 'structured-error-toast') {
@@ -47,6 +53,9 @@ function getOnClick(testId: string) {
   if (testId === 'tail-fast-batch') {
     return handleBatchRequest
   }
+  if (testId === 'pipeline-batch') {
+    return handlePipelineBatch
+  }
   return undefined
 }
 </script>

packages/evlog/package.json

@@ -58,6 +58,10 @@
     "./enrichers": {
       "types": "./dist/enrichers.d.mts",
       "import": "./dist/enrichers.mjs"
+    },
+    "./pipeline": {
+      "types": "./dist/pipeline.d.mts",
+      "import": "./dist/pipeline.mjs"
     }
   },
   "main": "./dist/index.mjs",
@@ -90,6 +94,9 @@
       ],
       "enrichers": [
         "./dist/enrichers.d.mts"
+      ],
+      "pipeline": [
+        "./dist/pipeline.d.mts"
       ]
     }
   },

packages/evlog/src/adapters/axiom.ts

@@ -34,8 +34,11 @@ export interface AxiomConfig {
  * }))
  * ```
  */
-export function createAxiomDrain(overrides?: Partial<AxiomConfig>): (ctx: DrainContext) => Promise<void> {
-  return async (ctx: DrainContext) => {
+export function createAxiomDrain(overrides?: Partial<AxiomConfig>): (ctx: DrainContext | DrainContext[]) => Promise<void> {
+  return async (ctx: DrainContext | DrainContext[]) => {
+    const contexts = Array.isArray(ctx) ? ctx : [ctx]
+    if (contexts.length === 0) return
+
     const runtimeConfig = getRuntimeConfig()
     // Support both runtimeConfig.evlog.axiom and runtimeConfig.axiom
     const evlogAxiom = runtimeConfig?.evlog?.axiom
@@ -55,11 +58,7 @@ export function createAxiomDrain(overrides?: Partial<AxiomConfig>): (ctx: DrainC
       return
     }
 
-    try {
-      await sendToAxiom(ctx.event, config as AxiomConfig)
-    } catch (error) {
-      console.error('[evlog/axiom] Failed to send event:', error)
-    }
+    await sendBatchToAxiom(contexts.map(c => c.event), config as AxiomConfig)
   }
 }
 

packages/evlog/src/adapters/posthog.ts

@@ -61,8 +61,11 @@ export function toPostHogEvent(event: WideEvent, config: PostHogConfig): PostHog
  * }))
  * ```
  */
-export function createPostHogDrain(overrides?: Partial<PostHogConfig>): (ctx: DrainContext) => Promise<void> {
-  return async (ctx: DrainContext) => {
+export function createPostHogDrain(overrides?: Partial<PostHogConfig>): (ctx: DrainContext | DrainContext[]) => Promise<void> {
+  return async (ctx: DrainContext | DrainContext[]) => {
+    const contexts = Array.isArray(ctx) ? ctx : [ctx]
+    if (contexts.length === 0) return
+
     const runtimeConfig = getRuntimeConfig()
     // Support both runtimeConfig.evlog.posthog and runtimeConfig.posthog
     const evlogPostHog = runtimeConfig?.evlog?.posthog
@@ -82,11 +85,7 @@ export function createPostHogDrain(overrides?: Partial<PostHogConfig>): (ctx: Dr
       return
     }
 
-    try {
-      await sendToPostHog(ctx.event, config as PostHogConfig)
-    } catch (error) {
-      console.error('[evlog/posthog] Failed to send event:', error)
-    }
+    await sendBatchToPostHog(contexts.map(c => c.event), config as PostHogConfig)
   }
 }
 

packages/evlog/src/adapters/sentry.ts

@@ -214,8 +214,11 @@ function buildEnvelopeBody(logs: SentryLog[], dsn: string): string {
  * }))
  * ```
  */
-export function createSentryDrain(overrides?: Partial<SentryConfig>): (ctx: DrainContext) => Promise<void> {
-  return async (ctx: DrainContext) => {
+export function createSentryDrain(overrides?: Partial<SentryConfig>): (ctx: DrainContext | DrainContext[]) => Promise<void> {
+  return async (ctx: DrainContext | DrainContext[]) => {
+    const contexts = Array.isArray(ctx) ? ctx : [ctx]
+    if (contexts.length === 0) return
+
     const runtimeConfig = getRuntimeConfig()
     const evlogSentry = runtimeConfig?.evlog?.sentry
     const rootSentry = runtimeConfig?.sentry
@@ -233,11 +236,7 @@ export function createSentryDrain(overrides?: Partial<SentryConfig>): (ctx: Drai
       return
     }
 
-    try {
-      await sendToSentry(ctx.event, config as SentryConfig)
-    } catch (error) {
-      console.error('[evlog/sentry] Failed to send log:', error)
-    }
+    await sendBatchToSentry(contexts.map(c => c.event), config as SentryConfig)
   }
 }
 

packages/evlog/src/pipeline.ts

@@ -0,0 +1,144 @@
+export interface DrainPipelineOptions<T = unknown> {
+  batch?: {
+    /** Default: 50 */
+    size?: number
+    /** Default: 5000 */
+    intervalMs?: number
+  }
+  retry?: {
+    /** Including the first attempt. Default: 3 */
+    maxAttempts?: number
+    /** Default: 'exponential' */
+    backoff?: 'exponential' | 'linear' | 'fixed'
+    /** Default: 1000 */
+    initialDelayMs?: number
+    /** Default: 30000 */
+    maxDelayMs?: number
+  }
+  /** Default: 1000 */
+  maxBufferSize?: number
+  onDropped?: (events: T[], error?: Error) => void
+}
+
+export interface PipelineDrainFn<T> {
+  (ctx: T): void
+  /** Flush all buffered events. Call on server shutdown. */
+  flush: () => Promise<void>
+  readonly pending: number
+}
+
+export function createDrainPipeline<T = unknown>(options?: DrainPipelineOptions<T>): (drain: (ctx: T | T[]) => void | Promise<void>) => PipelineDrainFn<T> {
+  const batchSize = options?.batch?.size ?? 50
+  const intervalMs = options?.batch?.intervalMs ?? 5000
+  const maxBufferSize = options?.maxBufferSize ?? 1000
+  const maxAttempts = options?.retry?.maxAttempts ?? 3
+  const backoffStrategy = options?.retry?.backoff ?? 'exponential'
+  const initialDelayMs = options?.retry?.initialDelayMs ?? 1000
+  const maxDelayMs = options?.retry?.maxDelayMs ?? 30000
+  const onDropped = options?.onDropped
+
+  return (drain: (ctx: T | T[]) => void | Promise<void>): PipelineDrainFn<T> => {
+    const buffer: T[] = []
+    let timer: ReturnType<typeof setTimeout> | null = null
+    let activeFlush: Promise<void> | null = null
+
+    function clearTimer(): void {
+      if (timer !== null) {
+        clearTimeout(timer)
+        timer = null
+      }
+    }
+
+    function scheduleFlush(): void {
+      if (timer !== null || activeFlush) return
+      timer = setTimeout(() => {
+        timer = null
+        if (!activeFlush) startFlush()
+      }, intervalMs)
+    }
+
+    function getRetryDelay(attempt: number): number {
+      let delay: number
+      switch (backoffStrategy) {
+        case 'linear':
+          delay = initialDelayMs * attempt
+          break
+        case 'fixed':
+          delay = initialDelayMs
+          break
+        case 'exponential':
+        default:
+          delay = initialDelayMs * 2 ** (attempt - 1)
+          break
+      }
+      return Math.min(delay, maxDelayMs)
+    }
+
+    async function sendWithRetry(batch: T[]): Promise<void> {
+      let lastError: Error | undefined
+      for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+        try {
+          await drain(batch)
+          return
+        } catch (error) {
+          lastError = error instanceof Error ? error : new Error(String(error))
+          if (attempt < maxAttempts) {
+            await new Promise<void>(r => setTimeout(r, getRetryDelay(attempt)))
+          }
+        }
+      }
+      onDropped?.(batch, lastError)
+    }
+
+    async function drainBuffer(): Promise<void> {
+      while (buffer.length > 0) {
+        const batch = buffer.splice(0, batchSize)
+        await sendWithRetry(batch)
+      }
+    }
+
+    function startFlush(): void {
+      if (activeFlush) return
+      activeFlush = drainBuffer().finally(() => {
+        activeFlush = null
+        if (buffer.length >= batchSize) {
+          startFlush()
+        } else if (buffer.length > 0) {
+          scheduleFlush()
+        }
+      })
+    }
+
+    function push(ctx: T): void {
+      if (buffer.length >= maxBufferSize) {
+        const dropped = buffer.splice(0, 1)
+        onDropped?.(dropped)
+      }
+      buffer.push(ctx)
+
+      if (buffer.length >= batchSize) {
+        clearTimer()
+        startFlush()
+      } else if (!activeFlush) {
+        scheduleFlush()
+      }
+    }
+
+    async function flush(): Promise<void> {
+      clearTimer()
+      if (activeFlush) {
+        await activeFlush
+      }
+      await drainBuffer()
+    }
+
+    const hookFn = push as PipelineDrainFn<T>
+    hookFn.flush = flush
+    Object.defineProperty(hookFn, 'pending', {
+      get: () => buffer.length,
+      enumerable: true,
+    })
+
+    return hookFn
+  }
+}