Database Storage for Webhooks — Complete Guide

-- Subscribers table
CREATE TABLE webhook_subscribers (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    url TEXT NOT NULL,
    secret TEXT NOT NULL,
    events TEXT[] NOT NULL DEFAULT '{}',
    is_active BOOLEAN DEFAULT true,
    description TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_subscribers_events
    ON webhook_subscribers USING GIN (events);
CREATE INDEX idx_subscribers_active
    ON webhook_subscribers (is_active) WHERE is_active = true;

-- Webhook events table
CREATE TABLE webhook_events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    event_type TEXT NOT NULL,
    source TEXT NOT NULL,
    payload JSONB NOT NULL,
    headers JSONB,
    idempotency_key TEXT UNIQUE,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_events_type
    ON webhook_events (event_type);
CREATE INDEX idx_events_created
    ON webhook_events (created_at DESC);
CREATE INDEX idx_events_idempotency
    ON webhook_events (idempotency_key);

-- Delivery logs table
CREATE TABLE webhook_delivery_logs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    event_id UUID REFERENCES webhook_events(id),
    subscriber_id UUID REFERENCES webhook_subscribers(id),
    status TEXT NOT NULL CHECK (status IN (
        'pending', 'delivered', 'failed', 'retrying'
    )),
    attempt INTEGER DEFAULT 1,
    max_attempts INTEGER DEFAULT 5,
    response_status INTEGER,
    response_body TEXT,
    error_message TEXT,
    duration_ms INTEGER,
    next_retry_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    completed_at TIMESTAMPTZ
);

CREATE INDEX idx_delivery_status
    ON webhook_delivery_logs (status, next_retry_at)
    WHERE status IN ('pending', 'retrying');
CREATE INDEX idx_delivery_event
    ON webhook_delivery_logs (event_id);
CREATE INDEX idx_delivery_subscriber
    ON webhook_delivery_logs (subscriber_id, created_at DESC);

-- Find webhooks needing retry
SELECT dl.*, s.url, s.secret, e.payload
FROM webhook_delivery_logs dl
JOIN webhook_subscribers s ON dl.subscriber_id = s.id
JOIN webhook_events e ON dl.event_id = e.id
WHERE dl.status = 'retrying'
  AND dl.next_retry_at <= NOW()
  AND dl.attempt < dl.max_attempts
ORDER BY dl.next_retry_at
LIMIT 100
FOR UPDATE SKIP LOCKED;

-- Delivery success rate by subscriber
SELECT
    s.id,
    s.url,
    COUNT(dl.id) AS total_deliveries,
    COUNT(dl.id) FILTER (WHERE dl.status = 'delivered') AS successful,
    ROUND(
        COUNT(dl.id) FILTER (WHERE dl.status = 'delivered')::numeric /
        NULLIF(COUNT(dl.id), 0) * 100, 2
    ) AS success_rate
FROM webhook_subscribers s
LEFT JOIN webhook_delivery_logs dl ON s.id = dl.subscriber_id
GROUP BY s.id, s.url
ORDER BY success_rate ASC NULLS LAST;

-- Recent events with delivery status
SELECT
    e.id AS event_id,
    e.event_type,
    e.created_at,
    COUNT(dl.id) AS delivery_attempts,
    BOOL_AND(dl.status = 'delivered') AS all_delivered
FROM webhook_events e
LEFT JOIN webhook_delivery_logs dl ON e.id = dl.event_id
WHERE e.created_at >= NOW() - INTERVAL '1 hour'
GROUP BY e.id
ORDER BY e.created_at DESC;

// MongoDB collections for webhook storage
const subscriberSchema = {
    url: String,
    secret: String,
    events: [String],
    isActive: Boolean,
    description: String,
    createdAt: Date,
    updatedAt: Date,
};

subscriberSchema.index({ events: 1 });
subscriberSchema.index({ isActive: 1 });

const eventSchema = {
    eventType: String,
    source: String,
    payload: Object,
    headers: Object,
    idempotencyKey: { type: String, unique: true, sparse: true },
    createdAt: Date,
};

eventSchema.index({ eventType: 1, createdAt: -1 });
eventSchema.index({ createdAt: -1 });

const deliveryLogSchema = {
    eventId: { type: ObjectId, ref: 'Event' },
    subscriberId: { type: ObjectId, ref: 'Subscriber' },
    status: { type: String, enum: ['pending', 'delivered', 'failed', 'retrying'] },
    attempt: Number,
    maxAttempts: Number,
    responseStatus: Number,
    responseBody: String,
    errorMessage: String,
    durationMs: Number,
    nextRetryAt: Date,
    createdAt: Date,
    completedAt: Date,
};

deliveryLogSchema.index({ status: 1, nextRetryAt: 1 });
deliveryLogSchema.index({ eventId: 1 });
deliveryLogSchema.index({ subscriberId: 1, createdAt: -1 });

const redis = require('redis');

class WebhookRedisStore {
    constructor(client) {
        this.client = client;
    }

    // Idempotency - 24 hour TTL
    async checkIdempotency(key) {
        const result = await this.client.set(
            `wh:idemp:${key}`,
            '1',
            { EX: 86400, NX: true }
        );
        return result !== null;
    }

    // Pending retry queue - sorted by retry time
    async addToRetryQueue(eventId, subscriberId, retryAt) {
        const score = new Date(retryAt).getTime();
        await this.client.zadd('wh:retry:queue', score, `${eventId}:${subscriberId}`);
    }

    async getPendingRetries(limit = 100) {
        const now = Date.now();
        const entries = await this.client.zrangebyscore(
            'wh:retry:queue',
            0,
            now,
            { LIMIT: [0, limit] }
        );
        if (entries.length > 0) {
            await this.client.zremrangebyscore('wh:retry:queue', 0, now);
        }
        return entries.map(e => {
            const [eventId, subscriberId] = e.split(':');
            return { eventId, subscriberId };
        });
    }

    // Recent events - capped list
    async addRecentEvent(event) {
        await this.client.lpush('wh:recent:events', JSON.stringify(event));
        await this.client.ltrim('wh:recent:events', 0, 999); // Keep 1000
    }

    async getRecentEvents(count = 50) {
        const events = await this.client.lrange('wh:recent:events', 0, count - 1);
        return events.map(e => JSON.parse(e));
    }

    // Rate limiting
    async checkRateLimit(subscriberId, maxPerMinute = 60) {
        const key = `wh:ratelimit:${subscriberId}`;
        const current = await this.client.incr(key);
        if (current === 1) {
            await this.client.expire(key, 60);
        }
        return current <= maxPerMinute;
    }
}

// Retention policy manager
class WebhookRetentionManager {
    constructor(db) {
        this.db = db;
    }

    async applyRetentionPolicy() {
        const now = new Date();
        const policies = [
            { table: 'webhook_events', retentionDays: 90 },
            { table: 'webhook_delivery_logs', retentionDays: 90 },
            { table: 'webhook_idempotency', retentionDays: 30 },
        ];

        for (const policy of policies) {
            const cutoff = new Date(
                now.getTime() - policy.retentionDays * 86400000
            );

            const deleted = await this.db.query(
                `DELETE FROM ${policy.table}
                 WHERE created_at < $1`,
                [cutoff]
            );

            console.log(
                `Cleaned ${deleted.rowCount} rows from ${policy.table}`
            );
        }
    }

    // Archive before delete
    async archiveEvents(before) {
        const events = await this.db.query(
            `SELECT * FROM webhook_events WHERE created_at < $1`,
            [before]
        );

        // Write to archive storage (S3, cold storage)
        await this.writeToArchive('webhooks-archive', events.rows);

        // Then delete from primary
        await this.db.query(
            `DELETE FROM webhook_events WHERE created_at < $1`,
            [before]
        );
    }
}