Event system

Warmbly's services communicate asynchronously through a pluggable event bus. The default self-host build uses NATS JetStream with JSON encoding; Kafka with Avro and Confluent Schema Registry is an opt-in build for cloud deployments. The realtime dashboard fanout is a separate, simpler pub/sub channel described at the end of this page.

Transport and codec

The bus is an abstraction (internal/infrastructure/eventbus/) with two providers, selected by EVENTBUS_PROVIDER:

• nats: NATS JetStream. One durable stream (default name warmbly) holds every topic as a subject under a prefix: topic jobs.worker-events becomes subject warmbly.jobs.worker-events. Messages are retained for 7 days, consumer groups map to JetStream durable consumers, and failed handlers are redelivered up to 10 times. This is what docker-compose.yml runs.
• kafka: Apache Kafka behind the kafka Go build tag. The default binaries do not include it (no librdkafka, no CGO), so EVENTBUS_PROVIDER=kafka on a default build fails at boot with a clear error. Every service that reads the bus publishes a second image for it, the same tag with a -kafka suffix: ghcr.io/warmbly/warmbly/backend:prod-kafka, and the same for consumer, worker, and tracking. Building your own needs GO_TAGS=kafka (Go) or CARGO_FEATURES=kafka (tracking).

Message encoding is orthogonal to transport, selected by CODEC_PROVIDER:

• json: plain JSON, no external dependencies, and the default for the self-host stack.
• avro: Avro with Confluent Schema Registry (SCHEMA_REGISTRY_URL plus optional key/secret), also behind the kafka build tag. The subject is the topic name plus -value, Confluent's default naming strategy, so warmup-events registers under warmup-events-value.

The Rust tracking service follows both switches. It speaks Kafka only when compiled with its kafka cargo feature, which is what the tracking:*-kafka image is, and it encodes with CODEC_PROVIDER on either transport. So a Kafka deployment on CODEC_PROVIDER=json needs no Schema Registry at all, and avro there is refused at boot without SCHEMA_REGISTRY_URL rather than failing at the first published event.

Topics

These are the real topic names. On NATS each is prefixed with the subject prefix (default warmbly.).

Each worker subscribes only to its own w.<worker-uuid> topic; the UUID is the worker's identity (derived from its public IP at install time). The consumer service reads jobs.worker-events with the consumer group consumer-group and tracking-events with the group tracking-consumer.

Worker commands

The backend drives workers with a {type, body} envelope (internal/models/event.go):

{
  "type": "SEND_EMAIL",
  "body": { ... }
}

Types: SEND_EMAIL, ADD_EMAIL, REMOVE_EMAIL, EMAIL_VALIDATION, WARMUP_ACTION, MESSAGE_SEEN, MAILBOX_IDENTITY. Each has a typed body struct in internal/models/ (for example models.SendEmail for SEND_EMAIL).

MESSAGE_SEEN carries a read or unread change made in the unified inbox out to the mailbox provider, one event per mailbox and at most models.SeenRelayChunk messages each. It answers with nothing: the store was written before the event was published, so the provider's copy is the only thing it changes, and a failure is logged rather than retried because the next sync reports whatever the provider actually holds. Each message carries all three providers' handles and the worker takes the ones its client uses: a provider message id for Gmail and Graph, an IMAP folder plus UID, and the immutable RFC Message-ID, which Graph re-resolves the live id from because a Graph id changes whenever a message moves. The state relayed is the one the row holds when the relay reads it back, not the one the request asked for, so a conflicting toggle leaves the provider agreeing with the store rather than with whoever published last.

MAILBOX_IDENTITY asks the worker holding a mailbox to read its send-as addresses, and one signature, from the provider. It answers on a Redis channel named by the request's process_id rather than on the worker-events topic, the same round trip EMAIL_VALIDATION uses, because the caller is an HTTP request waiting for it. The control plane makes this call itself in exactly one place, the OAuth handshake, where the token came from the consent the customer just completed and the mailbox has no worker yet.

Worker results

Workers publish the same envelope shape back on jobs.worker-events:

{
  "type": "NEW_EMAIL",
  "body": { ... }
}

Types: NEW_EMAIL, INBOUND_BOUNCE, REMOVE_EMAIL, FLAGS_ADD, FLAGS_REMOVE, UPDATE_EMAIL, UPDATE_MAILBOX, DELETE_MAILBOX, TOKEN_UPDATE, HISTORY_ID_UPDATE, GRAPH_DELTA_UPDATE, SYNC_STATE, EMAIL_SENT, EMAIL_FAILED, EMAIL_AUTH_ERROR, EMAIL_DISABLED, EMAIL_RATE_LIMITED, EMAIL_SERVER_ERROR, WORKER_HEALTH.

The consumer registers one handler per type (internal/app/consumer/events.go); unregistered types are logged and acknowledged rather than redelivered.

Every SEND_EMAIL is answered with exactly one per-task result: EMAIL_SENT (the provider accepted the message; the consumer records the wire Message-ID on the task) or EMAIL_FAILED (a SendEmailResult carrying the error code and message). The control plane stamps a campaign step sent when it hands the send to the worker, so EMAIL_FAILED is what walks that back: the consumer marks the task failed, clears the step's sent_at and counts the attempt on campaign_contact_progress, gives the send back to the campaign's daily counters, writes the failure to the campaign activity log, and reopens a campaign that completed while the send was in flight. After CampaignSendMaxAttempts (5) the lead is marked failed and routing drops it; a RECIPIENT_REJECTED result (refused at RCPT) skips the retries and is ingested as a bounce instead. Account-level conditions (EMAIL_AUTH_ERROR, EMAIL_DISABLED, EMAIL_RATE_LIMITED, EMAIL_SERVER_ERROR) are raised in addition to, never instead of, the per-task result; they carry an EmailErrorEvent and act on the mailbox. A worker that does not hold the mailbox yet leaves the send for a few redeliveries before reporting it failed, and the backend never publishes a send to a worker that is not heartbeating.

SYNC_STATE is the worker's relay of a mailbox's sync state (backfill progress and cursor, fair-use throttle, last-synced time). It carries the full state rather than a delta, so a lost event is repaired by the next one; the consumer writes it to email_sync_state and the backend hands it back inside ADD_EMAIL on the next load, which is what lets a replaced worker resume an import instead of restarting it.

Tracking events

Produced by the Rust tracking service when a pixel loads or a tracked link is clicked, consumed into campaign stats and deliverability data:

{
  "event_type": "EMAIL_OPENED",
  "task_id": "uuid",
  "original_url": "https://example.com/page",
  "link_id": "uuid",
  "timestamp": "2026-01-29T12:00:00Z",
  "user_agent": "Mozilla/5.0...",
  "ip_hash": "sha256...",
  "client_ip": "203.0.113.0",
  "scanner": "proofpoint",
  "scanner_probable": true
}

event_type is EMAIL_OPENED or EMAIL_CLICKED; original_url and link_id (the click ticket, which names the link's stored destination and anchor text) are set only for clicks; IPs are stored as hashes, never raw, and client_ip carries only the source network. scanner names the mail-filtering network the edge recognised, and scanner_probable says that network can also carry a person, which is what browser isolation makes true of the mail-security vendors. Both are absent from events written before those fields existed, and an absent scanner_probable reads as false, the conservative side. The struct is events.TrackingEvent in internal/events/schemas.go, mirrored in tracking/src/events.rs.

The consumer classifies each event before it counts. An open is automated when the user agent is a mail privacy proxy or missing, or when it arrives within tracking.machine_window_open_seconds of the step's dispatch; it is recorded with opened_machine and upgraded by a later human open. A click is automated for a missing user agent, for arriving inside tracking.machine_window_click_seconds of the same dispatch (30 by default, set independently of the open window and accepting the same 1 to 900 seconds; it ships shorter because a misjudged click costs an automation rather than a metric), or when the same source clicked another link of the same email within TrackingClickBurstSeconds; every click is logged per link in email_link_clicks with its reason, and only a human click stamps clicked_at, fires instant actions, or emits a webhook. A burst is only recognisable from its second click, so a human click's stamp and log row are written at once but its effects (evidence, instant actions, webhook, realtime event) run after the burst window plus a second, on the classification the click has by then; a burst recognised meanwhile relabels the earlier click and withdraws the stamp when no human click remains. A consumer restart inside the window loses only those deferred effects, and routing still follows the clicked branch at the next step boundary.

scanner overrides all of that: a request from a recognised mail-filtering network is automated whatever its user agent claims and however late it arrives, because a security gateway walks a message with an ordinary browser. scanner_probable qualifies it. A probable source is measured against tracking.machine_window_probable_seconds instead of the window for its event kind (whichever is wider), and outside that window it is judged exactly as an unrecognised source would be. That is what lets the vendor ASNs ship enabled: the delivery-time scan is classified automated, and the recipient who clicks through browser isolation an hour later still counts. A recipient who genuinely clicks inside the window is classified automated too, which is the cost of the widening.

The label carries one effect the consumer does not decide. The tracking service withholds the click's identification ticket from any recognised network, probable included, because the edge has no dispatch time and so cannot separate the delivery-time scan from the isolated human click. A probable click the consumer later counts as a person's is therefore still a click with no website-visit attribution behind it.

Website page views do not ride this topic. The tracking service forwards each accepted view to the backend's internal API (POST /api/v1/internal/page-hits) instead, because the backend is where the user agent and IP are turned into device and location, and the IP must not sit in a durable stream on the way there.

Analytics events

The backend records sends on two write-only streams for downstream analytics: email-events carries EMAIL_SENT records (task, account, campaign, contact, message ID, recipient), and warmup-events carries WARMUP_EMAIL_SENT records (sender and target accounts, whether it was a reply). Nothing in the repo consumes them today; they exist so an external pipeline can tap the stream.

Realtime fanout

Dashboard realtime does not ride the event bus. Backend and consumer publish JSON events on a dedicated pub/sub transport (internal/infrastructure/pubsub/), and the Elixir realtime service subscribes and pushes to WebSocket clients:

• Default: Redis pub/sub on the channel realtime:events. No configuration beyond a shared Redis.
• PUBSUB_ENABLED=true: Google Cloud Pub/Sub instead (requires GCP_PROJECT_ID); the backend creates a fixed set of topics and matching -sub subscriptions on boot.

Events embed a BaseEvent (event_type, user_id, timestamp) plus routing fields on the concrete payload (org_id, campaign_id, email_account_id, operation_id). The Elixir broadcaster routes purely on those body fields to the user:, org:, campaign:, account:, and bulk: channel topics. The event type list and permission gating are documented in the realtime API reference.

Configuration

# Transport
EVENTBUS_PROVIDER=nats            # nats | kafka (kafka needs the kafka build tag)
NATS_URL=nats://nats:4222
NATS_STREAM_NAME=warmbly          # optional, default warmbly
NATS_SUBJECT_PREFIX=warmbly       # optional, default warmbly
EVENTBUS_HANDLER_TIMEOUT=30s      # optional, per-message handler timeout

# Codec
CODEC_PROVIDER=json               # json | avro (avro needs the kafka build tag)

# Kafka path only
KAFKA_BOOTSTRAP_SERVERS=broker1:9092,broker2:9092
KAFKA_SASL_USERNAME=...
KAFKA_SASL_PASSWORD=...
SCHEMA_REGISTRY_URL=https://schema.example.com
SCHEMA_REGISTRY_KEY=...
SCHEMA_REGISTRY_SECRET=...
KAFKA_TRACKING_TOPIC=tracking-events   # optional, default tracking-events

# Realtime fanout
PUBSUB_ENABLED=false              # true switches Redis pub/sub to Google Cloud Pub/Sub

Code references

• Bus abstraction and providers: internal/infrastructure/eventbus/
• Codec selection: internal/infrastructure/codec/
• Topic names: internal/infrastructure/kafka/topics.go, internal/events/schemas.go
• Envelopes: internal/models/event.go
• Consumer dispatch: internal/app/consumer/events.go
• Tracking producer: tracking/src/nats.rs, tracking/src/kafka.rs
• Realtime fanout: internal/infrastructure/pubsub/, realtime/lib/realtime/redis/event_subscriber.ex