-- 1. Customer webhook endpoints configuration
CREATE TABLE webhook_endpoints (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id UUID NOT NULL,
  url VARCHAR(2048) NOT NULL,
  secret VARCHAR(255) NOT NULL, -- Shared secret for HMAC signing
  events TEXT[] NOT NULL,       -- e.g. ['order.created', 'payment.received']
  enabled BOOLEAN DEFAULT true,
  created_at TIMESTAMP DEFAULT NOW()
);
 
-- 2. Delivery logs and retry tracking
CREATE TABLE webhook_deliveries (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  endpoint_id UUID NOT NULL REFERENCES webhook_endpoints(id) ON DELETE CASCADE,
  event_id VARCHAR(255) NOT NULL,
  event_type VARCHAR(100) NOT NULL,
  payload JSONB NOT NULL,
  status VARCHAR(20) DEFAULT 'pending', -- 'pending', 'success', 'failed'
  attempts INT DEFAULT 0,
  last_attempt_at TIMESTAMP,
  response_status INT,
  response_body TEXT,
  next_retry_at TIMESTAMP,
  created_at TIMESTAMP DEFAULT NOW()
);

import crypto from "crypto";
import Queue from "bull";
import pg from "pg";
 
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
 
export class WebhookService {
  constructor() {
    this.queue = new Queue("outgoing-webhooks", process.env.REDIS_URL);
    this.setupDeliveryWorker();
  }
 
  // 1. Sign payload using HMAC-SHA256
  signPayload(payload, secret, timestamp) {
    const serialized = `${timestamp}.${JSON.stringify(payload)}`;
    return crypto
      .createHmac("sha256", secret)
      .update(serialized)
      .digest("hex");
  }
 
  // 2. Dispatch an event to all subscribed endpoints
  async dispatchEvent(customerId, eventType, data) {
    const { rows: endpoints } = await pool.query(
      `SELECT * FROM webhook_endpoints 
       WHERE customer_id = $1 AND enabled = true AND $2 = ANY(events)`,
      [customerId, eventType]
    );
 
    for (const endpoint of endpoints) {
      const eventId = `evt_${crypto.randomBytes(12).toString("hex")}`;
      const timestamp = Math.floor(Date.now() / 1000);
 
      const payload = {
        id: eventId,
        type: eventType,
        created: timestamp,
        data
      };
 
      // Record delivery attempt in DB
      await pool.query(
        `INSERT INTO webhook_deliveries (endpoint_id, event_id, event_type, payload, status)
         VALUES ($1, $2, $3, $4, 'pending')`,
        [endpoint.id, eventId, eventType, payload]
      );
 
      // Enqueue job with exponential backoff retries
      await this.queue.add("deliver-webhook", {
        endpointId: endpoint.id,
        eventId,
        url: endpoint.url,
        secret: endpoint.secret,
        payload,
        timestamp
      }, {
        attempts: 5,
        backoff: { type: "exponential", delay: 10000 } // 10s, 20s, 40s...
      });
    }
  }
 
  // 3. Worker delivering the HTTP POST requests
  setupDeliveryWorker() {
    this.queue.process("deliver-webhook", async (job) => {
      const { endpointId, eventId, url, secret, payload, timestamp } = job.data;
      const signature = this.signPayload(payload, secret, timestamp);
 
      try {
        const controller = new AbortController();
        const timeoutId = setTimeout(() => controller.abort(), 15000); // 15s timeout
 
        const response = await fetch(url, {
          method: "POST",
          headers: {
            "Content-Type": "application/json",
            "X-Webhook-Signature": `t=${timestamp},v1=${signature}`,
            "X-Webhook-ID": eventId,
            "User-Agent": "MyPlatform-Webhooks/1.0"
          },
          body: JSON.stringify(payload),
          signal: controller.signal
        });
 
        clearTimeout(timeoutId);
 
        const responseText = await response.text();
 
        // Update delivery log
        await pool.query(
          `UPDATE webhook_deliveries
           SET status = $1, attempts = attempts + 1, last_attempt_at = NOW(),
               response_status = $2, response_body = $3
           WHERE event_id = $4`,
          [response.ok ? "success" : "failed", response.status, responseText.slice(0, 1000), eventId]
        );
 
        if (!response.ok) {
          throw new Error(`Customer endpoint returned HTTP ${response.status}`);
        }
 
        return { success: true };
      } catch (err) {
        await pool.query(
          `UPDATE webhook_deliveries
           SET attempts = attempts + 1, last_attempt_at = NOW(), response_body = $1
           WHERE event_id = $2`,
          [err.message, eventId]
        );
        throw err; // Bull triggers automatic retry
      }
    });
  }
}

import express from "express";
import crypto from "crypto";
import Queue from "bull";
import Redis from "ioredis";
import Stripe from "stripe";
 
const app = express();
const redis = new Redis(process.env.REDIS_URL);
const webhookQueue = new Queue("incoming-webhooks", process.env.REDIS_URL);
const stripe = new Stripe(process.env.STRIPE_SECRET_KEY);
 
// 1. Stripe Webhook Route (Preserves raw byte body)
app.post(
  "/webhooks/stripe",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    const signature = req.headers["stripe-signature"];
 
    let event;
    try {
      event = stripe.webhooks.constructEvent(
        req.body,
        signature,
        process.env.STRIPE_WEBHOOK_SECRET
      );
    } catch (err) {
      console.error(`Invalid Stripe signature: ${err.message}`);
      return res.status(401).json({ error: "Invalid signature" });
    }
 
    // Idempotency check with Redis (TTL: 7 days)
    const isProcessed = await redis.get(`webhook:stripe:${event.id}`);
    if (isProcessed) {
      return res.status(200).json({ status: "already_processed" });
    }
 
    // Enqueue event for async execution
    await webhookQueue.add("process-stripe", {
      eventId: event.id,
      eventType: event.type,
      payload: event.data.object
    }, {
      jobId: event.id,
      attempts: 3
    });
 
    res.status(200).json({ received: true });
  }
);
 
// 2. GitHub Webhook Route
app.post(
  "/webhooks/github",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    const signature = req.headers["x-hub-signature-256"];
    const eventType = req.headers["x-github-event"];
    const deliveryId = req.headers["x-github-delivery"];
    const rawBody = req.body.toString("utf-8");
 
    // Verify HMAC-SHA256 signature
    const expected = "sha256=" + crypto
      .createHmac("sha256", process.env.GITHUB_WEBHOOK_SECRET)
      .update(rawBody)
      .digest("hex");
 
    if (!crypto.timingSafeEqual(Buffer.from(signature || ""), Buffer.from(expected))) {
      return res.status(401).json({ error: "Invalid GitHub signature" });
    }
 
    // Enqueue job
    await webhookQueue.add("process-github", {
      eventId: deliveryId,
      eventType,
      payload: JSON.parse(rawBody)
    }, {
      jobId: deliveryId,
      attempts: 3
    });
 
    res.status(200).json({ received: true });
  }
);
 
// 3. Background Queue Consumer Workers
webhookQueue.process("process-stripe", async (job) => {
  const { eventId, eventType, payload } = job.data;
 
  switch (eventType) {
    case "payment_intent.succeeded":
      console.log(`Fulfilling order for payment: ${payload.id}`);
      // Run database updates, invoice generation, etc.
      break;
 
    case "customer.subscription.deleted":
      console.log(`Canceling subscription: ${payload.id}`);
      break;
  }
 
  // Mark event as processed in Redis (expires after 7 days)
  await redis.set(`webhook:stripe:${eventId}`, "completed", "EX", 60 * 60 * 24 * 7);
});
 
webhookQueue.process("process-github", async (job) => {
  const { eventType, payload } = job.data;
 
  if (eventType === "push" && payload.ref === "refs/heads/main") {
    console.log(`Triggering CI/CD build for commit: ${payload.commits[0]?.id}`);
  }
});
 
app.listen(3000, () => console.log("⚡ Webhook server active on port 3000"));

// Get delivery logs for an endpoint
app.get("/api/webhooks/:endpointId/deliveries", async (req, res) => {
  const { rows } = await pool.query(
    `SELECT id, event_type, status, attempts, response_status, response_body, created_at
     FROM webhook_deliveries
     WHERE endpoint_id = $1
     ORDER BY created_at DESC
     LIMIT 50;`,
    [req.params.endpointId]
  );
  res.json({ deliveries: rows });
});
 
// Manually retry a failed delivery
app.post("/api/webhooks/deliveries/:deliveryId/retry", async (req, res) => {
  const { rows } = await pool.query(
    `SELECT d.*, e.url, e.secret 
     FROM webhook_deliveries d
     JOIN webhook_endpoints e ON d.endpoint_id = e.id
     WHERE d.id = $1`,
    [req.params.deliveryId]
  );
 
  const delivery = rows[0];
  if (!delivery) return res.status(404).json({ error: "Delivery record not found" });
 
  await webhookService.queue.add("deliver-webhook", {
    endpointId: delivery.endpoint_id,
    eventId: delivery.event_id,
    url: delivery.url,
    secret: delivery.secret,
    payload: delivery.payload,
    timestamp: Math.floor(Date.now() / 1000)
  });
 
  res.json({ message: "Retry attempt enqueued successfully." });
});