feat(sse): add @apimatic/sse package for Server-Sent Events

README.md

@@ -28,6 +28,7 @@ They provide common runtime utilities needed by SDKs to make API calls and handl
 | [@apimatic/pagination](packages/pagination)                           | [![npm shield](https://img.shields.io/npm/v/@apimatic/pagination)](https://www.npmjs.com/package/@apimatic/pagination)                           | Provides utilities to handle paginated API responses, including support for asynchronous iteration over pages or items. |
 | [@apimatic/proxy](packages/proxy)                                     | [![npm shield](https://img.shields.io/npm/v/@apimatic/proxy)](https://www.npmjs.com/package/@apimatic/proxy)                                     | Provides proxy configuration utilities for HTTP clients.                                                                |
 | [@apimatic/hmac-signature-verifier](packages/hmac-signature-verifier) | [![npm shield](https://img.shields.io/npm/v/@apimatic/hmac-signature-verifier)](https://www.npmjs.com/package/@apimatic/hmac-signature-verifier) | Provides HMAC signature verification utilities to secure HTTP requests.                                                        |
+| [@apimatic/sse](packages/sse)                                         | [![npm shield](https://img.shields.io/npm/v/@apimatic/sse)](https://www.npmjs.com/package/@apimatic/sse)                                         | Provides parsing and typed consumption for Server-Sent Events (SSE) API responses.                                      |
 
 ## Builds and Usage
 

packages/sse/README.md

@@ -0,0 +1,48 @@
+# @apimatic/sse
+
+Server-Sent Events (SSE) utilities for APIMatic-generated TypeScript SDKs:
+
+- **Spec-compliant incremental parser** (`SseParser`) — chunk-boundary
+  buffering (including split multi-byte UTF-8 and split CRLF), multi-line
+  `data:` folding, comment/heartbeat skipping, BOM handling, and the
+  WHATWG `id:`/`retry:` field rules.
+- **`SseStream<T>`** — a single-use async-iterable stream with
+  `withMetadata()`, abort-on-break cancellation, and single-use enforcement.
+- **Read timeout** — bounds the wait for the server between frames (never
+  consumer processing time); a stall surfaces as a typed `SseTimeoutError`.
+- **Typed errors** — `SseError` base, `SseTimeoutError` (idle stall),
+  `SseDecodeError` (a frame that couldn't be decoded). A mid-stream transport
+  drop surfaces the raw underlying error to the consumer.
+- **`createEventStream`** — the factory that core's
+  `RequestBuilder.callAsEventStream(...)` hook delegates to (mirroring how
+  `paginate()` delegates to `@apimatic/pagination`'s `createPagedData`):
+  executes the request as an SSE stream and buffers non-2xx error bodies back
+  to text.
+
+## Usage
+
+Generated SDK code (via `@apimatic/core`'s request builder):
+
+```ts
+import { createEventStream } from '@apimatic/sse';
+
+const req = this.createRequest('GET', '/paragraphs');
+// core's callAsEventStream hands the factory the request builder plus the
+// client-wide stream read timeout (ms).
+const { result: stream } = await req.callAsEventStream((req, streamReadTimeout) =>
+  createEventStream(
+    req,
+    requestOptions,
+    streamReadTimeout,
+    (raw) => ({ done: false, value: raw.data }) // a decoder, e.g. schemaSseDecoder(schema)
+  )
+);
+
+for await (const event of stream) {
+  console.log(event);
+}
+```
+
+The client-wide stream read timeout is a plain `number` (milliseconds),
+configured on the SDK like `timeout` — a `0` or unset value means no idle
+timeout.

packages/sse/jest.config.js

@@ -0,0 +1,9 @@
+const { jest: lernaAliases } = require('lerna-alias');
+
+module.exports = {
+  preset: 'ts-jest',
+  testEnvironment: 'node',
+  setupFiles: ['<rootDir>/test/setup.js'],
+  moduleNameMapper: lernaAliases(),
+  coverageReporters: [['lcov', { projectRoot: '../../' }]],
+};

packages/sse/package.json

@@ -0,0 +1,61 @@
+{
+  "name": "@apimatic/sse",
+  "version": "0.0.0",
+  "description": "Provides parsing and typed consumption for Server-Sent Events (SSE) API responses.",
+  "main": "lib/index.js",
+  "module": "es/index.js",
+  "types": "lib/index.d.ts",
+  "sideEffects": false,
+  "files": [
+    "umd",
+    "lib/**/*.js",
+    "lib/**/*.d.ts",
+    "es/**/*.js",
+    "es/**/*.d.ts",
+    "src",
+    "LICENSE.md"
+  ],
+  "engines": {
+    "node": ">=14.15.0 || >=16.0.0"
+  },
+  "scripts": {
+    "clean": "rimraf lib es umd tsconfig.tsbuildinfo",
+    "test": "jest --ci --coverage --maxWorkers=2 --passWithNoTests",
+    "build": "npm run clean && tsc && rollup -c --bundleConfigAsCjs && npm run annotate:es",
+    "annotate:es": "babel es --out-dir es --no-babelrc --plugins annotate-pure-calls",
+    "preversion": "npm run test",
+    "prepublishOnly": "npm run build",
+    "size": "size-limit",
+    "analyze": "size-limit --why",
+    "lint": "tslint --project .",
+    "lint:fix": "tslint --project . --fix",
+    "check-style": "prettier --check \"{src,test}/**/*.ts\"",
+    "check-style:fix": "prettier --write \"{src,test}/**/*.ts\""
+  },
+  "author": "APIMatic Ltd.",
+  "license": "MIT",
+  "size-limit": [
+    {
+      "path": "es/index.js",
+      "limit": "8 KB"
+    }
+  ],
+  "dependencies": {
+    "@apimatic/core-interfaces": "^0.2.14",
+    "@apimatic/json-bigint": "^1.2.0",
+    "@apimatic/schema": "^0.7.21",
+    "abort-controller": "^3.0.0",
+    "tslib": "2.8.1"
+  },
+  "devDependencies": {
+    "@types/node": "^17.0.29"
+  },
+  "publishConfig": {
+    "access": "public"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git@github.com:apimatic/apimatic-js-runtime.git",
+    "directory": "packages/sse"
+  }
+}

packages/sse/rollup.config.js

@@ -0,0 +1,34 @@
+import typescript from 'rollup-plugin-typescript2';
+const packageJson = require('./package.json');
+
+const getTsPlugin = ({ declaration = true, target } = {}) =>
+  typescript({
+    clean: true,
+    tsconfigOverride: {
+      compilerOptions: {
+        declaration,
+        ...(target && { target }),
+      },
+    },
+  });
+
+const getNpmConfig = ({ input, output, external }) => ({
+  input,
+  output,
+  preserveModules: true,
+  plugins: [getTsPlugin({ declaration: true })],
+  external,
+});
+
+export default [
+  getNpmConfig({
+    input: 'src/index.ts',
+    output: [
+      {
+        dir: 'es',
+        format: 'esm',
+      },
+    ],
+    external: Object.keys(packageJson.dependencies),
+  }),
+];

packages/sse/src/coreInterfaces.ts

@@ -0,0 +1,6 @@
+export {
+  ApiResponse,
+  HttpContext,
+  HttpInterceptorInterface,
+  RequestOptions,
+} from '@apimatic/core-interfaces';

packages/sse/src/createEventStream.ts

@@ -0,0 +1,150 @@
+import { AbortController } from 'abort-controller';
+import JSONBig from '@apimatic/json-bigint';
+import { Schema, validateAndMap } from '@apimatic/schema';
+import {
+  ApiResponse,
+  HttpContext,
+  HttpInterceptorInterface,
+  RequestOptions,
+} from './coreInterfaces';
+import { drainBodyToText, SseStream } from './sseStream';
+import { RawSseEvent, SseDecodeResult, SseDecoder } from './sseTypes';
+
+const SSE_CONTENT_TYPE = 'text/event-stream';
+
+/**
+ * Parses JSON with the same big-integer-safe reader the request builder's
+ * `callAsJson` uses, so streamed frames decode identically to non-streaming
+ * responses (integers beyond `Number.MAX_SAFE_INTEGER` keep their precision).
+ */
+const jsonBig = JSONBig();
+
+/** Cap on how much of a non-2xx error body is buffered into text. */
+const ERROR_BODY_MAX_BYTES = 64 * 1024;
+
+/** The subset of the request builder the SSE machinery needs. */
+export interface StreamCapableRequestBuilder {
+  accept(acceptHeaderValue: string): void;
+  intercept(
+    interceptor: HttpInterceptorInterface<RequestOptions | undefined>
+  ): void;
+  callAsStream(
+    requestOptions?: RequestOptions
+  ): Promise<ApiResponse<NodeJS.ReadableStream | Blob>>;
+}
+
+/**
+ * A decoder that validates and maps each event's `data` payload against
+ * `schema` (the same mechanism as the request builder's `callAsJson`), so a
+ * single decoder works for every response type — `string()`, `number()`,
+ * objects, arrays, etc. A frame that fails validation is surfaced as an
+ * `SseDecodeError`.
+ */
+export function schemaSseDecoder<T>(schema: Schema<T>): SseDecoder<T> {
+  return (raw: RawSseEvent): SseDecodeResult<T> => {
+    const parsed = parseFrameData(raw.data);
+    let mapped = validateAndMap(parsed, schema);
+    if (mapped.errors && typeof parsed !== 'string') {
+      // A plain-text frame can coincidentally be valid JSON (`123`, `true`,
+      // `null`); when the schema rejects the parsed form, try the raw text
+      // before failing so string schemas keep accepting such frames.
+      const rawMapped = validateAndMap(raw.data, schema);
+      if (!rawMapped.errors) {
+        mapped = rawMapped;
+      }
+    }
+    if (mapped.errors) {
+      throw new Error(
+        'SSE frame failed schema validation:\n' +
+          mapped.errors.map((e) => e.message).join('\n')
+      );
+    }
+    return { type: 'event', value: mapped.result };
+  };
+}
+
+/**
+ * Prepares a frame's `data` for schema validation. Structured payloads
+ * (objects, arrays, numbers) arrive as JSON text and are parsed with the
+ * big-integer-safe reader; a payload that isn't JSON — e.g. a plain-text
+ * `string()` frame — is validated as the raw string as-is.
+ */
+function parseFrameData(data: string): unknown {
+  try {
+    return jsonBig.parse(data);
+  } catch {
+    return data;
+  }
+}
+
+/**
+ * Executes a request as a server-sent event stream. Core's
+ * `callAsEventStream` hook delegates here.
+ */
+export async function createEventStream<T>(
+  requestBuilder: StreamCapableRequestBuilder,
+  requestOptions: RequestOptions | undefined,
+  streamReadTimeout: number | undefined,
+  decoder: SseDecoder<T>,
+  terminator?: string
+): Promise<ApiResponse<SseStream<T>>> {
+  const controller = new AbortController();
+  const userSignal = requestOptions?.abortSignal;
+  if (userSignal !== undefined) {
+    if (userSignal.aborted) {
+      controller.abort();
+    } else {
+      const forwardAbort = () => controller.abort();
+      userSignal.addEventListener('abort', forwardAbort);
+      controller.signal.addEventListener('abort', () =>
+        userSignal.removeEventListener('abort', forwardAbort)
+      );
+    }
+  }
+  const streamRequestOptions: RequestOptions = {
+    ...requestOptions,
+    abortSignal: controller.signal as unknown as RequestOptions['abortSignal'],
+  };
+
+  requestBuilder.accept(SSE_CONTENT_TYPE);
+  requestBuilder.intercept(bufferErrorBodies);
+
+  const wrappedDecoder = wrapDecoder(decoder, terminator);
+  const response = await requestBuilder
+    .callAsStream(streamRequestOptions)
+    .catch((error) => {
+      // Abort so the user-signal forwarding listener above is detached even
+      // when the request itself fails (e.g. a non-2xx ApiError).
+      controller.abort();
+      throw error;
+    });
+  const stream = new SseStream<T>(response.result, {
+    decoder: wrappedDecoder,
+    controller,
+    readTimeout: streamReadTimeout,
+  });
+  return { ...response, result: stream };
+}
+
+/** Drains a non-2xx stream body back into text before error handling. */
+const bufferErrorBodies: HttpInterceptorInterface<
+  RequestOptions | undefined
+> = async (request, options, next) => {
+  const context: HttpContext = await next(request, options);
+  const { response } = context;
+  if (response.statusCode >= 200 && response.statusCode < 300) {
+    return context;
+  }
+  const body = await drainBodyToText(response.body, ERROR_BODY_MAX_BYTES);
+  return { ...context, response: { ...response, body } };
+};
+
+function wrapDecoder<T>(
+  decoder: SseDecoder<T>,
+  terminator: string | undefined
+): SseDecoder<T> {
+  if (terminator === undefined) {
+    return decoder;
+  }
+  return (raw) => (raw.data === terminator ? { type: 'done' } : decoder(raw));
+}

packages/sse/src/errors/sseDecodeError.ts

@@ -0,0 +1,18 @@
+import { SseError } from './sseError';
+
+/** A frame's payload could not be decoded into the expected type. */
+export class SseDecodeError extends SseError {
+  /** The raw frame payload that failed to decode. */
+  public readonly rawFrame: string;
+  /** The underlying decode failure. */
+  public readonly cause: unknown;
+
+  constructor(rawFrame: string, cause: unknown) {
+    super(
+      'Failed to decode a Server-Sent Events frame. ' +
+        'See rawFrame for the offending payload.'
+    );
+    this.rawFrame = rawFrame;
+    this.cause = cause;
+  }
+}

packages/sse/src/errors/sseError.ts

@@ -0,0 +1,8 @@
+/** Base class for errors thrown while iterating a server-sent event stream. */
+export abstract class SseError extends Error {
+  constructor(message: string) {
+    super(message);
+    Object.setPrototypeOf(this, new.target.prototype);
+    this.name = new.target.name;
+  }
+}

packages/sse/src/errors/sseTimeoutError.ts

@@ -0,0 +1,15 @@
+import { SseError } from './sseError';
+
+/** The server stalled between frames longer than the read timeout. */
+export class SseTimeoutError extends SseError {
+  /** The idle window (ms) that elapsed without a frame arriving. */
+  public readonly idleTimeoutMs: number;
+
+  constructor(idleTimeoutMs: number) {
+    super(
+      `SSE stream idle timeout exceeded (${idleTimeoutMs}ms without a frame). ` +
+        'This bounds only the wait for the server, not consumer processing time.'
+    );
+    this.idleTimeoutMs = idleTimeoutMs;
+  }
+}

packages/sse/src/index.ts

@@ -0,0 +1,19 @@
+/**
+ * @apimatic/sse
+ *
+ * Server-Sent Events utilities for APIMatic-generated TypeScript SDKs.
+ */
+export { SseParser } from './sseParser';
+export { SseStream } from './sseStream';
+export type { SseStreamOptions } from './sseStream';
+export { SseError } from './errors/sseError';
+export { SseDecodeError } from './errors/sseDecodeError';
+export { SseTimeoutError } from './errors/sseTimeoutError';
+export { createEventStream, schemaSseDecoder } from './createEventStream';
+export type { StreamCapableRequestBuilder } from './createEventStream';
+export type {
+  RawSseEvent,
+  SseDecodeResult,
+  SseDecoder,
+  SseEvent,
+} from './sseTypes';