feat(realtime): add client event support for Pusher and Firebase transports

AGENTS.md

@@ -2,7 +2,7 @@
 
 This file is for AI agents working inside applications scaffolded by Arkstack. It assumes the agent has access to the generated project only, not to the Arkstack framework monorepo.
 
-Read `SKILLS.md` first, then use these workflows to combine the available skills safely.
+Read `SKILL.md` first, then use these workflows to combine the available skills safely.
 
 ## Operating Rules
 

packages/realtime/README.md

@@ -44,6 +44,28 @@ unsubscribe();
 await realtime.disconnect();
 ```
 
+### Client events
+
+Pusher private and presence channels can send ephemeral client events (often
+called whispers). Subscribe to the channel before sending an event:
+
+```ts
+const stopTyping = await realtime.listenForWhisper(
+  'private-room.7',
+  'typing',
+  ({ userId }) => console.log(`${userId} is typing`),
+);
+
+await realtime.whisper('private-room.7', 'typing', { userId: user.id });
+stopTyping();
+```
+
+The `client-` prefix is added automatically. Pusher sends client events over its
+private/presence channel. Firebase uses Realtime Database to publish the same
+ephemeral events across connected clients. Use `listen(channel, event, handler)`
+and `trigger(channel, event, payload)` when you need the lower-level APIs with
+exact event names.
+
 Each `notification` matches the payload broadcast by the server:
 
 ```ts
@@ -115,10 +137,16 @@ const realtime = createRealtime({
     projectId: import.meta.env.VITE_FIREBASE_PROJECT_ID,
     appId: import.meta.env.VITE_FIREBASE_APP_ID,
     messagingSenderId: import.meta.env.VITE_FIREBASE_SENDER_ID,
+    databaseURL: import.meta.env.VITE_FIREBASE_DATABASE_URL,
   },
 });
 ```
 
+Firebase client events are written below `arkstack/client-events` in Realtime
+Database and removed immediately after publishing. Configure Firebase Security
+Rules for the channels your authenticated clients may read and write. Override
+the root with `firebase.clientEventsPath` when needed.
+
 ## Custom transport
 
 Provide `transportFactory` to bridge any backend (a raw WebSocket, SSE, a test double, …):
@@ -146,6 +174,9 @@ const realtime = createRealtime({ transportFactory: () => transport });
 
 - `createRealtime(config)` — create a `RealtimeClient`. Config: `transport` (`'pusher'` | `'firebase'`), `event` (default `notification`), `channelPrefix` (default `user.`), `pusher`/`firebase` credentials, or a custom `transportFactory`.
 - `client.subscribe(channel, handler)` / `client.forUser(userId, handler)` — subscribe; returns an unsubscribe function.
+- `client.listen(channel, event, handler)` — listen for an arbitrary event.
+- `client.listenForWhisper(channel, event, handler)` / `client.whisper(channel, event, payload)` — receive and send Pusher client events.
+- `client.trigger(channel, event, payload)` — emit an exact event name through a transport that supports it.
 - `client.channelFor(userId)` — the per-user channel name.
 - `client.disconnect()` — tear down the transport connection.
 - `@arkstack/realtime/react` — `useNotifications(client, channel, { limit? })` → `{ notifications, latest, clear }`.

packages/realtime/src/RealtimeClient.ts

@@ -1,4 +1,4 @@
-import type { NotificationHandler, RealtimeConfig, RealtimeSubscription, RealtimeTransport } from './types'
+import type { NotificationHandler, RealtimeConfig, RealtimeEventHandler, RealtimeSubscription, RealtimeTransport } from './types'
 
 /**
  * Consumes Arkstack realtime notifications. Resolves a transport (Pusher,
@@ -15,18 +15,23 @@ export class RealtimeClient {
         this.channelPrefix = config.channelPrefix ?? 'user.'
     }
 
-    /** The channel name a given user's notifications are broadcast on. */
-    channelFor (userId: string | number): string {
+    /**
+     * The channel name a given user's notifications are broadcast on.
+     * 
+     * @param userId 
+     * @returns 
+     */
+    channelFor(userId: string | number): string {
         return `${this.channelPrefix}${userId}`
     }
 
-    private transport (): Promise<RealtimeTransport> {
+    private transport(): Promise<RealtimeTransport> {
         this.transportPromise ??= this.resolveTransport()
 
         return this.transportPromise
     }
 
-    private async resolveTransport (): Promise<RealtimeTransport> {
+    private async resolveTransport(): Promise<RealtimeTransport> {
         if (this.config.transportFactory) {
             return await this.config.transportFactory()
         }
@@ -56,25 +61,93 @@ export class RealtimeClient {
      * @param channel  The channel name (e.g. `user.7`).
      * @param handler  Called with each incoming notification.
      */
-    async subscribe (channel: string, handler: NotificationHandler): Promise<() => void> {
+    async subscribe(channel: string, handler: NotificationHandler): Promise<() => void> {
+        return await this.listen(channel, this.event, handler as RealtimeEventHandler)
+    }
+
+    /**
+     * Listen for an arbitrary event on a channel.
+     *
+     * @param channel
+     * @param event
+     * @param handler
+     * @returns
+     */
+    async listen<Payload = unknown>(
+        channel: string,
+        event: string,
+        handler: RealtimeEventHandler<Payload>,
+    ): Promise<() => void> {
         const transport = await this.transport()
-        const subscription: RealtimeSubscription = await transport.subscribe(channel, this.event, handler)
+        const subscription: RealtimeSubscription = await transport.subscribe(
+            channel,
+            event,
+            handler as RealtimeEventHandler,
+        )
 
         return () => subscription.unsubscribe()
     }
 
+    /**
+     * Listen for a client event (`client-{event}`).
+     *
+     * @param channel
+     * @param event
+     * @param handler
+     * @returns
+     */
+    async listenForWhisper<Payload = unknown>(
+        channel: string,
+        event: string,
+        handler: RealtimeEventHandler<Payload>,
+    ): Promise<() => void> {
+        return await this.listen(channel, this.clientEventName(event), handler)
+    }
+
+    /**
+     * Emit a client event (`client-{event}`) on a subscribed channel.
+     *
+     * @param channel
+     * @param event
+     * @param payload
+     */
+    async whisper<Payload = unknown>(channel: string, event: string, payload: Payload): Promise<void> {
+        await this.trigger(channel, this.clientEventName(event), payload)
+    }
+
+    /**
+     * Emit an event through a transport that supports client-originated events.
+     *
+     * @param channel
+     * @param event
+     * @param payload
+     */
+    async trigger<Payload = unknown>(channel: string, event: string, payload: Payload): Promise<void> {
+        const transport = await this.transport()
+
+        if (!transport.trigger) {
+            throw new Error('Realtime: the configured transport does not support client events')
+        }
+
+        await transport.trigger(channel, event, payload)
+    }
+
+    private clientEventName(event: string): string {
+        return event.startsWith('client-') ? event : `client-${event}`
+    }
+
     /**
      * Subscribe to a user's channel (`{channelPrefix}{userId}`).
      *
      * @param userId   The user id.
      * @param handler  Called with each incoming notification.
      */
-    async forUser (userId: string | number, handler: NotificationHandler): Promise<() => void> {
+    async forUser(userId: string | number, handler: NotificationHandler): Promise<() => void> {
         return await this.subscribe(this.channelFor(userId), handler)
     }
 
     /** Tear down the underlying transport connection. */
-    async disconnect (): Promise<void> {
+    async disconnect(): Promise<void> {
         if (!this.transportPromise) {
             return
         }
@@ -85,5 +158,7 @@ export class RealtimeClient {
     }
 }
 
-/** Create a {@link RealtimeClient}. */
-export const createRealtime = (config: RealtimeConfig = {}): RealtimeClient => new RealtimeClient(config)
+/** Create a new {@link RealtimeClient}. */
+export const createRealtime = (
+    config: RealtimeConfig = {}
+): RealtimeClient => new RealtimeClient(config)

packages/realtime/src/transports/firebase.ts

@@ -1,4 +1,4 @@
-import type { FirebaseClientConfig, NotificationHandler, RealtimeTransport } from '../types'
+import type { FirebaseClientConfig, RealtimeEventHandler, RealtimeTransport } from '../types'
 
 /** The slice of `firebase/messaging` this transport uses. */
 interface FirebaseMessagePayload {
@@ -7,6 +7,33 @@ interface FirebaseMessagePayload {
 
 type OnMessage = (messaging: unknown, next: (payload: FirebaseMessagePayload) => void) => () => void
 
+interface DatabaseReference {
+    readonly key?: string | null
+}
+
+interface DatabaseSnapshot {
+    val(): unknown
+}
+
+interface FirebaseClientEvent {
+    sender: string
+    payload: unknown
+}
+
+interface FirebaseDatabaseModule {
+    getDatabase(app: unknown, url?: string): unknown
+    ref(database: unknown, path: string): DatabaseReference
+    onChildAdded(reference: DatabaseReference, handler: (snapshot: DatabaseSnapshot) => void): () => void
+    push(reference: DatabaseReference, payload: unknown): Promise<DatabaseReference>
+    remove(reference: DatabaseReference): Promise<void>
+}
+
+const pathSegment = (value: string): string => encodeURIComponent(value).replace(/\./g, '%2E')
+
+const isFirebaseClientEvent = (value: unknown): value is FirebaseClientEvent => {
+    return typeof value === 'object' && value !== null && 'sender' in value && 'payload' in value
+}
+
 /**
  * Realtime transport backed by [Firebase Cloud Messaging](https://firebase.google.com/docs/cloud-messaging/js/receive)
  * foreground messages. `firebase` is an optional peer dependency imported lazily.
@@ -17,10 +44,12 @@ type OnMessage = (messaging: unknown, next: (payload: FirebaseMessagePayload) =>
 export const createFirebaseTransport = async (config: FirebaseClientConfig): Promise<RealtimeTransport> => {
     const appSpecifier = 'firebase/app'
     const messagingSpecifier = 'firebase/messaging'
+    const databaseSpecifier = 'firebase/database'
 
-    const [appMod, messagingMod] = await Promise.all([
+    const [appMod, messagingMod, databaseMod] = await Promise.all([
         import(appSpecifier),
         import(messagingSpecifier),
+        import(databaseSpecifier),
     ]).catch(() => {
         throw new Error(
             'The "firebase" package is required for the Firebase transport. Install it with `npm i firebase`.',
@@ -32,13 +61,25 @@ export const createFirebaseTransport = async (config: FirebaseClientConfig): Pro
         projectId: config.projectId,
         appId: config.appId,
         messagingSenderId: config.messagingSenderId,
+        databaseURL: config.databaseURL,
     })
 
     const messaging = messagingMod.getMessaging(app)
     const onMessage = messagingMod.onMessage as OnMessage
+    const messageUnsubscribers = new Set<() => void>()
+    const databaseUnsubscribers = new Set<() => void>()
+    const databaseApi = databaseMod as unknown as FirebaseDatabaseModule
+    const database = databaseApi.getDatabase(app, config.databaseURL)
+    const clientEventsPath = config.clientEventsPath ?? 'arkstack/client-events'
+    const clientId = globalThis.crypto?.randomUUID?.() ?? `${Date.now()}-${Math.random()}`
+
+    const clientEventReference = (channel: string, event: string): DatabaseReference => databaseApi.ref(
+        database,
+        `${clientEventsPath}/${pathSegment(channel)}/${pathSegment(event)}`,
+    )
 
     const transport: RealtimeTransport = {
-        subscribe(channel: string, event: string, handler: NotificationHandler) {
+        subscribe(channel: string, event: string, handler: RealtimeEventHandler) {
             const off = onMessage(messaging, (payload) => {
                 if (payload.data?.event !== event || !payload.data.payload) {
                     return
@@ -50,10 +91,47 @@ export const createFirebaseTransport = async (config: FirebaseClientConfig): Pro
                     /** Ignore malformed payloads. */
                 }
             })
+            messageUnsubscribers.add(off)
+            const offClientEvent = event.startsWith('client-')
+                ? databaseApi.onChildAdded(clientEventReference(channel, event), (snapshot) => {
+                    const clientEvent = snapshot.val()
+
+                    if (isFirebaseClientEvent(clientEvent) && clientEvent.sender !== clientId) {
+                        handler(clientEvent.payload)
+                    }
+                })
+                : undefined
+
+            if (offClientEvent) {
+                databaseUnsubscribers.add(offClientEvent)
+            }
 
-            return { channel, unsubscribe: off }
+            return {
+                channel,
+                unsubscribe() {
+                    off()
+                    messageUnsubscribers.delete(off)
+                    if (offClientEvent) {
+                        offClientEvent()
+                        databaseUnsubscribers.delete(offClientEvent)
+                    }
+                },
+            }
+        },
+        async trigger(channel: string, event: string, payload: unknown) {
+            const eventReference = await databaseApi.push(clientEventReference(channel, event), {
+                sender: clientId,
+                payload,
+            })
+
+            await databaseApi.remove(eventReference)
+        },
+        disconnect() {
+            messageUnsubscribers.forEach((off) => off())
+            messageUnsubscribers.clear()
+            databaseUnsubscribers.forEach((off) => off())
+            databaseUnsubscribers.clear()
         },
-        disconnect() { },
     }
 
     return transport