add real-time infrastructure (#137)
* setup infra for pusher service * setup client sdk for pusher service * fix: ping parse failure * fix: send pong back to client avoid disconnections every 2.5 minutes * increase replicas * feat: show presence and compose indicator
This commit was merged in pull request #137.
This commit is contained in:
Arjun Patel
@@ -0,0 +1,66 @@
|
||||
package pusher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
|
||||
"github.com/flowy-live/llink/internal/network"
|
||||
)
|
||||
|
||||
var ErrUnauthorized = errors.New("unauthorized")
|
||||
|
||||
// Authorizer validates whether a user can access a given channel.
|
||||
type Authorizer struct {
|
||||
networkSvc network.Service
|
||||
}
|
||||
|
||||
// NewAuthorizer creates a new channel authorizer.
|
||||
func NewAuthorizer(networkSvc network.Service) *Authorizer {
|
||||
return &Authorizer{networkSvc: networkSvc}
|
||||
}
|
||||
|
||||
// Authorize checks if the given humanID is allowed to subscribe to the channel.
|
||||
// Channel formats:
|
||||
// - network:{networkId}
|
||||
// - stream:{networkId}:{streamId}
|
||||
func (a *Authorizer) Authorize(ctx context.Context, channelID, humanID string) error {
|
||||
parts := strings.SplitN(channelID, ":", 2)
|
||||
if len(parts) < 2 {
|
||||
return ErrUnauthorized
|
||||
}
|
||||
|
||||
channelType := parts[0]
|
||||
rest := parts[1]
|
||||
|
||||
switch channelType {
|
||||
case "network":
|
||||
return a.authorizeNetwork(ctx, rest, humanID)
|
||||
case "stream":
|
||||
return a.authorizeStream(ctx, rest, humanID)
|
||||
default:
|
||||
return ErrUnauthorized
|
||||
}
|
||||
}
|
||||
|
||||
func (a *Authorizer) authorizeNetwork(ctx context.Context, networkID, humanID string) error {
|
||||
isMember, err := a.networkSvc.IsMember(ctx, networkID, humanID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !isMember {
|
||||
return ErrUnauthorized
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// authorizeStream expects rest to be "{networkId}:{streamId}".
|
||||
// We only check network membership — stream visibility is handled by network access.
|
||||
func (a *Authorizer) authorizeStream(ctx context.Context, rest, humanID string) error {
|
||||
parts := strings.SplitN(rest, ":", 2)
|
||||
if len(parts) < 2 {
|
||||
return ErrUnauthorized
|
||||
}
|
||||
networkID := parts[0]
|
||||
return a.authorizeNetwork(ctx, networkID, humanID)
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package pusher
|
||||
|
||||
// Channel tracks the local connections subscribed to a channel on this pod.
|
||||
// All methods are only called from the Hub goroutine — no locks needed.
|
||||
type Channel struct {
|
||||
id string
|
||||
members map[*Conn]string // conn → humanID
|
||||
}
|
||||
|
||||
func newChannel(id string) *Channel {
|
||||
return &Channel{
|
||||
id: id,
|
||||
members: make(map[*Conn]string),
|
||||
}
|
||||
}
|
||||
|
||||
func (ch *Channel) addMember(conn *Conn, humanID string) {
|
||||
ch.members[conn] = humanID
|
||||
}
|
||||
|
||||
func (ch *Channel) removeMember(conn *Conn) {
|
||||
delete(ch.members, conn)
|
||||
}
|
||||
|
||||
func (ch *Channel) isEmpty() bool {
|
||||
return len(ch.members) == 0
|
||||
}
|
||||
|
||||
// localHumanIDs returns the deduplicated set of humanIDs connected on this pod.
|
||||
func (ch *Channel) localHumanIDs() []string {
|
||||
seen := make(map[string]bool, len(ch.members))
|
||||
ids := make([]string, 0, len(ch.members))
|
||||
for _, hid := range ch.members {
|
||||
if !seen[hid] {
|
||||
seen[hid] = true
|
||||
ids = append(ids, hid)
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// hasHumanID returns true if the given humanID has at least one local connection.
|
||||
func (ch *Channel) hasHumanID(humanID string) bool {
|
||||
for _, hid := range ch.members {
|
||||
if hid == humanID {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// broadcast sends a message to all local connections except the excluded one.
|
||||
func (ch *Channel) broadcast(msg ServerMessage, exclude *Conn) {
|
||||
for conn := range ch.members {
|
||||
if conn != exclude {
|
||||
conn.Send(msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
package pusher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"sync"
|
||||
|
||||
"nhooyr.io/websocket"
|
||||
)
|
||||
|
||||
const sendBufferSize = 256
|
||||
|
||||
// Conn wraps a WebSocket connection with identity and a send buffer.
|
||||
type Conn struct {
|
||||
id string
|
||||
humanID string
|
||||
ws *websocket.Conn
|
||||
send chan []byte
|
||||
once sync.Once // ensures close logic runs once
|
||||
}
|
||||
|
||||
func newConn(id, humanID string, ws *websocket.Conn) *Conn {
|
||||
return &Conn{
|
||||
id: id,
|
||||
humanID: humanID,
|
||||
ws: ws,
|
||||
send: make(chan []byte, sendBufferSize),
|
||||
}
|
||||
}
|
||||
|
||||
// ReadPump reads messages from the WebSocket and forwards them to the hub.
|
||||
// It blocks until the connection is closed or the context is cancelled.
|
||||
func (c *Conn) ReadPump(ctx context.Context, hub *Hub) {
|
||||
defer hub.disconnect(c)
|
||||
|
||||
for {
|
||||
_, data, err := c.ws.Read(ctx)
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
slog.Debug("websocket read error", "connId", c.id, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Respond to keep-alive pings
|
||||
if string(data) == "ping" {
|
||||
if err := c.ws.Write(ctx, websocket.MessageText, []byte("pong")); err != nil {
|
||||
slog.Debug("websocket pong write error", "connId", c.id, "error", err)
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
var msg ClientMessage
|
||||
if err := json.Unmarshal(data, &msg); err != nil {
|
||||
c.sendError("invalid message format")
|
||||
continue
|
||||
}
|
||||
|
||||
switch msg.Type {
|
||||
case TypeSubscribe:
|
||||
if msg.Channel == "" {
|
||||
c.sendError("channel is required")
|
||||
continue
|
||||
}
|
||||
hub.subscribeCh <- &subscribeRequest{conn: c, channelID: msg.Channel}
|
||||
case TypeUnsubscribe:
|
||||
if msg.Channel == "" {
|
||||
c.sendError("channel is required")
|
||||
continue
|
||||
}
|
||||
hub.unsubscribeCh <- &unsubscribeRequest{conn: c, channelID: msg.Channel}
|
||||
case TypeMessage:
|
||||
if msg.Channel == "" {
|
||||
c.sendError("channel is required")
|
||||
continue
|
||||
}
|
||||
hub.broadcastCh <- &broadcastRequest{conn: c, channelID: msg.Channel, payload: msg.Payload}
|
||||
default:
|
||||
c.sendError("unknown message type: " + msg.Type)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WritePump drains the send buffer and writes messages to the WebSocket.
|
||||
func (c *Conn) WritePump(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case data, ok := <-c.send:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if err := c.ws.Write(ctx, websocket.MessageText, data); err != nil {
|
||||
slog.Debug("websocket write error", "connId", c.id, "error", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Send enqueues a ServerMessage to be written to the WebSocket.
|
||||
// If the send buffer is full, the connection is closed (slow client).
|
||||
func (c *Conn) Send(msg ServerMessage) {
|
||||
data, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
slog.Error("failed to marshal server message", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case c.send <- data:
|
||||
default:
|
||||
slog.Warn("slow client, closing connection", "connId", c.id, "humanId", c.humanID)
|
||||
c.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Close closes the WebSocket connection and the send channel.
|
||||
func (c *Conn) Close() {
|
||||
c.once.Do(func() {
|
||||
c.ws.Close(websocket.StatusNormalClosure, "closing")
|
||||
close(c.send)
|
||||
})
|
||||
}
|
||||
|
||||
func (c *Conn) sendError(msg string) {
|
||||
c.Send(ServerMessage{Type: TypeError, Message: msg})
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
package pusher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
)
|
||||
|
||||
type subscribeRequest struct {
|
||||
conn *Conn
|
||||
channelID string
|
||||
}
|
||||
|
||||
type unsubscribeRequest struct {
|
||||
conn *Conn
|
||||
channelID string
|
||||
}
|
||||
|
||||
type broadcastRequest struct {
|
||||
conn *Conn
|
||||
channelID string
|
||||
payload json.RawMessage
|
||||
}
|
||||
|
||||
type remoteEvent struct {
|
||||
channelID string
|
||||
event redisEvent
|
||||
}
|
||||
|
||||
// Hub manages all local WebSocket connections and channels on this pod.
|
||||
// All state mutations happen in a single goroutine via Go channels — no locks.
|
||||
type Hub struct {
|
||||
channels map[string]*Channel
|
||||
connChannels map[*Conn]map[string]bool // reverse index: conn → set of channel IDs
|
||||
|
||||
bridge *RedisBridge
|
||||
authorizer *Authorizer
|
||||
|
||||
subscribeCh chan *subscribeRequest
|
||||
unsubscribeCh chan *unsubscribeRequest
|
||||
broadcastCh chan *broadcastRequest
|
||||
disconnectCh chan *Conn
|
||||
remoteEventCh chan *remoteEvent
|
||||
}
|
||||
|
||||
// NewHub creates a new Hub.
|
||||
func NewHub(bridge *RedisBridge, authorizer *Authorizer) *Hub {
|
||||
return &Hub{
|
||||
channels: make(map[string]*Channel),
|
||||
connChannels: make(map[*Conn]map[string]bool),
|
||||
bridge: bridge,
|
||||
authorizer: authorizer,
|
||||
subscribeCh: make(chan *subscribeRequest, 256),
|
||||
unsubscribeCh: make(chan *unsubscribeRequest, 256),
|
||||
broadcastCh: make(chan *broadcastRequest, 256),
|
||||
disconnectCh: make(chan *Conn, 256),
|
||||
remoteEventCh: make(chan *remoteEvent, 256),
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts the hub event loop. Blocks until the context is cancelled.
|
||||
func (h *Hub) Run(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
|
||||
case req := <-h.subscribeCh:
|
||||
h.handleSubscribe(ctx, req)
|
||||
|
||||
case req := <-h.unsubscribeCh:
|
||||
h.handleUnsubscribe(ctx, req)
|
||||
|
||||
case req := <-h.broadcastCh:
|
||||
h.handleBroadcast(ctx, req)
|
||||
|
||||
case conn := <-h.disconnectCh:
|
||||
h.handleDisconnect(ctx, conn)
|
||||
|
||||
case evt := <-h.remoteEventCh:
|
||||
h.handleRemoteEvent(evt)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) {
|
||||
// Authorize channel access
|
||||
if err := h.authorizer.Authorize(ctx, req.channelID, req.conn.humanID); err != nil {
|
||||
req.conn.Send(ServerMessage{
|
||||
Type: TypeError,
|
||||
Channel: req.channelID,
|
||||
Message: "unauthorized",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Get or create local channel
|
||||
ch, ok := h.channels[req.channelID]
|
||||
if !ok {
|
||||
ch = newChannel(req.channelID)
|
||||
h.channels[req.channelID] = ch
|
||||
}
|
||||
|
||||
// Add to local channel
|
||||
ch.addMember(req.conn, req.conn.humanID)
|
||||
|
||||
// Track in reverse index
|
||||
if h.connChannels[req.conn] == nil {
|
||||
h.connChannels[req.conn] = make(map[string]bool)
|
||||
}
|
||||
h.connChannels[req.conn][req.channelID] = true
|
||||
|
||||
// Register in Redis and get global presence
|
||||
presence, err := h.bridge.Subscribe(ctx, req.channelID, req.conn.id, req.conn.humanID)
|
||||
if err != nil {
|
||||
slog.Error("redis subscribe failed", "channelId", req.channelID, "error", err)
|
||||
// Still send local presence as fallback
|
||||
presence = ch.localHumanIDs()
|
||||
}
|
||||
|
||||
// Send subscribed ack with presence snapshot
|
||||
req.conn.Send(ServerMessage{
|
||||
Type: TypeSubscribed,
|
||||
Channel: req.channelID,
|
||||
Presence: presence,
|
||||
})
|
||||
}
|
||||
|
||||
func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) {
|
||||
ch, ok := h.channels[req.channelID]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
ch.removeMember(req.conn)
|
||||
|
||||
// Remove from reverse index
|
||||
if chans, ok := h.connChannels[req.conn]; ok {
|
||||
delete(chans, req.channelID)
|
||||
}
|
||||
|
||||
// Update Redis
|
||||
if err := h.bridge.Unsubscribe(ctx, req.channelID, req.conn.id, req.conn.humanID); err != nil {
|
||||
slog.Error("redis unsubscribe failed", "channelId", req.channelID, "error", err)
|
||||
}
|
||||
|
||||
// Clean up empty local channel
|
||||
if ch.isEmpty() {
|
||||
delete(h.channels, req.channelID)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Hub) handleBroadcast(ctx context.Context, req *broadcastRequest) {
|
||||
ch, ok := h.channels[req.channelID]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
// Check that the sender is actually in the channel
|
||||
if _, isMember := ch.members[req.conn]; !isMember {
|
||||
req.conn.sendError("not subscribed to channel: " + req.channelID)
|
||||
return
|
||||
}
|
||||
|
||||
// Deliver to local connections (except sender)
|
||||
ch.broadcast(ServerMessage{
|
||||
Type: TypeMessage,
|
||||
Channel: req.channelID,
|
||||
HumanID: req.conn.humanID,
|
||||
Payload: req.payload,
|
||||
}, req.conn)
|
||||
|
||||
// Publish to Redis for other pods
|
||||
h.bridge.Broadcast(ctx, req.channelID, req.conn.humanID, req.payload)
|
||||
}
|
||||
|
||||
func (h *Hub) handleDisconnect(ctx context.Context, conn *Conn) {
|
||||
chans, ok := h.connChannels[conn]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
for channelID := range chans {
|
||||
ch, ok := h.channels[channelID]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
ch.removeMember(conn)
|
||||
|
||||
if err := h.bridge.Unsubscribe(ctx, channelID, conn.id, conn.humanID); err != nil {
|
||||
slog.Error("redis unsubscribe on disconnect failed", "channelId", channelID, "error", err)
|
||||
}
|
||||
|
||||
if ch.isEmpty() {
|
||||
delete(h.channels, channelID)
|
||||
}
|
||||
}
|
||||
|
||||
delete(h.connChannels, conn)
|
||||
}
|
||||
|
||||
func (h *Hub) handleRemoteEvent(evt *remoteEvent) {
|
||||
ch, ok := h.channels[evt.channelID]
|
||||
if !ok {
|
||||
return // no local connections care about this channel
|
||||
}
|
||||
|
||||
switch evt.event.Type {
|
||||
case TypeJoin:
|
||||
ch.broadcast(ServerMessage{
|
||||
Type: TypeJoin,
|
||||
Channel: evt.channelID,
|
||||
HumanID: evt.event.HumanID,
|
||||
}, nil)
|
||||
|
||||
case TypeLeave:
|
||||
ch.broadcast(ServerMessage{
|
||||
Type: TypeLeave,
|
||||
Channel: evt.channelID,
|
||||
HumanID: evt.event.HumanID,
|
||||
}, nil)
|
||||
|
||||
case TypeMessage:
|
||||
ch.broadcast(ServerMessage{
|
||||
Type: TypeMessage,
|
||||
Channel: evt.channelID,
|
||||
HumanID: evt.event.HumanID,
|
||||
Payload: evt.event.Payload,
|
||||
}, nil)
|
||||
}
|
||||
}
|
||||
|
||||
// disconnect sends a connection to the disconnect channel.
|
||||
func (h *Hub) disconnect(conn *Conn) {
|
||||
h.disconnectCh <- conn
|
||||
}
|
||||
@@ -0,0 +1,372 @@
|
||||
package pusher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
const (
|
||||
podHeartbeatInterval = 30 * time.Second
|
||||
podHeartbeatTTL = 60 * time.Second
|
||||
cleanupInterval = 60 * time.Second
|
||||
|
||||
// Redis key prefixes
|
||||
channelConnsPrefix = "pusher:ch:"
|
||||
channelConnsSuffix = ":conns"
|
||||
podKeyPrefix = "pusher:pod:"
|
||||
pubsubPrefix = "pusher:events:"
|
||||
)
|
||||
|
||||
// redisEvent is published/received via Redis Pub/Sub for cross-pod communication.
|
||||
type redisEvent struct {
|
||||
Type string `json:"type"` // "join", "leave", "message"
|
||||
HumanID string `json:"humanId,omitempty"` // who triggered the event
|
||||
PodID string `json:"podId,omitempty"` // originating pod
|
||||
Payload json.RawMessage `json:"payload,omitempty"` // for message events
|
||||
}
|
||||
|
||||
// RedisBridge handles cross-pod coordination via Redis Pub/Sub and presence tracking.
|
||||
type RedisBridge struct {
|
||||
client *redis.Client
|
||||
podID string
|
||||
hub *Hub // set after hub is created
|
||||
}
|
||||
|
||||
// NewRedisBridge creates a new Redis bridge for cross-pod coordination.
|
||||
func NewRedisBridge(client *redis.Client, podID string) *RedisBridge {
|
||||
return &RedisBridge{
|
||||
client: client,
|
||||
podID: podID,
|
||||
}
|
||||
}
|
||||
|
||||
// SetHub sets the hub reference. Called during initialization.
|
||||
func (rb *RedisBridge) SetHub(hub *Hub) {
|
||||
rb.hub = hub
|
||||
}
|
||||
|
||||
// --- Presence management (called by hub goroutine) ---
|
||||
|
||||
// Subscribe adds a connection to a channel in Redis.
|
||||
// Returns the current presence set for the channel.
|
||||
func (rb *RedisBridge) Subscribe(ctx context.Context, channelID, connID, humanID string) ([]string, error) {
|
||||
key := channelConnsKey(channelID)
|
||||
field := rb.connField(connID)
|
||||
|
||||
// Check if humanID was already present before adding
|
||||
existingMembers, err := rb.client.HVals(ctx, key).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return nil, fmt.Errorf("failed to get channel members: %w", err)
|
||||
}
|
||||
|
||||
wasPresent := containsString(existingMembers, humanID)
|
||||
|
||||
// Add this connection
|
||||
if err := rb.client.HSet(ctx, key, field, humanID).Err(); err != nil {
|
||||
return nil, fmt.Errorf("failed to add connection to channel: %w", err)
|
||||
}
|
||||
|
||||
// Publish join event if this is a new humanID in the channel
|
||||
if !wasPresent {
|
||||
rb.publishEvent(ctx, channelID, redisEvent{
|
||||
Type: TypeJoin,
|
||||
HumanID: humanID,
|
||||
PodID: rb.podID,
|
||||
})
|
||||
}
|
||||
|
||||
// Return deduplicated presence set
|
||||
allMembers, err := rb.client.HVals(ctx, key).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get channel members: %w", err)
|
||||
}
|
||||
return deduplicateStrings(allMembers), nil
|
||||
}
|
||||
|
||||
// Unsubscribe removes a connection from a channel in Redis.
|
||||
func (rb *RedisBridge) Unsubscribe(ctx context.Context, channelID, connID, humanID string) error {
|
||||
key := channelConnsKey(channelID)
|
||||
field := rb.connField(connID)
|
||||
|
||||
if err := rb.client.HDel(ctx, key, field).Err(); err != nil {
|
||||
return fmt.Errorf("failed to remove connection from channel: %w", err)
|
||||
}
|
||||
|
||||
// Check if this humanID is still present via other connections
|
||||
remainingMembers, err := rb.client.HVals(ctx, key).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return fmt.Errorf("failed to get remaining members: %w", err)
|
||||
}
|
||||
|
||||
if !containsString(remainingMembers, humanID) {
|
||||
rb.publishEvent(ctx, channelID, redisEvent{
|
||||
Type: TypeLeave,
|
||||
HumanID: humanID,
|
||||
PodID: rb.podID,
|
||||
})
|
||||
}
|
||||
|
||||
// Clean up empty channel hash
|
||||
if len(remainingMembers) == 0 {
|
||||
rb.client.Del(ctx, key)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Broadcast publishes a message event to all pods.
|
||||
func (rb *RedisBridge) Broadcast(ctx context.Context, channelID, humanID string, payload json.RawMessage) {
|
||||
rb.publishEvent(ctx, channelID, redisEvent{
|
||||
Type: TypeMessage,
|
||||
HumanID: humanID,
|
||||
PodID: rb.podID,
|
||||
Payload: payload,
|
||||
})
|
||||
}
|
||||
|
||||
// GetPresence returns the deduplicated humanIDs for the given channels.
|
||||
func (rb *RedisBridge) GetPresence(ctx context.Context, channelIDs []string) (map[string][]string, error) {
|
||||
result := make(map[string][]string, len(channelIDs))
|
||||
for _, chID := range channelIDs {
|
||||
members, err := rb.client.HVals(ctx, channelConnsKey(chID)).Result()
|
||||
if err != nil && err != redis.Nil {
|
||||
return nil, fmt.Errorf("failed to get presence for %s: %w", chID, err)
|
||||
}
|
||||
result[chID] = deduplicateStrings(members)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// --- Pub/Sub listener (runs in its own goroutine) ---
|
||||
|
||||
// Listen subscribes to Redis Pub/Sub and forwards events to the local hub.
|
||||
// Blocks until the context is cancelled.
|
||||
func (rb *RedisBridge) Listen(ctx context.Context) {
|
||||
pubsub := rb.client.PSubscribe(ctx, pubsubPrefix+"*")
|
||||
defer pubsub.Close()
|
||||
|
||||
ch := pubsub.Channel()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case msg, ok := <-ch:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
rb.handlePubSubMessage(msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (rb *RedisBridge) handlePubSubMessage(msg *redis.Message) {
|
||||
// Extract channel ID from topic: "pusher:events:{channelID}"
|
||||
channelID := strings.TrimPrefix(msg.Channel, pubsubPrefix)
|
||||
if channelID == "" {
|
||||
return
|
||||
}
|
||||
|
||||
var event redisEvent
|
||||
if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil {
|
||||
slog.Error("failed to parse pub/sub event", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
// Skip events originating from this pod — the local hub already handled them
|
||||
if event.PodID == rb.podID {
|
||||
return
|
||||
}
|
||||
|
||||
if rb.hub == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Forward to local hub for delivery to local WebSocket connections
|
||||
rb.hub.remoteEventCh <- &remoteEvent{
|
||||
channelID: channelID,
|
||||
event: event,
|
||||
}
|
||||
}
|
||||
|
||||
// --- Heartbeat + cleanup (runs in its own goroutine) ---
|
||||
|
||||
// Heartbeat maintains this pod's liveness key and cleans up stale pods.
|
||||
func (rb *RedisBridge) Heartbeat(ctx context.Context) {
|
||||
podKey := podKeyPrefix + rb.podID
|
||||
|
||||
// Initial heartbeat
|
||||
rb.client.Set(ctx, podKey, "alive", podHeartbeatTTL)
|
||||
|
||||
heartbeatTicker := time.NewTicker(podHeartbeatInterval)
|
||||
cleanupTicker := time.NewTicker(cleanupInterval)
|
||||
defer heartbeatTicker.Stop()
|
||||
defer cleanupTicker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
// On shutdown, remove our pod key and clean up our connections
|
||||
rb.client.Del(context.Background(), podKey)
|
||||
rb.cleanupPod(context.Background(), rb.podID)
|
||||
return
|
||||
case <-heartbeatTicker.C:
|
||||
rb.client.Set(ctx, podKey, "alive", podHeartbeatTTL)
|
||||
case <-cleanupTicker.C:
|
||||
rb.cleanupStalePods(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (rb *RedisBridge) cleanupStalePods(ctx context.Context) {
|
||||
// Scan all channel conn hashes for pod IDs, then check if those pods are still alive
|
||||
var cursor uint64
|
||||
knownPods := make(map[string]bool)
|
||||
alivePods := make(map[string]bool)
|
||||
|
||||
for {
|
||||
keys, nextCursor, err := rb.client.Scan(ctx, cursor, channelConnsPrefix+"*"+channelConnsSuffix, 100).Result()
|
||||
if err != nil {
|
||||
slog.Error("failed to scan channel keys", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
fields, err := rb.client.HKeys(ctx, key).Result()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, field := range fields {
|
||||
podID := extractPodID(field)
|
||||
if podID != "" {
|
||||
knownPods[podID] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Check which pods are still alive
|
||||
for podID := range knownPods {
|
||||
exists, err := rb.client.Exists(ctx, podKeyPrefix+podID).Result()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if exists > 0 {
|
||||
alivePods[podID] = true
|
||||
}
|
||||
}
|
||||
|
||||
// Clean up dead pods
|
||||
for podID := range knownPods {
|
||||
if !alivePods[podID] {
|
||||
slog.Info("cleaning up stale pod", "podId", podID)
|
||||
rb.cleanupPod(ctx, podID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (rb *RedisBridge) cleanupPod(ctx context.Context, podID string) {
|
||||
var cursor uint64
|
||||
for {
|
||||
keys, nextCursor, err := rb.client.Scan(ctx, cursor, channelConnsPrefix+"*"+channelConnsSuffix, 100).Result()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
fields, err := rb.client.HGetAll(ctx, key).Result()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
channelID := extractChannelID(key)
|
||||
for field, humanID := range fields {
|
||||
if extractPodID(field) == podID {
|
||||
rb.client.HDel(ctx, key, field)
|
||||
// Check if this humanID is now gone from the channel
|
||||
remaining, _ := rb.client.HVals(ctx, key).Result()
|
||||
if !containsString(remaining, humanID) {
|
||||
rb.publishEvent(ctx, channelID, redisEvent{
|
||||
Type: TypeLeave,
|
||||
HumanID: humanID,
|
||||
PodID: rb.podID,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// --- Helpers ---
|
||||
|
||||
func (rb *RedisBridge) connField(connID string) string {
|
||||
return rb.podID + ":" + connID
|
||||
}
|
||||
|
||||
func (rb *RedisBridge) publishEvent(ctx context.Context, channelID string, event redisEvent) {
|
||||
data, err := json.Marshal(event)
|
||||
if err != nil {
|
||||
slog.Error("failed to marshal event", "error", err)
|
||||
return
|
||||
}
|
||||
if err := rb.client.Publish(ctx, pubsubPrefix+channelID, data).Err(); err != nil {
|
||||
slog.Error("failed to publish event", "channelId", channelID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func channelConnsKey(channelID string) string {
|
||||
return channelConnsPrefix + channelID + channelConnsSuffix
|
||||
}
|
||||
|
||||
func extractChannelID(redisKey string) string {
|
||||
// "pusher:ch:{channelID}:conns" → channelID
|
||||
s := strings.TrimPrefix(redisKey, channelConnsPrefix)
|
||||
s = strings.TrimSuffix(s, channelConnsSuffix)
|
||||
return s
|
||||
}
|
||||
|
||||
func extractPodID(field string) string {
|
||||
// "{podID}:{connID}" → podID
|
||||
parts := strings.SplitN(field, ":", 2)
|
||||
if len(parts) == 2 {
|
||||
return parts[0]
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func containsString(slice []string, s string) bool {
|
||||
for _, v := range slice {
|
||||
if v == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func deduplicateStrings(slice []string) []string {
|
||||
seen := make(map[string]bool, len(slice))
|
||||
result := make([]string, 0, len(slice))
|
||||
for _, s := range slice {
|
||||
if !seen[s] {
|
||||
seen[s] = true
|
||||
result = append(result, s)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package pusher
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
|
||||
"github.com/flowy-live/llink/genproto/llink/pusher"
|
||||
"github.com/flowy-live/llink/internal/auth"
|
||||
"github.com/google/uuid"
|
||||
"nhooyr.io/websocket"
|
||||
)
|
||||
|
||||
// Server handles WebSocket upgrades and gRPC presence queries.
|
||||
type Server struct {
|
||||
pbpusher.UnimplementedPusherServiceServer
|
||||
|
||||
hub *Hub
|
||||
bridge *RedisBridge
|
||||
authSvc auth.AuthService
|
||||
}
|
||||
|
||||
// NewServer creates a new pusher server.
|
||||
func NewServer(hub *Hub, bridge *RedisBridge, authSvc auth.AuthService) *Server {
|
||||
return &Server{
|
||||
hub: hub,
|
||||
bridge: bridge,
|
||||
authSvc: authSvc,
|
||||
}
|
||||
}
|
||||
|
||||
// HandleWebSocket handles the WebSocket upgrade and connection lifecycle.
|
||||
func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
|
||||
// Authenticate via query param (WebSocket upgrade can't use custom headers)
|
||||
token := r.URL.Query().Get("token")
|
||||
if token == "" {
|
||||
http.Error(w, "token required", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
session, err := s.authSvc.GetSession(r.Context(), token)
|
||||
if err != nil {
|
||||
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
// Accept WebSocket upgrade
|
||||
ws, err := websocket.Accept(w, r, &websocket.AcceptOptions{
|
||||
// Allow all origins for now — CORS is handled at the gateway level
|
||||
InsecureSkipVerify: true,
|
||||
})
|
||||
if err != nil {
|
||||
slog.Error("websocket accept failed", "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
connID := uuid.New().String()
|
||||
conn := newConn(connID, session.HumanId, ws)
|
||||
|
||||
slog.Info("websocket connected", "connId", connID, "humanId", session.HumanId)
|
||||
|
||||
ctx, cancel := context.WithCancel(r.Context())
|
||||
defer cancel()
|
||||
|
||||
// WritePump runs in a separate goroutine
|
||||
go conn.WritePump(ctx)
|
||||
|
||||
// ReadPump blocks until the connection closes
|
||||
conn.ReadPump(ctx, s.hub)
|
||||
|
||||
slog.Info("websocket disconnected", "connId", connID, "humanId", session.HumanId)
|
||||
}
|
||||
|
||||
// BulkGetPresence implements the gRPC PusherService.
|
||||
func (s *Server) BulkGetPresence(ctx context.Context, req *pbpusher.BulkGetPresenceRequest) (*pbpusher.BulkGetPresenceResponse, error) {
|
||||
presence, err := s.bridge.GetPresence(ctx, req.ChannelIds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp := &pbpusher.BulkGetPresenceResponse{
|
||||
Presences: make(map[string]*pbpusher.ChannelPresence, len(presence)),
|
||||
}
|
||||
for chID, humanIDs := range presence {
|
||||
resp.Presences[chID] = &pbpusher.ChannelPresence{
|
||||
HumanIds: humanIDs,
|
||||
}
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package pusher
|
||||
|
||||
import "encoding/json"
|
||||
|
||||
// Client → Server message types
|
||||
const (
|
||||
TypeSubscribe = "subscribe"
|
||||
TypeUnsubscribe = "unsubscribe"
|
||||
TypeMessage = "message"
|
||||
)
|
||||
|
||||
// Server → Client message types
|
||||
const (
|
||||
TypeSubscribed = "subscribed"
|
||||
TypeJoin = "join"
|
||||
TypeLeave = "leave"
|
||||
// TypeMessage is reused for server → client messages
|
||||
TypeError = "error"
|
||||
)
|
||||
|
||||
// ClientMessage is a message sent from a WebSocket client to the server.
|
||||
type ClientMessage struct {
|
||||
Type string `json:"type"`
|
||||
Channel string `json:"channel,omitempty"`
|
||||
Payload json.RawMessage `json:"payload,omitempty"`
|
||||
}
|
||||
|
||||
// ServerMessage is a message sent from the server to a WebSocket client.
|
||||
type ServerMessage struct {
|
||||
Type string `json:"type"`
|
||||
Channel string `json:"channel,omitempty"`
|
||||
HumanID string `json:"humanId,omitempty"`
|
||||
Presence []string `json:"presence,omitempty"`
|
||||
Payload json.RawMessage `json:"payload,omitempty"`
|
||||
Message string `json:"message,omitempty"`
|
||||
}
|
||||
Reference in New Issue
Block a user