Inbox Pattern

The transactional outbox solves the producer's dual-write problem, atomic database write and event publish. But what about the consumer? The consumer has the mirror problem: it receives a message from the broker and needs to process it (write to its own database) and acknowledge it. If it processes but crashes before acknowledging, the message is redelivered, a duplicate.

The inbox pattern solves this. The consumer stores incoming messages in an "inbox" table within its own database. Processing the message and recording its arrival happen in the same database transaction. If the message ID already exists in the inbox, it's a duplicate and is skipped.

Think of the outbox as the sender's "sent mail" folder and the inbox as the receiver's "received mail" folder. Together, they form a reliable messaging pipeline: the outbox guarantees the event is published, and the inbox guarantees it's processed exactly once.

async function handleMessage(msg: IncomingMessage) {
 await db.transaction(async (tx) => {
 // 1. Check inbox for duplicate
 const existing = await tx.query(
 'SELECT 1 FROM inbox WHERE message_id = $1', [msg.id]
 );
 
 if (existing.rows.length > 0) {
 // Already processed, acknowledge and skip
 await msg.ack();
 return;
 }
 
 // 2. Record in inbox
 await tx.query(
 'INSERT INTO inbox (message_id, received_at, payload) VALUES ($1, NOW(), $2)',
 [msg.id, JSON.stringify(msg.body)]
 );
 
 // 3. Process business logic (in same transaction)
 await tx.query(
 'UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2',
 [msg.body.quantity, msg.body.productId]
 );
 });
 
 // 4. Acknowledge after successful commit
 await msg.ack();
}

The critical sequence:

1. Begin database transaction
2. Check inbox for existing message ID → skip if duplicate
3. Insert message ID into inbox
4. Execute business logic
5. Commit transaction (atomic: inbox insert + business logic)
6. Acknowledge message to broker

If the consumer crashes after step 5 but before step 6, the broker redelivers the message. But step 2 detects it's already in the inbox and skips reprocessing. No duplicate side effect.

Inbox table schema:

CREATE TABLE inbox (
 message_id VARCHAR(255) PRIMARY KEY,
 received_at TIMESTAMP NOT NULL DEFAULT NOW(),
 payload JSONB,
 processed_at TIMESTAMP
);

-- Index for cleanup queries
CREATE INDEX idx_inbox_received_at ON inbox(received_at);