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

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 at lines 163–176. This publisher wraps two adapters:

• WebSocketGatewayAdapter — pushes payloads to connected frontend clients
• KafkaNotificationRealtimePublisher — emits events to Kafka for downstream consumers

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) captures row inserts and immediately forwards them to the fan-out publisher.

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, the service constructs a FanoutRealtimeSender and injects it into NotificationEgressService:

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
FanoutRealtimeSender	Dual-adapter sender for egress worker	src/main.rs
NotificationEventsListener	PostgreSQL NOTIFY event consumer	src/main.rs
PgNotificationEventsReceiver	Async PostgreSQL LISTEN implementation	src/main.rs
ConnectionGatewayClient	Authenticated WebSocket gateway client	src/main.rs
MacroEventBrokerService	Kafka publishing infrastructure	src/main.rs
NotificationEgressService	Core egress orchestrator with realtime field	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:

// 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}