feat: PoC for PubSub

package.json

@@ -179,6 +179,7 @@
     "ulid": "2.3.0",
     "unixify": "1.0.0",
     "update-notifier": "6.0.2",
+    "urlpattern-polyfill": "^9.0.0",
     "uuid": "9.0.0",
     "wait-port": "1.0.4",
     "winston": "3.8.2",

src/lib/pubsub.mjs

@@ -0,0 +1,76 @@
+import { EventEmitter } from 'node:events'
+
+import { URLPattern } from 'urlpattern-polyfill'
+
+const pattern = new URLPattern({ pathname: '/.netlify/pubsub/:topic' })
+
+export class PubSubServer {
+  #messages = new EventEmitter()
+
+  /**
+   * @param {import("node:http").IncomingMessage} req
+   * @param {import("node:http").ServerResponse} res
+   * @returns {"skip" | undefined}
+   */
+  handleRequest(req, res) {
+    const match = pattern.exec(req.url, 'http://localhost')
+    if (!match) return 'skip'
+
+    const { topic } = match.pathname.groups
+    if (!topic) {
+      res.statusCode = 400
+      res.write('Missing topic in URL')
+      res.end()
+      return
+    }
+
+    if (req.method === 'GET') {
+      this.#subscribe(req, res, topic)
+      return true
+    }
+    if (req.method === 'POST') {
+      this.#publish(req, res, topic)
+      return true
+    }
+
+    res.statusCode = 405
+    res.end()
+    return true
+  }
+
+  /**
+   * @param {import("node:http").IncomingMessage} req
+   * @param {import("node:http").ServerResponse} res
+   * @param {string} topic
+   */
+  #subscribe(req, res, topic) {
+    res.setHeader('Content-Type', 'text/event-stream')
+    res.statusCode = 200
+
+    const sendPing = () => res.write('event: ping\n\n')
+    sendPing()
+
+    const pingInterval = setInterval(sendPing, 1000)
+
+    const sendMessage = (message) => res.write(`data: ${message}\n\n`)
+
+    this.#messages.on(topic, sendMessage)
+
+    req.on('end', () => {
+      clearInterval(pingInterval)
+      this.#messages.off(topic, sendMessage)
+    })
+  }
+
+  /**
+   * @param {import("node:http").IncomingMessage} req
+   * @param {import("node:http").ServerResponse} res
+   * * @param {string} topic
+   */
+  async #publish(req, res, topic) {
+    const message = await req.originalBody
+    this.#messages.emit(topic, message)
+    res.statusCode = 202
+    res.end()
+  }
+}

src/utils/proxy.mjs

@@ -28,6 +28,7 @@ import {
 } from '../lib/edge-functions/proxy.mjs'
 import { fileExistsAsync, isFileAsync } from '../lib/fs.mjs'
 import { DEFAULT_FUNCTION_URL_EXPRESSION } from '../lib/functions/registry.mjs'
+import { PubSubServer } from '../lib/pubsub.mjs'
 import renderErrorTemplate from '../lib/render-error-template.mjs'
 
 import { NETLIFYDEVLOG, NETLIFYDEVWARN, log, chalk } from './command-helpers.mjs'
@@ -585,8 +586,26 @@ const initializeProxy = async function ({ configPath, distDir, env, host, port,
   return handlers
 }
 
+/**
+ *
+ * @param {{ pubSubServer: import("../lib/pubsub.mjs").PubSubServer}} param0
+ * @param {http.IncomingMessage} req
+ * @param {http.ServerResponse} res
+ * @returns
+ */
 const onRequest = async (
-  { addonsUrls, edgeFunctionsProxy, env, functionsRegistry, functionsServer, proxy, rewriter, settings, siteInfo },
+  {
+    addonsUrls,
+    edgeFunctionsProxy,
+    env,
+    functionsRegistry,
+    functionsServer,
+    proxy,
+    pubSubServer,
+    rewriter,
+    settings,
+    siteInfo,
+  },
   req,
   res,
 ) => {
@@ -600,6 +619,10 @@ const onRequest = async (
     return proxy.web(req, res, { target: edgeFunctionsProxyURL })
   }
 
+  if (pubSubServer.handleRequest(req, res) !== 'skip') {
+    return
+  }
+
   const functionMatch = functionsRegistry && (await functionsRegistry.getFunctionForURLPath(req.url, req.method))
 
   if (functionMatch) {
@@ -720,6 +743,8 @@ export const startProxy = async function ({
     siteInfo,
   })
 
+  const pubSubServer = new PubSubServer()
+
   const rewriter = await createRewriter({
     distDir: settings.dist,
     projectDir,
@@ -736,6 +761,7 @@ export const startProxy = async function ({
     addonsUrls,
     functionsRegistry,
     functionsServer,
+    pubSubServer,
     edgeFunctionsProxy,
     siteInfo,
     env,

tests/integration/__fixtures__/dev-server-for-pubsub/netlify/edge-functions/guard.ts

@@ -0,0 +1,11 @@
+export default async (req: Request) => {
+  // but only authorized
+  if (req.headers.get('Authorization') !== 'Bearer foo') {
+    return new Response('Unauthorized', { status: 401 })
+  }
+}
+
+export const config = {
+  path: '/.netlify/pubsub/*',
+  method: 'POST',
+}

tests/integration/commands/dev/pubsub.test.ts

@@ -0,0 +1,50 @@
+import { describe, expect, test } from 'vitest'
+
+import { FixtureTestContext, setupFixtureTests } from '../../utils/fixture.js'
+import { pause } from '../../utils/pause.cjs'
+
+describe('pubsub', () => {
+  setupFixtureTests('dev-server-for-pubsub', { devServer: true }, () => {
+    test<FixtureTestContext>('pubsub flow', async ({ devServer }) => {
+      const subscriptionController = new AbortController()
+      const url = `http://localhost:${devServer.port}/.netlify/pubsub/events`
+      const subscription = await fetch(url, { signal: subscriptionController.signal })
+      expect(subscription.headers.get('content-type')).toEqual('text/event-stream')
+
+      const events: string[] = []
+      subscription.body
+        ?.pipeTo(
+          new WritableStream({
+            write(chunk) {
+              events.push(new TextDecoder().decode(chunk))
+            },
+          }),
+        )
+        // eslint-disable-next-line promise/prefer-await-to-callbacks,promise/prefer-await-to-then
+        .catch((error) => {
+          expect(error.message).toEqual('The operation was aborted.')
+        })
+
+      const unauthenticated = await fetch(url, {
+        method: 'POST',
+        body: 'foo',
+      })
+      expect(unauthenticated.status).toBe(401)
+      expect(events).toEqual(['event: ping\n\n'])
+
+      const authenticated = await fetch(`http://localhost:${devServer.port}/.netlify/pubsub/events`, {
+        method: 'POST',
+        body: 'foo',
+        headers: {
+          Authorization: 'Bearer foo',
+        },
+      })
+      expect(authenticated.status).toEqual(202)
+
+      await pause(10)
+      expect(events).toEqual(['event: ping\n\n', 'data: foo\n\n'])
+
+      subscriptionController.abort()
+    })
+  })
+})