Pass registry to ElectricClient and ElectricNamespace + unit test for…

… resource cleanup

clients/typescript/src/client/model/client.ts

@@ -5,7 +5,7 @@ import { Row, Statement } from '../../util'
 import { LiveResult, LiveResultContext } from './model'
 import { Notifier } from '../../notifiers'
 import { DatabaseAdapter } from '../../electric/adapter'
-import { Satellite } from '../../satellite'
+import { GlobalRegistry, Registry, Satellite } from '../../satellite'
 import { ShapeManager } from './shapes'
 
 export type ClientTables<DB extends DbSchema<any>> = {
@@ -64,9 +64,10 @@ export class ElectricClient<
     dbName: string,
     adapter: DatabaseAdapter,
     notifier: Notifier,
-    public readonly satellite: Satellite
+    public readonly satellite: Satellite,
+    registry: Registry | GlobalRegistry
   ) {
-    super(dbName, adapter, notifier)
+    super(dbName, adapter, notifier, registry)
     this.satellite = satellite
   }
 
@@ -76,7 +77,8 @@ export class ElectricClient<
     dbDescription: DB,
     adapter: DatabaseAdapter,
     notifier: Notifier,
-    satellite: Satellite
+    satellite: Satellite,
+    registry: Registry | GlobalRegistry
   ): ElectricClient<DB> {
     const tables = dbDescription.extendedTables
     const shapeManager = new ShapeManager(satellite)
@@ -109,6 +111,13 @@ export class ElectricClient<
       liveRaw: liveRaw.bind(null, adapter),
     }
 
-    return new ElectricClient(db, dbName, adapter, notifier, satellite)
+    return new ElectricClient(
+      db,
+      dbName,
+      adapter,
+      notifier,
+      satellite,
+      registry
+    )
   }
 }

clients/typescript/src/electric/index.ts

@@ -74,7 +74,8 @@ export const electrify = async <DB extends DbSchema<any>>(
     dbDescription,
     adapter,
     notifier,
-    satellite
+    satellite,
+    registry
   )
 
   if (satellite.connectivityState !== undefined) {

clients/typescript/src/electric/namespace.ts

@@ -3,23 +3,30 @@
 import { DatabaseAdapter } from './adapter'
 import { Notifier } from '../notifiers'
 import { ConnectivityState } from '../util/types'
-import { globalRegistry } from '../satellite'
+import { GlobalRegistry, Registry } from '../satellite'
 
 export class ElectricNamespace {
   dbName: string
   adapter: DatabaseAdapter
   notifier: Notifier
+  public readonly registry: Registry | GlobalRegistry
   private _isConnected: boolean
   public get isConnected(): boolean {
     return this._isConnected
   }
 
   private _stateChangeSubscription: string
 
-  constructor(dbName: string, adapter: DatabaseAdapter, notifier: Notifier) {
+  constructor(
+    dbName: string,
+    adapter: DatabaseAdapter,
+    notifier: Notifier,
+    registry: Registry | GlobalRegistry
+  ) {
     this.dbName = dbName
     this.adapter = adapter
     this.notifier = notifier
+    this.registry = registry
     this._isConnected = false
 
     this._stateChangeSubscription =
@@ -44,10 +51,10 @@ export class ElectricNamespace {
   /**
    * Cleans up the resources used by the `ElectricNamespace`.
    */
-  close(): void {
+  async close(): Promise<void> {
     this.notifier.unsubscribeFromConnectivityStateChanges(
       this._stateChangeSubscription
     )
-    globalRegistry.stop(this.dbName)
+    await this.registry.stop(this.dbName)
   }
 }

clients/typescript/src/satellite/index.ts

@@ -34,6 +34,10 @@ export type { ShapeSubscription } from './process'
 
 // `Registry` that starts one Satellite process per database.
 export interface Registry {
+  satellites: {
+    [key: DbName]: Satellite
+  }
+
   ensureStarted(
     dbName: DbName,
     dbDescription: DbSchema<any>,

clients/typescript/test/client/model/shapes.test.ts

@@ -4,7 +4,7 @@ import Database from 'better-sqlite3'
 import { schema } from '../generated'
 import { DatabaseAdapter } from '../../../src/drivers/better-sqlite3'
 import { SatelliteProcess } from '../../../src/satellite/process'
-import { MockSatelliteClient } from '../../../src/satellite/mock'
+import { MockRegistry, MockSatelliteClient } from '../../../src/satellite/mock'
 import { BundleMigrator } from '../../../src/migrators'
 import { MockNotifier } from '../../../src/notifiers'
 import { randomValue } from '../../../src/util'
@@ -46,6 +46,7 @@ async function makeContext(t: ExecutionContext<ContextType>) {
   const migrator = new BundleMigrator(adapter, migrations)
   const dbName = `.tmp/test-${randomValue()}.db`
   const notifier = new MockNotifier(dbName)
+  const registry = new MockRegistry()
 
   const satellite = new SatelliteProcess(
     dbName,
@@ -61,7 +62,8 @@ async function makeContext(t: ExecutionContext<ContextType>) {
     schema,
     adapter,
     notifier,
-    satellite
+    satellite,
+    registry
   )
   const Post = electric.db.Post
   const Items = electric.db.Items

clients/typescript/test/client/notifications.test.ts

@@ -3,6 +3,8 @@ import Database from 'better-sqlite3'
 import { electrify } from '../../src/drivers/better-sqlite3'
 import { schema } from './generated'
 import { MockRegistry } from '../../src/satellite/mock'
+import { EventNotifier } from '../../src/notifiers'
+import { mockElectricClient } from '../satellite/common'
 
 const conn = new Database(':memory:')
 const config = {
@@ -212,3 +214,30 @@ test.serial('deleteMany runs potentiallyChanged', async (t) => {
   const notifications = await runAndCheckNotifications(del)
   t.is(notifications, 1)
 })
+
+test.serial(
+  'electrification registers process and unregisters on close thereby releasing resources',
+  async (t) => {
+    const registry = new MockRegistry()
+    const electric = await mockElectricClient(conn, registry)
+
+    // Check that satellite is registered
+    const satellite = electric.satellite
+    t.is(registry.satellites[conn.name], satellite)
+
+    // Check that the listeners are registered
+    const notifier = electric.notifier as EventNotifier
+    t.assert(Object.keys(notifier._changeCallbacks).length > 0)
+    t.assert(Object.keys(notifier._connectivityStatusCallbacks).length > 0)
+
+    // Close the Electric client
+    await electric.close()
+
+    // Check that the listeners are unregistered
+    t.deepEqual(notifier._changeCallbacks, {})
+    t.deepEqual(notifier._connectivityStatusCallbacks, {})
+
+    // Check that the Satellite process is unregistered
+    t.assert(!registry.satellites.hasOwnProperty(conn.name))
+  }
+)

clients/typescript/test/frameworks/react.test.tsx

@@ -21,7 +21,7 @@ import {
 import { makeElectricContext } from '../../src/frameworks/react/provider'
 import { ElectricClient } from '../../src/client/model/client'
 import { schema, Electric } from '../client/generated'
-import { MockSatelliteProcess } from '../../src/satellite/mock'
+import { MockRegistry, MockSatelliteProcess } from '../../src/satellite/mock'
 import { Migrator } from '../../src/migrators'
 import { SocketFactory } from '../../src/sockets'
 import { SatelliteOpts } from '../../src/satellite/config'
@@ -56,12 +56,14 @@ test.beforeEach((t) => {
     {} as SocketFactory,
     {} as SatelliteOpts
   )
+  const registry = new MockRegistry()
   const dal = ElectricClient.create(
     'test.db',
     schema,
     adapter,
     notifier,
-    satellite
+    satellite,
+    registry
   )
 
   dal.db.Items.sync()

clients/typescript/test/satellite/common.ts

@@ -6,7 +6,7 @@ import { DatabaseAdapter } from '../../src/drivers/better-sqlite3'
 import { BundleMigrator } from '../../src/migrators'
 import { EventNotifier, MockNotifier } from '../../src/notifiers'
 import { MockSatelliteClient } from '../../src/satellite/mock'
-import { SatelliteProcess } from '../../src/satellite'
+import { GlobalRegistry, Registry, SatelliteProcess } from '../../src/satellite'
 import { TableInfo, initTableInfo } from '../support/satellite-helpers'
 import { satelliteDefaults, SatelliteOpts } from '../../src/satellite/config'
 import { Table, generateTableTriggers } from '../../src/migrators/triggers'
@@ -161,6 +161,7 @@ import { AuthState } from '../../src/auth'
 import { DbSchema, TableSchema } from '../../src/client/model/schema'
 import { PgBasicType } from '../../src/client/conversions/types'
 import { HKT } from '../../src/client/util/hkt'
+import { ElectricClient } from '../../src/client/model'
 
 // Speed up the intervals for testing.
 export const opts = Object.assign({}, satelliteDefaults, {
@@ -231,6 +232,32 @@ export const makeContext = async (
   }
 }
 
+export const mockElectricClient = async (
+  db: SqliteDB,
+  registry: Registry | GlobalRegistry,
+  options: Opts = opts
+): Promise<ElectricClient<any>> => {
+  const dbName = db.name
+  const adapter = new DatabaseAdapter(db)
+  const migrator = new BundleMigrator(adapter, migrations)
+  const notifier = new MockNotifier(dbName)
+  const client = new MockSatelliteClient()
+  const satellite = new SatelliteProcess(
+    dbName,
+    adapter,
+    migrator,
+    notifier,
+    client,
+    options
+  )
+
+  await satellite.start({ clientId: '', token: 'test-token' })
+  registry.satellites[dbName] = satellite
+
+  // @ts-ignore Mock Electric client that does not contain the DAL
+  return new ElectricClient({}, dbName, adapter, notifier, satellite, registry)
+}
+
 export const clean = async (t: ExecutionContext<{ dbName: string }>) => {
   const { dbName } = t.context