feat(realtime): add support to configure Broadcast Replay (#1623)

Author: edgurgel

packages/core/realtime-js/README.md

@@ -80,6 +80,7 @@ channel.subscribe((status, err) => {
 - `REALTIME_URL` is `'ws://localhost:4000/socket'` when developing locally and `'wss://<project_ref>.supabase.co/realtime/v1'` when connecting to your Supabase project.
 - `API_KEY` is a JWT whose claims must contain `exp` and `role` (existing database role).
 - Channel name can be any `string`.
+- Setting `private` to `true` means that the client will use RLS to determine if the user can connect or not to a given channel.
 
 ## Broadcast
 
@@ -106,9 +107,38 @@ channel.subscribe(async (status) => {
 
 ### Notes:
 
-- Setting `ack` to `true` means that the `channel.send` promise will resolve once server replies with acknowledgement that it received the broadcast message request.
+- Setting `ack` to `true` means that the `channel.send` promise will resolve once server replies with acknowledgment that it received the broadcast message request.
 - Setting `self` to `true` means that the client will receive the broadcast message it sent out.
-- Setting `private` to `true` means that the client will use RLS to determine if the user can connect or not to a given channel.
+
+### Broadcast Replay
+
+Broadcast Replay enables **private** channels to access messages that were sent earlier. Only messages published via [Broadcast From the Database](https://supabase.com/docs/guides/realtime/broadcast#trigger-broadcast-messages-from-your-database) are available for replay.
+
+You can configure replay with the following options:
+
+- **`since`** (Required): The epoch timestamp in milliseconds, specifying the earliest point from which messages should be retrieved.
+- **`limit`** (Optional): The number of messages to return. This must be a positive integer, with a maximum value of 25.
+
+Example:
+
+```typescript
+const twelveHours = 12 * 60 * 60 * 1000
+const twelveHoursAgo = Date.now() - twelveHours
+
+const config = { private: true, broadcast: { replay: { since: twelveHoursAgo, limit: 10 } } }
+
+supabase
+  .channel('main:room', { config })
+  .on('broadcast', { event: 'my_event' }, (payload) => {
+    if (payload?.meta?.replayed) {
+      console.log('This message was sent earlier:', payload)
+    } else {
+      console.log('This is a new message', payload)
+    }
+    // ...
+  })
+  .subscribe()
+```
 
 ## Presence
 

packages/core/realtime-js/src/RealtimeChannel.ts

@@ -11,13 +11,19 @@ import type {
 import * as Transformers from './lib/transformers'
 import { httpEndpointURL } from './lib/transformers'
 
+type ReplayOption = {
+  since: number
+  limit?: number
+}
+
 export type RealtimeChannelOptions = {
   config: {
     /**
      * self option enables client to receive message it broadcast
      * ack option instructs server to acknowledge that broadcast message was received
+     * replay option instructs server to replay broadcast messages
      */
-    broadcast?: { self?: boolean; ack?: boolean }
+    broadcast?: { self?: boolean; ack?: boolean; replay?: ReplayOption }
     /**
      * key option is used to track presence payload across clients
      */
@@ -203,6 +209,10 @@ export default class RealtimeChannel {
 
     this.broadcastEndpointURL = httpEndpointURL(this.socket.endPoint)
     this.private = this.params.config.private || false
+
+    if (!this.private && this.params.config?.broadcast?.replay) {
+      throw `tried to use replay on public channel '${this.topic}'. It must be a private channel.`
+    }
   }
 
   /** Subscribe registers your client with the server */
@@ -386,6 +396,10 @@ export default class RealtimeChannel {
     callback: (payload: {
       type: `${REALTIME_LISTEN_TYPES.BROADCAST}`
       event: string
+      meta?: {
+        replayed?: boolean
+        id: string
+      }
       [key: string]: any
     }) => void
   ): RealtimeChannel
@@ -395,6 +409,10 @@ export default class RealtimeChannel {
     callback: (payload: {
       type: `${REALTIME_LISTEN_TYPES.BROADCAST}`
       event: string
+      meta?: {
+        replayed?: boolean
+        id: string
+      }
       payload: T
     }) => void
   ): RealtimeChannel

packages/core/realtime-js/test/RealtimeClient.channels.test.ts

@@ -14,6 +14,31 @@ afterEach(() => {
 
 describe('channel', () => {
   let channel
+  test('throws error on replay for a public channel', () => {
+    expect(() => {
+      channel = testSetup.socket.channel('topic', {
+        config: { broadcast: { replay: { since: 0 } } },
+      })
+    }).toThrow(
+      "tried to use replay on public channel 'realtime:topic'. It must be a private channel."
+    )
+  })
+
+  test('returns channel with private channel using replay', () => {
+    channel = testSetup.socket.channel('topic', {
+      config: { private: true, broadcast: { replay: { since: 0 } } },
+    })
+
+    assert.deepStrictEqual(channel.socket, testSetup.socket)
+    assert.equal(channel.topic, 'realtime:topic')
+    assert.deepEqual(channel.params, {
+      config: {
+        broadcast: { replay: { since: 0 } },
+        presence: { key: '', enabled: false },
+        private: true,
+      },
+    })
+  })
 
   test('returns channel with given topic and params', () => {
     channel = testSetup.socket.channel('topic')