feat(cli): add event stream over the public event stream

README.md

@@ -1064,6 +1064,36 @@ that includes the current operation.
 response excerpt that the API records for a failed HTTP response. Human output
 escapes control characters in remote text.
 
+### Stream events
+
+`polymorfa event stream` follows a project's public event stream and prints
+each event as it happens. It needs a credential with `events:listen` and a team
+enrolled in the Event streams beta; otherwise it exits with the authorization
+code.
+
+```shell
+polymorfa event stream \
+  --owner project \
+  --project project_123 \
+  --type 'message.*' \
+  --json
+```
+
+`--json` writes one JSON line per event with its `cursor`, the event metadata,
+and the decoded webhook body (`null` when hosted message storage did not keep
+it). Human output prints the time, type, ID, and body, and prints a
+`--from <cursor>` hint on exit. Pass `--from` to resume within 24 hours. The
+command reconnects with backoff after a dropped connection. If events after
+`--from` expired before delivery, it reports the count and continues; use
+`--on-gap stop` to exit with the stream gap code instead. `--max-events` stops
+after a fixed number of events.
+
+Ctrl-C or SIGTERM stops the stream with exit code 0. If the output pipe is
+blocked, stopping discards queued output instead of waiting for the reader.
+
+Use `event stream` to consume events in a script. Use `listen` to forward exact
+bodies to a local webhook handler.
+
 ## Local event listening and forwarding
 
 `polymorfa listen` opens the project listener stream and can forward the exact

src/commands/event/stream.ts

@@ -0,0 +1,179 @@
+import {once} from 'node:events'
+import {Flags} from '@oclif/core'
+import type {EventStreamItem} from '@polymorfa/sdk'
+
+import {BaseCommand} from '../../base-command.js'
+import {CliConditionError} from '../../errors/conditions.js'
+
+export default class EventStream extends BaseCommand {
+  static override description =
+    "Stream a project's events as they happen, with automatic reconnect and resume. Requires events:listen and the Event streams beta."
+
+  static override examples = [
+    '<%= config.bin %> event stream --owner project --project 11111111-1111-4111-8111-111111111111 --type "message.*"',
+    '<%= config.bin %> event stream --owner project --project 11111111-1111-4111-8111-111111111111 --from "$CURSOR" --json',
+  ]
+
+  static override flags = {
+    from: Flags.string({
+      description:
+        'Resume after this cursor from a previous run; valid for 24 hours.',
+    }),
+    'max-events': Flags.integer({
+      description: 'Stop after this many events.',
+      min: 1,
+    }),
+    'on-gap': Flags.string({
+      default: 'continue',
+      description:
+        'Continue or stop when events after --from expired before delivery.',
+      options: ['continue', 'stop'],
+    }),
+    type: Flags.string({
+      description:
+        'Stream an event type or trailing-wildcard pattern such as message.*.',
+      multiple: true,
+    }),
+  }
+
+  async run(): Promise<void> {
+    const {flags} = await this.parse(EventStream)
+    const session = await this.developerSession(flags)
+    if (session.context.owner !== 'project') {
+      throw new CliConditionError(
+        'action_required',
+        'Event streams belong to a project. Pass --owner project and --project.',
+      )
+    }
+    const json = flags.json === true
+    const quiet = flags.quiet === true
+    const controller = new AbortController()
+    let interrupted = false
+    const stop = () => {
+      interrupted = true
+      controller.abort()
+    }
+    process.once('SIGINT', stop)
+    process.once('SIGTERM', stop)
+    // A consumer that closes the pipe (for example `| head -n 1`) ends the
+    // stream normally instead of failing the command.
+    const stdoutClosed = (error: unknown) => {
+      if (isClosedPipe(error)) controller.abort()
+      else throw error
+    }
+    process.stdout.on('error', stdoutClosed)
+    let gap: {missedEvents: number} | undefined
+    let delivered = 0
+    let lastCursor: string | undefined = flags.from
+    try {
+      const stream = session.client.events.stream({
+        ...(flags.type === undefined ? {} : {types: flags.type}),
+        ...(flags.from === undefined ? {} : {since: flags.from}),
+        signal: controller.signal,
+        onGap: (value) => {
+          gap = value
+          if (json) {
+            process.stdout.write(
+              `${JSON.stringify({
+                schemaVersion: 1,
+                type: 'gap',
+                reason: value.reason,
+                missedEvents: value.missedEvents,
+                requestedCursor: value.requestedCursor,
+              })}\n`,
+            )
+          } else {
+            process.stderr.write(
+              `${value.missedEvents} events after the resume cursor expired before delivery. Use "event list" to read them.\n`,
+            )
+          }
+          if (flags['on-gap'] === 'stop') controller.abort()
+        },
+        onReconnect: (_error, delayMs) => {
+          if (!json && !quiet) {
+            process.stderr.write(`Reconnecting in ${delayMs} ms...\n`)
+          }
+        },
+      })
+      for await (const item of stream) {
+        delivered += 1
+        lastCursor = item.cursor
+        if (json)
+          await writeStdout(
+            `${JSON.stringify(jsonLine(item))}\n`,
+            controller.signal,
+          )
+        else if (!quiet)
+          await writeStdout(`${humanLine(item)}\n`, controller.signal)
+        if (controller.signal.aborted) break
+        if (
+          flags['max-events'] !== undefined &&
+          delivered >= flags['max-events']
+        )
+          break
+      }
+    } finally {
+      process.removeListener('SIGINT', stop)
+      process.removeListener('SIGTERM', stop)
+      process.stdout.removeListener('error', stdoutClosed)
+      if (!json && !quiet && lastCursor !== undefined) {
+        process.stderr.write(`Resume with --from ${lastCursor}\n`)
+      }
+      // Aborting the drain wait does not cancel Node's pending stdout write.
+      // stdout.destroy() also leaves that write alive. After closing the
+      // iterator and removing listeners, discard the cancelled output so an
+      // unread pipe cannot keep an interrupted command running indefinitely.
+      if (interrupted && process.stdout.writableNeedDrain) process.exit(0)
+    }
+    if (gap !== undefined && flags['on-gap'] === 'stop') {
+      throw new CliConditionError(
+        'stream_gap',
+        `${gap.missedEvents} events expired before delivery; the stream stopped.`,
+      )
+    }
+  }
+}
+
+/**
+ * Waits for stdout to accept each line so a slow consumer pauses the stream
+ * instead of growing the process's output buffer without bound. An abort
+ * (Ctrl-C, SIGTERM or a closed stdout pipe) ends the wait.
+ */
+async function writeStdout(value: string, signal: AbortSignal): Promise<void> {
+  if (process.stdout.write(value)) return
+  try {
+    // SIGINT/SIGTERM abort the controller; stop waiting so the command exits
+    // even when the consumer never drains stdout.
+    await once(process.stdout, 'drain', {signal})
+  } catch (error) {
+    if (isClosedPipe(error)) return
+    if (signal.aborted) return
+    throw error
+  }
+}
+
+function isClosedPipe(error: unknown): boolean {
+  return (
+    error instanceof Error &&
+    'code' in error &&
+    (error.code === 'EPIPE' || error.code === 'ERR_STREAM_DESTROYED')
+  )
+}
+
+function jsonLine(item: EventStreamItem) {
+  return {
+    schemaVersion: 1,
+    type: 'event',
+    cursor: item.cursor,
+    event: {...item.event, payload: undefined},
+    webhook: item.webhook,
+  }
+}
+
+function humanLine(item: EventStreamItem): string {
+  const body =
+    item.webhook === null
+      ? `(no body: ${item.event.payloadAvailability})`
+      : JSON.stringify(item.webhook.payload)
+  return `${item.event.createdAt}\t${item.event.type}\t${item.event.id}\t${body}`
+}

test/commands/event-stream-process.test.ts

@@ -0,0 +1,50 @@
+import {fork} from 'node:child_process'
+import {once} from 'node:events'
+import {fileURLToPath} from 'node:url'
+import {expect, test} from 'vitest'
+
+test.each(['SIGINT', 'SIGTERM'] as const)(
+  '%s exits while a real stdout pipe remains unread',
+  async (signal) => {
+    const child = fork(
+      fileURLToPath(
+        new URL('../fixtures/event-stream-blocked-output.ts', import.meta.url),
+      ),
+      [],
+      {execArgv: ['--import', 'tsx'], stdio: ['ignore', 'pipe', 'pipe', 'ipc']},
+    )
+    const deadline = AbortSignal.timeout(5000)
+    const exited = once(child, 'exit')
+    let stderr = ''
+    child.stderr?.setEncoding('utf8').on('data', (value: string) => {
+      stderr += value
+    })
+    try {
+      const messages: unknown[] = await once(child, 'message', {
+        signal: deadline,
+      })
+      expect(messages[0]).toEqual({backpressured: true})
+      // Deliberately never read child.stdout, including after the signal.
+      child.kill(signal)
+      const result = await Promise.race([
+        exited,
+        new Promise<never>((_, reject) => {
+          deadline.addEventListener(
+            'abort',
+            () => reject(new Error(`Child did not exit: ${stderr}`)),
+            {once: true},
+          )
+        }),
+      ])
+      expect(result).toEqual([0, null])
+      expect(stderr).toBe('')
+    } finally {
+      if (child.exitCode === null && child.signalCode === null)
+        child.kill('SIGKILL')
+      await exited
+      child.stdout?.destroy()
+      child.stderr?.destroy()
+    }
+  },
+  10000,
+)