Replication maybe working.

Author: rkistner

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

@@ -1,3 +1,4 @@
+export * from './migrations/SlateDBMigrationAgent.js';
 export * from './module/SlateDBStorageModule.js';
 
 export * from './storage/storage-index.js';

modules/module-slatedb-storage/src/migrations/SlateDBMigrationAgent.ts

@@ -0,0 +1,46 @@
+import { framework, migrations } from '@powersync/service-core';
+
+class SlateDBMigrationLockManager extends framework.locks.AbstractLockManager {
+  private locked = false;
+
+  protected async acquireHandle(): Promise<framework.locks.LockHandle | null> {
+    if (this.locked) {
+      return null;
+    }
+    this.locked = true;
+    return {
+      refresh: async () => {},
+      release: async () => {
+        this.locked = false;
+      }
+    };
+  }
+}
+
+export class SlateDBMigrationAgent extends migrations.AbstractPowerSyncMigrationAgent {
+  readonly locks = new SlateDBMigrationLockManager({ name: 'slatedb_migrations' });
+
+  private state: framework.MigrationState | undefined;
+
+  readonly store: framework.MigrationStore = {
+    load: async () => this.state,
+    save: async (state) => {
+      this.state = state;
+    },
+    clear: async () => {
+      this.state = undefined;
+    }
+  };
+
+  getInternalScriptsDir(): string {
+    throw new Error('SlateDB storage has no internal migrations.');
+  }
+
+  override async loadInternalMigrations(): Promise<framework.Migration<migrations.PowerSyncMigrationContext>[]> {
+    return [];
+  }
+
+  async [Symbol.asyncDispose](): Promise<void> {
+    this.state = undefined;
+  }
+}

modules/module-slatedb-storage/src/module/SlateDBStorageModule.ts

@@ -1,6 +1,8 @@
 import { modules, system } from '@powersync/service-core';
 
+import { SlateDBMigrationAgent } from '../migrations/SlateDBMigrationAgent.js';
 import { SlateDBStorageProvider } from '../storage/storage-index.js';
+import { isSlateDBStorageConfig } from '../types/types.js';
 
 export class SlateDBStorageModule extends modules.AbstractModule {
   constructor() {
@@ -11,6 +13,10 @@ export class SlateDBStorageModule extends modules.AbstractModule {
 
   async initialize(context: system.ServiceContextContainer): Promise<void> {
     context.storageEngine.registerProvider(new SlateDBStorageProvider());
+
+    if (isSlateDBStorageConfig(context.configuration.storage)) {
+      context.migrations.registerMigrationAgent(new SlateDBMigrationAgent());
+    }
   }
 
   async teardown(): Promise<void> {

modules/module-slatedb-storage/src/storage/SlateDBBucketStorageFactory.ts

@@ -10,6 +10,7 @@ import {
   SlateDBPersistedReplicationStream,
   SlateDBReplicationStreamRecord
 } from './SlateDBPersistedSyncConfigContent.js';
+import { SlateDBSyncBucketStorage } from './SlateDBSyncBucketStorage.js';
 
 export type SlateDBBucketStorageOptions = {
   config: SlateDBStorageConfigDecoded;
@@ -21,6 +22,8 @@ export class SlateDBBucketStorageFactory extends storage.BucketStorageFactory {
 
   readonly replicationStreamNamePrefix: string;
   private store: Promise<SlateDBKVStore> | undefined;
+  private resolvedStore: SlateDBKVStore | undefined;
+  private activeStorageCache: SlateDBSyncBucketStorage | undefined;
 
   constructor(readonly options: SlateDBBucketStorageOptions) {
     super();
@@ -31,14 +34,27 @@ export class SlateDBBucketStorageFactory extends storage.BucketStorageFactory {
     this.store ??= SlateDBKVStore.open({
       path: this.options.config.path
     });
-    return this.store;
+    this.resolvedStore ??= await this.store;
+    return this.resolvedStore;
   }
 
   getInstance(
-    _replicationStream: storage.PersistedReplicationStream,
+    replicationStream: storage.PersistedReplicationStream,
     _options?: storage.GetIntanceOptions
   ): storage.SyncRulesBucketStorage {
-    throw notImplemented('getInstance');
+    if (!(replicationStream instanceof SlateDBPersistedReplicationStream)) {
+      throw new Error(`Expected SlateDBPersistedReplicationStream`);
+    }
+    if (this.activeStorageCache?.replicationStreamId == replicationStream.replicationStreamId) {
+      return this.activeStorageCache;
+    }
+    const store = this.resolvedStore;
+    if (store == null) {
+      throw new Error(`SlateDB store has not been initialized`);
+    }
+    const instance = new SlateDBSyncBucketStorage(this, store, replicationStream);
+    this.activeStorageCache = instance;
+    return instance;
   }
 
   async updateSyncRules(_options: storage.UpdateSyncRulesOptions): Promise<storage.PersistedReplicationStream> {
@@ -174,6 +190,7 @@ export class SlateDBBucketStorageFactory extends storage.BucketStorageFactory {
       const store = await this.store;
       await store[Symbol.asyncDispose]();
       this.store = undefined;
+      this.resolvedStore = undefined;
     }
   }
 

modules/module-slatedb-storage/src/storage/SlateDBKVStore.ts

@@ -63,10 +63,11 @@ export class SlateDBKVStore implements AsyncDisposable {
   ) {}
 
   static async open(options: SlateDBKVStoreOptions): Promise<SlateDBKVStore> {
-    await fs.mkdir(options.path, { recursive: true });
+    const rootPath = path.resolve(options.path);
+    await fs.mkdir(rootPath, { recursive: true });
 
     const objectStoreUrl = LOCAL_FILE_OBJECT_STORE_URL;
-    const dbPath = path.join(options.path, options.dbPath ?? DEFAULT_DB_PATH).replace(/^\/+/, '');
+    const dbPath = path.join(rootPath, options.dbPath ?? DEFAULT_DB_PATH).replace(/^\/+/, '');
     const objectStore = ObjectStore.resolve(objectStoreUrl);
 
     const dbBuilder = new DbBuilder(dbPath, objectStore);
@@ -93,6 +94,15 @@ export class SlateDBKVStore implements AsyncDisposable {
     await this.db.delete(toKeyBytes(key));
   }
 
+  async deletePrefix(prefix: SlateDBKey, options: { limit?: number } = {}): Promise<number> {
+    const deletes: SlateDBWriteOperation[] = [];
+    for await (const entry of this.scanPrefix(prefix, options)) {
+      deletes.push({ type: 'delete', key: entry.key });
+    }
+    await this.write(deletes);
+    return deletes.length;
+  }
+
   async write(operations: SlateDBWriteOperation[]): Promise<void> {
     if (operations.length == 0) {
       return;

modules/module-slatedb-storage/src/storage/SlateDBPersistedSyncConfigContent.ts

@@ -8,6 +8,10 @@ export interface SlateDBReplicationStreamRecord {
   state: storage.SyncRuleState;
   storageVersion: number;
   syncConfig: SlateDBSyncConfigRecord;
+  next_op_id?: string;
+  last_persisted_op?: string;
+  resume_lsn?: string | null;
+  no_checkpoint_before_lsn?: string | null;
   snapshot_done?: boolean;
   last_checkpoint_lsn: string | null;
   last_fatal_error?: string | null;