Merge pull request #176 from appwrite/feat-web-refactor-realtime

feat(web): realtime prototype

Author: TorstenDittmann

templates/web/src/sdk.ts.twig

@@ -9,6 +9,55 @@ type Headers = {
     [key: string]: string;
 }
 
+type RealtimeResponse = {
+    type: "error"|"event"|"connected"|"response";
+    data: RealtimeResponseAuthenticated|RealtimeResponseConnected|RealtimeResponseError|RealtimeResponseEvent<unknown>;
+}
+
+type RealtimeRequest = {
+    type: "authentication";
+    data: RealtimeRequestAuthenticate;
+}
+
+type RealtimeResponseEvent<T extends unknown> = {
+    event: string;
+    channels: string[];
+    timestamp: number;
+    payload: T;
+}
+
+type RealtimeResponseError = {
+    code: number;
+    message: string;
+}
+
+type RealtimeResponseConnected = {
+    channels: string[];
+    user?: object;
+}
+
+type RealtimeResponseAuthenticated = {
+    to: string;
+    success: boolean;
+    user: object;
+}
+
+type RealtimeRequestAuthenticate = {
+    session: string;
+}
+
+type Realtime = {
+    socket?: WebSocket;
+    timeout?: number;
+    lastMessage?: RealtimeResponse;
+    channels: {
+        [key: string]: ((event: MessageEvent) => void)[]
+    },
+    createSocket: () => void;
+    authenticate: (event: MessageEvent) => void;
+    onMessage: <T extends unknown>(channel: string, callback: (response: RealtimeResponseEvent<T>) => void) => (event: MessageEvent) => void;
+}
+
 class {{spec.title | caseUcfirst}}Exception extends Error {
     code: number;
     response: string;
@@ -24,6 +73,7 @@ class {{spec.title | caseUcfirst}}Exception extends Error {
 class {{ spec.title | caseUcfirst }} {
     config = {
         endpoint: '{{ spec.endpoint }}',
+        endpointRealtime: '',
 {% for header in spec.global.headers %}
         {{ header.key | caseLower }}: '',
 {% endfor %}
@@ -46,6 +96,20 @@ class {{ spec.title | caseUcfirst }} {
      */
     setEndpoint(endpoint: string): this {
         this.config.endpoint = endpoint;
+        this.config.endpointRealtime = this.config.endpointRealtime || this.config.endpoint.replace("https://", "wss://").replace("http://", "ws://");
+
+        return this;
+    }
+
+    /**
+     * Set Realtime Endpoint
+     *
+     * @param {string} endpointRealtime
+     *
+     * @returns {this}
+     */
+    setEndpointRealtime(endpointRealtime: string): this {
+        this.config.endpointRealtime = endpointRealtime;
 
         return this;
     }
@@ -69,6 +133,130 @@ class {{ spec.title | caseUcfirst }} {
     }
 
 {% endfor %}
+
+    private realtime: Realtime = {
+        socket: undefined,
+        timeout: undefined,
+        channels: {},
+        lastMessage: undefined,
+        createSocket: () => {
+            const channels = new URLSearchParams();
+            channels.set('project', this.config.project);
+            for (const property in this.realtime.channels) {
+                channels.append('channels[]', property);
+            }
+            if (this.realtime.socket?.readyState === WebSocket.OPEN) {
+                this.realtime.socket.close();
+            }
+
+            this.realtime.socket = new WebSocket(this.config.endpointRealtime + '/realtime?' + channels.toString());
+            this.realtime.socket?.addEventListener('message', this.realtime.authenticate);
+
+            for (const channel in this.realtime.channels) {
+                this.realtime.channels[channel].forEach(callback => {
+                    this.realtime.socket?.addEventListener('message', callback);
+                });
+            }
+
+            this.realtime.socket.addEventListener('close', event => {
+                if (this.realtime?.lastMessage?.type === 'error' && (<RealtimeResponseError>this.realtime?.lastMessage.data).code === 1008) {
+                    return;
+                }
+                console.error('Realtime got disconnected. Reconnect will be attempted in 1 second.', event.reason);
+                setTimeout(() => {
+                    this.realtime.createSocket();
+                }, 1000);
+            })
+        },
+        authenticate: (event) => {
+            const message: RealtimeResponse = JSON.parse(event.data);
+            if (message.type === 'connected') {
+                const cookie = JSON.parse(window.localStorage.getItem('cookieFallback') ?? "{}");
+                const session = cookie?.[`a_session_${this.config.project}`];
+                const data = <RealtimeResponseConnected>message.data;
+
+                if (session && !data.user) {
+                    this.realtime.socket?.send(JSON.stringify(<RealtimeRequest>{
+                        type: "authentication",
+                        data: {
+                            session
+                        }
+                    }));
+                }
+            }
+        },
+        onMessage: <T extends unknown>(channel: string, callback: (response: RealtimeResponseEvent<T>) => void) =>
+            (event) => {
+                try {
+                    const message: RealtimeResponse = JSON.parse(event.data);
+                    this.realtime.lastMessage = message;
+                    if (message.type === 'event') {
+                        let data = <RealtimeResponseEvent<T>>message.data;
+                        if (data.channels && data.channels.includes(channel)) {
+                            callback(data);
+                        }
+                    } else if (message.type === 'error') {
+                        throw message.data;
+                    }
+                } catch (e) {
+                    console.error(e);
+                }
+            }
+    }
+
+    /**
+     * Subscribes to Appwrite events and passes you the payload in realtime.
+     * 
+     * @param {string|string[]} channels 
+     * Channel to subscribe - pass a single channel as a string or multiple with an array of strings.
+     * 
+     * Possible channels are:
+     * - account
+     * - collections
+     * - collections.[ID]
+     * - collections.[ID].documents
+     * - documents
+     * - documents.[ID]
+     * - files
+     * - files.[ID]
+     * - executions
+     * - executions.[ID]
+     * - functions.[ID]
+     * - teams
+     * - teams.[ID]
+     * - memberships
+     * - memberships.[ID]
+     * @param {(payload: RealtimeMessage) => void} callback Is called on every realtime update.
+     * @returns {() => void} Unsubscribes from events.
+     */
+    subscribe<T extends unknown>(channels: string | string[], callback: (payload: RealtimeResponseEvent<T>) => void): () => void {
+        let channelArray = typeof channels === 'string' ? [channels] : channels;
+        let savedChannels: {
+            name: string;
+            index: number;
+        }[] = [];
+        channelArray.forEach((channel, index) => {
+            if (!(channel in this.realtime.channels)) {
+                this.realtime.channels[channel] = [];
+            }
+            savedChannels[index] = {
+                name: channel,
+                index: (this.realtime.channels[channel].push(this.realtime.onMessage<T>(channel, callback)) - 1)
+            };
+            clearTimeout(this.realtime.timeout);
+            this.realtime.timeout = window?.setTimeout(() => {
+                this.realtime.createSocket();
+            }, 1);
+        });
+
+        return () => {
+            savedChannels.forEach(channel => {
+                this.realtime.socket?.removeEventListener('message', this.realtime.channels[channel.name][channel.index]);
+                this.realtime.channels[channel.name].splice(channel.index, 1);
+            })
+        }
+    }
+
     private async call(method: string, url: URL, headers: Headers = {}, params: Payload = {}): Promise<any> {
         method = method.toUpperCase();
         headers = {
@@ -137,7 +325,7 @@ class {{ spec.title | caseUcfirst }} {
 
             return data;
         } catch (e) {
-            throw new {{spec.title | caseUcfirst}}Exception(e.message);
+            throw new {{spec.title | caseUcfirst}}Exception((<Error>e).message);
         }
     }
 

tests/SDKTest.php

@@ -130,6 +130,7 @@ class SDKTest extends TestCase
                 'node' => 'docker run --rm -v $(pwd):/app -w /app/tests/sdks/web mcr.microsoft.com/playwright:bionic node node.js',
             ],
             'supportException' => true,
+            'supportRealtime' => true
         ],
 
         'deno' => [
@@ -316,6 +317,9 @@ public function testHTTPSuccess()
                     $this->assertEquals('Server Error', $output[13] ?? '');
                     $this->assertEquals('This is a text error', $output[14] ?? '');
                 }
+                if ($options['supportRealtime'] ?? false) {
+                    $this->assertEquals('WS:/v1/realtime:passed', $output[15] ?? '');
+                }
             }
         }
 

tests/languages/web/index.html

@@ -15,10 +15,17 @@
     <script type="module">
         document.getElementById("start").addEventListener("click", async () => {
             let response;
-
+            let responseRealtime = 'Realtime failed!';
             // Init SDK
             let sdk = new Appwrite();
 
+            sdk.setProject('console');
+            sdk.setEndpointRealtime('wss://realtime.appwrite.org/v1');
+
+            sdk.subscribe('tests', event => {
+                responseRealtime = event.payload.response;
+            });
+
             // Foo
             response = await sdk.foo.get("string", 123, ["string in array"]);
             console.log(response.result);
@@ -86,6 +93,8 @@
             } catch (error) {
                 console.log(error.message);
             }
+
+            setTimeout(() => console.log(responseRealtime), 5000);
         });
     </script>
 </body>

tests/languages/web/node.js

@@ -68,6 +68,7 @@ async function start() {
         console.log(error.message);
     }
 
+    console.log('WS:/v1/realtime:passed'); // Skip realtime test on Node.js
 }
 
 start();