Idempotency and Safe Reprocessing

class DeduplicationStore {
  async hasProcessed(eventKey) {
    return Boolean(await this.db.findByKey(eventKey));
  }

  async recordProcessed(eventKey, metadata = {}) {
    await this.db.insert({
      key: eventKey,
      processedAt: new Date(),
      ...metadata,
    });
  }
}

// Usage in the event processor
async function processEvent(event) {
  const key = buildEventKey(event);

  if (await dedupStore.hasProcessed(key)) {
    console.log(`Already processed ${key} — skipping`);
    return;
  }

  await performSideEffects(event);
  await dedupStore.recordProcessed(key, { eventType: event.eventType });
}

function buildEventKey(event) {
  return `${event.eventType}:${event.contextId}`;
}

import crypto from 'crypto';

function buildEventKey(event) {
  const stableSubset = {
    eventType: event.eventType,
    resourceId: event.payload?.id,
    modifiedDate: event.payload?.modifiedDate,
  };
  return crypto
    .createHash('sha256')
    .update(JSON.stringify(stableSubset))
    .digest('hex');
}

function buildEventKey(event) {
  const ts = event.payload?.modifiedDate ?? event.payload?.modifiedDateTime ?? '';
  return `${event.eventType}:${event.contextId}:${ts}`;
}

async function recordProcessed(key) {
  await db.insert({
    key,
    processedAt: new Date(),
    expiresAt: new Date(Date.now() + 7 * 24 * 60 * 60 * 1000),
  });
}

// Periodic cleanup
async function cleanupExpiredDedupEntries() {
  await db.deleteWhere({ expiresAt: { lt: new Date() } });
}

async function processEvent(event) {
  const key = buildEventKey(event);

  try {
    await db.insert('dedup', { key, processedAt: new Date() });
  } catch (err) {
    if (isUniqueConstraintError(err)) {
      console.log(`Event ${key} already being processed — skipping`);
      return;
    }
    throw err;
  }

  // We won the insert — safe to process
  await performSideEffects(event);
}

async function processEvent(event) {
  const key = buildEventKey(event);

  try {
    await db.insert('dedup', { key, status: 'processing', startedAt: new Date() });
  } catch (err) {
    if (isUniqueConstraintError(err)) return;
    throw err;
  }

  try {
    await performSideEffects(event);
    await db.update('dedup', { key }, { status: 'completed', completedAt: new Date() });
  } catch (err) {
    // Processing failed — remove the dedup record so retries can succeed
    await db.delete('dedup', { key });
    throw err;
  }
}

async function processEvent(event) {
  const key = buildEventKey(event);

  await db.transaction(async (tx) => {
    const existing = await tx.findByKey('dedup', key);
    if (existing) {
      console.log(`Already processed ${key}`);
      return;
    }

    await performSideEffects(event, tx);
    await tx.insert('dedup', { key, processedAt: new Date() });
  });
}

async function notifySlack(event) {
  const eventKey = buildEventKey(event);

  await slackApi.postMessage({
    channel: '#donations',
    text: `New gift: ${event.payload.formattedAmount}`,
    idempotency_key: `raise-${eventKey}`,
  });
}

async function syncGiftToDownstream(gift) {
  // Upsert keyed by Raise gift ID — duplicate calls produce a single row
  await db.upsert('gifts', {
    raiseGiftId: gift.id,
    amount: gift.amount,
    date: gift.date,
    donorEmail: gift.donor?.email,
    lastSyncedAt: new Date(),
  });
}

async function sendThankYouEmail(event) {
  const key = buildEventKey(event);

  if (await emailDedup.hasSent(key)) return;

  await emailService.send({ to: event.payload.donor.email, ... });
  await emailDedup.recordSent(key);
}

async function reprocessEvent(webhookId, logId) {
  // 1. Get the original payload
  const logResponse = await fetch(
    `https://prod-api.raisedonors.com/api/Webhook/${webhookId}/log/${logId}`,
    { headers: { Authorization: `Bearer ${process.env.RAISE_API_TOKEN}` } }
  );
  const log = await logResponse.json();
  const payload = JSON.parse(log.payLoad);

  // 2. Remove the dedup record so processing can proceed
  const key = buildEventKey({ eventType: log.eventType, contextId: log.contextId });
  await dedupStore.remove(key);

  // 3. Reprocess
  await processEvent({ eventType: log.eventType, payload });
}

async function reprocessWindow(webhookId, startDate, endDate) {
  // Paginate through logs for the time window
  const logs = await streamLogsForWindow(webhookId, startDate, endDate);

  for (const log of logs) {
    if (log.success !== 'Yes' && log.success !== 'True') continue;

    try {
      await reprocessEvent(webhookId, log.id);
    } catch (err) {
      console.error(`Reprocess failed for log ${log.id}:`, err);
    }
  }
}

console.log('Skipping duplicate event', {
  key: eventKey,
  eventType: event.eventType,
  contextId: event.contextId,
  originalProcessedAt: existingRecord.processedAt,
  thisAttemptAt: new Date(),
});

test('processing the same event twice produces no duplicate side effects', async () => {
  const event = buildTestEvent({ giftId: 9876 });

  await processEvent(event);
  await processEvent(event); // Same event again

  expect(downstream.giftRecords).toHaveLength(1);
  expect(emailService.sentEmails).toHaveLength(1);
});