24964376795

Committed 26 Apr 2026 06:52PM UTC coverage: 90.069% (-0.3%) from 90.372%

Build # 24964376795

Build Type

Pull #13

github

Committed by

web-flow

Commit Message

Merge a40c3174a into c7a0902b0

Pull Request Pull Request #13: Docker fixes

Coverage Stats

1506 of 1799 branches covered (83.71%)

Branch coverage included in aggregate %.

955 of 1086 new or added lines in 34 files covered. (87.94%)

7935 of 8683 relevant lines covered (91.39%)

18.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.61

/internal/handler/ws.go

package handler

import (
        "context"
        "encoding/json"
        "log/slog"
        "net/http"
        "time"

        "github.com/coder/websocket"

        "github.com/DigitalTolk/ex/internal/events"
        "github.com/DigitalTolk/ex/internal/middleware"
        "github.com/DigitalTolk/ex/internal/pubsub"
        "github.com/DigitalTolk/ex/internal/service"
)

// inboundMessage is the shape of a client → server WebSocket frame. Only
// "typing" is currently understood; anything else is dropped. We keep
// the shape small and JSON-tolerant — unknown fields are ignored so the
// protocol can grow without breaking older clients.
type inboundMessage struct {
        Type           string `json:"type"`
        ParentID       string `json:"parentID"`
        ParentType     string `json:"parentType"` // "channel" | "conversation"
}

const wsKeepAliveInterval = 30 * time.Second

// WSHandler serves a WebSocket connection for real-time updates.
type WSHandler struct {
        broker      *pubsub.Broker
        chanSvc     *service.ChannelService
        convSvc     *service.ConversationService
        presenceSvc *service.PresenceService
        publisher   service.Publisher
}

// NewWSHandler creates a WSHandler.
func NewWSHandler(broker *pubsub.Broker, chanSvc *service.ChannelService, convSvc *service.ConversationService, presenceSvc *service.PresenceService) *WSHandler {
        return &WSHandler{broker: broker, chanSvc: chanSvc, convSvc: convSvc, presenceSvc: presenceSvc}
}

// SetPublisher wires a publisher for inbound ephemeral events (typing
// indicator). Optional — when nil, inbound typing is dropped.
func (h *WSHandler) SetPublisher(p service.Publisher) { h.publisher = p }

// Connect upgrades the HTTP connection to a WebSocket for the authenticated
// user. Authentication is handled via the "token" query parameter by the auth
// middleware.
func (h *WSHandler) Connect(w http.ResponseWriter, r *http.Request) {
        userID := middleware.UserIDFromContext(r.Context())
        if userID == "" {
                writeError(w, http.StatusUnauthorized, "unauthorized", "authentication required")
                return
        }

        conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
                InsecureSkipVerify: true, // allow any origin in dev; tighten in production
        })
        if err != nil {
                slog.Error("ws: accept", "error", err, "userID", userID)
                return
        }
        defer func() { _ = conn.Close(websocket.StatusNormalClosure, "") }()

        conn.SetReadLimit(4096)

        client := h.broker.RegisterClient(userID)
        defer func() {
                if dropped := client.DropCount(); dropped > 0 {
                        slog.Warn("ws: events dropped", "userID", userID, "dropped", dropped)
                }
                h.broker.UnregisterClient(userID)
                if h.presenceSvc != nil {
                        h.presenceSvc.OnDisconnect(context.Background(), userID)
                }
        }()

        // Subscribe to user's channels and conversations (fetched concurrently).
        var channels []string

        type subResult struct {
                channels []string
                err      error
        }
        chanCh := make(chan subResult, 1)
        convCh := make(chan subResult, 1)

        go func() {
                uc, err := h.chanSvc.ListUserChannels(r.Context(), userID)
                var chs []string
                for _, c := range uc {
                        chs = append(chs, pubsub.ChannelName(c.ChannelID))
                }
                chanCh <- subResult{chs, err}
        }()
        go func() {
                uc, err := h.convSvc.ListUserConversations(r.Context(), userID)
                var chs []string
                for _, c := range uc {
                        chs = append(chs, pubsub.ConversationName(c.ConversationID))
                }
                convCh <- subResult{chs, err}
        }()

        cr := <-chanCh
        if cr.err != nil {
                slog.Error("ws: list channels", "error", cr.err, "userID", userID)
        } else {
                channels = append(channels, cr.channels...)
        }

        cvr := <-convCh
        if cvr.err != nil {
                slog.Error("ws: list conversations", "error", cvr.err, "userID", userID)
        } else {
                channels = append(channels, cvr.channels...)
        }

        // Subscribe to the user's personal channel for direct notifications
        // (e.g. new conversation created).
        channels = append(channels, pubsub.UserChannel(userID))

        // Subscribe to global broadcast channels (channel events, emoji catalog,
        // online presence) so all connected users receive these updates.
        channels = append(channels,
                pubsub.GlobalChannelEvents(),
                pubsub.GlobalEmojiEvents(),
                pubsub.PresenceEvents(),
                pubsub.UserEvents(),
        )

        if len(channels) > 0 {
                h.broker.Subscribe(userID, channels)
        }

        // Mark the user online AFTER subscribing to PresenceEvents so the publish
        // reaches the user's own client (and all other connected clients) instead
        // of being dispatched before any subscriber is wired up.
        if h.presenceSvc != nil {
                h.presenceSvc.OnConnect(r.Context(), userID)
        }

        ctx, cancel := context.WithCancel(r.Context())
        defer cancel()

        // Read loop: parse incoming JSON frames so the typing indicator can
        // fan out via the same pubsub fabric as ordinary events. Unknown
        // frames are silently dropped — the protocol is forward-compatible.
        go func() {
                defer cancel()
                for {
                        _, data, err := conn.Read(ctx)
                        if err != nil {
                                return
                        }
                        h.handleInbound(ctx, userID, data)
                }
        }()

        ticker := time.NewTicker(wsKeepAliveInterval)
        defer ticker.Stop()

        if err := writePing(ctx, conn); err != nil {
                return
        }

        for {
                select {
                case <-ctx.Done():
                        return
                case <-client.Done():
                        return
                case data := <-client.Events:
                        if err := conn.Write(ctx, websocket.MessageText, data); err != nil {
                                return
                        }
                case <-ticker.C:
                        if err := writePing(ctx, conn); err != nil {
                                return
                        }
                }
        }
}

func writePing(ctx context.Context, conn *websocket.Conn) error {
        evt, _ := events.NewEvent(events.EventPing, map[string]int64{"ts": time.Now().UnixMilli()})
        data, _ := json.Marshal(evt)
        return conn.Write(ctx, websocket.MessageText, data)
}

// handleInbound dispatches a single client → server frame. Currently
// only the "typing" event is recognised; everything else is ignored.
func (h *WSHandler) handleInbound(ctx context.Context, userID string, raw []byte) {
        var msg inboundMessage
        if err := json.Unmarshal(raw, &msg); err != nil {
                return
        }
        switch msg.Type {
        case "typing":
                h.publishTyping(ctx, userID, msg)
        }
}

// publishTyping broadcasts a typing event to the parent's pubsub topic
// after verifying the sender is a member. Membership check prevents a
// stranger from spamming a channel they can't read.
func (h *WSHandler) publishTyping(ctx context.Context, userID string, msg inboundMessage) {
        if h.publisher == nil || msg.ParentID == "" {
                return
        }
        var topic string
        switch msg.ParentType {
        case service.ParentChannel:
                if h.chanSvc == nil {
                        return
                }
                // CheckAccess silently no-ops if the membership exists; an error
                // means the user isn't allowed in this channel — drop the event.
                if !h.chanSvc.IsMember(ctx, userID, msg.ParentID) {
                        return
                }
                topic = pubsub.ChannelName(msg.ParentID)
        case service.ParentConversation:
                if h.convSvc == nil {
                        return
                }
                if !h.convSvc.IsParticipant(ctx, userID, msg.ParentID) {
                        return
                }
                topic = pubsub.ConversationName(msg.ParentID)
        default:
                return
        }
        evt, err := events.NewEvent(events.EventTyping, map[string]any{
                "userID":     userID,
                "parentID":   msg.ParentID,
                "parentType": msg.ParentType,
        })
        if err != nil {
                return
        }
        _ = h.publisher.Publish(ctx, topic, evt)
}

1	package handler
2
3	import (
4	"context"
5	"encoding/json"
6	"log/slog"
7	"net/http"
8	"time"
9
10	"github.com/coder/websocket"
11
12	"github.com/DigitalTolk/ex/internal/events"
13	"github.com/DigitalTolk/ex/internal/middleware"
14	"github.com/DigitalTolk/ex/internal/pubsub"
15	"github.com/DigitalTolk/ex/internal/service"
16	)
17
18	// inboundMessage is the shape of a client → server WebSocket frame. Only
19	// "typing" is currently understood; anything else is dropped. We keep
20	// the shape small and JSON-tolerant — unknown fields are ignored so the
21	// protocol can grow without breaking older clients.
22	type inboundMessage struct {
23	Type string `json:"type"`
24	ParentID string `json:"parentID"`
25	ParentType string `json:"parentType"` // "channel" \| "conversation"
26	}
27
28	const wsKeepAliveInterval = 30 * time.Second
29
30	// WSHandler serves a WebSocket connection for real-time updates.
31	type WSHandler struct {
32	broker *pubsub.Broker
33	chanSvc *service.ChannelService
34	convSvc *service.ConversationService
35	presenceSvc *service.PresenceService
36	publisher service.Publisher
37	}
38
39	// NewWSHandler creates a WSHandler.
40	func NewWSHandler(broker pubsub.Broker, chanSvc service.ChannelService, convSvc service.ConversationService, presenceSvc service.PresenceService) *WSHandler {	2✔
41	return &WSHandler{broker: broker, chanSvc: chanSvc, convSvc: convSvc, presenceSvc: presenceSvc}	2✔
42	}	2✔
43
44	// SetPublisher wires a publisher for inbound ephemeral events (typing
45	// indicator). Optional — when nil, inbound typing is dropped.
46	func (h *WSHandler) SetPublisher(p service.Publisher) { h.publisher = p }	9✔
47
48	// Connect upgrades the HTTP connection to a WebSocket for the authenticated
49	// user. Authentication is handled via the "token" query parameter by the auth
50	// middleware.
51	func (h WSHandler) Connect(w http.ResponseWriter, r http.Request) {	2✔
52	userID := middleware.UserIDFromContext(r.Context())	2✔
53	if userID == "" {	3✔
54	writeError(w, http.StatusUnauthorized, "unauthorized", "authentication required")	1✔
55	return	1✔
56	}	1✔
57
58	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{	1✔
59	InsecureSkipVerify: true, // allow any origin in dev; tighten in production	1✔
60	})	1✔
61	if err != nil {	1✔
62	slog.Error("ws: accept", "error", err, "userID", userID)	×
63	return	×
64	}	×
65	defer func() { _ = conn.Close(websocket.StatusNormalClosure, "") }()	2✔
66
67	conn.SetReadLimit(4096)	1✔
68		1✔
69	client := h.broker.RegisterClient(userID)	1✔
70	defer func() {	2✔
71	if dropped := client.DropCount(); dropped > 0 {	1✔
72	slog.Warn("ws: events dropped", "userID", userID, "dropped", dropped)	×
73	}	×
74	h.broker.UnregisterClient(userID)	1✔
75	if h.presenceSvc != nil {	2✔
76	h.presenceSvc.OnDisconnect(context.Background(), userID)	1✔
77	}	1✔
78	}()
79
80	// Subscribe to user's channels and conversations (fetched concurrently).
81	var channels []string	1✔
82		1✔
83	type subResult struct {	1✔
84	channels []string	1✔
85	err error	1✔
86	}	1✔
87	chanCh := make(chan subResult, 1)	1✔
88	convCh := make(chan subResult, 1)	1✔
89		1✔
90	go func() {	2✔
91	uc, err := h.chanSvc.ListUserChannels(r.Context(), userID)	1✔
92	var chs []string	1✔
93	for _, c := range uc {	1✔
94	chs = append(chs, pubsub.ChannelName(c.ChannelID))	×
95	}	×
96	chanCh <- subResult{chs, err}	1✔
97	}()
98	go func() {	2✔
99	uc, err := h.convSvc.ListUserConversations(r.Context(), userID)	1✔
100	var chs []string	1✔
101	for _, c := range uc {	1✔
102	chs = append(chs, pubsub.ConversationName(c.ConversationID))	×
103	}	×
104	convCh <- subResult{chs, err}	1✔
105	}()
106
107	cr := <-chanCh	1✔
108	if cr.err != nil {	1✔
109	slog.Error("ws: list channels", "error", cr.err, "userID", userID)	×
110	} else {	1✔
111	channels = append(channels, cr.channels...)	1✔
112	}	1✔
113
114	cvr := <-convCh	1✔
115	if cvr.err != nil {	1✔
116	slog.Error("ws: list conversations", "error", cvr.err, "userID", userID)	×
117	} else {	1✔
118	channels = append(channels, cvr.channels...)	1✔
119	}	1✔
120
121	// Subscribe to the user's personal channel for direct notifications
122	// (e.g. new conversation created).
123	channels = append(channels, pubsub.UserChannel(userID))	1✔
124		1✔
125	// Subscribe to global broadcast channels (channel events, emoji catalog,	1✔
126	// online presence) so all connected users receive these updates.	1✔
127	channels = append(channels,	1✔
128	pubsub.GlobalChannelEvents(),	1✔
129	pubsub.GlobalEmojiEvents(),	1✔
130	pubsub.PresenceEvents(),	1✔
131	pubsub.UserEvents(),	1✔
132	)	1✔
133		1✔
134	if len(channels) > 0 {	2✔
135	h.broker.Subscribe(userID, channels)	1✔
136	}	1✔
137
138	// Mark the user online AFTER subscribing to PresenceEvents so the publish
139	// reaches the user's own client (and all other connected clients) instead
140	// of being dispatched before any subscriber is wired up.
141	if h.presenceSvc != nil {	2✔
142	h.presenceSvc.OnConnect(r.Context(), userID)	1✔
143	}	1✔
144
145	ctx, cancel := context.WithCancel(r.Context())	1✔
146	defer cancel()	1✔
147		1✔
148	// Read loop: parse incoming JSON frames so the typing indicator can	1✔
149	// fan out via the same pubsub fabric as ordinary events. Unknown	1✔
150	// frames are silently dropped — the protocol is forward-compatible.	1✔
151	go func() {	2✔
152	defer cancel()	1✔
153	for {	2✔
154	_, data, err := conn.Read(ctx)	1✔
155	if err != nil {	2✔
156	return	1✔
157	}	1✔
NEW 158	h.handleInbound(ctx, userID, data)	×
159	}
160	}()
161
162	ticker := time.NewTicker(wsKeepAliveInterval)	1✔
163	defer ticker.Stop()	1✔
164		1✔
165	if err := writePing(ctx, conn); err != nil {	1✔
166	return	×
167	}	×
168
169	for {	3✔
170	select {	2✔
171	case <-ctx.Done():	1✔
172	return	1✔
173	case <-client.Done():	×
174	return	×
175	case data := <-client.Events:	1✔
176	if err := conn.Write(ctx, websocket.MessageText, data); err != nil {	1✔
177	return	×
178	}	×
179	case <-ticker.C:	×
180	if err := writePing(ctx, conn); err != nil {	×
181	return	×
182	}	×
183	}
184	}
185	}
186
187	func writePing(ctx context.Context, conn *websocket.Conn) error {	1✔
188	evt, _ := events.NewEvent(events.EventPing, map[string]int64{"ts": time.Now().UnixMilli()})	1✔
189	data, _ := json.Marshal(evt)	1✔
190	return conn.Write(ctx, websocket.MessageText, data)	1✔
191	}	1✔
192
193	// handleInbound dispatches a single client → server frame. Currently
194	// only the "typing" event is recognised; everything else is ignored.
195	func (h *WSHandler) handleInbound(ctx context.Context, userID string, raw []byte) {	9✔
196	var msg inboundMessage	9✔
197	if err := json.Unmarshal(raw, &msg); err != nil {	10✔
198	return	1✔
199	}	1✔
200	switch msg.Type {	8✔
201	case "typing":	7✔
202	h.publishTyping(ctx, userID, msg)	7✔
203	}
204	}
205
206	// publishTyping broadcasts a typing event to the parent's pubsub topic
207	// after verifying the sender is a member. Membership check prevents a
208	// stranger from spamming a channel they can't read.
209	func (h *WSHandler) publishTyping(ctx context.Context, userID string, msg inboundMessage) {	7✔
210	if h.publisher == nil \|\| msg.ParentID == "" {	9✔
211	return	2✔
212	}	2✔
213	var topic string	5✔
214	switch msg.ParentType {	5✔
215	case service.ParentChannel:	2✔
216	if h.chanSvc == nil {	2✔
NEW 217	return	×
NEW 218	}	×
219	// CheckAccess silently no-ops if the membership exists; an error
220	// means the user isn't allowed in this channel — drop the event.
221	if !h.chanSvc.IsMember(ctx, userID, msg.ParentID) {	3✔
222	return	1✔
223	}	1✔
224	topic = pubsub.ChannelName(msg.ParentID)	1✔
225	case service.ParentConversation:	2✔
226	if h.convSvc == nil {	2✔
NEW 227	return	×
NEW 228	}	×
229	if !h.convSvc.IsParticipant(ctx, userID, msg.ParentID) {	3✔
230	return	1✔
231	}	1✔
232	topic = pubsub.ConversationName(msg.ParentID)	1✔
233	default:	1✔
234	return	1✔
235	}
236	evt, err := events.NewEvent(events.EventTyping, map[string]any{	2✔
237	"userID": userID,	2✔
238	"parentID": msg.ParentID,	2✔
239	"parentType": msg.ParentType,	2✔
240	})	2✔
241	if err != nil {	2✔
NEW 242	return	×
NEW 243	}	×
244	_ = h.publisher.Publish(ctx, topic, evt)	2✔
245	}

DigitalTolk / ex / 24964376795

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous