bunker-admin 7895ce683e Tonne of debugging - getting ready for the production builds

2026-02-16 10:44:18 -07:00

25 KiB

Raw Blame History

Email Queue System

Overview

The email queue system manages asynchronous email sending for advocacy campaigns using BullMQ and Redis. It provides reliable email delivery, retry logic, job monitoring, and comprehensive tracking of email campaign effectiveness.

Key Capabilities:

BullMQ integration: Redis-backed job queue for email processing
Automatic retry logic: Failed emails retried with exponential backoff
Job status tracking: Monitor queued, active, completed, and failed jobs
Rate limiting: Prevent SMTP server overload
Email tracking: Track sent emails per campaign
Admin monitoring: Real-time queue statistics and job management
Test mode: Send to MailHog instead of SMTP for testing

Use Cases:

Bulk email sending for advocacy campaigns
Reliable email delivery with retry
Email campaign effectiveness tracking
SMTP server load management
Development email testing

Architecture

graph TD
    A[Public User] -->|Send Email| B[CampaignPage]
    B -->|POST /api/public/campaigns/send-email| C[Campaign Service]
    C -->|Add Job| D[Email Queue Service]
    D -->|Create Job| E[(BullMQ Redis)]

    F[Email Worker] -->|Poll Jobs| E
    F -->|Process Job| G{Send Email}
    G -->|Success| H[Email Service - SMTP]
    G -->|Failure| I[Retry Logic]

    H -->|Track| J[(CampaignEmail Model)]
    I -->|Backoff| E

    K[Admin User] -->|Monitor| L[EmailQueuePage]
    L -->|GET /api/email-queue/stats| D
    L -->|Pause/Resume| D
    L -->|Clean Jobs| D

    M[Prometheus] -->|Scrape| N[Metrics Endpoint]
    N -->|cm_email_queue_size| E

    style E fill:#fff4e1
    style J fill:#e1f5ff

Flow Description:

User sends email → Campaign service adds job to BullMQ queue
Worker polls queue → Picks up job for processing
Email sent via SMTP → Nodemailer sends email
Success → Job marked completed, email tracked in database
Failure → Job retried with exponential backoff (3 attempts)
Admin monitors → View queue stats, pause/resume, clean old jobs

Database Models

CampaignEmail Model

See CampaignEmail Model Documentation for full schema.

Key Fields:

Field	Type	Description
`id`	String (UUID)	Primary key
`campaignId`	String	Associated campaign
`recipientEmail`	String	Email recipient
`recipientName`	String?	Recipient name
`senderEmail`	String	Sender email address
`senderName`	String	Sender name
`subject`	String	Email subject line
`body`	String (Text)	Email body content
`status`	Enum	QUEUED, SENT, FAILED
`jobId`	String?	BullMQ job ID
`sentAt`	DateTime?	When email was sent
`failureReason`	String?	Error message if failed

Indexes:

campaignId, status — For campaign email stats
jobId — For job status lookups
sentAt — For time-based queries

Related Models:

Campaign — Campaign association
Representative — Email recipients

API Endpoints

Admin Endpoints

See Email Queue Module API Reference for full details.

Method	Endpoint	Auth	Description
GET	`/api/email-queue/stats`	SUPER_ADMIN, INFLUENCE_ADMIN	Get queue statistics
POST	`/api/email-queue/pause`	SUPER_ADMIN, INFLUENCE_ADMIN	Pause queue processing
POST	`/api/email-queue/resume`	SUPER_ADMIN, INFLUENCE_ADMIN	Resume queue processing
POST	`/api/email-queue/clean`	SUPER_ADMIN	Clean completed/failed jobs
POST	`/api/email-queue/retry/:jobId`	SUPER_ADMIN, INFLUENCE_ADMIN	Retry failed job

Public Endpoints

Email queue jobs are created via campaign email endpoints (no direct public access).

Configuration

Environment Variables

Variable	Type	Default	Description
`REDIS_HOST`	string	localhost	Redis hostname
`REDIS_PORT`	number	6379	Redis port
`REDIS_PASSWORD`	string	-	Redis password (required)
`SMTP_HOST`	string	-	SMTP server hostname
`SMTP_PORT`	number	587	SMTP server port
`SMTP_USER`	string	-	SMTP username
`SMTP_PASS`	string	-	SMTP password
`SMTP_FROM_EMAIL`	string	-	Default sender email
`SMTP_FROM_NAME`	string	-	Default sender name
`EMAIL_TEST_MODE`	boolean	false	Send to MailHog instead of SMTP
`EMAIL_QUEUE_CONCURRENCY`	number	5	Max concurrent email workers

BullMQ Configuration

// api/src/services/email-queue.service.ts

const queueOptions = {
  connection: {
    host: process.env.REDIS_HOST,
    port: parseInt(process.env.REDIS_PORT || '6379'),
    password: process.env.REDIS_PASSWORD
  },
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 5000 // 5s, 25s, 125s
    },
    removeOnComplete: {
      age: 86400, // Keep completed jobs for 24h
      count: 1000
    },
    removeOnFail: {
      age: 604800 // Keep failed jobs for 7 days
    }
  }
};

Worker Configuration

const workerOptions = {
  connection: queueOptions.connection,
  concurrency: parseInt(process.env.EMAIL_QUEUE_CONCURRENCY || '5'),
  limiter: {
    max: 60, // Max 60 emails
    duration: 60000 // per minute
  }
};

Admin Workflow

1. View Queue Statistics

[Screenshot: EmailQueuePage with queue stats cards]

Steps:

Navigate to Influence > Email Queue
View queue statistics:
- Waiting: Jobs queued for processing
- Active: Jobs currently being processed
- Completed: Successfully sent emails
- Failed: Failed emails requiring attention
Monitor queue health (green if waiting < 100)

Code Example (EmailQueuePage.tsx):

const [stats, setStats] = useState({
  waiting: 0,
  active: 0,
  completed: 0,
  failed: 0,
  paused: false
});

useEffect(() => {
  const fetchStats = async () => {
    const { data } = await api.get('/email-queue/stats');
    setStats(data);
  };

  fetchStats();

  // Refresh every 5 seconds
  const interval = setInterval(fetchStats, 5000);
  return () => clearInterval(interval);
}, []);

return (
  <Row gutter={16}>
    <Col span={6}>
      <Card>
        <Statistic
          title="Waiting"
          value={stats.waiting}
          valueStyle={{ color: stats.waiting > 100 ? '#cf1322' : '#3f8600' }}
        />
      </Card>
    </Col>
    <Col span={6}>
      <Card>
        <Statistic title="Active" value={stats.active} />
      </Card>
    </Col>
    <Col span={6}>
      <Card>
        <Statistic title="Completed" value={stats.completed} />
      </Card>
    </Col>
    <Col span={6}>
      <Card>
        <Statistic
          title="Failed"
          value={stats.failed}
          valueStyle={{ color: stats.failed > 0 ? '#cf1322' : undefined }}
        />
      </Card>
    </Col>
  </Row>
);

2. Pause/Resume Queue

[Screenshot: EmailQueuePage with pause/resume buttons]

Steps:

Click Pause Queue button
Queue stops processing new jobs
Active jobs complete normally
Status indicator shows "Paused"
Click Resume Queue to restart processing

Use Cases:

Temporary SMTP server maintenance
Stop email sending during testing
Prevent email sending during off-hours

Code Example (email-queue.service.ts):

async pauseQueue(): Promise<void> {
  await this.queue.pause();
  logger.info('Email queue paused');
}

async resumeQueue(): Promise<void> {
  await this.queue.resume();
  logger.info('Email queue resumed');
}

async isPaused(): Promise<boolean> {
  return this.queue.isPaused();
}

3. Clean Completed Jobs

[Screenshot: EmailQueuePage with clean jobs button]

Steps:

Click Clean Jobs dropdown
Select cleanup type:
- Completed (>24h): Remove old successful jobs
- Failed (>7d): Remove old failed jobs
- All Completed: Remove all successful jobs
Confirm cleanup
Jobs removed from queue, stats updated

Code Example (email-queue.routes.ts):

router.post('/clean', requireRole('SUPER_ADMIN', 'INFLUENCE_ADMIN'), async (req, res) => {
  try {
    const { type } = req.body; // 'completed', 'failed', 'all-completed'

    let count = 0;

    if (type === 'completed') {
      count = await queue.clean(86400000, 1000, 'completed'); // 24h
    } else if (type === 'failed') {
      count = await queue.clean(604800000, 1000, 'failed'); // 7d
    } else if (type === 'all-completed') {
      count = await queue.clean(0, 0, 'completed'); // All
    }

    logger.info(`Cleaned ${count} ${type} jobs`);

    res.json({ count });
  } catch (error) {
    logger.error('Failed to clean jobs:', error);
    res.status(500).json({ error: 'Failed to clean jobs' });
  }
});

4. Retry Failed Jobs

[Screenshot: Failed jobs table with retry buttons]

Steps:

Scroll to Failed Jobs section
View failed job details (error message, recipient)
Click Retry button on specific job
Job re-queued for processing
Monitor in Active tab

Bulk Retry:

Select multiple failed jobs (checkboxes)
Click Retry Selected button
All selected jobs re-queued

Code Example (email-queue.service.ts):

async retryFailedJob(jobId: string): Promise<void> {
  const job = await this.queue.getJob(jobId);

  if (!job) {
    throw new Error('Job not found');
  }

  if (await job.isFailed()) {
    await job.retry();
    logger.info(`Retrying job ${jobId}`);
  } else {
    throw new Error('Job is not failed');
  }
}

async retryAllFailed(): Promise<number> {
  const failed = await this.queue.getFailed();
  let count = 0;

  for (const job of failed) {
    await job.retry();
    count++;
  }

  logger.info(`Retried ${count} failed jobs`);
  return count;
}

Public Workflow

1. Send Campaign Email

[Screenshot: CampaignPage with email sending form]

User Journey:

User selects representatives to email
Fills in sender details (name, email)
Reviews/edits email content (if allowed)
Clicks Send Email button
System creates email jobs (one per recipient)
Jobs added to BullMQ queue
User sees confirmation message

Code Example (campaigns-public.routes.ts):

router.post('/send-email', async (req, res) => {
  try {
    const {
      campaignId,
      senderName,
      senderEmail,
      postalCode,
      representativeIds,
      customMessage
    } = req.body;

    const campaign = await prisma.campaign.findUnique({
      where: { id: campaignId }
    });

    if (!campaign || campaign.status !== 'ACTIVE') {
      return res.status(400).json({ error: 'Campaign not active' });
    }

    const representatives = await prisma.representative.findMany({
      where: { id: { in: representativeIds } }
    });

    // Create email jobs
    const emailJobs = [];

    for (const rep of representatives) {
      const emailData = {
        campaignId,
        recipientEmail: rep.email,
        recipientName: rep.name,
        senderEmail,
        senderName,
        subject: processTemplate(campaign.emailSubjectTemplate, {
          senderName,
          recipientName: rep.name,
          postalCode
        }),
        body: customMessage || processTemplate(campaign.emailBodyTemplate, {
          senderName,
          senderEmail,
          recipientName: rep.name,
          recipientEmail: rep.email,
          postalCode
        })
      };

      // Add to queue
      const job = await emailQueueService.addEmail(emailData);

      emailJobs.push(job);
    }

    res.json({
      success: true,
      emailsQueued: emailJobs.length
    });
  } catch (error) {
    logger.error('Failed to queue campaign emails:', error);
    res.status(500).json({ error: 'Failed to send emails' });
  }
});

2. Job Processing

Worker Processing Logic:

// api/src/services/email-queue.service.ts

import { Worker } from 'bullmq';
import { emailService } from './email.service';

const worker = new Worker('campaign-emails', async (job) => {
  const {
    campaignId,
    recipientEmail,
    recipientName,
    senderEmail,
    senderName,
    subject,
    body
  } = job.data;

  try {
    // Send email via nodemailer
    await emailService.send({
      to: recipientEmail,
      from: {
        email: process.env.SMTP_FROM_EMAIL!,
        name: process.env.SMTP_FROM_NAME!
      },
      replyTo: {
        email: senderEmail,
        name: senderName
      },
      subject,
      html: body
    });

    // Update database record
    await prisma.campaignEmail.update({
      where: { jobId: job.id },
      data: {
        status: 'SENT',
        sentAt: new Date()
      }
    });

    logger.info(`Sent campaign email ${job.id} to ${recipientEmail}`);

    // Update Prometheus metric
    metrics.campaignEmailsSent.inc({ campaign_id: campaignId });

    return { success: true };
  } catch (error) {
    logger.error(`Failed to send email ${job.id}:`, error);

    // Update database record
    await prisma.campaignEmail.update({
      where: { jobId: job.id },
      data: {
        status: 'FAILED',
        failureReason: error.message
      }
    });

    throw error; // Let BullMQ handle retry
  }
}, workerOptions);

worker.on('completed', (job) => {
  logger.info(`Job ${job.id} completed`);
});

worker.on('failed', (job, err) => {
  logger.error(`Job ${job?.id} failed:`, err);
});

Volunteer Workflow

Not applicable — email queue is system-level.

Code Examples

Backend: Email Queue Service

// api/src/services/email-queue.service.ts

import { Queue, QueueEvents } from 'bullmq';
import { logger } from '../utils/logger';
import { prisma } from '../config/database';

export class EmailQueueService {
  private queue: Queue;
  private queueEvents: QueueEvents;

  constructor() {
    const connection = {
      host: process.env.REDIS_HOST!,
      port: parseInt(process.env.REDIS_PORT || '6379'),
      password: process.env.REDIS_PASSWORD
    };

    this.queue = new Queue('campaign-emails', {
      connection,
      defaultJobOptions: {
        attempts: 3,
        backoff: {
          type: 'exponential',
          delay: 5000
        },
        removeOnComplete: {
          age: 86400,
          count: 1000
        },
        removeOnFail: {
          age: 604800
        }
      }
    });

    this.queueEvents = new QueueEvents('campaign-emails', { connection });

    this.setupEventHandlers();
  }

  private setupEventHandlers(): void {
    this.queueEvents.on('completed', ({ jobId }) => {
      logger.info(`Email job ${jobId} completed`);
    });

    this.queueEvents.on('failed', ({ jobId, failedReason }) => {
      logger.error(`Email job ${jobId} failed: ${failedReason}`);
    });
  }

  async addEmail(data: any): Promise<{ jobId: string }> {
    // Create database record
    const emailRecord = await prisma.campaignEmail.create({
      data: {
        ...data,
        status: 'QUEUED'
      }
    });

    // Add job to queue
    const job = await this.queue.add('send-email', data, {
      jobId: emailRecord.id
    });

    // Update database with job ID
    await prisma.campaignEmail.update({
      where: { id: emailRecord.id },
      data: { jobId: job.id }
    });

    logger.info(`Queued email job ${job.id}`);

    return { jobId: job.id! };
  }

  async getStats(): Promise<any> {
    const counts = await this.queue.getJobCounts();

    return {
      waiting: counts.waiting || 0,
      active: counts.active || 0,
      completed: counts.completed || 0,
      failed: counts.failed || 0,
      paused: await this.queue.isPaused()
    };
  }

  async pauseQueue(): Promise<void> {
    await this.queue.pause();
  }

  async resumeQueue(): Promise<void> {
    await this.queue.resume();
  }

  async clean(grace: number, limit: number, type: string): Promise<number> {
    return this.queue.clean(grace, limit, type as any);
  }
}

export const emailQueueService = new EmailQueueService();

Frontend: Queue Stats Dashboard

// admin/src/pages/EmailQueuePage.tsx

import React, { useState, useEffect } from 'react';
import { Card, Row, Col, Statistic, Button, Space, message } from 'antd';
import { PlayCircleOutlined, PauseCircleOutlined, ClearOutlined } from '@ant-design/icons';
import { api } from '../../lib/api';

const EmailQueuePage: React.FC = () => {
  const [stats, setStats] = useState<any>(null);
  const [loading, setLoading] = useState(false);

  const fetchStats = async () => {
    const { data } = await api.get('/email-queue/stats');
    setStats(data);
  };

  useEffect(() => {
    fetchStats();
    const interval = setInterval(fetchStats, 5000);
    return () => clearInterval(interval);
  }, []);

  const handlePause = async () => {
    setLoading(true);
    try {
      await api.post('/email-queue/pause');
      message.success('Queue paused');
      fetchStats();
    } catch (error) {
      message.error('Failed to pause queue');
    } finally {
      setLoading(false);
    }
  };

  const handleResume = async () => {
    setLoading(true);
    try {
      await api.post('/email-queue/resume');
      message.success('Queue resumed');
      fetchStats();
    } catch (error) {
      message.error('Failed to resume queue');
    } finally {
      setLoading(false);
    }
  };

  const handleClean = async (type: string) => {
    setLoading(true);
    try {
      const { data } = await api.post('/email-queue/clean', { type });
      message.success(`Cleaned ${data.count} jobs`);
      fetchStats();
    } catch (error) {
      message.error('Failed to clean jobs');
    } finally {
      setLoading(false);
    }
  };

  if (!stats) return <Card loading />;

  return (
    <Space direction="vertical" size="large" style={{ width: '100%' }}>
      <Card title="Queue Statistics">
        <Row gutter={16}>
          <Col span={6}>
            <Statistic
              title="Waiting"
              value={stats.waiting}
              valueStyle={{ color: stats.waiting > 100 ? '#cf1322' : '#3f8600' }}
            />
          </Col>
          <Col span={6}>
            <Statistic title="Active" value={stats.active} />
          </Col>
          <Col span={6}>
            <Statistic title="Completed" value={stats.completed} />
          </Col>
          <Col span={6}>
            <Statistic
              title="Failed"
              value={stats.failed}
              valueStyle={{ color: stats.failed > 0 ? '#cf1322' : undefined }}
            />
          </Col>
        </Row>
      </Card>

      <Card title="Queue Controls">
        <Space>
          {stats.paused ? (
            <Button
              type="primary"
              icon={<PlayCircleOutlined />}
              onClick={handleResume}
              loading={loading}
            >
              Resume Queue
            </Button>
          ) : (
            <Button
              icon={<PauseCircleOutlined />}
              onClick={handlePause}
              loading={loading}
            >
              Pause Queue
            </Button>
          )}

          <Button
            icon={<ClearOutlined />}
            onClick={() => handleClean('completed')}
            loading={loading}
          >
            Clean Completed
          </Button>

          <Button
            danger
            icon={<ClearOutlined />}
            onClick={() => handleClean('failed')}
            loading={loading}
          >
            Clean Failed
          </Button>
        </Space>
      </Card>
    </Space>
  );
};

export default EmailQueuePage;

Troubleshooting

Emails Stuck in Queue

Symptoms:

Waiting count increases but active/completed don't
Jobs not processing

Solutions:

Check worker status → docker compose logs api | grep "Worker"
Verify Redis connection → docker compose exec redis redis-cli ping
Check SMTP configuration → test with /api/auth/test-email
Restart worker → docker compose restart api

Debugging:

# Check Redis keys
docker compose exec redis redis-cli --pass $REDIS_PASSWORD
> KEYS bull:campaign-emails:*

# Check worker logs
docker compose logs -f api | grep "Email worker"

# Check queue status
curl -H "Authorization: Bearer $TOKEN" http://localhost:4000/api/email-queue/stats

High Failure Rate

Symptoms:

Many jobs failing
Failed count increasing rapidly

Solutions:

Check SMTP credentials → verify username/password
Review failure reasons → check failureReason field in database
Check SMTP server status → verify server is reachable
Review rate limits → may be hitting SMTP server limits

Common Failure Reasons:

535 Authentication failed → Invalid SMTP credentials
550 Mailbox unavailable → Recipient email doesn't exist
421 Too many connections → Reduce concurrency
Connection timeout → SMTP server unreachable

Code Fix (email.service.ts):

// Add better error handling
async send(options: EmailOptions): Promise<void> {
  try {
    await this.transporter.sendMail(options);
  } catch (error) {
    if (error.responseCode === 535) {
      throw new Error('SMTP authentication failed - check credentials');
    } else if (error.responseCode === 550) {
      throw new Error('Recipient mailbox unavailable');
    } else if (error.code === 'ETIMEDOUT') {
      throw new Error('SMTP server connection timeout');
    } else {
      throw error;
    }
  }
}

Redis Connection Issues

Symptoms:

Error: "ECONNREFUSED" or "NOAUTH"
Queue operations fail

Solutions:

Verify Redis is running → docker compose ps redis
Check Redis password → ensure REDIS_PASSWORD matches docker-compose.yml
Check Redis port → default 6379
Verify Redis auth → docker compose exec redis redis-cli --pass $REDIS_PASSWORD ping

Fix Redis Auth:

# docker-compose.yml
services:
  redis:
    image: redis:7-alpine
    command: redis-server --requirepass ${REDIS_PASSWORD}
    ports:
      - "6379:6379"

Performance Considerations

Concurrency Tuning

Worker Concurrency:

// Adjust based on SMTP server limits
const workerOptions = {
  concurrency: 5, // Process 5 emails simultaneously
  limiter: {
    max: 60, // Max 60 emails per minute
    duration: 60000
  }
};

SMTP Server Limits:

Gmail: 100 emails/day (consumer), 2000/day (Workspace)
SendGrid: Varies by plan (40k/day free tier)
AWS SES: 14 emails/second, 200 emails/day (sandbox)

Queue Monitoring

Prometheus Metrics:

import { Counter, Gauge } from 'prom-client';

export const campaignEmailsQueued = new Counter({
  name: 'cm_campaign_emails_queued_total',
  help: 'Total campaign emails queued',
  labelNames: ['campaign_id']
});

export const campaignEmailsSent = new Counter({
  name: 'cm_campaign_emails_sent_total',
  help: 'Total campaign emails sent',
  labelNames: ['campaign_id']
});

export const emailQueueSize = new Gauge({
  name: 'cm_email_queue_size',
  help: 'Current email queue size',
  labelNames: ['status']
});

// Update gauge every 30 seconds
setInterval(async () => {
  const stats = await emailQueueService.getStats();

  emailQueueSize.set({ status: 'waiting' }, stats.waiting);
  emailQueueSize.set({ status: 'active' }, stats.active);
  emailQueueSize.set({ status: 'failed' }, stats.failed);
}, 30000);

Database Optimization

Index Strategy:

CREATE INDEX idx_campaign_email_status ON campaign_emails (status);
CREATE INDEX idx_campaign_email_campaign_id ON campaign_emails (campaign_id);
CREATE INDEX idx_campaign_email_sent_at ON campaign_emails (sent_at);

Query Optimization:

// Paginated campaign email stats
const emails = await prisma.campaignEmail.findMany({
  where: { campaignId },
  select: {
    id: true,
    recipientEmail: true,
    status: true,
    sentAt: true,
    failureReason: true
  },
  orderBy: { createdAt: 'desc' },
  take: 100,
  skip: page * 100
});

Backend Modules

Email Queue Module — Full API reference
Email Service — SMTP configuration
Campaigns Module — Campaign integration

Frontend Pages

EmailQueuePage — Admin queue monitoring
CampaignsPage — Campaign management

Database Models

CampaignEmail — Email tracking schema
Campaign — Campaign schema

Configuration

Environment Variables — SMTP/Redis configuration
BullMQ Documentation — Official BullMQ docs

Monitoring

Prometheus Metrics — Email queue metrics
Grafana Dashboards — Queue visualization

25 KiB Raw Blame History

Email Queue System

Overview

Architecture

Database Models

CampaignEmail Model

API Endpoints

Admin Endpoints

Public Endpoints

Configuration

Environment Variables

BullMQ Configuration

Worker Configuration

Admin Workflow

1. View Queue Statistics

2. Pause/Resume Queue

3. Clean Completed Jobs

4. Retry Failed Jobs

Public Workflow

1. Send Campaign Email

2. Job Processing

Volunteer Workflow

Code Examples

Backend: Email Queue Service

Frontend: Queue Stats Dashboard

Troubleshooting

Emails Stuck in Queue

High Failure Rate

Redis Connection Issues

Performance Considerations

Concurrency Tuning

Queue Monitoring

Database Optimization

Related Documentation

Backend Modules

Frontend Pages

Database Models

Configuration

Monitoring

25 KiB

Raw Blame History