Merge pull request #1512 from jetstreamapp/chore/add-email-storage

chore: add email storage

Author: paustint

apps/api/src/app/controllers/mailgun-webhook.controller.ts

@@ -0,0 +1,219 @@
+import { ENV, logger, prisma } from '@jetstream/api-config';
+import type { Request, Response } from '@jetstream/api-types';
+import { getErrorMessageAndStackObj } from '@jetstream/shared/utils';
+import crypto from 'crypto';
+import { LRUCache } from 'lru-cache';
+import { z } from 'zod';
+
+// Cache for tracking used webhook tokens to prevent replay attacks
+// Tokens expire after 15 minutes, same as our timestamp validation window
+const WEBHOOK_TOKEN_CACHE = new LRUCache<string, boolean>({
+  max: 10000, // Store up to 10k recent tokens
+  ttl: 1000 * 60 * 15, // 15 minutes
+});
+
+// Maximum age for webhook timestamps (in seconds)
+const MAX_TIMESTAMP_AGE_SECONDS = 60 * 15; // 15 minutes
+
+// Mailgun webhook payload schema based on their documentation
+const MailgunWebhookSignatureSchema = z.object({
+  timestamp: z.string(),
+  token: z.string(),
+  signature: z.string(),
+});
+
+const MailgunDeliveryStatusSchema = z
+  .object({
+    code: z.number().optional(),
+    message: z.string().optional(),
+    description: z.string().optional(),
+    'enhanced-code': z.string().optional(),
+    'attempt-no': z.number().optional(),
+    'mx-host': z.string().optional(),
+    'session-seconds': z.number().optional(),
+    tls: z.boolean().optional(),
+    'certificate-verified': z.boolean().optional(),
+  })
+  .optional();
+
+const MailgunEnvelopeSchema = z
+  .object({
+    sender: z.string().optional(),
+    'sending-ip': z.string().optional(),
+    transport: z.string().optional(),
+    targets: z.string().optional(),
+  })
+  .optional();
+
+const MailgunMessageHeadersSchema = z
+  .object({
+    to: z.string().optional(),
+    from: z.string().optional(),
+    subject: z.string().optional(),
+    'message-id': z.string().optional(),
+  })
+  .optional();
+
+const MailgunMessageSchema = z
+  .object({
+    headers: MailgunMessageHeadersSchema,
+    size: z.number().optional(),
+    attachments: z.array(z.any()).optional(),
+  })
+  .optional();
+
+const MailgunFlagsSchema = z
+  .object({
+    'is-test-mode': z.boolean().optional(),
+    'is-routed': z.boolean().optional(),
+    'is-authenticated': z.boolean().optional(),
+    'is-system-test': z.boolean().optional(),
+  })
+  .optional();
+
+const MailgunEventDataSchema = z.object({
+  id: z.string().optional(),
+  event: z.string(),
+  timestamp: z.number(),
+  'log-level': z.string().optional(),
+  recipient: z.string(),
+  'recipient-domain': z.string().optional(),
+  'recipient-provider': z.string().optional(),
+  'delivery-status': MailgunDeliveryStatusSchema,
+  envelope: MailgunEnvelopeSchema,
+  message: MailgunMessageSchema,
+  flags: MailgunFlagsSchema,
+  tags: z.array(z.string()).optional(),
+  'user-variables': z.record(z.string(), z.any()).optional(),
+});
+
+const MailgunWebhookPayloadSchema = z.object({
+  signature: MailgunWebhookSignatureSchema,
+  'event-data': MailgunEventDataSchema,
+});
+
+export const routeDefinition = {
+  webhook: {
+    controllerFn: () => mailgunWebhookHandler,
+  },
+};
+
+const mailgunWebhookHandler = async (req: Request, res: Response) => {
+  try {
+    // Parse and validate the webhook payload
+    const rawBody = req.body as Buffer;
+    const parseResult = MailgunWebhookPayloadSchema.safeParse(JSON.parse(rawBody.toString()));
+
+    if (!parseResult.success) {
+      logger.warn({ error: parseResult.error }, 'Invalid Mailgun webhook payload');
+      return res.status(400).send('Invalid payload');
+    }
+
+    const { signature, 'event-data': eventData } = parseResult.data;
+
+    // Verify webhook signature if signing key is configured
+    if (ENV.MAILGUN_WEBHOOK_SIGNING_KEY) {
+      // Check timestamp freshness to prevent replay attacks
+      const timestampAge = Math.abs(Date.now() / 1000 - parseInt(signature.timestamp));
+      if (timestampAge > MAX_TIMESTAMP_AGE_SECONDS) {
+        logger.warn(
+          { timestamp: signature.timestamp, age: timestampAge },
+          'Mailgun webhook timestamp too old or too far in the future',
+        );
+        return res.status(403).send('Invalid timestamp');
+      }
+
+      // Check if this token has already been used (replay attack prevention)
+      if (WEBHOOK_TOKEN_CACHE.has(signature.token)) {
+        logger.warn({ token: signature.token }, 'Mailgun webhook token already used (replay attack)');
+        return res.status(403).send('Token already used');
+      }
+
+      // Verify the signature
+      const isValid = verifyWebhookSignature({
+        timestamp: signature.timestamp,
+        token: signature.token,
+        signature: signature.signature,
+        signingKey: ENV.MAILGUN_WEBHOOK_SIGNING_KEY,
+      });
+
+      if (!isValid) {
+        logger.warn({ timestamp: signature.timestamp }, 'Invalid Mailgun webhook signature');
+        return res.status(403).send('Invalid signature');
+      }
+
+      // Cache the token to prevent replay attacks
+      WEBHOOK_TOKEN_CACHE.set(signature.token, true);
+    } else {
+      logger.warn('Mailgun webhook signing key not configured - skipping signature verification');
+      return res.status(500).send('Webhook signing key not configured');
+    }
+
+    // Extract recipient domain from recipient email
+    const recipientDomain = eventData['recipient-domain'] || eventData.recipient.split('@')[1] || 'unknown';
+
+    // Store the webhook event in the database
+    await prisma.mailgunWebhookEvent.create({
+      data: {
+        // Event metadata
+        eventId: eventData.id,
+        event: eventData.event,
+        timestamp: new Date(eventData.timestamp * 1000),
+        logLevel: eventData['log-level'],
+
+        // Recipient information
+        recipient: eventData.recipient,
+        recipientDomain,
+        recipientProvider: eventData['recipient-provider'],
+
+        // Message information
+        subject: eventData.message?.headers?.subject,
+        messageId: eventData.message?.headers?.['message-id'],
+        fromAddress: eventData.message?.headers?.from,
+        toAddress: eventData.message?.headers?.to,
+        messageSize: eventData.message?.size,
+
+        // Delivery status
+        deliveryCode: eventData['delivery-status']?.code,
+        deliveryMessage: eventData['delivery-status']?.message,
+        deliveryDescription: eventData['delivery-status']?.description,
+        deliveryEnhancedCode: eventData['delivery-status']?.['enhanced-code'],
+        deliveryAttemptNo: eventData['delivery-status']?.['attempt-no'],
+        deliveryMxHost: eventData['delivery-status']?.['mx-host'],
+        deliverySessionSeconds: eventData['delivery-status']?.['session-seconds'],
+        deliveryTls: eventData['delivery-status']?.tls,
+        deliveryCertVerified: eventData['delivery-status']?.['certificate-verified'],
+
+        // Envelope information
+        envelopeSender: eventData.envelope?.sender,
+        envelopeSendingIp: eventData.envelope?.['sending-ip'],
+        envelopeTransport: eventData.envelope?.transport,
+
+        // Metadata
+        tags: eventData.tags || [],
+        userVariables: eventData['user-variables'],
+        flags: eventData.flags,
+      },
+    });
+
+    res.status(200).end();
+  } catch (err) {
+    logger.error(getErrorMessageAndStackObj(err), 'Error processing Mailgun webhook');
+    return res.status(500).send(`Error processing Mailgun webhook`);
+  }
+};
+
+function verifyWebhookSignature({
+  timestamp,
+  token,
+  signature,
+  signingKey,
+}: {
+  timestamp: string;
+  token: string;
+  signature: string;
+  signingKey: string;
+}): boolean {
+  const encodedToken = crypto.createHmac('sha256', signingKey).update(timestamp.concat(token)).digest('hex');
+  return encodedToken === signature;
+}

apps/api/src/app/routes/webhook.routes.ts

@@ -1,11 +1,19 @@
+import { createRateLimit } from '@jetstream/api-config';
 import { raw } from 'body-parser';
 import express, { Router } from 'express';
 import { routeDefinition as billingController } from '../controllers/billing.controller';
+import { routeDefinition as mailgunWebhookController } from '../controllers/mailgun-webhook.controller';
 
 const routes: express.Router = Router();
 
+const webhookRateLimit = createRateLimit('webhook', {
+  windowMs: 1000 * 60 * 1, // 1 minute
+  limit: 1000,
+});
+
 routes.use(raw({ type: 'application/json' }));
 
-routes.post('/stripe', billingController.webhook.controllerFn());
+routes.post('/stripe', webhookRateLimit, billingController.webhook.controllerFn());
+routes.post('/mailgun', webhookRateLimit, mailgunWebhookController.webhook.controllerFn());
 
 export default routes;

libs/api-config/src/lib/env-config.ts

@@ -188,6 +188,7 @@ const envSchema = z.object({
   JETSTREAM_EMAIL_FROM_NAME: z.string().optional().default(''),
   JETSTREAM_EMAIL_REPLY_TO: z.string().optional().default(''),
   MAILGUN_API_KEY: z.string().optional(),
+  MAILGUN_WEBHOOK_SIGNING_KEY: z.string().optional(),
   /**
    * Salesforce Org Connections
    * Connected App OAuth2 for connecting orgs

libs/auth/server/src/lib/auth.constants.ts

@@ -5,3 +5,4 @@ export const EMAIL_VERIFICATION_TOKEN_DURATION_HOURS = 1;
 export const DELETE_AUTH_ACTIVITY_DAYS = 365;
 export const DELETE_EMAIL_ACTIVITY_DAYS = 180;
 export const DELETE_TOKEN_DAYS = 3;
+export const DELETE_MAILGUN_WEBHOOK_DAYS = 30;

libs/auth/server/src/lib/auth.db.service.ts

@@ -39,6 +39,7 @@ import { actionDisplayName, methodDisplayName } from './auth-logging.db.service'
 import {
   DELETE_AUTH_ACTIVITY_DAYS,
   DELETE_EMAIL_ACTIVITY_DAYS,
+  DELETE_MAILGUN_WEBHOOK_DAYS,
   DELETE_TOKEN_DAYS,
   PASSWORD_RESET_DURATION_MINUTES,
 } from './auth.constants';
@@ -132,6 +133,11 @@ export async function pruneExpiredRecords() {
       expiresAt: { lte: addDays(startOfDay(new Date()), -DELETE_TOKEN_DAYS) },
     },
   });
+  await prisma.mailgunWebhookEvent.deleteMany({
+    where: {
+      createdAt: { lte: addDays(startOfDay(new Date()), -DELETE_MAILGUN_WEBHOOK_DAYS) },
+    },
+  });
 }
 
 async function findUserByProviderId(provider: OauthProviderType, providerAccountId: string) {

prisma/migrations/20260123014634_email_webhook/migration.sql

@@ -0,0 +1,52 @@
+-- CreateTable
+CREATE TABLE "mailgun_webhook_event" (
+    "id" UUID NOT NULL DEFAULT uuid_generate_v4(),
+    "eventId" VARCHAR(255),
+    "event" VARCHAR(50) NOT NULL,
+    "timestamp" TIMESTAMP(3) NOT NULL,
+    "logLevel" VARCHAR(20),
+    "recipient" VARCHAR(255) NOT NULL,
+    "recipientDomain" VARCHAR(255) NOT NULL,
+    "recipientProvider" VARCHAR(100),
+    "subject" VARCHAR(500),
+    "messageId" VARCHAR(255),
+    "fromAddress" VARCHAR(255),
+    "toAddress" VARCHAR(255),
+    "messageSize" INTEGER,
+    "deliveryCode" INTEGER,
+    "deliveryMessage" VARCHAR(500),
+    "deliveryDescription" VARCHAR(500),
+    "deliveryEnhancedCode" VARCHAR(50),
+    "deliveryAttemptNo" INTEGER,
+    "deliveryMxHost" VARCHAR(255),
+    "deliverySessionSeconds" DOUBLE PRECISION,
+    "deliveryTls" BOOLEAN,
+    "deliveryCertVerified" BOOLEAN,
+    "envelopeSender" VARCHAR(255),
+    "envelopeSendingIp" VARCHAR(50),
+    "envelopeTransport" VARCHAR(20),
+    "tags" TEXT[] DEFAULT ARRAY[]::TEXT[],
+    "userVariables" JSON,
+    "flags" JSON,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "mailgun_webhook_event_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "mailgun_webhook_event_createdAt_idx" ON "mailgun_webhook_event"("createdAt");
+
+-- CreateIndex
+CREATE INDEX "mailgun_webhook_event_event_createdAt_idx" ON "mailgun_webhook_event"("event", "createdAt");
+
+-- CreateIndex
+CREATE INDEX "mailgun_webhook_event_recipient_createdAt_idx" ON "mailgun_webhook_event"("recipient", "createdAt");
+
+-- CreateIndex
+CREATE INDEX "mailgun_webhook_event_recipientDomain_createdAt_idx" ON "mailgun_webhook_event"("recipientDomain", "createdAt");
+
+-- CreateIndex
+CREATE INDEX "mailgun_webhook_event_deliveryCode_event_idx" ON "mailgun_webhook_event"("deliveryCode", "event");
+
+-- CreateIndex
+CREATE INDEX "mailgun_webhook_event_timestamp_idx" ON "mailgun_webhook_event"("timestamp");

prisma/schema.prisma

@@ -337,6 +337,57 @@ model EmailActivity {
   createdAt  DateTime @default(now())
 }
 
+model MailgunWebhookEvent {
+  id        String   @id @default(dbgenerated("uuid_generate_v4()")) @db.Uuid
+  eventId   String?  @db.VarChar(255)
+  event     String   @db.VarChar(50)
+  timestamp DateTime
+  logLevel  String?  @db.VarChar(20)
+
+  // Recipient information
+  recipient         String  @db.VarChar(255)
+  recipientDomain   String  @db.VarChar(255)
+  recipientProvider String? @db.VarChar(100)
+
+  // Message information
+  subject     String? @db.VarChar(500)
+  messageId   String? @db.VarChar(255)
+  fromAddress String? @db.VarChar(255)
+  toAddress   String? @db.VarChar(255)
+  messageSize Int?
+
+  // Delivery status
+  deliveryCode           Int?
+  deliveryMessage        String?  @db.VarChar(500)
+  deliveryDescription    String?  @db.VarChar(500)
+  deliveryEnhancedCode   String?  @db.VarChar(50)
+  deliveryAttemptNo      Int?
+  deliveryMxHost         String?  @db.VarChar(255)
+  deliverySessionSeconds Float?
+  deliveryTls            Boolean?
+  deliveryCertVerified   Boolean?
+
+  // Envelope information
+  envelopeSender    String? @db.VarChar(255)
+  envelopeSendingIp String? @db.VarChar(50)
+  envelopeTransport String? @db.VarChar(20)
+
+  // Additional metadata
+  tags          String[] @default([])
+  userVariables Json?    @db.Json
+  flags         Json?    @db.Json
+
+  createdAt DateTime @default(now())
+
+  @@index([createdAt])
+  @@index([event, createdAt])
+  @@index([recipient, createdAt])
+  @@index([recipientDomain, createdAt])
+  @@index([deliveryCode, event])
+  @@index([timestamp])
+  @@map("mailgun_webhook_event")
+}
+
 model UserPreference {
   id                     String   @id @default(dbgenerated("uuid_generate_v4()")) @db.Uuid
   user                   User     @relation(fields: [userId], references: [id], onDelete: Cascade)