iistwin/server/routes/auth.sse.routes.ts

import { Router } from "express";
import crypto from 'crypto';
import { storage } from "../storage";
import { authenticateToken, type AuthenticatedRequest } from "../middleware/auth.middleware";
import { sseLimiter, eventBus, type SSEConnection } from "./shared";
import { withTenant } from "../db";

interface SseTokenEntry {
  userId: number;
  organizationId: number;
  expiresAt: number;
}

const sseTokens = new Map<string, SseTokenEntry>();
const SSE_TOKEN_TTL_MS = 30_000;

setInterval(() => {
  const now = Date.now();
  for (const [t, entry] of Array.from(sseTokens.entries())) {
    if (entry.expiresAt < now) sseTokens.delete(t);
  }
}, 60_000).unref();

export function registerSseRoutes(router: Router): void {
  // POST /api/events/token — exchange long-lived JWT for a one-time SSE token
  router.post('/api/events/token',
    authenticateToken,
    (req: AuthenticatedRequest, res) => {
      const token = crypto.randomUUID();
      sseTokens.set(token, {
        userId: req.user!.id,
        organizationId: req.organizationId!,
        expiresAt: Date.now() + SSE_TOKEN_TTL_MS,
      });
      res.json({ token });
    }
  );

  // GET /api/events — SSE stream, authenticated via one-time token
  router.get('/api/events',
    sseLimiter,
    async (req, res) => {
      try {
        const rawToken = req.query.token as string;
        if (!rawToken) {
          return res.status(401).json({ error: 'Токен не предоставлен' });
        }

        const now = Date.now();
        const entry = sseTokens.get(rawToken);
        sseTokens.delete(rawToken);

        if (!entry || entry.expiresAt < now) {
          return res.status(401).json({ error: 'Недействительный или просроченный SSE-токен' });
        }

        const user = await withTenant(entry.organizationId, () =>
          storage.getUser(entry.userId)
        );
        if (!user || !user.isActive) {
          return res.status(401).json({ error: 'Пользователь не найден или неактивен' });
        }

        res.writeHead(200, {
          'Content-Type': 'text/event-stream',
          'Cache-Control': 'no-cache',
          'Connection': 'keep-alive',
          'Access-Control-Allow-Origin': '*',
          'Access-Control-Allow-Headers': 'Cache-Control'
        });

        const connectionId = `${Date.now()}-${Math.random()}`;
        const connection: SSEConnection = {
          id: connectionId,
          userId: user.id,
          organizationId: user.organizationId,
          res
        };

        eventBus.addConnection(connection);
        res.write(`data: {"type":"connected","userId":${user.id},"organizationId":${user.organizationId}}\n\n`);

        req.on('close', () => {
          eventBus.removeConnection(connectionId);
        });

        req.on('error', (error) => {
          console.error(`SSE connection error ${connectionId}:`, error);
          eventBus.removeConnection(connectionId);
        });
      } catch (error) {
        console.error('SSE endpoint error:', error);
        res.status(500).json({ error: 'Ошибка подключения к событиям' });
      }
    }
  );
}