Idempotent Consumers and the Inbox Pattern

async def handle(event: dict) -> None:
    async with db.transaction():
        inserted = await db.fetchval(
            "INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING RETURNING 1",
            event["id"])
        if not inserted:
            return                       # duplicate: already applied
        await apply_payment(db, event)   # side effects in the same transaction
# ack the message only after the transaction commits

@KafkaListener(topics = "payments.captured")
@Transactional
public void on(PaymentCaptured event) {
    if (inbox.markProcessed(event.eventId()) == 0) return;   // INSERT ... ON CONFLICT DO NOTHING
    ledger.recordCapture(event);
}

export async function handle(event: PaymentCaptured) {
  await db.$transaction(async (tx) => {
    const r = await tx.$executeRaw`INSERT INTO processed_events (event_id) VALUES (${event.id}) ON CONFLICT DO NOTHING`;
    if (r === 0) return;                 // duplicate
    await tx.ledgerEntry.create({ data: toEntry(event) });
  });
}