Spanglish/backend/src/lib/emailQueue.ts

// 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 { 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 TemplateEmailJobParams {
  templateSlug: string;
  to: string;
  toName?: string;
  variables: Record<string, any>;
  locale?: string;
  eventId?: string;
  sentBy?: string;
}

interface ClaimedJob {
  id: string;
  params: TemplateEmailJobParams;
}

export interface QueueStatus {
  queued: number;
  processing: boolean;
  sentInLastHour: number;
  maxPerHour: number;
}

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

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

// Lazy reference to emailService to avoid circular imports
let _emailService: any = null;

function getEmailService() {
  if (!_emailService) {
    throw new Error('[EmailQueue] Email service not initialized. Call initEmailQueue() first.');
  }
  return _emailService;
}

/**
 * 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 {
  return parseInt(process.env.MAX_EMAILS_PER_HOUR || '30', 10);
}

/**
 * Clean up timestamps older than 1 hour
 */
function cleanOldTimestamps(): void {
  const oneHourAgo = Date.now() - 3_600_000;
  while (sentTimestamps.length > 0 && sentTimestamps[0] <= oneHourAgo) {
    sentTimestamps.shift();
  }
}

// ==================== 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. 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();
  insertJob(id, params)
    .then(() => scheduleProcessing())
    .catch((err) => console.error('[EmailQueue] Failed to enqueue email:', err?.message || err));
  return id;
}

/**
 * Add multiple email jobs to the queue at once.
 * Returns array of job IDs.
 */
export function enqueueBulkEmails(paramsList: TemplateEmailJobParams[]): string[] {
  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 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: Number(row?.count || 0),
    processing,
    sentInLastHour: sentTimestamps.length,
    maxPerHour: getMaxPerHour(),
  };
}

// ==================== Processing ====================

function scheduleProcessing(): void {
  if (processing) return;
  processing = true;
  // Start processing on next tick to not block the caller
  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> {
  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.
  // - 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 (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
    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.`
    );
    processTimer = setTimeout(() => processNext(), waitMs);
    return;
  }

  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}. ` +
      `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 just record it on the queue row and move on.
  }

  // Small delay between sends to be gentle on the email server
  processTimer = setTimeout(() => processNext(), 200);
}

/**
 * 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) {
    clearTimeout(processTimer);
    processTimer = null;
  }
  processing = false;
  console.log('[EmailQueue] Stopped. Pending jobs remain persisted in the database.');
}