Change Detection Best Practices

// ❌ Anti-pattern: advance early
await setCheckpoint(customerId, 'user_sync', latestSeen); // What if processing crashes?
for (const user of users) {
  await processUser(user); // Crash here → checkpoint is past these users
}

// ✅ Correct: advance after all processing succeeds
let latestSeen = currentCheckpoint;
for (const user of users) {
  await processUser(user);
  const u = new Date(user.updated_at);
  if (u > latestSeen) latestSeen = u;
}
await setCheckpoint(customerId, 'user_sync', latestSeen); // After processing

async function setCheckpointWithBackup(customerId, resource, checkpoint) {
  // Primary store
  await db.upsert('sync_checkpoints', {
    customer_id: customerId,
    resource,
    checkpoint,
  });

  // Backup store (different region, different provider)
  await backupStore.set(`checkpoint:${customerId}:${resource}`, checkpoint);
}

// Destination has an upsert operation keyed by external ID
await externalSystem.upsertUser({
  external_id: `vomo-${user.id}`,
  email: user.email,
  first_name: user.first_name,
  last_name: user.last_name,
});

async function sendWelcomeEmailOnce(userId) {
  const key = `welcome_email:${userId}`;
  const alreadySent = await externalDb.dedupExists(key);
  if (alreadySent) return;

  await emailService.sendWelcome(userId);
  await externalDb.recordDedup(key);
}

async function updateUserIfNewer(externalUserId, sourceUpdatedAt, changes) {
  const current = await externalSystem.getUser(externalUserId);
  if (new Date(current.lastSyncedAt) >= new Date(sourceUpdatedAt)) {
    // Already up-to-date or newer — skip
    return { skipped: true };
  }

  await externalSystem.updateUser(externalUserId, { ...changes, lastSyncedAt: sourceUpdatedAt });
  return { updated: true };
}

async function pollUserChanges(customerId) {
  const traceId = generateTraceId();
  logger.info('Poll started', { customerId, traceId, resource: 'users' });

  try {
    const result = await doPoll(customerId, traceId);
    logger.info('Poll completed', { customerId, traceId, result });
    return result;
  } catch (err) {
    logger.error('Poll failed', { customerId, traceId, error: err.message });
    throw err;
  }
}

async function processUserChange(customerId, user, traceId) {
  await db.recordProcessing({
    customer_id: customerId,
    resource: 'user',
    record_id: user.id,
    record_updated_at: user.updated_at,
    processed_at: new Date(),
    trace_id: traceId,
    outcome: 'success', // or 'failed' on error path
  });
}

async function replayPollCycle(customerId, fromCheckpoint, toCheckpoint) {
  // Run the polling logic against the time window, but emit to a dry-run destination
  const dryRun = new DryRunDestination();

  const result = await runPollLogic({
    customerId,
    fromCheckpoint,
    toCheckpoint,
    destination: dryRun,
  });

  return {
    result,
    operationsThatWouldHaveRun: dryRun.recordedOperations,
  };
}