Webhook Handler Examples

const express = require('express');
const crypto = require('crypto');
const app = express();

// Important: preserve raw body for signature verification
app.use(express.json({
  verify: (req, res, buf) => {
    req.rawBody = buf.toString();
  }
}));

const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

function verifySignature(payload, signature, secret) {
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(payload)
    .digest('hex');
  
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  );
}

// Idempotency tracking (use Redis in production)
const processedEvents = new Set();

app.post('/webhooks/ai-interview', async (req, res) => {
  // 1. Verify signature
  const signature = req.headers['x-webhook-signature'];
  if (!verifySignature(req.rawBody, signature, WEBHOOK_SECRET)) {
    console.error('Invalid signature');
    return res.status(401).send('Invalid signature');
  }

  // 2. Check idempotency
  const eventKey = `${req.body.event}:${req.body.session_id}`;
  if (processedEvents.has(eventKey)) {
    console.log('Already processed:', eventKey);
    return res.status(200).send('Already processed');
  }

  // 3. Respond quickly (queue for async processing)
  res.status(200).send('OK');

  // 4. Process asynchronously
  processedEvents.add(eventKey);
  
  try {
    await processWebhook(req.body);
  } catch (error) {
    console.error('Webhook processing failed:', error);
    processedEvents.delete(eventKey); // Retry on next delivery
  }
});

async function processWebhook(payload) {
  const { event, session_id, invitee_email } = payload;

  if (event === 'interview.completed') {
    console.log(`Interview completed for ${invitee_email}`);
    
    // Download transcript
    if (payload.transcript_url) {
      const transcriptRes = await fetch(payload.transcript_url);
      const transcript = await transcriptRes.json();
      
      // Store in database
      await db.interviews.create({
        session_id,
        participant_email: invitee_email,
        transcript,
        duration_minutes: payload.total_minutes,
        completed_at: payload.timestamp,
      });
    }
    
    // Notify team via Slack
    await notifySlack({
      text: `New interview completed: ${invitee_email}`,
      session_id,
    });
    
  } else if (event === 'session.feedback_submitted') {
    console.log(`Feedback submitted for ${session_id}`);
    
    await db.feedback.create({
      session_id,
      ...payload.feedback,
    });
  }
}

async function notifySlack({ text, session_id }) {
  await fetch(process.env.SLACK_WEBHOOK_URL, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      text,
      blocks: [
        {
          type: 'section',
          text: {
            type: 'mrkdwn',
            text: `*${text}*\nSession: \`${session_id}\``,
          },
        },
      ],
    }),
  });
}

app.listen(3000, () => {
  console.log('Webhook server running on port 3000');
});

import hmac
import hashlib
import os
from flask import Flask, request
from datetime import datetime

app = Flask(__name__)

WEBHOOK_SECRET = os.environ['WEBHOOK_SECRET'].encode()

# Idempotency tracking (use Redis in production)
processed_events = set()

def verify_signature(payload, signature):
    expected_signature = hmac.new(
        WEBHOOK_SECRET,
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(signature, expected_signature)

@app.route('/webhooks/ai-interview', methods=['POST'])
def handle_webhook():
    # 1. Verify signature
    signature = request.headers.get('X-Webhook-Signature')
    payload = request.get_data()
    
    if not verify_signature(payload, signature):
        return 'Invalid signature', 401
    
    # 2. Parse data
    data = request.get_json()
    
    # 3. Check idempotency
    event_key = f"{data['event']}:{data['session_id']}"
    if event_key in processed_events:
        return 'Already processed', 200
    
    # 4. Respond quickly
    processed_events.add(event_key)
    
    # 5. Process webhook (in background thread/queue in production)
    try:
        process_webhook(data)
    except Exception as e:
        print(f"Error processing webhook: {e}")
        processed_events.remove(event_key)
        return 'Processing failed', 500
    
    return 'OK', 200

def process_webhook(payload):
    event = payload['event']
    session_id = payload['session_id']
    
    if event == 'interview.completed':
        print(f"Interview completed: {session_id}")
        
        # Download transcript
        if payload.get('transcript_url'):
            import requests
            response = requests.get(payload['transcript_url'])
            transcript = response.json()
            
            # Store in database
            from models import db, Interview
            interview = Interview(
                session_id=session_id,
                participant_email=payload['invitee_email'],
                transcript=transcript,
                duration_minutes=payload.get('total_minutes'),
                completed_at=datetime.fromisoformat(payload['timestamp'])
            )
            db.session.add(interview)
            db.session.commit()
        
        # Send notification
        notify_team(f"Interview completed: {payload['invitee_email']}")
        
    elif event == 'session.feedback_submitted':
        print(f"Feedback submitted: {session_id}")
        
        from models import db, Feedback
        feedback = Feedback(
            session_id=session_id,
            **payload['feedback']
        )
        db.session.add(feedback)
        db.session.commit()

def notify_team(message):
    # Implement your notification logic
    print(f"NOTIFICATION: {message}")

if __name__ == '__main__':
    app.run(port=3000)

// pages/api/webhooks/ai-interview.ts
import type { NextApiRequest, NextApiResponse } from 'next';
import crypto from 'crypto';
import { prisma } from '@/lib/prisma';

const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET!;

function verifySignature(payload: string, signature: string): boolean {
  const expectedSignature = crypto
    .createHmac('sha256', WEBHOOK_SECRET)
    .update(payload)
    .digest('hex');
  
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  );
}

export const config = {
  api: {
    bodyParser: false, // We need raw body for signature verification
  },
};

async function getRawBody(req: NextApiRequest): Promise<string> {
  return new Promise((resolve, reject) => {
    let data = '';
    req.on('data', chunk => { data += chunk; });
    req.on('end', () => resolve(data));
    req.on('error', reject);
  });
}

export default async function handler(
  req: NextApiRequest,
  res: NextApiResponse
) {
  if (req.method !== 'POST') {
    return res.status(405).json({ error: 'Method not allowed' });
  }

  try {
    // Get raw body
    const rawBody = await getRawBody(req);
    
    // Verify signature
    const signature = req.headers['x-webhook-signature'] as string;
    if (!signature || !verifySignature(rawBody, signature)) {
      return res.status(401).json({ error: 'Invalid signature' });
    }

    // Parse payload
    const payload = JSON.parse(rawBody);

    // Check idempotency
    const existing = await prisma.webhookEvent.findUnique({
      where: {
        event_session: {
          event: payload.event,
          session_id: payload.session_id,
        },
      },
    });

    if (existing) {
      return res.status(200).json({ message: 'Already processed' });
    }

    // Store event
    await prisma.webhookEvent.create({
      data: {
        event: payload.event,
        session_id: payload.session_id,
        payload,
      },
    });

    // Process webhook
    await processWebhook(payload);

    return res.status(200).json({ message: 'OK' });
    
  } catch (error) {
    console.error('Webhook error:', error);
    return res.status(500).json({ error: 'Internal server error' });
  }
}

async function processWebhook(payload: any) {
  if (payload.event === 'interview.completed') {
    // Fetch transcript
    if (payload.transcript_url) {
      const response = await fetch(payload.transcript_url);
      const transcript = await response.json();

      // Store in database
      await prisma.interview.create({
        data: {
          sessionId: payload.session_id,
          participantEmail: payload.invitee_email,
          transcript,
          durationMinutes: payload.total_minutes,
          completedAt: new Date(payload.timestamp),
        },
      });
    }
  }
}

const { Queue } = require('bullmq');

const webhookQueue = new Queue('webhooks', {
  connection: {
    host: 'localhost',
    port: 6379,
  },
});

app.post('/webhooks', async (req, res) => {
  // Verify signature...
  
  // Add to queue
  await webhookQueue.add('process', req.body);
  
  // Respond immediately
  res.status(200).send('OK');
});

// Worker process
const { Worker } = require('bullmq');

new Worker('webhooks', async (job) => {
  await processWebhook(job.data);
}, {
  connection: { host: 'localhost', port: 6379 }
});

const Redis = require('ioredis');
const redis = new Redis();

async function isProcessed(eventKey) {
  return await redis.exists(eventKey);
}

async function markProcessed(eventKey) {
  await redis.setex(eventKey, 86400, '1'); // 24h TTL
}

app.post('/webhooks', async (req, res) => {
  const eventKey = `webhook:${req.body.event}:${req.body.session_id}`;
  
  if (await isProcessed(eventKey)) {
    return res.status(200).send('Already processed');
  }
  
  await markProcessed(eventKey);
  // Process...
});

async function processWebhook(payload, retryCount = 0) {
  try {
    // Process webhook...
  } catch (error) {
    if (retryCount < 3) {
      console.log(`Retry ${retryCount + 1}/3`);
      await new Promise(r => setTimeout(r, 1000 * Math.pow(2, retryCount)));
      return processWebhook(payload, retryCount + 1);
    }
    
    // Log failure for manual review
    await db.failedWebhooks.create({
      payload,
      error: error.message,
      retries: retryCount,
    });
    
    throw error;
  }
}

const prometheus = require('prom-client');

const webhookCounter = new prometheus.Counter({
  name: 'webhooks_received_total',
  help: 'Total webhooks received',
  labelNames: ['event', 'status'],
});

app.post('/webhooks', async (req, res) => {
  try {
    // Process...
    webhookCounter.inc({ event: req.body.event, status: 'success' });
    res.status(200).send('OK');
  } catch (error) {
    webhookCounter.inc({ event: req.body.event, status: 'failure' });
    res.status(500).send('Error');
  }
});

# Start your server
npm start

# In another terminal, start ngrok
ngrok http 3000

# Use the HTTPS URL in webhook settings
https://abc123.ngrok.io/webhooks/ai-interview