From e048fc157080b36f40fbfc4440db51367a2d374b Mon Sep 17 00:00:00 2001 From: Dries Augustyns Date: Mon, 20 Apr 2026 21:30:52 +0200 Subject: [PATCH] feat: implement meter event processing with queue for Stripe billing --- apps/api/src/jobs/meter-processor.ts | 58 +++++++++++++++++++ apps/api/src/jobs/worker.ts | 6 ++ apps/api/src/services/MeterService.ts | 80 ++------------------------- apps/api/src/services/QueueService.ts | 39 +++++++++++++ packages/types/src/jobs/email.ts | 10 ++++ 5 files changed, 118 insertions(+), 75 deletions(-) create mode 100644 apps/api/src/jobs/meter-processor.ts diff --git a/apps/api/src/jobs/meter-processor.ts b/apps/api/src/jobs/meter-processor.ts new file mode 100644 index 0000000..212581f --- /dev/null +++ b/apps/api/src/jobs/meter-processor.ts @@ -0,0 +1,58 @@ +/** + * Background Job: Meter Processor + * Reliably records Stripe billing meter events with automatic retries + */ + +import type {MeterEventJobData} from '@plunk/types'; +import {type Job, Worker} from 'bullmq'; +import signale from 'signale'; + +import {STRIPE_ENABLED, STRIPE_METER_EVENT_NAME} from '../app/constants.js'; +import {stripe} from '../app/stripe.js'; +import {meterQueue} from '../services/QueueService.js'; + +async function processMeterEvent(job: Job): Promise { + const {customerId, value, idempotencyKey} = job.data; + + if (!STRIPE_ENABLED || !stripe) { + return; + } + + await stripe.billing.meterEvents.create({ + event_name: STRIPE_METER_EVENT_NAME, + payload: { + stripe_customer_id: customerId, + value: value.toString(), + }, + ...(idempotencyKey && {identifier: idempotencyKey}), + }); + + signale.debug(`[METER-WORKER] Recorded ${value} email(s) for customer ${customerId}`); +} + +export function createMeterWorker(): Worker { + const worker = new Worker( + meterQueue.name, + async (job: Job) => { + await processMeterEvent(job); + }, + { + connection: meterQueue.opts.connection, + concurrency: 5, + limiter: { + max: 50, + duration: 1000, // Stay well within Stripe's rate limits + }, + }, + ); + + worker.on('failed', (job, error) => { + signale.error(`[METER-WORKER] Job ${job?.id} failed (attempt ${job?.attemptsMade}):`, error); + }); + + worker.on('error', error => { + signale.error('[METER-WORKER] Worker error:', error); + }); + + return worker; +} diff --git a/apps/api/src/jobs/worker.ts b/apps/api/src/jobs/worker.ts index 5735c0e..9bc8264 100644 --- a/apps/api/src/jobs/worker.ts +++ b/apps/api/src/jobs/worker.ts @@ -15,6 +15,7 @@ import {createCampaignWorker} from './campaign-processor.js'; import {createDomainVerificationWorker} from './domain-verification-processor.js'; import {createEmailWorker} from './email-processor.js'; import {createImportWorker} from './import-processor.js'; +import {createMeterWorker} from './meter-processor.js'; import {createScheduledCampaignWorker} from './scheduled-processor.js'; import {createSegmentCountWorker} from './segment-count-processor.js'; import {createWorkflowWorker} from './workflow-processor-queue.js'; @@ -70,6 +71,11 @@ async function startWorkers() { workers.push({name: 'api-request-cleanup', worker: apiRequestCleanupWorker}); signale.success('[WORKER] API request cleanup worker started'); + // Start meter worker + const meterWorker = createMeterWorker(); + workers.push({name: 'meter', worker: meterWorker}); + signale.success('[WORKER] Meter worker started'); + signale.success('[WORKER] All workers started successfully'); } catch (error) { signale.error('[WORKER] Failed to start workers:', error); diff --git a/apps/api/src/services/MeterService.ts b/apps/api/src/services/MeterService.ts index 995d2ed..5be15ab 100644 --- a/apps/api/src/services/MeterService.ts +++ b/apps/api/src/services/MeterService.ts @@ -1,98 +1,28 @@ import signale from 'signale'; -import {STRIPE_ENABLED, STRIPE_METER_EVENT_NAME} from '../app/constants.js'; -import {stripe} from '../app/stripe.js'; +import {STRIPE_ENABLED} from '../app/constants.js'; +import {QueueService} from './QueueService.js'; -/** - * Meter Service - * Handles recording usage events to Stripe billing meters for pay-as-you-go pricing - */ export class MeterService { - /** - * Record an email sent event to Stripe meter - * This is used for usage-based billing where customers pay per email sent - * - * @param customerId - Stripe customer ID - * @param value - Number of emails sent (default: 1) - * @param idempotencyKey - Optional unique identifier to prevent duplicate recording - */ public static async recordEmailSent(customerId: string, value = 1, idempotencyKey?: string): Promise { - // Skip if billing is disabled (self-hosted mode) - if (!STRIPE_ENABLED || !stripe) { + if (!STRIPE_ENABLED) { return; } - // Skip if no customer ID (not subscribed) if (!customerId) { return; } try { - await stripe.billing.meterEvents.create({ - event_name: STRIPE_METER_EVENT_NAME, - payload: { - stripe_customer_id: customerId, - value: value.toString(), // Must be a string - }, - ...(idempotencyKey && {identifier: idempotencyKey}), - }); - - signale.debug(`[METER] Recorded ${value} email(s) sent for customer ${customerId}`); + await QueueService.queueMeterEvent(customerId, value, idempotencyKey); } catch (error) { - // Log error but don't throw - we don't want billing issues to break email sending - signale.error('[METER] Failed to record email sent event:', error); - - // If it's a rate limit error, we might want to retry - if (error instanceof Error && error.message.includes('429')) { - signale.warn('[METER] Rate limited - consider implementing queue-based meter recording'); - } + signale.error('[METER] Failed to enqueue meter event:', error); } } - /** - * Record multiple emails in a single batch (for campaign sending) - * More efficient than individual calls when sending bulk emails - * - * @param customerId - Stripe customer ID - * @param count - Number of emails sent in this batch - * @param batchId - Unique identifier for this batch - */ public static async recordEmailBatch(customerId: string, count: number, batchId: string): Promise { if (count <= 0) return; await this.recordEmailSent(customerId, count, `batch_${batchId}`); } - - /** - * Get current usage for a customer in the current billing period - * Useful for displaying usage dashboards or implementing soft limits - * - * @param customerId - Stripe customer ID - * @param meterId - The meter ID from Stripe dashboard - * @param startTime - Start of period (Unix timestamp) - * @param endTime - End of period (Unix timestamp) - */ - public static async getUsageSummary( - meterId: string, - customerId: string, - startTime: number, - endTime: number, - ): Promise { - if (!STRIPE_ENABLED || !stripe) { - return null; - } - - try { - const summaries = await stripe.billing.meters.listEventSummaries(meterId, { - customer: customerId, - start_time: startTime, - end_time: endTime, - }); - - return summaries; - } catch (error) { - signale.error('[METER] Failed to get usage summary:', error); - return null; - } - } } diff --git a/apps/api/src/services/QueueService.ts b/apps/api/src/services/QueueService.ts index 89ddf12..747d5e2 100644 --- a/apps/api/src/services/QueueService.ts +++ b/apps/api/src/services/QueueService.ts @@ -7,6 +7,7 @@ import type { CampaignBatchJobData, ContactImportJobData, DomainVerificationJobData, + MeterEventJobData, ScheduledCampaignJobData, SegmentCountJobData, SendEmailJobData, @@ -158,6 +159,19 @@ export const bulkContactQueue = new Queue('bulk-contac }, }); +export const meterQueue = new Queue('meter', { + connection: redisConnection, + defaultJobOptions: { + attempts: 10, + backoff: { + type: 'exponential', + delay: 5000, + }, + removeOnComplete: 5000, + removeOnFail: 10000, + }, +}); + /** * Queue Service - Centralized queue management */ @@ -313,6 +327,23 @@ export class QueueService { }; } + /** + * Queue a Stripe meter event for reliable delivery with retries + */ + public static async queueMeterEvent( + customerId: string, + value: number, + idempotencyKey?: string, + ): Promise> { + return meterQueue.add( + 'record-meter-event', + {customerId, value, idempotencyKey}, + { + jobId: idempotencyKey ? `meter-${idempotencyKey}` : undefined, + }, + ); + } + /** * Queue bulk contact action job */ @@ -390,6 +421,7 @@ export class QueueService { domainVerificationCounts, apiRequestCleanupCounts, bulkContactCounts, + meterCounts, ] = await Promise.all([ emailQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), campaignQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), @@ -400,6 +432,7 @@ export class QueueService { domainVerificationQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), apiRequestCleanupQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), bulkContactQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), + meterQueue.getJobCounts('waiting', 'active', 'completed', 'failed', 'delayed'), ]); return { @@ -412,6 +445,7 @@ export class QueueService { domainVerification: domainVerificationCounts, apiRequestCleanup: apiRequestCleanupCounts, bulkContact: bulkContactCounts, + meter: meterCounts, }; } @@ -429,6 +463,7 @@ export class QueueService { domainVerificationQueue.pause(), apiRequestCleanupQueue.pause(), bulkContactQueue.pause(), + meterQueue.pause(), ]); } @@ -446,6 +481,7 @@ export class QueueService { domainVerificationQueue.resume(), apiRequestCleanupQueue.resume(), bulkContactQueue.resume(), + meterQueue.resume(), ]); } @@ -472,6 +508,8 @@ export class QueueService { domainVerificationQueue.clean(gracePeriod * 7, 50, 'failed'), bulkContactQueue.clean(gracePeriod, 50, 'completed'), bulkContactQueue.clean(gracePeriod * 7, 100, 'failed'), + meterQueue.clean(gracePeriod * 30, 5000, 'completed'), // Keep 30 days for billing audit + meterQueue.clean(gracePeriod * 30, 10000, 'failed'), ]); } @@ -557,6 +595,7 @@ export class QueueService { domainVerificationQueue.close(), apiRequestCleanupQueue.close(), bulkContactQueue.close(), + meterQueue.close(), ]); } } diff --git a/packages/types/src/jobs/email.ts b/packages/types/src/jobs/email.ts index 0bfd20b..a3a57d5 100644 --- a/packages/types/src/jobs/email.ts +++ b/packages/types/src/jobs/email.ts @@ -9,3 +9,13 @@ export interface SendEmailJobData { emailId: string; } + +/** + * Job data for recording a Stripe meter event + * Used by: meterQueue worker + */ +export interface MeterEventJobData { + customerId: string; + value: number; + idempotencyKey?: string; +}