[WIP] Stream during snapshot

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

@@ -1,2 +1,3 @@
 export * from './Logger.js';
 export { Logger } from 'winston';
+export { createLogger, format, transports } from 'winston';

libs/lib-services/src/migrations/AbstractMigrationAgent.ts

@@ -1,11 +1,12 @@
 import { LockManager } from '../locks/LockManager.js';
-import { logger } from '../logger/Logger.js';
+import { logger as defaultLogger, Logger } from '../logger/logger-index.js';
 import * as defs from './migration-definitions.js';
 
 export type MigrationParams<Generics extends MigrationAgentGenerics = MigrationAgentGenerics> = {
   count?: number;
   direction: defs.Direction;
   migrationContext?: Generics['MIGRATION_CONTEXT'];
+  logger?: Logger;
 };
 
 type WriteLogsParams = {
@@ -20,10 +21,12 @@ export type MigrationAgentGenerics = {
 export type RunMigrationParams<Generics extends MigrationAgentGenerics = MigrationAgentGenerics> = MigrationParams & {
   migrations: defs.Migration<Generics['MIGRATION_CONTEXT']>[];
   maxLockWaitMs?: number;
+  logger?: Logger;
 };
 
 type ExecuteParams = RunMigrationParams & {
   state?: defs.MigrationState;
+  logger: Logger;
 };
 
 export const DEFAULT_MAX_LOCK_WAIT_MS = 3 * 60 * 1000; // 3 minutes
@@ -46,9 +49,11 @@ export abstract class AbstractMigrationAgent<Generics extends MigrationAgentGene
   async run(params: RunMigrationParams) {
     await this.init();
 
+    const logger = params.logger ?? defaultLogger;
+
     const { direction, migrations, migrationContext } = params;
     // Only one process should execute this at a time.
-    logger.info('Acquiring lock for migrations');
+    logger.debug('Acquiring lock for migrations');
     const lockHandle = await this.locks.acquire({ max_wait_ms: params.maxLockWaitMs ?? DEFAULT_MAX_LOCK_WAIT_MS });
 
     if (!lockHandle) {
@@ -75,22 +80,24 @@ export abstract class AbstractMigrationAgent<Generics extends MigrationAgentGene
         direction,
         migrations,
         state,
-        migrationContext
+        migrationContext,
+        logger
       });
 
       await this.writeLogsToStore({
         log_stream: logStream,
         state
       });
     } finally {
-      logger.info('Releasing migration lock');
+      logger.debug('Releasing migration lock');
       await releaseLock();
       process.removeListener('beforeExit', releaseLock);
-      logger.info('Done with migrations');
+      logger.debug('Done with migrations');
     }
   }
 
   protected async *execute(params: ExecuteParams): AsyncGenerator<defs.ExecutedMigration> {
+    const logger = params.logger;
     const internalMigrations = await this.loadInternalMigrations();
     let migrations = [...internalMigrations, ...params.migrations];
 

modules/module-mongodb-storage/src/migrations/db/migrations/1764667093139-current-data-cleanup.ts

@@ -0,0 +1,43 @@
+import { migrations } from '@powersync/service-core';
+import * as storage from '../../../storage/storage-index.js';
+import { MongoStorageConfig } from '../../../types/types.js';
+
+const INDEX_NAME = 'pending_delete';
+
+export const up: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    await db.current_data.createIndex(
+      {
+        '_id.g': 1,
+        pending_delete: 1
+      },
+      {
+        partialFilterExpression: { pending_delete: { $exists: true } },
+        name: INDEX_NAME
+      }
+    );
+  } finally {
+    await db.client.close();
+  }
+};
+
+export const down: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    if (await db.current_data.indexExists(INDEX_NAME)) {
+      await db.current_data.dropIndex(INDEX_NAME);
+    }
+  } finally {
+    await db.client.close();
+  }
+};