feat: add puter.peer to sdk, create PeerService in backend (#2664)

* feat: add puter.peer to sdk, create PeerService in backend

* cloudflare turn, typedefs, fallback ice, bugfixes

* minr tweak

* restrict peerservice to api.puter.com, make customIndentifier more detailed

---------

Co-authored-by: ProgrammerIn-wonderland <30693865+ProgrammerIn-wonderland@users.noreply.github.com>

src/backend/src/CoreModule.js

@@ -381,6 +381,9 @@ const install = async ({ context, services, app, useapi, modapi }) => {
     const { PermissionShortcutService } = require('./services/auth/PermissionShortcutService');
     services.registerService('permission-shortcut', PermissionShortcutService);
 
+    const { PeerService } = require('./services/PeerService');
+    services.registerService('peer', PeerService);
+
 };
 
 const install_legacy = async ({ services }) => {

src/backend/src/services/PeerService.js

@@ -0,0 +1,77 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import configurable_auth from '../middleware/configurable_auth.js';
+import { Endpoint } from '../util/expressutil.js';
+import BaseService from './BaseService.js';
+
+export class PeerService extends BaseService {
+    '__on_install.routes' (_, { app }) {
+        Endpoint({
+            route: '/peer/signaller-info',
+            methods: ['GET'],
+            subdomain: "api",
+            handler: async (req, res) => {
+                res.json({
+                    url: this.config.signaller_url,
+                    fallbackIce: this.config.fallback_ice,
+                });
+            },
+        }).attach(app);
+
+        Endpoint({
+            route: '/peer/generate-turn',
+            methods: ['POST'],
+            mw: [configurable_auth()],
+            subdomain: "api",
+            handler: async (req, res) => {
+                if ( ! this.config.cloudflare_turn ) {
+                    res.status(500).send({ error: 'TURN is not configured' });
+                    return;
+                }
+                let response = await fetch(
+                    `https://rtc.live.cloudflare.com/v1/turn/keys/${this.config.cloudflare_turn.turn_key_id}/credentials/generate-ice-servers`,
+                    {
+                        headers: {
+                            Authorization: `Bearer ${this.config.cloudflare_turn.turn_key_api_token}`,
+                            'Content-Type': 'application/json',
+                        },
+                        method: 'POST',
+                        body: JSON.stringify({
+                            ttl: this.config.cloudflare_turn.ttl_ms,
+                            customIdentifier: `${req.actor.type.user.uuid}:${req.actor.type?.app?.uid ?? 'global'}`,
+                        }),
+                    },
+                );
+
+                if ( ! response.ok ) {
+                    res.status(500).send({ error: 'Failed to generate TURN credentials' });
+                    return;
+                }
+
+                const { iceServers } = await response.json();
+
+                res.json({
+                    ttl: this.config.cloudflare_turn.ttl_ms,
+                    iceServers,
+                });
+            },
+        }).attach(app);
+    }
+}

src/puter-js/index.d.ts

@@ -11,6 +11,7 @@ import type { KV, KVIncrementPath, KVPair } from './types/modules/kv.d.ts';
 import type { Networking, PSocket, PTLSSocket } from './types/modules/networking.d.ts';
 import type { OS } from './types/modules/os.d.ts';
 import type { Perms } from './types/modules/perms.d.ts';
+import type Peer, { PuterPeerConnection, PuterPeerServer } from './types/modules/peer.d.ts';
 import type { AlertButton, AppConnection, AppConnectionCloseEvent, CancelAwarePromise, ContextMenuItem, ContextMenuOptions, DirectoryPickerOptions, FilePickerOptions, LaunchAppOptions, MenuItem, MenubarOptions, ThemeData, UI, WindowOptions } from './types/modules/ui.d.ts';
 import type Util, { UtilRPC } from './types/modules/util.d.ts';
 import type { WorkerDeployment, WorkerInfo, WorkersHandler } from './types/modules/workers.d.ts';
@@ -77,8 +78,11 @@ export type {
     OS,
     PaginatedResult,
     PaginationOptions,
+    Peer,
     Perms,
     PSocket,
+    PuterPeerConnection,
+    PuterPeerServer,
     PTLSSocket,
     Puter,
     PuterEnvironment,

src/puter-js/src/index.js

@@ -21,6 +21,7 @@ import Perms from './modules/Perms.js';
 import UI from './modules/UI.js';
 import Util from './modules/Util.js';
 import { WorkersHandler } from './modules/Workers.js';
+import Peer from './modules/Peer.js';
 
 class SimpleLogger {
     constructor (fields = {}) {
@@ -172,6 +173,7 @@ const puterInit = (function () {
             this.registerModule('perms', Perms);
             this.registerModule('drivers', Drivers);
             this.registerModule('debug', Debug);
+            this.registerModule('peer', Peer);
 
             // Path
             this.path = path;

src/puter-js/src/modules/Peer.js

@@ -0,0 +1,467 @@
+class PuterPeerServerConnectionEvent extends Event {
+    conn;
+    user;
+    constructor (connection, user) {
+        super('connection');
+        this.conn = connection;
+        this.user = user;
+    }
+}
+
+class PuterPeerConnectionMessageEvent extends Event {
+    data;
+    constructor (message) {
+        super('message');
+        this.data = message;
+    }
+}
+
+class PuterPeerConnectionOpenEvent extends Event {
+    constructor () {
+        super('open');
+    }
+}
+
+class PuterPeerConnectionCloseEvent extends Event {
+    reason;
+    constructor (reason = undefined) {
+        super('close');
+        this.reason = reason;
+    }
+}
+
+class PuterPeerConnectionErrorEvent extends Event {
+    error;
+    constructor (error) {
+        super('error');
+        this.error = error;
+    }
+}
+
+class PuterPeerServer extends EventTarget {
+    #wsconn;
+    #oncreateresolve;
+
+    /** @type {Map<string, PuterPeerConnection>} */
+    #connections = new Map();
+    invitecode;
+    #peerConfig;
+
+    constructor (peerConfig) {
+        super();
+        this.#peerConfig = peerConfig;
+        this.#wsconn = new WebSocket(peerConfig.signallerUrl);
+    }
+
+    async start () {
+        await new Promise((resolve, reject) => {
+            this.#wsconn.onopen = resolve;
+            this.#wsconn.onerror = reject;
+            this.#wsconn.onclose = () => {
+                reject(new Error('Connection closed unexpectedly'));
+            };
+        });
+
+        this.#wsconn.onmessage = (event) => {
+            let data = JSON.parse(event.data);
+            this.#message(data);
+        };
+
+        this.#wsconn.onclose = () => {
+            // what should we do here?
+        };
+
+        this.#wsconn.send(
+            JSON.stringify({
+                server: {
+                    create: {
+                        authToken: this.#peerConfig.authToken,
+                    },
+                },
+            }),
+        );
+
+        const { invitecode } = await new Promise((resolve, reject) => {
+            this.#oncreateresolve = (data) => {
+                if ( data.success ) {
+                    resolve({
+                        invitecode: data.invitecode,
+                    });
+                    this.#oncreateresolve = null;
+                    this.invitecode = data.invitecode;
+                } else {
+                    reject(new Error(data.error));
+                }
+            };
+            setTimeout(
+                () => reject(new Error('Server creation timed out')),
+                15000,
+            );
+        });
+
+        return invitecode;
+    }
+
+    async #message (data) {
+        if ( ! data.server ) return;
+        if ( data.server.create ) {
+            this.#oncreateresolve(data.server.create);
+            return;
+        }
+
+        if ( data.server.connect ) {
+            let uuid = data.server.connect.id;
+            let connection = new PuterPeerConnection(this.#peerConfig);
+            this.#connections.set(uuid, connection);
+            connection.peerconnection.onicecandidate = (e) => {
+                if ( e.candidate ) {
+                    this.#wsconn.send(
+                        JSON.stringify({
+                            server: {
+                                candidate: {
+                                    id: uuid,
+                                    candidate: e.candidate,
+                                },
+                            },
+                        }),
+                    );
+                }
+            };
+            this.dispatchEvent(
+                new PuterPeerServerConnectionEvent(
+                    connection,
+                    data.server.connect.user,
+                ),
+            );
+        }
+
+        if ( data.server.candidate ) {
+            let uuid = data.server.candidate.id;
+            let connection = this.#connections.get(uuid);
+            if ( connection ) {
+                await connection.addIceCandidate(
+                    data.server.candidate.candidate,
+                );
+            }
+        }
+
+        if ( data.server.offer ) {
+            let uuid = data.server.offer.id;
+            let connection = this.#connections.get(uuid);
+            if ( connection ) {
+                await connection.setRemoteDescription(
+                    new RTCSessionDescription(data.server.offer.offer),
+                );
+            }
+
+            const answer = await connection.createAnswer();
+            this.#wsconn.send(
+                JSON.stringify({
+                    server: {
+                        answer: {
+                            id: uuid,
+                            answer,
+                        },
+                    },
+                }),
+            );
+        }
+    }
+
+    close () {
+        for ( const [uuid, connection] of this.#connections ) {
+            connection.close();
+        }
+        this.#wsconn.onclose = null;
+        this.#wsconn.close();
+    }
+}
+
+class PuterPeerConnection extends EventTarget {
+    #wsconn;
+    peerconnection;
+    #peerConfig;
+    #datachannel;
+    connected = false;
+    closed = false;
+    #bufferedMessages = [];
+    constructor (peerConfig) {
+        super();
+        this.#peerConfig = peerConfig;
+        this.peerconnection = new RTCPeerConnection({
+            iceServers: peerConfig.iceServers,
+        });
+        this.#datachannel = this.peerconnection.createDataChannel('channel-1', { negotiated: true, id: 2 });
+        this.#datachannel.onmessage = (evt) => {
+            this.dispatchEvent(new PuterPeerConnectionMessageEvent(evt.data));
+        };
+        this.#datachannel.onopen = () => {
+            this.connected = true;
+            for ( const message of this.#bufferedMessages ) {
+                this.send(message);
+            }
+            this.#bufferedMessages = [];
+            this.dispatchEvent(new PuterPeerConnectionOpenEvent());
+            this.#closews();
+        };
+        this.#datachannel.onclose = () => {
+            this.#doclose(undefined, undefined);
+        };
+        this.#datachannel.onerror = (evt) => {
+            this.#doclose(undefined, evt.error);
+        };
+    }
+
+    #closews () {
+        if ( this.#wsconn ) {
+            this.#wsconn.onclose = null;
+            this.#wsconn.close();
+            this.#wsconn = null;
+        }
+    }
+
+    async connect (invitecode) {
+        this.#wsconn = new WebSocket(this.#peerConfig.signallerUrl);
+        await new Promise((resolve, reject) => {
+            this.#wsconn.onopen = resolve;
+            this.#wsconn.onerror = reject;
+            this.#wsconn.onclose = () => {
+                reject(new Error('Connection closed unexpectedly'));
+            };
+        });
+        this.#wsconn.onopen = null;
+        this.#wsconn.onerror = null;
+        // post initial connect close
+        this.#wsconn.onclose = () => {
+            this.#doclose(undefined, new Error('Connection closed unexpectedly before peer offer was sent'));
+        };
+
+        this.#wsconn.send(
+            JSON.stringify({
+                client: {
+                    connect: {
+                        authToken: this.#peerConfig.authToken,
+                        invitecode,
+                    },
+                },
+            }),
+        );
+
+        this.peerconnection.onicecandidate = (evt) => {
+            this.#wsconn.send(
+                JSON.stringify({
+                    client: {
+                        candidate: {
+                            candidate: evt.candidate,
+                        },
+                    },
+                }),
+            );
+        };
+
+        this.#wsconn.onmessage = async (evt) => {
+            let msg = JSON.parse(evt.data).client;
+            if ( ! msg ) return;
+            if ( msg.answer ) {
+                this.setRemoteDescription(msg.answer.answer);
+            }
+            if ( msg.candidate ) {
+                this.addIceCandidate(msg.candidate.candidate);
+            }
+            if ( msg.connect ) {
+                if ( msg.connect.success ) {
+                    const offer = await this.createOffer();
+                    this.#wsconn.send(
+                        JSON.stringify({
+                            client: {
+                                offer: {
+                                    offer,
+                                },
+                            },
+                        }),
+                    );
+                } else {
+                    this.#doclose(undefined, new Error(msg.connect.error));
+                }
+            }
+            if ( msg.disconnect && !this.connected ) {
+                this.#doclose(msg.disconnect.reason);
+            }
+        };
+    }
+
+    #doclose (reason, error) {
+        if ( this.closed ) return;
+        this.closed = true;
+        this.connected = false;
+        if ( this.#wsconn ) this.#closews();
+        if ( this.#datachannel ) {
+            this.#datachannel.onclose = null;
+            this.#datachannel.close();
+        }
+        if ( this.peerconnection ) {
+            this.peerconnection.close();
+        }
+        if ( error ) this.dispatchEvent(new PuterPeerConnectionErrorEvent(error));
+        this.dispatchEvent(new PuterPeerConnectionCloseEvent(reason));
+    }
+
+    close (reason) {
+        this.#doclose(reason, undefined);
+    }
+
+    async createOffer () {
+        const offer = await this.peerconnection.createOffer();
+        await this.peerconnection.setLocalDescription(offer);
+        return offer;
+    }
+
+    async createAnswer () {
+        const answer = await this.peerconnection.createAnswer();
+        await this.peerconnection.setLocalDescription(answer);
+        return answer;
+    }
+
+    async setRemoteDescription (description) {
+        await this.peerconnection.setRemoteDescription(description);
+    }
+
+    async addIceCandidate (candidate) {
+        await this.peerconnection.addIceCandidate(candidate);
+    }
+
+    send ( message ) {
+        if ( ! this.connected ) {
+            this.#bufferedMessages.push(message);
+            return;
+        }
+        this.#datachannel.send(message);
+    }
+}
+
+class Peer {
+    #signallerUrl;
+    #turnServers;
+    #fallbackIceServers;
+    #turnTTL;
+    #turnStartedAt;
+    #turnFailed;
+    /**
+     * Creates a new instance with the given authentication token, API origin, and app ID,
+     *
+     * @class
+     * @param {string} authToken - Token used to authenticate the user.
+     * @param {string} APIOrigin - Origin of the API server. Used to build the API endpoint URLs.
+     * @param {string} appID - ID of the app to use.
+     */
+    constructor (puter) {
+        this.puter = puter;
+        this.authToken = puter.authToken;
+        this.APIOrigin = puter.APIOrigin;
+        this.appID = puter.appID;
+    }
+
+    /**
+     * Sets a new authentication token.
+     *
+     * @param {string} authToken - The new authentication token.
+     * @memberof [OS]
+     * @returns {void}
+     */
+    setAuthToken (authToken) {
+        this.authToken = authToken;
+    }
+
+    /**
+     * Sets the API origin.
+     *
+     * @param {string} APIOrigin - The new API origin.
+     * @memberof [Apps]
+     * @returns {void}
+     */
+    setAPIOrigin (APIOrigin) {
+        this.APIOrigin = APIOrigin;
+    }
+
+    async ensureTurnRelays () {
+        if ( this.#turnFailed ) return;
+        if ( this.#turnServers && Date.now() - this.#turnStartedAt < this.#turnTTL * 1000 ) return;
+
+        const response = await fetch(`${this.APIOrigin}/peer/generate-turn`, {
+            method: 'POST',
+            headers: {
+                'Content-Type': 'application/json',
+                'Authorization': `Bearer ${this.authToken}`,
+            },
+        });
+
+        if ( ! response.ok ) {
+            this.#turnFailed = true;
+            return;
+        }
+
+        const { iceServers, ttl, fallbackIce } = await response.json();
+        this.#fallbackIceServers = fallbackIce;
+        this.#turnServers = iceServers;
+        this.#turnTTL = ttl;
+        this.#turnStartedAt = Date.now();
+    }
+
+    async #loadMetadata () {
+        if ( this.#signallerUrl ) return;
+        const response = await fetch(`${this.APIOrigin}/peer/signaller-info`);
+        if ( ! response.ok ) {
+            throw new Error('Failed to get signaller info from Puter.');
+        }
+        const { url } = await response.json();
+        this.#signallerUrl = url;
+    }
+
+    async #authenticateForPeerAction (action) {
+        if ( this.authToken || this.puter.env !== 'web' ) return;
+        try {
+            await this.puter.ui.authenticateWithPuter();
+        } catch (e) {
+            throw new Error(`Need authentication to ${action} but failed to authenticate with Puter.`);
+        }
+    }
+
+    async #resolvePeerConfig (options) {
+        await this.#loadMetadata();
+        let iceServers;
+        if ( options?.iceServers ) {
+            iceServers = options.iceServers;
+        } else {
+            await this.ensureTurnRelays();
+            if ( this.#turnServers ) {
+                iceServers = this.#turnServers;
+            } else {
+                iceServers = this.#fallbackIceServers;
+                console.warn('Unable to use TURN relays. Some connections may fail.');
+            }
+        }
+
+        return {
+            authToken: this.authToken,
+            iceServers,
+            signallerUrl: this.#signallerUrl,
+        };
+    }
+    async serve (options) {
+        await this.#authenticateForPeerAction('create a server');
+        const peerConfig = await this.#resolvePeerConfig(options);
+        const server = new PuterPeerServer(peerConfig);
+        await server.start();
+        return server;
+    }
+
+    async connect (invitecode, options) {
+        await this.#authenticateForPeerAction('connect to a server');
+        const peerConfig = await this.#resolvePeerConfig(options);
+        const conn = new PuterPeerConnection(peerConfig);
+        await conn.connect(invitecode);
+        return conn;
+    }
+}
+
+export default Peer;

src/puter-js/types/modules/peer.d.ts

@@ -0,0 +1,98 @@
+export interface PuterPeerOptions {
+    iceServers?: RTCIceServer[];
+}
+
+export interface PuterPeerUser extends Record<string, unknown> {}
+
+export type PuterPeerMessage = string | Blob | ArrayBuffer | ArrayBufferView;
+export type PuterPeerDescription = RTCSessionDescription | RTCSessionDescriptionInit;
+export type PuterPeerIceCandidate = RTCIceCandidate | RTCIceCandidateInit;
+
+export class PuterPeerServerConnectionEvent extends Event {
+    readonly conn: PuterPeerConnection;
+    readonly user: PuterPeerUser;
+}
+
+export class PuterPeerConnectionMessageEvent extends Event {
+    readonly data: unknown;
+}
+
+export class PuterPeerConnectionOpenEvent extends Event {}
+
+export class PuterPeerConnectionCloseEvent extends Event {
+    readonly reason?: unknown;
+}
+
+export class PuterPeerConnectionErrorEvent extends Event {
+    readonly error: unknown;
+}
+
+export interface PuterPeerServerEventMap {
+    connection: PuterPeerServerConnectionEvent;
+}
+
+export interface PuterPeerConnectionEventMap {
+    open: PuterPeerConnectionOpenEvent;
+    message: PuterPeerConnectionMessageEvent;
+    close: PuterPeerConnectionCloseEvent;
+    error: PuterPeerConnectionErrorEvent;
+}
+
+export class PuterPeerServer extends EventTarget {
+    invitecode?: string;
+
+    start (): Promise<string>;
+    message (data: unknown): Promise<void>;
+
+    addEventListener(type: string, callback: EventListenerOrEventListenerObject | null, options?: boolean | AddEventListenerOptions): void;
+    addEventListener<K extends keyof PuterPeerServerEventMap>(
+        type: K,
+        listener: (this: PuterPeerServer, ev: PuterPeerServerEventMap[K]) => unknown,
+        options?: boolean | AddEventListenerOptions,
+    ): void;
+    removeEventListener(type: string, callback: EventListenerOrEventListenerObject | null, options?: boolean | EventListenerOptions): void;
+    removeEventListener<K extends keyof PuterPeerServerEventMap>(
+        type: K,
+        listener: (this: PuterPeerServer, ev: PuterPeerServerEventMap[K]) => unknown,
+        options?: boolean | EventListenerOptions,
+    ): void;
+}
+
+export class PuterPeerConnection extends EventTarget {
+    peerconnection: RTCPeerConnection;
+    connected: boolean;
+    closed: boolean;
+
+    connect (invitecode: string): Promise<void>;
+    close (reason?: unknown): void;
+    createOffer (): Promise<RTCSessionDescriptionInit>;
+    createAnswer (): Promise<RTCSessionDescriptionInit>;
+    setRemoteDescription (description: PuterPeerDescription): void;
+    addIceCandidate (candidate: PuterPeerIceCandidate): void;
+    send (message: PuterPeerMessage): void;
+
+    addEventListener(type: string, callback: EventListenerOrEventListenerObject | null, options?: boolean | AddEventListenerOptions): void;
+    addEventListener<K extends keyof PuterPeerConnectionEventMap>(
+        type: K,
+        listener: (this: PuterPeerConnection, ev: PuterPeerConnectionEventMap[K]) => unknown,
+        options?: boolean | AddEventListenerOptions,
+    ): void;
+    removeEventListener(type: string, callback: EventListenerOrEventListenerObject | null, options?: boolean | EventListenerOptions): void;
+    removeEventListener<K extends keyof PuterPeerConnectionEventMap>(
+        type: K,
+        listener: (this: PuterPeerConnection, ev: PuterPeerConnectionEventMap[K]) => unknown,
+        options?: boolean | EventListenerOptions,
+    ): void;
+}
+
+export default class Peer {
+    authToken?: string | null;
+    APIOrigin: string;
+    appID?: string;
+
+    setAuthToken (authToken: string): void;
+    setAPIOrigin (APIOrigin: string): void;
+    ensureTurnRelays (): Promise<void>;
+    serve (options?: PuterPeerOptions): Promise<PuterPeerServer>;
+    connect (invitecode: string, options?: PuterPeerOptions): Promise<PuterPeerConnection>;
+}

src/puter-js/types/puter.d.ts

@@ -9,6 +9,7 @@ import type { Hosting } from './modules/hosting.d.ts';
 import type { KV } from './modules/kv.d.ts';
 import type { Networking } from './modules/networking.d.ts';
 import type { OS } from './modules/os.d.ts';
+import type Peer from './modules/peer.d.ts';
 import type { Perms } from './modules/perms.d.ts';
 import type { UI } from './modules/ui.d.ts';
 import type Util from './modules/util.d.ts';
@@ -55,6 +56,7 @@ export class Puter {
     perms: Perms;
     drivers: Drivers;
     debug: Debug;
+    peer: Peer | null;
     path: {
         join: (...parts: string[]) => string;
         dirname: (p: string) => string;