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. // When channel_ids is empty, returns all connected humanIDs across all channels // under the key "_all" — useful for checking overall online status. func (s *Server) BulkGetPresence(ctx context.Context, req *pbpusher.BulkGetPresenceRequest) (*pbpusher.BulkGetPresenceResponse, error) { // Empty channel_ids → return all connected humans if len(req.ChannelIds) == 0 { humanIDs, err := s.bridge.GetAllConnectedHumanIDs(ctx) if err != nil { return nil, err } return &pbpusher.BulkGetPresenceResponse{ Presences: map[string]*pbpusher.ChannelPresence{ "_all": {HumanIds: humanIDs}, }, }, nil } 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 }