Webhook Flow

// Provider: webhook registration with challenge
app.post('/api/webhooks', async (req, res) => {
  const { url, events } = req.body;

  // Generate verification challenge
  const challenge = crypto.randomBytes(32).toString('hex');
  const webhookId = generateId();

  // Send verification request to consumer
  try {
    const verificationResult = await axios.post(url, {
      type: 'verification',
      webhookId,
      challenge,
    }, {
      headers: { 'Content-Type': 'application/json' },
      timeout: 10000,
    });

    // Verify consumer responded with correct challenge
    if (verificationResult.data.challenge === challenge) {
      // Webhook verified successfully
      saveWebhook(webhookId, url, events);
      res.status(201).json({
        id: webhookId,
        status: 'verified',
        secret: generateSecret(),
      });
    } else {
      res.status(400).json({
        error: 'Verification failed: challenge mismatch',
      });
    }
  } catch (error) {
    res.status(400).json({
      error: 'Verification failed: cannot reach consumer URL',
    });
  }
});

// Consumer: handle verification
app.post('/webhooks/verify', (req, res) => {
  const { type, webhookId, challenge } = req.body;

  if (type === 'verification') {
    // Respond with the challenge to verify
    res.json({ challenge });
  }
});

async function deliverWebhook(webhook, event) {
  const payload = {
    id: generateEventId(),
    type: event.type,
    created_at: new Date().toISOString(),
    data: event.data,
  };

  const payloadStr = JSON.stringify(payload);
  const signature = crypto
    .createHmac('sha256', webhook.secret)
    .update(payloadStr)
    .digest('hex');

  const delivery = {
    webhookId: webhook.id,
    eventId: payload.id,
    url: webhook.url,
    payload: payloadStr,
    signature,
    attempt: 1,
    maxAttempts: 5,
    status: 'pending',
  };

  for (let attempt = 1; attempt <= delivery.maxAttempts; attempt++) {
    delivery.attempt = attempt;
    
    try {
      const response = await axios.post(webhook.url, payload, {
        headers: {
          'Content-Type': 'application/json',
          'X-Webhook-ID': webhook.id,
          'X-Event-ID': payload.id,
          'X-Signature-256': signature,
          'X-Delivery-Attempt': attempt,
        },
        timeout: 15000,
      });

      if (response.status >= 200 && response.status < 300) {
        delivery.status = 'delivered';
        logDelivery(delivery);
        return { success: true, delivery };
      }
    } catch (error) {
      delivery.lastError = error.message;
      delivery.status = 'failed';
      logDelivery(delivery);

      if (attempt < delivery.maxAttempts) {
        // Exponential backoff
        await sleep(Math.min(1000 * Math.pow(2, attempt - 1), 60000));
      }
    }
  }

  delivery.status = 'permanently_failed';
  logDelivery(delivery);
  return { success: false, delivery };
}

function logDelivery(delivery) {
  console.log(
    `Delivery ${delivery.eventId} to ${delivery.url}: ` +
    `${delivery.status} (attempt ${delivery.attempt})`
  );
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

from flask import Flask, request, jsonify
import hashlib
import hmac
import threading
from queue import Queue

app = Flask(__name__)
webhook_queue = Queue()

@app.route('/webhooks/handler', methods=['POST'])
def handle_webhook():
    # Verify signature
    signature = request.headers.get('X-Signature-256')
    payload = request.get_data()
    
    if not verify_signature(payload, signature):
        return jsonify({"error": "Invalid signature"}), 401

    # Acknowledge immediately
    event_id = request.headers.get('X-Event-ID')
    
    # Queue for async processing
    webhook_queue.put({
        'event_id': event_id,
        'type': request.json.get('type'),
        'data': request.json.get('data'),
        'headers': dict(request.headers),
    })

    return jsonify({"status": "accepted", "event_id": event_id}), 200

def verify_signature(payload, signature):
    if not signature:
        return False
    
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected, signature)

def webhook_worker():
    while True:
        webhook = webhook_queue.get()
        try:
            process_event(webhook['type'], webhook['data'])
        except Exception as e:
            log_error(f"Failed to process {webhook['event_id']}: {e}")
        finally:
            webhook_queue.task_done()

# Start worker thread
threading.Thread(target=webhook_worker, daemon=True).start()

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