Event Bus

ATHELOS uses an event bus for background work. Server actions emit events; a worker process picks them up and dispatches them to handlers. This keeps the web request fast and lets long-running work happen out of band.

Why an event bus

Some operations shouldn't block a web request:

• Sending an email takes 500ms to 2 seconds
• Generating a report PDF takes 3 to 10 seconds
• Embedding an article for semantic search takes 200 to 500ms
• Notifying a webhook endpoint has unpredictable latency

If these run inside the request, the user waits. If they run in the background, the user gets an immediate response and the work happens behind the scenes.

The event bus is how ATHELOS separates "respond to the user" from "do the work."

The architecture

Server action (web)
  -> emitEvent(type, payload, { jobId })
  -> BullMQ Queue.add("event.dispatch", envelope)
  -> Redis
  -> Worker picks up the job
  -> dispatchEvent(envelope)
  -> handlers[type](payload)
  -> audit log write + optional side effect

Every step is asynchronous except the initial emitEvent call, which returns immediately.

Emitting an event

From any server action:

import { emitEvent } from "@/lib/events/emit";

await emitEvent("finding.critical", {
  findingId: finding.id,
  engagementId: finding.engagementId,
  severity: finding.severity,
});

The call returns as soon as the event is queued. It doesn't wait for handlers to run.

Event types

Events use dot-namespaced names:

This list grows as features are added.

The payload

Payloads carry IDs and minimal metadata, not full records:

{
  findingId: "uuid",
  engagementId: "uuid",
  severity: "critical",
}

The handler re-fetches the full record from the database if it needs it. This keeps payloads small and avoids stale data.

Job IDs

Each event has a unique job ID. BullMQ uses these to deduplicate — if the same event fires twice with the same ID, only one job is created.

**Job IDs cannot contain :. ** BullMQ uses colons internally for Redis key separators. ATHELOS sanitizes job IDs before passing them to BullMQ:

function sanitizeJobId(raw: string): string {
  return raw.replace(/[^a-zA-Z0-9._-]/g, "-");
}

So finding.critical:abc123 becomes finding.critical-abc123.

The worker

The worker is a separate Node.js process. It listens on the BullMQ queue and dispatches events as they arrive.

Start it:

pnpm dev:worker

In production, it runs as a systemd service or a separate container.

The worker is stateless — you can run multiple workers for throughput. BullMQ ensures each job is processed exactly once.

Handlers

Handlers are a typed map. Each event type has one handler:

export const handlers = {
  "finding.critical": async (payload) => {
    // Send notification, add to exec summary, etc.
  },
  "report.generated": async (payload) => {
    // Log, optionally trigger PDF pre-render
  },
  // ...
};

When an event fires, the worker looks up the handler by type and calls it. If no handler exists, the event is logged and skipped.

Audit logging

Every event writes one row to audit_log, regardless of whether a handler exists:

• Who triggered the event (user ID)
• What event type
• The payload
• When it happened

This gives a complete timeline of everything that happened in the system.

Retries

BullMQ retries failed jobs automatically:

• 3 attempts by default
• Exponential backoff — 1s, 4s, 16s
• After all retries fail, the job moves to the dead letter queue

Failed jobs are visible in the Redis queue. You can inspect them and manually retry if needed.

Dead letter queue

Jobs that exhaust retries go to a "failed" set in BullMQ. These need investigation — a handler that consistently fails indicates a bug.

List failed jobs:

docker exec soc_redis redis-cli -a <password> --scan --pattern "bull:event.dispatch:failed"

Delayed events

Some events should fire in the future. For example: "remind me about this invoice in 30 days if unpaid."

BullMQ supports delayed jobs:

await emitEvent("invoice.reminder", payload, {
  delayMs: 30 * 24 * 60 * 60 * 1000,
});

The job sits in Redis until the delay passes, then fires.

Scheduled events

Recurring events use BullMQ's repeatable jobs. For example, "check for expiring certs every day at 8am":

await emitEvent("certs.check", {}, {
  repeat: { pattern: "0 8 * * *" },
});

BullMQ schedules these automatically.

Monitoring

The worker logs every dispatched event:

[ATHELOS] [worker] dispatched finding.critical (job 42)

Failed jobs log the error and stack trace.

For a visual dashboard, run Bull Board alongside the worker — a web UI for BullMQ queues showing pending, active, completed, and failed jobs.

Testing

To emit a test event:

node -e "
const { emitEvent } = require('./apps/worker/dist/events/emit');
emitEvent('test.event', { hello: 'world' });
"

The worker should log the dispatch. Use this to verify the pipeline is wired correctly.

When events fail

If the worker isn't running, events queue up in Redis but never dispatch. Symptom: notifications don't fire, PDFs don't generate, but nothing errors in the web app.

Check the worker is running:

ps aux | grep worker

Or in Docker:

docker ps | grep worker

If it's down, start it. Queued events will be processed.

Where to go next

• Notifications — what events trigger
• Scheduled backups — the backup event
• Reference: environment variables — worker config