added HandleSSEConnection

internal/interface/sse/consumer.go

@@ -14,9 +14,11 @@ type SSEConsumer struct {
 	queue      *queue.Queue
 }
 
-func NewSSEConsumer(redisAddr string, dispatcher *ConnectionDispatcher) (*SSEConsumer, error) {
+func NewSSEConsumer(redisAddr, redisPassword string, redisDB int, dispatcher *ConnectionDispatcher) (*SSEConsumer, error) {
 	w := redisdb.NewWorker(
 		redisdb.WithAddr(redisAddr),
+		redisdb.WithDB(redisDB),
+		redisdb.WithPassword(redisPassword),
 		redisdb.WithChannel(RedisKeySSEEvents),
 		redisdb.WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			var sseMessage SSEEvent

internal/interface/sse/dispatcher.go

@@ -51,6 +51,7 @@ func (r *ConnectionDispatcher) RemoveConnection(userID, deviceID string) {
 		if conn.DeviceID != deviceID {
 			afterDropConnections = append(afterDropConnections, conn)
 		} else {
+			fmt.Println("ЙОУ")
 			close(conn.Channel)
 		}
 	}

internal/interface/sse/producer.go

@@ -24,9 +24,11 @@ func (r *SSEMessageTask) Bytes() []byte {
 	return b
 }
 
-func NewSSEProducer(redisAddr string) (*SSEProducer, error) {
+func NewSSEProducer(redisAddr, redisPassword string, redisDB int) (*SSEProducer, error) {
 	worker := redisdb.NewWorker(
 		redisdb.WithAddr(redisAddr),
+		redisdb.WithDB(redisDB),
+		redisdb.WithPassword(redisPassword),
 		redisdb.WithChannel(RedisKeySSEEvents),
 		redisdb.WithRunFunc(func(ctx context.Context, m core.TaskMessage) error {
 			return nil

internal/interface/sse/service.go

@@ -1,9 +1,12 @@
 package sse
 
 import (
+	"bufio"
+	"encoding/json"
 	"fmt"
 	"github.com/gofiber/fiber/v2"
 	"go.uber.org/zap"
+	"time"
 )
 
 type SSEService struct {
@@ -13,15 +16,15 @@ type SSEService struct {
 	logger     *zap.Logger
 }
 
-func NewSSEService(redisAddr string, bufferSize int, logger *zap.Logger) (*SSEService, error) {
+func NewSSEService(redisAddr, redisPassword string, redisDB int, bufferSize int, logger *zap.Logger) (*SSEService, error) {
 	dispatcher := NewConnectionDispatcher(bufferSize)
 
-	producer, err := NewSSEProducer(redisAddr)
+	producer, err := NewSSEProducer(redisAddr, redisPassword, redisDB)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create producer: %w", err)
 	}
 
-	consumer, err := NewSSEConsumer(redisAddr, dispatcher)
+	consumer, err := NewSSEConsumer(redisAddr, redisPassword, redisDB, dispatcher)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create consumer: %w", err)
 	}
@@ -52,5 +55,59 @@ func (s *SSEService) SendEvent(userID, eventType string, data interface{}) error
 
 // todo для тех кто потребляет контент
 func (s *SSEService) HandleSSEConnection(ctx *fiber.Ctx, userID, deviceID string) error {
+	ctx.Set(fiber.HeaderContentType, "text/event-stream")
+	ctx.Set("Cache-Control", "no-cache")
+	ctx.Set("Connection", "keep-alive")
+	ctx.Set("Transfer-Encoding", "chunked")
+
+	conn := s.dispatcher.AddConnection(userID, deviceID)
+	mainCtx := ctx.Context().Done()
+
+	ctx.Status(fiber.StatusOK).Context().SetBodyStreamWriter(func(w *bufio.Writer) {
+		defer s.dispatcher.RemoveConnection(userID, deviceID)
+		pingTicker := time.NewTicker(5 * time.Second)
+		defer pingTicker.Stop()
+
+		for {
+			select {
+			case <-mainCtx:
+				s.logger.Info("Client disconnected", zap.String("userID", userID), zap.String("deviceID", deviceID))
+				return
+			case event, ok := <-conn.Channel:
+				if !ok {
+					s.logger.Info("Connection channel closed", zap.String("userID", userID), zap.String("deviceID", deviceID))
+					return
+				}
+
+				payload, err := json.Marshal(map[string]interface{}{
+					"event": event.EventType,
+					"data":  event.Data,
+				})
+				if err != nil {
+					s.logger.Error("Failed to marshal event", zap.Error(err))
+					continue
+				}
+
+				if _, err := fmt.Fprintf(w, "data: %s\n\n", payload); err != nil {
+					s.logger.Error("Failed to write event to stream", zap.Error(err))
+					return
+				}
+				if err := w.Flush(); err != nil {
+					s.logger.Error("Failed to flush SSE stream", zap.Error(err))
+					return
+				}
+			case <-pingTicker.C:
+				if _, err := fmt.Fprintf(w, "data: %s\n\n", `{"event":"ping"}`); err != nil {
+					s.logger.Error("Failed to write ping to stream", zap.Error(err))
+					return
+				}
+				if err := w.Flush(); err != nil {
+					s.logger.Error("Failed to flush ping", zap.Error(err))
+					return
+				}
+			}
+		}
+	})
+
 	return nil
 }