import WebSocket, { Server as WebSocketServer } from 'ws';
import http from 'http';
import { verifyToken } from './auth';

const server = http.createServer();
const wss = new WebSocketServer({ server });

interface AuthenticatedWebSocket extends WebSocket {
  userId?: string;
  isAlive?: boolean;
}

// Track active connections per user
const userConnections = new Map<string, Set<AuthenticatedWebSocket>>();

// Middleware-style authentication
function verifyClient(
  info: { origin: string; secure: boolean; req: http.IncomingMessage },
  callback: (success: boolean, code?: number, message?: string) => void,
) {
  const url = new URL(info.req.url || '', 'http://localhost');
  const token = url.searchParams.get('token');

  if (!token) {
    callback(false, 4001, 'Token required');
    return;
  }

  try {
    const decoded = verifyToken(token);
    (info.req as any).userId = decoded.sub;
    callback(true);
  } catch (err) {
    callback(false, 4001, 'Invalid token');
  }
}

const wss = new WebSocketServer({
  server,
  verifyClient,
  clientTracking: false, // Manage connections manually for multi-server setups
});

wss.on('connection', (ws: AuthenticatedWebSocket, req) => {
  const userId = (req as any).userId;
  ws.userId = userId;
  ws.isAlive = true;

  // Track connection
  if (!userConnections.has(userId)) {
    userConnections.set(userId, new Set());
  }
  userConnections.get(userId)!.add(ws);

  console.log(`User ${userId} connected`);

  // Heartbeat
  ws.on('pong', () => {
    ws.isAlive = true;
  });

  // Handle incoming messages
  ws.on('message', async (data: WebSocket.Data) => {
    try {
      const message = JSON.parse(data.toString());

      switch (message.type) {
        case 'chat':
          broadcastToUser(message.recipientId, {
            type: 'chat',
            from: userId,
            text: message.text,
            timestamp: new Date(),
          });
          break;

        case 'presence':
          broadcastToUser(userId, {
            type: 'user_status',
            userId,
            status: message.status,
          });
          break;

        default:
          ws.send(JSON.stringify({ error: 'Unknown message type' }));
      }
    } catch (err) {
      ws.send(JSON.stringify({ error: 'Invalid message format' }));
    }
  });

  // Handle errors
  ws.on('error', (err) => {
    console.error(`WebSocket error for user ${userId}:`, err);
  });

  // Handle close
  ws.on('close', (code, reason) => {
    const connections = userConnections.get(userId);
    if (connections) {
      connections.delete(ws);
      if (connections.size === 0) {
        userConnections.delete(userId);
      }
    }
    console.log(`User ${userId} disconnected (code: ${code})`);
  });
});

// Heartbeat interval (detects dead connections)
setInterval(() => {
  wss.clients.forEach((ws: AuthenticatedWebSocket) => {
    if (ws.isAlive === false) {
      ws.terminate();
      return;
    }
    ws.isAlive = false;
    ws.ping();
  });
}, 30000);

function broadcastToUser(userId: string, message: any) {
  const connections = userConnections.get(userId);
  if (connections) {
    connections.forEach((ws) => {
      if (ws.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify(message));
      }
    });
  }
}

server.listen(3000, () => {
  console.log('WebSocket server listening on port 3000');
});

// Spring Boot WebSocket Server with STOMP
// pom.xml: spring-boot-starter-websocket, jjwt

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Controller;
import org.springframework.web.socket.config.annotation.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue");
        registry.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }
}

@Controller
@EnableScheduling
public class WebSocketController {

    private final SimpMessagingTemplate messagingTemplate;
    private final Map<String, Set<String>> userSessions = new ConcurrentHashMap<>();

    public WebSocketController(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    @MessageMapping("/chat")
    public void handleChat(@Payload ChatMessage message, Principal principal) {
        String fromUserId = principal.getName();
        ChatMessage outgoing = new ChatMessage(
            "chat", fromUserId, message.getText(), new Date()
        );
        // Route to recipient's personal queue
        messagingTemplate.convertAndSendToUser(
            message.getRecipientId(), "/queue/messages", outgoing
        );
    }

    @MessageMapping("/presence")
    public void handlePresence(@Payload PresenceMessage message, Principal principal) {
        String userId = principal.getName();
        messagingTemplate.convertAndSendToUser(
            userId, "/queue/status",
            new StatusMessage(userId, message.getStatus())
        );
    }

    // Heartbeat: Spring STOMP handles ping/pong at the protocol level
    @Scheduled(fixedDelay = 30000)
    public void sendHeartbeat() {
        messagingTemplate.convertAndSend("/topic/heartbeat", Map.of("type", "ping"));
    }

    public record ChatMessage(String type, String from, String text, Date timestamp) {
        public String getRecipientId() { return ""; } // set via payload
        public String getText() { return text; }
        public String getRecipientId(String s) { return s; }
    }
    public record PresenceMessage(String status) {
        public String getStatus() { return status; }
    }
    public record StatusMessage(String userId, String status) {}
}

# FastAPI WebSocket Server
# pip install fastapi uvicorn websockets python-jose

import asyncio
import json
from collections import defaultdict
from datetime import datetime
from typing import Dict, Set

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, status
from jose import jwt, JWTError

app = FastAPI()

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

# Track active connections per user
user_connections: Dict[str, Set[WebSocket]] = defaultdict(set)


def verify_token(token: str) -> dict:
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Query(...)):
    # Authenticate before accepting
    try:
        payload = verify_token(token)
        user_id = payload["sub"]
    except JWTError:
        await websocket.close(code=4001)
        return

    await websocket.accept()
    user_connections[user_id].add(websocket)
    print(f"User {user_id} connected")

    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)

            if message["type"] == "chat":
                await broadcast_to_user(
                    message["recipientId"],
                    {
                        "type": "chat",
                        "from": user_id,
                        "text": message["text"],
                        "timestamp": datetime.utcnow().isoformat(),
                    },
                )
            elif message["type"] == "presence":
                await broadcast_to_user(
                    user_id,
                    {"type": "user_status", "userId": user_id, "status": message["status"]},
                )
            else:
                await websocket.send_text(json.dumps({"error": "Unknown message type"}))

    except WebSocketDisconnect:
        user_connections[user_id].discard(websocket)
        if not user_connections[user_id]:
            del user_connections[user_id]
        print(f"User {user_id} disconnected")
    except Exception as e:
        print(f"WebSocket error for user {user_id}: {e}")
        user_connections[user_id].discard(websocket)


async def broadcast_to_user(user_id: str, message: dict):
    connections = user_connections.get(user_id, set())
    dead = set()
    for ws in connections:
        try:
            await ws.send_text(json.dumps(message))
        except Exception:
            dead.add(ws)
    connections -= dead


# Heartbeat task
async def heartbeat_task():
    while True:
        await asyncio.sleep(30)
        for user_id, connections in list(user_connections.items()):
            dead = set()
            for ws in connections:
                try:
                    await ws.send_text(json.dumps({"type": "ping"}))
                except Exception:
                    dead.add(ws)
            connections -= dead


@app.on_event("startup")
async def startup():
    asyncio.create_task(heartbeat_task())

// ASP.NET Core WebSocket Server
// No extra packages needed — built-in WebSocket support

using System.Collections.Concurrent;
using System.Net.WebSockets;
using System.Text;
using System.Text.Json;

var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();

app.UseWebSockets(new WebSocketOptions { KeepAliveInterval = TimeSpan.FromSeconds(30) });

// Track connections per user
var userConnections = new ConcurrentDictionary<string, ConcurrentBag<WebSocket>>();

app.Map("/ws", async (HttpContext context) =>
{
    if (!context.WebSockets.IsWebSocketRequest)
    {
        context.Response.StatusCode = 400;
        return;
    }

    var token = context.Request.Query["token"].ToString();
    var userId = ValidateToken(token); // implement your JWT validation
    if (userId == null)
    {
        context.Response.StatusCode = 401;
        return;
    }

    var ws = await context.WebSockets.AcceptWebSocketAsync();
    userConnections.GetOrAdd(userId, _ => new ConcurrentBag<WebSocket>()).Add(ws);
    Console.WriteLine($"User {userId} connected");

    var buffer = new byte[4096];
    try
    {
        while (ws.State == WebSocketState.Open)
        {
            var result = await ws.ReceiveAsync(buffer, CancellationToken.None);
            if (result.MessageType == WebSocketMessageType.Close)
                break;

            var json = Encoding.UTF8.GetString(buffer, 0, result.Count);
            var message = JsonSerializer.Deserialize<JsonElement>(json);
            var type = message.GetProperty("type").GetString();

            if (type == "chat")
            {
                var recipientId = message.GetProperty("recipientId").GetString()!;
                var outgoing = JsonSerializer.Serialize(new
                {
                    type = "chat",
                    from = userId,
                    text = message.GetProperty("text").GetString(),
                    timestamp = DateTime.UtcNow,
                });
                await BroadcastToUser(userConnections, recipientId, outgoing);
            }
        }
    }
    finally
    {
        var bag = userConnections.GetValueOrDefault(userId);
        // ConcurrentBag doesn't support remove; swap to a cleaner structure in production
        Console.WriteLine($"User {userId} disconnected");
        await ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "Bye", CancellationToken.None);
    }
});

static async Task BroadcastToUser(
    ConcurrentDictionary<string, ConcurrentBag<WebSocket>> connections,
    string userId, string message)
{
    if (!connections.TryGetValue(userId, out var sockets)) return;
    var bytes = Encoding.UTF8.GetBytes(message);
    foreach (var ws in sockets)
    {
        if (ws.State == WebSocketState.Open)
            await ws.SendAsync(bytes, WebSocketMessageType.Text, true, CancellationToken.None);
    }
}

static string? ValidateToken(string token)
{
    // Implement JWT validation; return userId or null
    return null;
}

app.Run();

class WebSocketClient {
  private ws: WebSocket | null = null;
  private url: string;
  private token: string;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 5;
  private reconnectDelay = 1000;

  constructor(url: string, token: string) {
    this.url = url;
    this.token = token;
  }

  connect(): Promise<void> {
    return new Promise((resolve, reject) => {
      try {
        this.ws = new WebSocket(
          `${this.url}?token=${encodeURIComponent(this.token)}`,
        );

        this.ws.onopen = () => {
          console.log('Connected');
          this.reconnectAttempts = 0;
          resolve();
        };

        this.ws.onmessage = (event) => {
          const message = JSON.parse(event.data);
          this.handleMessage(message);
        };

        this.ws.onerror = (event) => {
          console.error('WebSocket error:', event);
          reject(new Error('Connection failed'));
        };

        this.ws.onclose = () => {
          console.log('Disconnected');
          this.attemptReconnect();
        };
      } catch (err) {
        reject(err);
      }
    });
  }

  send(message: any) {
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(message));
    } else {
      console.warn('WebSocket not connected');
    }
  }

  private attemptReconnect() {
    if (this.reconnectAttempts < this.maxReconnectAttempts) {
      this.reconnectAttempts++;
      const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
      console.log(`Reconnecting in ${delay}ms...`);
      setTimeout(() => this.connect().catch(console.error), delay);
    }
  }

  private handleMessage(message: any) {
    switch (message.type) {
      case 'chat':
        console.log(`Message from ${message.from}: ${message.text}`);
        break;
      default:
        console.log('Received:', message);
    }
  }

  disconnect() {
    if (this.ws) {
      this.ws.close(1000, 'Normal closure');
    }
  }
}

// Usage
const client = new WebSocketClient(
  'ws://localhost:3000/ws',
  'your-jwt-token',
);
await client.connect();
client.send({ type: 'chat', recipientId: 'user123', text: 'Hello!' });

// Java WebSocket Client using Jakarta WebSocket API
// Maven: jakarta.websocket-api, tyrus-standalone-client

import jakarta.websocket.*;
import java.net.URI;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@ClientEndpoint
public class WebSocketClient {

    private Session session;
    private final String url;
    private final String token;
    private int reconnectAttempts = 0;
    private final int maxReconnectAttempts = 5;
    private final long reconnectDelayMs = 1000;
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public WebSocketClient(String url, String token) {
        this.url = url;
        this.token = token;
    }

    public CompletableFuture<Void> connect() {
        var future = new CompletableFuture<Void>();
        try {
            var container = ContainerProvider.getWebSocketContainer();
            container.connectToServer(this, URI.create(url + "?token=" + token));
            future.complete(null);
        } catch (Exception e) {
            future.completeExceptionally(e);
        }
        return future;
    }

    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        this.reconnectAttempts = 0;
        System.out.println("Connected");
    }

    @OnMessage
    public void onMessage(String messageJson) {
        // Parse and handle message
        System.out.println("Received: " + messageJson);
    }

    @OnClose
    public void onClose(Session session, CloseReason reason) {
        System.out.println("Disconnected: " + reason);
        attemptReconnect();
    }

    @OnError
    public void onError(Session session, Throwable error) {
        System.err.println("WebSocket error: " + error.getMessage());
    }

    public void send(String messageJson) {
        if (session != null && session.isOpen()) {
            session.getAsyncRemote().sendText(messageJson);
        } else {
            System.out.println("WebSocket not connected");
        }
    }

    private void attemptReconnect() {
        if (reconnectAttempts < maxReconnectAttempts) {
            long delay = reconnectDelayMs * (1L << reconnectAttempts);
            reconnectAttempts++;
            System.out.println("Reconnecting in " + delay + "ms...");
            scheduler.schedule(() -> connect(), delay, TimeUnit.MILLISECONDS);
        }
    }

    public void disconnect() {
        if (session != null && session.isOpen()) {
            try { session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, "Normal closure")); }
            catch (Exception ignored) {}
        }
    }
}

// Usage
var client = new WebSocketClient("ws://localhost:3000/ws", "your-jwt-token");
client.connect().join();
client.send("{\"type\":\"chat\",\"recipientId\":\"user123\",\"text\":\"Hello!\"}");

# Python WebSocket Client
# pip install websockets

import asyncio
import json
import math
from urllib.parse import quote

import websockets
from websockets.exceptions import ConnectionClosed


class WebSocketClient:
    def __init__(self, url: str, token: str):
        self.url = url
        self.token = token
        self.ws = None
        self.reconnect_attempts = 0
        self.max_reconnect_attempts = 5
        self.reconnect_delay = 1.0

    async def connect(self):
        uri = f"{self.url}?token={quote(self.token)}"
        self.ws = await websockets.connect(uri)
        self.reconnect_attempts = 0
        print("Connected")
        asyncio.create_task(self._receive_loop())

    async def _receive_loop(self):
        try:
            async for raw in self.ws:
                message = json.loads(raw)
                self._handle_message(message)
        except ConnectionClosed:
            print("Disconnected")
            await self._attempt_reconnect()

    def _handle_message(self, message: dict):
        if message.get("type") == "chat":
            print(f"Message from {message['from']}: {message['text']}")
        else:
            print("Received:", message)

    async def send(self, message: dict):
        if self.ws and not self.ws.closed:
            await self.ws.send(json.dumps(message))
        else:
            print("WebSocket not connected")

    async def _attempt_reconnect(self):
        if self.reconnect_attempts < self.max_reconnect_attempts:
            delay = self.reconnect_delay * (2 ** self.reconnect_attempts)
            self.reconnect_attempts += 1
            print(f"Reconnecting in {delay:.0f}ms...")
            await asyncio.sleep(delay / 1000)
            await self.connect()

    async def disconnect(self):
        if self.ws:
            await self.ws.close(1000, "Normal closure")


# Usage
async def main():
    client = WebSocketClient("ws://localhost:3000/ws", "your-jwt-token")
    await client.connect()
    await client.send({"type": "chat", "recipientId": "user123", "text": "Hello!"})
    await asyncio.sleep(5)
    await client.disconnect()

asyncio.run(main())

// C# WebSocket Client (.NET 8)
using System.Net.WebSockets;
using System.Text;
using System.Text.Json;

public class WebSocketClient
{
    private ClientWebSocket? _ws;
    private readonly string _url;
    private readonly string _token;
    private int _reconnectAttempts = 0;
    private const int MaxReconnectAttempts = 5;
    private const int ReconnectDelayMs = 1000;

    public WebSocketClient(string url, string token)
    {
        _url = url;
        _token = token;
    }

    public async Task ConnectAsync()
    {
        _ws = new ClientWebSocket();
        await _ws.ConnectAsync(new Uri($"{_url}?token={Uri.EscapeDataString(_token)}"), CancellationToken.None);
        _reconnectAttempts = 0;
        Console.WriteLine("Connected");
        _ = ReceiveLoopAsync();
    }

    private async Task ReceiveLoopAsync()
    {
        var buffer = new byte[4096];
        try
        {
            while (_ws!.State == WebSocketState.Open)
            {
                var result = await _ws.ReceiveAsync(buffer, CancellationToken.None);
                if (result.MessageType == WebSocketMessageType.Close) break;

                var json = Encoding.UTF8.GetString(buffer, 0, result.Count);
                var message = JsonSerializer.Deserialize<JsonElement>(json);
                HandleMessage(message);
            }
        }
        catch (Exception ex)
        {
            Console.Error.WriteLine($"WebSocket error: {ex.Message}");
        }
        Console.WriteLine("Disconnected");
        await AttemptReconnectAsync();
    }

    private void HandleMessage(JsonElement message)
    {
        var type = message.GetProperty("type").GetString();
        if (type == "chat")
            Console.WriteLine($"Message from {message.GetProperty("from").GetString()}: {message.GetProperty("text").GetString()}");
        else
            Console.WriteLine($"Received: {message}");
    }

    public async Task SendAsync(object message)
    {
        if (_ws?.State == WebSocketState.Open)
        {
            var json = JsonSerializer.Serialize(message);
            await _ws.SendAsync(Encoding.UTF8.GetBytes(json), WebSocketMessageType.Text, true, CancellationToken.None);
        }
        else
        {
            Console.WriteLine("WebSocket not connected");
        }
    }

    private async Task AttemptReconnectAsync()
    {
        if (_reconnectAttempts < MaxReconnectAttempts)
        {
            var delay = ReconnectDelayMs * (int)Math.Pow(2, _reconnectAttempts);
            _reconnectAttempts++;
            Console.WriteLine($"Reconnecting in {delay}ms...");
            await Task.Delay(delay);
            await ConnectAsync();
        }
    }

    public async Task DisconnectAsync()
    {
        if (_ws != null)
            await _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "Normal closure", CancellationToken.None);
    }
}

// Usage
var client = new WebSocketClient("ws://localhost:3000/ws", "your-jwt-token");
await client.ConnectAsync();
await client.SendAsync(new { type = "chat", recipientId = "user123", text = "Hello!" });

import express from 'express';
import { verifyToken } from './auth';

const app = express();

interface SSEClient {
  userId: string;
  res: express.Response;
  lastEventId?: string;
}

const sseClients = new Map<string, Set<SSEClient>>();

// SSE endpoint
app.get('/events', (req, res) => {
  const token = req.query.token as string;

  if (!token) {
    res.status(401).json({ error: 'Token required' });
    return;
  }

  try {
    const decoded = verifyToken(token);
    const userId = decoded.sub;

    // Set SSE headers
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');
    res.setHeader('Access-Control-Allow-Origin', '*');

    // Support reconnection with Last-Event-ID
    const lastEventId = req.headers['last-event-id'] as string | undefined;

    const client: SSEClient = { userId, res, lastEventId };

    // Track client
    if (!sseClients.has(userId)) {
      sseClients.set(userId, new Set());
    }
    sseClients.get(userId)!.add(client);

    console.log(`SSE client connected: ${userId}`);

    // Send initial heartbeat
    res.write(': connected\n\n');

    // Handle client disconnect
    res.on('close', () => {
      const clients = sseClients.get(userId);
      if (clients) {
        clients.delete(client);
        if (clients.size === 0) {
          sseClients.delete(userId);
        }
      }
      console.log(`SSE client disconnected: ${userId}`);
    });

    res.on('error', (err) => {
      console.error(`SSE error for user ${userId}:`, err);
      res.end();
    });
  } catch (err) {
    res.status(401).json({ error: 'Invalid token' });
  }
});

// Broadcast function
function broadcastEventToUser(userId: string, eventType: string, data: any, eventId?: string) {
  const clients = sseClients.get(userId);
  if (clients) {
    clients.forEach((client) => {
      if (!client.res.closed) {
        client.res.write(`event: ${eventType}\n`);
        if (eventId) {
          client.res.write(`id: ${eventId}\n`);
        }
        client.res.write(`data: ${JSON.stringify(data)}\n\n`);
      }
    });
  }
}

// Example: Broadcast a notification
app.post('/notify', (req, res) => {
  const { userId, message } = req.body;
  broadcastEventToUser(userId, 'notification', message);
  res.json({ sent: true });
});

// Heartbeat to prevent proxy timeout
setInterval(() => {
  sseClients.forEach((clients) => {
    clients.forEach((client) => {
      if (!client.res.closed) {
        client.res.write(': heartbeat\n\n');
      }
    });
  });
}, 30000);

app.listen(3000, () => {
  console.log('SSE server on port 3000');
});

// Spring Boot SSE Server using SseEmitter
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@RestController
@RequestMapping
public class SseController {

    private final SseService sseService;

    public SseController(SseService sseService) {
        this.sseService = sseService;
    }

    @GetMapping("/events")
    public SseEmitter subscribe(@RequestParam String token) {
        String userId = validateToken(token); // implement JWT validation
        if (userId == null) throw new SecurityException("Invalid token");

        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        sseService.addClient(userId, emitter);

        emitter.onCompletion(() -> sseService.removeClient(userId, emitter));
        emitter.onTimeout(() -> sseService.removeClient(userId, emitter));
        emitter.onError(e -> sseService.removeClient(userId, emitter));

        try {
            emitter.send(SseEmitter.event().comment("connected"));
        } catch (IOException e) {
            emitter.completeWithError(e);
        }

        return emitter;
    }

    @PostMapping("/notify")
    public Map<String, Boolean> notify(@RequestBody Map<String, Object> body) {
        String userId = (String) body.get("userId");
        Object message = body.get("message");
        sseService.broadcastToUser(userId, "notification", message);
        return Map.of("sent", true);
    }

    private String validateToken(String token) { return null; /* implement */ }
}

@Service
class SseService {

    private final Map<String, Set<SseEmitter>> clients = new ConcurrentHashMap<>();

    public void addClient(String userId, SseEmitter emitter) {
        clients.computeIfAbsent(userId, k -> Collections.synchronizedSet(new HashSet<>())).add(emitter);
        System.out.println("SSE client connected: " + userId);
    }

    public void removeClient(String userId, SseEmitter emitter) {
        Set<SseEmitter> set = clients.get(userId);
        if (set != null) {
            set.remove(emitter);
            if (set.isEmpty()) clients.remove(userId);
        }
        System.out.println("SSE client disconnected: " + userId);
    }

    public void broadcastToUser(String userId, String eventType, Object data) {
        Set<SseEmitter> emitters = clients.getOrDefault(userId, Set.of());
        Set<SseEmitter> dead = new HashSet<>();
        for (SseEmitter emitter : emitters) {
            try {
                emitter.send(SseEmitter.event().name(eventType).data(data));
            } catch (IOException e) {
                dead.add(emitter);
            }
        }
        dead.forEach(e -> removeClient(userId, e));
    }

    @Scheduled(fixedDelay = 30000)
    public void sendHeartbeat() {
        clients.forEach((userId, emitters) -> {
            Set<SseEmitter> dead = new HashSet<>();
            for (SseEmitter emitter : emitters) {
                try {
                    emitter.send(SseEmitter.event().comment("heartbeat"));
                } catch (IOException e) {
                    dead.add(emitter);
                }
            }
            dead.forEach(e -> removeClient(userId, e));
        });
    }
}

# FastAPI SSE Server
# pip install fastapi uvicorn sse-starlette

import asyncio
import json
from collections import defaultdict
from typing import AsyncGenerator, Dict, Set

from fastapi import FastAPI, Query, Request
from fastapi.responses import JSONResponse
from jose import jwt, JWTError
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

sse_clients: Dict[str, Set[asyncio.Queue]] = defaultdict(set)


def verify_token(token: str) -> dict:
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])


@app.get("/events")
async def sse_endpoint(request: Request, token: str = Query(...)):
    try:
        payload = verify_token(token)
        user_id = payload["sub"]
    except JWTError:
        return JSONResponse({"error": "Invalid token"}, status_code=401)

    queue: asyncio.Queue = asyncio.Queue()
    sse_clients[user_id].add(queue)
    print(f"SSE client connected: {user_id}")

    async def event_generator() -> AsyncGenerator:
        yield {"event": "connected", "data": ""}
        try:
            while True:
                if await request.is_disconnected():
                    break
                try:
                    event = await asyncio.wait_for(queue.get(), timeout=30)
                    yield event
                except asyncio.TimeoutError:
                    # Heartbeat
                    yield {"comment": "heartbeat"}
        finally:
            sse_clients[user_id].discard(queue)
            if not sse_clients[user_id]:
                del sse_clients[user_id]
            print(f"SSE client disconnected: {user_id}")

    return EventSourceResponse(event_generator())


@app.post("/notify")
async def notify(body: dict):
    user_id = body["userId"]
    message = body["message"]
    await broadcast_to_user(user_id, "notification", message)
    return {"sent": True}


async def broadcast_to_user(user_id: str, event_type: str, data, event_id: str = None):
    queues = list(sse_clients.get(user_id, []))
    event = {"event": event_type, "data": json.dumps(data)}
    if event_id:
        event["id"] = event_id
    for queue in queues:
        await queue.put(event)

// ASP.NET Core SSE Server (.NET 8)
using System.Collections.Concurrent;
using System.Text;
using System.Text.Json;

var builder = WebApplication.CreateBuilder(args);
var app = builder.Build();

var sseClients = new ConcurrentDictionary<string, ConcurrentBag<HttpResponse>>();

app.MapGet("/events", async (HttpContext context, string token) =>
{
    var userId = ValidateToken(token);
    if (userId == null) { context.Response.StatusCode = 401; return; }

    context.Response.Headers["Content-Type"] = "text/event-stream";
    context.Response.Headers["Cache-Control"] = "no-cache";
    context.Response.Headers["Connection"] = "keep-alive";
    context.Response.Headers["Access-Control-Allow-Origin"] = "*";

    var res = context.Response;
    sseClients.GetOrAdd(userId, _ => new ConcurrentBag<HttpResponse>()).Add(res);
    Console.WriteLine($"SSE client connected: {userId}");

    await res.WriteAsync(": connected\n\n");
    await res.Body.FlushAsync();

    try
    {
        // Keep connection open until client disconnects
        var tcs = new TaskCompletionSource();
        context.RequestAborted.Register(() => tcs.TrySetResult());
        await tcs.Task;
    }
    finally
    {
        Console.WriteLine($"SSE client disconnected: {userId}");
    }
});

app.MapPost("/notify", async (HttpContext context) =>
{
    var body = await JsonSerializer.DeserializeAsync<JsonElement>(context.Request.Body);
    var userId = body.GetProperty("userId").GetString()!;
    var message = body.GetProperty("message");
    await BroadcastToUser(sseClients, userId, "notification", message.ToString());
    await context.Response.WriteAsJsonAsync(new { sent = true });
});

// Heartbeat background service
_ = Task.Run(async () =>
{
    while (true)
    {
        await Task.Delay(30000);
        foreach (var (userId, responses) in sseClients)
        {
            foreach (var res in responses)
            {
                try { await res.WriteAsync(": heartbeat\n\n"); await res.Body.FlushAsync(); }
                catch { /* client disconnected */ }
            }
        }
    }
});

static async Task BroadcastToUser(
    ConcurrentDictionary<string, ConcurrentBag<HttpResponse>> clients,
    string userId, string eventType, string data, string? eventId = null)
{
    if (!clients.TryGetValue(userId, out var responses)) return;
    var sb = new StringBuilder();
    sb.Append($"event: {eventType}\n");
    if (eventId != null) sb.Append($"id: {eventId}\n");
    sb.Append($"data: {data}\n\n");
    var payload = sb.ToString();
    foreach (var res in responses)
    {
        try { await res.WriteAsync(payload); await res.Body.FlushAsync(); }
        catch { /* client disconnected */ }
    }
}

static string? ValidateToken(string token) => null; // implement JWT validation

app.Run();

class SSEClient {
  private eventSource: EventSource | null = null;
  private url: string;
  private token: string;
  private reconnectAttempts = 0;

  constructor(url: string, token: string) {
    this.url = url;
    this.token = token;
  }

  connect() {
    const url = `${this.url}?token=${encodeURIComponent(this.token)}`;
    this.eventSource = new EventSource(url);

    this.eventSource.addEventListener('notification', (event) => {
      const data = JSON.parse(event.data);
      console.log('Notification:', data);
    });

    this.eventSource.addEventListener('message', (event) => {
      const data = JSON.parse(event.data);
      console.log('Message:', data);
    });

    this.eventSource.onerror = (err) => {
      console.error('EventSource error:', err);
      if (this.eventSource?.readyState === EventSource.CLOSED) {
        this.attemptReconnect();
      }
    };

    console.log('SSE connected');
  }

  private attemptReconnect() {
    this.reconnectAttempts++;
    const delay = 1000 * Math.pow(2, Math.min(this.reconnectAttempts - 1, 5));
    console.log(`Reconnecting in ${delay}ms...`);
    setTimeout(() => this.connect(), delay);
  }

  disconnect() {
    if (this.eventSource) {
      this.eventSource.close();
    }
  }
}

// Usage
const client = new SSEClient('http://localhost:3000/events', 'your-jwt-token');
client.connect();

// Java SSE Client using OkHttp
// Maven: com.squareup.okhttp3:okhttp

import okhttp3.*;
import okio.BufferedSource;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

public class SseClient {

    private final OkHttpClient httpClient;
    private final String url;
    private final String token;
    private int reconnectAttempts = 0;
    private final int maxReconnectAttempts = 5;

    public SseClient(String url, String token) {
        this.url = url;
        this.token = token;
        this.httpClient = new OkHttpClient.Builder()
                .readTimeout(0, TimeUnit.MILLISECONDS) // No timeout for streaming
                .build();
    }

    public void connect() {
        Request request = new Request.Builder()
                .url(url + "?token=" + token)
                .build();

        httpClient.newCall(request).enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                System.err.println("SSE connection failed: " + e.getMessage());
                attemptReconnect();
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                reconnectAttempts = 0;
                System.out.println("SSE connected");
                try (ResponseBody body = response.body()) {
                    if (body == null) return;
                    BufferedSource source = body.source();
                    String eventType = "";
                    while (!source.exhausted()) {
                        String line = source.readUtf8Line();
                        if (line == null) break;
                        if (line.startsWith("event:")) {
                            eventType = line.substring(6).trim();
                        } else if (line.startsWith("data:")) {
                            String data = line.substring(5).trim();
                            handleEvent(eventType, data);
                            eventType = "";
                        }
                    }
                }
                attemptReconnect();
            }
        });
    }

    private void handleEvent(String eventType, String data) {
        switch (eventType) {
            case "notification" -> System.out.println("Notification: " + data);
            case "message" -> System.out.println("Message: " + data);
            default -> System.out.println("Event [" + eventType + "]: " + data);
        }
    }

    private void attemptReconnect() {
        if (reconnectAttempts < maxReconnectAttempts) {
            long delay = 1000L * (1L << Math.min(reconnectAttempts, 5));
            reconnectAttempts++;
            System.out.println("Reconnecting in " + delay + "ms...");
            try { Thread.sleep(delay); } catch (InterruptedException ignored) {}
            connect();
        }
    }

    public void disconnect() {
        httpClient.dispatcher().cancelAll();
    }
}

// Usage
var client = new SseClient("http://localhost:3000/events", "your-jwt-token");
client.connect();

# Python SSE Client
# pip install httpx

import asyncio
import json
import math

import httpx


class SSEClient:
    def __init__(self, url: str, token: str):
        self.url = url
        self.token = token
        self.reconnect_attempts = 0

    async def connect(self):
        stream_url = f"{self.url}?token={self.token}"
        print("SSE connected")
        try:
            async with httpx.AsyncClient(timeout=None) as client:
                async with client.stream("GET", stream_url) as response:
                    self.reconnect_attempts = 0
                    event_type = ""
                    async for line in response.aiter_lines():
                        if line.startswith("event:"):
                            event_type = line[6:].strip()
                        elif line.startswith("data:"):
                            data = line[5:].strip()
                            self._handle_event(event_type, data)
                            event_type = ""
        except Exception as e:
            print(f"EventSource error: {e}")
            await self._attempt_reconnect()

    def _handle_event(self, event_type: str, data: str):
        parsed = json.loads(data) if data.startswith("{") else data
        if event_type == "notification":
            print("Notification:", parsed)
        elif event_type == "message":
            print("Message:", parsed)
        else:
            print(f"Event [{event_type}]:", parsed)

    async def _attempt_reconnect(self):
        delay = 1000 * (2 ** min(self.reconnect_attempts, 5)) / 1000
        self.reconnect_attempts += 1
        print(f"Reconnecting in {delay:.0f}s...")
        await asyncio.sleep(delay)
        await self.connect()


# Usage
async def main():
    client = SSEClient("http://localhost:3000/events", "your-jwt-token")
    await client.connect()

asyncio.run(main())

// C# SSE Client (.NET 8)
using System.Net.Http.Headers;
using System.Text.Json;

public class SseClient
{
    private readonly HttpClient _httpClient;
    private readonly string _url;
    private readonly string _token;
    private int _reconnectAttempts = 0;
    private const int MaxReconnectAttempts = 5;

    public SseClient(string url, string token)
    {
        _url = url;
        _token = token;
        _httpClient = new HttpClient { Timeout = System.Threading.Timeout.InfiniteTimeSpan };
    }

    public async Task ConnectAsync(CancellationToken cancellationToken = default)
    {
        var uri = $"{_url}?token={Uri.EscapeDataString(_token)}";
        Console.WriteLine("SSE connected");

        try
        {
            using var response = await _httpClient.GetAsync(uri, HttpCompletionOption.ResponseHeadersRead, cancellationToken);
            response.EnsureSuccessStatusCode();
            _reconnectAttempts = 0;

            using var stream = await response.Content.ReadAsStreamAsync(cancellationToken);
            using var reader = new StreamReader(stream);

            string eventType = "";
            while (!reader.EndOfStream && !cancellationToken.IsCancellationRequested)
            {
                var line = await reader.ReadLineAsync(cancellationToken);
                if (line == null) break;

                if (line.StartsWith("event:"))
                    eventType = line[6..].Trim();
                else if (line.StartsWith("data:"))
                {
                    var data = line[5..].Trim();
                    HandleEvent(eventType, data);
                    eventType = "";
                }
            }
        }
        catch (Exception ex)
        {
            Console.Error.WriteLine($"EventSource error: {ex.Message}");
            await AttemptReconnectAsync(cancellationToken);
        }
    }

    private void HandleEvent(string eventType, string data)
    {
        switch (eventType)
        {
            case "notification":
                Console.WriteLine($"Notification: {data}");
                break;
            case "message":
                Console.WriteLine($"Message: {data}");
                break;
            default:
                Console.WriteLine($"Event [{eventType}]: {data}");
                break;
        }
    }

    private async Task AttemptReconnectAsync(CancellationToken cancellationToken)
    {
        if (_reconnectAttempts < MaxReconnectAttempts)
        {
            var delay = 1000 * (int)Math.Pow(2, Math.Min(_reconnectAttempts, 5));
            _reconnectAttempts++;
            Console.WriteLine($"Reconnecting in {delay}ms...");
            await Task.Delay(delay, cancellationToken);
            await ConnectAsync(cancellationToken);
        }
    }
}

// Usage
var client = new SseClient("http://localhost:3000/events", "your-jwt-token");
await client.ConnectAsync();