feat: Base Setup for realtime

.env.local.example

@@ -97,4 +97,7 @@ UNKEY_ROOT_KEY=""
 ############################ NEXT_PUBLIC VARIABLES ############################
 NEXT_PUBLIC_WEBAPP_URL=http://localhost:3000
 NEXT_PUBLIC_STORAGE_URL=http://localhost:3200
-NEXT_PUBLIC_PLATFORM_URL=http://localhost:3300
+NEXT_PUBLIC_PLATFORM_URL=http://localhost:3300
+NEXT_PUBLIC_REALTIME_APP_KEY=secretsecretsecret
+NEXT_PUBLIC_REALTIME_HOST=localhost
+NEXT_PUBLIC_REALTIME_PORT=3904

apps/web/package.json

@@ -38,6 +38,7 @@
     "@trpc/react-query": "10.45.2",
     "@trpc/server": "10.45.2",
     "@u22n/platform": "workspace:^",
+    "@u22n/realtime": "workspace:^",
     "@u22n/tiptap": "workspace:^",
     "@u22n/utils": "workspace:^",
     "@uidotdev/usehooks": "^2.4.1",

apps/web/src/app/[orgShortCode]/convo/layout.tsx

@@ -22,6 +22,8 @@ import {
 import { useState } from 'react';
 import Link from 'next/link';
 import { useGlobalStore } from '@/src/providers/global-store-provider';
+import { useEffect } from 'react';
+import { useRealtime } from '@/src/providers/realtime-provider';
 
 export default function Layout({
   children
@@ -30,6 +32,41 @@ export default function Layout({
   const { setSidebarExpanded } = usePreferencesState();
   const isMobile = useIsMobile();
   const [showHidden, setShowHidden] = useState(false);
+  // TODO: Implement the realtime event handlers and update query cache accordingly
+
+  const client = useRealtime();
+
+  useEffect(() => {
+    client.on('convo:new', async ({ publicId }) => {
+      //TODO: Handle new convo added
+      console.info('New convo added', publicId);
+    });
+
+    client.on('convo:hidden', async ({ publicId }) => {
+      // TODO: Handle convo updated
+      console.info('Convo Hidden', publicId);
+    });
+
+    client.on('convo:deleted', async ({ publicId }) => {
+      // TODO: Handle convo deleted
+      console.info('Convo Delete', publicId);
+    });
+
+    client.on(
+      'convo:entry:new',
+      async ({ convoPublicId, convoEntryPublicId }) => {
+        // TODO: Handle new convo entry
+        console.info('New convo entry', convoPublicId, convoEntryPublicId);
+      }
+    );
+
+    return () => {
+      client.off('convo:new');
+      client.off('convo:hidden');
+      client.off('convo:deleted');
+      client.off('convo:entry:new');
+    };
+  }, [client]);
 
   return (
     <div className="grid h-full w-full grid-cols-3 gap-0">

apps/web/src/app/[orgShortCode]/layout.tsx

@@ -6,6 +6,7 @@ import { buttonVariants } from '@/src/components/shadcn-ui/button';
 import { SpinnerGap } from '@phosphor-icons/react';
 import Link from 'next/link';
 import { api } from '@/src/lib/trpc';
+import { RealtimeProvider } from '@/src/providers/realtime-provider';
 
 export default function Layout({
   children,
@@ -68,7 +69,9 @@ export default function Layout({
         <div className="h-full max-h-full w-fit">
           <Sidebar />
         </div>
-        <div className="flex h-full w-full flex-row p-0">{children}</div>
+        <div className="flex h-full w-full flex-row p-0">
+          <RealtimeProvider>{children}</RealtimeProvider>
+        </div>
       </div>
     </GlobalStoreProvider>
   );

apps/web/src/providers/realtime-provider.tsx

@@ -0,0 +1,58 @@
+import {
+  createContext,
+  useContext,
+  useEffect,
+  useMemo,
+  type PropsWithChildren
+} from 'react';
+import RealtimeClient from '@u22n/realtime/client';
+import { useGlobalStore } from './global-store-provider';
+import { env } from 'next-runtime-env';
+import { toast } from 'sonner';
+
+const realtimeContext = createContext<RealtimeClient | null>(null);
+
+const appKey = env('NEXT_PUBLIC_REALTIME_APP_KEY')!;
+const host = env('NEXT_PUBLIC_REALTIME_HOST')!;
+const port = Number(env('NEXT_PUBLIC_REALTIME_PORT')!);
+const PLATFORM_URL = env('NEXT_PUBLIC_PLATFORM_URL')!;
+
+export function RealtimeProvider({ children }: PropsWithChildren) {
+  const orgShortCode = useGlobalStore((state) => state.currentOrg.shortCode);
+
+  const client = useMemo(
+    () =>
+      new RealtimeClient({
+        appKey,
+        host,
+        port,
+        authEndpoint: `${PLATFORM_URL}/realtime/auth`
+      }),
+    []
+  );
+
+  useEffect(() => {
+    void client.connect({ orgShortCode }).catch(() => {
+      toast.error(
+        'Uninbox encountered an error while trying to connect to the realtime server'
+      );
+    });
+    return () => {
+      client.disconnect();
+    };
+  }, [client, orgShortCode]);
+
+  return (
+    <realtimeContext.Provider value={client}>
+      {children}
+    </realtimeContext.Provider>
+  );
+}
+
+export function useRealtime() {
+  const client = useContext(realtimeContext);
+  if (!client) {
+    throw new Error('useRealtime must be used within RealtimeProvider');
+  }
+  return client;
+}

packages/realtime/client.ts

@@ -4,6 +4,10 @@ import type { z } from 'zod';
 
 export default class RealtimeClient {
   private client: Pusher | null = null;
+  #preConnectEventHandlers = new Map<keyof EventDataMap, Function[]>();
+  #preConnectBroadcastHandlers = new Map<keyof EventDataMap, Function[]>();
+  #connectionTimeout: NodeJS.Timeout | null = null;
+
   constructor(
     private config: {
       appKey: string;
@@ -12,6 +16,7 @@ export default class RealtimeClient {
       authEndpoint: string;
     }
   ) {}
+
   public async connect({ orgShortCode }: { orgShortCode: string }) {
     if (this.client) return;
     const client = new Pusher(this.config.appKey, {
@@ -44,13 +49,22 @@ export default class RealtimeClient {
 
     client.signin();
     this.client = client;
+    this.bindEvents();
     return new Promise<void>((resolve, reject) => {
+      this.#connectionTimeout = setTimeout(() => {
+        this.client = null;
+        reject(new Error('Connection timeout'));
+      }, 10000);
+
       client.bind('pusher:signin_success', () => {
         client.unbind('pusher:signin_success');
+        if (this.#connectionTimeout) clearTimeout(this.#connectionTimeout);
         resolve();
       });
+
       client.bind('pusher:error', (e: unknown) => {
         this.client = null;
+        if (this.#connectionTimeout) clearTimeout(this.#connectionTimeout);
         reject(e);
       });
     });
@@ -59,6 +73,7 @@ export default class RealtimeClient {
   public disconnect() {
     if (this.client) {
       this.client.disconnect();
+      if (this.#connectionTimeout) clearTimeout(this.#connectionTimeout);
       this.client = null;
     }
   }
@@ -67,24 +82,64 @@ export default class RealtimeClient {
     event: T,
     callback: (data: z.infer<EventDataMap[T]>) => Promise<void>
   ) {
-    if (!this.client) return;
-    this.client.bind(event, (e: unknown) =>
-      callback(eventDataMaps[event].parse(e))
-    );
+    if (!this.client) {
+      const existing = this.#preConnectEventHandlers.get(event) ?? [];
+      existing.push(callback);
+      this.#preConnectEventHandlers.set(event, existing);
+    } else {
+      this.client.bind(event, (e: unknown) =>
+        callback(eventDataMaps[event].parse(e))
+      );
+    }
   }
 
   public off<const T extends keyof EventDataMap>(event: T) {
-    if (!this.client) return;
-    this.client.unbind(event);
+    if (!this.client) {
+      // eslint-disable-next-line drizzle/enforce-delete-with-where
+      this.#preConnectEventHandlers.delete(event);
+    } else {
+      this.client.unbind(event);
+    }
   }
 
   public onBroadcast<const T extends keyof EventDataMap>(
     event: T,
     callback: (data: z.infer<EventDataMap[T]>) => Promise<void>
   ) {
+    if (!this.client) {
+      const existing = this.#preConnectBroadcastHandlers.get(event) ?? [];
+      existing.push(callback);
+      this.#preConnectBroadcastHandlers.set(event, existing);
+    } else {
+      this.client
+        .subscribe('broadcasts')
+        .bind(event, (e: unknown) => callback(eventDataMaps[event].parse(e)));
+    }
+  }
+
+  public get isConnected() {
+    return !!this.client;
+  }
+
+  private bindEvents() {
     if (!this.client) return;
-    this.client
-      .subscribe('broadcasts')
-      .bind(event, (e: unknown) => callback(eventDataMaps[event].parse(e)));
+    for (const [event, handlers] of this.#preConnectEventHandlers) {
+      handlers.forEach((handler) => {
+        if (!this.client) return;
+        this.client.bind(event, (e: unknown) =>
+          handler(eventDataMaps[event].parse(e))
+        );
+      });
+    }
+    for (const [event, handlers] of this.#preConnectBroadcastHandlers) {
+      handlers.forEach((handler) => {
+        if (!this.client) return;
+        this.client
+          .subscribe('broadcasts')
+          .bind(event, (e: unknown) => handler(eventDataMaps[event].parse(e)));
+      });
+    }
+    this.#preConnectEventHandlers.clear();
+    this.#preConnectBroadcastHandlers.clear();
   }
 }