Incremental migration

Author: jmgasper

data-migration/README.md

@@ -23,6 +23,7 @@ This tool provides a robust framework for migrating data from JSON files to a Po
 - **Detailed Logging**: Configurable logging levels
 - **Migration Statistics**: Comprehensive reporting of migration results
 - **Validation Testing**: Verify data integrity after migration
+- **Incremental Updates**: Date-filtered migrations with selective field updates for efficient data synchronization
 
 ## Project Structure
 
@@ -99,39 +100,88 @@ npx prisma migrate dev
 ```
 
 ### Running the Migration
-```
+```bash
+# Run full migration (default)
 npm run migrate
 
+# Run incremental migration with date filter
+MIGRATION_MODE=incremental INCREMENTAL_SINCE_DATE=2024-01-15T00:00:00Z npm run migrate
+
+# Run incremental migration with selective field updates
+MIGRATION_MODE=incremental INCREMENTAL_SINCE_DATE=2024-01-15T00:00:00Z INCREMENTAL_FIELDS=status,updatedAt npm run migrate
+```
+
+For more details on incremental migrations, see the `Incremental Updates` section below.
+
+```bash
 # Additional commands
 npm run migrate:reset # Reset the db and run the migration tool
 ```
+
 ### Configuration Options
-You can configure the migration behavior through environment variables:
+
+The migration tool is configurable through environment variables. You can set these in your `.env` file or pass them directly on the command line.
+
+**Database Configuration**
 ```
-# Database connection
 DATABASE_URL=postgresql://username:password@localhost:5432/database_name
+```
 
-# Migration settings
+**Migration Settings**
+```
 DATA_DIRECTORY=./data
 BATCH_SIZE=100
 CONCURRENCY_LIMIT=10
 LOG_LEVEL=info
+```
 
-# Migration behavior
+**Migration Behavior**
+```
 SKIP_MISSING_REQUIRED=false
 USE_TRANSACTIONS=true
 CHALLENGE_COUNTERS_ONLY=false
+MIGRATION_MODE=full
+INCREMENTAL_SINCE_DATE=
+INCREMENTAL_FIELDS=
 MIGRATORS_ONLY=
+```
 
-# Migration attribution
+**Migration Attribution**
+```
 CREATED_BY=migration
 UPDATED_BY=migration
 ```
-`SKIP_MISSING_REQUIRED` skips the record if required fields are missing. When `false`, default values for required fields must be configured in `src/config.js`  
-Logfiles are by default stored in `logs/migration.log`  
-It can be configured using the env variable `LOG_FILE`  
-Log levels(increasing level of information): `error`, `warn`, `info`, `debug`  
-Further migration configuration can also be done in `src/config.js`
+
+`SKIP_MISSING_REQUIRED` skips the record if required fields are missing. When `false`, default values for required fields must be configured in `src/config.js`.  
+`MIGRATION_MODE` controls the migration strategy. Set to `full` for complete data loads or `incremental` for date-filtered updates. Defaults to `full`. See the `Incremental Updates` section below for detailed usage.  
+`INCREMENTAL_SINCE_DATE` specifies the cutoff date for incremental migrations (ISO 8601 format, e.g., `2024-01-15T00:00:00Z`). Only records with `updatedAt` or `updated` fields after this date are processed. Required when `MIGRATION_MODE=incremental`.  
+`INCREMENTAL_FIELDS` is an optional comma-separated list of field names (e.g., `status,updatedAt,name`) that restricts which fields are updated during incremental migrations. When omitted, all fields are updated. The fields `updatedAt` and `updatedBy` are always included. Useful for targeted updates like status changes or counter refreshes.  
+Logfiles are by default stored in `logs/migration.log`.  
+You can set a custom location with the `LOG_FILE` environment variable.  
+Log levels (in increasing verbosity): `error`, `warn`, `info`, `debug`.  
+Further migration configuration can also be done in `src/config.js`.
+
+## Incremental Updates
+
+Incremental updates let you run a full migration once and then keep the database in sync with smaller, targeted refreshes. After the initial load, you can filter subsequent runs to only process records changed after a specific date. The migrators handle both updates to existing rows and insertion of new records while leaving untouched data in place. This will help cut down on the time needed to migrate the data on the final cutover date.
+
+### How It Works
+1. **Date filtering**: Only records with `updatedAt` or `updated` values later than `INCREMENTAL_SINCE_DATE` are loaded into memory.
+2. **Selective field updates**: When `INCREMENTAL_FIELDS` is set, only those fields are updated on matching database records; otherwise, all fields are considered for updates.
+3. **Upsert behavior**: New records are inserted in full, while existing records receive partial or full updates based on your configuration.
+
+All standard validation rules, dependency checks, and relational guarantees remain in effect while running in incremental mode.
+
+### Configuration
+- `MIGRATION_MODE`: Set to `incremental` to enable this workflow (defaults to `full`).
+- `INCREMENTAL_SINCE_DATE`: ISO 8601 timestamp that defines the cutoff date (e.g., `2024-01-15T00:00:00Z`). Only records updated after this value are processed.
+- `INCREMENTAL_FIELDS`: Optional comma-separated list to limit which fields are updated (e.g., `status,updatedAt,name`). When omitted, all fields are updated; when set, the tool automatically includes `updatedAt` and `updatedBy`.
+
+### Limitations
+- Records without `updatedAt` or `updated` fields will be skipped in incremental mode (a warning is logged).
+- The `INCREMENTAL_FIELDS` configuration applies globally to all migrators; model-specific field lists are not currently supported.
+- Deleted records in the source data are not removed from the database; incremental mode only handles updates and inserts.
+- Dependency validation still requires related records to exist; ensure dependent models are included in the incremental run.
 
 ### Updating Challenge Counters Only
 

data-migration/src/config.js

@@ -7,6 +7,13 @@ const parseListEnv = value => {
   return list.length ? list : null;
 };
 
+const ALLOWED_MIGRATION_MODES = new Set(['full', 'incremental']);
+const parseMigrationMode = value => {
+  if (!value) return 'full';
+  const normalized = String(value).toLowerCase();
+  return ALLOWED_MIGRATION_MODES.has(normalized) ? normalized : 'full';
+};
+
 // Default configuration with fallbacks
 module.exports = {
   // Database connection
@@ -21,6 +28,11 @@ module.exports = {
   // Migration behavior
   SKIP_MISSING_REQUIRED: process.env.SKIP_MISSING_REQUIRED === 'true',
   USE_TRANSACTIONS: process.env.USE_TRANSACTIONS !== 'false',
+
+  // Incremental migration settings
+  MIGRATION_MODE: parseMigrationMode(process.env.MIGRATION_MODE),
+  INCREMENTAL_SINCE_DATE: process.env.INCREMENTAL_SINCE_DATE || null,
+  INCREMENTAL_FIELDS: parseListEnv(process.env.INCREMENTAL_FIELDS),
 
   // Migration attribution
   CREATED_BY: process.env.CREATED_BY || 'migration',
@@ -409,3 +421,8 @@ module.exports = {
     },
   }, 
 };
+
+Object.defineProperty(module.exports, 'parseMigrationMode', {
+  value: parseMigrationMode,
+  enumerable: false
+});

data-migration/src/migrationManager.js

@@ -1,20 +1,43 @@
 const { PrismaClient, Prisma } = require('@prisma/client');
-const config = require('./config');
+const configModule = require('./config');
 const fs = require('fs');
 const path = require('path');
 
+const { parseMigrationMode } = configModule;
+const baseConfig = { ...configModule };
+
+if (typeof parseMigrationMode === 'function') {
+  delete baseConfig.parseMigrationMode;
+}
+
 /**
  * Core migration manager that orchestrates the migration process
  */
 class MigrationManager {
   constructor(userConfig = {}) {
+    const overrideConfig = { ...userConfig };
+    let pendingMigrationModeWarning = null;
+
+    if (Object.prototype.hasOwnProperty.call(overrideConfig, 'MIGRATION_MODE') && typeof parseMigrationMode === 'function') {
+      const rawMigrationMode = overrideConfig.MIGRATION_MODE;
+      const normalizedMode = parseMigrationMode(rawMigrationMode);
+      overrideConfig.MIGRATION_MODE = normalizedMode;
+
+      const isNullishOverride = rawMigrationMode === undefined || rawMigrationMode === null;
+      const matchesAllowedString = typeof rawMigrationMode === 'string' && rawMigrationMode.toLowerCase() === normalizedMode;
+
+      if (!isNullishOverride && !matchesAllowedString) {
+        pendingMigrationModeWarning = rawMigrationMode;
+      }
+    }
+
     // Merge default config with user-provided config
     this.config = {
-      ...config,
-      ...userConfig,
+      ...baseConfig,
+      ...overrideConfig,
       defaultValues: {
-        ...config.defaultValues,
-        ...(userConfig.defaultValues || {})
+        ...baseConfig.defaultValues,
+        ...(overrideConfig.defaultValues || {})
       }
     };
 
@@ -29,6 +52,20 @@ class MigrationManager {
     this.dependencies = {};
     // this.logger = this.createLogger(this.config.LOG_LEVEL);
     this.logger = this.createLogger(this.config.LOG_LEVEL, this.config.LOG_FILE || 'migration.log');
+    if (pendingMigrationModeWarning) {
+      this.logger.warn(`Invalid MIGRATION_MODE "${pendingMigrationModeWarning}" provided. Defaulting to 'full'.`);
+    }
+
+    if (this.config.MIGRATION_MODE === 'incremental') {
+      const sinceDate = this.config.INCREMENTAL_SINCE_DATE;
+
+      if (!sinceDate) {
+        this.logger.warn('Incremental migration mode is set but no INCREMENTAL_SINCE_DATE is configured.');
+      } else if (Number.isNaN(Date.parse(sinceDate))) {
+        this.logger.warn(`INCREMENTAL_SINCE_DATE "${sinceDate}" is not a valid ISO date string and will be ignored by subsequent phases.`);
+        this.config.INCREMENTAL_SINCE_DATE = null;
+      }
+    }
     this.migrators = [];
   }
 
@@ -112,6 +149,13 @@ class MigrationManager {
     return this.dependencies[dependencyModel]?.has(id) || false;
   }
 
+  /**
+   * Determine if the migration is running in incremental mode
+   */
+  isIncrementalMode() {
+    return this.config.MIGRATION_MODE === 'incremental';
+  }
+
   /**
    * Stores nested data for a specific model
    * @param {String} modelName 
@@ -303,4 +347,4 @@ class MigrationManager {
   }
 }
 
-module.exports = { MigrationManager };
+module.exports = { MigrationManager };

data-migration/src/migrators/_baseMigrator.js

@@ -36,8 +36,22 @@ class BaseMigrator {
   /**
    * Load data for this migrator
    * @returns {Array} The loaded data
-   */
+  */
   async loadData() {
+    const isIncremental = this.manager.isIncrementalMode();
+    const sinceDate = this.manager.config.INCREMENTAL_SINCE_DATE ?? null;
+
+    if (isIncremental) {
+      this.manager.logger.debug(`Loading ${this.modelName} data with incremental filter since ${sinceDate || 'unspecified date'}`);
+      return await loadData(
+        this.manager.config.DATA_DIRECTORY,
+        this.getFileName(),
+        this.isElasticsearch,
+        sinceDate
+      );
+    }
+
+    this.manager.logger.debug(`Loading ${this.modelName} data without incremental filtering`);
     return await loadData(this.manager.config.DATA_DIRECTORY, this.getFileName(), this.isElasticsearch);
   }
 
@@ -47,15 +61,31 @@ class BaseMigrator {
    */
   async migrate() {
     this.manager.logger.info(`Migrating ${this.modelName} data...`);
+    const isIncremental = this.manager.isIncrementalMode();
+    const sinceDate = this.manager.config.INCREMENTAL_SINCE_DATE ?? 'unspecified date';
+    const incrementalFields = Array.isArray(this.manager.config.INCREMENTAL_FIELDS)
+      ? this.manager.config.INCREMENTAL_FIELDS
+      : [];
+
+    if (isIncremental) {
+      this.manager.logger.info(`Running in INCREMENTAL mode since ${sinceDate}`);
+      if (incrementalFields.length) {
+        this.manager.logger.info(`Updating only fields: ${incrementalFields.join(', ')}`);
+      }
+    } else {
+      this.manager.logger.info('Running in FULL migration mode');
+    }
+
     const data = await this.loadData();
 
     // Allow subclasses to perform pre-processing
     const processedData = await this.beforeMigration(data);
 
     const processFn = this.createProcessFunction();
     const result = await this.manager.processBatch(processedData, processFn);
-    
-    this.manager.logger.info(`Migrated ${result.processed} ${this.modelName} records (skipped ${result.skipped})`);
+
+    const modeLabel = isIncremental ? 'incremental' : 'full';
+    this.manager.logger.info(`Migrated ${result.processed} ${this.modelName} records (skipped ${result.skipped}) in ${modeLabel} mode`);
 
     // Allow subclasses to perform post-processing
     await this.afterMigration(result);
@@ -93,10 +123,16 @@ class BaseMigrator {
     return async (batch, prisma, uniqueTracker) => {
       let processed = 0;
       let skipped = 0;
-      
+
       // Initialize unique trackers if needed
       this.initializeUniqueTrackers(uniqueTracker);
-      
+
+      if (this.manager.isIncrementalMode()) {
+        this.manager.logger.debug(`Processing ${this.modelName} batch in incremental mode`);
+      } else {
+        this.manager.logger.debug(`Processing ${this.modelName} batch in full migration mode`);
+      }
+
       for (const _record of batch) {
 
         const record = this.beforeValidation(_record);
@@ -137,7 +173,9 @@ class BaseMigrator {
         const finalModelData = this.customizeRecordData(modelData);
 
         // Create upsert data
-        const upsertData = this.createUpsertData(finalModelData, this.getIdField());
+        const upsertData = this.manager.isIncrementalMode()
+          ? this.createIncrementalUpsertData(finalModelData, this.getIdField())
+          : this.createUpsertData(finalModelData, this.getIdField());
 
         // Allow subclasses to modify upsert data if needed
         const finalUpsertData = this.customizeUpsertData(upsertData, record);
@@ -243,6 +281,61 @@ class BaseMigrator {
     };
   }
 
+  /**
+   * Create upsert data when running in incremental mode. Only configured incremental fields
+   * are updated while new records still receive the full dataset.
+   * @param {Object} record The processed record data
+   * @param {string} idField The ID field name
+   * @returns {{ where: Object, update: Object, create: Object }} The incremental upsert payload
+   */
+  createIncrementalUpsertData(record, idField) {
+    const incrementalFields = Array.isArray(this.manager.config.INCREMENTAL_FIELDS)
+      ? this.manager.config.INCREMENTAL_FIELDS
+      : [];
+
+    if (!incrementalFields.length) {
+      return this.createUpsertData(record, idField);
+    }
+
+    const updateData = {};
+    const missingFields = [];
+
+    for (const field of incrementalFields) {
+      if (field === idField) {
+        continue;
+      }
+
+      if (record[field] !== undefined) {
+        updateData[field] = record[field];
+      } else {
+        updateData[field] = Prisma.skip;
+        missingFields.push(field);
+      }
+    }
+
+    updateData.updatedAt = record.updatedAt ? new Date(record.updatedAt) : new Date();
+    updateData.updatedBy = record.updatedBy;
+
+    if (missingFields.length) {
+      this._missingIncrementalFieldWarnings = this._missingIncrementalFieldWarnings || new Set();
+      for (const field of missingFields) {
+        if (!this._missingIncrementalFieldWarnings.has(field)) {
+          this._missingIncrementalFieldWarnings.add(field);
+          this.manager.logger.warn(`Configured incremental field "${field}" is missing on some ${this.modelName} records; skipping updates for this field`);
+        }
+      }
+    }
+
+    const createData = { ...record };
+    createData.createdAt = record.createdAt ? new Date(record.createdAt) : new Date();
+
+    return {
+      where: { [idField]: record[idField] },
+      update: updateData,
+      create: createData
+    };
+  }
+
   /**
    * Initialize unique trackers for this model
    */
@@ -384,4 +477,4 @@ class BaseMigrator {
   }
 }
 
-module.exports = { BaseMigrator };
+module.exports = { BaseMigrator };