Skip to main content

Server-Sent Events (SSE)

SseManager maintains a registry of open HTTP streams and broadcasts StreamEvent objects to all clients subscribed to a matching topic. The server controls which topics a connection may receive — clients cannot self-declare channels.

Connection lifecycle​

Setup​

Enable SSE when creating AuthTools:

const tools = new AuthTools(bus, { sse: true });

Mount the tools router (exposes GET /tools/stream):

app.use('/tools', createToolsRouter(tools, {
  authMiddleware: auth.middleware(),
}));

1. Setup in your Backend​

First, ensure SSE is enabled when creating your AuthTools instance, and that the tools router is mounted with your auth middleware.

import express from 'express';
import { AuthConfigurator, AuthTools, createToolsRouter } from 'awesome-node-auth';

const app = express();
app.use(express.json());

// Initialize core auth
const auth = new AuthConfigurator(config, userStore);
app.use('/auth', auth.router());

// Enable SSE in the tools configuration
const tools = new AuthTools(bus, { sse: true });

// Mount the tools router, protected by auth.middleware()
app.use('/tools', createToolsRouter(tools, {
  authMiddleware: auth.middleware(),
}));

// Now you can broadcast events from anywhere in your app:
app.post('/api/messages', auth.middleware(), (req, res) => {
  const { recipientId, text } = req.body;
  
  // 🎯 Broadcast a custom event to a specific user
  tools.sseManager?.broadcastToUser(recipientId, {
    type: 'new_message',
    data: { 
      from: req.user.id, 
      text 
    }
  });

  res.sendStatus(200);
});

2. Connecting from the Frontend​

When a user logs in, they can connect to the /tools/stream endpoint. Because the router is protected by auth.middleware(), the user will only receive events that match their own user:{id}, their tenant:{id}, or global topics.

Vanilla JavaScript Example​

// The topics query param tells the server which channels you want to listen to.
// The server will verify you have permission to access them.
const currentUserId = '123';
const evtSource = new EventSource(`/tools/stream?topics=user:${currentUserId}`, { 
  withCredentials: true 
});

// Listen for a specific event type (matches the `type` field in broadcast)
evtSource.addEventListener('new_message', (event) => {
  const payload = JSON.parse(event.data);
  console.log(`New message from ${payload.data.from}: ${payload.data.text}`);
});

// The library automatically sends ping events every 30s to keep the connection alive
evtSource.addEventListener('ping', () => {
  console.debug('SSE connection is alive');
});

// Handle errors and reconnections
evtSource.onerror = (err) => {
  console.error("SSE Error:", err);
  // Browsers automatically try to reconnect EventSources, 
  // but you can call evtSource.close() to stop it.
};

React Example​

A practical way to consume SSE in a React application using a custom hook:

import { useEffect, useState } from 'react';

export function useMessages(userId: string) {
  const [messages, setMessages] = useState([]);

  useEffect(() => {
    if (!userId) return;

    const evtSource = new EventSource(`http://api.myapp.com/tools/stream?topics=user:${userId}`, {
      withCredentials: true, // Crucial if you use cookie-based auth
    });

    evtSource.addEventListener('new_message', (event) => {
      const payload = JSON.parse(event.data);
      setMessages((prev) => [...prev, payload.data]);
    });

    return () => {
      evtSource.close(); // Cleanup when unmounting
    };
  }, [userId]);

  return messages;
}

// Usage in a component:
function ChatApp({ currentUser }) {
  const messages = useMessages(currentUser.id);
  
  return (
    <ul>
      {messages.map((msg, i) => <li key={i}>{msg.text}</li>)}
    </ul>
  );
}

Proxy Configuration (Nginx / Traefik / Apache) ⚠️​

If your Node.js application is running behind a reverse proxy, you must ensure the proxy doesn't buffer the /tools/stream endpoint. Otherwise, the proxy will hold onto the events and not send them to the client in real-time.

Nginx​

Nginx buffers HTTP responses by default. You must explicitly disable it for the SSE endpoint.

Example Nginx config:

location /tools/stream {
    proxy_pass http://localhost:3000;
    
    # Required for SSE: disable buffering and caching
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;
    proxy_cache off;
    chunked_transfer_encoding off;
    
    # Keep idle connections open longer
    proxy_read_timeout 24h;
}

Traefik​

Traefik natively supports SSE (Server-Sent Events) out of the box and does not buffer responses by default. In most setups, no explicit configuration is needed.

However, keep these two things in mind:

  1. Ensure you haven't globally applied a Buffering middleware to the router serving node-auth.
  2. If your SSE connections are dropping prematurely, check if you have an aggressive readTimeout set on your Traefik entrypoints or transport configurations.

Topic hierarchy​

The server enforces topic access — restrict which topics each user/connection may subscribe to via authMiddleware.

TopicDescription
globalAll connected clients
tenant:{tenantId}All clients in a tenant
tenant:{tenantId}:role:{role}All clients with a role in a tenant
tenant:{tenantId}:group:{groupId}All clients in a group
user:{userId}A specific user
session:{sessionId}A specific session
custom:{namespace}Application-defined namespace

StreamEvent structure​

interface StreamEvent<T = unknown> {
  id: string;           // auto-generated UUID
  type: string;         // event type name
  timestamp: string;    // ISO 8601
  topic: string;        // channel this was published to
  data: T;              // arbitrary payload
  userId?: string;
  tenantId?: string;
  metadata?: Record<string, unknown>;
}

SseManager API​

When sse: true is passed to AuthTools, the manager is available at tools.sseManager:

class SseManager {
  readonly heartbeatIntervalMs: number;  // default: 30000

  addConnection(
    res: Response,
    topics: string[],
    meta?: { userId?: string; tenantId?: string },
  ): string;  // returns connectionId

  removeConnection(connectionId: string): void;

  broadcast(topic: string, event: Omit<StreamEvent, 'id' | 'timestamp' | 'topic'>): void;

  broadcastToUser(userId: string, event: Omit<StreamEvent, 'id' | 'timestamp' | 'topic'>): void;

  broadcastToTenant(tenantId: string, event: Omit<StreamEvent, 'id' | 'timestamp' | 'topic'>): void;

  get connectionCount(): number;
}