Forked from mcandeia/warp

Signed-off-by: Marcos Candeia <[email protected]>

.github/workflows/publish.yaml

@@ -0,0 +1,19 @@
+name: Publish
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+
+    permissions:
+      contents: read
+      id-token: write
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Publish package
+        run: npx jsr publish

README.md

@@ -1,2 +1,127 @@
-# warp
-A simple tool that allows your locally running HTTP(s) servers to have a public URL, serving as an easy-to-self-host alternative to services like `ngrok`
+# Warp
+
+**Warp** is a simple tool that allows your locally running HTTP(s) servers to have a public URL, serving as an easy-to-self-host alternative to services like `ngrok`. Warp is implemented in Deno with the goal of providing flexibility and minimal dependencies.
+
+The project has two main components:
+
+- **Server**: Deployable on a server, it connects to the outside world and is accessible from any domain.
+- **Client**: Runs locally to connect a given HTTP endpoint running on a local or non-public network.
+
+## Server
+
+The Warp server opens a single HTTP port to which the Warp client connects and upgrades to a WebSocket connection. Each request to this HTTP port is forwarded (based on the client's HOST header) to the corresponding connected Warp client connection, which then serves the request.
+
+### Usage
+
+To start the Warp server, import the `start` function from the Warp package and call it with the appropriate configuration.
+
+#### Example
+
+```typescript
+import { start } from "jsr:@mcandeia/warp";
+
+const port = 8080; // The port where the Warp server will listen
+const apiKeys = ["YOUR_API_KEY1", "YOUR_API_KEY2"]; // Array of API keys for authentication
+
+start({ port, apiKeys });
+```
+
+#### Parameters
+
+- `port`: The port number where the Warp server will listen for connections.
+- `apiKeys`: An array of API keys used for client authentication.
+
+## Client
+
+The Warp client connects to the Warp server. Upon connection, the client shares the given API key and the domain it wants to receive requests for.
+
+### Usage
+
+To connect a client to the Warp server, import the `connect` function from the Warp package and call it with the appropriate configuration.
+
+#### Example
+
+```typescript
+import { connect } from "jsr:@mcandeia/warp";
+
+const port = 3000; // The local port you want to expose
+const domain = "www.your.domain.com"; // The domain name for your service
+const server = "wss://YOUR_SERVER"; // The WebSocket URL of your Warp server
+const token = "YOUR_TOKEN"; // The authentication token
+
+const { registered, closed } = await connect({
+  domain,
+  localAddr: `http://localhost:${port}`,
+  server,
+  token,
+});
+
+await registered;
+console.log("Client registered successfully");
+
+closed.then(() => {
+  console.log("Connection closed");
+});
+```
+
+#### Parameters
+
+- `domain`: The domain name that will be used to access your localhost service.
+- `localAddr`: The local address of the service you want to expose (e.g., `http://localhost:3000`).
+- `server`: The WebSocket URL of your Warp server (e.g., `wss://YOUR_SERVER`).
+- `token`: The authentication token for connecting to the Warp server.
+
+#### Return Values
+
+- `registered`: A promise that resolves when the client has successfully registered with the server.
+- `closed`: A promise that resolves when the connection to the server is closed.
+
+## Example Workflow
+
+Here’s a complete example of setting up a Warp server and client:
+
+### Server
+
+```typescript
+import { start } from "jsr:@mcandeia/warp";
+
+const port = 8080;
+const apiKeys = ["YOUR_API_KEY1", "YOUR_API_KEY2"];
+
+start({ port, apiKeys });
+```
+
+### Client
+
+```typescript
+import { connect } from "jsr:@mcandeia/warp";
+
+const port = 3000;
+const domain = "www.your.domain.com";
+const server = "wss://YOUR_SERVER";
+const token = "YOUR_TOKEN";
+
+(async () => {
+  const { registered, closed } = await connect({
+    domain,
+    localAddr: `http://localhost:${port}`,
+    server,
+    token,
+  });
+
+  await registered;
+  console.log("Client registered successfully");
+
+  closed.then(() => {
+    console.log("Connection closed");
+  });
+})();
+```
+
+## Troubleshooting
+
+### Common Issues
+
+- **Invalid API Key**: Ensure that the API key you are using is listed in the `apiKeys` array on the server.
+- **Connection Refused**: Check that the server is running and accessible at the specified WebSocket URL.
+- **Domain Not Accessible**: Ensure that the domain name is correctly configured and pointing to the Warp server.

channel.ts

@@ -0,0 +1,121 @@
+import { Queue } from "./queue.ts";
+
+export interface Channel<T> {
+    close(): void;
+    send(value: T): Promise<void>;
+    recv(): AsyncIterableIterator<T>;
+}
+
+export const makeChan = <T>(): Channel<T> => {
+    const queue: Queue<{ value: T, resolve: () => void }> = new Queue();
+    const ctrl = new AbortController();
+    const abortPromise = Promise.withResolvers<void>();
+    ctrl.signal.onabort = () => {
+        abortPromise.resolve();
+    }
+
+    const send = (value: T): Promise<void> => {
+        return new Promise((resolve, reject) => {
+            if (ctrl.signal.aborted) reject(new Error("Channel is closed"));
+            queue.push({ value, resolve });
+        });
+    };
+
+    const close = () => {
+        ctrl.abort();
+    };
+
+    const recv = async function* (): AsyncIterableIterator<T> {
+        while (true) {
+            if (ctrl.signal.aborted) {
+                return;
+            }
+            try {
+                const next = await queue.pop({ signal: ctrl.signal });
+                next.resolve();
+                yield next.value;
+            } catch (_err) {
+                if (ctrl.signal.aborted) {
+                    return;
+                }
+                throw _err;
+            }
+        }
+    };
+
+    return { send, recv, close };
+};
+
+export interface DuplexChannel<TSend, TReceive> {
+    in: Channel<TReceive>
+    out: Channel<TSend>
+}
+
+export const makeWebSocket = <TSend, TReceive>(socket: WebSocket, parse: boolean = true): Promise<DuplexChannel<TSend, TReceive>> => {
+    const sendChan = makeChan<TSend>();
+    const recvChan = makeChan<TReceive>();
+    const ch = Promise.withResolvers<DuplexChannel<TSend, TReceive>>();
+    socket.onclose = () => {
+        sendChan.close();
+        recvChan.close();
+    }
+    socket.onerror = (err) => {
+        socket.close();
+        ch.reject(err);
+    }
+    socket.onmessage = async (msg) => {
+        let eventData = msg.data;
+        const target = msg?.target;
+        if (
+            target && "binaryType" in target &&
+            target.binaryType === "blob" && typeof eventData === "object" &&
+            "text" in eventData
+        ) {
+            eventData = await eventData.text();
+        }
+        const message = parse ? JSON.parse(eventData) : eventData;
+        await recvChan.send(message);
+    }
+    socket.onopen = async () => {
+        ch.resolve({ in: recvChan, out: sendChan });
+        for await (const message of sendChan.recv()) {
+            try {
+                socket.send(parse ? JSON.stringify(message) : message as ArrayBuffer);
+            } catch (_err) {
+                console.error("error sending message through socket", message);
+            }
+        }
+        socket.close();
+    }
+    return ch.promise;
+}
+
+export const makeReadableStream = (ch: Channel<Uint8Array>): ReadableStream<Uint8Array> => {
+    return new ReadableStream({
+        async start(controller) {
+            for await (const content of ch.recv()) {
+                controller.enqueue(content);
+            }
+            controller.close();
+        },
+        cancel() {
+            ch.close();
+        },
+    })
+}
+export const makeChanStream = (stream: ReadableStream): Channel<Uint8Array> => {
+    const chan = makeChan<Uint8Array>();
+
+    // Consume the transformed stream to trigger the pipeline
+    const reader = stream.getReader();
+    const processStream = async () => {
+        while (true) {
+            const { done, value } = await reader.read();
+            if (done) break;
+            await chan.send(value);
+        }
+        chan.close();
+    };
+    processStream().catch(console.error);
+    return chan;
+};

client.ts

@@ -0,0 +1,81 @@
+import { type Channel, makeWebSocket } from "./channel.ts";
+import { handleServerMessage } from "./handlers.client.ts";
+import type { ClientMessage, ClientState, ServerMessage } from "./messages.ts";
+
+/**
+ * Options for establishing a connection.
+ * @typedef {Object} ConnectOptions
+ * @property {string} token - The authentication token for connecting to the server.
+ * @property {string} domain - The domain to register the connection with.
+ * @property {string} server - The WebSocket server URL.
+ * @property {string} localAddr - The local address for the WebSocket connection.
+ */
+export interface ConnectOptions {
+    token: string;
+    domain: string;
+    server: string;
+    localAddr: string;
+}
+
+/**
+ * Represents a connection status object.
+ * @typedef {Object} Connected
+ * @property {Promise<void>} closed - A promise that resolves when the connection is closed.
+ * @property {Promise<void>} registered - A promise that resolves when the connection is registered.
+ */
+export interface Connected {
+    closed: Promise<void>;
+    registered: Promise<void>;
+}
+
+/**
+ * Establishes a WebSocket connection with the server.
+ * @param {ConnectOptions} opts - Options for establishing the connection.
+ * @returns {Promise<Connected>} A promise that resolves with the connection status.
+ */
+export const connect = async (opts: ConnectOptions): Promise<Connected> => {
+    const closed = Promise.withResolvers<void>();
+    const registered = Promise.withResolvers<void>();
+    const client = typeof Deno.createHttpClient === "function" ? Deno.createHttpClient({
+        allowHost: true,
+        proxy: {
+            url: opts.localAddr,
+        }
+    }) : undefined;
+
+    const socket = new WebSocket(`${opts.server}/_connect`);
+    const ch = await makeWebSocket<ClientMessage, ServerMessage>(socket);
+    await ch.out.send({
+        id: crypto.randomUUID(),
+        type: "register",
+        apiKey: opts.token,
+        domain: opts.domain,
+    });
+    const requestBody: Record<string, Channel<Uint8Array>> = {};
+    const wsMessages: Record<string, Channel<ArrayBuffer>> = {};
+
+    (async () => {
+        const state: ClientState = {
+            client,
+            localAddr: opts.localAddr,
+            live: false,
+            requestBody,
+            wsMessages,
+            ch,
+        }
+        for await (const message of ch.in.recv()) {
+            try {
+                await handleServerMessage(state, message);
+                if (state.live) {
+                    registered.resolve();
+                }
+            } catch (err) {
+                console.error(new Date(), "error handling message", err);
+                break;
+            }
+        }
+        closed.resolve();
+    })()
+    return { closed: closed.promise, registered: registered.promise };
+}
+

deno.json

@@ -0,0 +1,8 @@
+{
+  "name": "@deco/warp",
+  "version": "0.1.0",
+  "exports": "./mod.ts",
+  "tasks": {
+    "start": "deno run -A main.ts"
+  }
+}