# Architecture of the Macro WebSocket Gateway for Real-Time Communication

> Explore the Macro WebSocket gateway architecture. Learn how the decoupled design separates connection management from event distribution for efficient real-time communication.

- Repository: [Macro/macro](https://github.com/macro-inc/macro)
- Tags: architecture
- Published: 2026-08-14

---

**The Macro WebSocket gateway uses a decoupled architecture where the Connection‑Gateway service manages WebSocket connections, and downstream services publish real-time events via an HTTP client to a `/message/batch_send` endpoint that fans out to connected clients.**

Macro's real-time communication layer is built around a dedicated **Connection‑Gateway** service that isolates WebSocket connection handling from message production. This design keeps business services stateless while centralizing connection state in a single, horizontally scalable service.

## WebSocket Entry Point and Connection Lifecycle

The gateway exposes a WebSocket endpoint at `GET /` in [`services/connection_gateway/src/api/connection/mod.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/api/connection/mod.rs). The `ws_handler` function upgrades HTTPS requests to WebSocket connections and delegates to `handle_websocket_connection` for session management.

When a client connects, the gateway generates a unique `connection_id` (UUID) and registers the socket with the **ConnectionManager** stored in the shared `ApiContext`. The registration captures three critical pieces of state:

- **Entity identity** — the Macro user ID and entity type (`User`)
- **Message channel** — a `tokio::mpsc::Receiver<OutgoingMessage>` for pushing messages to this client
- **Abort handle** — for cleanly terminating the forwarder task on disconnect

## Bidirectional Message Flow: Forwarder and Receiver Loops

Each WebSocket spawns two concurrent Tokio tasks coordinated via `tokio::select!`:

- **`forwarder`** — reads from the internal `Receiver` channel and writes `OutgoingMessage` values to the WebSocket sink
- **`handle_websocket_stream`** — reads inbound WebSocket messages from the client and translates them into domain events

This split ensures that slow consumers don't block outbound broadcasts, and that inbound message processing remains independent of outbound delivery backpressure.

## Decoupled Message Distribution via HTTP

Services never communicate directly with WebSocket connections. Instead, they use the **ConnectionGatewayClient** defined in [`crates/connection_gateway_client/src/client/mod.rs`](https://github.com/macro-inc/macro/blob/main/crates/connection_gateway_client/src/client/mod.rs) — a thin HTTP wrapper that POSTs JSON payloads to the gateway's `/message/batch_send` endpoint.

The client constructs a `BatchSendMessageBody` containing:

```rust
// From crates/connection_gateway_client/src/client/mod.rs
pub struct BatchSendMessageBody {
    pub message_type: String,        // e.g., "comms_message", "typing_indicator"
    pub payload: serde_json::Value,  // Domain-specific JSON
    pub targets: Vec<EntityTarget>,  // List of users to receive this message
}

```

The client returns a `Vec<MessageReceipt>` indicating delivery status per target user.

## Real-Time Publisher Adapters

Service-level publishers wrap the client to expose domain-specific APIs. The **ConnectionGatewayChannelRealtimePublisher** in [`crates/channels/src/outbound/connection_gateway_realtime.rs`](https://github.com/macro-inc/macro/blob/main/crates/channels/src/outbound/connection_gateway_realtime.rs) demonstrates this pattern:

1. Accepts a domain `effect` (new message, reaction, typing indicator)
2. Serializes to JSON and selects the appropriate `message_type` string
3. Invokes `client.batch_send_message()` with target user IDs extracted from channel membership

This adapter pattern appears across services — the notification crate implements a similar wrapper in [`crates/notification/src/outbound/websocket.rs`](https://github.com/macro-inc/macro/blob/main/crates/notification/src/outbound/websocket.rs).

## Connection Cleanup and Presence Tracking

When a WebSocket disconnects or either task fails, `tokio::select!` resolves and triggers cleanup:

1. The forwarder task terminates
2. `ConnectionManager.remove_connection()` deletes the registry entry
3. A "last-online" guard updates the user's presence timestamp in the database

The in-memory connection registry in [`services/connection_gateway/src/context.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/context.rs) remains lightweight — it stores only routing metadata and channels, not message history or user profiles.

## Practical Implementation: Publishing a Real-Time Update

```rust
use connection_gateway_client::ConnectionGatewayClient;
use macro_user_id::user_id::MacroUserIdStr;
use serde_json::json;

// Initialize with internal auth key from Doppler
let client = ConnectionGatewayClient::new(
    std::env::var("CG_INTERNAL_AUTH_KEY").unwrap(),
    "https://connection-gateway.mycompany.com".to_string(),
);

let users = vec![
    MacroUserIdStr::new("user-123"),
    MacroUserIdStr::new("user-456"),
];

let payload = json!({
    "channel_id": "chan-99",
    "message": {
        "id": "msg-abc",
        "content": "Hello, real-time world!"
    }
});

let receipts = client
    .batch_send_message(
        "comms_message".to_string(),
        payload,
        users.iter()
            .map(|u| model_entity::EntityType::User.with_entity_str(u.as_ref()))
            .collect()
    )
    .await
    .expect("failed to send realtime message");

println!("Delivered to {:?} users", receipts.len());

```

## WebSocket Handler Structure

```rust
use axum::extract::ws::WebSocketUpgrade;
use uuid::Uuid;

async fn ws_handler(ws: WebSocketUpgrade) -> impl IntoResponse {
    ws.on_upgrade(|socket| async move {
        let (sink, stream) = socket.split();
        let (sender, receiver) = tokio::sync::mpsc::channel::<OutgoingMessage>(100);
        let connection_id = Uuid::new_v4().to_string();

        CONNECTION_MANAGER
            .add_connection(
                EntityType::User
                    .with_entity_str("user-123")
                    .with_connection_str(&connection_id),
                sender.clone(),
                // abort handle...
            )
            .await
            .unwrap();

        let forwarder = tokio::spawn(async move { forwarder(sink, receiver).await });
        let receiver = handle_websocket_stream(stream, sender.clone()).fuse();

        tokio::select! {
            _ = forwarder => {}
            _ = receiver => {}
        }

        CONNECTION_MANAGER.remove_connection(&connection_id).await.unwrap();
    })
}

```

## Key Source Files

| Component | Path | Responsibility |
|-----------|------|--------------|
| HTTP client SDK | [`crates/connection_gateway_client/src/client/mod.rs`](https://github.com/macro-inc/macro/blob/main/crates/connection_gateway_client/src/client/mod.rs) | Posts batched messages to gateway |
| WebSocket server | [`services/connection_gateway/src/api/connection/mod.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/api/connection/mod.rs) | Accepts connections, runs I/O loops |
| Channel publisher adapter | [`crates/channels/src/outbound/connection_gateway_realtime.rs`](https://github.com/macro-inc/macro/blob/main/crates/channels/src/outbound/connection_gateway_realtime.rs) | Domain-specific publishing API |
| Notification adapter | [`crates/notification/src/outbound/websocket.rs`](https://github.com/macro-inc/macro/blob/main/crates/notification/src/outbound/websocket.rs) | Notification payload distribution |
| Connection registry | [`services/connection_gateway/src/context.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/context.rs) | In-memory session state |

## Summary

- **Connection-Gateway** centralizes WebSocket state while business services remain stateless HTTP clients
- **Two-task architecture** (`forwarder` + `receiver`) handles bidirectional flow with clean failure isolation
- **HTTP-based distribution** decouples message producers from connection management, enabling independent scaling
- **Adapter pattern** in each service (`ConnectionGatewayChannelRealtimePublisher`) hides client complexity behind domain APIs
- **Registry cleanup** maintains accurate presence data without persistent connection state

## Frequently Asked Questions

### How does the Macro WebSocket gateway handle high fan-out for broadcast messages?

The gateway receives batched messages via HTTP and fans out using individual `tokio::mpsc::Sender` channels stored per connection. Each `forwarder` task runs independently, so slow clients only block their own channel — broadcasts to other users proceed concurrently. The HTTP endpoint accepts large target lists, letting services batch millions of recipients in a single request while the gateway handles parallel delivery.

### What happens when a service needs to send real-time updates but the Connection‑Gateway is temporarily unavailable?

The `ConnectionGatewayClient` returns `Result<Vec<MessageReceipt>>` from `batch_send_message`. Services can inspect this result and implement retry logic or dead-letter queues as needed. The client uses `reqwest` with standard timeouts; because the gateway holds no critical business state, temporary unavailability only delays delivery rather than losing data.

### Why does Macro use HTTP from services to the gateway instead of direct WebSocket connections?

HTTP provides **looser coupling** and **better load balancing**. Services don't need persistent connections to the gateway, can use standard HTTP middleware (retries, circuit breakers, metrics), and the gateway can scale independently without service reconfiguration. This matches the **CQRS** pattern: services produce commands (messages) via HTTP; the gateway handles the read-side optimization of live connection fan-out.

### How does the gateway identify which connections belong to a specific user?

The `ConnectionManager` indexes connections by `EntityType` and entity string (the Macro user ID). When `batch_send_message` receives targets, it looks up active `Sender` channels for each user and dispatches only to online sessions. Offline users simply receive no message — the calling service receives a receipt indicating zero successful deliveries for that user.