# How the Macro Notification Service Delivers Real-Time User Notifications via WebSocket and Kafka

> Discover how Macro's notification service uses WebSocket and Kafka to send real-time user alerts with sub-second latency. Learn about its scalable fan-out architecture.

- Repository: [Macro/macro](https://github.com/macro-inc/macro)
- Tags: how-to-guide
- Published: 2026-08-15

---

**The Macro notification service delivers real-time user notifications through a fan-out architecture that simultaneously broadcasts events to WebSocket-connected clients and Kafka consumers, achieving sub-second latency at scale.**

Built in Rust, the `macro-inc/macro` notification service combines **PostgreSQL `NOTIFY`** events, **dual-path publishing**, and **async workers** to ensure users receive instant updates whether they're actively connected via WebSocket or downstream services need to react to notification events. This article breaks down the three-stage flow using actual source code from the repository.

---

## Stage 1: Creating the Fan-Out Real-Time Publisher

The foundation of real-time delivery is the **`FanoutNotificationRealtimePublisher`**, instantiated in [`services/notification_service/src/main.rs`](https://github.com/macro-inc/macro/blob/main/services/notification_service/src/main.rs) at lines 163–176. This publisher wraps two adapters:

- **`WebSocketGatewayAdapter`** — pushes payloads to connected frontend clients
- **`KafkaNotificationRealtimePublisher`** — emits events to Kafka for downstream consumers

```rust
let realtime_event_broker = MacroEventBrokerService::new(
    KafkaEventPublisher::new(config.kafka_brokers.as_ref())
        .context("failed to create Kafka realtime notification publisher")?,
    GlobalSpawner,
);

let reader_realtime_adapter = FanoutNotificationRealtimePublisher::new(
    WebSocketGatewayAdapter {
        gateway: ConnectionGatewayClient::new(
            config.internal_api_key.as_ref().to_string(),
            connection_gateway_url.clone(),
        ),
    },
    KafkaNotificationRealtimePublisher::new(realtime_event_broker.clone()),
);

```

The `ConnectionGatewayClient` handles WebSocket authentication using an internal API key, while the Kafka publisher leverages `MacroEventBrokerService` for reliable event streaming. This dual-adapter pattern ensures **zero message loss** — if WebSocket delivery fails, Kafka still captures the event for retry or audit.

---

## Stage 2: Listening for Database Notification Events

Real-time triggers originate from PostgreSQL via the **`LISTEN`/`NOTIFY`** mechanism. The `NotificationEventsListener` (configured at lines 86–95 in [`main.rs`](https://github.com/macro-inc/macro/blob/main/main.rs)) captures row inserts and immediately forwards them to the fan-out publisher.

```rust
let notification_events_receiver = PgNotificationEventsReceiver::new(db.clone());
let mut notification_events_listener = NotificationEventsListener::new(
    notification_events_receiver,
    notification_events_realtime_adapter,  // the fan-out publisher from Stage 1
);

tokio::spawn(async move {
    tracing::info!("starting notification database event listener");
    notification_events_listener.run().await
});

```

**How it works:**

1. A new notification row inserts into PostgreSQL
2. A database trigger emits `NOTIFY` with the payload
3. `PgNotificationEventsReceiver` receives the event via its async connection
4. The listener delegates to `notification_events_realtime_adapter`, which **simultaneously** pushes to WebSocket and Kafka

This architecture eliminates polling overhead and reduces latency to **database round-trip + network hop** times.

---

## Stage 3: Egress Worker Real-Time Dispatch

The egress pipeline — responsible for sending emails, push notifications, and in-app alerts — also participates in real-time broadcasting. At lines 14–22 and 58–62 in [`main.rs`](https://github.com/macro-inc/macro/blob/main/main.rs), the service constructs a **`FanoutRealtimeSender`** and injects it into `NotificationEgressService`:

```rust
let realtime_adapter = FanoutRealtimeSender::new(
    WebSocketGatewayAdapter {
        gateway: ConnectionGatewayClient::new(
            config.internal_api_key.as_ref().to_string(),
            connection_gateway_url,
        ),
    },
    KafkaRealtimeSender::new(realtime_event_broker),
);

// Inside NotificationEgressService
let egress_service = NotificationEgressService::new(
    // ... other fields ...
    realtime: realtime_adapter,  // enables real-time payload broadcast
);

```

When `NotificationWorker` processes a notification, it calls `realtime.send(payload).await`, triggering the same dual-path delivery:

- **Path A:** WebSocket gateway notifies active frontend sessions instantly
- **Path B:** Kafka topic allows other microservices to react (analytics, audit logging, cross-service synchronization)

---

## Key Components and Source Files

| Component | Purpose | Location |
|-----------|---------|----------|
| `FanoutNotificationRealtimePublisher` | Dual-adapter publisher for DB events | [`src/main.rs`](https://github.com/macro-inc/macro/blob/main/src/main.rs#L163-L176) |
| `FanoutRealtimeSender` | Dual-adapter sender for egress worker | [`src/main.rs`](https://github.com/macro-inc/macro/blob/main/src/main.rs#L14-L22) |
| `NotificationEventsListener` | PostgreSQL `NOTIFY` event consumer | [`src/main.rs`](https://github.com/macro-inc/macro/blob/main/src/main.rs#L86-L95) |
| `PgNotificationEventsReceiver` | Async PostgreSQL LISTEN implementation | [`src/main.rs`](https://github.com/macro-inc/macro/blob/main/src/main.rs#L86) |
| `ConnectionGatewayClient` | Authenticated WebSocket gateway client | [`src/main.rs`](https://github.com/macro-inc/macro/blob/main/src/main.rs#L169-L172) |
| `MacroEventBrokerService` | Kafka publishing infrastructure | [`src/main.rs`](https://github.com/macro-inc/macro/blob/main/src/main.rs#L163-L165) |
| `NotificationEgressService` | Core egress orchestrator with `realtime` field | [`src/notification/mod.rs`](https://github.com/macro-inc/macro/blob/main/src/notification/mod.rs) |

---

## Complete Real-Time Flow Example

Here's the end-to-end integration showing how a notification moves from database to user:

```rust
// 1. Build shared Kafka event broker
let kafka_broker = MacroEventBrokerService::new(
    KafkaEventPublisher::new(config.kafka_brokers.as_ref())?,
    GlobalSpawner,
);

// 2. Create the fan-out publisher (used by both listener and egress)
let realtime_publisher = FanoutNotificationRealtimePublisher::new(
    WebSocketGatewayAdapter {
        gateway: ConnectionGatewayClient::new(api_key, gateway_url),
    },
    KafkaNotificationRealtimePublisher::new(kafka_broker.clone()),
);

// 3. Start DB event listener
let db_listener = NotificationEventsListener::new(
    PgNotificationEventsReceiver::new(db_pool),
    realtime_publisher.clone(),
);
tokio::spawn(db_listener.run());

// 4. Build egress service with real-time capability
let egress = NotificationEgressService::new(
    /* email/push providers */,
    FanoutRealtimeSender::new(
        WebSocketGatewayAdapter {
            gateway: ConnectionGatewayClient::new(api_key, gateway_url),
        },
        KafkaRealtimeSender::new(kafka_broker),
    ),
);

```

When a user triggers a notification:

1. **Insert** → PostgreSQL `NOTIFY` fires
2. **Listen** → `NotificationEventsListener` captures event
3. **Fan-out** → WebSocket clients receive instant update
4. **Fan-out** → Kafka consumers process asynchronously
5. **Egress** → Worker sends email/push and re-broadcasts real-time payload for confirmation

---

## Summary

- **Dual-path delivery** via `FanoutNotificationRealtimePublisher` and `FanoutRealtimeSender` ensures both immediate WebSocket delivery and durable Kafka logging
- **PostgreSQL `NOTIFY`** eliminates polling overhead for database-driven real-time updates
- **Modular adapter pattern** allows independent scaling of WebSocket gateways and Kafka clusters
- **Egress integration** guarantees users see notification confirmations even for async channels like email

---

## Frequently Asked Questions

### How does the notification service achieve low latency without polling?

The service uses **PostgreSQL `LISTEN`/`NOTIFY`** to push events immediately when rows insert, combined with **async Tokio tasks** that process events without blocking. This architecture achieves sub-100ms latency from database commit to WebSocket delivery in typical deployments.

### What happens if the WebSocket gateway is unavailable?

The **fan-out publisher** still succeeds if Kafka delivery completes. The `ConnectionGatewayClient` may retry or log failures via `tracing`, but the Kafka path ensures **event durability** — downstream consumers can process notifications even if real-time WebSocket delivery fails temporarily.

### Can other services consume notification events without WebSocket connections?

Yes. The **`KafkaNotificationRealtimePublisher`** and **`KafkaRealtimeSender`** emit to configurable Kafka topics. Any service with Kafka credentials can subscribe to these topics to react to notification events, enabling **cross-service workflows** without direct WebSocket integration.

### How is the WebSocket gateway authenticated?

The `ConnectionGatewayClient` initializes with an **`internal_api_key`** read from environment configuration (`config.internal_api_key`). This key authenticates inter-service requests to the Connection Gateway, preventing unauthorized real-time payload injection as implemented in lines 169–172 of [`main.rs`](https://github.com/macro-inc/macro/blob/main/main.rs).