vynte-infra/plunk
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Added platform emails for billing limits and disabled projects

Browse Source
This commit is contained in:
Dries Augustyns
2025-12-12 12:28:53 +01:00
parent cb40669385
commit 2485d2ff1d
29 changed files with 768 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/__tests__/integration/actions.test.ts
+7
View File
@@ -80,6 +80,7 @@ describe('Actions API Integration Tests', () => {
it('should validate subject and body required when no template', () => {
const result = ActionSchemas.send.safeParse({
to: '[email protected]',
from: '[email protected]',
// Missing subject, body, and template
});
@@ -169,6 +170,7 @@ describe('Actions API Integration Tests', () => {
it('should accept to as string (backward compatible)', () => {
const result = ActionSchemas.send.safeParse({
to: '[email protected]',
from: '[email protected]',
subject: 'Test',
body: 'Test',
});
@@ -182,6 +184,7 @@ describe('Actions API Integration Tests', () => {
name: 'Jane Doe',
email: '[email protected]',
},
from: '[email protected]',
subject: 'Test',
body: 'Test',
});
@@ -194,6 +197,7 @@ describe('Actions API Integration Tests', () => {
to: {
email: '[email protected]',
},
from: '[email protected]',
subject: 'Test',
body: 'Test',
});
@@ -204,6 +208,7 @@ describe('Actions API Integration Tests', () => {
it('should accept to as array of strings', () => {
const result = ActionSchemas.send.safeParse({
to: ['[email protected]', '[email protected]'],
from: '[email protected]',
subject: 'Test',
body: 'Test',
});
@@ -217,6 +222,7 @@ describe('Actions API Integration Tests', () => {
{name: 'Jane Doe', email: '[email protected]'},
{name: 'John Smith', email: '[email protected]'},
],
from: '[email protected]',
subject: 'Test',
body: 'Test',
});
@@ -227,6 +233,7 @@ describe('Actions API Integration Tests', () => {
it('should accept to as mixed array of strings and objects', () => {
const result = ActionSchemas.send.safeParse({
to: ['[email protected]', {name: 'John Smith', email: '[email protected]'}],
from: '[email protected]',
subject: 'Test',
body: 'Test',
});
apps/api/src/app.ts
+7 -1
View File
@@ -14,12 +14,13 @@ import {
GOOGLE_OAUTH_ENABLED,
LANDING_URI,
NODE_ENV,
PLUNK_ENABLED,
PORT,
S3_ENABLED,
SMTP_ENABLED,
STRIPE_ENABLED,
TRACKING_TOGGLE_ENABLED,
WIKI_URI,
WIKI_URI
} from './app/constants.js';
import {Actions} from './controllers/Actions.js';
import {Activity} from './controllers/Activity.js';
@@ -351,6 +352,11 @@ void prisma.$connect().then(async () => {
? 'Per-project tracking toggle enabled'
: 'Always tracking or always no-tracking',
},
{
name: 'Platform emails',
enabled: PLUNK_ENABLED,
details: PLUNK_ENABLED ? 'Platform email notifications enabled' : 'PLUNK_API_KEY not configured',
},
];
const rows = features.map(f => ({
apps/api/src/app/constants.ts
+4
View File
@@ -94,3 +94,7 @@ export const SMTP_PORT_SUBMISSION = Number(validateEnv('PORT_SUBMISSION', '587')
// Enable SMTP features only when explicitly enabled via env or when a non-default domain is configured
export const SMTP_ENABLED =
process.env.SMTP_ENABLED === 'true' || (SMTP_DOMAIN !== 'localhost' && NODE_ENV !== 'development');
export const PLUNK_API_KEY = validateEnv('PLUNK_API_KEY', '');
export const PLUNK_FROM_ADDRESS = validateEnv('PLUNK_FROM_ADDRESS', '');
export const PLUNK_ENABLED = PLUNK_API_KEY !== '' && PLUNK_FROM_ADDRESS !== '';
apps/api/src/controllers/Actions.ts
+15 -8
View File
@@ -9,7 +9,7 @@ import {ContactService} from '../services/ContactService.js';
import {DomainService} from '../services/DomainService.js';
import {EmailService} from '../services/EmailService.js';
import {EventService} from '../services/EventService.js';
import {NotFound} from '../exceptions/index.js';
import {NotFound, ValidationError} from '../exceptions/index.js';
import {CatchAsync} from '../utils/asyncHandler.js';
/**
@@ -223,14 +223,21 @@ export class Actions {
templateId = templateRecord.id;
}
// Verify 'from' domain is verified if provided
const senderEmail = emailFrom || '[email protected]'; // Default sender
// Only verify custom domains (not the default [email protected])
if (emailFrom && emailFrom !== '[email protected]') {
await DomainService.verifyEmailDomain(emailFrom, auth.projectId);
if (!emailFrom) {
throw new ValidationError(
[
{
field: 'from',
message: 'Sender email is required either in request or template',
code: 'required',
},
],
'Could not parse sender email',
);
}
await DomainService.verifyEmailDomain(emailFrom, auth.projectId);
const replyToEmail = emailReplyTo;
const timestamp = new Date();
@@ -281,7 +288,7 @@ export class Actions {
contactId: contact.id,
subject: renderedSubject,
body: renderedBody,
from: senderEmail,
from: emailFrom,
fromName: emailFromName,
toName: recipient.name,
replyTo: replyToEmail,
apps/api/src/services/ActivityService.ts
+2 -1
View File
@@ -3,6 +3,7 @@ import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {redis} from '../database/redis.js';
import {Keys} from './keys.js';
/**
* Activity types that can be tracked
@@ -171,7 +172,7 @@ export class ActivityService {
*/
public static async getStats(projectId: string, startDate?: Date, endDate?: Date): Promise<ActivityStats> {
// Try to get from cache
const cacheKey = `activity:stats:${projectId}:${startDate?.getTime() || 'all'}:${endDate?.getTime() || 'now'}`;
const cacheKey = Keys.Activity.stats(projectId, startDate?.getTime() || 'all', endDate?.getTime() || 'now');
try {
const cached = await redis.get(cacheKey);
apps/api/src/services/AnalyticsService.ts
+4 -3
View File
@@ -1,5 +1,6 @@
import {prisma} from '../database/prisma.js';
import {redis} from '../database/redis.js';
import {Keys} from './keys.js';
/**
* Time series data point for analytics
@@ -57,7 +58,7 @@ export class AnalyticsService {
const limitedStartDate = effectiveStartDate < maxStartDate ? maxStartDate : effectiveStartDate;
// Check cache first
const cacheKey = `analytics:timeseries:${projectId}:${limitedStartDate.toISOString()}:${effectiveEndDate.toISOString()}`;
const cacheKey = Keys.Analytics.timeseries(projectId, limitedStartDate.toISOString(), effectiveEndDate.toISOString());
const cached = await redis.get(cacheKey);
if (cached) {
return JSON.parse(cached);
@@ -190,7 +191,7 @@ export class AnalyticsService {
const effectiveEndDate = endDate || now;
// Check cache
const cacheKey = `analytics:campaignStats:${projectId}:${effectiveStartDate.toISOString()}:${effectiveEndDate.toISOString()}`;
const cacheKey = Keys.Analytics.campaignStats(projectId, effectiveStartDate.toISOString(), effectiveEndDate.toISOString());
const cached = await redis.get(cacheKey);
if (cached) {
return JSON.parse(cached);
@@ -294,7 +295,7 @@ export class AnalyticsService {
const effectiveEndDate = endDate || now;
// Check cache
const cacheKey = `analytics:topEvents:${projectId}:${limit}:${effectiveStartDate.toISOString()}:${effectiveEndDate.toISOString()}`;
const cacheKey = Keys.Analytics.topEvents(projectId, limit, effectiveStartDate.toISOString(), effectiveEndDate.toISOString());
const cached = await redis.get(cacheKey);
if (cached) {
return JSON.parse(cached);
apps/api/src/services/BillingLimitService.ts
+138 -3
View File
@@ -1,10 +1,13 @@
import {EmailSourceType} from '@plunk/db';
import {BillingLimitExceededEmail, BillingLimitWarningEmail, sendPlatformEmail} from '@plunk/email';
import React from 'react';
import signale from 'signale';
import {STRIPE_ENABLED} from '../app/constants.js';
import {DASHBOARD_URI, LANDING_URI, STRIPE_ENABLED} from '../app/constants.js';
import {stripe} from '../app/stripe.js';
import {prisma} from '../database/prisma.js';
import {redis} from '../database/redis.js';
import {Keys} from './keys.js';
import {NtfyService} from './NtfyService.js';
/**
@@ -238,6 +241,9 @@ export class BillingLimitService {
EmailSourceType.TRANSACTIONAL, // Use generic type for notification
);
// Send email notification
await this.sendLimitExceededEmail(projectId, project.name, totalUsage, freeLimit, 'Free Tier (All Types)');
return {
allowed: false,
warning: false,
@@ -259,6 +265,16 @@ export class BillingLimitService {
percentage,
EmailSourceType.TRANSACTIONAL, // Use generic type for notification
);
// Send email notification (only once per month)
await this.sendWarningEmail(
projectId,
project.name,
totalUsage,
freeLimit,
percentage,
'Free Tier (All Types)',
);
}
return {
@@ -299,6 +315,9 @@ export class BillingLimitService {
if (project) {
// Send notification about limit exceeded
await NtfyService.notifyBillingLimitExceeded(project.name, projectId, usage, limit, sourceType);
// Send email notification
await this.sendLimitExceededEmail(projectId, project.name, usage, limit, sourceType);
}
return {
@@ -322,7 +341,17 @@ export class BillingLimitService {
});
if (project) {
await NtfyService.notifyBillingLimitApproaching(project.name, projectId, usage, limit, percentage, sourceType);
await NtfyService.notifyBillingLimitApproaching(
project.name,
projectId,
usage,
limit,
percentage,
sourceType,
);
// Send email notification (only once per month)
await this.sendWarningEmail(projectId, project.name, usage, limit, percentage, sourceType);
}
}
@@ -484,7 +513,7 @@ export class BillingLimitService {
const now = new Date();
const year = now.getFullYear();
const month = String(now.getMonth() + 1).padStart(2, '0');
return `billing:usage:${projectId}:${sourceType}:${year}-${month}`;
return Keys.Billing.usage(projectId, sourceType, year, month);
}
/**
@@ -496,4 +525,110 @@ export class BillingLimitService {
const end = new Date(now.getFullYear(), now.getMonth() + 1, 1);
return {start, end};
}
/**
* Send billing limit warning email to project members
*/
private static async sendWarningEmail(
projectId: string,
projectName: string,
usage: number,
limit: number,
percentage: number,
sourceType: string,
): Promise<void> {
try {
// Check if we've already sent this warning email this month
const now = new Date();
const year = now.getFullYear();
const month = String(now.getMonth() + 1).padStart(2, '0');
const cacheKey = Keys.Billing.warningEmail(projectId, sourceType, year, month);
const alreadySent = await redis.get(cacheKey);
if (alreadySent === '1') {
return;
}
const members = await prisma.membership.findMany({
where: {projectId},
include: {user: {select: {email: true}}},
});
const emails = members.map(m => m.user.email);
if (emails.length === 0) {
return;
}
const template = React.createElement(BillingLimitWarningEmail, {
projectName,
projectId,
usage,
limit,
percentage,
sourceType,
dashboardUrl: DASHBOARD_URI,
landingUrl: LANDING_URI,
});
await Promise.all(emails.map(email => sendPlatformEmail(email, 'Billing Limit Warning', template)));
// Mark that we've sent the warning email (expires at end of month)
const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
await redis.setex(cacheKey, ttl, '1');
} catch (error) {
signale.error(`[BILLING_LIMIT] Failed to send warning email:`, error);
}
}
/**
* Send billing limit exceeded email to project members
*/
private static async sendLimitExceededEmail(
projectId: string,
projectName: string,
usage: number,
limit: number,
sourceType: string,
): Promise<void> {
try {
// Check if we've already sent this warning email this month
const now = new Date();
const year = now.getFullYear();
const month = String(now.getMonth() + 1).padStart(2, '0');
const cacheKey = Keys.Billing.limitEmail(projectId, sourceType, year, month);
const alreadySent = await redis.get(cacheKey);
if (alreadySent === '1') {
return;
}
const members = await prisma.membership.findMany({
where: {projectId},
include: {user: {select: {email: true}}},
});
const emails = members.map(m => m.user.email);
if (emails.length === 0) {
return;
}
const template = React.createElement(BillingLimitExceededEmail, {
projectName,
projectId,
usage,
limit,
sourceType,
dashboardUrl: DASHBOARD_URI,
landingUrl: LANDING_URI,
});
await Promise.all(emails.map(email => sendPlatformEmail(email, 'Billing Limit Exceeded', template)));
// Mark that we've sent the limit email (expires at end of month)
const endOfMonth = new Date(now.getFullYear(), now.getMonth() + 1, 1);
const ttl = Math.floor((endOfMonth.getTime() - now.getTime()) / 1000);
await redis.setex(cacheKey, ttl, '1');
} catch (error) {
signale.error(`[BILLING_LIMIT] Failed to send limit exceeded email:`, error);
}
}
}
apps/api/src/services/EventService.ts
+3 -2
View File
@@ -5,6 +5,7 @@ import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {redis} from '../database/redis.js';
import {Keys} from './keys.js';
import {WorkflowExecutionService} from './WorkflowExecutionService.js';
@@ -49,7 +50,7 @@ export class EventService {
* Should be called when workflows are enabled/disabled or updated
*/
public static async invalidateWorkflowCache(projectId: string): Promise<void> {
const cacheKey = `workflows:enabled:${projectId}`;
const cacheKey = Keys.Workflow.enabled(projectId);
try {
await redis.del(cacheKey);
} catch (error) {
@@ -323,7 +324,7 @@ export class EventService {
data?: Record<string, unknown>,
): Promise<void> {
// Try to get workflows from cache
const cacheKey = `workflows:enabled:${projectId}`;
const cacheKey = Keys.Workflow.enabled(projectId);
let workflows;
try {
apps/api/src/services/SecurityService.ts
+29 -10
View File
@@ -1,9 +1,13 @@
import {ProjectDisabledEmail, sendPlatformEmail} from '@plunk/email';
import React from 'react';
import signale from 'signale';
import {prisma} from '../database/prisma.js';
import {redis} from '../database/redis.js';
import {Keys} from './keys.js';
import {NtfyService} from './NtfyService.js';
import {QueueService} from './QueueService.js';
import {DASHBOARD_URI, LANDING_URI} from '../app/constants.js';
/**
* Security thresholds for bounce and complaint rates
@@ -45,7 +49,6 @@ interface SecurityStatus {
}
export class SecurityService {
private static readonly CACHE_PREFIX = 'security';
private static readonly CACHE_TTL = 300; // 5 minutes
/**
@@ -54,7 +57,7 @@ export class SecurityService {
public static async getSecurityStatus(projectId: string): Promise<SecurityStatus> {
try {
// Try to get from cache first
const cacheKey = this.getCacheKey(projectId, 'rates');
const cacheKey = Keys.Security.rates(projectId);
const cached = await redis.get(cacheKey);
if (cached) {
@@ -137,7 +140,7 @@ export class SecurityService {
*/
public static async invalidateCache(projectId: string): Promise<void> {
try {
const cacheKey = this.getCacheKey(projectId, 'rates');
const cacheKey = Keys.Security.rates(projectId);
await redis.del(cacheKey);
} catch (error) {
signale.error(`[SECURITY] Failed to invalidate cache for project ${projectId}:`, error);
@@ -167,13 +170,6 @@ export class SecurityService {
};
}
/**
* Get cache key for security metrics
*/
private static getCacheKey(projectId: string, type: 'rates'): string {
return `${this.CACHE_PREFIX}:${projectId}:${type}`;
}
/**
* Calculate bounce and complaint rates for a project
*/
@@ -345,6 +341,29 @@ export class SecurityService {
// Send urgent notification about project suspension
await NtfyService.notifyProjectDisabledForSecurity(project.name, projectId, status.violations);
// Send email notification to project members
try {
const members = await prisma.membership.findMany({
where: {projectId},
include: {user: {select: {email: true}}},
});
const emails = members.map(m => m.user.email);
if (emails.length > 0) {
const template = React.createElement(ProjectDisabledEmail, {
projectName: project.name,
projectId,
violations: status.violations,
dashboardUrl: DASHBOARD_URI,
landingUrl: LANDING_URI,
});
await Promise.all(
emails.map(email => sendPlatformEmail(email, 'Project Disabled - Security Risk', template)),
);
}
} catch (emailError) {
signale.error(`[SECURITY] Failed to send project disabled email:`, emailError);
}
} catch (error) {
signale.error(`[SECURITY] Failed to disable project ${projectId}:`, error);
}
apps/api/src/services/__tests__/EmailService.test.ts
+4 -15
View File
@@ -1,4 +1,4 @@
import {describe, it, expect, beforeEach, vi} from 'vitest';
import {beforeEach, describe, expect, it, vi} from 'vitest';
import {EmailSourceType, EmailStatus} from '@plunk/db';
import {ActionSchemas} from '@plunk/shared';
import {EmailService} from '../EmailService';
@@ -669,7 +669,6 @@ describe('EmailService', () => {
// ========================================
describe('Attachment Schema Validation', () => {
it('should validate attachment count limit (max 10)', () => {
const tooManyAttachments = Array.from({length: 11}, (_, i) => ({
filename: `file${i}.txt`,
content: Buffer.from('content').toString('base64'),
@@ -690,7 +689,6 @@ describe('EmailService', () => {
});
it('should validate attachment size limit (10MB total)', () => {
// Exceeds ~13.3M base64 chars limit
const largeContent = 'A'.repeat(14000000);
@@ -711,11 +709,11 @@ describe('EmailService', () => {
});
it('should accept attachments within size limit', () => {
const validContent = Buffer.from('Small file content').toString('base64');
const result = ActionSchemas.send.safeParse({
to: '[email protected]',
from: '[email protected]',
subject: 'Test',
body: 'Test',
attachments: [
@@ -731,7 +729,6 @@ describe('EmailService', () => {
});
it('should reject attachment with missing required fields', () => {
const result = ActionSchemas.send.safeParse({
to: '[email protected]',
subject: 'Test',
@@ -748,7 +745,6 @@ describe('EmailService', () => {
});
it('should reject attachment with empty filename', () => {
const result = ActionSchemas.send.safeParse({
to: '[email protected]',
subject: 'Test',
@@ -766,7 +762,6 @@ describe('EmailService', () => {
});
it('should reject attachment with filename exceeding 255 chars', () => {
const tooLongFilename = 'a'.repeat(256) + '.pdf';
const result = ActionSchemas.send.safeParse({
@@ -786,18 +781,12 @@ describe('EmailService', () => {
});
it('should accept valid attachment with various content types', () => {
const contentTypes = [
'application/pdf',
'image/png',
'image/jpeg',
'text/plain',
'application/zip',
];
const contentTypes = ['application/pdf', 'image/png', 'image/jpeg', 'text/plain', 'application/zip'];
for (const contentType of contentTypes) {
const result = ActionSchemas.send.safeParse({
to: '[email protected]',
from: '[email protected]',
subject: 'Test',
body: 'Test',
attachments: [
apps/api/src/services/__tests__/EventService.test.ts
+2 -1
View File
@@ -1,6 +1,7 @@
import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest';
import {WorkflowExecutionStatus, WorkflowTriggerType} from '@plunk/db';
import {EventService} from '../EventService';
import {Keys} from '../keys';
import {factories, getPrismaClient} from '../../../../../test/helpers';
// Mock Redis for caching tests - must be inline to avoid hoisting issues
@@ -503,7 +504,7 @@ describe('EventService', () => {
const {redis} = await import('../../database/redis');
// Set cache
const cacheKey = `workflows:enabled:${projectId}`;
const cacheKey = Keys.Workflow.enabled(projectId);
await redis.set(cacheKey, JSON.stringify([{id: 'test'}]));
// Verify cache exists
apps/api/src/services/__tests__/WorkflowService.test.ts
+4 -3
View File
@@ -1,6 +1,7 @@
import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest';
import {WorkflowExecutionStatus, WorkflowStepType, WorkflowTriggerType} from '@plunk/db';
import {WorkflowService} from '../WorkflowService';
import {Keys} from '../keys';
import {factories, getPrismaClient} from '../../../../../test/helpers';
// Mock Redis for caching tests - must be inline to avoid hoisting issues
@@ -130,7 +131,7 @@ describe('WorkflowService', () => {
it('should invalidate cache when creating enabled workflow', async () => {
const {redis} = await import('../../database/redis');
const cacheKey = `workflows:enabled:${projectId}`;
const cacheKey = Keys.Workflow.enabled(projectId);
// Set cache
await redis.set(cacheKey, JSON.stringify([{id: 'old'}]));
@@ -286,7 +287,7 @@ describe('WorkflowService', () => {
it('should invalidate cache when enabling workflow', async () => {
const {redis} = await import('../../database/redis');
const cacheKey = `workflows:enabled:${projectId}`;
const cacheKey = Keys.Workflow.enabled(projectId);
const workflow = await factories.createWorkflow({
projectId,
@@ -326,7 +327,7 @@ describe('WorkflowService', () => {
it('should invalidate cache when deleting enabled workflow', async () => {
const {redis} = await import('../../database/redis');
const cacheKey = `workflows:enabled:${projectId}`;
const cacheKey = Keys.Workflow.enabled(projectId);
const workflow = await factories.createWorkflow({
projectId,
apps/api/src/services/keys.ts
+37
View File
@@ -15,4 +15,41 @@ export const Keys = {
return `domain:project:${projectId}`;
},
},
Billing: {
usage(projectId: string, sourceType: string, year: number, month: string): string {
return `billing:usage:${projectId}:${sourceType}:${year}-${month}`;
},
warningEmail(projectId: string, sourceType: string, year: number, month: string): string {
return `billing:warning_email:${projectId}:${sourceType}:${year}-${month}`;
},
limitEmail(projectId: string, sourceType: string, year: number, month: string): string {
return `billing:limit_email:${projectId}:${sourceType}:${year}-${month}`;
},
},
Security: {
rates(projectId: string): string {
return `security:${projectId}:rates`;
},
},
Activity: {
stats(projectId: string, startTime: number | string, endTime: number | string): string {
return `activity:stats:${projectId}:${startTime}:${endTime}`;
},
},
Analytics: {
timeseries(projectId: string, startDate: string, endDate: string): string {
return `analytics:timeseries:${projectId}:${startDate}:${endDate}`;
},
campaignStats(projectId: string, startDate: string, endDate: string): string {
return `analytics:campaignStats:${projectId}:${startDate}:${endDate}`;
},
topEvents(projectId: string, limit: number, startDate: string, endDate: string): string {
return `analytics:topEvents:${projectId}:${limit}:${startDate}:${endDate}`;
},
},
Workflow: {
enabled(projectId: string): string {
return `workflows:enabled:${projectId}`;
},
},
} as const;