feat: auto migration resume with FETCH_ALL (#31)

* First implementation of FETCH_ALL

* Optimize query: do not fetch all inserted ids

* Add type-safety with explicit message to use FETCH_ALL

* Feed doc

* Bump 1.6.0

* chore: remove useless cast

* refactor: improve typings

* remove require()

* [auto] run prettier

---------

Co-authored-by: hugo.prunaux <hugo.prunaux@gmail.com>

Author: pp0rtal

CHANGLOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 1.6.0 (2026-02-25)
+
+- Automatic resume for empty queries using `FETCH_ALL` - https://github.com/360Learning/mongo-bulk-data-migration/pull/31
+
 ## 1.5.2 (2026-02-25)
 
 - Patch nested $set restore - https://github.com/360Learning/mongo-bulk-data-migration/pull/29

README.md

@@ -142,7 +142,7 @@ new MongoBulkDataMigration<Score>({
   id: 'scores_total_new_field',
   collectionName: 'scores',
   projection: { scoreA: 1, scoreB: 1 },
-  query: {},
+  query: FETCH_ALL,
   update: (doc) => {
     $set: {
       total: doc.scoreA + doc.scoreB;
@@ -151,6 +151,24 @@ new MongoBulkDataMigration<Score>({
 });
 ```
 
+### Automatic resume (`query: FETCH_ALL`)
+
+If you don't can't provide a query to fetch only non-migrated documents (like `newField:{$exists:false}`), you can let MBDM continue according the last `_id` processed:
+
+```ts
+import { MongoBulkDataMigration, FETCH_ALL } from "@360-l/mongo-bulk-data-migration";
+...
+new MongoBulkDataMigration<Score>({
+    db,
+    id: "update_all_totals",
+    collectionName: "scores",
+    projection: { total:1 },
+    // Automatic resume, iso {}
+    query: FETCH_ALL,
+    update: () => ({ total: total + 1 }),
+});
+```
+
 ### Delete documents (`update: DELETE_OPERATION`)
 
 This migration will delete doc having negative `total`.

__tests__/MongoBulkDataMigration.rollback.test.ts

@@ -280,6 +280,7 @@ describe('MongoBulkDataMigration', () => {
     describe('rollback twice', () => {
       it('[non idempotent scripts] should restore the initial documents', async () => {
         await collection.insertMany([{ key: 1 }, { key: 2 }, { key: 3 }]);
+        // @ts-expect-error Use FETCH_ALL instead of an empty query {}
         const dataMigration = new MongoBulkDataMigration({
           ...DM_DEFAULT_SETUP,
           query: {},
@@ -316,6 +317,7 @@ describe('MongoBulkDataMigration', () => {
 
       it('[idempotent scripts] should restore the initial documents of the first migration', async () => {
         await collection.insertMany([{ key: 1 }, { key: 2 }, { key: 3 }]);
+        // @ts-expect-error Use FETCH_ALL instead of an empty query {}
         const dataMigration = new MongoBulkDataMigration({
           ...DM_DEFAULT_SETUP,
           query: {},

__tests__/MongoBulkDataMigration.update.test.ts

@@ -1,6 +1,7 @@
 import _ from 'lodash';
-import type { Collection, Db, Document, ObjectId, UpdateFilter } from 'mongodb';
-import { MongoBulkDataMigration, DELETE_OPERATION } from '../src';
+// import { ObjectId } from 'bson';
+import { Collection, Db, Document, ObjectId, UpdateFilter } from 'mongodb';
+import { MongoBulkDataMigration, DELETE_OPERATION, FETCH_ALL } from '../src';
 import { INITIAL_BULK_INFOS } from '../src/lib/AbstractBulkOperationResults';
 import { LoggerInterface } from '../src/types';
 
@@ -523,6 +524,85 @@ describe('MongoBulkDataMigration', () => {
     });
   });
 
+  describe('FETCH_ALL', () => {
+    it('should resume migration only on documents inserted after the last migrated one', async () => {
+      await collection.insertMany([{ key: 1 }, { key: 2 }, { key: 3 }]);
+      const update = { $set: { key: 10 } };
+
+      // First run: migrate all existing documents
+      const firstRun = new MongoBulkDataMigration({
+        ...DM_DEFAULT_SETUP,
+        query: FETCH_ALL,
+        update,
+      });
+      await firstRun.update();
+
+      // ---- Simulation of a migration resume ---
+      // New documents arrive after first migration
+      await collection.insertMany([{ key: 4 }, { key: 5 }]);
+
+      // Second run: resume with FETCH_ALL
+      const secondRun = new MongoBulkDataMigration({
+        ...DM_DEFAULT_SETUP,
+        query: FETCH_ALL,
+        update,
+      });
+      const resumeResults = await secondRun.update();
+
+      expect(resumeResults).toEqual({
+        ...INITIAL_BULK_INFOS,
+        nMatched: 2,
+        nModified: 2,
+      });
+      const documents = await collection
+        .find({}, { projection: { _id: 0 } })
+        .toArray();
+      expect(documents).toEqual([
+        { key: 10 },
+        { key: 10 },
+        { key: 10 },
+        { key: 10 },
+        { key: 10 },
+      ]);
+    });
+
+    // Limitation: the FETCH_ALL relies on the last _id fetch, this is assumed for performances reasons
+    it('[limitation] should not migrate a document with an older _id inserted after first migration', async () => {
+      await collection.insertMany([{ key: 1 }, { key: 2 }, { key: 3 }]);
+      const update = { $set: { key: 10 } };
+
+      // First run: migrate all existing documents
+      const firstRun = new MongoBulkDataMigration({
+        ...DM_DEFAULT_SETUP,
+        query: FETCH_ALL,
+        update,
+      });
+      await firstRun.update();
+
+      // ---- Simulation of a migration resume ---
+      // A document with an old _id is inserted (e.g. restored from a backup)
+      const oldId = new ObjectId('000000000000000000000001');
+      await collection.insertOne({ _id: oldId, key: 99 });
+
+      // Second run: resume with FETCH_ALL — old _id should be skipped
+      const secondRun = new MongoBulkDataMigration({
+        ...DM_DEFAULT_SETUP,
+        query: FETCH_ALL,
+        update,
+      });
+      const resumeResults = await secondRun.update();
+
+      expect(resumeResults).toEqual({
+        ...INITIAL_BULK_INFOS,
+        nMatched: 0,
+        nModified: 0,
+      });
+      // The old document was not migrated
+      const oldDoc = await collection.findOne({ _id: oldId });
+      expect(oldDoc?.key).toEqual(99);
+    });
+  });
+
   describe('#delete', () => {
     it('should perform the delete operation', async () => {
       await collection.insertMany([{ key: 1 }, { key: 2 }, { key: 3 }]);

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@360-l/mongo-bulk-data-migration",
-  "version": "1.5.2",
+  "version": "1.6.0",
   "description": "MongoDB bulk data migration for node scripts",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",

src/MongoBulkDataMigration.ts

@@ -20,13 +20,15 @@ import type {
   RollbackDocument,
   RollBackUpdateObject,
 } from './types';
-import type { Collection, Document, ObjectId, WithId } from 'mongodb';
+import type { Collection, Document, Filter, ObjectId, WithId } from 'mongodb';
 
 const DEFAULT_BULK_SIZE = 5000;
 const COUNT_TOO_LONG_WARNING_THRESHOLD_MS = 30000;
 const COLLECTION_VALIDATION_LEVEL = 'moderate';
 /** Fully delete collection, use with operation:DELETE_COLLECTION */
 export const DELETE_COLLECTION = Symbol();
+/** Fetches all documents excluding already rolled-back ones, use with query:FETCH_ALL */
+export const FETCH_ALL = Symbol();
 const defaultLogger = {
   info: (...args: unknown[]) => {
     if (process.env.NODE_ENV === 'test') {
@@ -43,8 +45,10 @@ const defaultLogger = {
   },
 };
 
-export default class MongoBulkDataMigration<TSchema extends Document>
-  implements RollbackableUpdate
+export default class MongoBulkDataMigration<
+  TSchema extends Document,
+  TQuery extends Filter<TSchema> | typeof FETCH_ALL = Filter<TSchema>,
+> implements RollbackableUpdate
 {
   private readonly options: DataMigrationOptions<TSchema> = {
     arrayFilters: [],
@@ -68,7 +72,12 @@ export default class MongoBulkDataMigration<TSchema extends Document>
    * @see <a href="/doc/softwareDesigns/bulkDataMigration/index.md">More information in the software design document</a>
    * @param config
    */
-  constructor(config: DataMigrationConfig<TSchema>) {
+  constructor(
+    ...args: [keyof TQuery] extends [never]
+      ? ['Use FETCH_ALL instead of an empty query {}']
+      : [config: DataMigrationConfig<TSchema, TQuery>]
+  ) {
+    const config = args[0] as DataMigrationConfig<TSchema, TQuery>;
     this.id = config.id;
     this.collectionName = config.collectionName;
     Object.assign(this.options, { ...config.options });
@@ -123,8 +132,10 @@ export default class MongoBulkDataMigration<TSchema extends Document>
     }
 
     await this.lowerValidationLevel('update');
-    const { cursor, totalEntries } =
-      await this.getCursorAndCount(migrationCollection);
+    const { cursor, totalEntries } = await this.getCursorAndCount(
+      migrationCollection,
+      rollbackCollection,
+    );
     const formattedTotalEntries =
       totalEntries === NO_COUNT_AVAILABLE
         ? 'N/A (dontCount option ON)'
@@ -189,8 +200,13 @@ export default class MongoBulkDataMigration<TSchema extends Document>
     return bulkMigration.getResults();
   }
 
-  private async getCursorAndCount(migrationCollection: Collection<TSchema>) {
-    const cursor = getCursor(this.migrationInfos);
+  private async getCursorAndCount(
+    migrationCollection: Collection<TSchema>,
+    rollbackCollection: Collection<TSchema>,
+  ) {
+    const resolvedQuery = await this.resolveQuery(rollbackCollection);
+
+    const cursor = getCursor(resolvedQuery, this.migrationInfos);
     const countTakingTooLongTimeout = setTimeout(
       () =>
         this.logger.warn(
@@ -201,11 +217,14 @@ export default class MongoBulkDataMigration<TSchema extends Document>
     );
     const totalEntries = this.options.dontCount
       ? NO_COUNT_AVAILABLE
-      : await getTotalEntries(this.migrationInfos);
+      : await getTotalEntries(resolvedQuery);
     clearTimeout(countTakingTooLongTimeout);
     return { cursor, totalEntries };
 
-    function getCursor({ query, projection }: MigrationInfos<TSchema>) {
+    function getCursor(
+      query: Filter<TSchema> | MongoPipeline,
+      { projection }: MigrationInfos<TSchema>,
+    ) {
       if (isPipeline(query)) {
         const pipelineWithProjection = query.concat(
           _.isEmpty(projection) ? [] : [{ $project: projection }],
@@ -215,7 +234,7 @@ export default class MongoBulkDataMigration<TSchema extends Document>
       return migrationCollection.find(query, { projection });
     }
 
-    async function getTotalEntries({ query }: MigrationInfos<TSchema>) {
+    async function getTotalEntries(query: Filter<TSchema> | MongoPipeline) {
       if (isPipeline(query)) {
         const pipelineComputeTotal = query.concat({ $count: 'totalEntries' });
         const cursorComputeTotal =
@@ -283,6 +302,23 @@ export default class MongoBulkDataMigration<TSchema extends Document>
     }
   }
 
+  private async resolveQuery(
+    rollbackCollection: Collection<TSchema>,
+  ): Promise<Filter<TSchema> | MongoPipeline> {
+    if (this.migrationInfos.query !== FETCH_ALL) {
+      return this.migrationInfos.query;
+    }
+    const lastBackup = await rollbackCollection
+      .find({}, { projection: { _id: 1 } })
+      .sort({ _id: -1 })
+      .limit(1)
+      .next();
+
+    return lastBackup
+      ? ({ _id: { $gt: lastBackup._id } } as Filter<TSchema>)
+      : {};
+  }
+
   async rollback(): Promise<BulkOperationResult> {
     if (!this.options.rollbackable) {
       this.logger.warn('Calling rollback() on a non rollbackable script');

src/index.ts

@@ -1,3 +1,3 @@
 export { DELETE_OPERATION } from './lib/MigrationBulk';
-export { DELETE_COLLECTION } from './MongoBulkDataMigration';
+export { DELETE_COLLECTION, FETCH_ALL } from './MongoBulkDataMigration';
 export { default as MongoBulkDataMigration } from './MongoBulkDataMigration';

src/types.ts

@@ -7,7 +7,7 @@ import type {
   Document,
 } from 'mongodb';
 import type { DELETE_OPERATION } from './lib/MigrationBulk';
-import { DELETE_COLLECTION } from './MongoBulkDataMigration';
+import { DELETE_COLLECTION, FETCH_ALL } from './MongoBulkDataMigration';
 
 export type DataMigrationOptions<TSchema> = {
   /** Array filters to use in case of a migration on nested object in arrays */
@@ -52,7 +52,7 @@ export type MigrationInfos<TSchema extends Document> = {
   operation?: typeof DELETE_COLLECTION;
   projection: FindOptions<TSchema>['projection'];
   rollback?: (backup?: RollbackDocument['backup']) => RollBackUpdateObject;
-  query: Filter<TSchema> | MongoPipeline;
+  query: Filter<TSchema> | MongoPipeline | typeof FETCH_ALL;
   update:
     | UpdateFilter<TSchema>
     | typeof DELETE_OPERATION
@@ -61,16 +61,19 @@ export type MigrationInfos<TSchema extends Document> = {
       ) => Promise<UpdateFilter<TSchema>> | UpdateFilter<TSchema>);
 };
 
-export type DataMigrationConfig<TSchema extends Document> =
+export type DataMigrationConfig<
+  TSchema extends Document,
+  TQuery extends Filter<TSchema> | typeof FETCH_ALL = Filter<TSchema>,
+> =
   | DMInstanceSpecialOperation<TSchema>
-  | DMInstanceSpecialOperationDropDocument<TSchema>
+  | DMInstanceSpecialOperationDropDocument<TSchema, TQuery>
   | DMInstanceAggregate<TSchema>
-  | DMInstanceFilter<TSchema>;
+  | DMInstanceFilter<TSchema, TQuery>;
 
-type DMInstanceSpecialOperationDropDocument<TSchema> = Omit<
-  DMInstanceFilter<TSchema>,
-  'projection'
-> & {
+type DMInstanceSpecialOperationDropDocument<
+  TSchema extends Document,
+  TQuery extends Filter<TSchema> | typeof FETCH_ALL = Filter<TSchema>,
+> = Omit<DMInstanceFilter<TSchema, TQuery>, 'projection'> & {
   update: typeof DELETE_OPERATION;
 };
 
@@ -85,13 +88,15 @@ export type DMInstanceSpecialOperation<TSchema> = Pick<
   operation: typeof DELETE_COLLECTION;
 };
 
-export type DMInstanceFilter<TSchema extends Document> =
-  DMInstanceBase<TSchema> & {
-    /** Projected properties (and backed up values) */
-    projection: FindOptions<TSchema>['projection'];
-    /** Mongo query for documents to migrate OR mongo aggregation pipeline */
-    query: Filter<TSchema>;
-  };
+export type DMInstanceFilter<
+  TSchema extends Document,
+  TQuery extends Filter<TSchema> | typeof FETCH_ALL = Filter<TSchema>,
+> = DMInstanceBase<TSchema> & {
+  /** Projected properties (and backed up values) */
+  projection: FindOptions<TSchema>['projection'];
+  /** Mongo query for documents to migrate, or FETCH_ALL to resume excluding already-migrated documents */
+  query: TQuery;
+};
 
 type DMInstanceBase<TSchema> = {
   /** Targeted collection */