How to Build a Shopify Webhook Replay System That Never Loses Events

Shopify
   |
   v
[Ingestion Endpoint]   <-- ACK immediately, store raw payload
   |
   v
[Event Store (DB)]     <-- durable record of every event
   |
   v
[Processing Queue]     <-- background worker picks up events
   |
   v
[Handler]              <-- idempotent processing logic
   |
   +-- Success --> mark 'succeeded'
   +-- Failure --> retry with backoff --> Dead Letter Queue

app.post('/webhooks', express.raw({ type: 'application/json' }), async (req, res) => {
  const hmac = req.headers['x-shopify-hmac-sha256'];
  const webhookId = req.headers['x-shopify-webhook-id'];
  const topic = req.headers['x-shopify-topic'];

  // Verify HMAC first
  if (!verifyHmac(req.body, hmac)) {
    return res.status(401).send('Unauthorized');
  }

  // Store immediately, respond fast
  await db.query(`
    INSERT INTO webhook_events (shopify_id, topic, payload, hmac, status)
    VALUES ($1, $2, $3, $4, 'pending')
    ON CONFLICT (shopify_id) DO NOTHING
  `, [webhookId, topic, req.body.toString(), hmac]);

  res.status(200).send('OK');
});

CREATE TABLE webhook_events (
  id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  shopify_id      VARCHAR(255) UNIQUE,
  topic           VARCHAR(100) NOT NULL,
  payload         JSONB NOT NULL,
  hmac            VARCHAR(255),
  status          VARCHAR(20) DEFAULT 'pending',
  retry_count     INT DEFAULT 0,
  last_error      TEXT,
  created_at      TIMESTAMPTZ DEFAULT NOW(),
  processed_at    TIMESTAMPTZ
);

CREATE INDEX idx_webhook_status ON webhook_events(status);
CREATE INDEX idx_webhook_topic  ON webhook_events(topic);
CREATE INDEX idx_webhook_created ON webhook_events(created_at);

async function processWebhookEvents() {
  const events = await db.query(`
    SELECT * FROM webhook_events
    WHERE status = 'pending'
    ORDER BY created_at ASC
    LIMIT 50
    FOR UPDATE SKIP LOCKED
  `);

  for (const event of events.rows) {
    try {
      await handleEvent(event.topic, JSON.parse(event.payload));

      await db.query(`
        UPDATE webhook_events
        SET status = 'succeeded', processed_at = NOW()
        WHERE id = $1
      `, [event.id]);

    } catch (err) {
      const newRetryCount = event.retry_count + 1;
      const newStatus = newRetryCount >= 5 ? 'dead' : 'pending';

      await db.query(`
        UPDATE webhook_events
        SET status = $1, retry_count = $2, last_error = $3
        WHERE id = $4
      `, [newStatus, newRetryCount, err.message, event.id]);
    }
  }
}

function getRetryDelay(retryCount) {
  const baseDelays = [30, 120, 600, 3600, 21600]; // seconds
  const base = (baseDelays[retryCount] || 21600) * 1000;
  const jitter = Math.random() * 1000;
  return base + jitter;
}

async function replayEvents({ topic, status = 'failed', from, to, batchSize = 100 }) {
  const events = await db.query(`
    SELECT * FROM webhook_events
    WHERE status = $1
      AND ($2::text IS NULL OR topic = $2)
      AND ($3::timestamptz IS NULL OR created_at >= $3)
      AND ($4::timestamptz IS NULL OR created_at <= $4)
    ORDER BY created_at ASC
    LIMIT $5
  `, [status, topic, from, to, batchSize]);

  for (const event of events.rows) {
    // Reset to pending so the worker picks it up
    await db.query(`
      UPDATE webhook_events
      SET status = 'pending', retry_count = 0, last_error = NULL
      WHERE id = $1
    `, [event.id]);
  }

  return events.rowCount;
}

// Usage: replay all failed orders/paid from the last 24 hours
await replayEvents({
  topic: 'orders/paid',
  status: 'failed',
  from: new Date(Date.now() - 86400000)
});

async function handleOrderCreate(payload) {
  const shopifyOrderId = payload.id;

  // Check if already processed
  const existing = await db.query(
    'SELECT id FROM orders WHERE shopify_order_id = $1',
    [shopifyOrderId]
  );

  if (existing.rows.length > 0) {
    console.log(`Order ${shopifyOrderId} already processed. Skipping.`);
    return;
  }

  // Safe to process
  await db.query(
    'INSERT INTO orders (shopify_order_id, ...) VALUES ($1, ...)',
    [shopifyOrderId, ...]
  );
}

// Query DLQ events
const dlqEvents = await db.query(`
  SELECT topic, COUNT(*) as count, MIN(created_at) as oldest
  FROM webhook_events
  WHERE status = 'dead'
  GROUP BY topic
  ORDER BY count DESC
`);