Idempotent Message Handlers: Surviving Retries

def generate_idempotency_key(event: dict, producer_message_id: str) -> str:
    """
    Combine producer ID with business context for debugging and scoping.
    """
    return f"{event['aggregate_type']}:{event['aggregate_id']}:{producer_message_id}"

# Example usage:
event = {
    "aggregate_type": "Order",
    "aggregate_id": "12345"
}
msg_id = "msg-a1b2c3d4-e5f6-7890"

key = generate_idempotency_key(event, msg_id)
print(key)

# Output: Order:12345:msg-a1b2c3d4-e5f6-7890

// Redis-based idempotency store with atomic check-and-lock
// Note: The GET-then-SET pattern below has a race window between the two calls.
// For high-concurrency scenarios, use a Lua script to make the entire operation atomic,
// or accept that concurrent handlers may both attempt processing (one will fail on markCompleted).
class IdempotencyStore {
  constructor(private redis: Redis) {}

  async checkAndLock(key: string, ttlSeconds: number): Promise<{
    isDuplicate: boolean;
    previousResult?: any;
    lockAcquired: boolean;
  }> {
    const existing = await this.redis.get(`idempotency:${key}`);

    if (existing) {
      const record = JSON.parse(existing);
      if (record.status === 'completed') {
        return { isDuplicate: true, previousResult: record.result, lockAcquired: false };
      }
      if (record.status === 'processing') {
        return { isDuplicate: true, previousResult: undefined, lockAcquired: false };
      }
    }

    // Atomic set-if-not-exists with TTL
    const acquired = await this.redis.set(
      `idempotency:${key}`,
      JSON.stringify({ status: 'processing', createdAt: new Date() }),
      'EX', ttlSeconds,
      'NX'
    );

    return { isDuplicate: false, lockAcquired: acquired === 'OK' };
  }

  async markCompleted(key: string, result: any, ttlSeconds: number): Promise<void> {
    await this.redis.set(
      `idempotency:${key}`,
      JSON.stringify({ status: 'completed', result, completedAt: new Date() }),
      'EX', ttlSeconds
    );
  }

  async markFailed(key: string): Promise<void> {
    // Delete lock so retry can proceed
    await this.redis.del(`idempotency:${key}`);
  }
}

-- Idempotency as unique constraint on business table (PostgreSQL)
CREATE TABLE payments (
    id SERIAL PRIMARY KEY,
    idempotency_key VARCHAR(255) UNIQUE NOT NULL,
    order_id INTEGER NOT NULL,
    amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(20) NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT NOW()
);

-- Atomic check-and-insert: duplicate key silently returns existing row
INSERT INTO payments (idempotency_key, order_id, amount, status)
VALUES ($1, $2, $3, 'pending')
ON CONFLICT (idempotency_key) DO UPDATE
SET status = payments.status  -- No-op update to return existing row
RETURNING id, (xmax = 0) as was_inserted;

-- was_inserted = true means new row, false means duplicate detected

// Base class for idempotent message handlers
abstract class IdempotentHandler<T, R> {
  constructor(
    protected idempotencyStore: IdempotencyStore,
    protected config: { ttlSeconds: number }
  ) {}

  async handle(message: Message<T>): Promise<HandlerResult<R>> {
    const key = this.getIdempotencyKey(message);

    // Step 1: Check for duplicate
    const check = await this.idempotencyStore.checkAndLock(key, this.config.ttlSeconds);

    if (check.isDuplicate) {
      if (check.previousResult !== undefined) {
        return { success: true, result: check.previousResult, wasRetry: true };
      }
      throw new ConcurrentProcessingError(`Message ${message.id} being processed`);
    }

    // Step 2: Process (subclass implements this)
    try {
      const result = await this.process(message);
      await this.idempotencyStore.markCompleted(key, result, this.config.ttlSeconds);
      return { success: true, result, wasRetry: false };
    } catch (error) {
      await this.idempotencyStore.markFailed(key);
      if (this.isRetryable(error as Error)) throw error;
      return { success: false, error: error as Error, wasRetry: false };
    }
  }

  protected getIdempotencyKey(message: Message<T>): string {
    return message.id;
  }

  protected abstract process(message: Message<T>): Promise<R>;

  protected isRetryable(error: Error): boolean {
    return error instanceof TransientError;
  }
}

-- Outbox table for reliable event publishing (PostgreSQL)
CREATE TABLE outbox (
    id VARCHAR(255) PRIMARY KEY,           -- Use message idempotency key
    aggregate_type VARCHAR(100) NOT NULL,  -- e.g., 'Order', 'Payment'
    aggregate_id VARCHAR(255) NOT NULL,    -- Business entity ID
    event_type VARCHAR(100) NOT NULL,      -- e.g., 'OrderPaid', 'PaymentFailed'
    payload JSONB NOT NULL,                -- Event data
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    published_at TIMESTAMP                 -- NULL until successfully published
);

CREATE INDEX idx_outbox_unpublished ON outbox (created_at) WHERE published_at IS NULL;

// Handler: update state + insert outbox in single transaction
async processPayment(orderId: string, paymentId: string, amount: number): Promise<void> {
  const check = await this.idempotencyStore.checkAndLock(`payment:${paymentId}`, 7 * 24 * 60 * 60);
  if (check.isDuplicate) return;

  await this.db.transaction(async (tx) => {
    // Update business state
    await tx.query(
      `UPDATE orders SET status = 'paid', paid_amount = $2, payment_id = $3
       WHERE id = $1 AND status = 'pending'`,
      [orderId, amount, paymentId]
    );

    // Insert outbox event (same transaction)
    await tx.query(
      `INSERT INTO outbox (id, aggregate_type, aggregate_id, event_type, payload, created_at)
       VALUES ($1, 'Order', $2, 'OrderPaid', $3, NOW())`,
      [paymentId, orderId, JSON.stringify({ orderId, amount, paymentId })]
    );
  });

  await this.idempotencyStore.markCompleted(`payment:${paymentId}`, { orderId, status: 'paid' }, 7 * 24 * 60 * 60);
}

// Publisher: poll outbox and publish to message queue
async publishPendingEvents(): Promise<void> {
  const pending = await this.db.query(
    `SELECT * FROM outbox WHERE published_at IS NULL
     ORDER BY created_at LIMIT 100 FOR UPDATE SKIP LOCKED`
  );

  for (const record of pending.rows) {
    try {
      await this.messageQueue.publish(record.event_type, {
        id: record.id,  // Idempotency key for downstream consumers
        payload: record.payload,
      });
      await this.db.query(`UPDATE outbox SET published_at = NOW() WHERE id = $1`, [record.id]);
    } catch (error) {
      // Will retry on next poll - idempotent downstream handles duplicates
      console.error(`Failed to publish ${record.id}`, error);
    }
  }
}

// Saga with idempotent step execution
class IdempotentSaga {
  async executeStep(sagaId: string, stepIndex: number, stepFn: () => Promise<any>): Promise<any> {
    const saga = await this.loadSaga(sagaId);
    const step = saga.steps[stepIndex];

    // Idempotency: skip if already completed
    if (step.status === 'completed') return step.result;
    if (step.status === 'compensated') throw new Error(`Step ${step.name} already compensated`);

    try {
      const result = await stepFn();
      await this.updateStepStatus(sagaId, stepIndex, { status: 'completed', result, completedAt: new Date() });
      return result;
    } catch (error) {
      await this.updateStepStatus(sagaId, stepIndex, { status: 'failed', error: (error as Error).message });
      throw error;
    }
  }

  async compensateStep(sagaId: string, stepIndex: number, compensateFn: () => Promise<void>): Promise<void> {
    const saga = await this.loadSaga(sagaId);
    const step = saga.steps[stepIndex];

    if (step.status !== 'completed') return;  // Nothing to compensate
    if (step.status === 'compensated') return;  // Already compensated

    await this.updateStepStatus(sagaId, stepIndex, { status: 'compensating' });
    try {
      await compensateFn();
      await this.updateStepStatus(sagaId, stepIndex, { status: 'compensated' });
    } catch (error) {
      await this.updateStepStatus(sagaId, stepIndex, { status: 'failed', error: `Compensation failed: ${(error as Error).message}` });
      throw error;
    }
  }
}

// Order fulfillment saga: reserve -> charge -> ship
const orderSaga = new IdempotentSaga();

async function fulfillOrder(orderId: string): Promise<void> {
  const sagaId = `order-fulfillment:${orderId}`;

  try {
    // Step 0: Reserve inventory (compensation: release)
    await orderSaga.executeStep(sagaId, 0, () => inventoryService.reserve(orderId));

    // Step 1: Charge payment (compensation: refund)
    await orderSaga.executeStep(sagaId, 1, () => paymentService.charge(orderId));

    // Step 2: Ship order (compensation: cancel shipment)
    await orderSaga.executeStep(sagaId, 2, () => shippingService.ship(orderId));
  } catch (error) {
    // Payment or shipping failed - compensate in reverse order
    await orderSaga.compensateStep(sagaId, 1, () => paymentService.refund(orderId));
    await orderSaga.compensateStep(sagaId, 0, () => inventoryService.release(orderId));
    throw error;
  }
}

import { SQS } from 'aws-sdk';

// SQS FIFO queue provides built-in deduplication
// But only within 5-minute window!

class SQSFIFOHandler {
  constructor(private sqs: SQS) {}

  async sendWithDeduplication(
    queueUrl: string,
    message: any,
    deduplicationId: string,  // Your idempotency key
    messageGroupId: string    // For ordering
  ): Promise<void> {
    await this.sqs.sendMessage({
      QueueUrl: queueUrl,
      MessageBody: JSON.stringify(message),
      MessageDeduplicationId: deduplicationId,  // SQS dedupes on this
      MessageGroupId: messageGroupId,
    }).promise();

    // SQS will reject duplicate deduplicationId within 5 minutes
    // After 5 minutes, duplicates are accepted!
  }
}

// Important: SQS deduplication is NOT enough for idempotency
// It only prevents duplicate sends within 5 minutes
// You still need handler-side idempotency for:
// - Redeliveries after visibility timeout
// - Messages sent > 5 minutes apart
// - Consumer-side retries

class SQSConsumer {
  async processMessage(sqsMessage: SQS.Message): Promise<void> {
    const message = JSON.parse(sqsMessage.Body!);

    // DO NOT rely solely on SQS MessageId - changes on redelivery
    // Use your own idempotency key from message content
    const idempotencyKey = message.idempotencyKey;

    // Still need application-level idempotency
    const check = await this.idempotencyStore.checkAndLock(idempotencyKey, 7 * 24 * 60 * 60);

    if (check.isDuplicate) {
      // ACK without processing
      await this.sqs.deleteMessage({
        QueueUrl: this.queueUrl,
        ReceiptHandle: sqsMessage.ReceiptHandle!,
      }).promise();
      return;
    }

    // Process...
  }
}

import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';

// Kafka exactly-once: idempotent producer + transactional consumer
// But consumer idempotency is still your responsibility

class IdempotentKafkaConsumer {
  private consumer: Consumer;

  async processMessage({ topic, partition, message }: EachMessagePayload): Promise<void> {
    // Kafka provides unique offset per partition
    // Can use topic:partition:offset as idempotency key
    // BUT: rebalances can cause re-processing of same offset

    const kafkaOffset = `${topic}:${partition}:${message.offset}`;
    const payload = JSON.parse(message.value!.toString());

    // Prefer application-level idempotency key if available
    const idempotencyKey = payload.idempotencyKey || kafkaOffset;

    // Pattern: Store processed offsets in database
    // Commit offset only after database transaction completes
    await this.db.transaction(async (tx) => {
      // Check if already processed
      const existing = await tx.query(
        'SELECT 1 FROM processed_offsets WHERE key = $1',
        [idempotencyKey]
      );

      if (existing.rows.length > 0) {
        return;  // Already processed
      }

      // Process business logic
      await this.processBusinessLogic(tx, payload);

      // Mark as processed (same transaction)
      await tx.query(
        'INSERT INTO processed_offsets (key, processed_at) VALUES ($1, NOW())',
        [idempotencyKey]
      );
    });

    // Offset commit happens automatically after handler returns
    // If handler throws, offset not committed, message redelivered
  }
}

// Kafka idempotent producer (prevents duplicate sends)
const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092'],
});

const producer = kafka.producer({
  idempotent: true,  // Enable idempotent producer
  maxInFlightRequests: 5,  // Required for idempotent
  transactionalId: 'my-transactional-producer',  // For transactions
});

describe('PaymentHandler idempotency', () => {
  let handler: PaymentHandler;
  let idempotencyStore: IdempotencyStore;
  let paymentGateway: MockPaymentGateway;

  beforeEach(() => {
    idempotencyStore = new InMemoryIdempotencyStore();
    paymentGateway = new MockPaymentGateway();
    handler = new PaymentHandler(idempotencyStore, paymentGateway);
  });

  it('should process message exactly once', async () => {
    const message = createPaymentMessage({ amount: 100 });

    // First call - should process
    await handler.handle(message);
    expect(paymentGateway.chargeCount).toBe(1);

    // Second call - should be idempotent
    await handler.handle(message);
    expect(paymentGateway.chargeCount).toBe(1);  // Still 1!

    // Third call - still idempotent
    await handler.handle(message);
    expect(paymentGateway.chargeCount).toBe(1);
  });

  it('should return same result on duplicate', async () => {
    const message = createPaymentMessage({ amount: 100 });

    const result1 = await handler.handle(message);
    const result2 = await handler.handle(message);

    expect(result2).toEqual(result1);
    expect(result2.wasRetry).toBe(true);
  });

  it('should allow retry after failure', async () => {
    const message = createPaymentMessage({ amount: 100 });

    // First call fails
    paymentGateway.shouldFail = true;
    await expect(handler.handle(message)).rejects.toThrow();
    expect(paymentGateway.chargeCount).toBe(1);

    // Retry should attempt again (not cached failure)
    paymentGateway.shouldFail = false;
    await handler.handle(message);
    expect(paymentGateway.chargeCount).toBe(2);
  });

  it('should handle concurrent duplicate delivery', async () => {
    const message = createPaymentMessage({ amount: 100 });

    // Simulate concurrent processing
    const results = await Promise.allSettled([
      handler.handle(message),
      handler.handle(message),
      handler.handle(message),
    ]);

    // Only one should succeed with actual processing
    const successes = results.filter(r => r.status === 'fulfilled');
    expect(paymentGateway.chargeCount).toBe(1);

    // Others should either succeed with cached result or fail with concurrent error
  });

  it('should not dedupe different messages', async () => {
    const message1 = createPaymentMessage({ id: 'pay-1', amount: 100 });
    const message2 = createPaymentMessage({ id: 'pay-2', amount: 200 });

    await handler.handle(message1);
    await handler.handle(message2);

    expect(paymentGateway.chargeCount).toBe(2);  // Both processed
  });
});

// Chaos testing for idempotency
describe('PaymentHandler chaos tests', () => {
  it('should maintain idempotency under random failures', async () => {
    const chaosGateway = new ChaosPaymentGateway({ failureRate: 0.2 });
    const handler = new PaymentHandler(
      new ChaosIdempotencyStore({ failureRate: 0.3 }),
      chaosGateway
    );

    const message = createPaymentMessage({ amount: 100 });

    // Attempt processing 100 times with random failures
    for (let i = 0; i < 100; i++) {
      try {
        await handler.handle(message);
      } catch {
        // Expected under chaos - idempotency store or gateway failed
      }
    }

    // Verify external effect happened exactly once despite failures
    expect(chaosGateway.successfulCharges).toBe(1);
  });
});