中断後の変更ストリームの再開

// A resume token is the _id field of a change event
// Example structure (your values will differ):
// {
//   _id: {
//     _data: '8264CE3B7200000002463C6F5A100...'
//   }
// }

// Store it as-is — do not parse or modify the _data value

const CHECKPOINT_COLLECTION = 'changeStreamCheckpoints';
const STREAM_ID = 'orderProcessorV1';

for await (const change of changeStream) {
  // Process the event
  await processOrderChange(change);

  // Save token AFTER successful processing
  await db.collection(CHECKPOINT_COLLECTION).updateOne(
    { streamId: STREAM_ID },
    { $set: { resumeToken: change._id, processedAt: new Date() } },
    { upsert: true }
  );
}

async function startOrResumeStream() {
  // Load the last saved resume token
  const checkpoint = await db.collection('changeStreamCheckpoints')
    .findOne({ streamId: 'orderProcessorV1' });

  const watchOptions = checkpoint
    ? { resumeAfter: checkpoint.resumeToken }
    : {}; // Start from current position if no token

  const changeStream = db.collection('orders').watch([], watchOptions);

  for await (const change of changeStream) {
    await processOrderChange(change);
    await saveResumeToken(change._id);
  }
}

// resumeAfter — standard use case, does not work after invalidate tokens
db.collection('orders').watch([], { resumeAfter: savedToken });

// startAfter — can also start after an invalidate event
// (use when the collection may have been dropped and recreated)
db.collection('orders').watch([], { startAfter: savedToken });

const { Timestamp } = require('mongodb');

// Replay events since a specific time
const startTime = new Timestamp({ t: Math.floor(Date.now() / 1000) - 3600, i: 1 }); // 1 hour ago

const changeStream = db.collection('orders').watch([], {
  startAtOperationTime: startTime
});

// Events from 1 hour ago will be delivered

async function startOrResumeWithFallback() {
  const checkpoint = await loadResumeToken();
  try {
    const options = checkpoint ? { resumeAfter: checkpoint } : {};
    const stream = db.collection('orders').watch([], options);
    for await (const change of stream) {
      await processChange(change);
      await saveResumeToken(change._id);
    }
  } catch (error) {
    if (error.code === 286) { // ChangeStreamHistoryLost
      console.warn('Resume token expired — starting from now and running catch-up scan');
      await deleteResumeToken();
      await catchUpScan(); // full collection scan to catch missed changes
      await startOrResumeWithFallback();
    } else {
      throw error;
    }
  }
}

async function processOrderChange(change) {
  if (change.operationType === 'insert') {
    const { fullDocument } = change;
    // Idempotent upsert — safe to replay:
    await db.collection('orderSummaries').updateOne(
      { _id: fullDocument._id },       // match by _id
      { $set: { ...summarize(fullDocument) } }, // idempotent set
      { upsert: true }                 // create if not exists
    );
  }
}

// The driver automatically resumes after network blips
// No code needed in your loop for this case:
for await (const change of changeStream) {
  // If the network drops and reconnects, the driver resumes automatically
  // and continues delivering events from where it left off.
  await processChange(change);
  await saveCheckpoint(change._id); // Still save tokens for process restarts
}

class ReliableChangeStreamConsumer {
  constructor(collection, handler) {
    this.collection = collection;
    this.handler = handler;
    this.running = false;
  }

  async start() {
    this.running = true;
    while (this.running) {
      const token = await loadToken();
      const stream = this.collection.watch([], token ? { resumeAfter: token } : {});
      try {
        for await (const change of stream) {
          await this.handler(change);
          await saveToken(change._id);
        }
      } catch (e) {
        if (e.code === 286) { await deleteToken(); continue; } // expired, restart fresh
        if (!this.running) break; // shutting down
        throw e;
      } finally {
        await stream.close();
      }
    }
  }

  stop() { this.running = false; }
}

// Correct: save token AFTER processing, not before
for await (const change of changeStream) {
  await processChange(change);        // process first
  await saveToken(change._id);        // token saved only after success
  // If crash happens here, the event was already processed — safe
}

// Wrong: saving token before processing
for await (const change of changeStream) {
  await saveToken(change._id);        // token saved
  await processChange(change);        // if crash here, event is skipped on resume!
}