fix(realtime): subscribed state

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

@@ -179,6 +179,10 @@ export default class RealtimeChannel {
   broadcastEndpointURL: string
   private: boolean
   presence: RealtimePresence
+  private _postgresChangesSystemReady = false
+  private _subscribeCallback: ((status: REALTIME_SUBSCRIBE_STATES, err?: Error) => void) | null =
+    null
+  private _subscribeTimeoutId: ReturnType<typeof setTimeout> | null = null
   /** @internal */
   channelAdapter: ChannelAdapter
 
@@ -253,8 +257,39 @@ export default class RealtimeChannel {
 
     this.channelAdapter = new ChannelAdapter(this.socket.socketAdapter, topic, this.params)
     this.presence = new RealtimePresence(this)
+    this.channelAdapter.on(REALTIME_LISTEN_TYPES.SYSTEM, (payload: unknown) => {
+      if (
+        !payload ||
+        typeof payload !== 'object' ||
+        !('extension' in payload) ||
+        payload.extension !== REALTIME_LISTEN_TYPES.POSTGRES_CHANGES ||
+        !('status' in payload)
+      ) {
+        return
+      }
+
+      if (payload.status !== 'ok') {
+        this._failSubscribeState(
+          REALTIME_SUBSCRIBE_STATES.CHANNEL_ERROR,
+          new Error(
+            'message' in payload && typeof payload.message === 'string'
+              ? payload.message
+              : `postgres_changes system status: ${String(payload.status)}`
+          )
+        )
+        return
+      }
+
+      this._postgresChangesSystemReady = true
+      this._emitSubscribed()
+    })
+
+    this._onError(() => {
+      this._resetSubscribeState()
+    })
 
     this._onClose(() => {
+      this._resetSubscribeState()
       this.socket._remove(this)
     })
 
@@ -277,6 +312,10 @@ export default class RealtimeChannel {
    * Log the full `err` so its `cause`, `name`, and any structured fields aren't hidden
    * behind `err.message`.
    *
+   * For channels that only bind `postgres_changes` callbacks, `SUBSCRIBED` is emitted after the
+   * server confirms the Postgres changefeed is ready. Channels with `broadcast` or `presence`
+   * callbacks keep the existing `SUBSCRIBED` timing.
+   *
    * @category Realtime
    *
    * @example Handling errors
@@ -297,6 +336,7 @@ export default class RealtimeChannel {
       this.socket.connect()
     }
     if (this.channelAdapter.isClosed()) {
+      this._resetSubscribeState()
       const {
         config: { broadcast, presence, private: isPrivate },
       } = this.params
@@ -320,6 +360,7 @@ export default class RealtimeChannel {
       }
 
       this._onError((reason: unknown) => {
+        this._resetSubscribeState()
         callback?.(REALTIME_SUBSCRIBE_STATES.CHANNEL_ERROR, normalizeChannelError(reason))
       })
 
@@ -341,14 +382,16 @@ export default class RealtimeChannel {
             return
           }
 
-          this._updatePostgresBindings(postgres_changes, callback)
+          this._updatePostgresBindings(postgres_changes, callback, timeout)
         })
         .receive('error', (error: { [key: string]: any }) => {
+          this._resetSubscribeState()
           this.state = CHANNEL_STATES.errored
           const message = Object.values(error).join(', ') || 'error'
           callback?.(REALTIME_SUBSCRIBE_STATES.CHANNEL_ERROR, new Error(message, { cause: error }))
         })
         .receive('timeout', () => {
+          this._resetSubscribeState()
           callback?.(REALTIME_SUBSCRIBE_STATES.TIMED_OUT)
         })
     }
@@ -357,7 +400,8 @@ export default class RealtimeChannel {
 
   private _updatePostgresBindings(
     postgres_changes: PostgresChangesFilters['postgres_changes'],
-    callback?: (status: REALTIME_SUBSCRIBE_STATES, err?: Error) => void
+    callback?: (status: REALTIME_SUBSCRIBE_STATES, err?: Error) => void,
+    timeout = this.timeout
   ) {
     const clientPostgresBindings = this.bindings.postgres_changes
     const bindingsLen = clientPostgresBindings?.length ?? 0
@@ -382,6 +426,7 @@ export default class RealtimeChannel {
           id: serverPostgresFilter.id,
         })
       } else {
+        this._resetSubscribeState()
         this.unsubscribe()
         this.state = CHANNEL_STATES.errored
 
@@ -396,7 +441,7 @@ export default class RealtimeChannel {
     this.bindings.postgres_changes = newPostgresBindings
 
     if (this.state != CHANNEL_STATES.errored && callback) {
-      callback(REALTIME_SUBSCRIBE_STATES.SUBSCRIBED)
+      this._onSubscribeOk(callback, timeout)
     }
   }
 
@@ -1121,6 +1166,68 @@ export default class RealtimeChannel {
     return normalizedServer === normalizedClient
   }
 
+  private _onSubscribeOk(
+    callback: (status: REALTIME_SUBSCRIBE_STATES, err?: Error) => void,
+    timeout = this.timeout
+  ) {
+    if (!this._shouldWaitForPostgresChangesSystem()) {
+      callback(REALTIME_SUBSCRIBE_STATES.SUBSCRIBED)
+      return
+    }
+
+    this._subscribeCallback = callback
+    this._startSubscribeTimeout(timeout)
+    this._emitSubscribed()
+  }
+
+  private _emitSubscribed() {
+    if (!this._subscribeCallback || !this._postgresChangesSystemReady) {
+      return
+    }
+
+    const callback = this._subscribeCallback
+    this._resetSubscribeState()
+    callback?.(REALTIME_SUBSCRIBE_STATES.SUBSCRIBED)
+  }
+
+  private _failSubscribeState(status: REALTIME_SUBSCRIBE_STATES, err?: Error) {
+    if (!this._subscribeCallback) {
+      return
+    }
+
+    const callback = this._subscribeCallback
+    this._resetSubscribeState()
+    this.state = CHANNEL_STATES.errored
+    callback?.(status, err)
+  }
+
+  private _startSubscribeTimeout(timeout: number) {
+    if (this._subscribeTimeoutId) {
+      clearTimeout(this._subscribeTimeoutId)
+    }
+
+    this._subscribeTimeoutId = setTimeout(() => {
+      this._failSubscribeState(REALTIME_SUBSCRIBE_STATES.TIMED_OUT)
+    }, timeout)
+  }
+
+  private _resetSubscribeState() {
+    this._postgresChangesSystemReady = false
+    this._subscribeCallback = null
+    if (this._subscribeTimeoutId) {
+      clearTimeout(this._subscribeTimeoutId)
+      this._subscribeTimeoutId = null
+    }
+  }
+
+  private _shouldWaitForPostgresChangesSystem() {
+    return (
+      (this.bindings.postgres_changes?.length ?? 0) > 0 &&
+      (this.bindings.broadcast?.length ?? 0) === 0 &&
+      (this.bindings.presence?.length ?? 0) === 0
+    )
+  }
+
   /** @internal */
   private _getPayloadRecords(payload: any) {
     const records = {