Architecture of the Macro WebSocket Gateway for Real-Time Communication

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. 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 — a thin HTTP wrapper that POSTs JSON payloads to the gateway's /message/batch_send endpoint.

The client constructs a BatchSendMessageBody containing:

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

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 remains lightweight — it stores only routing metadata and channels, not message history or user profiles.

Practical Implementation: Publishing a Real-Time Update

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

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	Posts batched messages to gateway
WebSocket server	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	Domain-specific publishing API
Notification adapter	crates/notification/src/outbound/websocket.rs	Notification payload distribution
Connection registry	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.

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