From de7ed84fcfddde16a9297c947048e9876712f6fa Mon Sep 17 00:00:00 2001 From: Dries Augustyns Date: Wed, 18 Feb 2026 15:23:31 +0100 Subject: [PATCH] feat: Add billing for inbound --- apps/api/src/controllers/Users.ts | 3 ++ apps/api/src/controllers/Webhooks.ts | 42 ++++++++++++++++++- apps/api/src/services/BillingLimitService.ts | 31 +++++++++++--- .../migration.sql | 5 +++ packages/db/prisma/schema.prisma | 2 + packages/shared/src/schemas/index.ts | 1 + packages/types/src/api/billing.ts | 1 + 7 files changed, 78 insertions(+), 7 deletions(-) create mode 100644 packages/db/prisma/migrations/20260218141637_add_inbound_email_tracking/migration.sql diff --git a/apps/api/src/controllers/Users.ts b/apps/api/src/controllers/Users.ts index c688d86..1a2f2a0 100644 --- a/apps/api/src/controllers/Users.ts +++ b/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, ); diff --git a/apps/api/src/controllers/Webhooks.ts b/apps/api/src/controllers/Webhooks.ts index a6ce39b..8283af1 100644 --- a/apps/api/src/controllers/Webhooks.ts +++ b/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, ); diff --git a/apps/api/src/services/BillingLimitService.ts b/apps/api/src/services/BillingLimitService.ts index fe95dc3..ad4584a 100644 --- a/apps/api/src/services/BillingLimitService.ts +++ b/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 { 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( diff --git a/packages/db/prisma/migrations/20260218141637_add_inbound_email_tracking/migration.sql b/packages/db/prisma/migrations/20260218141637_add_inbound_email_tracking/migration.sql new file mode 100644 index 0000000..1bee406 --- /dev/null +++ b/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; diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index f49b2c2..bf364d8 100644 --- a/packages/db/prisma/schema.prisma +++ b/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 { diff --git a/packages/shared/src/schemas/index.ts b/packages/shared/src/schemas/index.ts index cc570d2..2aa61f1 100644 --- a/packages/shared/src/schemas/index.ts +++ b/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; diff --git a/packages/types/src/api/billing.ts b/packages/types/src/api/billing.ts index 9dc62b5..1cae02d 100644 --- a/packages/types/src/api/billing.ts +++ b/packages/types/src/api/billing.ts @@ -20,6 +20,7 @@ export interface BillingLimitsResponse { workflows: CategoryUsage; campaigns: CategoryUsage; transactional: CategoryUsage; + inbound: CategoryUsage; currency: string | null; }