Build a video webhook handler that survives duplicates (Express + Postgres)

// app.js: THE BUG
const express = require('express');
const app = express();
app.use(express.json());          // ❌ consumes + reparses the body globally

app.post('/webhooks/video', (req, res) => {
  // req.body is parsed; the raw bytes are gone; signature check will always fail
});

// app.js: THE FIX
const express = require('express');
const app = express();

// raw buffer ONLY for the webhook route, registered before any json() middleware
app.post('/webhooks/video',
  express.raw({ type: 'application/json' }),
  handleVideoWebhook
);

app.use(express.json()); // normal parsing for everything else

// verify.js
const crypto = require('crypto');

function verifySignature(rawBody, header, secret) {
  // header like: "t=1735689600,v1=4f2c...". Adapt to your provider's format.
  const parts = Object.fromEntries(header.split(',').map(kv => kv.split('=')));
  const timestamp = parts.t;
  const sent = parts.v1;
  if (!timestamp || !sent) return false;

  // reject old timestamps (replay protection): 5 min tolerance
  const ageSec = Math.abs(Date.now() / 1000 - Number(timestamp));
  if (ageSec > 300) return false;

  const signedPayload = `${timestamp}.${rawBody.toString('utf8')}`;
  const expected = crypto
    .createHmac('sha256', secret)
    .update(signedPayload)
    .digest('hex');

  const a = Buffer.from(sent, 'hex');
  const b = Buffer.from(expected, 'hex');
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

module.exports = { verifySignature };

// handleVideoWebhook.js (top half)
const { verifySignature } = require('./verify');
const SECRET = process.env.WEBHOOK_SECRET;

async function handleVideoWebhook(req, res) {
  const sigHeader = req.get('webhook-signature') || '';
  if (!verifySignature(req.body, sigHeader, SECRET)) {
    return res.status(401).send('bad signature');   // reject, don't process
  }
  const event = JSON.parse(req.body.toString('utf8'));
  // ... step 3 below
}

-- schema.sql
CREATE TABLE webhook_events (
  event_id    text PRIMARY KEY,        -- provider's unique event id
  type        text NOT NULL,
  raw         jsonb NOT NULL,          -- persist the payload for replay/debug
  received_at timestamptz NOT NULL DEFAULT now(),
  processed   boolean NOT NULL DEFAULT false
);

// handleVideoWebhook.js (continued)
const { pool } = require('./db');
const { enqueue } = require('./queue');

  // inside handleVideoWebhook, after JSON.parse:
  const eventId = event.id;
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    // INSERT ... ON CONFLICT DO NOTHING returns 0 rows if we've seen it
    const ins = await client.query(
      `INSERT INTO webhook_events (event_id, type, raw)
       VALUES ($1, $2, $3) ON CONFLICT (event_id) DO NOTHING`,
      [eventId, event.type, event]
    );

    if (ins.rowCount === 0) {
      await client.query('COMMIT');
      return res.status(200).send('duplicate ignored');   // idempotent no-op
    }

    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }

// handleVideoWebhook.js (end)
  await enqueue({ eventId, type: event.type });   // durable queue (SQS/BullMQ/etc.)
  return res.status(202).send('accepted');
}

// worker.js
const { dequeue } = require('./queue');
const { pool } = require('./db');

async function processEvent({ eventId }) {
  const { rows } = await pool.query(
    'SELECT raw, processed FROM webhook_events WHERE event_id = $1', [eventId]
  );
  if (!rows.length || rows[0].processed) return;     // gone or already done

  const event = rows[0].raw;
  switch (event.type) {
    case 'video.asset.ready':
      await markVideoReady(event.data.asset_id, event.data.playback_id);
      break;
    case 'video.asset.errored':
      await markVideoFailed(event.data.asset_id, event.data.error);
      break;
    // ordering is NOT guaranteed: reconcile by asset_id, don't assume sequence
  }

  await pool.query(
    'UPDATE webhook_events SET processed = true WHERE event_id = $1', [eventId]
  );
}

dequeue(processEvent);

❌ Returning 500 on a business error
   A 500 tells the provider to RETRY. If the event itself is bad
   (malformed asset, constraint violation), you've made a poison
   message that retries forever and floods your logs.
   ✅ Ack 200 once persisted; handle business failures in the worker
      with your own retries + a dead-letter queue.

❌ Treating webhooks as the only source of truth
   If your endpoint is down for an hour, those events can be lost
   for good. ✅ Run a periodic job that lists assets from the
   provider and reconciles against your DB.

# replay a captured event twice; second should be a no-op
curl -X POST localhost:3000/webhooks/video \
  -H 'content-type: application/json' \
  -H "webhook-signature: t=$(date +%s),v1=$(...)" \
  --data-binary @event.json
# → 202 accepted
# run it again → 200 duplicate ignored