Server-Sent Events (SSE) в Node.js

data: Текст сообщения
 
id: 42
event: update
data: {"temp": 22, "humidity": 60}

const eventSource = new EventSource('/events');
eventSource.onmessage = function(event) {
  console.log('Получено событие:', event.data);
};

// Сервер отправляет события с ID
res.write(`id: ${eventId}\n`);
res.write(`data: ${JSON.stringify(eventData)}\n\n`);

app.get('/events', (req, res) => {
  const lastEventId = req.headers['last-event-id'] || '0';
  // Отправляем все события с ID > lastEventId
  const missedEvents = getEventsSince(lastEventId);
  
  missedEvents.forEach(event => {
    res.write(`id: ${event.id}\n`);
    res.write(`data: ${JSON.stringify(event.data)}\n\n`);
  });
  
  // Продолжаем отправку новых событий...
});

// Сервер указывает интервал переподключения в миллисекундах
res.write(`retry: 5000\n`);
res.write(`data: Переподключение произойдет через 5 секунд\n\n`);

// Пинг для поддержания соединения
setInterval(() => {
  res.write(`: ping\n\n`);  // События, начинающиеся с двоеточия, игнорируются клиентом
}, 30000);

const http = require('http');
 
const server = http.createServer((req, res) => {
  if (req.url === '/events') {
    // Устанавливаем необходимые заголовки для SSE
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive'
    });
    
    // Отправляем сообщение о подключении
    res.write(`data: Соединение установлено\n\n`);
    
    // Периодически отправляем сообщения
    const intervalId = setInterval(() => {
      const data = JSON.stringify({ time: new Date().toISOString() });
      res.write(`data: ${data}\n\n`);
    }, 1000);
    
    // Отслеживаем закрытие соединения
    req.on('close', () => {
      clearInterval(intervalId);
      console.log('Клиент отключился');
    });
  } else {
    res.writeHead(404);
    res.end();
  }
});
 
server.listen(3000, () => {
  console.log('SSE-сервер запущен на порту 3000');
});

class SSEManager {
  constructor() {
    this.clients = new Map();
    this.eventHistory = new Map();
    this.historySize = 100;  // Сколько событий храним для повторной отправки
  }
  
  addClient(id, res, channel = 'default') {
    if (!this.clients.has(channel)) {
      this.clients.set(channel, new Map());
    }
    
    this.clients.get(channel).set(id, res);
    
    // Отправляем пропущенные события, если они есть
    const history = this.eventHistory.get(channel) || [];
    history.forEach(event => {
      res.write(event);
    });
    
    return () => {
      if (this.clients.has(channel)) {
        this.clients.get(channel).delete(id);
      }
    };
  }
  
  broadcast(data, eventType = null, channel = 'default') {
    if (!this.clients.has(channel)) return 0;
    
    const clients = this.clients.get(channel);
    const eventId = Date.now();
    let message = `id: ${eventId}\n`;
    
    if (eventType) {
      message += `event: ${eventType}\n`;
    }
    
    if (typeof data === 'object') {
      message += `data: ${JSON.stringify(data)}\n\n`;
    } else {
      message += `data: ${data}\n\n`;
    }
    
    // Сохраняем событие в истории
    if (!this.eventHistory.has(channel)) {
      this.eventHistory.set(channel, []);
    }
    
    const history = this.eventHistory.get(channel);
    history.push(message);
    
    // Ограничиваем размер истории
    while (history.length > this.historySize) {
      history.shift();
    }
    
    // Отправляем всем подключенным клиентам
    let sentCount = 0;
    for (const client of clients.values()) {
      try {
        client.write(message);
        sentCount++;
      } catch (error) {
        console.error('Ошибка отправки события:', error);
      }
    }
    
    return sentCount;
  }
}

const http = require('http');
const crypto = require('crypto');
const sseManager = new SSEManager();
 
const server = http.createServer((req, res) => {
  if (req.url.startsWith('/events')) {
    // Парсим параметры запроса
    const url = new URL(req.url, [INLINE]http://${req.headers.host}[/INLINE]);
    const channel = url.searchParams.get('channel') || 'default';
    
    // Устанавливаем заголовки
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
      'Access-Control-Allow-Origin': '*'  // В реальном проекте настройте более безопасно
    });
    
    // Генерируем уникальный ID для клиента
    const clientId = crypto.randomUUID();
    
    // Регистрируем клиента и получаем функцию для отписки
    const cleanup = sseManager.addClient(clientId, res, channel);
    
    // Отправляем сообщение о подключении
    res.write(`data: Подключено к каналу ${channel}\n\n`);
    
    // Настраиваем пинг для поддержания соединения
    const pingInterval = setInterval(() => {
      res.write(`: ping\n\n`);
    }, 30000);
    
    // Отслеживаем закрытие соединения
    req.on('close', () => {
      clearInterval(pingInterval);
      cleanup();
      console.log(`Клиент ${clientId} отключился от канала ${channel}`);
    });
  } else if (req.url.startsWith('/publish') && req.method === 'POST') {
    // Эндпоинт для публикации событий
    let body = '';
    
    req.on('data', chunk => {
      body += chunk.toString();
    });
    
    req.on('end', () => {
      try {
        const { channel = 'default', event, data } = JSON.parse(body);
        const sent = sseManager.broadcast(data, event, channel);
        
        res.writeHead(200, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ success: true, sent }));
      } catch (error) {
        res.writeHead(400, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ success: false, error: error.message }));
      }
    });
  } else {
    res.writeHead(404);
    res.end('Not Found');
  }
});

// Добавляем отслеживание времени жизни соединения
addClient(id, res, channel = 'default') {
  const client = {
    res,
    connectedAt: Date.now(),
    lastActivity: Date.now()
  };
  
  if (!this.clients.has(channel)) {
    this.clients.set(channel, new Map());
  }
  
  this.clients.get(channel).set(id, client);
  
  // Отправляем пропущенные события...
  
  return () => {
    if (this.clients.has(channel)) {
      this.clients.get(channel).delete(id);
    }
  };
}
 
// Периодически проверяем "зависшие" соединения
monitorConnections(maxIdleTime = 3600000) {  // 1 час по умолчанию
  return setInterval(() => {
    const now = Date.now();
    let closedConnections = 0;
    
    for (const [channel, clients] of this.clients.entries()) {
      for (const [clientId, client] of clients.entries()) {
        if (now - client.lastActivity > maxIdleTime) {
          try {
            client.res.end();  // Принудительно закрываем "зависшее" соединение
            clients.delete(clientId);
            closedConnections++;
          } catch (err) {
            console.error(`Ошибка при закрытии соединения: ${err.message}`);
          }
        }
      }
    }
    
    if (closedConnections > 0) {
      console.log(`Закрыто ${closedConnections} неактивных соединений`);
    }
  }, 300000);  // Проверка каждые 5 минут
}

// Ограничиваем количество соединений с одного IP
const clientsPerIp = new Map();
const MAX_CLIENTS_PER_IP = 10;
 
server.on('request', (req, res) => {
  if (req.url.startsWith('/events')) {
    const ip = req.headers['x-forwarded-for'] || req.socket.remoteAddress;
    
    if (!clientsPerIp.has(ip)) {
      clientsPerIp.set(ip, 0);
    }
    
    const currentCount = clientsPerIp.get(ip);
    if (currentCount >= MAX_CLIENTS_PER_IP) {
      res.writeHead(429, { 'Content-Type': 'text/plain' });
      res.end('Too Many Connections');
      return;
    }
    
    clientsPerIp.set(ip, currentCount + 1);
    
    // Уменьшаем счетчик при отключении
    req.on('close', () => {
      const newCount = clientsPerIp.get(ip) - 1;
      if (newCount <= 0) {
        clientsPerIp.delete(ip);
      } else {
        clientsPerIp.set(ip, newCount);
      }
    });
  }
  
  // Продолжаем обработку запроса...
});

const jwt = require('jsonwebtoken');
const JWT_SECRET = process.env.JWT_SECRET || 'ваш_секретный_ключ';
 
function verifyToken(req) {
  // Получаем токен из query-параметра или cookie
  const token = req.query.token || req.cookies.token;
  if (!token) return null;
  
  try {
    return jwt.verify(token, JWT_SECRET);
  } catch (err) {
    return null;
  }
}
 
// В обработчике запроса
if (req.url.startsWith('/events')) {
  const user = verifyToken(req);
  if (!user) {
    res.writeHead(401, { 'Content-Type': 'text/plain' });
    res.end('Unauthorized');
    return;
  }
  
  // Используем ID пользователя как part канала для персональных уведомлений
  const userChannel = `user-${user.id}`;
  
  // Продолжаем обработку запроса...
}

class ConnectionManager {
  constructor(redisClient) {
    this.connections = new Map();
    this.redis = redisClient;
  }
  
  async addConnection(userId, connectionId, res) {
    // Сохраняем соединение в памяти
    this.connections.set(connectionId, { userId, res, timestamp: Date.now() });
    
    // Сохраняем состояние в Redis
    await this.redis.sadd(`user:${userId}:connections`, connectionId);
    await this.redis.set(`connection:${connectionId}:info`, JSON.stringify({
      userId,
      userAgent: req.headers['user-agent'],
      connectedAt: Date.now()
    }));
    
    return connectionId;
  }
  
  async removeConnection(connectionId) {
    const connection = this.connections.get(connectionId);
    if (!connection) return false;
    
    // Удаляем из памяти
    this.connections.delete(connectionId);
    
    // Удаляем из Redis
    await this.redis.srem(`user:${connection.userId}:connections`, connectionId);
    await this.redis.del(`connection:${connectionId}:info`);
    
    return true;
  }
}

function setupHeartbeat(res, interval = 30000) {
  return setInterval(() => {
    res.write(`: heartbeat ${Date.now()}\n\n`);
  }, interval);
}

// На сервере
const pendingPings = new Map();
 
function sendPing(clientId, res) {
  const pingId = Date.now();
  pendingPings.set(`${clientId}:${pingId}`, { timestamp: Date.now() });
  res.write(`event: ping\ndata: ${pingId}\n\n`);
  
  // Если через 30 секунд нет ответа, считаем соединение мертвым
  setTimeout(() => {
    if (pendingPings.has(`${clientId}:${pingId}`)) {
      pendingPings.delete(`${clientId}:${pingId}`);
      disconnectClient(clientId);
    }
  }, 30000);
}

eventSource.addEventListener('ping', function(e) {
  fetch('/ping-response', {
    method: 'POST',
    body: JSON.stringify({ pingId: e.data }),
    headers: { 'Content-Type': 'application/json' }
  });
});

function sseMiddleware() {
return (req, res, next) => {
  res.sseSetup = function() {
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive'
    });
  };
 
  res.sseSend = function(data, event, id) {
    let payload = '';
    
    if (id) payload += `id: ${id}\n`;
    if (event) payload += `event: ${event}\n`;
    
    if (typeof data === 'object') {
      payload += `data: ${JSON.stringify(data)}\n\n`;
    } else {
      payload += `data: ${data}\n\n`;
    }
    
    res.write(payload);
  };
  
  next();
};
}

const express = require('express');
const app = express();
 
app.use(sseMiddleware());
 
app.get('/events', (req, res) => {
// Настраиваем SSE-соединение
res.sseSetup();
 
// Отправляем приветственное сообщение
res.sseSend('Соединение установлено');
 
// Настраиваем периодическую отправку данных
const interval = setInterval(() => {
  res.sseSend({ time: new Date().toISOString() }, 'time-update');
}, 1000);
 
// Обрабатываем закрытие соединения
req.on('close', () => {
  clearInterval(interval);
});
});

class SSERouter {
constructor() {
  this.channels = new Map();
  this.clients = new Map();
}
 
channel(name, handler) {
  this.channels.set(name, handler);
  return this;
}
 
middleware() {
  return (req, res, next) => {
    const channelName = req.params.channel || req.query.channel;
    
    if (!channelName || !this.channels.has(channelName)) {
      return next();
    }
    
    const clientId = req.query.clientId || Date.now().toString();
    const handler = this.channels.get(channelName);
    
    res.sseSetup();
    
    // Регистрируем клиента
    if (!this.clients.has(channelName)) {
      this.clients.set(channelName, new Map());
    }
    
    this.clients.get(channelName).set(clientId, res);
    
    // Запускаем обработчик для этого канала
    handler(req, res, clientId);
    
    // Убираем клиента при отключении
    req.on('close', () => {
      if (this.clients.has(channelName)) {
        this.clients.get(channelName).delete(clientId);
      }
    });
  };
}
}

const sseRouter = new SSERouter();
 
// Определяем каналы событий
sseRouter.channel('news', (req, res, clientId) => {
// Логика работы с каналом новостей
// ...
});
 
sseRouter.channel('stats', (req, res, clientId) => {
// Логика работы с каналом статистики
// ...
});
 
// Подключаем роутер к приложению
app.get('/sse/:channel', sseMiddleware(), sseRouter.middleware());

// Клиент
const eventSource = new EventSource('/events?token=eyJhbGciOiJ...');
 
// Сервер
app.get('/events', (req, res) => {
const token = req.query.token;
if (!validateToken(token)) {
  return res.status(401).end('Unauthorized');
}
// Продолжаем обработку запроса...
});

// Аутентифицируем пользователя и устанавливаем куки
app.post('/login', (req, res) => {
const user = authenticateUser(req.body);
if (user) {
  res.cookie('auth_token', generateToken(user), {
    httpOnly: true,
    secure: true,
    sameSite: 'strict'
  });
  res.json({ success: true });
}
});
 
// Проверка при установлении SSE-соединения
app.get('/events', (req, res) => {
const token = req.cookies.auth_token;
if (!validateToken(token)) {
  return res.status(401).end('Unauthorized');
}
// Продолжаем обработку запроса...
});

function authorizeChannel(user, channel) {
// Проверяем права пользователя на конкретный канал
const userChannels = getUserPermissions(user.id);
return userChannels.includes(channel);
}
 
app.get('/events/:channel', (req, res) => {
const user = verifyToken(req.cookies.auth_token);
if (!user) {
  return res.status(401).end('Unauthorized');
}
 
const channel = req.params.channel;
if (!authorizeChannel(user, channel)) {
  return res.status(403).end('Forbidden');
}
 
// Пользователь авторизован для этого канала
setupSSE(req, res, channel, user);
});

// На клиенте
function logout() {
// Закрываем SSE-соединение
if (eventSource) {
  eventSource.close();
  eventSource = null;
}
// Выполняем выход
fetch('/logout', { method: 'POST' });
}

class AnalyticsEnabledSSE {
constructor(analyticsService) {
  this.analytics = analyticsService;
}
 
handleConnection(req, res, userId) {
  const connectionId = crypto.randomUUID();
  const metadata = {
    userId,
    userAgent: req.headers['user-agent'],
    ip: req.ip,
    referrer: req.headers.referer,
    timestamp: Date.now()
  };
  
  // Отправляем событие подключения в аналитику
  this.analytics.trackEvent('sse_connection_opened', {
    connectionId,
    ...metadata
  });
  
  // Отслеживаем отключение
  req.on('close', () => {
    const duration = Date.now() - metadata.timestamp;
    this.analytics.trackEvent('sse_connection_closed', {
      connectionId,
      duration,
      timestamp: Date.now()
    });
  });
  
  return connectionId;
}
}

// Отправляем уведомление через SSE
function sendNotification(userId, message) {
const eventId = crypto.randomUUID();
const timestamp = Date.now();
 
// Записываем в аналитику факт отправки
analytics.trackEvent('notification_sent', {
  eventId,
  userId,
  timestamp
});
 
// Отправляем через SSE с идентификатором события
sse.send(userId, {
  type: 'notification',
  message,
  eventId  // Включаем ID в само сообщение
});
}
 
// На клиенте отслеживаем реакцию
eventSource.addEventListener('notification', function(e) {
const data = JSON.parse(e.data);
// Отправляем аналитику о получении
fetch('/analytics/event', {
  method: 'POST',
  body: JSON.stringify({
    type: 'notification_received',
    eventId: data.eventId,
    timestamp: Date.now()
  })
});
 
// При клике по уведомлению
notificationElement.addEventListener('click', () => {
  fetch('/analytics/event', {
    method: 'POST',
    body: JSON.stringify({
      type: 'notification_clicked',
      eventId: data.eventId,
      timestamp: Date.now()
    })
  });
});
});

res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Cache-Control', 'no-cache');
res.setHeader('Connection', 'keep-alive');
res.setHeader('Access-Control-Allow-Origin', 'https://ваш-домен.ru');
res.setHeader('Access-Control-Allow-Credentials', 'true');

const cors = require('cors');
 
// Настройка CORS для SSE-эндпоинта
const sseOptions = {
  origin: 'https://ваш-домен.ru',  // или массив доменов
  credentials: true,
  methods: 'GET',  // Для SSE используется только GET
};
 
app.get('/events', cors(sseOptions), (req, res) => {
  // Настройка SSE...
});

// Периодически отправляем метрики балансировщику
setInterval(() => {
  const metrics = {
    connections: countActiveSSEConnections(),
    cpuUsage: process.cpuUsage(),
    memoryUsage: process.memoryUsage(),
    serverId: process.env.SERVER_ID
  };
  
  fetch('http://load-balancer/metrics', {
    method: 'POST',
    body: JSON.stringify(metrics),
    headers: { 'Content-Type': 'application/json' }
  });
}, 5000);

http {
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
    proxy_connect_timeout 300s;
}

proxy_set_header Connection "";
proxy_buffering off;

const redis = require('redis');
const publisher = redis.createClient();
const subscriber = redis.createClient();
 
// Класс для управления SSE в кластере
class ClusteredSSE {
constructor() {
  this.connections = new Map();
  this.setupSubscriber();
}
 
setupSubscriber() {
  subscriber.subscribe('sse-events');
  subscriber.on('message', (channel, message) => {
    try {
      const event = JSON.parse(message);
      this.broadcastToLocalClients(event);
    } catch (err) {
      console.error('Ошибка при обработке сообщения:', err);
    }
  });
}
 
broadcastToLocalClients({ channel, data, eventType, eventId }) {
  // Отправляем только тем клиентам, которые подключены к ЭТОМУ экземпляру
  const channelClients = this.connections.get(channel) || new Map();
  
  for (const [clientId, res] of channelClients.entries()) {
    try {
      let message = '';
      if (eventId) message += `id: ${eventId}\n`;
      if (eventType) message += `event: ${eventType}\n`;
      message += `data: ${typeof data === 'object' ? JSON.stringify(data) : data}\n\n`;
      
      res.write(message);
    } catch (err) {
      console.error(`Ошибка отправки клиенту ${clientId}:`, err);
    }
  }
}
 
// Публикация события для ВСЕХ серверов в кластере
publishEvent(channel, data, eventType, eventId) {
  const event = { channel, data, eventType, eventId };
  publisher.publish('sse-events', JSON.stringify(event));
}
 
// Регистрация локального подключения
addConnection(clientId, channel, res) {
  if (!this.connections.has(channel)) {
    this.connections.set(channel, new Map());
  }
  this.connections.get(channel).set(clientId, res);
}
}

// Мониторинг потребления памяти
function trackMemoryUsage() {
  const memUsage = process.memoryUsage();
  console.log(`Использование памяти: ${Math.round(memUsage.heapUsed / 1024 / 1024)} МБ`);
  
  if (memUsage.heapUsed > MEM_THRESHOLD) {
    console.log('Превышен порог использования памяти, перезапуск...');
    // Сохраняем состояние и инициируем плавный перезапуск
  }
}

// Буферизация событий
const eventBuffer = new Map();
 
function scheduleEvent(channel, data) {
  if (!eventBuffer.has(channel)) {
    eventBuffer.set(channel, []);
    
    setTimeout(() => {
      const events = eventBuffer.get(channel);
      eventBuffer.delete(channel);
      if (events.length > 0) {
        sendBatchEvents(channel, events);
      }
    }, 50);
  }
  
  eventBuffer.get(channel).push(data);
}

// Включение сжатия в Express
const compression = require('compression');
app.use(compression({
  filter: (req) => req.url.startsWith('/events')
}));

function createNotification(userId, type, data) {
  const notification = {
    id: crypto.randomUUID(),
    userId,
    type,
    data,
    created: Date.now(),
    read: false
  };
  
  // Сохраняем в БД
  db.notifications.insert(notification);
  
  // Публикуем событие через Redis для кластеризованной архитектуры
  redisClient.publish('notifications', JSON.stringify(notification));
  
  return notification;
}

subscriber.on('message', (channel, message) => {
  const notification = JSON.parse(message);
  const userConnections = clients.get(notification.userId) || [];
  
  // Отправляем через SSE только активным соединениям этого пользователя
  userConnections.forEach(client => {
    client.res.sseSend(notification, 'notification', notification.id);
  });
});

eventSource.addEventListener('notification', function(e) {
  const notification = JSON.parse(e.data);
  showPopup(notification.data.title, notification.data.message);
  
  // Обновляем счетчик непрочитанных
  updateUnreadCounter();
});

// Серверная часть - отслеживание активности
const userActivities = new Map();
 
app.post('/activity', (req, res) => {
const { userId, status, typing } = req.body;
userActivities.set(userId, { status, typing, timestamp: Date.now() });
 
// Распространяем статус всем пользователям
broadcastUserActivity(userId);
res.send({ success: true });
});
 
function broadcastUserActivity(userId) {
const activity = userActivities.get(userId);
if (!activity) return;
 
const event = {
  type: 'user_activity',
  userId,
  status: activity.status,
  typing: activity.typing
};
 
// Отправляем через SSE
sseManager.broadcast(event, 'activity');
}

eventSource.addEventListener('activity', function(e) {
const data = JSON.parse(e.data);
if (data.typing) {
  showTypingIndicator(data.userId);
} else {
  hideTypingIndicator(data.userId);
}
 
updateUserStatus(data.userId, data.status);
});
 
// При вводе сообщения отправляем статус
messageInput.addEventListener('keydown', debounce(() => {
fetch('/activity', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ userId, typing: true })
});
}, 300));

function collectMetrics() {
  return {
    memory: process.memoryUsage(),
    cpu: process.cpuUsage(),
    uptime: process.uptime(),
    activeConnections: getActiveConnectionsCount(),
    queueSize: getTaskQueueSize(),
    timestamp: Date.now()
  };
}
 
// Отправка метрик через SSE каждую секунду
const metricsInterval = setInterval(() => {
  const metrics = collectMetrics();
  sseManager.broadcast(metrics, 'metrics', 'system');
}, 1000);

const memoryChart = new Chart(ctx, {
  type: 'line',
  data: {
    labels: [],
    datasets: [{
      label: 'Heap Used (MB)',
      data: []
    }]
  }
});
 
eventSource.addEventListener('metrics', function(e) {
  const metrics = JSON.parse(e.data);
  const heapUsedMB = Math.round(metrics.memory.heapUsed / 1024 / 1024);
  
  // Добавляем новую точку на график
  memoryChart.data.labels.push(new Date().toLocaleTimeString());
  memoryChart.data.datasets[0].data.push(heapUsedMB);
  
  // Убираем старые точки для экономии памяти
  if (memoryChart.data.labels.length > 50) {
    memoryChart.data.labels.shift();
    memoryChart.data.datasets[0].data.shift();
  }
  
  memoryChart.update();
});

async function streamLargeDataset(req, res) {
// Настраиваем SSE
res.writeHead(200, {
  'Content-Type': 'text/event-stream',
  'Cache-Control': 'no-cache',
  'Connection': 'keep-alive'
});
 
// Получаем курсор из базы данных
const cursor = db.collection('analytics')
  .find(req.query.filter)
  .sort(req.query.sort)
  .batchSize(100);
 
let count = 0;
let batch = [];
 
// Итерируемся по результатам
while (await cursor.hasNext()) {
  const doc = await cursor.next();
  batch.push(doc);
  count++;
  
  // Отправляем пакетами по 50 документов
  if (batch.length >= 50) {
    res.write(`data: ${JSON.stringify({
      type: 'batch',
      items: batch,
      progress: count,
      done: false
    })}
 
`);
    
    batch = [];
    // Небольшая пауза для обработки на клиенте
    await new Promise(resolve => setTimeout(resolve, 10));
  }
}
 
// Отправляем последний пакет
if (batch.length > 0) {
  res.write(`data: ${JSON.stringify({
    type: 'batch',
    items: batch,
    progress: count,
    done: true
  })}
 
`);
}
 
// Сигнализируем о завершении
res.write(`data: ${JSON.stringify({
  type: 'complete',
  totalCount: count
})}
 
`);
}

eventSource.addEventListener('message', function(e) {
const data = JSON.parse(e.data);
 
if (data.type === 'batch') {
  // Добавляем строки в таблицу
  data.items.forEach(item => {
    addRowToTable(item);
  });
  
  // Обновляем индикатор прогресса
  updateProgressBar(data.progress, data.totalCount);
} else if (data.type === 'complete') {
  // Завершаем загрузку
  hideProgressBar();
  showCompletionMessage(data.totalCount);
}
});