Refactor monolithic modules and harden booking, email, and auth infrastructure.

Split oversized frontend API client, email service, and admin/booking pages into focused modules while preserving import surfaces, and add Redis-backed queues, stale booking cleanup, stronger auth, and scale deployment configs. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-25 07:12:59 +00:00
parent f0e2de2834
commit 613bd7be1d
75 changed files with 7702 additions and 5580 deletions
@@ -1,17 +1,16 @@
-// In-memory email queue with rate limiting
-// Processes emails asynchronously in the background without blocking the request thread
+// Durable email queue with rate limiting.
+// Jobs are persisted in the `email_queue` DB table so they survive process
+// restarts. Emails are processed asynchronously in the background without
+// blocking the request thread.

-import { generateId } from './utils.js';
+import { eq, and, asc, sql } from 'drizzle-orm';
+import { db, dbGet, emailQueue } from '../db/index.js';
+import { generateId, getNow } from './utils.js';
+import { isRedisEnabled } from './redis.js';
+import { getRateLimiter } from './stores/rateLimiter.js';

 // ==================== Types ====================

-export interface EmailJob {
-  id: string;
-  type: 'template';
-  params: TemplateEmailJobParams;
-  addedAt: number;
-}
-
 export interface TemplateEmailJobParams {
  templateSlug: string;
  to: string;
@@ -22,6 +21,11 @@ export interface TemplateEmailJobParams {
  sentBy?: string;
 }

+interface ClaimedJob {
+  id: string;
+  params: TemplateEmailJobParams;
+}
+
 export interface QueueStatus {
  queued: number;
  processing: boolean;
@@ -31,7 +35,8 @@ export interface QueueStatus {

 // ==================== Queue State ====================

-const queue: EmailJob[] = [];
+// Tracks send timestamps for the per-process (non-Redis) sliding-window rate
+// limit. The job backlog itself lives in the database, not in memory.
 const sentTimestamps: number[] = [];
 let processing = false;
 let processTimer: ReturnType<typeof setTimeout> | null = null;
@@ -41,7 +46,6 @@ let _emailService: any = null;

 function getEmailService() {
  if (!_emailService) {
-    // Dynamic import to avoid circular dependency
    throw new Error('[EmailQueue] Email service not initialized. Call initEmailQueue() first.');
  }
  return _emailService;
@@ -50,12 +54,31 @@ function getEmailService() {
 /**
 * Initialize the email queue with a reference to the email service.
 * Must be called once at startup.
+ *
+ * Also performs crash recovery: any jobs left in the 'processing' state by a
+ * previous (crashed) process are reset to 'pending' so they get retried.
 */
 export function initEmailQueue(emailService: any): void {
  _emailService = emailService;
+  recoverProcessingJobs()
+    .then((recovered) => {
+      if (recovered > 0) {
+        console.log(`[EmailQueue] Recovered ${recovered} in-flight job(s) after restart`);
+      }
+      scheduleProcessing();
+    })
+    .catch((err) => console.error('[EmailQueue] Recovery failed:', err?.message || err));
  console.log('[EmailQueue] Initialized');
 }

+async function recoverProcessingJobs(): Promise<number> {
+  const result: any = await (db as any)
+    .update(emailQueue)
+    .set({ status: 'pending' })
+    .where(eq((emailQueue as any).status, 'processing'));
+  return result?.changes ?? result?.rowCount ?? 0;
+}
+
 // ==================== Rate Limiting ====================

 function getMaxPerHour(): number {
@@ -74,19 +97,26 @@ function cleanOldTimestamps(): void {

 // ==================== Queue Operations ====================

+async function insertJob(id: string, params: TemplateEmailJobParams): Promise<void> {
+  await (db as any).insert(emailQueue).values({
+    id,
+    params: JSON.stringify(params),
+    status: 'pending',
+    attempts: 0,
+    createdAt: getNow(),
+  });
+}
+
 /**
 * Add a single email job to the queue.
- * Returns the job ID.
+ * Returns the job ID. Persistence happens asynchronously so the caller is not
+ * blocked; processing is scheduled once the row is written.
 */
 export function enqueueEmail(params: TemplateEmailJobParams): string {
  const id = generateId();
-  queue.push({
-    id,
-    type: 'template',
-    params,
-    addedAt: Date.now(),
-  });
-  scheduleProcessing();
+  insertJob(id, params)
+    .then(() => scheduleProcessing())
+    .catch((err) => console.error('[EmailQueue] Failed to enqueue email:', err?.message || err));
  return id;
 }

@@ -95,31 +125,32 @@ export function enqueueEmail(params: TemplateEmailJobParams): string {
 * Returns array of job IDs.
 */
 export function enqueueBulkEmails(paramsList: TemplateEmailJobParams[]): string[] {
-  const ids: string[] = [];
-  for (const params of paramsList) {
-    const id = generateId();
-    queue.push({
-      id,
-      type: 'template',
-      params,
-      addedAt: Date.now(),
-    });
-    ids.push(id);
-  }
-  if (ids.length > 0) {
-    console.log(`[EmailQueue] Queued ${ids.length} emails for background processing`);
-    scheduleProcessing();
-  }
+  const ids = paramsList.map(() => generateId());
+  if (ids.length === 0) return ids;
+
+  Promise.all(paramsList.map((params, i) => insertJob(ids[i], params)))
+    .then(() => {
+      console.log(`[EmailQueue] Queued ${ids.length} emails for background processing`);
+      scheduleProcessing();
+    })
+    .catch((err) => console.error('[EmailQueue] Failed to enqueue bulk emails:', err?.message || err));
+
  return ids;
 }

 /**
 * Get current queue status
 */
-export function getQueueStatus(): QueueStatus {
+export async function getQueueStatus(): Promise<QueueStatus> {
  cleanOldTimestamps();
+  const row = await dbGet<any>(
+    (db as any)
+      .select({ count: sql<number>`count(*)` })
+      .from(emailQueue)
+      .where(eq((emailQueue as any).status, 'pending'))
+  );
  return {
-    queued: queue.length,
+    queued: Number(row?.count || 0),
    processing,
    sentInLastHour: sentTimestamps.length,
    maxPerHour: getMaxPerHour(),
@@ -135,46 +166,125 @@ function scheduleProcessing(): void {
  setImmediate(() => processNext());
 }

+/**
+ * Atomically claim the oldest pending job by flipping its status to
+ * 'processing'. Returns null if there is nothing to do. The conditional update
+ * (WHERE status='pending') guards against two workers claiming the same row.
+ */
+async function claimNextJob(): Promise<ClaimedJob | null> {
+  for (let attempt = 0; attempt < 5; attempt++) {
+    const row = await dbGet<any>(
+      (db as any)
+        .select({ id: (emailQueue as any).id, params: (emailQueue as any).params })
+        .from(emailQueue)
+        .where(eq((emailQueue as any).status, 'pending'))
+        .orderBy(asc((emailQueue as any).createdAt))
+        .limit(1)
+    );
+    if (!row) return null;
+
+    const result: any = await (db as any)
+      .update(emailQueue)
+      .set({ status: 'processing' })
+      .where(and(
+        eq((emailQueue as any).id, row.id),
+        eq((emailQueue as any).status, 'pending')
+      ));
+
+    const affected = result?.changes ?? result?.rowCount ?? 0;
+    if (affected > 0) {
+      try {
+        return { id: row.id, params: JSON.parse(row.params) };
+      } catch {
+        // Corrupt params: mark failed and move on rather than crash-looping.
+        await markJob(row.id, 'failed', 'Invalid job params (JSON parse failed)');
+        continue;
+      }
+    }
+    // Lost the race for this row; try the next pending one.
+  }
+  return null;
+}
+
+async function markJob(id: string, status: 'sent' | 'failed', error?: string | null): Promise<void> {
+  const update: any = { status, processedAt: getNow() };
+  if (status === 'failed') {
+    update.attempts = sql`${(emailQueue as any).attempts} + 1`;
+    if (error) update.lastError = error.slice(0, 1000);
+  }
+  await (db as any).update(emailQueue).set(update).where(eq((emailQueue as any).id, id));
+}
+
+async function releaseJob(id: string): Promise<void> {
+  await (db as any)
+    .update(emailQueue)
+    .set({ status: 'pending' })
+    .where(eq((emailQueue as any).id, id));
+}
+
 async function processNext(): Promise<void> {
-  if (queue.length === 0) {
+  let job: ClaimedJob | null;
+  try {
+    job = await claimNextJob();
+  } catch (error: any) {
+    // Database error while claiming: back off and retry rather than stop.
+    console.error('[EmailQueue] Failed to claim next job:', error?.message || error);
+    processTimer = setTimeout(() => processNext(), 5_000);
+    return;
+  }
+
+  if (!job) {
    processing = false;
    console.log('[EmailQueue] Queue empty. Processing stopped.');
    return;
  }

-  // Rate limit check
+  // Rate limit check.
+  // - Without Redis: per-process sliding window.
+  // - With Redis: a shared hourly counter so the cap applies across all
+  //   instances rather than once per replica.
  cleanOldTimestamps();
  const maxPerHour = getMaxPerHour();
+  let waitMs = 0;

-  if (sentTimestamps.length >= maxPerHour) {
+  if (isRedisEnabled()) {
+    const result = await getRateLimiter().consume('email:hourly', maxPerHour, 3_600_000);
+    if (!result.allowed) {
+      waitMs = (result.retryAfter ?? 60) * 1000 + 500; // 500ms buffer
+    }
+  } else if (sentTimestamps.length >= maxPerHour) {
    // Calculate when the oldest timestamp in the window expires
-    const waitMs = sentTimestamps[0] + 3_600_000 - Date.now() + 500; // 500ms buffer
+    waitMs = sentTimestamps[0] + 3_600_000 - Date.now() + 500; // 500ms buffer
+  }
+
+  if (waitMs > 0) {
+    // Put the claimed job back so it is retried after the cooldown.
+    await releaseJob(job.id);
    console.log(
      `[EmailQueue] Rate limit reached (${maxPerHour}/hr). ` +
-      `Pausing for ${Math.ceil(waitMs / 1000)}s. ${queue.length} email(s) remaining.`
+      `Pausing for ${Math.ceil(waitMs / 1000)}s.`
    );
    processTimer = setTimeout(() => processNext(), waitMs);
    return;
  }

-  // Dequeue and process
-  const job = queue.shift()!;
-
  try {
    const emailService = getEmailService();
    await emailService.sendTemplateEmail(job.params);
    sentTimestamps.push(Date.now());
+    await markJob(job.id, 'sent');
    console.log(
      `[EmailQueue] Sent email ${job.id} to ${job.params.to}. ` +
-      `Queue: ${queue.length} remaining. Sent this hour: ${sentTimestamps.length}/${maxPerHour}`
+      `Sent this hour: ${sentTimestamps.length}/${maxPerHour}`
    );
  } catch (error: any) {
+    await markJob(job.id, 'failed', error?.message || String(error));
    console.error(
      `[EmailQueue] Failed to send email ${job.id} to ${job.params.to}:`,
      error?.message || error
    );
-    // The sendTemplateEmail method already logs the failure in the email_logs table,
-    // so we don't need to retry here. The error is logged and we move on.
+    // The sendTemplateEmail method already logs the failure in the email_logs
+    // table, so we just record it on the queue row and move on.
  }

  // Small delay between sends to be gentle on the email server
@@ -182,7 +292,8 @@ async function processNext(): Promise<void> {
 }

 /**
- * Stop processing (for graceful shutdown)
+ * Stop processing (for graceful shutdown). In-flight and pending jobs remain
+ * persisted in the database and resume on the next startup.
 */
 export function stopQueue(): void {
  if (processTimer) {
@@ -190,5 +301,5 @@ export function stopQueue(): void {
    processTimer = null;
  }
  processing = false;
-  console.log(`[EmailQueue] Stopped. ${queue.length} email(s) remaining in queue.`);
+  console.log('[EmailQueue] Stopped. Pending jobs remain persisted in the database.');
 }