Sending Webhooks

// Database models
model WebhookEndpoint {
  id          String   @id @default(cuid())
  userId      String
  url         String
  secret      String   // For signing
  events      String[] // Events to receive
  isActive    Boolean  @default(true)
  createdAt   DateTime @default(now())
  updatedAt   DateTime @updatedAt

  user        User     @relation(fields: [userId], references: [id])
  deliveries  WebhookDelivery[]
}

model WebhookDelivery {
  id          String   @id @default(cuid())
  endpointId  String
  eventType   String
  payload     Json
  statusCode  Int?
  response    String?
  attempts    Int      @default(0)
  deliveredAt DateTime?
  createdAt   DateTime @default(now())

  endpoint    WebhookEndpoint @relation(fields: [endpointId], references: [id])
}

// src/services/webhookService.js
import crypto from 'crypto';
import { prisma } from '../lib/prisma.js';
import { webhookQueue } from '../queues/webhook.js';

export const WebhookService = {
  // Create a new webhook endpoint
  async createEndpoint(userId, data) {
    const secret = crypto.randomBytes(32).toString('hex');

    return prisma.webhookEndpoint.create({
      data: {
        userId,
        url: data.url,
        secret,
        events: data.events,
      },
    });
  },

  // List user's endpoints
  async listEndpoints(userId) {
    return prisma.webhookEndpoint.findMany({
      where: { userId },
      orderBy: { createdAt: 'desc' },
    });
  },

  // Update endpoint
  async updateEndpoint(id, userId, data) {
    return prisma.webhookEndpoint.update({
      where: { id, userId },
      data: {
        url: data.url,
        events: data.events,
        isActive: data.isActive,
      },
    });
  },

  // Regenerate secret
  async regenerateSecret(id, userId) {
    const secret = crypto.randomBytes(32).toString('hex');

    return prisma.webhookEndpoint.update({
      where: { id, userId },
      data: { secret },
      select: { id: true, secret: true },
    });
  },

  // Delete endpoint
  async deleteEndpoint(id, userId) {
    return prisma.webhookEndpoint.delete({
      where: { id, userId },
    });
  },

  // Dispatch event to all relevant endpoints
  async dispatch(eventType, payload, options = {}) {
    const { userId, resourceId } = options;

    // Find endpoints subscribed to this event
    const endpoints = await prisma.webhookEndpoint.findMany({
      where: {
        isActive: true,
        events: { has: eventType },
        ...(userId && { userId }),
      },
    });

    // Queue delivery for each endpoint
    for (const endpoint of endpoints) {
      await webhookQueue.add('deliver', {
        endpointId: endpoint.id,
        eventType,
        payload: {
          id: crypto.randomUUID(),
          type: eventType,
          createdAt: new Date().toISOString(),
          data: payload,
        },
      }, {
        attempts: 5,
        backoff: {
          type: 'exponential',
          delay: 60000, // 1 minute, then 2, 4, 8, 16
        },
      });
    }

    return { queued: endpoints.length };
  },

  // Create signature for payload
  createSignature(payload, secret, timestamp) {
    const signedPayload = `${timestamp}.${JSON.stringify(payload)}`;
    return crypto
      .createHmac('sha256', secret)
      .update(signedPayload)
      .digest('hex');
  },
};

// src/queues/webhookWorker.js
import { Worker } from 'bullmq';
import { prisma } from '../lib/prisma.js';
import { WebhookService } from '../services/webhookService.js';

const worker = new Worker('webhooks', async (job) => {
  const { endpointId, eventType, payload } = job.data;

  // Get endpoint details
  const endpoint = await prisma.webhookEndpoint.findUnique({
    where: { id: endpointId },
  });

  if (!endpoint || !endpoint.isActive) {
    return { skipped: true, reason: 'Endpoint inactive or deleted' };
  }

  // Create delivery record
  const delivery = await prisma.webhookDelivery.create({
    data: {
      endpointId,
      eventType,
      payload,
    },
  });

  // Generate signature
  const timestamp = Math.floor(Date.now() / 1000);
  const signature = WebhookService.createSignature(
    payload,
    endpoint.secret,
    timestamp
  );

  try {
    // Send webhook
    const controller = new AbortController();
    const timeout = setTimeout(() => controller.abort(), 30000); // 30s timeout

    const response = await fetch(endpoint.url, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-Webhook-Signature': signature,
        'X-Webhook-Timestamp': timestamp.toString(),
        'X-Webhook-ID': payload.id,
        'User-Agent': 'MyApp-Webhooks/1.0',
      },
      body: JSON.stringify(payload),
      signal: controller.signal,
    });

    clearTimeout(timeout);

    const responseText = await response.text().catch(() => '');

    // Update delivery record
    await prisma.webhookDelivery.update({
      where: { id: delivery.id },
      data: {
        statusCode: response.status,
        response: responseText.slice(0, 1000), // Truncate
        attempts: job.attemptsMade + 1,
        deliveredAt: response.ok ? new Date() : null,
      },
    });

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${responseText.slice(0, 200)}`);
    }

    return { success: true, statusCode: response.status };
  } catch (error) {
    // Update delivery with error
    await prisma.webhookDelivery.update({
      where: { id: delivery.id },
      data: {
        attempts: job.attemptsMade + 1,
        response: error.message.slice(0, 1000),
      },
    });

    throw error; // Re-throw to trigger retry
  }
}, {
  connection: { host: 'localhost', port: 6379 },
  concurrency: 10,
});

worker.on('failed', async (job, err) => {
  console.error(`Webhook delivery failed: ${job.id}`, err.message);

  // Disable endpoint after too many failures
  if (job.attemptsMade >= 5) {
    const { endpointId } = job.data;
    await prisma.webhookEndpoint.update({
      where: { id: endpointId },
      data: { isActive: false },
    });
    console.log(`Disabled webhook endpoint ${endpointId} after repeated failures`);
  }
});

// src/routes/webhooks.js
import express from 'express';
import { WebhookService } from '../services/webhookService.js';
import { authenticate } from '../middleware/auth.js';
import { z } from 'zod';

const router = express.Router();

const createEndpointSchema = z.object({
  url: z.string().url(),
  events: z.array(z.string()).min(1),
});

// List webhook endpoints
router.get('/endpoints', authenticate, async (req, res) => {
  const endpoints = await WebhookService.listEndpoints(req.user.id);

  // Don't expose full secret
  const sanitized = endpoints.map(e => ({
    ...e,
    secret: e.secret.slice(0, 8) + '...',
  }));

  res.json({ data: sanitized });
});

// Create endpoint
router.post('/endpoints', authenticate, async (req, res) => {
  const data = createEndpointSchema.parse(req.body);
  const endpoint = await WebhookService.createEndpoint(req.user.id, data);

  res.status(201).json({
    data: endpoint,
    message: 'Save this secret - it won\'t be shown again',
  });
});

// Update endpoint
router.patch('/endpoints/:id', authenticate, async (req, res) => {
  const endpoint = await WebhookService.updateEndpoint(
    req.params.id,
    req.user.id,
    req.body
  );
  res.json({ data: endpoint });
});

// Regenerate secret
router.post('/endpoints/:id/rotate-secret', authenticate, async (req, res) => {
  const result = await WebhookService.regenerateSecret(req.params.id, req.user.id);
  res.json({
    data: result,
    message: 'Save this new secret - it won\'t be shown again',
  });
});

// Delete endpoint
router.delete('/endpoints/:id', authenticate, async (req, res) => {
  await WebhookService.deleteEndpoint(req.params.id, req.user.id);
  res.status(204).send();
});

// List deliveries (for debugging)
router.get('/endpoints/:id/deliveries', authenticate, async (req, res) => {
  const deliveries = await prisma.webhookDelivery.findMany({
    where: {
      endpointId: req.params.id,
      endpoint: { userId: req.user.id },
    },
    orderBy: { createdAt: 'desc' },
    take: 50,
  });

  res.json({ data: deliveries });
});

// Retry a failed delivery
router.post('/deliveries/:id/retry', authenticate, async (req, res) => {
  const delivery = await prisma.webhookDelivery.findFirst({
    where: {
      id: req.params.id,
      endpoint: { userId: req.user.id },
    },
    include: { endpoint: true },
  });

  if (!delivery) {
    return res.status(404).json({ error: 'Delivery not found' });
  }

  await webhookQueue.add('deliver', {
    endpointId: delivery.endpointId,
    eventType: delivery.eventType,
    payload: delivery.payload,
  });

  res.json({ message: 'Retry queued' });
});

export default router;

// In your application code
import { WebhookService } from '../services/webhookService.js';

// When an order is created
async function createOrder(userId, orderData) {
  const order = await OrderService.create(userId, orderData);

  // Dispatch webhook
  await WebhookService.dispatch('order.created', {
    order: {
      id: order.id,
      total: order.total,
      items: order.items,
      createdAt: order.createdAt,
    },
  }, { userId });

  return order;
}

// When a user is updated
async function updateUser(userId, updates) {
  const user = await UserService.update(userId, updates);

  await WebhookService.dispatch('user.updated', {
    user: {
      id: user.id,
      email: user.email,
      name: user.name,
    },
  }, { userId });

  return user;
}

// Customer's code to verify your webhooks
const crypto = require('crypto');

function verifyWebhook(payload, signature, timestamp, secret) {
  // Check timestamp (prevent replay attacks)
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp)) > 300) { // 5 min tolerance
    throw new Error('Webhook timestamp too old');
  }

  // Verify signature
  const signedPayload = `${timestamp}.${JSON.stringify(payload)}`;
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(signedPayload)
    .digest('hex');

  if (!crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  )) {
    throw new Error('Invalid webhook signature');
  }

  return payload;
}

// Express handler
app.post('/webhooks/myapp', express.json(), (req, res) => {
  try {
    const payload = verifyWebhook(
      req.body,
      req.headers['x-webhook-signature'],
      req.headers['x-webhook-timestamp'],
      process.env.MYAPP_WEBHOOK_SECRET
    );

    console.log('Received event:', payload.type);
    // Handle event...

    res.status(200).send('OK');
  } catch (err) {
    console.error('Webhook error:', err.message);
    res.status(400).send('Invalid webhook');
  }
});

// src/constants/webhookEvents.js
export const WebhookEvents = {
  // User events
  USER_CREATED: 'user.created',
  USER_UPDATED: 'user.updated',
  USER_DELETED: 'user.deleted',

  // Order events
  ORDER_CREATED: 'order.created',
  ORDER_PAID: 'order.paid',
  ORDER_SHIPPED: 'order.shipped',
  ORDER_DELIVERED: 'order.delivered',
  ORDER_CANCELLED: 'order.cancelled',

  // Subscription events
  SUBSCRIPTION_CREATED: 'subscription.created',
  SUBSCRIPTION_RENEWED: 'subscription.renewed',
  SUBSCRIPTION_CANCELLED: 'subscription.cancelled',
  SUBSCRIPTION_EXPIRED: 'subscription.expired',

  // Payment events
  PAYMENT_SUCCEEDED: 'payment.succeeded',
  PAYMENT_FAILED: 'payment.failed',
  REFUND_CREATED: 'refund.created',
};

export const AllWebhookEvents = Object.values(WebhookEvents);