diff --git a/routers/web/websocket/websocket.go b/routers/web/websocket/websocket.go index cc9c75bc3c3..553ffa7191d 100644 --- a/routers/web/websocket/websocket.go +++ b/routers/web/websocket/websocket.go @@ -9,7 +9,6 @@ import ( "time" "gitea.dev/modules/graceful" - "gitea.dev/modules/json" "gitea.dev/modules/log" "gitea.dev/services/context" "gitea.dev/services/pubsub" @@ -30,23 +29,6 @@ const ( closeCodeUnauthenticated gitea_ws.StatusCode = 3000 ) -// filterLogout forwards a session-free logout only to the targeted connection -// (its own session, or every session when SessionID is empty) and drops it for -// the rest. Non-logout messages pass through untouched. -func filterLogout(eventType string, eventDataBytes []byte, connSessionID string) []byte { - if eventType != websocket_service.EventLogout { - return eventDataBytes - } - var lm websocket_service.UserEventMessage[websocket_service.LogoutEventData] - if err := json.Unmarshal(eventDataBytes, &lm); err != nil { - return eventDataBytes - } - if lm.EventData.SessionID == "" || lm.EventData.SessionID == connSessionID { - return []byte(`{"eventType":"logout"}`) - } - return nil -} - func Serve(ctx *context.Context) { // Answer plain GETs (health checks, crawlers) here; letting Accept reject them // would log an error per request. Same reply it would have sent. @@ -57,6 +39,14 @@ func Serve(ctx *context.Context) { return } + if !ctx.IsSigned { + rejectUnauthenticated(ctx) + return + } + + ch, cancel := pubsub.DefaultBroker.Subscribe(pubsub.UserTopic(ctx.Doer.ID)) // before the snapshot, so no change falls between them + defer cancel() + conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, nil) if err != nil { log.Error("websocket: accept failed: %v", err) @@ -64,14 +54,7 @@ func Serve(ctx *context.Context) { } defer conn.CloseNow() //nolint:errcheck // best-effort close - if !ctx.IsSigned { - _ = conn.Close(closeCodeUnauthenticated, "unauthenticated") - return - } - sessionID := ctx.Session.ID() - ch, cancel := pubsub.DefaultBroker.Subscribe(pubsub.UserTopic(ctx.Doer.ID)) - defer cancel() // Ping requires a concurrent reader to observe the pong frame; CloseRead // spawns one and cancels its context when the peer goes away. @@ -80,6 +63,30 @@ func Serve(ctx *context.Context) { pingTicker := time.NewTicker(pingInterval) defer pingTicker.Stop() + send := func(brokerPayload []byte) error { + eventDataBytes := websocket_service.EventForSession(brokerPayload, sessionID) + if eventDataBytes == nil { + return nil + } + writeCtx, cancelWrite := gocontext.WithTimeout(wsCtx, writeTimeout) // a stalled peer must not starve the ping ticker + defer cancelWrite() + err := conn.Write(writeCtx, gitea_ws.MessageText, eventDataBytes) + if err != nil { + log.Trace("websocket: write failed: %v", err) + } + return err + } + + var backlog [][]byte // published before the snapshot read, so it must not follow the snapshot + for len(ch) > 0 { + backlog = append(backlog, <-ch) + } + for _, brokerPayload := range append(backlog, websocket_service.UserData(wsCtx, ctx.Doer)...) { + if err := send(brokerPayload); err != nil { + return + } + } + for { select { case <-wsCtx.Done(): @@ -98,20 +105,19 @@ func Serve(ctx *context.Context) { if !ok { return } - eventType, eventDataBytes := websocket_service.ExtractUserEventMessage(brokerPayload) - eventDataBytes = filterLogout(eventType, eventDataBytes, sessionID) - if eventDataBytes == nil { - continue - } - // Bound the write so a stalled/slow peer can't block this goroutine - // indefinitely and starve the ping ticker. - writeCtx, cancelWrite := gocontext.WithTimeout(wsCtx, writeTimeout) - err := conn.Write(writeCtx, gitea_ws.MessageText, eventDataBytes) - cancelWrite() - if err != nil { - log.Trace("websocket: write failed: %v", err) + if err := send(brokerPayload); err != nil { return } } } } + +// rejectUnauthenticated upgrades only to close, so clients stop reconnecting. +func rejectUnauthenticated(ctx *context.Context) { + conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, nil) + if err != nil { + log.Error("websocket: accept failed: %v", err) + return + } + _ = conn.Close(closeCodeUnauthenticated, "unauthenticated") +} diff --git a/routers/web/websocket/websocket_test.go b/routers/web/websocket/websocket_test.go deleted file mode 100644 index 421fb2f7633..00000000000 --- a/routers/web/websocket/websocket_test.go +++ /dev/null @@ -1,60 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package websocket - -import ( - "testing" - - "gitea.dev/services/websocket" - - "github.com/stretchr/testify/assert" -) - -func TestFilterLogout(t *testing.T) { - cases := []struct { - name string - brokerMsg []byte - connSessID string - want []byte // expected payload forwarded to the client - }{ - { - name: "originating session gets a session-free logout", - brokerMsg: websocket.MakeUserEventMessage("logout", websocket.LogoutEventData{SessionID: "sess-A"}), - connSessID: "sess-A", - want: []byte(`{"eventType":"logout"}`), - }, - { - name: "other session is dropped", - brokerMsg: websocket.MakeUserEventMessage("logout", websocket.LogoutEventData{SessionID: "sess-A"}), - connSessID: "sess-B", - want: nil, - }, - { - name: "empty sessionID reaches every session", - brokerMsg: websocket.MakeUserEventMessage("logout", nil), - connSessID: "sess-A", - want: []byte(`{"eventType":"logout"}`), - }, - { - name: "non-logout message passes through unchanged", - brokerMsg: websocket.MakeUserEventMessage("other", map[string]any{"k": "v"}), - connSessID: "sess-A", - want: []byte(`{"eventType":"other","eventData":{"k":"v"}}`), - }, - { - name: "malformed JSON with logout marker passes through unchanged", - brokerMsg: []byte("any type\nany data"), - connSessID: "sess-A", - want: []byte("any data"), - }, - } - - for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - eventType, eventData := websocket.ExtractUserEventMessage(tc.brokerMsg) - out := filterLogout(eventType, eventData, tc.connSessID) - assert.Equal(t, tc.want, out) - }) - } -} diff --git a/services/pubsub/broker_test.go b/services/pubsub/broker_test.go index d2fc36e9dc1..b09d70ed628 100644 --- a/services/pubsub/broker_test.go +++ b/services/pubsub/broker_test.go @@ -29,7 +29,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur b := newBroker(t) ch, cancel := b.Subscribe(t.Name()) defer cancel() - waitSubscribed(t, b, t.Name()) b.Publish(t.Name(), []byte("hello")) assert.Equal(t, []byte("hello"), recvWithin(t, ch, recvTimeout)) @@ -44,7 +43,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur defer cancel() channels[i] = ch } - waitSubscribed(t, b, t.Name()) // all local subscribers share one Redis SUBSCRIBE b.Publish(t.Name(), []byte("broadcast")) for i, ch := range channels { @@ -59,8 +57,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur defer cancelA() chB, cancelB := b.Subscribe(topicB) defer cancelB() - waitSubscribed(t, b, topicA) - waitSubscribed(t, b, topicB) b.Publish(topicA, []byte("only-a")) assert.Equal(t, []byte("only-a"), recvWithin(t, chA, recvTimeout)) @@ -100,7 +96,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur defer cancelSlow() fast, cancelFast := b.Subscribe(t.Name()) defer cancelFast() - waitSubscribed(t, b, t.Name()) // Drain fast concurrently so it keeps up while slow's buffer fills. got := make(chan struct{}, 1) @@ -147,23 +142,6 @@ func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Dur }) } -// waitSubscribed blocks until Redis registers the subscription, since -// RedisBroker.Subscribe returns before the server acks SUBSCRIBE and a publish -// right after it would race it. Topics are per-scenario, so any subscriber is -// this one. No-op for MemoryBroker, which registers synchronously. -func waitSubscribed(t *testing.T, b Broker, topic string) { - t.Helper() - rb, ok := b.(*RedisBroker) - if !ok { - return - } - channel := redisChannelForTopic(topic) - require.Eventually(t, func() bool { - res, err := rb.client.PubSubNumSub(t.Context(), channel).Result() - return err == nil && res[channel] > 0 - }, 5*time.Second, 10*time.Millisecond, "redis did not register a subscriber on %q", topic) -} - // recvWithin returns the next message or fails if none arrives before timeout. func recvWithin(t *testing.T, ch <-chan []byte, timeout time.Duration) []byte { t.Helper() diff --git a/services/pubsub/redis.go b/services/pubsub/redis.go index eb7798298e2..f1b90368ddd 100644 --- a/services/pubsub/redis.go +++ b/services/pubsub/redis.go @@ -17,10 +17,11 @@ import ( ) const ( - redisPingTimeout = 3 * time.Second - redisPingRetries = 10 - redisPingRetryDelay = time.Second - redisPublishTimeout = 2 * time.Second + redisPingTimeout = 3 * time.Second + redisPingRetries = 10 + redisPingRetryDelay = time.Second + redisPublishTimeout = 2 * time.Second + redisSubscribeTimeout = 2 * time.Second ) // RedisBroker fans out across processes via Redis pub/sub. Each topic is @@ -94,12 +95,15 @@ func (b *RedisBroker) Subscribe(topic string) (<-chan []byte, func()) { // other Subscribe/cancel calls aren't blocked on the network round-trip. // graceful.ShutdownContext so the reader loop dies cleanly on Gitea // shutdown even if every local subscriber has already cancelled. - // readLoop consumes the SUBSCRIBE ack; don't wait for it here, that would put - // a Redis round-trip in the WebSocket handshake to close a sub-millisecond - // window in which a publish is missed - harmless, since a client receives - // nothing at all until its handshake completes. ctx, cancelCtx := context.WithCancel(graceful.GetManager().ShutdownContext()) ps := b.client.Subscribe(ctx, redisChannelForTopic(topic)) + if _, err := ps.ReceiveTimeout(ctx, redisSubscribeTimeout); err != nil { // the ack, so state read afterwards misses no publish + log.Error("pubsub redis: subscribe to %q: %v", topic, err) + cancelCtx() + _ = ps.Close() + sub.close() + return sub.ch, func() {} + } b.mu.Lock() if existing, exists := b.topics[topic]; exists { // Another goroutine won the create race; merge into theirs and discard ours. diff --git a/services/pubsub/redis_test.go b/services/pubsub/redis_test.go index f1ad3a7512e..7b9855c5995 100644 --- a/services/pubsub/redis_test.go +++ b/services/pubsub/redis_test.go @@ -59,7 +59,6 @@ func TestRedisBroker(t *testing.T) { ch, cancel := subscriber.Subscribe(t.Name()) defer cancel() - waitSubscribed(t, subscriber, t.Name()) publisher.Publish(t.Name(), []byte("cross-process")) diff --git a/services/uinotification/notify.go b/services/uinotification/notify.go index b0aa8c1ac32..1823da09e3e 100644 --- a/services/uinotification/notify.go +++ b/services/uinotification/notify.go @@ -55,7 +55,10 @@ func handler(items ...issueNotificationOpts) []issueNotificationOpts { for _, opts := range items { notifiedIDs, err := activities_model.CreateOrUpdateIssueNotifications(ctx, opts.IssueID, opts.CommentID, opts.NotificationAuthorID, opts.ReceiverID) if err != nil { - log.Error("Was unable to create issue notification: %v", err) + // the issue can be gone by now, e.g. its repository was deleted + if !issues_model.IsErrIssueNotExist(err) { + log.Error("Unable to create issue notification: %v", err) + } continue } for _, userID := range notifiedIDs { diff --git a/services/websocket/events.go b/services/websocket/events.go deleted file mode 100644 index c7c3ec052eb..00000000000 --- a/services/websocket/events.go +++ /dev/null @@ -1,53 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package websocket - -import ( - "bytes" - - "gitea.dev/modules/json" - "gitea.dev/modules/setting" - "gitea.dev/services/pubsub" -) - -// Wire contract with web_src/js/user-events.sharedworker.ts — keep in sync. -const ( - EventNotificationCount = "notification-count" - EventStopwatches = "stopwatches" - EventLogout = "logout" -) - -type UserEventMessage[T any] struct { - EventType string `json:"eventType"` - EventData T `json:"eventData"` -} - -func publishUserEvent(userID int64, eventType string, eventData any) { - if pubsub.DefaultBroker == nil { - return - } - b := MakeUserEventMessage(eventType, eventData) - if b == nil { - return - } - pubsub.DefaultBroker.Publish(pubsub.UserTopic(userID), b) -} - -func MakeUserEventMessage(eventType string, eventData any) []byte { - buf := &bytes.Buffer{} - buf.WriteString(eventType) - buf.WriteByte('\n') - err := json.MarshalWriteRelaxed(buf, &UserEventMessage[any]{EventType: eventType, EventData: eventData}) - payloadBytes := bytes.TrimSuffix(buf.Bytes(), []byte("\n")) // json v1 adds extra "\n" but we don't want it - if err != nil { - setting.PanicInDevOrTesting("websocket: marshal event: %v", err) - return nil - } - return payloadBytes -} - -func ExtractUserEventMessage(b []byte) (string, []byte) { - eventType, eventDataBytes, _ := bytes.Cut(b, []byte("\n")) - return string(eventType), eventDataBytes -} diff --git a/services/websocket/logout.go b/services/websocket/logout.go new file mode 100644 index 00000000000..90e65d9de47 --- /dev/null +++ b/services/websocket/logout.go @@ -0,0 +1,10 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +const eventLogout = "logout" // keep in sync with web_src/js/types.ts + +func PublishLogout(userID int64, sessionID string) { + publishUserEvent(userID, func() []byte { return makeEventMessage(eventLogout, sessionID, nil) }) +} diff --git a/services/websocket/logout_publisher.go b/services/websocket/logout_publisher.go deleted file mode 100644 index 58c0bb3962a..00000000000 --- a/services/websocket/logout_publisher.go +++ /dev/null @@ -1,12 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package websocket - -type LogoutEventData struct { - SessionID string `json:"sessionID,omitempty"` -} - -func PublishLogout(userID int64, sessionID string) { - publishUserEvent(userID, EventLogout, LogoutEventData{SessionID: sessionID}) -} diff --git a/services/websocket/notification_notifier.go b/services/websocket/notification.go similarity index 59% rename from services/websocket/notification_notifier.go rename to services/websocket/notification.go index 7013dc2b637..f82ce2be5e8 100644 --- a/services/websocket/notification_notifier.go +++ b/services/websocket/notification.go @@ -9,31 +9,24 @@ import ( activities_model "gitea.dev/models/activities" "gitea.dev/models/db" "gitea.dev/modules/log" - notify_service "gitea.dev/services/notify" - "gitea.dev/services/pubsub" ) -type notificationCountEventData struct { - Count int64 `json:"count"` -} - -type wsNotifier struct { - notify_service.NullNotifier -} - -var _ notify_service.Notifier = &wsNotifier{} +const eventNotificationCount = "notification-count" // keep in sync with web_src/js/types.ts func (n *wsNotifier) NotificationCountChange(ctx context.Context, userID int64) { - if !pubsub.DefaultBroker.HasTopicSubscribers(pubsub.UserTopic(userID)) { - return - } + publishUserEvent(userID, func() []byte { return notificationCountEvent(ctx, userID) }) +} + +func notificationCountEvent(ctx context.Context, userID int64) []byte { count, err := db.Count[activities_model.Notification](ctx, activities_model.FindNotificationOptions{ UserID: userID, Status: []activities_model.NotificationStatus{activities_model.NotificationStatusUnread}, }) if err != nil { log.Error("websocket: count notifications for user %d: %v", userID, err) - return + return nil } - publishUserEvent(userID, EventNotificationCount, notificationCountEventData{Count: count}) + return makeEventMessage(eventNotificationCount, "", struct { + Count int64 `json:"count"` + }{count}) } diff --git a/services/websocket/notifier.go b/services/websocket/notifier.go deleted file mode 100644 index ec09c4f0d2f..00000000000 --- a/services/websocket/notifier.go +++ /dev/null @@ -1,18 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package websocket - -import ( - notify_service "gitea.dev/services/notify" - "gitea.dev/services/pubsub" -) - -func Init() error { - // the pubsub broker must be ready before the notifier starts publishing to it - if err := pubsub.Init(); err != nil { - return err - } - notify_service.RegisterNotifier(&wsNotifier{}) - return nil -} diff --git a/services/websocket/stopwatch_notifier.go b/services/websocket/stopwatch.go similarity index 63% rename from services/websocket/stopwatch_notifier.go rename to services/websocket/stopwatch.go index 5d81cbd139e..b9f8d83ad42 100644 --- a/services/websocket/stopwatch_notifier.go +++ b/services/websocket/stopwatch.go @@ -10,28 +10,32 @@ import ( issues_model "gitea.dev/models/issues" user_model "gitea.dev/models/user" "gitea.dev/modules/log" + "gitea.dev/modules/setting" "gitea.dev/modules/util" "gitea.dev/services/convert" - "gitea.dev/services/pubsub" ) -func (n *wsNotifier) StopwatchChanged(ctx context.Context, user *user_model.User) { - if !pubsub.DefaultBroker.HasTopicSubscribers(pubsub.UserTopic(user.ID)) { - return - } +const eventStopwatches = "stopwatches" // keep in sync with web_src/js/types.ts +func (n *wsNotifier) StopwatchChanged(ctx context.Context, user *user_model.User) { + publishUserEvent(user.ID, func() []byte { return stopwatchesEvent(ctx, user) }) +} + +func stopwatchesEvent(ctx context.Context, user *user_model.User) []byte { + if !setting.Service.EnableTimetracking { + return nil + } sws, err := issues_model.GetUserStopwatches(ctx, user.ID, db.ListOptions{}) if err != nil { log.Error("websocket: GetUserStopwatches %d: %v", user.ID, err) - return + return nil } - apiStopWatches, err := convert.ToStopWatches(ctx, user, sws) if err != nil { if !issues_model.IsErrIssueNotExist(err) { log.Error("websocket: ToStopWatches: %v", err) } - return + return nil } - publishUserEvent(user.ID, EventStopwatches, util.SliceNilAsEmpty(apiStopWatches)) + return makeEventMessage(eventStopwatches, "", util.SliceNilAsEmpty(apiStopWatches)) } diff --git a/services/websocket/websocket.go b/services/websocket/websocket.go new file mode 100644 index 00000000000..43048bf7731 --- /dev/null +++ b/services/websocket/websocket.go @@ -0,0 +1,71 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +import ( + "bytes" + "context" + + user_model "gitea.dev/models/user" + "gitea.dev/modules/json" + "gitea.dev/modules/setting" + notify_service "gitea.dev/services/notify" + "gitea.dev/services/pubsub" +) + +type wsNotifier struct { + notify_service.NullNotifier +} + +var _ notify_service.Notifier = &wsNotifier{} + +func Init() error { + // the pubsub broker must be ready before the notifier starts publishing to it + if err := pubsub.Init(); err != nil { + return err + } + notify_service.RegisterNotifier(&wsNotifier{}) + return nil +} + +func UserData(ctx context.Context, user *user_model.User) (data [][]byte) { + for _, event := range [][]byte{notificationCountEvent(ctx, user.ID), stopwatchesEvent(ctx, user)} { + if event != nil { + data = append(data, event) + } + } + return data +} + +func EventForSession(brokerPayload []byte, sessionID string) []byte { + target, eventData, _ := bytes.Cut(brokerPayload, []byte("\n")) + if len(target) > 0 && string(target) != sessionID { + return nil + } + return eventData +} + +// makeEventMessage prefixes the client JSON with its target session, empty for all sessions. +func makeEventMessage(eventType, targetSession string, eventData any) []byte { + buf := bytes.NewBufferString(targetSession + "\n") + err := json.MarshalWriteRelaxed(buf, &struct { + EventType string `json:"eventType"` + EventData any `json:"eventData,omitzero"` + }{eventType, eventData}) + if err != nil { + setting.PanicInDevOrTesting("websocket: marshal event: %v", err) + return nil + } + return buf.Bytes() +} + +func publishUserEvent(userID int64, build func() []byte) { + topic := pubsub.UserTopic(userID) + if !pubsub.DefaultBroker.HasTopicSubscribers(topic) { + return + } + if event := build(); event != nil { + pubsub.DefaultBroker.Publish(topic, event) + } +} diff --git a/services/websocket/websocket_test.go b/services/websocket/websocket_test.go new file mode 100644 index 00000000000..5d7d367a637 --- /dev/null +++ b/services/websocket/websocket_test.go @@ -0,0 +1,19 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestEventForSessionTargetsAndKeepsEmptyData(t *testing.T) { + targeted := makeEventMessage("targeted-event", "sess-A", nil) + assert.JSONEq(t, `{"eventType":"targeted-event"}`, string(EventForSession(targeted, "sess-A"))) + assert.Nil(t, EventForSession(targeted, "sess-B")) + + broadcast := makeEventMessage("list-event", "", []int{}) + assert.JSONEq(t, `{"eventType":"list-event","eventData":[]}`, string(EventForSession(broadcast, "sess-A"))) +} diff --git a/tests/e2e/events.test.ts b/tests/e2e/events.test.ts index d48bcccc3bc..026e489ebfd 100644 --- a/tests/e2e/events.test.ts +++ b/tests/e2e/events.test.ts @@ -23,7 +23,7 @@ test.describe('events', () => { await expect(page.locator('html[data-user-events-connected]')).toBeAttached(); await apiCreateIssue(request, {owner, repo: repoName, title: 'events-notif', headers: apiUserHeaders(commenter)}); - await expect(badge).toBeVisible(); + await expect(badge).toHaveText('1'); }); test('stopwatch appears via real-time push and stops from its popup', async ({page, request}) => { @@ -51,6 +51,7 @@ test.describe('events', () => { await expect(stopwatch).toBeVisible(); await stopwatch.click(); + await expect(page.locator('.stopwatch-link')).toHaveAttribute('href', `/${name}/${name}/issues/1`); await page.getByRole('button', {name: 'Stop Timer'}).click(); await expect(stopwatch).toBeHidden(); }); diff --git a/tests/integration/websocket_test.go b/tests/integration/websocket_test.go new file mode 100644 index 00000000000..fed6db20605 --- /dev/null +++ b/tests/integration/websocket_test.go @@ -0,0 +1,79 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package integration + +import ( + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + + activities_model "gitea.dev/models/activities" + "gitea.dev/models/unittest" + "gitea.dev/modules/json" + "gitea.dev/modules/setting" + "gitea.dev/modules/test" + "gitea.dev/services/pubsub" + "gitea.dev/tests" + + "github.com/coder/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type staleBacklogBroker struct { + pubsub.Broker +} + +func (b *staleBacklogBroker) Subscribe(topic string) (<-chan []byte, func()) { + ch, cancel := b.Broker.Subscribe(topic) + for range 9 { + b.Publish(topic, countEvent(-1)) + } + return ch, cancel +} + +func countEvent(count int) []byte { + return fmt.Appendf(nil, "\n"+`{"eventType":"notification-count","eventData":{"count":%d}}`, count) +} + +func TestWebSocketSnapshotFollowsBacklog(t *testing.T) { + defer tests.PrepareTestEnv(t)() + defer test.MockVariableValue(&pubsub.DefaultBroker, pubsub.Broker(&staleBacklogBroker{pubsub.NewMemoryBroker()}))() + session := loginUser(t, "user2") + server := httptest.NewServer(testWebRoutes) + defer server.Close() + + header := http.Header{} + header.Set("Cookie", session.GetRawCookie(setting.SessionConfig.CookieName).String()) + conn, _, err := websocket.Dial(t.Context(), strings.Replace(server.URL, "http", "ws", 1)+"/-/ws", &websocket.DialOptions{HTTPHeader: header}) //nolint:bodyclose // coder/websocket owns the body, it is nil after the upgrade + require.NoError(t, err) + defer conn.CloseNow() + + const sentinelCount = -2 + lastCount := 0 + for { + _, payload, err := conn.Read(t.Context()) + require.NoError(t, err) + var event struct { + EventType string `json:"eventType"` + EventData json.Value `json:"eventData"` + } + require.NoError(t, json.Unmarshal(payload, &event)) + if event.EventType == "stopwatches" { + pubsub.DefaultBroker.Publish(pubsub.UserTopic(2), countEvent(sentinelCount)) + continue + } + var data struct { + Count int `json:"count"` + } + require.NoError(t, json.Unmarshal(event.EventData, &data)) + if data.Count == sentinelCount { + break + } + lastCount = data.Count + } + assert.Equal(t, unittest.GetCount(t, &activities_model.Notification{UserID: 2, Status: activities_model.NotificationStatusUnread}), lastCount) +} diff --git a/web_src/js/features/notification.ts b/web_src/js/features/notification.ts index f18567e62ba..274bc66e670 100644 --- a/web_src/js/features/notification.ts +++ b/web_src/js/features/notification.ts @@ -26,8 +26,7 @@ export function initNotificationCount() { let pollerStarted = false; onUserEvent('notification-count', (msg) => { receiveUpdateCount(msg.eventData.count) }); // no await - // On each (re)connect, reconcile the count from the server to recover any push dropped during the connect gap. - onUserEvent('worker-connected', () => { updateNotificationCount(); updateNotificationTable() }); // no await + onUserEvent('worker-attached', () => { updateNotificationCount(); updateNotificationTable() }); // no await onUserEvent('worker-unavailable', () => { if (pollerStarted) return; pollerStarted = true; diff --git a/web_src/js/features/stopwatch.ts b/web_src/js/features/stopwatch.ts index 2651856420d..43498a12210 100644 --- a/web_src/js/features/stopwatch.ts +++ b/web_src/js/features/stopwatch.ts @@ -76,8 +76,7 @@ export const initStopwatch = () => registerGlobalInitFunc('initActiveStopwatchNo let pollerStarted = false; onUserEvent('stopwatches', (msg) => updateStopwatchData(msg.eventData)); - // On each (re)connect, reconcile stopwatch state from the server to recover any push dropped during the connect gap. - onUserEvent('worker-connected', () => { updateStopwatch() }); // no await + onUserEvent('worker-attached', () => { updateStopwatch() }); // no await onUserEvent('worker-unavailable', () => { if (pollerStarted) return; pollerStarted = true; diff --git a/web_src/js/types.ts b/web_src/js/types.ts index 5793cfd2372..833ef089aaf 100644 --- a/web_src/js/types.ts +++ b/web_src/js/types.ts @@ -87,7 +87,7 @@ export type StopwatchData = { seconds: number, }; -// keep in sync with services/websocket/events.go +// keep in sync with services/websocket export type ServerUserEventMessage = {eventType: 'notification-count', eventData: {count: number}} | {eventType: 'stopwatches', eventData: Array} | @@ -97,7 +97,8 @@ export const serverUserEventTypes = ['notification-count', 'stopwatches', 'logou export type UserEventMessage = ServerUserEventMessage | {eventType: 'worker-unavailable'} | - {eventType: 'worker-connected'}; + {eventType: 'worker-connected'} | + {eventType: 'worker-attached'}; // joined an open socket after its connect snapshot export type UserEventType = UserEventMessage['eventType']; diff --git a/web_src/js/user-events.sharedworker.ts b/web_src/js/user-events.sharedworker.ts index 2a4ca963c32..bf2540a0eaf 100644 --- a/web_src/js/user-events.sharedworker.ts +++ b/web_src/js/user-events.sharedworker.ts @@ -96,9 +96,6 @@ class WsSource { this.ws.addEventListener('open', () => { this.reconnectDelay = 1000; this.failuresWithoutConnect = 0; - // Pushes fired while no client was subscribed (initial connect gap, or a - // reconnect window) are dropped server-side, so tell clients to reconcile - // their state from the server on every fresh connection. this.source.notifyClientsUserEvent({eventType: 'worker-connected'}); }); @@ -191,6 +188,7 @@ const wsSourcesByUrl = new Map(); const openWs = wsSourcesByUrl.get(url); if (openWs?.ws?.readyState === WebSocket.OPEN) { postUserEventMessage(port, {eventType: 'worker-connected'}); + postUserEventMessage(port, {eventType: 'worker-attached'}); } return; }