feat(client): Shape subscription status notifier

.changeset/young-plants-try.md

@@ -0,0 +1,5 @@
+---
+"electric-sql": patch
+---
+
+Add notifier method `subscribeToShapeSubscriptionSyncStatusChanges` for listening to shape subscription status updates

clients/typescript/src/notifiers/event.ts

@@ -19,14 +19,18 @@ import {
   Notifier,
   PotentialChangeCallback,
   PotentialChangeNotification,
+  ShapeSubscriptionSyncStatusChangeCallback,
+  ShapeSubscriptionSyncStatusChangeNotification,
   UnsubscribeFunction,
 } from './index'
+import { SyncStatus } from '../client/model/shapes'
 
 export const EVENT_NAMES = {
   authChange: 'auth:changed',
   actualDataChange: 'data:actually:changed',
   potentialDataChange: 'data:potentially:changed',
   connectivityStateChange: 'network:connectivity:changed',
+  shapeSubscriptionStatusChange: 'shape:status:changed',
 }
 
 // Initialise global emitter to be shared between all
@@ -201,6 +205,41 @@ export class EventNotifier implements Notifier {
     }
   }
 
+  shapeSubscriptionSyncStatusChanged(
+    dbName: string,
+    key: string,
+    status: SyncStatus
+  ): void {
+    if (!this._hasDbName(dbName)) {
+      return
+    }
+
+    this._emitShapeSubscriptionSyncStatusChange(dbName, key, status)
+  }
+
+  subscribeToShapeSubscriptionSyncStatusChanges(
+    callback: ShapeSubscriptionSyncStatusChangeCallback
+  ): UnsubscribeFunction {
+    const thisHasDbName = this._hasDbName.bind(this)
+
+    const wrappedCallback = (
+      notification: ShapeSubscriptionSyncStatusChangeNotification
+    ) => {
+      if (thisHasDbName(notification.dbName)) {
+        callback(notification)
+      }
+    }
+
+    this._subscribe(EVENT_NAMES.shapeSubscriptionStatusChange, wrappedCallback)
+
+    return () => {
+      this._unsubscribe(
+        EVENT_NAMES.shapeSubscriptionStatusChange,
+        wrappedCallback
+      )
+    }
+  }
+
   _getDbNames(): DbName[] {
     const idx = this.attachedDbIndex
 
@@ -261,6 +300,18 @@ export class EventNotifier implements Notifier {
     return notification
   }
 
+  _emitShapeSubscriptionSyncStatusChange(
+    dbName: DbName,
+    key: string,
+    status: SyncStatus
+  ): ShapeSubscriptionSyncStatusChangeNotification {
+    const notification = { dbName, key, status }
+
+    this._emit(EVENT_NAMES.shapeSubscriptionStatusChange, notification)
+
+    return notification
+  }
+
   _emit(eventName: string, notification: Notification) {
     this.events.emit(eventName, notification)
   }

clients/typescript/src/notifiers/index.ts

@@ -1,4 +1,5 @@
 import { AuthState } from '../auth/index'
+import { SyncStatus } from '../client/model/shapes'
 import { QualifiedTablename } from '../util/tablename'
 import {
   ConnectivityState,
@@ -39,11 +40,18 @@ export interface ConnectivityStateChangeNotification {
   connectivityState: ConnectivityState
 }
 
+export interface ShapeSubscriptionSyncStatusChangeNotification {
+  dbName: DbName
+  key: string
+  status: SyncStatus
+}
+
 export type Notification =
   | AuthStateNotification
   | ChangeNotification
   | PotentialChangeNotification
   | ConnectivityStateChangeNotification
+  | ShapeSubscriptionSyncStatusChangeNotification
 
 export type AuthStateCallback = (notification: AuthStateNotification) => void
 export type ChangeCallback = (notification: ChangeNotification) => void
@@ -54,11 +62,16 @@ export type ConnectivityStateChangeCallback = (
   notification: ConnectivityStateChangeNotification
 ) => void
 
+export type ShapeSubscriptionSyncStatusChangeCallback = (
+  notification: ShapeSubscriptionSyncStatusChangeNotification
+) => void
+
 export type NotificationCallback =
   | AuthStateCallback
   | ChangeCallback
   | PotentialChangeCallback
   | ConnectivityStateChangeCallback
+  | ShapeSubscriptionSyncStatusChangeCallback
 
 export type UnsubscribeFunction = () => void
 
@@ -129,4 +142,17 @@ export interface Notifier {
   subscribeToConnectivityStateChanges(
     callback: ConnectivityStateChangeCallback
   ): UnsubscribeFunction
+
+  // Notification for shape subscription sync status changes.
+  // Every notification will include a key that uniquely identifies the
+  // shape for which the sync status changed, as well as the new sync status.
+  shapeSubscriptionSyncStatusChanged(
+    dbName: DbName,
+    key: string,
+    status: SyncStatus
+  ): void
+
+  subscribeToShapeSubscriptionSyncStatusChanges(
+    callback: ShapeSubscriptionSyncStatusChangeCallback
+  ): UnsubscribeFunction
 }

clients/typescript/src/satellite/mock.ts

@@ -61,8 +61,8 @@ import {
 } from '../_generated/protocol/satellite'
 import { ShapeSubscription } from './process'
 import { DbSchema } from '../client/model/schema'
-import { getAllTablesForShape } from './shapes'
 import { SyncStatus } from '../client/model/shapes'
+import { getTableNamesForShapes } from './shapes/shapeManager'
 
 export const MOCK_BEHIND_WINDOW_LSN = 42
 export const MOCK_INTERNAL_ERROR = 27
@@ -276,7 +276,7 @@ export class MockSatelliteClient
     const shapeReqToUuid: Record<string, string> = {}
 
     for (const shape of shapes) {
-      const tables = getAllTablesForShape(shape.definition, 'main')
+      const tables = getTableNamesForShapes([shape.definition], 'main')
       for (const { tablename } of tables) {
         if (tablename === 'failure' || tablename === 'Items') {
           return Promise.resolve({

clients/typescript/src/satellite/process.ts

@@ -176,7 +176,12 @@ export class SatelliteProcess implements Satellite {
     this.relations = {}
 
     this.previousShapeSubscriptions = []
-    this.subscriptionManager = new ShapeManager()
+    this.subscriptionManager = new ShapeManager(
+      this.notifier.shapeSubscriptionSyncStatusChanged.bind(
+        this.notifier,
+        this.dbName
+      )
+    )
 
     this._throttledSnapshot = throttle(
       this._mutexSnapshot.bind(this),
@@ -446,6 +451,7 @@ export class SatelliteProcess implements Satellite {
 
       if (error) throw error
 
+      // persist subscription metadata
       await this._setMeta('subscriptions', this.subscriptionManager.serialize())
 
       return {
@@ -470,7 +476,7 @@ export class SatelliteProcess implements Satellite {
       )
     } else {
       return this.unsubscribeIds(
-        this.subscriptionManager.getServerID(target.shapes)
+        this.subscriptionManager.getServerIDsForShapes(target.shapes)
       )
     }
   }
@@ -482,6 +488,8 @@ export class SatelliteProcess implements Satellite {
 
     // If the server didn't send an error, we persist the fact the subscription was deleted.
     this.subscriptionManager.unsubscribeMade(subscriptionIds)
+
+    // persist subscription metadata
     await this.adapter.run(
       this._setMetaStatement(
         'subscriptions',
@@ -501,6 +509,7 @@ export class SatelliteProcess implements Satellite {
       [],
       subsData.subscriptionId
     )
+
     const toBeUnsubbed = afterApply()
     if (toBeUnsubbed.length > 0) await this.unsubscribeIds(toBeUnsubbed)
   }
@@ -1549,6 +1558,7 @@ export class SatelliteProcess implements Satellite {
       ...stmts,
       ...this._enableTriggers(affectedTables)
     )
+
     this.subscriptionManager.goneBatchDelivered(subscriptionIds)
 
     this._notifyChanges(fakeOplogEntries, 'remote')