# How the Macro Connection Gateway Handles WebSockets and Real-Time Messaging

> Discover how the Macro connection gateway uses WebSockets and Redis Pub/Sub for real-time messaging, powering live updates and notifications on the platform.

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

---

**The Macro connection gateway upgrades HTTP requests to persistent WebSocket connections and broadcasts real-time messages to clients via a Redis Pub/Sub architecture, enabling live document updates and scheduled-action notifications across the platform.**

The connection gateway is a critical service in the `macro-inc/macro` repository that manages persistent client connections and real-time message delivery. Written in Rust using the Axum framework, this service enables live collaboration features by bridging WebSocket clients with backend services through a scalable Pub/Sub infrastructure.

## WebSocket Upgrade and Connection Management

The gateway exposes a single HTTP endpoint at `/connection-gateway` that upgrades incoming requests to WebSocket connections. When a client connects, the handler in [`services/connection_gateway/src/router.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/router.rs) calls `axum::extract::ws::upgrade` to switch the protocol from HTTP to WebSocket.

The upgraded socket is wrapped in a `Connection` struct that stores the client’s identity—including the user ID and session token—and maintains a sender channel for outbound messages. This design keeps the connection state encapsulated while allowing asynchronous message delivery to the client.

## Connection Registry Architecture

All active WebSocket connections are tracked in a thread-safe registry defined in [`services/connection_gateway/src/registry.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/registry.rs). The implementation uses a `HashMap<ConnectionId, Sender>` protected by an `RwLock`, enabling concurrent reads across multiple tasks while ensuring safe mutation during connection establishment and teardown.

When a new WebSocket is created, the gateway inserts its sender handle into this map using a unique connection identifier. When the socket closes—either through client disconnection or server shutdown—the entry is removed from the registry to prevent memory leaks and message delivery attempts to dead connections.

## Redis Pub/Sub Message Broadcasting

Real-time message distribution relies on Redis Pub/Sub channels. Each gateway instance subscribes to a tenant-specific channel named `connection-gateway` via the subscriber implementation in [`services/connection_gateway/src/redis_subscriber.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/redis_subscriber.rs).

When any service publishes a payload—such as a new comment, a scheduled-action update, or a typing indicator—the gateway receives the message from Redis, looks up all matching target sockets in the registry, and forwards the JSON payload through each socket’s sender channel. This decouples message producers from direct WebSocket management.

## Publishing Messages from Services

Downstream services do not communicate directly with Redis. Instead, they use the `connection_gateway_client` crate located at [`crates/connection_gateway_client/src/lib.rs`](https://github.com/macro-inc/macro/blob/main/crates/connection_gateway_client/src/lib.rs). This client provides a thin wrapper around an authenticated HTTP POST call to the gateway’s `/publish` endpoint.

The `ConnectionGatewayClient::publish` method marshals the request into a JSON payload and transmits it over HTTP. The gateway validates an internal API secret, then republishes the message onto the Redis channel for broadcasting to connected clients. This abstraction ensures consistent authentication and message formatting across all services.

## Real-World Use Cases

### Live Comment and Anchor Updates

When a user creates, edits, or deletes a comment or anchor in a document, the document-storage service publishes a `ConnectionMessage` via the client crate. As implemented in [`services/document_storage_service/src/api/annotations/create_comment.rs`](https://github.com/macro-inc/macro/blob/main/services/document_storage_service/src/api/annotations/create_comment.rs), these events trigger immediate updates to every client that has the document open, ensuring collaborative editing remains synchronized.

### Scheduled-Action State Changes

The scheduled-action service sends live status updates—such as "action started" or "action completed"—through the gateway. The implementation in [`services/scheduled_action/src/outbound/conn_gateway_live_updates.rs`](https://github.com/macro-inc/macro/blob/main/services/scheduled_action/src/outbound/conn_gateway_live_updates.rs) demonstrates how long-running background jobs communicate progress to frontend components in real time.

### End-to-End WebSocket Testing

The sync-service test suite validates the entire messaging pipeline. Located at [`services/sync-service/tests/e2e.test.ts`](https://github.com/macro-inc/macro/blob/main/services/sync-service/tests/e2e.test.ts), these tests spin up real WebSocket clients, register them with the gateway, and assert that messages published by backend services are received within milliseconds.

## Horizontal Scaling and Resilience

The gateway architecture supports horizontal scaling through stateless design principles. The `/publish` HTTP endpoint is stateless; any gateway instance can handle publication requests because Redis serves as the single source of truth for message distribution.

Multiple gateway instances can run behind a load balancer, with all instances subscribing to the same Redis channels. This guarantees that every connected client receives every broadcast exactly once, regardless of which gateway node manages their specific WebSocket connection.

For graceful shutdown, the `Drop` implementation for the `Connection` struct automatically removes the client from the registry and unsubscribes the Redis listener when a gateway pod stops. This prevents dangling references and ensures clean disconnection without message loss.

## Code Examples

The following examples demonstrate how to publish messages from a service and how clients connect to receive real-time updates.

```rust
// Publishing a message from a backend service
use connection_gateway_client::client::ConnectionGatewayClient;
use connection_gateway_client::message::ConnectionMessage;

let client = ConnectionGatewayClient::new(
    std::env::var("INTERNAL_API_SECRET_KEY")?,
    std::env::var("CONNECTION_GATEWAY_URL")?,
);

let msg = ConnectionMessage {
    tenant_id: "tenant-123".into(),
    target: ConnectionMessageTarget::User("user-456".into()),
    payload: serde_json::json!({ "type": "comment_created", "comment_id": "c789" }),
};

client.publish(msg).await?;

```

```typescript
// Connecting from a web client
const ws = new WebSocket(
  `${import.meta.env.VITE_CONNECTION_GATEWAY_WS_URL}/connection-gateway`
);

ws.addEventListener('open', () => {
  console.log('WebSocket connection established');
});

ws.addEventListener('message', (event) => {
  const data = JSON.parse(event.data);
  // Handle real-time updates (e.g., new comment, typing indicator)
  console.log('Received:', data);
});

ws.addEventListener('close', () => {
  console.log('WebSocket closed');
});

```

## Summary

- The gateway upgrades HTTP connections to WebSockets at `/connection-gateway` using Axum’s upgrade mechanism in [`services/connection_gateway/src/router.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/router.rs).
- Active connections are stored in a thread-safe `HashMap<ConnectionId, Sender>` within [`services/connection_gateway/src/registry.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/registry.rs).
- Messages are broadcast via Redis Pub/Sub through [`services/connection_gateway/src/redis_subscriber.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/redis_subscriber.rs), enabling horizontal scaling across multiple gateway instances.
- Services publish messages using the `connection_gateway_client` crate rather than connecting directly to Redis, ensuring consistent authentication via the `/publish` endpoint.
- Real-time features—including live document comments and scheduled-action updates—rely on this architecture to deliver sub-second updates to connected clients.
- Graceful shutdown is handled through Rust’s `Drop` trait, which cleans up registry entries and Redis subscriptions automatically.

## Frequently Asked Questions

### How does the connection gateway scale horizontally?

The gateway achieves horizontal scalability by maintaining stateless HTTP endpoints for message publishing and delegating message distribution to Redis. Multiple gateway instances can run behind a load balancer, with each instance subscribing to the same Redis Pub/Sub channels. Since all connection state is local to each instance (stored in the registry) while broadcasts originate from Redis, the system scales linearly with the number of clients and messages.

### What Redis channel does the gateway use for broadcasting?

According to the source code in [`services/connection_gateway/src/redis_subscriber.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/redis_subscriber.rs), the gateway subscribes to a channel named `connection-gateway` for each tenant. All backend services publish messages to this channel, and every active gateway instance receives these messages to forward to their connected WebSocket clients.

### How do backend services publish messages without talking directly to Redis?

Services utilize the `connection_gateway_client` crate defined in [`crates/connection_gateway_client/src/lib.rs`](https://github.com/macro-inc/macro/blob/main/crates/connection_gateway_client/src/lib.rs). This client provides a `ConnectionGatewayClient::publish` method that sends an authenticated HTTP POST request to the gateway’s `/publish` endpoint. The gateway then validates the internal API secret and republishes the payload to Redis, abstracting the messaging infrastructure from service developers.

### How is the WebSocket connection upgraded from HTTP?

The upgrade process occurs in [`services/connection_gateway/src/router.rs`](https://github.com/macro-inc/macro/blob/main/services/connection_gateway/src/router.rs) using `axum::extract::ws::upgrade`. When a client requests `/connection-gateway`, the Axum server switches the protocol to WebSocket, wraps the socket in a `Connection` struct containing the user identity and a sender channel, and registers the connection in the thread-safe registry for message delivery.