Feat/add realtime api url config property and auth and register auth route notification methods

packages/notifications/package.json

@@ -45,11 +45,19 @@
     "peerDependencies": {
         "@arkstack/contract": "workspace:^",
         "@arkstack/database": "workspace:^",
+        "@arkstack/driver-express": "workspace:^",
+        "@arkstack/driver-h3": "workspace:^",
         "@kanun-hq/plugin-phone": "catalog:",
         "firebase-admin": "^13.0.0",
         "pusher": "^5.2.0"
     },
     "peerDependenciesMeta": {
+        "@arkstack/driver-express": {
+            "optional": true
+        },
+        "@arkstack/driver-h3": {
+            "optional": true
+        },
         "pusher": {
             "optional": true
         },
@@ -64,4 +72,4 @@
         "@types/africastalking": "^0.8.0",
         "@types/nodemailer": "^7.0.11"
     }
-}
+}

packages/notifications/src/Contracts/RealtimeDriver.ts

@@ -17,6 +17,11 @@ export interface RealtimeDriver {
         event: string,
         payload: RealtimeNotificationPayload,
     ): Promise<unknown>
+    auth?(
+        socketId: string,
+        channel: string,
+        data?: unknown
+    ): unknown | Promise<unknown>
 }
 
 export type RealtimeNotificationDriver<T extends RealtimeDriverName> = T extends 'firebase'

packages/notifications/src/drivers/realtime/PusherRealtimeDriver.ts

@@ -6,6 +6,20 @@ import { env } from '@arkstack/common'
 /** The slice of the `pusher` server SDK this driver uses. */
 interface PusherClient {
     trigger(channel: string | string[], event: string, data: unknown): Promise<unknown>
+    authorizeChannel(
+        socketId: string,
+        channel: string,
+        data?: {
+            user_id: string
+            user_info?: {
+                [key: string]: any
+            }
+        }
+    ): {
+        auth: string
+        channel_data?: string
+        shared_secret?: string
+    }
 }
 
 type PusherConstructor = new (options: {
@@ -56,4 +70,64 @@ export class PusherRealtimeDriver implements RealtimeDriver {
         // Pusher's `trigger` fans out to multiple channels when given an array.
         return await client.trigger(channel, event, payload)
     }
+
+    /**
+     * Authourize a pusher channel
+     * 
+     * @param socketId 
+     * @param channel 
+     * @returns 
+     */
+    async auth(
+        socketId: string,
+        channel: string,
+        data?: Parameters<PusherClient['authorizeChannel']>[2]
+    ): Promise<ReturnType<PusherClient['authorizeChannel']>> {
+        const client = await this.client()
+
+        return client.authorizeChannel(socketId, channel, data)
+    }
+
+    /**
+     * Register a realtime autorization route
+     * 
+     * @param authEndpoint 
+     * @param middleware 
+     */
+    async registerAuthRoute(
+        authEndpoint: string = '/realtime/auth',
+        middleware?: unknown | unknown[],
+    ): Promise<void> {
+        const client = await this.client()
+        const drivers = {
+            express: '@arkstack/driver-express',
+            h3: '@arkstack/driver-express'
+        }
+
+        for (const [name, path] of Object.entries(drivers)) {
+            const midsPath = `${path}/middlewares`
+            try {
+                const { Router } = await import(path)
+                const { auth } = await import(midsPath)
+
+                const middlewares = new Set(Array.isArray(middleware)
+                    ? middleware.concat(auth)
+                    : [middleware, auth]
+                )
+
+                Router.post(authEndpoint, ({ clearRequest }: any) => {
+                    const channel = clearRequest.input('channel_name')
+                    const socketId = clearRequest.input('socket_id')
+
+                    return client.authorizeChannel(socketId, channel)
+                }).middleware(Array.from(middlewares))
+
+                break
+            } catch (e) {
+                console.error(
+                    `Failed to register auth route for ${name}: ${e instanceof Error ? e.message : e}`
+                )
+            }
+        }
+    }
 }

packages/realtime/src/transports/pusher.ts

@@ -17,9 +17,14 @@ type PusherConstructor = new (key: string, options: Record<string, unknown>) =>
 /**
  * Realtime transport backed by [pusher-js](https://github.com/pusher/pusher-js).
  * The SDK is an optional peer dependency imported lazily, so consumers only pull
- * it in when they use the Pusher transport.
+ * it in when they use the Pusher transport. 
+ * 
+ * @param config 
+ * @returns 
  */
-export const createPusherTransport = async (config: PusherClientConfig): Promise<RealtimeTransport> => {
+export const createPusherTransport = async (
+    config: PusherClientConfig
+): Promise<RealtimeTransport> => {
     const specifier = 'pusher-js'
     const mod = await import(specifier).catch(() => {
         throw new Error(
@@ -31,7 +36,7 @@ export const createPusherTransport = async (config: PusherClientConfig): Promise
     const client = new Pusher(config.key, {
         cluster: config.cluster ?? 'mt1',
         forceTLS: config.forceTLS ?? true,
-        authEndpoint: config.authEndpoint,
+        authEndpoint: config.authEndpoint ?? `${config.apiBase}/realtime/auth`,
         auth: config.auth,
     })
 

packages/realtime/src/types.ts

@@ -22,7 +22,7 @@ export type NotificationHandler = (notification: RealtimeNotification) => void
 /** A live subscription to one channel; call `unsubscribe()` to stop listening. */
 export interface RealtimeSubscription {
     channel: string
-    unsubscribe (): void
+    unsubscribe(): void
 }
 
 /**
@@ -31,18 +31,29 @@ export interface RealtimeSubscription {
  * supplied via {@link RealtimeConfig.transportFactory} for custom backends/tests.
  */
 export interface RealtimeTransport {
-    subscribe (
+    subscribe(
         channel: string,
         event: string,
         handler: NotificationHandler,
     ): RealtimeSubscription | Promise<RealtimeSubscription>
-    disconnect (): void | Promise<void>
+    disconnect(): void | Promise<void>
 }
 
 export interface PusherClientConfig {
     key: string
     cluster?: string
-    /** Endpoint that authorizes private/presence channels. */
+    /** 
+     * Your API's base URL, if provisioned, private/presence channels 
+     * will be automatically authorized.
+     * 
+     * Will be ignored if {@link authEndpoint} is provisioned.
+     */
+    apiBase?: string
+    /** 
+     * Endpoint that authorizes private/presence channels. 
+     * 
+     * If provisioned, {@link apiBase} will be ignored.
+     */
     authEndpoint?: string
     auth?: { headers?: Record<string, string>, params?: Record<string, string> }
     forceTLS?: boolean