feat: Add billing for inbound

apps/api/src/controllers/Users.ts

@@ -330,6 +330,7 @@ export class Users {
         billingLimitWorkflows: true,
         billingLimitCampaigns: true,
         billingLimitTransactional: true,
+        billingLimitInbound: true,
       },
     });
 
@@ -349,6 +350,7 @@ export class Users {
         billingLimitWorkflows: data.workflows,
         billingLimitCampaigns: data.campaigns,
         billingLimitTransactional: data.transactional,
+        billingLimitInbound: data.inbound,
       },
     });
 
@@ -360,6 +362,7 @@ export class Users {
         workflows: project.billingLimitWorkflows,
         campaigns: project.billingLimitCampaigns,
         transactional: project.billingLimitTransactional,
+        inbound: project.billingLimitInbound,
       },
       data,
     );

apps/api/src/controllers/Webhooks.ts

@@ -1,6 +1,6 @@
 import {Controller, Post} from '@overnightjs/core';
 import type {Prisma} from '@plunk/db';
-import {EmailStatus} from '@plunk/db';
+import {EmailSourceType, EmailStatus} from '@plunk/db';
 import type {Request, Response} from 'express';
 import signale from 'signale';
 import type Stripe from 'stripe';
@@ -8,8 +8,10 @@ import type Stripe from 'stripe';
 import {STRIPE_ENABLED, STRIPE_WEBHOOK_SECRET} from '../app/constants.js';
 import {stripe} from '../app/stripe.js';
 import {prisma} from '../database/prisma.js';
+import {BillingLimitService} from '../services/BillingLimitService.js';
 import {ContactService} from '../services/ContactService.js';
 import {EventService} from '../services/EventService.js';
+import {MeterService} from '../services/MeterService.js';
 import {NtfyService} from '../services/NtfyService.js';
 import {SecurityService} from '../services/SecurityService.js';
 import {CatchAsync} from '../utils/asyncHandler.js';
@@ -122,6 +124,16 @@ export class Webhooks {
             for (const domainRecord of domainRecords) {
               signale.info(`[WEBHOOK] Processing inbound email for project: ${domainRecord.project.name}`);
 
+              // Check billing limits before processing inbound email
+              const limitCheck = await BillingLimitService.checkLimit(domainRecord.projectId, EmailSourceType.INBOUND);
+
+              if (!limitCheck.allowed) {
+                signale.warn(
+                  `[WEBHOOK] Inbound email blocked for project ${domainRecord.project.name}: ${limitCheck.message}`,
+                );
+                continue; // Skip this project but continue processing for other projects
+              }
+
               // Find or create a contact for the sender in this project
               let contact;
               if (senderEmail) {
@@ -133,6 +145,32 @@ export class Webhooks {
                 );
               }
 
+              // Create an Email record for tracking (no actual email content since it's inbound)
+              const inboundEmail = await prisma.email.create({
+                data: {
+                  projectId: domainRecord.projectId,
+                  contactId: contact!.id,
+                  subject: body.mail?.commonHeaders?.subject || '(No subject)',
+                  body: '', // Inbound emails don't have body content in our system
+                  from: recipientEmail, // The recipient address that received the email
+                  sourceType: EmailSourceType.INBOUND,
+                  status: EmailStatus.DELIVERED, // Inbound emails are already delivered
+                  deliveredAt: new Date(body.mail?.timestamp || new Date()),
+                },
+              });
+
+              // Increment usage counter in cache
+              await BillingLimitService.incrementUsage(domainRecord.projectId, EmailSourceType.INBOUND);
+
+              // Record Stripe metering if project has customer
+              if (domainRecord.project.customer) {
+                await MeterService.recordEmailSent(
+                  domainRecord.project.customer,
+                  1, // Inbound emails count as 1 credit
+                  `email_${inboundEmail.id}`,
+                );
+              }
+
               // Prepare event data with all inbound email details
               const eventData = {
                 messageId: body.mail?.messageId,
@@ -158,7 +196,7 @@ export class Webhooks {
                 domainRecord.projectId,
                 'email.received',
                 contact?.id,
-                undefined, // No emailId for inbound emails (they're not sent by us)
+                inboundEmail.id, // Link the event to the inbound email record
                 eventData,
               );
 

apps/api/src/services/BillingLimitService.ts

@@ -157,6 +157,7 @@ export class BillingLimitService {
           billingLimitWorkflows: true,
           billingLimitCampaigns: true,
           billingLimitTransactional: true,
+          billingLimitInbound: true,
         },
       });
 
@@ -183,6 +184,9 @@ export class BillingLimitService {
         case EmailSourceType.TRANSACTIONAL:
           limit = project.billingLimitTransactional;
           break;
+        case EmailSourceType.INBOUND:
+          limit = project.billingLimitInbound;
+          break;
         default:
           limit = null;
       }
@@ -191,7 +195,8 @@ export class BillingLimitService {
       const hasCustomLimits =
         project.billingLimitWorkflows !== null ||
         project.billingLimitCampaigns !== null ||
-        project.billingLimitTransactional !== null;
+        project.billingLimitTransactional !== null ||
+        project.billingLimitInbound !== null;
 
       // Free tier projects (no subscription and no custom limits): enforce total 1000 email/month limit
       // Only enforce free tier limits if billing is enabled
@@ -368,6 +373,7 @@ export class BillingLimitService {
           billingLimitWorkflows: true,
           billingLimitCampaigns: true,
           billingLimitTransactional: true,
+          billingLimitInbound: true,
         },
       });
 
@@ -390,10 +396,11 @@ export class BillingLimitService {
       }
 
       // Get usage for all categories in parallel
-      const [workflowUsage, campaignUsage, transactionalUsage] = await Promise.all([
+      const [workflowUsage, campaignUsage, transactionalUsage, inboundUsage] = await Promise.all([
         this.getUsage(projectId, EmailSourceType.WORKFLOW),
         this.getUsage(projectId, EmailSourceType.CAMPAIGN),
         this.getUsage(projectId, EmailSourceType.TRANSACTIONAL),
+        this.getUsage(projectId, EmailSourceType.INBOUND),
       ]);
 
       // Helper to calculate category usage
@@ -412,18 +419,19 @@ export class BillingLimitService {
       const hasCustomLimits =
         project.billingLimitWorkflows !== null ||
         project.billingLimitCampaigns !== null ||
-        project.billingLimitTransactional !== null;
+        project.billingLimitTransactional !== null ||
+        project.billingLimitInbound !== null;
 
       // Free tier projects (no subscription and no custom limits): show total usage with shared limit
       // Only show free tier limits if billing is enabled
       if (STRIPE_ENABLED && !project.subscription && !hasCustomLimits) {
-        const totalUsage = workflowUsage + campaignUsage + transactionalUsage;
+        const totalUsage = workflowUsage + campaignUsage + transactionalUsage + inboundUsage;
         const limit = this.FREE_TIER_TOTAL_LIMIT;
         const percentage = (totalUsage / limit) * 100;
         const isWarning = percentage >= this.WARNING_THRESHOLD * 100;
         const isBlocked = totalUsage >= limit;
 
-        // For free tier, show the same limit and total usage for all three categories
+        // For free tier, show the same limit and total usage for all four categories
         // This makes it clear in the UI that it's a shared limit
         const sharedUsageInfo: CategoryUsage = {
           limit,
@@ -437,6 +445,7 @@ export class BillingLimitService {
           workflows: sharedUsageInfo,
           campaigns: sharedUsageInfo,
           transactional: sharedUsageInfo,
+          inbound: sharedUsageInfo,
           currency,
         };
       }
@@ -446,6 +455,7 @@ export class BillingLimitService {
         workflows: calculateCategoryUsage(workflowUsage, project.billingLimitWorkflows),
         campaigns: calculateCategoryUsage(campaignUsage, project.billingLimitCampaigns),
         transactional: calculateCategoryUsage(transactionalUsage, project.billingLimitTransactional),
+        inbound: calculateCategoryUsage(inboundUsage, project.billingLimitInbound),
         currency,
       };
     } catch (error) {
@@ -470,6 +480,7 @@ export class BillingLimitService {
         this.getCacheKey(projectId, EmailSourceType.WORKFLOW),
         this.getCacheKey(projectId, EmailSourceType.CAMPAIGN),
         this.getCacheKey(projectId, EmailSourceType.TRANSACTIONAL),
+        this.getCacheKey(projectId, EmailSourceType.INBOUND),
       ];
 
       await Promise.all(keys.map(key => redis.del(key)));
@@ -493,11 +504,13 @@ export class BillingLimitService {
       workflows: number | null;
       campaigns: number | null;
       transactional: number | null;
+      inbound: number | null;
     },
     newLimits: {
       workflows: number | null;
       campaigns: number | null;
       transactional: number | null;
+      inbound: number | null;
     },
   ): Promise<void> {
     try {
@@ -531,6 +544,14 @@ export class BillingLimitService {
         );
       }
 
+      // Check inbound limit
+      if (oldLimits.inbound !== newLimits.inbound) {
+        keysToDelete.push(
+          Keys.Billing.warningEmail(projectId, EmailSourceType.INBOUND, year, month),
+          Keys.Billing.limitEmail(projectId, EmailSourceType.INBOUND, year, month),
+        );
+      }
+
       if (keysToDelete.length > 0) {
         await Promise.all(keysToDelete.map(key => redis.del(key)));
         signale.debug(

packages/db/prisma/migrations/20260218141637_add_inbound_email_tracking/migration.sql

@@ -0,0 +1,5 @@
+-- AlterEnum
+ALTER TYPE "EmailSourceType" ADD VALUE 'INBOUND';
+
+-- AlterTable
+ALTER TABLE "projects" ADD COLUMN     "billingLimitInbound" INTEGER;

packages/db/prisma/schema.prisma

@@ -53,6 +53,7 @@ model Project {
   billingLimitWorkflows     Int? // Max workflow emails per month
   billingLimitCampaigns     Int? // Max campaign emails per month
   billingLimitTransactional Int? // Max transactional emails per month
+  billingLimitInbound       Int? // Max inbound emails per month
 
   // Email Tracking
   tracking TrackingMode @default(ENABLED) // Open and click tracking mode
@@ -750,6 +751,7 @@ enum EmailSourceType {
   TRANSACTIONAL // Sent via API call
   CAMPAIGN // Sent as part of broadcast
   WORKFLOW // Sent via workflow automation
+  INBOUND // Received via SES inbound
 }
 
 enum EmailStatus {

packages/shared/src/schemas/index.ts

@@ -439,6 +439,7 @@ export const BillingLimitSchemas = {
     workflows: z.coerce.number().int().positive().nullable(),
     campaigns: z.coerce.number().int().positive().nullable(),
     transactional: z.coerce.number().int().positive().nullable(),
+    inbound: z.coerce.number().int().positive().nullable(),
   }),
 } as const;
 

packages/types/src/api/billing.ts

@@ -20,6 +20,7 @@ export interface BillingLimitsResponse {
   workflows: CategoryUsage;
   campaigns: CategoryUsage;
   transactional: CategoryUsage;
+  inbound: CategoryUsage;
   currency: string | null;
 }