feat: implement meter event processing with queue for Stripe billing

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<MeterEventJobData>): Promise<void> {
+  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<MeterEventJobData>(
+    meterQueue.name,
+    async (job: Job<MeterEventJobData>) => {
+      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;
+}

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);

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_ENABLED} from '../app/constants.js';
-import {stripe} from '../app/stripe.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<void> {
-    // Skip if billing is disabled (self-hosted mode)
+    if (!STRIPE_ENABLED) {
-    if (!STRIPE_ENABLED || !stripe) {
       return;
     }
 
-    // Skip if no customer ID (not subscribed)
     if (!customerId) {
       return;
     }
 
     try {
-      await stripe.billing.meterEvents.create({
+      await QueueService.queueMeterEvent(customerId, value, idempotencyKey);
-        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}`);
     } catch (error) {
-      // Log error but don't throw - we don't want billing issues to break email sending
+      signale.error('[METER] Failed to enqueue meter event:', error);
-      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');
-      }
     }
   }
 
-  /**
-   * 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<void> {
     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<unknown> {
-    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;
-    }
-  }
 }

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<BulkContactActionJobData>('bulk-contac
   },
 });
 
+export const meterQueue = new Queue<MeterEventJobData>('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<Job<MeterEventJobData>> {
+    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(),
     ]);
   }
 }

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;
+}