From cb3321477339a394f5c9eaecbc17d0446593e213 Mon Sep 17 00:00:00 2001 From: Epid Date: Mon, 23 Mar 2026 23:31:17 +0300 Subject: [PATCH 01/10] =?UTF-8?q?feat(websocket):=20Phase=201=20=E2=80=94?= =?UTF-8?q?=20replace=20SSE=20notification=20count=20with=20WebSocket?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a thin in-memory pubsub broker and a SharedWorker-based WebSocket client to deliver real-time notification count updates. This replaces the SSE path for notification-count events with a persistent WebSocket connection shared across all tabs. New files: - services/pubsub/broker.go: fan-out pubsub broker (DefaultBroker singleton) - services/websocket/notifier.go: polls notification counts, publishes to broker - routers/web/websocket/websocket.go: /-/ws endpoint, per-user topic subscription - web_src/js/features/websocket.sharedworker.ts: SharedWorker with exponential backoff reconnect (50ms initial, 10s max, reconnect on close and error) Modified files: - routers/init.go: register websocket_service.Init() - routers/web/web.go: add GET /-/ws route - services/context/response.go: add Hijack() to forward http.Hijacker so coder/websocket can upgrade the connection - web_src/js/features/notification.ts: port from SSE SharedWorker to WS SharedWorker - webpack.config.ts: add websocket.sharedworker entry point Part of RFC #36942. --- routers/init.go | 2 + routers/web/web.go | 2 + routers/web/websocket/websocket.go | 53 +++++++ services/context/response.go | 11 ++ services/pubsub/broker.go | 65 ++++++++ services/websocket/notifier.go | 76 +++++++++ web_src/js/features/notification.ts | 45 ++---- web_src/js/features/websocket.sharedworker.ts | 144 ++++++++++++++++++ webpack.config.ts | 3 + 9 files changed, 369 insertions(+), 32 deletions(-) create mode 100644 routers/web/websocket/websocket.go create mode 100644 services/pubsub/broker.go create mode 100644 services/websocket/notifier.go create mode 100644 web_src/js/features/websocket.sharedworker.ts diff --git a/routers/init.go b/routers/init.go index 2ed7a57e5c..f6775dd8fe 100644 --- a/routers/init.go +++ b/routers/init.go @@ -54,6 +54,7 @@ import ( "code.gitea.io/gitea/services/task" "code.gitea.io/gitea/services/uinotification" "code.gitea.io/gitea/services/webhook" + websocket_service "code.gitea.io/gitea/services/websocket" ) func mustInit(fn func() error) { @@ -160,6 +161,7 @@ func InitWebInstalled(ctx context.Context) { mustInit(task.Init) mustInit(repo_migrations.Init) eventsource.GetManager().Init() + mustInit(websocket_service.Init) mustInitCtx(ctx, mailer_incoming.Init) mustInitCtx(ctx, syncAppConfForGit) diff --git a/routers/web/web.go b/routers/web/web.go index a76a68ed80..8aa26c1f36 100644 --- a/routers/web/web.go +++ b/routers/web/web.go @@ -41,6 +41,7 @@ import ( "code.gitea.io/gitea/routers/web/user" user_setting "code.gitea.io/gitea/routers/web/user/setting" "code.gitea.io/gitea/routers/web/user/setting/security" + gitea_websocket "code.gitea.io/gitea/routers/web/websocket" auth_service "code.gitea.io/gitea/services/auth" "code.gitea.io/gitea/services/context" "code.gitea.io/gitea/services/forms" @@ -588,6 +589,7 @@ func registerWebRoutes(m *web.Router, webAuth *AuthMiddleware) { }, reqSignOut) m.Any("/user/events", routing.MarkLongPolling, events.Events) + m.Get("/-/ws", gitea_websocket.Serve) m.Group("/login/oauth", func() { m.Group("", func() { diff --git a/routers/web/websocket/websocket.go b/routers/web/websocket/websocket.go new file mode 100644 index 0000000000..e0fc955cfc --- /dev/null +++ b/routers/web/websocket/websocket.go @@ -0,0 +1,53 @@ +// Copyright 2024 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +import ( + "encoding/json" + "fmt" + + "code.gitea.io/gitea/modules/log" + "code.gitea.io/gitea/services/context" + "code.gitea.io/gitea/services/pubsub" + + gitea_ws "github.com/coder/websocket" + "github.com/coder/websocket/wsjson" +) + +// Serve handles WebSocket upgrade and event delivery for the signed-in user. +func Serve(ctx *context.Context) { + if !ctx.IsSigned { + ctx.Status(401) + return + } + + conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, &gitea_ws.AcceptOptions{ + InsecureSkipVerify: false, + }) + if err != nil { + log.Error("websocket: accept failed: %v", err) + return + } + defer conn.CloseNow() //nolint:errcheck + + topic := fmt.Sprintf("user-%d", ctx.Doer.ID) + ch, cancel := pubsub.DefaultBroker.Subscribe(topic) + defer cancel() + + wsCtx := ctx.Req.Context() + for { + select { + case <-wsCtx.Done(): + return + case msg, ok := <-ch: + if !ok { + return + } + if err := wsjson.Write(wsCtx, conn, json.RawMessage(msg)); err != nil { + log.Trace("websocket: write failed: %v", err) + return + } + } + } +} diff --git a/services/context/response.go b/services/context/response.go index c7368ebc6f..ac86820d70 100644 --- a/services/context/response.go +++ b/services/context/response.go @@ -4,6 +4,8 @@ package context import ( + "bufio" + "net" "net/http" web_types "code.gitea.io/gitea/modules/web/types" @@ -67,6 +69,15 @@ func (r *Response) WriteHeader(statusCode int) { } } +// Hijack implements http.Hijacker by forwarding to the underlying ResponseWriter. +// This is required for WebSocket upgrades. +func (r *Response) Hijack() (net.Conn, *bufio.ReadWriter, error) { + if h, ok := r.ResponseWriter.(http.Hijacker); ok { + return h.Hijack() + } + return nil, nil, http.ErrNotSupported +} + // Flush flushes cached data func (r *Response) Flush() { if f, ok := r.ResponseWriter.(http.Flusher); ok { diff --git a/services/pubsub/broker.go b/services/pubsub/broker.go new file mode 100644 index 0000000000..c2f2dc1026 --- /dev/null +++ b/services/pubsub/broker.go @@ -0,0 +1,65 @@ +// Copyright 2024 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package pubsub + +import ( + "sync" +) + +// Broker is a simple in-memory pub/sub broker. +// It supports fan-out: one Publish call delivers the message to all active subscribers. +type Broker struct { + mu sync.RWMutex + subs map[string][]chan []byte +} + +// DefaultBroker is the global singleton used by both routers and notifiers. +var DefaultBroker = NewBroker() + +// NewBroker creates a new in-memory Broker. +func NewBroker() *Broker { + return &Broker{ + subs: make(map[string][]chan []byte), + } +} + +// Subscribe returns a channel that receives messages published to topic. +// Call the returned cancel function to unsubscribe. +func (b *Broker) Subscribe(topic string) (<-chan []byte, func()) { + ch := make(chan []byte, 8) + + b.mu.Lock() + b.subs[topic] = append(b.subs[topic], ch) + b.mu.Unlock() + + cancel := func() { + b.mu.Lock() + defer b.mu.Unlock() + subs := b.subs[topic] + for i, sub := range subs { + if sub == ch { + b.subs[topic] = append(subs[:i], subs[i+1:]...) + break + } + } + close(ch) + } + return ch, cancel +} + +// Publish sends msg to all subscribers of topic. +// Non-blocking: slow subscribers are skipped. +func (b *Broker) Publish(topic string, msg []byte) { + b.mu.RLock() + subs := b.subs[topic] + b.mu.RUnlock() + + for _, ch := range subs { + select { + case ch <- msg: + default: + // subscriber too slow — skip + } + } +} diff --git a/services/websocket/notifier.go b/services/websocket/notifier.go new file mode 100644 index 0000000000..85c98bfb7b --- /dev/null +++ b/services/websocket/notifier.go @@ -0,0 +1,76 @@ +// Copyright 2024 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +import ( + "context" + "fmt" + "time" + + activities_model "code.gitea.io/gitea/models/activities" + "code.gitea.io/gitea/modules/graceful" + "code.gitea.io/gitea/modules/json" + "code.gitea.io/gitea/modules/log" + "code.gitea.io/gitea/modules/process" + "code.gitea.io/gitea/modules/setting" + "code.gitea.io/gitea/modules/timeutil" + "code.gitea.io/gitea/services/pubsub" +) + +type notificationCountEvent struct { + Type string `json:"type"` + Count int64 `json:"count"` +} + +func userTopic(userID int64) string { + return fmt.Sprintf("user-%d", userID) +} + +// Init starts the background goroutine that polls notification counts +// and pushes updates to connected WebSocket clients. +func Init() error { + go graceful.GetManager().RunWithShutdownContext(run) + return nil +} + +func run(ctx context.Context) { + ctx, _, finished := process.GetManager().AddTypedContext(ctx, "Service: WebSocket", process.SystemProcessType, true) + defer finished() + + if setting.UI.Notification.EventSourceUpdateTime <= 0 { + return + } + + then := timeutil.TimeStampNow().Add(-2) + timer := time.NewTicker(setting.UI.Notification.EventSourceUpdateTime) + defer timer.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + now := timeutil.TimeStampNow().Add(-2) + + uidCounts, err := activities_model.GetUIDsAndNotificationCounts(ctx, then, now) + if err != nil { + log.Error("websocket: GetUIDsAndNotificationCounts: %v", err) + continue + } + + for _, uidCount := range uidCounts { + msg, err := json.Marshal(notificationCountEvent{ + Type: "notification-count", + Count: uidCount.Count, + }) + if err != nil { + continue + } + pubsub.DefaultBroker.Publish(userTopic(uidCount.UserID), msg) + } + + then = now + } + } +} diff --git a/web_src/js/features/notification.ts b/web_src/js/features/notification.ts index 915f65f88d..e31fd5231e 100644 --- a/web_src/js/features/notification.ts +++ b/web_src/js/features/notification.ts @@ -5,12 +5,12 @@ import {logoutFromWorker} from '../modules/worker.ts'; const {appSubUrl, notificationSettings, assetVersionEncoded} = window.config; let notificationSequenceNumber = 0; -async function receiveUpdateCount(event: MessageEvent<{type: string, data: string}>) { +async function receiveUpdateCount(event: MessageEvent<{type: string, count: number}>) { try { - const data = JSON.parse(event.data.data); - for (const count of document.querySelectorAll('.notification_count')) { - count.classList.toggle('tw-hidden', data.Count === 0); - count.textContent = `${data.Count}`; + const {count} = event.data; + for (const el of document.querySelectorAll('.notification_count')) { + el.classList.toggle('tw-hidden', count === 0); + el.textContent = `${count}`; } await updateNotificationTable(); } catch (error) { @@ -21,55 +21,38 @@ async function receiveUpdateCount(event: MessageEvent<{type: string, data: strin export function initNotificationCount() { if (!document.querySelector('.notification_count')) return; - let usingPeriodicPoller = false; const startPeriodicPoller = (timeout: number, lastCount?: number) => { if (timeout <= 0 || !Number.isFinite(timeout)) return; - usingPeriodicPoller = true; lastCount = lastCount ?? getCurrentCount(); setTimeout(async () => { await updateNotificationCountWithCallback(startPeriodicPoller, timeout, lastCount); }, timeout); }; - if (notificationSettings.EventSourceUpdateTime > 0 && window.EventSource && window.SharedWorker) { - // Try to connect to the event source via the shared worker first - const worker = new SharedWorker(`${window.__webpack_public_path__}js/eventsource.sharedworker.js?v=${assetVersionEncoded}`, 'notification-worker'); + if (notificationSettings.EventSourceUpdateTime > 0 && window.SharedWorker) { + // Connect via WebSocket SharedWorker (one connection shared across all tabs) + const wsUrl = `${window.location.origin}${appSubUrl}/-/ws`.replace(/^http/, 'ws'); + const worker = new SharedWorker(`${window.__webpack_public_path__}js/websocket.sharedworker.js?v=${assetVersionEncoded}`, 'notification-worker'); worker.addEventListener('error', (event) => { console.error('worker error', event); }); worker.port.addEventListener('messageerror', () => { console.error('unable to deserialize message'); }); - worker.port.postMessage({ - type: 'start', - url: `${window.location.origin}${appSubUrl}/user/events`, - }); - worker.port.addEventListener('message', (event: MessageEvent<{type: string, data: string}>) => { + worker.port.postMessage({type: 'start', url: wsUrl}); + worker.port.addEventListener('message', (event: MessageEvent<{type: string, count: number, message?: string}>) => { if (!event.data || !event.data.type) { console.error('unknown worker message event', event); return; } if (event.data.type === 'notification-count') { receiveUpdateCount(event); // no await - } else if (event.data.type === 'no-event-source') { - // browser doesn't support EventSource, falling back to periodic poller - if (!usingPeriodicPoller) startPeriodicPoller(notificationSettings.MinTimeout); } else if (event.data.type === 'error') { console.error('worker port event error', event.data); } else if (event.data.type === 'logout') { - if (event.data.data !== 'here') { - return; - } - worker.port.postMessage({ - type: 'close', - }); + worker.port.postMessage({type: 'close'}); worker.port.close(); logoutFromWorker(); - } else if (event.data.type === 'close') { - worker.port.postMessage({ - type: 'close', - }); - worker.port.close(); } }); worker.port.addEventListener('error', (e) => { @@ -77,9 +60,7 @@ export function initNotificationCount() { }); worker.port.start(); window.addEventListener('beforeunload', () => { - worker.port.postMessage({ - type: 'close', - }); + worker.port.postMessage({type: 'close'}); worker.port.close(); }); diff --git a/web_src/js/features/websocket.sharedworker.ts b/web_src/js/features/websocket.sharedworker.ts new file mode 100644 index 0000000000..88d4870f01 --- /dev/null +++ b/web_src/js/features/websocket.sharedworker.ts @@ -0,0 +1,144 @@ +// One WebSocket connection per URL, shared across all tabs via SharedWorker. +// Messages from the server are JSON objects broadcast to all connected ports. +export {}; // make this a module to avoid global scope conflicts with other sharedworker files + +const RECONNECT_DELAY_INITIAL = 50; +const RECONNECT_DELAY_MAX = 10000; + +class WsSource { + url: string; + ws: WebSocket | null; + clients: MessagePort[]; + reconnectTimer: ReturnType | null; + reconnectDelay: number; + + constructor(url: string) { + this.url = url; + this.ws = null; + this.clients = []; + this.reconnectTimer = null; + this.reconnectDelay = RECONNECT_DELAY_INITIAL; + this.connect(); + } + + connect() { + this.ws = new WebSocket(this.url); + + this.ws.addEventListener('open', () => { + this.reconnectDelay = RECONNECT_DELAY_INITIAL; + this.broadcast({type: 'status', message: `connected to ${this.url}`}); + }); + + this.ws.addEventListener('message', (event: MessageEvent) => { + try { + const msg = JSON.parse(event.data); + this.broadcast(msg); + } catch { + // ignore malformed JSON + } + }); + + this.ws.addEventListener('close', () => { + this.ws = null; + this.scheduleReconnect(); + }); + + this.ws.addEventListener('error', () => { + this.broadcast({type: 'error', message: 'websocket error'}); + this.ws = null; + this.scheduleReconnect(); + }); + } + + scheduleReconnect() { + if (this.clients.length === 0 || this.reconnectTimer !== null) return; + this.reconnectDelay = Math.min(this.reconnectDelay * 2, RECONNECT_DELAY_MAX); + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + this.connect(); + }, this.reconnectDelay); + } + + register(port: MessagePort) { + if (this.clients.includes(port)) return; + this.clients.push(port); + port.postMessage({type: 'status', message: `registered to ${this.url}`}); + } + + deregister(port: MessagePort): number { + const idx = this.clients.indexOf(port); + if (idx >= 0) this.clients.splice(idx, 1); + return this.clients.length; + } + + close() { + if (this.reconnectTimer !== null) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + this.ws?.close(); + this.ws = null; + } + + broadcast(msg: unknown) { + for (const port of this.clients) { + port.postMessage(msg); + } + } +} + +const sourcesByUrl = new Map(); +const sourcesByPort = new Map(); + +(self as unknown as SharedWorkerGlobalScope).addEventListener('connect', (e: MessageEvent) => { + for (const port of e.ports) { + port.addEventListener('message', (event: MessageEvent) => { + if (event.data.type === 'start') { + const {url} = event.data; + let source = sourcesByUrl.get(url); + if (source) { + source.register(port); + sourcesByPort.set(port, source); + return; + } + source = sourcesByPort.get(port); + if (source) { + const count = source.deregister(port); + if (count === 0) { + source.close(); + sourcesByUrl.set(source.url, null); + } + } + source = new WsSource(url); + source.register(port); + sourcesByUrl.set(url, source); + sourcesByPort.set(port, source); + } else if (event.data.type === 'close') { + const source = sourcesByPort.get(port); + if (!source) return; + const count = source.deregister(port); + if (count === 0) { + source.close(); + sourcesByUrl.set(source.url, null); + sourcesByPort.set(port, null); + } + } else if (event.data.type === 'status') { + const source = sourcesByPort.get(port); + if (!source) { + port.postMessage({type: 'status', message: 'not connected'}); + return; + } + port.postMessage({ + type: 'status', + message: `url: ${source.url} readyState: ${source.ws?.readyState ?? 'null'}`, + }); + } else { + port.postMessage({ + type: 'error', + message: `received but don't know how to handle: ${JSON.stringify(event.data)}`, + }); + } + }); + port.start(); + } +}); diff --git a/webpack.config.ts b/webpack.config.ts index e3ef996909..cd601d6653 100644 --- a/webpack.config.ts +++ b/webpack.config.ts @@ -79,6 +79,9 @@ export default { 'eventsource.sharedworker': [ fileURLToPath(new URL('web_src/js/features/eventsource.sharedworker.ts', import.meta.url)), ], + 'websocket.sharedworker': [ + fileURLToPath(new URL('web_src/js/features/websocket.sharedworker.ts', import.meta.url)), + ], ...(!isProduction && { devtest: [ fileURLToPath(new URL('web_src/js/standalone/devtest.ts', import.meta.url)), From 607343812a88281f4d26e0f61ef1ea962c911737 Mon Sep 17 00:00:00 2001 From: Epid Date: Mon, 23 Mar 2026 23:46:17 +0300 Subject: [PATCH 02/10] chore: add github.com/coder/websocket dependency to go.mod --- go.mod | 1 + go.sum | 2 ++ 2 files changed, 3 insertions(+) diff --git a/go.mod b/go.mod index 24c18d5703..9070eeee91 100644 --- a/go.mod +++ b/go.mod @@ -36,6 +36,7 @@ require ( github.com/caddyserver/certmagic v0.25.1 github.com/charmbracelet/git-lfs-transfer v0.1.1-0.20251013092601-6327009efd21 github.com/chi-middleware/proxy v1.1.1 + github.com/coder/websocket v1.8.14 github.com/dimiro1/reply v0.0.0-20200315094148-d0136a4c9e21 github.com/dlclark/regexp2 v1.11.5 github.com/dsnet/compress v0.0.2-0.20230904184137-39efe44ab707 diff --git a/go.sum b/go.sum index 02e6532542..ccd2012c34 100644 --- a/go.sum +++ b/go.sum @@ -229,6 +229,8 @@ github.com/chzyer/test v1.0.0/go.mod h1:2JlltgoNkt4TW/z9V/IzDdFaMTM2JPIi26O1pF38 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8= github.com/cloudflare/circl v1.6.3/go.mod h1:2eXP6Qfat4O/Yhh8BznvKnJ+uzEoTQ6jVKJRn81BiS4= +github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= +github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE= github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk= github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= From 3076f902eab533735bc948d235e976a728cfa99d Mon Sep 17 00:00:00 2001 From: Epid Date: Tue, 24 Mar 2026 00:07:56 +0300 Subject: [PATCH 03/10] fix(websocket): use gitea modules/json, write raw bytes directly --- routers/web/websocket/websocket.go | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/routers/web/websocket/websocket.go b/routers/web/websocket/websocket.go index e0fc955cfc..6feb81008d 100644 --- a/routers/web/websocket/websocket.go +++ b/routers/web/websocket/websocket.go @@ -4,7 +4,6 @@ package websocket import ( - "encoding/json" "fmt" "code.gitea.io/gitea/modules/log" @@ -12,7 +11,6 @@ import ( "code.gitea.io/gitea/services/pubsub" gitea_ws "github.com/coder/websocket" - "github.com/coder/websocket/wsjson" ) // Serve handles WebSocket upgrade and event delivery for the signed-in user. @@ -29,7 +27,7 @@ func Serve(ctx *context.Context) { log.Error("websocket: accept failed: %v", err) return } - defer conn.CloseNow() //nolint:errcheck + defer conn.CloseNow() //nolint:errcheck // CloseNow is best-effort; error is intentionally ignored topic := fmt.Sprintf("user-%d", ctx.Doer.ID) ch, cancel := pubsub.DefaultBroker.Subscribe(topic) @@ -44,7 +42,7 @@ func Serve(ctx *context.Context) { if !ok { return } - if err := wsjson.Write(wsCtx, conn, json.RawMessage(msg)); err != nil { + if err := conn.Write(wsCtx, gitea_ws.MessageText, msg); err != nil { log.Trace("websocket: write failed: %v", err) return } From 096bdd0902640fd2550fa2bb4b20cacb9b6d16ca Mon Sep 17 00:00:00 2001 From: Epid Date: Tue, 24 Mar 2026 01:22:03 +0300 Subject: [PATCH 04/10] fix(websocket): avoid data race with timeutil.MockUnset in tests Replace timeutil.TimeStampNow() calls in the websocket notifier with a nowTS() helper that reads time.Now().Unix() directly. TimeStampNow reads a package-level mock variable that TestIncomingEmail writes concurrently, causing a race detected by the race detector in test-pgsql CI. --- services/websocket/notifier.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/services/websocket/notifier.go b/services/websocket/notifier.go index 85c98bfb7b..af00d75d03 100644 --- a/services/websocket/notifier.go +++ b/services/websocket/notifier.go @@ -18,6 +18,12 @@ import ( "code.gitea.io/gitea/services/pubsub" ) +// nowTS returns the current time as a TimeStamp using the real wall clock, +// avoiding data races with timeutil.MockUnset during tests. +func nowTS() timeutil.TimeStamp { + return timeutil.TimeStamp(time.Now().Unix()) +} + type notificationCountEvent struct { Type string `json:"type"` Count int64 `json:"count"` @@ -42,7 +48,7 @@ func run(ctx context.Context) { return } - then := timeutil.TimeStampNow().Add(-2) + then := nowTS().Add(-2) timer := time.NewTicker(setting.UI.Notification.EventSourceUpdateTime) defer timer.Stop() @@ -51,7 +57,7 @@ func run(ctx context.Context) { case <-ctx.Done(): return case <-timer.C: - now := timeutil.TimeStampNow().Add(-2) + now := nowTS().Add(-2) uidCounts, err := activities_model.GetUIDsAndNotificationCounts(ctx, then, now) if err != nil { From 6aba649e956abcbc3924be465ea3da8496bf2863 Mon Sep 17 00:00:00 2001 From: Epid Date: Tue, 24 Mar 2026 03:57:43 +0300 Subject: [PATCH 05/10] chore: run make tidy to add coder/websocket to go-licenses.json assets/go-licenses.json was missing the license entry for the newly added github.com/coder/websocket dependency. Running make tidy regenerates this file via build/generate-go-licenses.go. --- assets/go-licenses.json | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/assets/go-licenses.json b/assets/go-licenses.json index 30f56e5f87..cf98d95118 100644 --- a/assets/go-licenses.json +++ b/assets/go-licenses.json @@ -374,6 +374,11 @@ "path": "github.com/cloudflare/circl/LICENSE", "licenseText": "Copyright (c) 2019 Cloudflare. All rights reserved.\n\nRedistribution and use in source and binary forms, with or without\nmodification, are permitted provided that the following conditions are\nmet:\n\n * Redistributions of source code must retain the above copyright\nnotice, this list of conditions and the following disclaimer.\n * Redistributions in binary form must reproduce the above\ncopyright notice, this list of conditions and the following disclaimer\nin the documentation and/or other materials provided with the\ndistribution.\n * Neither the name of Cloudflare nor the names of its\ncontributors may be used to endorse or promote products derived from\nthis software without specific prior written permission.\n\nTHIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n\"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\nLIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\nA PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\nOWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\nSPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\nLIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\nDATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\nTHEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\nOF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\n========================================================================\n\nCopyright (c) 2009 The Go Authors. All rights reserved.\n\nRedistribution and use in source and binary forms, with or without\nmodification, are permitted provided that the following conditions are\nmet:\n\n * Redistributions of source code must retain the above copyright\nnotice, this list of conditions and the following disclaimer.\n * Redistributions in binary form must reproduce the above\ncopyright notice, this list of conditions and the following disclaimer\nin the documentation and/or other materials provided with the\ndistribution.\n * Neither the name of Google Inc. nor the names of its\ncontributors may be used to endorse or promote products derived from\nthis software without specific prior written permission.\n\nTHIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n\"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\nLIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\nA PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\nOWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\nSPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\nLIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\nDATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\nTHEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\nOF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" }, + { + "name": "github.com/coder/websocket", + "path": "github.com/coder/websocket/LICENSE.txt", + "licenseText": "Copyright (c) 2025 Coder\n\nPermission to use, copy, modify, and distribute this software for any\npurpose with or without fee is hereby granted, provided that the above\ncopyright notice and this permission notice appear in all copies.\n\nTHE SOFTWARE IS PROVIDED \"AS IS\" AND THE AUTHOR DISCLAIMS ALL WARRANTIES\nWITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF\nMERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR\nANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES\nWHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN\nACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF\nOR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.\n" + }, { "name": "github.com/couchbase/go-couchbase", "path": "github.com/couchbase/go-couchbase/LICENSE", From 89e508419af5381253e3bb58e5a9c37b75d91fc5 Mon Sep 17 00:00:00 2001 From: Epid Date: Tue, 24 Mar 2026 04:44:00 +0300 Subject: [PATCH 06/10] fix(websocket): address silverwind review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Move /-/ws route inside reqSignIn middleware group; remove manual ctx.IsSigned check from handler (auth is now enforced by the router) - Fix scheduleReconnect() to schedule using current delay then double, so first reconnect fires after 50ms not 100ms (reported by silverwind) - Replace sourcesByPort.set(port, null) with delete() to prevent MessagePort retention after tab close (memory leak fix) - Centralize topic naming in pubsub.UserTopic() — removes duplication between the notifier and the WebSocket handler - Skip DB polling in notifier when broker has no active subscribers to avoid unnecessary load on idle instances - Hold RLock for the full Publish fan-out loop to prevent a race where cancel() closes a channel between slice read and send --- routers/web/web.go | 4 ++- routers/web/websocket/websocket.go | 11 ++------ services/pubsub/broker.go | 26 ++++++++++++++++--- services/websocket/notifier.go | 12 ++++----- web_src/js/features/websocket.sharedworker.ts | 15 ++++++----- 5 files changed, 42 insertions(+), 26 deletions(-) diff --git a/routers/web/web.go b/routers/web/web.go index 8aa26c1f36..5dad91fc0a 100644 --- a/routers/web/web.go +++ b/routers/web/web.go @@ -589,7 +589,9 @@ func registerWebRoutes(m *web.Router, webAuth *AuthMiddleware) { }, reqSignOut) m.Any("/user/events", routing.MarkLongPolling, events.Events) - m.Get("/-/ws", gitea_websocket.Serve) + m.Group("", func() { + m.Get("/-/ws", gitea_websocket.Serve) + }, reqSignIn) m.Group("/login/oauth", func() { m.Group("", func() { diff --git a/routers/web/websocket/websocket.go b/routers/web/websocket/websocket.go index 6feb81008d..cfa146e347 100644 --- a/routers/web/websocket/websocket.go +++ b/routers/web/websocket/websocket.go @@ -4,8 +4,6 @@ package websocket import ( - "fmt" - "code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/services/context" "code.gitea.io/gitea/services/pubsub" @@ -14,12 +12,8 @@ import ( ) // Serve handles WebSocket upgrade and event delivery for the signed-in user. +// Authentication is enforced by the reqSignIn middleware in the router. func Serve(ctx *context.Context) { - if !ctx.IsSigned { - ctx.Status(401) - return - } - conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, &gitea_ws.AcceptOptions{ InsecureSkipVerify: false, }) @@ -29,8 +23,7 @@ func Serve(ctx *context.Context) { } defer conn.CloseNow() //nolint:errcheck // CloseNow is best-effort; error is intentionally ignored - topic := fmt.Sprintf("user-%d", ctx.Doer.ID) - ch, cancel := pubsub.DefaultBroker.Subscribe(topic) + ch, cancel := pubsub.DefaultBroker.Subscribe(pubsub.UserTopic(ctx.Doer.ID)) defer cancel() wsCtx := ctx.Req.Context() diff --git a/services/pubsub/broker.go b/services/pubsub/broker.go index c2f2dc1026..9143742489 100644 --- a/services/pubsub/broker.go +++ b/services/pubsub/broker.go @@ -4,6 +4,7 @@ package pubsub import ( + "fmt" "sync" ) @@ -48,14 +49,33 @@ func (b *Broker) Subscribe(topic string) (<-chan []byte, func()) { return ch, cancel } +// UserTopic returns the pub/sub topic name for a given user ID. +// Centralised here so the notifier and the WebSocket handler always agree on the format. +func UserTopic(userID int64) string { + return fmt.Sprintf("user-%d", userID) +} + +// HasSubscribers reports whether the broker has at least one active subscriber across all topics. +func (b *Broker) HasSubscribers() bool { + b.mu.RLock() + defer b.mu.RUnlock() + for _, subs := range b.subs { + if len(subs) > 0 { + return true + } + } + return false +} + // Publish sends msg to all subscribers of topic. // Non-blocking: slow subscribers are skipped. +// The RLock is held for the entire fan-out to prevent a race where cancel() +// closes a channel between the slice read and the send. func (b *Broker) Publish(topic string, msg []byte) { b.mu.RLock() - subs := b.subs[topic] - b.mu.RUnlock() + defer b.mu.RUnlock() - for _, ch := range subs { + for _, ch := range b.subs[topic] { select { case ch <- msg: default: diff --git a/services/websocket/notifier.go b/services/websocket/notifier.go index af00d75d03..d8f64ef7d5 100644 --- a/services/websocket/notifier.go +++ b/services/websocket/notifier.go @@ -5,7 +5,6 @@ package websocket import ( "context" - "fmt" "time" activities_model "code.gitea.io/gitea/models/activities" @@ -29,10 +28,6 @@ type notificationCountEvent struct { Count int64 `json:"count"` } -func userTopic(userID int64) string { - return fmt.Sprintf("user-%d", userID) -} - // Init starts the background goroutine that polls notification counts // and pushes updates to connected WebSocket clients. func Init() error { @@ -57,6 +52,11 @@ func run(ctx context.Context) { case <-ctx.Done(): return case <-timer.C: + if !pubsub.DefaultBroker.HasSubscribers() { + then = nowTS().Add(-2) + continue + } + now := nowTS().Add(-2) uidCounts, err := activities_model.GetUIDsAndNotificationCounts(ctx, then, now) @@ -73,7 +73,7 @@ func run(ctx context.Context) { if err != nil { continue } - pubsub.DefaultBroker.Publish(userTopic(uidCount.UserID), msg) + pubsub.DefaultBroker.Publish(pubsub.UserTopic(uidCount.UserID), msg) } then = now diff --git a/web_src/js/features/websocket.sharedworker.ts b/web_src/js/features/websocket.sharedworker.ts index 88d4870f01..491c7f2a07 100644 --- a/web_src/js/features/websocket.sharedworker.ts +++ b/web_src/js/features/websocket.sharedworker.ts @@ -52,11 +52,12 @@ class WsSource { scheduleReconnect() { if (this.clients.length === 0 || this.reconnectTimer !== null) return; - this.reconnectDelay = Math.min(this.reconnectDelay * 2, RECONNECT_DELAY_MAX); + const delay = this.reconnectDelay; this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; this.connect(); - }, this.reconnectDelay); + }, delay); + this.reconnectDelay = Math.min(this.reconnectDelay * 2, RECONNECT_DELAY_MAX); } register(port: MessagePort) { @@ -87,8 +88,8 @@ class WsSource { } } -const sourcesByUrl = new Map(); -const sourcesByPort = new Map(); +const sourcesByUrl = new Map(); +const sourcesByPort = new Map(); (self as unknown as SharedWorkerGlobalScope).addEventListener('connect', (e: MessageEvent) => { for (const port of e.ports) { @@ -106,7 +107,7 @@ const sourcesByPort = new Map(); const count = source.deregister(port); if (count === 0) { source.close(); - sourcesByUrl.set(source.url, null); + sourcesByUrl.delete(source.url); } } source = new WsSource(url); @@ -119,8 +120,8 @@ const sourcesByPort = new Map(); const count = source.deregister(port); if (count === 0) { source.close(); - sourcesByUrl.set(source.url, null); - sourcesByPort.set(port, null); + sourcesByUrl.delete(source.url); + sourcesByPort.delete(port); } } else if (event.data.type === 'status') { const source = sourcesByPort.get(port); From 1537d8f74a8df99ac201d651df43704c6c0a0c9b Mon Sep 17 00:00:00 2001 From: Epid Date: Tue, 24 Mar 2026 10:58:02 +0300 Subject: [PATCH 07/10] fix(websocket): auth via IsSigned check instead of reqSignIn middleware reqSignIn sends a 303 redirect which breaks WebSocket upgrade; use the same pattern as /user/events: register the route without middleware and return 401 inside the handler when the user is not signed in. Also fix copyright year to 2026 in all three new Go files and add a console.warn for malformed JSON in the SharedWorker. --- routers/web/web.go | 4 +--- routers/web/websocket/websocket.go | 9 +++++++-- services/pubsub/broker.go | 2 +- services/websocket/notifier.go | 2 +- web_src/js/features/websocket.sharedworker.ts | 2 +- 5 files changed, 11 insertions(+), 8 deletions(-) diff --git a/routers/web/web.go b/routers/web/web.go index 5dad91fc0a..8aa26c1f36 100644 --- a/routers/web/web.go +++ b/routers/web/web.go @@ -589,9 +589,7 @@ func registerWebRoutes(m *web.Router, webAuth *AuthMiddleware) { }, reqSignOut) m.Any("/user/events", routing.MarkLongPolling, events.Events) - m.Group("", func() { - m.Get("/-/ws", gitea_websocket.Serve) - }, reqSignIn) + m.Get("/-/ws", gitea_websocket.Serve) m.Group("/login/oauth", func() { m.Group("", func() { diff --git a/routers/web/websocket/websocket.go b/routers/web/websocket/websocket.go index cfa146e347..b4d9619f6d 100644 --- a/routers/web/websocket/websocket.go +++ b/routers/web/websocket/websocket.go @@ -1,9 +1,11 @@ -// Copyright 2024 The Gitea Authors. All rights reserved. +// Copyright 2026 The Gitea Authors. All rights reserved. // SPDX-License-Identifier: MIT package websocket import ( + "net/http" + "code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/services/context" "code.gitea.io/gitea/services/pubsub" @@ -12,8 +14,11 @@ import ( ) // Serve handles WebSocket upgrade and event delivery for the signed-in user. -// Authentication is enforced by the reqSignIn middleware in the router. func Serve(ctx *context.Context) { + if !ctx.IsSigned { + ctx.Status(http.StatusUnauthorized) + return + } conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, &gitea_ws.AcceptOptions{ InsecureSkipVerify: false, }) diff --git a/services/pubsub/broker.go b/services/pubsub/broker.go index 9143742489..1a8bef5321 100644 --- a/services/pubsub/broker.go +++ b/services/pubsub/broker.go @@ -1,4 +1,4 @@ -// Copyright 2024 The Gitea Authors. All rights reserved. +// Copyright 2026 The Gitea Authors. All rights reserved. // SPDX-License-Identifier: MIT package pubsub diff --git a/services/websocket/notifier.go b/services/websocket/notifier.go index d8f64ef7d5..2d93ae49ea 100644 --- a/services/websocket/notifier.go +++ b/services/websocket/notifier.go @@ -1,4 +1,4 @@ -// Copyright 2024 The Gitea Authors. All rights reserved. +// Copyright 2026 The Gitea Authors. All rights reserved. // SPDX-License-Identifier: MIT package websocket diff --git a/web_src/js/features/websocket.sharedworker.ts b/web_src/js/features/websocket.sharedworker.ts index 491c7f2a07..f8cc635570 100644 --- a/web_src/js/features/websocket.sharedworker.ts +++ b/web_src/js/features/websocket.sharedworker.ts @@ -34,7 +34,7 @@ class WsSource { const msg = JSON.parse(event.data); this.broadcast(msg); } catch { - // ignore malformed JSON + console.warn('websocket.sharedworker: received non-JSON message', event.data); } }); From f2450cc6e19eb31e722d18cfc38d1e262f8014c4 Mon Sep 17 00:00:00 2001 From: Epid Date: Tue, 24 Mar 2026 13:51:16 +0300 Subject: [PATCH 08/10] fix(websocket): remove export{} from sharedworker entry point, rename worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove `export {}` which caused webpack to tree-shake the entire SharedWorker bundle, resulting in an empty JS file with no connect handler — root cause of WebSocket never opening - Rename SharedWorker instance from 'notification-worker' to 'notification-worker-ws' to force browser to create a fresh worker instance instead of reusing a cached empty one --- web_src/js/features/notification.ts | 2 +- web_src/js/features/websocket.sharedworker.ts | 2 -- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/web_src/js/features/notification.ts b/web_src/js/features/notification.ts index e31fd5231e..cb025b9fd7 100644 --- a/web_src/js/features/notification.ts +++ b/web_src/js/features/notification.ts @@ -32,7 +32,7 @@ export function initNotificationCount() { if (notificationSettings.EventSourceUpdateTime > 0 && window.SharedWorker) { // Connect via WebSocket SharedWorker (one connection shared across all tabs) const wsUrl = `${window.location.origin}${appSubUrl}/-/ws`.replace(/^http/, 'ws'); - const worker = new SharedWorker(`${window.__webpack_public_path__}js/websocket.sharedworker.js?v=${assetVersionEncoded}`, 'notification-worker'); + const worker = new SharedWorker(`${window.__webpack_public_path__}js/websocket.sharedworker.js?v=${assetVersionEncoded}`, 'notification-worker-ws'); worker.addEventListener('error', (event) => { console.error('worker error', event); }); diff --git a/web_src/js/features/websocket.sharedworker.ts b/web_src/js/features/websocket.sharedworker.ts index f8cc635570..9925ba5bd6 100644 --- a/web_src/js/features/websocket.sharedworker.ts +++ b/web_src/js/features/websocket.sharedworker.ts @@ -1,7 +1,5 @@ // One WebSocket connection per URL, shared across all tabs via SharedWorker. // Messages from the server are JSON objects broadcast to all connected ports. -export {}; // make this a module to avoid global scope conflicts with other sharedworker files - const RECONNECT_DELAY_INITIAL = 50; const RECONNECT_DELAY_MAX = 10000; From c1ba1828086a7e25056402869ecb59866c4483ef Mon Sep 17 00:00:00 2001 From: Epid Date: Tue, 24 Mar 2026 22:25:26 +0300 Subject: [PATCH 09/10] fix(websocket): declare sharedworker as ES module and fix port cleanup - Add export{} to declare websocket.sharedworker.ts as an ES module, preventing TypeScript TS2451 redeclaration errors caused by global scope conflicts with eventsource.sharedworker.ts - Always delete port from sourcesByPort on close regardless of remaining subscriber count, preventing MessagePort keys from leaking in the Map --- web_src/js/features/websocket.sharedworker.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/web_src/js/features/websocket.sharedworker.ts b/web_src/js/features/websocket.sharedworker.ts index 9925ba5bd6..539aa46037 100644 --- a/web_src/js/features/websocket.sharedworker.ts +++ b/web_src/js/features/websocket.sharedworker.ts @@ -1,5 +1,7 @@ // One WebSocket connection per URL, shared across all tabs via SharedWorker. // Messages from the server are JSON objects broadcast to all connected ports. +export {}; // make this a module to avoid global scope conflicts with other sharedworker files + const RECONNECT_DELAY_INITIAL = 50; const RECONNECT_DELAY_MAX = 10000; @@ -116,10 +118,10 @@ const sourcesByPort = new Map(); const source = sourcesByPort.get(port); if (!source) return; const count = source.deregister(port); + sourcesByPort.delete(port); if (count === 0) { source.close(); sourcesByUrl.delete(source.url); - sourcesByPort.delete(port); } } else if (event.data.type === 'status') { const source = sourcesByPort.get(port); From 36b28c7e04c683afbe4688ae597a80a8578ce5b0 Mon Sep 17 00:00:00 2001 From: Epid Date: Mon, 30 Mar 2026 06:39:04 +0300 Subject: [PATCH 10/10] fix(websocket): add WsSource to eventsource.sharedworker, remove websocket.sharedworker - Add WsSource class to eventsource.sharedworker.ts for WebSocket transport - Remove websocket.sharedworker.ts (functionality merged into eventsource.sharedworker.ts) --- .../js/features/eventsource.sharedworker.ts | 87 +++++++++++ web_src/js/features/websocket.sharedworker.ts | 145 ------------------ 2 files changed, 87 insertions(+), 145 deletions(-) delete mode 100644 web_src/js/features/websocket.sharedworker.ts diff --git a/web_src/js/features/eventsource.sharedworker.ts b/web_src/js/features/eventsource.sharedworker.ts index 816cd7020a..58b371e6a0 100644 --- a/web_src/js/features/eventsource.sharedworker.ts +++ b/web_src/js/features/eventsource.sharedworker.ts @@ -69,8 +69,82 @@ class Source { } } +// WsSource provides a WebSocket transport alongside EventSource. +// It delivers real-time notification-count pushes using the same client list +// as the associated Source, normalising messages to the SSE event format so +// that callers do not need to know which transport delivered the event. +class WsSource { + wsUrl: string; + ws: WebSocket | null; + source: Source; + reconnectTimer: ReturnType | null; + reconnectDelay: number; + + constructor(wsUrl: string, source: Source) { + this.wsUrl = wsUrl; + this.source = source; + this.ws = null; + this.reconnectTimer = null; + this.reconnectDelay = 50; + this.connect(); + } + + connect() { + this.ws = new WebSocket(this.wsUrl); + + this.ws.addEventListener('open', () => { + this.reconnectDelay = 50; + }); + + this.ws.addEventListener('message', (event: MessageEvent) => { + try { + const msg = JSON.parse(event.data); + if (msg.type === 'notification-count') { + // Normalise to SSE event format so the receiver is transport-agnostic. + this.source.notifyClients({ + type: 'notification-count', + data: JSON.stringify({Count: msg.count}), + }); + } + } catch { + // ignore malformed messages + } + }); + + this.ws.addEventListener('close', () => { + this.ws = null; + this.scheduleReconnect(); + }); + + this.ws.addEventListener('error', () => { + this.ws = null; + this.scheduleReconnect(); + }); + } + + scheduleReconnect() { + if (this.reconnectTimer !== null) return; + const delay = this.reconnectDelay; + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + this.connect(); + }, delay); + this.reconnectDelay = Math.min(this.reconnectDelay * 2, 10000); + } + + close() { + if (this.reconnectTimer !== null) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + this.ws?.close(); + this.ws = null; + } +} + const sourcesByUrl = new Map(); const sourcesByPort = new Map(); +const wsSourcesByUrl = new Map(); (self as unknown as SharedWorkerGlobalScope).addEventListener('connect', (e: MessageEvent) => { for (const port of e.ports) { @@ -102,6 +176,11 @@ const sourcesByPort = new Map(); if (count === 0) { source.close(); sourcesByUrl.set(source.url, null); + const ws = wsSourcesByUrl.get(source.url); + if (ws) { + ws.close(); + wsSourcesByUrl.set(source.url, null); + } } } // Create a new Source @@ -109,6 +188,9 @@ const sourcesByPort = new Map(); source.register(port); sourcesByUrl.set(url, source); sourcesByPort.set(port, source); + // Start WebSocket alongside EventSource for real-time notification pushes. + const wsUrl = url.replace(/^http/, 'ws').replace(/\/user\/events$/, '/-/ws'); + wsSourcesByUrl.set(url, new WsSource(wsUrl, source)); } else if (event.data.type === 'listen') { const source = sourcesByPort.get(port)!; source.listen(event.data.eventType); @@ -121,6 +203,11 @@ const sourcesByPort = new Map(); source.close(); sourcesByUrl.set(source.url, null); sourcesByPort.set(port, null); + const ws = wsSourcesByUrl.get(source.url); + if (ws) { + ws.close(); + wsSourcesByUrl.set(source.url, null); + } } } else if (event.data.type === 'status') { const source = sourcesByPort.get(port); diff --git a/web_src/js/features/websocket.sharedworker.ts b/web_src/js/features/websocket.sharedworker.ts deleted file mode 100644 index 539aa46037..0000000000 --- a/web_src/js/features/websocket.sharedworker.ts +++ /dev/null @@ -1,145 +0,0 @@ -// One WebSocket connection per URL, shared across all tabs via SharedWorker. -// Messages from the server are JSON objects broadcast to all connected ports. -export {}; // make this a module to avoid global scope conflicts with other sharedworker files - -const RECONNECT_DELAY_INITIAL = 50; -const RECONNECT_DELAY_MAX = 10000; - -class WsSource { - url: string; - ws: WebSocket | null; - clients: MessagePort[]; - reconnectTimer: ReturnType | null; - reconnectDelay: number; - - constructor(url: string) { - this.url = url; - this.ws = null; - this.clients = []; - this.reconnectTimer = null; - this.reconnectDelay = RECONNECT_DELAY_INITIAL; - this.connect(); - } - - connect() { - this.ws = new WebSocket(this.url); - - this.ws.addEventListener('open', () => { - this.reconnectDelay = RECONNECT_DELAY_INITIAL; - this.broadcast({type: 'status', message: `connected to ${this.url}`}); - }); - - this.ws.addEventListener('message', (event: MessageEvent) => { - try { - const msg = JSON.parse(event.data); - this.broadcast(msg); - } catch { - console.warn('websocket.sharedworker: received non-JSON message', event.data); - } - }); - - this.ws.addEventListener('close', () => { - this.ws = null; - this.scheduleReconnect(); - }); - - this.ws.addEventListener('error', () => { - this.broadcast({type: 'error', message: 'websocket error'}); - this.ws = null; - this.scheduleReconnect(); - }); - } - - scheduleReconnect() { - if (this.clients.length === 0 || this.reconnectTimer !== null) return; - const delay = this.reconnectDelay; - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - this.connect(); - }, delay); - this.reconnectDelay = Math.min(this.reconnectDelay * 2, RECONNECT_DELAY_MAX); - } - - register(port: MessagePort) { - if (this.clients.includes(port)) return; - this.clients.push(port); - port.postMessage({type: 'status', message: `registered to ${this.url}`}); - } - - deregister(port: MessagePort): number { - const idx = this.clients.indexOf(port); - if (idx >= 0) this.clients.splice(idx, 1); - return this.clients.length; - } - - close() { - if (this.reconnectTimer !== null) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - } - this.ws?.close(); - this.ws = null; - } - - broadcast(msg: unknown) { - for (const port of this.clients) { - port.postMessage(msg); - } - } -} - -const sourcesByUrl = new Map(); -const sourcesByPort = new Map(); - -(self as unknown as SharedWorkerGlobalScope).addEventListener('connect', (e: MessageEvent) => { - for (const port of e.ports) { - port.addEventListener('message', (event: MessageEvent) => { - if (event.data.type === 'start') { - const {url} = event.data; - let source = sourcesByUrl.get(url); - if (source) { - source.register(port); - sourcesByPort.set(port, source); - return; - } - source = sourcesByPort.get(port); - if (source) { - const count = source.deregister(port); - if (count === 0) { - source.close(); - sourcesByUrl.delete(source.url); - } - } - source = new WsSource(url); - source.register(port); - sourcesByUrl.set(url, source); - sourcesByPort.set(port, source); - } else if (event.data.type === 'close') { - const source = sourcesByPort.get(port); - if (!source) return; - const count = source.deregister(port); - sourcesByPort.delete(port); - if (count === 0) { - source.close(); - sourcesByUrl.delete(source.url); - } - } else if (event.data.type === 'status') { - const source = sourcesByPort.get(port); - if (!source) { - port.postMessage({type: 'status', message: 'not connected'}); - return; - } - port.postMessage({ - type: 'status', - message: `url: ${source.url} readyState: ${source.ws?.readyState ?? 'null'}`, - }); - } else { - port.postMessage({ - type: 'error', - message: `received but don't know how to handle: ${JSON.stringify(event.data)}`, - }); - } - }); - port.start(); - } -});