Queues & Messaging

Some work shouldn't block the request that triggered it. When a parent submits an admission, the API doesn't need to wait for the confirmation email to send before it answers — that would make the request slow and fragile, hostage to whether an email provider happens to be up. Instead the API hands the email off to a background job and replies immediately. Something else picks the job up and does the slow part later.

That hand-off is what this page is about. The backend runs two separate messaging systems, and the single most useful thing you can learn here is which one to reach for:

• BullMQ — durable, retried job queues backed by Redis. This is fire-and-forget: the API drops a job on a queue and walks away. A separate app picks it up. Email, audit-log writes, and Microsoft 365 sync all flow through here.
• RabbitMQ microservices — request/response between services over AMQP. The caller sends a message and waits for an answer. Today this powers liveness checks — the server asks each background app "are you alive?" and reads the reply.

They're easy to confuse because the same background apps use both. Keep the question simple: do I need an answer back? If no, it's a BullMQ job. If yes, it's a RabbitMQ message.

The apps

Most jobs aren't processed by the main API server — they're processed by two dedicated background apps that exist for exactly this. The server is the producer; these are the consumers:

• apps/worker — sends email, and will host most general background processors.
• apps/audit — writes audit-log entries to the database.

Each one boots as a microservice listening on its own durable RabbitMQ queue. "Durable" means the queue survives a broker restart — messages aren't lost if RabbitMQ blips. Here's the worker's bootstrap:

app.connectMicroservice<MicroserviceOptions>({
  transport: Transport.RMQ,
  options: {
    urls: [configService.getOrThrow<string>('rabbitmq.url')],
    queue: 'naalya-worker',
    queueOptions: { durable: true },
  },
});

await app.startAllMicroservices();

The audit app is identical, just listening on naalya-audit. Note what this queue is for: it's the RabbitMQ request/response channel, not a BullMQ job queue. They both happen to involve "queues", which is exactly why people mix them up — these are two different systems that share a name.

RabbitMQ

Reach for RabbitMQ microservice transport when the server needs a reply from another service. The clearest example is the health check. The server's /health endpoint needs to know whether the worker and audit apps are actually alive, so it sends each one a ping and waits for an uptime/memory report to come back.

The server talks to the background apps as clients, registered in the Health module. Each client is wired to one of those durable queues, and addressed by a token from @app/shared (WORKER_SERVICE, AUDIT_SERVICE):

ClientsModule.registerAsync([
  {
    name: WORKER_SERVICE,
    inject: [ConfigService],
    useFactory: (config: ConfigService<EnvTypes>) => ({
      transport: Transport.RMQ,
      options: {
        urls: [config.getOrThrow('rabbitmq.url', { infer: true })],
        queue: 'naalya-worker',
        queueOptions: { durable: true },
      },
    }),
  },
  // ...and the same shape again for AUDIT_SERVICE → 'naalya-audit'
]);

To actually make the round trip, the service uses client.send(pattern, payload). The message pattern is just a string both sides agree on — here HEALTH_PING_PATTERN, which is 'health.ping'. send returns an RxJS Observable, so firstValueFrom turns the reply into a promise, and a timeout guards against a service that never answers:

const result = await firstValueFrom(
  client.send(HEALTH_PING_PATTERN, {}).pipe(
    timeout(WORKER_PING_TIMEOUT_MS),     // give up after 5s
    catchError(() => of(null)),          // a down service → null, not a crash
  ),
);

On the other end, the worker replies to that same pattern with @MessagePattern. Whatever the handler returns becomes the response the server awaits:

@MessagePattern(HEALTH_PING_PATTERN)
ping() {
  const mem = process.memoryUsage();
  return { uptime: process.uptime(), pid: process.pid, memory: { /* ... */ } };
}

That send here / @MessagePattern there is the whole request/response shape. The audit app responds to the exact same 'health.ping' pattern — the message pattern is the address, the queue routes it.

BullMQ

This is the workhorse, and the one you'll write most often. BullMQ stores jobs in Redis, hands them to a processor when one is free, and — crucially — retries a job if it throws. That retry safety net is the whole reason email and audit writes go through a queue instead of running inline.

It's configured once, globally, in the server's root module. The important line is the default retry policy:

BullModule.forRootAsync({
  inject: [ConfigService],
  useFactory: (config: ConfigService<EnvTypes>) => ({
    connection: { /* parsed from REDIS_URL */ },
    defaultJobOptions: { attempts: 3 },   // every job retries up to 3×
  }),
}),

Because attempts: 3 is the default, every job you enqueue gets three tries for free before BullMQ marks it failed — no per-job config needed. When something does fail, you can see it in the Bull Board dashboard mounted at /queues, which lists every queue with live counts of waiting, active, completed, and failed jobs.

The queues

Eight queues carry the background work. The pattern is always the same: a producer service on the server adds jobs, a processor consumes them.

Health watches the same pipes

GET /health (public) returns a status-only rollup; GET /health/detailed (platform admins) adds process metrics, transport latency, and per-queue counts. The rollups are queue-aware: worker is UP only if its transport and the email + kb-embedding queues are healthy, audit needs the audit-log queue, server needs microsoft-sync + social-agent — and any queue with ≥ 100 waiting jobs reports DOWN. If a queue backs up, the health page is where it shows first.

The email pipeline

Email is the cleanest worked example, so let's follow one job from the API all the way to "sent". There are two halves: a producer on the server and a processor in the worker.

Step 1: The producer enqueues a job

The producing module registers the queue, and a service injects it with @InjectQueue. To enqueue, you call queue.add(jobName, payload). Here's the entire EmailService:

@Injectable()
export class EmailService {
  private readonly logger = new Logger(EmailService.name);

  constructor(@InjectQueue(EMAIL_QUEUE) private readonly emailQueue: Queue) {}

  send(payload: EmailJobPayload): void {
    this.emailQueue.add(payload.type, payload).catch((err) => {
      this.logger.error('Failed to enqueue email job', err);
    });
  }
}

Two conventions in those few lines are worth copying into every producer you write:

• The job name is a meaningful discriminator. Email passes payload.type (the EmailType enum value) as the name, so each job shows up in Bull Board labelled by what it is. The audit queue uses the literal 'write'.
• Enqueuing never throws into the caller. send returns void and swallows-then-logs any error from .add(). A parent submitting an application should not get a 500 just because Redis hiccupped — the API request and the email's fate are decoupled on purpose.

The queue has to be registered for @InjectQueue to find it, and registered again with Bull Board so it appears on the dashboard. Both happen in the module:

@Global()
@Module({
  imports: [
    BullModule.registerQueue({ name: EMAIL_QUEUE }),
    BullBoardModule.forFeature({ name: EMAIL_QUEUE, adapter: BullMQAdapter }),
  ],
  providers: [EmailService],
  exports: [EmailService],
})
export class EmailModule {}

Step 2: The payload is a typed, discriminated union

What actually rides on the queue is an EmailJobPayload — a discriminated union keyed on type. "Discriminated" means TypeScript uses the type field to figure out which variant you have, so each email kind carries its own shape of data and nothing else. Ask for data.otp on a welcome email and the compiler stops you.

export enum EmailType {
  WELCOME, PASSWORD_RESET, APPLICATION_STATUS, ADMISSION_RECEIVED,
  INQUIRY_CONFIRMATION, STAFF_LOGIN_OTP, SOCIAL_AGENT_ESCALATION,
}

// the OTP variant — each type narrows `data` to exactly its fields:
interface StaffLoginOtpEmailPayload extends BaseEmailPayload {
  type: EmailType.STAFF_LOGIN_OTP;
  data: { firstName: string; otp: string; expiresInMinutes: number };
}

This type lives in @app/shared, so the server (producer) and the worker (processor) compile against the same definition — the contract is enforced by the type system, not by hope.

Step 3: The processor consumes the job

Over in the worker, a processor picks the job up. A processor is a class decorated with @Processor(QUEUE) that extends WorkerHost and implements process(job). BullMQ calls process for each job; if it throws, BullMQ retries (up to those three attempts).

@Processor(EMAIL_QUEUE)
export class EmailProcessor extends WorkerHost {
  constructor(private readonly resendService: ResendService) { super(); }

  async process(job: Job<EmailJobPayload>): Promise<void> {
    const payload = job.data;
    const template = this.getTemplate(payload);   // picks a React Email template by type
    await this.resendService.sendEmail({
      to: payload.to, subject: payload.subject, react: template,
    });
  }
}

getTemplate() switches on payload.type to choose a React Email template, and the actual send goes through ResendService, which wraps the Resend API. The audit processor follows the exact same shape on the audit-log queue, except its process writes a row to the audit_log table instead of sending mail.

Which one?

When you sit down to add async work, this is the decision in one table:

Most of what you'll write is the first row. The mechanics never change: define the contract in @app/shared, enqueue from a producer service on the server, consume in a @Processor in the worker or audit app.

Where to go next