Abstract classes for bucket storage factory

.changeset/nasty-bananas-pretend.md

@@ -0,0 +1,9 @@
+---
+'@powersync/service-module-postgres-storage': minor
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core-tests': minor
+'@powersync/service-module-mongodb': minor
+'@powersync/service-core': minor
+---
+
+Refactor `BucketStorageFactory` and `PersistedSyncRulesContent` to be abstract classes instead of interfaces.

modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts

@@ -1,8 +1,6 @@
-import { SqlSyncRules } from '@powersync/service-sync-rules';
-
 import { GetIntanceOptions, storage } from '@powersync/service-core';
 
-import { BaseObserver, ErrorCode, logger, ServiceError } from '@powersync/lib-services-framework';
+import { ErrorCode, ServiceError } from '@powersync/lib-services-framework';
 import { v4 as uuid } from 'uuid';
 
 import * as lib_mongo from '@powersync/lib-service-mongodb';
@@ -11,18 +9,15 @@ import { mongo } from '@powersync/lib-service-mongodb';
 import { PowerSyncMongo } from './implementation/db.js';
 import { SyncRuleDocument } from './implementation/models.js';
 import { MongoPersistedSyncRulesContent } from './implementation/MongoPersistedSyncRulesContent.js';
-import { MongoSyncBucketStorage, MongoSyncBucketStorageOptions } from './implementation/MongoSyncBucketStorage.js';
+import { MongoSyncBucketStorage } from './implementation/MongoSyncBucketStorage.js';
 import { generateSlotName } from '../utils/util.js';
 import { MongoChecksumOptions } from './implementation/MongoChecksums.js';
 
 export interface MongoBucketStorageOptions {
   checksumOptions?: Omit<MongoChecksumOptions, 'storageConfig'>;
 }
 
-export class MongoBucketStorage
-  extends BaseObserver<storage.BucketStorageFactoryListener>
-  implements storage.BucketStorageFactory
-{
+export class MongoBucketStorage extends storage.BucketStorageFactory {
   private readonly client: mongo.MongoClient;
   private readonly session: mongo.ClientSession;
   // TODO: This is still Postgres specific and needs to be reworked
@@ -91,33 +86,13 @@ export class MongoBucketStorage
     };
   }
 
-  async configureSyncRules(options: storage.UpdateSyncRulesOptions) {
-    const next = await this.getNextSyncRulesContent();
-    const active = await this.getActiveSyncRulesContent();
-
-    if (next?.sync_rules_content == options.content) {
-      logger.info('Sync rules from configuration unchanged');
-      return { updated: false };
-    } else if (next == null && active?.sync_rules_content == options.content) {
-      logger.info('Sync rules from configuration unchanged');
-      return { updated: false };
-    } else {
-      logger.info('Sync rules updated from configuration');
-      const persisted_sync_rules = await this.updateSyncRules(options);
-      return { updated: true, persisted_sync_rules, lock: persisted_sync_rules.current_lock ?? undefined };
-    }
-  }
-
   async restartReplication(sync_rules_group_id: number) {
     const next = await this.getNextSyncRulesContent();
     const active = await this.getActiveSyncRulesContent();
 
     if (next != null && next.id == sync_rules_group_id) {
       // We need to redo the "next" sync rules
-      await this.updateSyncRules({
-        content: next.sync_rules_content,
-        validate: false
-      });
+      await this.updateSyncRules(next.asUpdateOptions());
       // Pro-actively stop replicating
       await this.db.sync_rules.updateOne(
         {
@@ -133,10 +108,7 @@ export class MongoBucketStorage
       await this.db.notifyCheckpoint();
     } else if (next == null && active?.id == sync_rules_group_id) {
       // Slot removed for "active" sync rules, while there is no "next" one.
-      await this.updateSyncRules({
-        content: active.sync_rules_content,
-        validate: false
-      });
+      await this.updateSyncRules(active.asUpdateOptions());
 
       // In this case we keep the old one as active for clients, so that that existing clients
       // can still get the latest data while we replicate the new ones.
@@ -173,19 +145,6 @@ export class MongoBucketStorage
   }
 
   async updateSyncRules(options: storage.UpdateSyncRulesOptions): Promise<MongoPersistedSyncRulesContent> {
-    if (options.validate) {
-      // Parse and validate before applying any changes
-      SqlSyncRules.fromYaml(options.content, {
-        // No schema-based validation at this point
-        schema: undefined,
-        defaultSchema: 'not_applicable', // Not needed for validation
-        throwOnError: true
-      });
-    } else {
-      // We do not validate sync rules at this point.
-      // That is done when using the sync rules, so that the diagnostics API can report the errors.
-    }
-
     let rules: MongoPersistedSyncRulesContent | undefined = undefined;
 
     await this.session.withTransaction(async () => {
@@ -219,7 +178,7 @@ export class MongoBucketStorage
       const doc: SyncRuleDocument = {
         _id: id,
         storage_version: storageVersion,
-        content: options.content,
+        content: options.config.yaml,
         last_checkpoint: null,
         last_checkpoint_lsn: null,
         no_checkpoint_before: null,
@@ -258,11 +217,6 @@ export class MongoBucketStorage
     return new MongoPersistedSyncRulesContent(this.db, doc);
   }
 
-  async getActiveSyncRules(options: storage.ParseSyncRulesOptions): Promise<storage.PersistedSyncRules | null> {
-    const content = await this.getActiveSyncRulesContent();
-    return content?.parsed(options) ?? null;
-  }
-
   async getNextSyncRulesContent(): Promise<MongoPersistedSyncRulesContent | null> {
     const doc = await this.db.sync_rules.findOne(
       {
@@ -277,11 +231,6 @@ export class MongoBucketStorage
     return new MongoPersistedSyncRulesContent(this.db, doc);
   }
 
-  async getNextSyncRules(options: storage.ParseSyncRulesOptions): Promise<storage.PersistedSyncRules | null> {
-    const content = await this.getNextSyncRulesContent();
-    return content?.parsed(options) ?? null;
-  }
-
   async getReplicatingSyncRules(): Promise<storage.PersistedSyncRulesContent[]> {
     const docs = await this.db.sync_rules
       .find({

modules/module-mongodb-storage/src/storage/implementation/MongoPersistedSyncRulesContent.ts

@@ -1,49 +1,32 @@
 import { mongo } from '@powersync/lib-service-mongodb';
 import { storage } from '@powersync/service-core';
-import { SqlSyncRules } from '@powersync/service-sync-rules';
-import { MongoPersistedSyncRules } from './MongoPersistedSyncRules.js';
 import { MongoSyncRulesLock } from './MongoSyncRulesLock.js';
 import { PowerSyncMongo } from './db.js';
 import { getMongoStorageConfig, SyncRuleDocument } from './models.js';
 import { ErrorCode, ServiceError } from '@powersync/lib-services-framework';
 
-export class MongoPersistedSyncRulesContent implements storage.PersistedSyncRulesContent {
-  public readonly slot_name: string;
-
-  public readonly id: number;
-  public readonly sync_rules_content: string;
-  public readonly last_checkpoint_lsn: string | null;
-  public readonly last_fatal_error: string | null;
-  public readonly last_fatal_error_ts: Date | null;
-  public readonly last_keepalive_ts: Date | null;
-  public readonly last_checkpoint_ts: Date | null;
-  public readonly active: boolean;
-  public readonly storageVersion: number;
-
+export class MongoPersistedSyncRulesContent extends storage.PersistedSyncRulesContent {
   public current_lock: MongoSyncRulesLock | null = null;
 
   constructor(
     private db: PowerSyncMongo,
     doc: mongo.WithId<SyncRuleDocument>
   ) {
-    this.id = doc._id;
-    this.sync_rules_content = doc.content;
-    this.last_checkpoint_lsn = doc.last_checkpoint_lsn;
-    // Handle legacy values
-    this.slot_name = doc.slot_name ?? `powersync_${this.id}`;
-    this.last_fatal_error = doc.last_fatal_error;
-    this.last_fatal_error_ts = doc.last_fatal_error_ts;
-    this.last_checkpoint_ts = doc.last_checkpoint_ts;
-    this.last_keepalive_ts = doc.last_keepalive_ts;
-    this.active = doc.state == 'ACTIVE';
-    this.storageVersion = doc.storage_version ?? storage.LEGACY_STORAGE_VERSION;
+    super({
+      id: doc._id,
+      sync_rules_content: doc.content,
+      last_checkpoint_lsn: doc.last_checkpoint_lsn,
+      // Handle legacy values
+      slot_name: doc.slot_name ?? `powersync_${doc._id}`,
+      last_fatal_error: doc.last_fatal_error,
+      last_fatal_error_ts: doc.last_fatal_error_ts,
+      last_checkpoint_ts: doc.last_checkpoint_ts,
+      last_keepalive_ts: doc.last_keepalive_ts,
+      active: doc.state == 'ACTIVE',
+      storageVersion: doc.storage_version ?? storage.LEGACY_STORAGE_VERSION
+    });
   }
 
-  /**
-   * Load the storage config.
-   *
-   * This may throw if the persisted storage version is not supported.
-   */
   getStorageConfig() {
     const storageConfig = getMongoStorageConfig(this.storageVersion);
     if (storageConfig == null) {
@@ -55,16 +38,6 @@ export class MongoPersistedSyncRulesContent implements storage.PersistedSyncRule
     return storageConfig;
   }
 
-  parsed(options: storage.ParseSyncRulesOptions) {
-    return new MongoPersistedSyncRules(
-      this.id,
-      SqlSyncRules.fromYaml(this.sync_rules_content, options),
-      this.last_checkpoint_lsn,
-      this.slot_name,
-      this.getStorageConfig()
-    );
-  }
-
   async lock() {
     const lock = await MongoSyncRulesLock.createLock(this.db, this);
     this.current_lock = lock;

modules/module-mongodb-storage/src/storage/storage-index.ts

@@ -2,7 +2,6 @@ export * from './implementation/db.js';
 export * from './implementation/models.js';
 export * from './implementation/MongoBucketBatch.js';
 export * from './implementation/MongoIdSequence.js';
-export * from './implementation/MongoPersistedSyncRules.js';
 export * from './implementation/MongoPersistedSyncRulesContent.js';
 export * from './implementation/MongoStorageProvider.js';
 export * from './implementation/MongoSyncBucketStorage.js';

modules/module-mongodb-storage/test/src/storage_compacting.test.ts

@@ -1,7 +1,7 @@
 import { bucketRequest, bucketRequests, register, TEST_TABLE, test_utils } from '@powersync/service-core-tests';
 import { describe, expect, test } from 'vitest';
 import { INITIALIZED_MONGO_STORAGE_FACTORY } from './util.js';
-import { storage, SyncRulesBucketStorage } from '@powersync/service-core';
+import { storage, SyncRulesBucketStorage, updateSyncRulesFromYaml } from '@powersync/service-core';
 
 describe('Mongo Sync Bucket Storage Compact', () => {
   register.registerCompactTests(INITIALIZED_MONGO_STORAGE_FACTORY);
@@ -38,14 +38,14 @@ describe('Mongo Sync Bucket Storage Compact', () => {
 
     const setup = async () => {
       await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY();
-      const syncRules = await factory.updateSyncRules({
-        content: `
+      const syncRules = await factory.updateSyncRules(
+        updateSyncRulesFromYaml(`
 bucket_definitions:
   by_user:
     parameters: select request.user_id() as user_id
     data: [select * from test where owner_id = bucket.user_id]
-    `
-      });
+    `)
+      );
       const bucketStorage = factory.getInstance(syncRules);
       const { checkpoint } = await populate(bucketStorage);
 
@@ -89,14 +89,14 @@ bucket_definitions:
       const { factory } = await setup();
 
       // Not populate another version (bucket definition name changed)
-      const syncRules = await factory.updateSyncRules({
-        content: `
+      const syncRules = await factory.updateSyncRules(
+        updateSyncRulesFromYaml(`
 bucket_definitions:
   by_user2:
     parameters: select request.user_id() as user_id
     data: [select * from test where owner_id = bucket.user_id]
-    `
-      });
+    `)
+      );
       const bucketStorage = factory.getInstance(syncRules);
 
       await populate(bucketStorage);

modules/module-mongodb-storage/test/src/storage_sync.test.ts

@@ -1,4 +1,4 @@
-import { storage } from '@powersync/service-core';
+import { storage, updateSyncRulesFromYaml } from '@powersync/service-core';
 import { bucketRequest, register, TEST_TABLE, test_utils } from '@powersync/service-core-tests';
 import { describe, expect, test } from 'vitest';
 import { INITIALIZED_MONGO_STORAGE_FACTORY, TEST_STORAGE_VERSIONS } from './util.js';
@@ -12,15 +12,17 @@ function registerSyncStorageTests(storageFactory: storage.TestStorageFactory, st
     // but large enough in size to be split over multiple returned chunks.
     // Similar to the above test, but splits over 1MB chunks.
     await using factory = await storageFactory();
-    const syncRules = await factory.updateSyncRules({
-      content: `
+    const syncRules = await factory.updateSyncRules(
+      updateSyncRulesFromYaml(
+        `
     bucket_definitions:
       global:
         data:
           - SELECT id, description FROM "%"
     `,
-      storageVersion
-    });
+        { storageVersion }
+      )
+    );
     const bucketStorage = factory.getInstance(syncRules);
     const globalBucket = bucketRequest(syncRules, 'global[]');
 

modules/module-mongodb/test/src/change_stream_utils.ts

@@ -11,6 +11,7 @@ import {
   STORAGE_VERSION_CONFIG,
   SyncRulesBucketStorage,
   TestStorageOptions,
+  updateSyncRulesFromYaml,
   utils
 } from '@powersync/service-core';
 import { METRICS_HELPER, test_utils } from '@powersync/service-core-tests';
@@ -99,11 +100,9 @@ export class ChangeStreamTestContext {
   }
 
   async updateSyncRules(content: string) {
-    const syncRules = await this.factory.updateSyncRules({
-      content: content,
-      validate: true,
-      storageVersion: this.storageVersion
-    });
+    const syncRules = await this.factory.updateSyncRules(
+      updateSyncRulesFromYaml(content, { validate: true, storageVersion: this.storageVersion })
+    );
     this.syncRulesId = syncRules.id;
     this.storage = this.factory.getInstance(syncRules);
     return this.storage!;

modules/module-mssql/test/src/CDCStreamTestContext.ts

@@ -6,7 +6,8 @@ import {
   LEGACY_STORAGE_VERSION,
   OplogEntry,
   storage,
-  SyncRulesBucketStorage
+  SyncRulesBucketStorage,
+  updateSyncRulesFromYaml
 } from '@powersync/service-core';
 import { METRICS_HELPER, test_utils } from '@powersync/service-core-tests';
 import { clearTestDb, getClientCheckpoint, TEST_CONNECTION_OPTIONS } from './util.js';
@@ -73,11 +74,9 @@ export class CDCStreamTestContext implements AsyncDisposable {
   }
 
   async updateSyncRules(content: string) {
-    const syncRules = await this.factory.updateSyncRules({
-      content: content,
-      validate: true,
-      storageVersion: LEGACY_STORAGE_VERSION
-    });
+    const syncRules = await this.factory.updateSyncRules(
+      updateSyncRulesFromYaml(content, { validate: true, storageVersion: LEGACY_STORAGE_VERSION })
+    );
     this.storage = this.factory.getInstance(syncRules);
     return this.storage!;
   }