[ECO-5260] feat: initial local proxy implementation

Author: ttypic

.github/workflows/check.yml

@@ -0,0 +1,21 @@
+name: Check
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 20
+      - run: npm ci
+
+      - name: Run test
+        run: npm run test

src/async-queue.ts

@@ -0,0 +1,35 @@
+type AsyncTask = () => Promise<void>;
+
+export class AsyncQueue {
+  private readonly queue: AsyncTask[] = [];
+  private processing: boolean = false;
+
+  enqueue(task: AsyncTask) {
+    return new Promise<void>((resolve, reject) => {
+      this.queue.push(async () => {
+        try {
+          await task();
+          resolve();
+        } catch (e) {
+          reject(e);
+        }
+      });
+      this.processNext();
+    });
+  }
+
+  private async processNext() {
+    if (this.processing || this.queue.length === 0) return;
+
+    this.processing = true;
+
+    const task = this.queue.shift();
+
+    try {
+      await task!!();
+    } finally {
+      this.processing = false;
+      this.processNext();
+    }
+  }
+}

src/index.ts

@@ -0,0 +1 @@
+export * from './proxy.js';

src/proxy.test.ts

@@ -0,0 +1,91 @@
+import { beforeAll, describe, expect, it } from 'vitest';
+import WebSocket, { WebSocketServer } from 'ws';
+import { createInterceptingProxy, type InterceptingProxy } from './proxy.js';
+import http from 'http';
+
+describe('messages integration', () => {
+  let proxy: InterceptingProxy;
+
+  beforeAll(async () => {
+    const server = await startEchoServer();
+    proxy = createInterceptingProxy({
+      realtimeHost: 'localhost',
+      restHost: 'localhost',
+      port: 8080,
+      tls: false,
+    });
+    await proxy.start();
+    return () => new Promise<void>((resolve) => server.close(() => resolve()));
+  });
+
+  it('test', async () => {
+    const unregister = proxy.registerRealtimeMiddleware(async (message, messageType) => {
+      if (messageType === 'incoming') {
+        return { command: 'replace', replacement: { action: 3 } };
+      } else {
+        return { command: 'keep' };
+      }
+    });
+
+    const testWs = new WebSocket(`ws://localhost:${proxy.options.port}`);
+
+    await openConnection(testWs);
+
+    await runAndObserve(
+      async () => testWs.send(JSON.stringify({ action: 1 })),
+      async () => {
+        const message = await proxy.observeNextIncomingProtocolMessage();
+        const rawWebsocketMessage = await waitForNextReceivedMessage(testWs);
+        expect(message).toEqual({ action: 3 });
+        expect(rawWebsocketMessage.toString()).toBe('{"action":3}');
+      },
+    );
+
+    unregister();
+    testWs.close();
+  });
+});
+
+async function startEchoServer(port = 8080) {
+  const echoServer = http.createServer((req, res) => {
+    const bodyChunks: any[] = [];
+    req.on('data', (chunk) => bodyChunks.push(chunk));
+    req.on('end', () => {
+      const requestData = Buffer.concat(bodyChunks).toString();
+      res.writeHead(200);
+      res.end(requestData);
+    });
+  });
+
+  const wss = new WebSocketServer({ server: echoServer, path: '/' });
+
+  wss.on('connection', (ws) => {
+    ws.on('message', (message) => {
+      ws.send(message);
+    });
+  });
+
+  await new Promise<void>((resolve) => {
+    echoServer.listen(port, () => resolve());
+  });
+
+  return echoServer;
+}
+
+async function runAndObserve(runner: () => Promise<void>, observer: () => Promise<void>) {
+  const observerPromise = observer();
+  await runner();
+  await observerPromise;
+}
+
+async function openConnection(ws: WebSocket) {
+  return new Promise<void>((resolve) => ws.once('open', () => resolve()));
+}
+
+async function waitForNextReceivedMessage(ws: WebSocket) {
+  return new Promise<WebSocket.RawData>((resolve) => {
+    ws.once('message', (data) => {
+      resolve(data);
+    });
+  });
+}