feat: add stream helper (#395)

This PR adds a helper to support streaming responses in Netlify
Functions. The decorator handles all the `awslambda` things under the
hood, all that devs have to do is return a `NodeJS.Readable` or a Web
Stream as the `body`.

It also updates the Node.js version to v14, so that `pipeline` is
available, which makes this a breaking change technically - but as
@ascorbic notes, not an actual one.

---------

Co-authored-by: Eduardo Bouças <mail@eduardoboucas.com>

Author: Skn0tt

.github/workflows/workflow.yml

@@ -13,12 +13,12 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, macOS-latest, windows-latest]
-        node-version: [8.17.0, '*']
+        node-version: [14.0.0, '*']
         exclude:
           - os: macOS-latest
-            node-version: 8.17.0
+            node-version: 14.0.0
           - os: windows-latest
-            node-version: 8.17.0
+            node-version: 14.0.0
       fail-fast: false
     steps:
       - name: Git checkout

package-lock.json

@@ -66,6 +66,6 @@
     "typescript": "^4.4.4"
   },
   "engines": {
-    "node": ">=8.3.0"
+    "node": ">=14.0.0"
   }
 }

package.json

@@ -1,6 +1,6 @@
 import type { Context } from './context.js'
 import type { Event } from './event.js'
-import type { Response, BuilderResponse } from './response.js'
+import type { Response, BuilderResponse, StreamingResponse } from './response.js'
 
 export interface HandlerCallback<ResponseType extends Response = Response> {
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -17,3 +17,7 @@ export interface BackgroundHandler<C extends Context = Context> {
 
 export type Handler = BaseHandler<Response, Context>
 export type BuilderHandler = BaseHandler<BuilderResponse, Context>
+
+export interface StreamingHandler {
+  (event: Event, context: Context): Promise<StreamingResponse>
+}

src/function/handler.ts

@@ -1,4 +1,4 @@
 export { Context as HandlerContext } from './context.js'
 export { Event as HandlerEvent } from './event.js'
-export { BuilderHandler, Handler, BackgroundHandler, HandlerCallback } from './handler.js'
-export { BuilderResponse, Response as HandlerResponse } from './response.js'
+export { BuilderHandler, Handler, BackgroundHandler, HandlerCallback, StreamingHandler } from './handler.js'
+export { BuilderResponse, Response as HandlerResponse, StreamingResponse } from './response.js'

src/function/index.ts

@@ -1,3 +1,5 @@
+import type { PipelineSource } from 'node:stream'
+
 export interface Response {
   statusCode: number
   headers?: {
@@ -12,3 +14,8 @@ export interface Response {
 export interface BuilderResponse extends Response {
   ttl?: number
 }
+
+export interface StreamingResponse extends Omit<Response, 'body'> {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  body?: string | PipelineSource<any>
+}

src/function/response.ts

@@ -0,0 +1,70 @@
+import { pipeline as pipelineSync } from 'node:stream'
+import { promisify } from 'node:util'
+
+import type { Handler, HandlerEvent, HandlerContext, StreamingHandler, StreamingResponse } from '../function/index.js'
+
+// Node v14 doesn't have node:stream/promises
+const pipeline = promisify(pipelineSync)
+
+declare global {
+  // eslint-disable-next-line @typescript-eslint/no-namespace
+  namespace awslambda {
+    function streamifyResponse(
+      handler: (event: HandlerEvent, responseStream: NodeJS.WritableStream, context: HandlerContext) => Promise<void>,
+    ): Handler
+
+    // eslint-disable-next-line @typescript-eslint/no-namespace
+    namespace HttpResponseStream {
+      function from(stream: NodeJS.WritableStream, metadata: Omit<StreamingResponse, 'body'>): NodeJS.WritableStream
+    }
+  }
+}
+
+/**
+ * Enables streaming responses. `body` accepts a Node.js `Readable` stream or a WHATWG `ReadableStream`.
+ *
+ * @example
+ * ```
+ * const { Readable } = require('stream');
+ *
+ * export const handler = stream(async (event, context) => {
+ *   const stream = Readable.from(Buffer.from(JSON.stringify(event)))
+ *   return {
+ *     statusCode: 200,
+ *     body: stream,
+ *   }
+ * })
+ * ```
+ *
+ * @example
+ * ```
+ * export const handler = stream(async (event, context) => {
+ *   const response = await fetch('https://api.openai.com/', { ... })
+ *   // ...
+ *   return {
+ *     statusCode: 200,
+ *     body: response.body, // Web stream
+ *   }
+ * })
+ * ```
+ *
+ * @param handler
+ * @see https://ntl.fyi/streaming-func
+ */
+const stream = (handler: StreamingHandler): Handler =>
+  awslambda.streamifyResponse(async (event, responseStream, context) => {
+    const { body, ...httpResponseMetadata } = await handler(event, context)
+
+    const responseBody = awslambda.HttpResponseStream.from(responseStream, httpResponseMetadata)
+
+    if (typeof body === 'undefined') {
+      responseBody.end()
+    } else if (typeof body === 'string') {
+      responseBody.write(body)
+      responseBody.end()
+    } else {
+      await pipeline(body, responseBody)
+    }
+  })
+
+export { stream }

src/lib/stream.ts

@@ -1,3 +1,4 @@
 export { builder } from './lib/builder.js'
 export { schedule } from './lib/schedule.js'
+export { stream } from './lib/stream.js'
 export * from './function/index.js'

src/main.ts

@@ -11,7 +11,7 @@
     // "disableReferencedProjectLoad": true,             /* Reduce the number of projects loaded automatically by TypeScript. */
 
     /* Language and Environment */
-    "target": "es5" /* Set the JavaScript language version for emitted JavaScript and include compatible library declarations. */,
+    "target": "ES2020" /* Set the JavaScript language version for emitted JavaScript and include compatible library declarations. */,
     // "lib": [],                                        /* Specify a set of bundled library declaration files that describe the target runtime environment. */
     // "jsx": "preserve",                                /* Specify what JSX code is generated. */
     // "experimentalDecorators": true,                   /* Enable experimental support for TC39 stage 2 draft decorators. */