[Modules] Feat: Replication Events

.changeset/orange-eagles-tap.md

@@ -0,0 +1,5 @@
+---
+'@powersync/lib-services-framework': minor
+---
+
+Added disposable listeners and observers

.changeset/popular-snails-cough.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-core': minor
+'@powersync/service-sync-rules': minor
+---
+
+Added ability to emit data replication events

.changeset/sour-turkeys-collect.md

@@ -0,0 +1,7 @@
+---
+'@powersync/service-module-postgres': patch
+'@powersync/service-rsocket-router': patch
+'@powersync/service-types': patch
+---
+
+Updates from Replication events changes

libs/lib-services/src/utils/BaseObserver.ts

@@ -1,6 +1,10 @@
 import { v4 as uuid } from 'uuid';
 
-export class BaseObserver<T> {
+export interface ObserverClient<T> {
+  registerListener(listener: Partial<T>): () => void;
+}
+
+export class BaseObserver<T> implements ObserverClient<T> {
   protected listeners: { [id: string]: Partial<T> };
 
   constructor() {

libs/lib-services/src/utils/DisposableObserver.ts

@@ -0,0 +1,37 @@
+import { BaseObserver, ObserverClient } from './BaseObserver.js';
+
+export interface DisposableListener {
+  /**
+   * Event which is fired when the `[Symbol.disposed]` method is called.
+   */
+  disposed: () => void;
+}
+
+export interface DisposableObserverClient<T extends DisposableListener> extends ObserverClient<T>, Disposable {
+  /**
+   * Registers a listener that is automatically disposed when the parent is disposed.
+   * This is useful for disposing nested listeners.
+   */
+  registerManagedListener: (parent: DisposableObserverClient<DisposableListener>, cb: Partial<T>) => () => void;
+}
+
+export class DisposableObserver<T extends DisposableListener>
+  extends BaseObserver<T>
+  implements DisposableObserverClient<T>
+{
+  registerManagedListener(parent: DisposableObserverClient<DisposableListener>, cb: Partial<T>) {
+    const disposer = this.registerListener(cb);
+    parent.registerListener({
+      disposed: () => {
+        disposer();
+      }
+    });
+    return disposer;
+  }
+
+  [Symbol.dispose]() {
+    this.iterateListeners((cb) => cb.disposed?.());
+    // Delete all callbacks
+    Object.keys(this.listeners).forEach((key) => delete this.listeners[key]);
+  }
+}

libs/lib-services/src/utils/utils-index.ts

@@ -1,2 +1,3 @@
 export * from './BaseObserver.js';
+export * from './DisposableObserver.js';
 export * from './environment-variables.js';

libs/lib-services/test/src/DisposeableObserver.test.ts

@@ -0,0 +1,58 @@
+import { describe, expect, test } from 'vitest';
+
+import { DisposableListener, DisposableObserver } from '../../src/utils/DisposableObserver.js';
+
+describe('DisposableObserver', () => {
+  test('it should dispose all listeners on dispose', () => {
+    const listener = new DisposableObserver();
+
+    let wasDisposed = false;
+    listener.registerListener({
+      disposed: () => {
+        wasDisposed = true;
+      }
+    });
+
+    listener[Symbol.dispose]();
+
+    expect(wasDisposed).equals(true);
+    expect(Object.keys(listener['listeners']).length).equals(0);
+  });
+
+  test('it should dispose nested listeners for managed listeners', () => {
+    interface ParentListener extends DisposableListener {
+      childCreated: (child: DisposableObserver<any>) => void;
+    }
+    class ParentObserver extends DisposableObserver<ParentListener> {
+      createChild() {
+        const child = new DisposableObserver();
+        this.iterateListeners((cb) => cb.childCreated?.(child));
+      }
+    }
+
+    const parent = new ParentObserver();
+    let aChild: DisposableObserver<any> | null = null;
+
+    parent.registerListener({
+      childCreated: (child) => {
+        aChild = child;
+        child.registerManagedListener(parent, {
+          test: () => {
+            // this does nothing
+          }
+        });
+      }
+    });
+
+    parent.createChild();
+
+    // The managed listener should add a `disposed` listener
+    expect(Object.keys(parent['listeners']).length).equals(2);
+    expect(Object.keys(aChild!['listeners']).length).equals(1);
+
+    parent[Symbol.dispose]();
+    expect(Object.keys(parent['listeners']).length).equals(0);
+    // The listener attached to the child should be disposed when the parent was disposed
+    expect(Object.keys(aChild!['listeners']).length).equals(0);
+  });
+});

modules/module-postgres/src/module/PostgresModule.ts

@@ -1,20 +1,13 @@
-import {
-  api,
-  auth,
-  ConfigurationFileSyncRulesProvider,
-  replication,
-  system,
-  TearDownOptions
-} from '@powersync/service-core';
+import { api, auth, ConfigurationFileSyncRulesProvider, modules, replication, system } from '@powersync/service-core';
 import * as jpgwire from '@powersync/service-jpgwire';
-import * as types from '../types/types.js';
 import { PostgresRouteAPIAdapter } from '../api/PostgresRouteAPIAdapter.js';
 import { SupabaseKeyCollector } from '../auth/SupabaseKeyCollector.js';
-import { WalStreamReplicator } from '../replication/WalStreamReplicator.js';
 import { ConnectionManagerFactory } from '../replication/ConnectionManagerFactory.js';
+import { PgManager } from '../replication/PgManager.js';
 import { PostgresErrorRateLimiter } from '../replication/PostgresErrorRateLimiter.js';
 import { cleanUpReplicationSlot } from '../replication/replication-utils.js';
-import { PgManager } from '../replication/PgManager.js';
+import { WalStreamReplicator } from '../replication/WalStreamReplicator.js';
+import * as types from '../types/types.js';
 
 export class PostgresModule extends replication.ReplicationModule<types.PostgresConnectionConfig> {
   constructor() {
@@ -70,7 +63,7 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
     };
   }
 
-  async teardown(options: TearDownOptions): Promise<void> {
+  async teardown(options: modules.TearDownOptions): Promise<void> {
     const normalisedConfig = this.resolveConfig(this.decodedConfig!);
     const connectionManager = new PgManager(normalisedConfig, {
       idleTimeout: 30_000,

modules/module-postgres/src/replication/WalStream.ts

@@ -1,11 +1,11 @@
-import * as pgwire from '@powersync/service-jpgwire';
-import * as util from '../utils/pgwire_utils.js';
 import { container, errors, logger } from '@powersync/lib-services-framework';
+import { getUuidReplicaIdentityBson, Metrics, SourceEntityDescriptor, storage } from '@powersync/service-core';
+import * as pgwire from '@powersync/service-jpgwire';
 import { DatabaseInputRow, SqliteRow, SqlSyncRules, TablePattern, toSyncRulesRow } from '@powersync/service-sync-rules';
+import * as pg_utils from '../utils/pgwire_utils.js';
+import { PgManager } from './PgManager.js';
 import { getPgOutputRelation, getRelId } from './PgRelation.js';
-import { getUuidReplicaIdentityBson, Metrics, SourceEntityDescriptor, storage } from '@powersync/service-core';
 import { checkSourceConfiguration, getReplicationIdentityColumns } from './replication-utils.js';
-import { PgManager } from './PgManager.js';
 
 export const ZERO_LSN = '00000000/00000000';
 export const PUBLICATION_NAME = 'powersync';
@@ -60,7 +60,7 @@ export class WalStream {
           // Ping to speed up cancellation of streaming replication
           // We're not using pg_snapshot here, since it could be in the middle of
           // an initial replication transaction.
-          const promise = util.retriedQuery(
+          const promise = pg_utils.retriedQuery(
             this.connections.pool,
             `SELECT * FROM pg_logical_emit_message(false, 'powersync', 'ping')`
           );
@@ -347,7 +347,6 @@ WHERE  oid = $1::regclass`,
         for (let table of tables) {
           await this.snapshotTable(batch, db, table);
           await batch.markSnapshotDone([table], lsn);
-
           await touch();
         }
       }
@@ -395,7 +394,7 @@ WHERE  oid = $1::regclass`,
         throw new Error(`Aborted initial replication of ${this.slot_name}`);
       }
 
-      for (let record of WalStream.getQueryData(rows)) {
+      for (const record of WalStream.getQueryData(rows)) {
         // This auto-flushes when the batch reaches its size limit
         await batch.save({
           tag: 'insert',
@@ -406,6 +405,7 @@ WHERE  oid = $1::regclass`,
           afterReplicaId: getUuidReplicaIdentityBson(record, table.replicaIdColumns)
         });
       }
+
       at += rows.length;
       Metrics.getInstance().rows_replicated_total.add(rows.length);
 
@@ -495,7 +495,7 @@ WHERE  oid = $1::regclass`,
 
       if (msg.tag == 'insert') {
         Metrics.getInstance().rows_replicated_total.add(1);
-        const baseRecord = util.constructAfterRecord(msg);
+        const baseRecord = pg_utils.constructAfterRecord(msg);
         return await batch.save({
           tag: 'insert',
           sourceTable: table,
@@ -508,8 +508,8 @@ WHERE  oid = $1::regclass`,
         Metrics.getInstance().rows_replicated_total.add(1);
         // "before" may be null if the replica id columns are unchanged
         // It's fine to treat that the same as an insert.
-        const before = util.constructBeforeRecord(msg);
-        const after = util.constructAfterRecord(msg);
+        const before = pg_utils.constructBeforeRecord(msg);
+        const after = pg_utils.constructAfterRecord(msg);
         return await batch.save({
           tag: 'update',
           sourceTable: table,
@@ -520,7 +520,7 @@ WHERE  oid = $1::regclass`,
         });
       } else if (msg.tag == 'delete') {
         Metrics.getInstance().rows_replicated_total.add(1);
-        const before = util.constructBeforeRecord(msg)!;
+        const before = pg_utils.constructBeforeRecord(msg)!;
 
         return await batch.save({
           tag: 'delete',
@@ -592,7 +592,6 @@ WHERE  oid = $1::regclass`,
         // chunkLastLsn may come from normal messages in the chunk,
         // or from a PrimaryKeepalive message.
         const { messages, lastLsn: chunkLastLsn } = chunk;
-
         for (const msg of messages) {
           if (msg.tag == 'relation') {
             await this.handleRelation(batch, getPgOutputRelation(msg), true);
@@ -609,7 +608,7 @@ WHERE  oid = $1::regclass`,
             }
 
             count += 1;
-            const result = await this.writeChange(batch, msg);
+            await this.writeChange(batch, msg);
           }
         }
 

modules/module-postgres/src/replication/WalStreamReplicationJob.ts

@@ -1,6 +1,6 @@
-import { MissingReplicationSlotError, WalStream } from './WalStream.js';
 import { container } from '@powersync/lib-services-framework';
 import { PgManager } from './PgManager.js';
+import { MissingReplicationSlotError, WalStream } from './WalStream.js';
 
 import { replication } from '@powersync/service-core';
 import { ConnectionManagerFactory } from './ConnectionManagerFactory.js';

modules/module-postgres/src/replication/WalStreamReplicator.ts

@@ -1,7 +1,7 @@
-import { storage, replication } from '@powersync/service-core';
-import { WalStreamReplicationJob } from './WalStreamReplicationJob.js';
+import { replication, storage } from '@powersync/service-core';
 import { ConnectionManagerFactory } from './ConnectionManagerFactory.js';
 import { cleanUpReplicationSlot } from './replication-utils.js';
+import { WalStreamReplicationJob } from './WalStreamReplicationJob.js';
 
 export interface WalStreamReplicatorOptions extends replication.AbstractReplicatorOptions {
   connectionFactory: ConnectionManagerFactory;

modules/module-postgres/test/src/slow_tests.test.ts

@@ -2,16 +2,16 @@ import * as bson from 'bson';
 import { afterEach, describe, expect, test } from 'vitest';
 import { WalStream, WalStreamOptions } from '../../src/replication/WalStream.js';
 import { env } from './env.js';
-import { TEST_CONNECTION_OPTIONS, clearTestDb, connectPgPool, getClientCheckpoint } from './util.js';
+import { clearTestDb, connectPgPool, getClientCheckpoint, TEST_CONNECTION_OPTIONS } from './util.js';
 
 import * as pgwire from '@powersync/service-jpgwire';
 import { SqliteRow } from '@powersync/service-sync-rules';
 
 import { mapOpEntry, MongoBucketStorage } from '@/storage/storage-index.js';
-import * as timers from 'node:timers/promises';
+import { reduceBucket, validateCompactedBucket } from '@core-tests/bucket_validation.js';
 import { MONGO_STORAGE_FACTORY, StorageFactory } from '@core-tests/util.js';
 import { PgManager } from '@module/replication/PgManager.js';
-import { reduceBucket, validateCompactedBucket } from '@core-tests/bucket_validation.js';
+import * as timers from 'node:timers/promises';
 
 describe('slow tests - mongodb', function () {
   // These are slow, inconsistent tests.
@@ -82,7 +82,7 @@ bucket_definitions:
       - SELECT * FROM "test_data"
 `;
     const syncRules = await f.updateSyncRules({ content: syncRuleContent });
-    const storage = f.getInstance(syncRules);
+    using storage = f.getInstance(syncRules);
     abortController = new AbortController();
     const options: WalStreamOptions = {
       abort_signal: abortController.signal,
@@ -234,7 +234,7 @@ bucket_definitions:
       - SELECT id, description FROM "test_data"
 `;
       const syncRules = await f.updateSyncRules({ content: syncRuleContent });
-      const storage = f.getInstance(syncRules);
+      using storage = f.getInstance(syncRules);
 
       // 1. Setup some base data that will be replicated in initial replication
       await pool.query(`CREATE TABLE test_data(id uuid primary key default uuid_generate_v4(), description text)`);

modules/module-postgres/test/src/util.ts

@@ -1,11 +1,11 @@
+import { connectMongo } from '@core-tests/util.js';
 import * as types from '@module/types/types.js';
 import * as pg_utils from '@module/utils/pgwire_utils.js';
+import { logger } from '@powersync/lib-services-framework';
 import { BucketStorageFactory, Metrics, MongoBucketStorage, OpId } from '@powersync/service-core';
 import * as pgwire from '@powersync/service-jpgwire';
-import { env } from './env.js';
 import { pgwireRows } from '@powersync/service-jpgwire';
-import { logger } from '@powersync/lib-services-framework';
-import { connectMongo } from '@core-tests/util.js';
+import { env } from './env.js';
 
 // The metrics need to be initialized before they can be used
 await Metrics.initialise({
@@ -35,7 +35,9 @@ export const INITIALIZED_MONGO_STORAGE_FACTORY: StorageFactory = async () => {
 
   await db.clear();
 
-  return new MongoBucketStorage(db, { slot_name_prefix: 'test_' });
+  return new MongoBucketStorage(db, {
+    slot_name_prefix: 'test_'
+  });
 };
 
 export async function clearTestDb(db: pgwire.PgClient) {

modules/module-postgres/test/src/wal_stream_utils.ts

@@ -1,9 +1,9 @@
-import { BucketStorageFactory, SyncRulesBucketStorage } from '@powersync/service-core';
-import * as pgwire from '@powersync/service-jpgwire';
-import { TEST_CONNECTION_OPTIONS, clearTestDb, getClientCheckpoint } from './util.js';
-import { WalStream, WalStreamOptions, PUBLICATION_NAME } from '@module/replication/WalStream.js';
 import { fromAsync } from '@core-tests/stream_utils.js';
 import { PgManager } from '@module/replication/PgManager.js';
+import { PUBLICATION_NAME, WalStream, WalStreamOptions } from '@module/replication/WalStream.js';
+import { BucketStorageFactory, SyncRulesBucketStorage } from '@powersync/service-core';
+import * as pgwire from '@powersync/service-jpgwire';
+import { clearTestDb, getClientCheckpoint, TEST_CONNECTION_OPTIONS } from './util.js';
 
 /**
  * Tests operating on the wal stream need to configure the stream and manage asynchronous
@@ -20,16 +20,12 @@ export function walStreamTest(
     const connectionManager = new PgManager(TEST_CONNECTION_OPTIONS, {});
 
     await clearTestDb(connectionManager.pool);
-    const context = new WalStreamTestContext(f, connectionManager);
-    try {
-      await test(context);
-    } finally {
-      await context.dispose();
-    }
+    await using context = new WalStreamTestContext(f, connectionManager);
+    await test(context);
   };
 }
 
-export class WalStreamTestContext {
+export class WalStreamTestContext implements AsyncDisposable {
   private _walStream?: WalStream;
   private abortController = new AbortController();
   private streamPromise?: Promise<void>;
@@ -41,10 +37,11 @@ export class WalStreamTestContext {
     public connectionManager: PgManager
   ) {}
 
-  async dispose() {
+  async [Symbol.asyncDispose]() {
     this.abortController.abort();
     await this.streamPromise;
     await this.connectionManager.destroy();
+    this.storage?.[Symbol.dispose]();
   }
 
   get pool() {