chore (client): expose a close method on the ElectricNamespace to dis…

…pose its resources (#703)

This PR addresses the first task of #243.
As suggested, the `ElectricNamespace` (and thus also the
`ElectricClient` as it extends the namespace) now exposes a `close`
method that can be called by the user to dispose all resources used by
the Electric client. This `close` method calls the `stop` on the global
registry such that it also stops the client's satellite process.

.changeset/brave-colts-change.md

@@ -0,0 +1,5 @@
+---
+"electric-sql": patch
+---
+
+Dispose listeners when Electric client is closed.

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>> = {
@@ -59,27 +59,26 @@ interface RawQueries {
 export class ElectricClient<
   DB extends DbSchema<any>
 > extends ElectricNamespace {
-  private _satellite: Satellite
-  public get satellite(): Satellite {
-    return this._satellite
-  }
-
   private constructor(
     public db: ClientTables<DB> & RawQueries,
+    dbName: string,
     adapter: DatabaseAdapter,
     notifier: Notifier,
-    satellite: Satellite
+    public readonly satellite: Satellite,
+    registry: Registry | GlobalRegistry
   ) {
-    super(adapter, notifier)
-    this._satellite = satellite
+    super(dbName, adapter, notifier, registry)
+    this.satellite = satellite
   }
 
   // Builds the DAL namespace from a `dbDescription` object
   static create<DB extends DbSchema<any>>(
+    dbName: string,
     dbDescription: DB,
     adapter: DatabaseAdapter,
     notifier: Notifier,
-    satellite: Satellite
+    satellite: Satellite,
+    registry: Registry | GlobalRegistry
   ): ElectricClient<DB> {
     const tables = dbDescription.extendedTables
     const shapeManager = new ShapeManager(satellite)
@@ -112,6 +111,13 @@ export class ElectricClient<
       liveRaw: liveRaw.bind(null, adapter),
     }
 
-    return new ElectricClient(db, adapter, notifier, satellite)
+    return new ElectricClient(
+      db,
+      dbName,
+      adapter,
+      notifier,
+      satellite,
+      registry
+    )
   }
 }

clients/typescript/src/electric/index.ts

@@ -70,10 +70,12 @@ export const electrify = async <DB extends DbSchema<any>>(
   )
 
   const electric = ElectricClient.create(
+    dbName,
     dbDescription,
     adapter,
     notifier,
-    satellite
+    satellite,
+    registry
   )
 
   if (satellite.connectivityState !== undefined) {

clients/typescript/src/electric/namespace.ts

@@ -3,37 +3,40 @@
 import { DatabaseAdapter } from './adapter'
 import { Notifier } from '../notifiers'
 import { ConnectivityState } from '../util/types'
+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
   }
 
-  constructor(adapter: DatabaseAdapter, notifier: Notifier) {
+  private _stateChangeSubscription: string
+
+  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
 
-    // XXX if you're implementing VAX-799, see the note below and maybe refactor
-    // this out of here whilst cleaning up the subscription.
-
-    // we need to set isConnected before the first event is emitted,
-    // otherwise application might be out of sync with satellite state.
-    this.notifier.subscribeToConnectivityStateChanges(
-      ({ connectivityState }) => {
-        this.setIsConnected(connectivityState)
-      }
-    )
+    this._stateChangeSubscription =
+      this.notifier.subscribeToConnectivityStateChanges(
+        ({ connectivityState }) => {
+          this.setIsConnected(connectivityState)
+        }
+      )
   }
 
-  // XXX this `isConnected` property is now only used via the ElectricClient.
-  // Now ... because the connectivity state change subscription is wired up
-  // here, we proxy this property from a dynamic `isConnected` getter on the
-  // ElectricClient. All of which is a bit unecessary and something of a
-  // code smell. As is the subscription above not being cleaned up.
   setIsConnected(connectivityState: ConnectivityState): void {
     this._isConnected = connectivityState === 'connected'
   }
@@ -44,4 +47,14 @@ export class ElectricNamespace {
   potentiallyChanged(): void {
     this.notifier.potentiallyChanged()
   }
+
+  /**
+   * Cleans up the resources used by the `ElectricNamespace`.
+   */
+  async close(): Promise<void> {
+    this.notifier.unsubscribeFromConnectivityStateChanges(
+      this._stateChangeSubscription
+    )
+    await this.registry.stop(this.dbName)
+  }
 }

clients/typescript/src/notifiers/event.ts

@@ -239,7 +239,7 @@ export class EventNotifier implements Notifier {
 
     this._unsubscribe(EVENT_NAMES.connectivityStateChange, callback)
 
-    delete this._changeCallbacks[key]
+    delete this._connectivityStatusCallbacks[key]
   }
 
   _getDbNames(): 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,
@@ -56,7 +57,14 @@ async function makeContext(t: ExecutionContext<ContextType>) {
     satelliteDefaults
   )
 
-  const electric = ElectricClient.create(schema, adapter, notifier, satellite)
+  const electric = ElectricClient.create(
+    dbName,
+    schema,
+    adapter,
+    notifier,
+    satellite,
+    registry
+  )
   const Post = electric.db.Post
   const Items = electric.db.Items
   const User = electric.db.User

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,7 +56,15 @@ test.beforeEach((t) => {
     {} as SocketFactory,
     {} as SatelliteOpts
   )
-  const dal = ElectricClient.create(schema, adapter, notifier, satellite)
+  const registry = new MockRegistry()
+  const dal = ElectricClient.create(
+    'test.db',
+    schema,
+    adapter,
+    notifier,
+    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'
@@ -179,6 +179,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, {
@@ -249,6 +250,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