refactor: Replace SSE with WebSocket

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

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=

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)

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() {

routers/web/websocket/websocket.go

@@ -0,0 +1,51 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"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"
+)
+
+// 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 // CloseNow is best-effort; error is intentionally ignored
+
+	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 := conn.Write(wsCtx, gitea_ws.MessageText, msg); err != nil {
+				log.Trace("websocket: write failed: %v", err)
+				return
+			}
+		}
+	}
+}

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 {

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
+		}
+	}
+}

services/websocket/notifier.go

@@ -0,0 +1,82 @@
+// 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"
+)
+
+// 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"`
+}
+
+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 := nowTS().Add(-2)
+	timer := time.NewTicker(setting.UI.Notification.EventSourceUpdateTime)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			now := nowTS().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
+		}
+	}
+}

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,65 +21,46 @@ 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) => {
       console.error('worker port error', e);
     });
     worker.port.start();
     window.addEventListener('beforeunload', () => {
-      worker.port.postMessage({
-        type: 'close',
-      });
+      worker.port.postMessage({type: 'close'});
       worker.port.close();
     });