98 lines
2.7 KiB
Go
98 lines
2.7 KiB
Go
package realtime
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log/slog"
|
|
"strings"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
const sseChannelPrefix = "sse:"
|
|
|
|
// SSEPublisher publishes SSE events to Redis for cross-process delivery.
|
|
// Used by the worker to send events that reach SSE-connected clients via the service.
|
|
type SSEPublisher struct {
|
|
client *redis.Client
|
|
logger *slog.Logger
|
|
}
|
|
|
|
// NewSSEPublisher creates a publisher that sends SSE events via Redis pub/sub.
|
|
func NewSSEPublisher(client *redis.Client, logger *slog.Logger) *SSEPublisher {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &SSEPublisher{
|
|
client: client,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
// SendToUser publishes an SSE event to a user-specific channel.
|
|
// The service's SSE subscriber picks this up and delivers to the connected client.
|
|
func (p *SSEPublisher) SendToUser(userID string, event *SSEEvent) error {
|
|
return p.SendToChannel("user:"+userID, event)
|
|
}
|
|
|
|
// SendToChannel publishes an SSE event to a named channel via Redis.
|
|
func (p *SSEPublisher) SendToChannel(channel string, event *SSEEvent) error {
|
|
data, err := json.Marshal(event)
|
|
if err != nil {
|
|
return fmt.Errorf("marshal SSE event: %w", err)
|
|
}
|
|
|
|
redisChannel := sseChannelPrefix + channel
|
|
if err := p.client.Publish(context.Background(), redisChannel, data).Err(); err != nil {
|
|
p.logger.Error("failed to publish SSE event to Redis",
|
|
"channel", channel, "event_type", event.Type, "error", err)
|
|
return fmt.Errorf("publish SSE event: %w", err)
|
|
}
|
|
|
|
p.logger.Debug("published SSE event to Redis",
|
|
"channel", channel, "event_type", event.Type)
|
|
return nil
|
|
}
|
|
|
|
// RunSSESubscriber subscribes to all SSE Redis channels and forwards events
|
|
// to the local SSEHub for delivery to connected clients.
|
|
// Blocks until ctx is cancelled.
|
|
func RunSSESubscriber(ctx context.Context, client *redis.Client, hub *SSEHub, logger *slog.Logger) error {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
|
|
// PSubscribe to all SSE channels
|
|
pubsub := client.PSubscribe(ctx, sseChannelPrefix+"*")
|
|
defer pubsub.Close()
|
|
|
|
logger.Info("SSE Redis subscriber started", "pattern", sseChannelPrefix+"*")
|
|
|
|
ch := pubsub.Channel()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
logger.Info("SSE Redis subscriber stopping")
|
|
return ctx.Err()
|
|
case msg, ok := <-ch:
|
|
if !ok {
|
|
return fmt.Errorf("SSE Redis subscription closed")
|
|
}
|
|
|
|
// Extract channel name by stripping the "sse:" prefix
|
|
channel := strings.TrimPrefix(msg.Channel, sseChannelPrefix)
|
|
|
|
var event SSEEvent
|
|
if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil {
|
|
logger.Error("failed to parse SSE event from Redis",
|
|
"channel", channel, "error", err)
|
|
continue
|
|
}
|
|
|
|
// Deliver to local SSE hub
|
|
hub.SendToChannel(channel, &event)
|
|
}
|
|
}
|
|
}
|