event: messageCreated
id: 42
data: {"userId": 10, "text": "Привіт, SSE!"}
retry: 5000

event: userJoined
id: 43
data: {"userId": 15, "username": "Andriy"}

: це коментар (heartbeat)

data: {"metric": "cpu", "value": 45}

// src/events/events.controller.ts
import { Controller, Sse } from '@nestjs/common';
import { Observable, interval, map } from 'rxjs';

interface MessageEvent {
  data: string | object;
  id?: string;
  type?: string;
  retry?: number;
}

@Controller('events')
export class EventsController {
  /**
   * SSE endpoint: відправляє поточний час кожні 2 секунди
   */
  @Sse('stream')
  streamEvents(): Observable<MessageEvent> {
    return interval(2000).pipe(
      map((index) => ({
        data: {
          message: 'Поточний час',
          timestamp: new Date().toISOString(),
          index,
        },
        id: index.toString(),
      })),
    );
  }
}

// client/sse-client.ts
class SSEClient {
  private eventSource: EventSource | null = null;

  constructor(private url: string) {}

  /**
   * Підключитися до SSE потоку
   */
  connect() {
    this.eventSource = new EventSource(this.url);

    // Обробник успішного підключення
    this.eventSource.addEventListener('open', () => {
      console.log('[SSE] З\'єднання встановлено');
    });

    // Обробник для події типу 'message' (за замовчуванням)
    this.eventSource.addEventListener('message', (event: MessageEvent) => {
      const data = JSON.parse(event.data);
      console.log('[SSE] Отримано подію:', data);
      console.log('[SSE] Event ID:', event.lastEventId);
      
      // Відобразити дані у UI
      this.displayData(data);
    });

    // Обробник помилок та розривів з'єднання
    this.eventSource.addEventListener('error', (error) => {
      console.error('[SSE] Помилка з\'єднання:', error);

      if (this.eventSource?.readyState === EventSource.CLOSED) {
        console.log('[SSE] З\'єднання закрито сервером');
      } else if (this.eventSource?.readyState === EventSource.CONNECTING) {
        console.log('[SSE] Спроба автоматичного reconnect...');
      }
    });
  }

  /**
   * Закрити SSE з'єднання
   */
  disconnect() {
    if (this.eventSource) {
      this.eventSource.close();
      this.eventSource = null;
      console.log('[SSE] З\'єднання закрито клієнтом');
    }
  }

  private displayData(data: any) {
    const container = document.getElementById('sse-data');
    if (container) {
      const item = document.createElement('div');
      item.textContent = `${data.message}: ${data.timestamp}`;
      container.appendChild(item);
    }
  }
}

// Використання
const client = new SSEClient('http://localhost:3001/events/stream');
client.connect();

// Закрити при виході зі сторінки
window.addEventListener('beforeunload', () => {
  client.disconnect();
});

// src/monitoring/monitoring.controller.ts
import { Controller, Sse } from '@nestjs/common';
import { Observable, interval, map } from 'rxjs';
import { MonitoringService } from './monitoring.service';

@Controller('monitoring')
export class MonitoringController {
  constructor(private readonly monitoringService: MonitoringService) {}

  /**
   * Відправка метрик системи кожні 5 секунд
   */
  @Sse('metrics')
  streamMetrics(): Observable<MessageEvent> {
    return interval(5000).pipe(
      map(async () => {
        // Отримати поточні метрики з сервісу
        const metrics = await this.monitoringService.getCurrentMetrics();

        return {
          data: {
            cpu: metrics.cpuUsage,
            memory: metrics.memoryUsage,
            activeConnections: metrics.activeConnections,
            requestsPerSecond: metrics.rps,
          },
          id: Date.now().toString(),
        };
      }),
    );
  }
}

const eventSource = new EventSource('http://localhost:3001/monitoring/metrics');

eventSource.addEventListener('message', (event) => {
  const metrics = JSON.parse(event.data);

  // Оновити графік у dashboard
  updateChart('cpu', metrics.cpu);
  updateChart('memory', metrics.memory);
  
  // Оновити лічильники
  document.getElementById('active-connections')!.textContent = metrics.activeConnections;
  document.getElementById('rps')!.textContent = metrics.requestsPerSecond;
});

// src/notifications/notifications.controller.ts
import { Controller, Sse } from '@nestjs/common';
import { Observable, Subject } from 'rxjs';
import { OnEvent } from '@nestjs/event-emitter';

interface NotificationEvent {
  type: string;
  title: string;
  message: string;
  userId: number;
  timestamp: Date;
}

@Controller('notifications')
export class NotificationsController {
  // Subject для ручного емітування подій
  private notificationsSubject = new Subject<MessageEvent>();

  /**
   * SSE endpoint для нотифікацій
   */
  @Sse('stream')
  streamNotifications(): Observable<MessageEvent> {
    return this.notificationsSubject.asObservable();
  }

  /**
   * Обробник події з іншого модуля (наприклад, новий коментар створено)
   */
  @OnEvent('comment.created')
  handleCommentCreated(payload: { postId: string; authorId: number; text: string }) {
    // Знайти власника поста
    const postOwnerId = this.getPostOwnerId(payload.postId);

    if (postOwnerId !== payload.authorId) {
      // Емітувати SSE подію для власника поста
      this.notificationsSubject.next({
        data: {
          type: 'COMMENT',
          title: 'Новий коментар',
          message: `Користувач залишив коментар під вашим постом`,
          postId: payload.postId,
        },
        id: Date.now().toString(),
        type: 'notification', // Тип SSE події
      });
    }
  }

  /**
   * Обробник події про новий лайк
   */
  @OnEvent('post.liked')
  handlePostLiked(payload: { postId: string; userId: number }) {
    const postOwnerId = this.getPostOwnerId(payload.postId);

    if (postOwnerId !== payload.userId) {
      this.notificationsSubject.next({
        data: {
          type: 'LIKE',
          title: 'Новий лайк',
          message: `Ваш пост сподобався користувачу`,
          postId: payload.postId,
        },
        id: Date.now().toString(),
        type: 'notification',
      });
    }
  }

  private getPostOwnerId(postId: string): number {
    // Запит до БД або кешу
    return 123;
  }
}

const eventSource = new EventSource('http://localhost:3001/notifications/stream');

// Обробник для події типу 'notification'
eventSource.addEventListener('notification', (event) => {
  const notification = JSON.parse(event.data);

  // Показати браузерне повідомлення
  if ('Notification' in window && Notification.permission === 'granted') {
    new Notification(notification.title, {
      body: notification.message,
      icon: '/notification-icon.png',
      tag: notification.type, // Унікальний тег для групування
    });
  }

  // Додати до списку нотифікацій у UI
  addNotificationToList(notification);

  // Оновити badge з кількістю непрочитаних
  incrementUnreadCount();
});

// src/events/events.controller.ts
import { Controller, Sse } from '@nestjs/common';
import { Observable, interval, merge, map } from 'rxjs';

@Controller('events')
export class EventsController {
  private notificationsSubject = new Subject<MessageEvent>();
  private alertsSubject = new Subject<MessageEvent>();

  /**
   * Об'єднаний потік: heartbeat + notifications + alerts
   */
  @Sse('combined')
  streamCombined(): Observable<MessageEvent> {
    // Heartbeat кожні 30 секунд
    const heartbeat$ = interval(30000).pipe(
      map(() => ({
        type: 'heartbeat',
        data: { timestamp: new Date().toISOString() },
      })),
    );

    // Notifications (ручні події)
    const notifications$ = this.notificationsSubject.asObservable();

    // Alerts (ручні події)
    const alerts$ = this.alertsSubject.asObservable();

    // Об'єднати всі потоки
    return merge(heartbeat$, notifications$, alerts$);
  }

  /**
   * Методи для емітування подій з інших сервісів
   */
  emitNotification(data: any) {
    this.notificationsSubject.next({
      type: 'notification',
      data,
      id: Date.now().toString(),
    });
  }

  emitAlert(data: any) {
    this.alertsSubject.next({
      type: 'alert',
      data,
      id: Date.now().toString(),
    });
  }
}

// src/events/events.controller.ts
import { Controller, Sse, Query } from '@nestjs/common';
import { Observable, filter } from 'rxjs';

@Controller('events')
export class EventsController {
  private eventsSubject = new Subject<MessageEvent & { userId?: number }>();

  /**
   * SSE endpoint з фільтрацією за userId
   */
  @Sse('user-events')
  streamUserEvents(@Query('userId') userId: string): Observable<MessageEvent> {
    const userIdNum = parseInt(userId, 10);

    return this.eventsSubject.asObservable().pipe(
      // Фільтрувати: пропускати лише події для цього userId або глобальні події
      filter((event) => !event.userId || event.userId === userIdNum),
      // Видалити службове поле userId перед відправкою клієнту
      map(({ userId, ...rest }) => rest),
    );
  }

  /**
   * Емітувати подію для конкретного користувача
   */
  emitToUser(userId: number, data: any) {
    this.eventsSubject.next({
      userId, // Службове поле для фільтрації
      type: 'user-event',
      data,
      id: Date.now().toString(),
    });
  }

  /**
   * Емітувати глобальну подію (всім)
   */
  emitGlobal(data: any) {
    this.eventsSubject.next({
      type: 'global-event',
      data,
      id: Date.now().toString(),
    });
  }
}

const userId = getCurrentUserId();
const eventSource = new EventSource(`http://localhost:3001/events/user-events?userId=${userId}`);

eventSource.addEventListener('user-event', (event) => {
  console.log('Подія для мене:', JSON.parse(event.data));
});

eventSource.addEventListener('global-event', (event) => {
  console.log('Глобальна подія:', JSON.parse(event.data));
});

// client/components/LiveDashboard.tsx
import React, { useEffect, useState } from 'react';
import { Line } from 'react-chartjs-2';

interface Metrics {
  timestamp: string;
  cpu: { usage: number };
  memory: { usagePercent: number };
}

export const LiveDashboard: React.FC = () => {
  const [metrics, setMetrics] = useState<Metrics[]>([]);
  const [isConnected, setIsConnected] = useState(false);

  useEffect(() => {
    const eventSource = new EventSource('http://localhost:3001/monitoring/metrics');

    eventSource.addEventListener('open', () => {
      console.log('[Dashboard] З\'єднання встановлено');
      setIsConnected(true);
    });

    eventSource.addEventListener('metrics', (event) => {
      const data: Metrics = JSON.parse(event.data);

      setMetrics((prev) => {
        const updated = [...prev, data];
        // Зберігати останні 20 точок для графіка
        return updated.slice(-20);
      });
    });

    eventSource.addEventListener('error', () => {
      console.error('[Dashboard] Помилка з\'єднання');
      setIsConnected(false);
    });

    return () => {
      eventSource.close();
    };
  }, []);

  const chartData = {
    labels: metrics.map((m) => new Date(m.timestamp).toLocaleTimeString()),
    datasets: [
      {
        label: 'CPU Usage (%)',
        data: metrics.map((m) => m.cpu.usage),
        borderColor: 'rgb(75, 192, 192)',
        tension: 0.1,
      },
      {
        label: 'Memory Usage (%)',
        data: metrics.map((m) => m.memory.usagePercent),
        borderColor: 'rgb(255, 99, 132)',
        tension: 0.1,
      },
    ],
  };

  return (
    <div className="dashboard">
      <div className="status">
        {isConnected ? (
          <span className="badge badge-success">🟢 Connected</span>
        ) : (
          <span className="badge badge-error">🔴 Disconnected</span>
        )}
      </div>

      <div className="chart-container">
        <Line data={chartData} options={{ animation: false }} />
      </div>

      {metrics.length > 0 && (
        <div className="current-metrics">
          <div className="metric-card">
            <h3>CPU</h3>
            <p className="metric-value">{metrics[metrics.length - 1].cpu.usage.toFixed(2)}%</p>
          </div>
          <div className="metric-card">
            <h3>Memory</h3>
            <p className="metric-value">
              {metrics[metrics.length - 1].memory.usagePercent.toFixed(2)}%
            </p>
          </div>
        </div>
      )}
    </div>
  );
};

// src/uploads/uploads.controller.ts
import {
  Controller,
  Post,
  UploadedFile,
  UseInterceptors,
  Sse,
  Query,
} from '@nestjs/common';
import { FileInterceptor } from '@nestjs/platform-express';
import { Observable, Subject } from 'rxjs';
import { filter, map } from 'rxjs/operators';

interface ProgressEvent {
  uploadId: string;
  progress: number; // 0-100
  bytesUploaded: number;
  totalBytes: number;
  status: 'processing' | 'completed' | 'error';
  message?: string;
}

@Controller('uploads')
export class UploadsController {
  // Subject для емітування прогресу
  private progressSubject = new Subject<ProgressEvent>();

  /**
   * SSE endpoint для відстеження прогресу конкретного завантаження
   */
  @Sse('progress')
  streamProgress(@Query('uploadId') uploadId: string): Observable<MessageEvent> {
    return this.progressSubject.asObservable().pipe(
      // Фільтрувати лише події для цього uploadId
      filter((event) => event.uploadId === uploadId),
      map((event) => ({
        data: event,
        id: Date.now().toString(),
      })),
    );
  }

  /**
   * Завантаження файлу з емітуванням прогресу
   */
  @Post('upload')
  @UseInterceptors(FileInterceptor('file'))
  async uploadFile(@UploadedFile() file: Express.Multer.File) {
    const uploadId = this.generateUploadId();

    // Емітувати початковий статус
    this.emitProgress(uploadId, {
      progress: 0,
      bytesUploaded: 0,
      totalBytes: file.size,
      status: 'processing',
      message: 'Початок обробки файлу',
    });

    // Симуляція обробки файлу порціями (у реальності — обробка через stream)
    const chunkSize = Math.floor(file.size / 10); // 10 частин
    let bytesProcessed = 0;

    for (let i = 1; i <= 10; i++) {
      await this.sleep(500); // Імітація обробки

      bytesProcessed += chunkSize;
      const progress = Math.min(Math.round((bytesProcessed / file.size) * 100), 100);

      this.emitProgress(uploadId, {
        progress,
        bytesUploaded: bytesProcessed,
        totalBytes: file.size,
        status: 'processing',
        message: `Обробка частини ${i}/10`,
      });
    }

    // Завершення
    this.emitProgress(uploadId, {
      progress: 100,
      bytesUploaded: file.size,
      totalBytes: file.size,
      status: 'completed',
      message: 'Файл успішно завантажено',
    });

    return {
      success: true,
      uploadId,
      filename: file.originalname,
      size: file.size,
    };
  }

  private emitProgress(uploadId: string, data: Omit<ProgressEvent, 'uploadId'>) {
    this.progressSubject.next({
      uploadId,
      ...data,
    });
  }

  private generateUploadId(): string {
    return `upload-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
  }

  private sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }
}

// client/components/FileUpload.tsx
import React, { useState } from 'react';

export const FileUpload: React.FC = () => {
  const [file, setFile] = useState<File | null>(null);
  const [uploading, setUploading] = useState(false);
  const [progress, setProgress] = useState(0);
  const [message, setMessage] = useState('');

  const handleFileSelect = (e: React.ChangeEvent<HTMLInputElement>) => {
    if (e.target.files && e.target.files[0]) {
      setFile(e.target.files[0]);
    }
  };

  const handleUpload = async () => {
    if (!file) return;

    setUploading(true);
    setProgress(0);
    setMessage('Підготовка...');

    // Відправити файл
    const formData = new FormData();
    formData.append('file', file);

    try {
      const response = await fetch('http://localhost:3001/uploads/upload', {
        method: 'POST',
        body: formData,
      });

      const result = await response.json();
      const { uploadId } = result;

      // Підключитися до SSE для відстеження прогресу
      const eventSource = new EventSource(
        `http://localhost:3001/uploads/progress?uploadId=${uploadId}`
      );

      eventSource.addEventListener('message', (event) => {
        const data = JSON.parse(event.data);

        setProgress(data.progress);
        setMessage(data.message);

        if (data.status === 'completed') {
          eventSource.close();
          setUploading(false);
          alert('Файл успішно завантажено!');
        } else if (data.status === 'error') {
          eventSource.close();
          setUploading(false);
          alert(`Помилка: ${data.message}`);
        }
      });

      eventSource.addEventListener('error', () => {
        eventSource.close();
        setUploading(false);
        alert('Помилка з\'єднання');
      });
    } catch (error) {
      console.error('Помилка завантаження:', error);
      setUploading(false);
    }
  };

  return (
    <div className="file-upload">
      <input type="file" onChange={handleFileSelect} disabled={uploading} />

      <button onClick={handleUpload} disabled={!file || uploading}>
        {uploading ? 'Завантаження...' : 'Завантажити'}
      </button>

      {uploading && (
        <div className="progress-container">
          <div className="progress-bar" style={{ width: `${progress}%` }}>
            {progress}%
          </div>
          <p className="progress-message">{message}</p>
        </div>
      )}
    </div>
  );
};

// src/auth/auth.controller.ts
import { Controller, Post, Res, Body } from '@nestjs/common';
import { Response } from 'express';
import { AuthService } from './auth.service';

@Controller('auth')
export class AuthController {
  constructor(private authService: AuthService) {}

  @Post('login')
  async login(
    @Body() credentials: { email: string; password: string },
    @Res({ passthrough: true }) response: Response,
  ) {
    const user = await this.authService.validateUser(credentials.email, credentials.password);

    if (!user) {
      throw new UnauthorizedException('Invalid credentials');
    }

    const token = this.authService.generateToken(user);

    // Встановити HttpOnly cookie
    response.cookie('auth_token', token, {
      httpOnly: true, // Не доступний через JavaScript
      secure: process.env.NODE_ENV === 'production', // Лише HTTPS у production
      sameSite: 'strict',
      maxAge: 7 * 24 * 60 * 60 * 1000, // 7 днів
    });

    return { success: true, user: { id: user.id, email: user.email } };
  }
}

// src/events/events.controller.ts
import { Controller, Sse, Req, UnauthorizedException } from '@nestjs/common';
import { Request } from 'express';
import { Observable } from 'rxjs';
import { JwtService } from '@nestjs/jwt';

@Controller('events')
export class EventsController {
  constructor(private jwtService: JwtService) {}

  @Sse('stream')
  streamEvents(@Req() request: Request): Observable<MessageEvent> {
    // Витягти токен з cookie
    const token = request.cookies['auth_token'];

    if (!token) {
      throw new UnauthorizedException('No authentication token');
    }

    try {
      // Валідувати токен
      const payload = this.jwtService.verify(token, {
        secret: process.env.JWT_SECRET,
      });

      const userId = payload.sub;

      // Повернути потік подій для цього користувача
      return this.getEventsStreamForUser(userId);
    } catch (error) {
      throw new UnauthorizedException('Invalid token');
    }
  }

  private getEventsStreamForUser(userId: number): Observable<MessageEvent> {
    // Фільтрований потік подій для користувача
    return this.eventsSubject.asObservable().pipe(
      filter((event) => event.userId === userId),
      map((event) => ({
        data: event.data,
        id: event.id,
      })),
    );
  }
}

// EventSource автоматично відправляє cookies, якщо вказати withCredentials
const eventSource = new EventSource('http://localhost:3001/events/stream', {
  withCredentials: true, // Відправляти cookies cross-origin
});

eventSource.addEventListener('message', (event) => {
  console.log('Подія:', JSON.parse(event.data));
});

@Sse('stream')
streamEvents(@Query('token') token: string): Observable<MessageEvent> {
  if (!token) {
    throw new UnauthorizedException('No token provided');
  }

  try {
    const payload = this.jwtService.verify(token);
    const userId = payload.sub;

    return this.getEventsStreamForUser(userId);
  } catch (error) {
    throw new UnauthorizedException('Invalid token');
  }
}

const token = localStorage.getItem('accessToken');
const eventSource = new EventSource(`http://localhost:3001/events/stream?token=${token}`);

@Sse('stream')
streamEvents(): Observable<MessageEvent> {
  const heartbeat$ = interval(15000).pipe(
    map(() => ({
      type: 'heartbeat',
      data: { timestamp: new Date().toISOString() },
    })),
  );

  const events$ = this.eventsSubject.asObservable();

  return merge(heartbeat$, events$);
}

this.eventsSubject.next({
  data: { message: 'Hello' },
  id: Date.now().toString(), // Унікальний ID
});

@Sse('stream')
streamEvents(@Headers('last-event-id') lastEventId?: string): Observable<MessageEvent> {
  const sinceId = lastEventId ? parseInt(lastEventId, 10) : 0;

  // Відправити пропущені події з кешу або БД
  const missedEvents = this.getCachedEventsSince(sinceId);

  const missed$ = from(missedEvents);
  const live$ = this.eventsSubject.asObservable();

  // Спочатку пропущені, потім live
  return concat(missed$, live$);
}

// React
useEffect(() => {
  const eventSource = new EventSource('/events/stream');

  // Обробники подій...

  return () => {
    eventSource.close(); // Cleanup при unmount
  };
}, []);

@Controller('events')
export class EventsController {
  private activeConnections = 0;

  @Sse('stream')
  streamEvents(): Observable<MessageEvent> {
    this.activeConnections++;
    console.log(`[SSE] Активних з'єднань: ${this.activeConnections}`);

    return this.eventsSubject.asObservable().pipe(
      finalize(() => {
        this.activeConnections--;
        console.log(`[SSE] Клієнт відключився. Залишилось: ${this.activeConnections}`);
      }),
    );
  }

  @Get('metrics')
  getMetrics() {
    return {
      activeSSEConnections: this.activeConnections,
    };
  }
}

# Nginx конфігурація
location /events/ {
    proxy_pass http://localhost:3001;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding off;
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 3600s; # 1 година timeout
}

@Sse('stream')
streamEvents(): Observable<MessageEvent> {
  return interval(2000).pipe(
    map((index) => {
      const event = {
        data: { message: 'Test', index },
        id: index.toString(),
      };

      console.log('[SSE] Відправка події:', JSON.stringify(event));
      return event;
    }),
  );
}

@Sse('stream')
streamEvents(): Observable<MessageEvent> {
  const connectionId = Math.random().toString(36);

  console.log(`[SSE] Клієнт ${connectionId} підключився`);

  return this.eventsSubject.asObservable().pipe(
    finalize(() => {
      console.log(`[SSE] Клієнт ${connectionId} відключився`);
      // Cleanup: видалити з Map, звільнити ресурси
    }),
  );
}