diff --git a/assets/go-licenses.json b/assets/go-licenses.json index 96e4407c1c..fbcfc207b9 100644 --- a/assets/go-licenses.json +++ b/assets/go-licenses.json @@ -404,6 +404,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", diff --git a/custom/conf/app.example.ini b/custom/conf/app.example.ini index d93bbd91c1..5e518a1be6 100644 --- a/custom/conf/app.example.ini +++ b/custom/conf/app.example.ini @@ -1312,6 +1312,21 @@ LEVEL = Info ;PROJECT_BOARD_BASIC_KANBAN_TYPE = To Do, In Progress, Done ;PROJECT_BOARD_BUG_TRIAGE_TYPE = Needs Triage, High Priority, Low Priority, Closed +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;[websocket] +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; +;; Backend used to fan out websocket events (notification count, stopwatch, logout). +;; "memory" works for a single process, use "redis" when running multiple Gitea processes. +;PUBSUB_TYPE = memory +;; +;; Redis connection string, used when PUBSUB_TYPE = redis +;; e.g. redis://127.0.0.1:6379/0 +;; When PUBSUB_TYPE is `redis` and this is left empty, it falls back to the shared [redis] CONN_STR. +;PUBSUB_CONN_STR = + ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;[cors] @@ -1471,10 +1486,6 @@ LEVEL = Info ;MIN_TIMEOUT = 10s ;MAX_TIMEOUT = 60s ;TIMEOUT_STEP = 10s -;; -;; This setting determines how often the db is queried to get the latest notification counts. -;; If the browser client supports EventSource and SharedWorker, a SharedWorker will be used in preference to polling notification. Set to -1 to disable the EventSource -;EVENT_SOURCE_UPDATE_TIME = 10s ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; diff --git a/go.mod b/go.mod index 0de3ba3e92..2b5b24ede3 100644 --- a/go.mod +++ b/go.mod @@ -34,6 +34,7 @@ require ( github.com/caddyserver/certmagic v0.25.4 github.com/charmbracelet/git-lfs-transfer v0.1.1-0.20260309112543-12416315a635 github.com/chi-middleware/proxy v1.1.1 + github.com/coder/websocket v1.8.14 github.com/dlclark/regexp2/v2 v2.5.0 github.com/dsnet/compress v0.0.2-0.20230904184137-39efe44ab707 github.com/dustin/go-humanize v1.0.1 diff --git a/go.sum b/go.sum index d0a3b8da6a..398d402968 100644 --- a/go.sum +++ b/go.sum @@ -209,6 +209,8 @@ github.com/clipperhouse/uax29/v2 v2.7.0 h1:+gs4oBZ2gPfVrKPthwbMzWZDaAFPGYK72F0NJ github.com/clipperhouse/uax29/v2 v2.7.0/go.mod h1:EFJ2TJMRUaplDxHKj1qAEhCtQPW2tJSwu5BF98AuoVM= 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= diff --git a/models/activities/notification.go b/models/activities/notification.go index c786811fcd..188106faac 100644 --- a/models/activities/notification.go +++ b/models/activities/notification.go @@ -336,30 +336,33 @@ func GetUIDsAndNotificationCounts(ctx context.Context, since, until timeutil.Tim return res, db.GetEngine(ctx).SQL(sql, NotificationStatusUnread, since, until).Find(&res) } -// SetIssueReadBy sets issue to be read by given user. -func SetIssueReadBy(ctx context.Context, issueID, userID int64) error { +// SetIssueReadBy sets issue to be read by given user. The bool result is true +// when the unread count actually decreased, so callers can skip a push on no-op. +func SetIssueReadBy(ctx context.Context, issueID, userID int64) (bool, error) { if err := issues_model.UpdateIssueUserByRead(ctx, userID, issueID); err != nil { - return err + return false, err } return setIssueNotificationStatusReadIfUnread(ctx, userID, issueID) } -func setIssueNotificationStatusReadIfUnread(ctx context.Context, userID, issueID int64) error { +func setIssueNotificationStatusReadIfUnread(ctx context.Context, userID, issueID int64) (bool, error) { notification, err := GetIssueNotification(ctx, userID, issueID) // ignore if not exists if err != nil { - return nil + return false, nil } if notification.Status != NotificationStatusUnread { - return nil + return false, nil } notification.Status = NotificationStatusRead - _, err = db.GetEngine(ctx).ID(notification.ID).Cols("status").Update(notification) - return err + if _, err := db.GetEngine(ctx).ID(notification.ID).Cols("status").Update(notification); err != nil { + return false, err + } + return true, nil } // SetRepoReadBy sets repo to be visited by given user. @@ -407,12 +410,13 @@ func GetNotificationByID(ctx context.Context, notificationID int64) (*Notificati return notification, nil } -// UpdateNotificationStatuses updates the statuses of all of a user's notifications that are of the currentStatus type to the desiredStatus -func UpdateNotificationStatuses(ctx context.Context, user *user_model.User, currentStatus, desiredStatus NotificationStatus) error { +// UpdateNotificationStatuses updates the statuses of all of a user's notifications +// that are of the currentStatus type to the desiredStatus. Returns the number of +// rows actually changed so callers can skip downstream work on a no-op. +func UpdateNotificationStatuses(ctx context.Context, user *user_model.User, currentStatus, desiredStatus NotificationStatus) (int64, error) { n := &Notification{Status: desiredStatus, UpdatedBy: user.ID} - _, err := db.GetEngine(ctx). + return db.GetEngine(ctx). Where("user_id = ? AND status = ?", user.ID, currentStatus). Cols("status", "updated_by", "updated_unix"). Update(n) - return err } diff --git a/models/activities/notification_list.go b/models/activities/notification_list.go index 70bf6811bc..97170263b2 100644 --- a/models/activities/notification_list.go +++ b/models/activities/notification_list.go @@ -69,25 +69,30 @@ func (opts FindNotificationOptions) ToOrders() string { // CreateOrUpdateIssueNotifications creates an issue notification // for each watcher, or updates it if already exists // receiverID > 0 just send to receiver, else send to all watcher -func CreateOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, notificationAuthorID, receiverID int64) error { - return db.WithTx(ctx, func(ctx context.Context) error { - return createOrUpdateIssueNotifications(ctx, issueID, commentID, notificationAuthorID, receiverID) +// Returns the set of user IDs whose notification rows were created or updated. +func CreateOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, notificationAuthorID, receiverID int64) ([]int64, error) { + var notifiedIDs []int64 + err := db.WithTx(ctx, func(ctx context.Context) error { + var innerErr error + notifiedIDs, innerErr = createOrUpdateIssueNotifications(ctx, issueID, commentID, notificationAuthorID, receiverID) + return innerErr }) + return notifiedIDs, err } -func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, notificationAuthorID, receiverID int64) error { +func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, notificationAuthorID, receiverID int64) ([]int64, error) { // init var toNotify container.Set[int64] notifications, err := db.Find[Notification](ctx, FindNotificationOptions{ IssueID: issueID, }) if err != nil { - return err + return nil, err } issue, err := issues_model.GetIssueByID(ctx, issueID) if err != nil { - return err + return nil, err } if receiverID > 0 { @@ -97,19 +102,19 @@ func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, n toNotify = make(container.Set[int64], 32) issueWatches, err := issues_model.GetIssueWatchersIDs(ctx, issueID, true) if err != nil { - return err + return nil, err } toNotify.AddMultiple(issueWatches...) if !(issue.IsPull && issues_model.HasWorkInProgressPrefix(issue.Title)) { repoWatches, err := repo_model.GetRepoWatchersIDs(ctx, issue.RepoID) if err != nil { - return err + return nil, err } toNotify.AddMultiple(repoWatches...) } issueParticipants, err := issue.GetParticipantIDsByIssue(ctx) if err != nil { - return err + return nil, err } toNotify.AddMultiple(issueParticipants...) @@ -118,19 +123,19 @@ func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, n // explicit unwatch on issue issueUnWatches, err := issues_model.GetIssueWatchersIDs(ctx, issueID, false) if err != nil { - return err + return nil, err } for _, id := range issueUnWatches { toNotify.Remove(id) } } - err = issue.LoadRepo(ctx) - if err != nil { - return err + if err := issue.LoadRepo(ctx); err != nil { + return nil, err } // notify + notifiedIDs := make([]int64, 0, len(toNotify)) for userID := range toNotify { issue.Repo.Units = nil user, err := user_model.GetUserByID(ctx, userID) @@ -139,7 +144,7 @@ func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, n continue } - return err + return nil, err } if issue.IsPull && !access_model.CheckRepoUnitUser(ctx, issue.Repo, user, unit.TypePullRequests) { continue @@ -149,16 +154,16 @@ func createOrUpdateIssueNotifications(ctx context.Context, issueID, commentID, n } if notificationExists(notifications, issue.ID, userID) { - if err = updateIssueNotification(ctx, userID, issue.ID, commentID, notificationAuthorID); err != nil { - return err - } - continue + err = updateIssueNotification(ctx, userID, issue.ID, commentID, notificationAuthorID) + } else { + err = createIssueNotification(ctx, userID, issue, commentID, notificationAuthorID) } - if err = createIssueNotification(ctx, userID, issue, commentID, notificationAuthorID); err != nil { - return err + if err != nil { + return nil, err } + notifiedIDs = append(notifiedIDs, userID) } - return nil + return notifiedIDs, nil } // NotificationList contains a list of notifications diff --git a/models/activities/notification_test.go b/models/activities/notification_test.go index eaf96bf31e..4f978449e2 100644 --- a/models/activities/notification_test.go +++ b/models/activities/notification_test.go @@ -20,7 +20,8 @@ func TestCreateOrUpdateIssueNotifications(t *testing.T) { assert.NoError(t, unittest.PrepareTestDatabase()) issue := unittest.AssertExistsAndLoadBean(t, &issues_model.Issue{ID: 1}) - assert.NoError(t, activities_model.CreateOrUpdateIssueNotifications(t.Context(), issue.ID, 0, 2, 0)) + _, err := activities_model.CreateOrUpdateIssueNotifications(t.Context(), issue.ID, 0, 2, 0) + assert.NoError(t, err) // User 9 is inactive, thus notifications for user 1 and 4 are created notf := unittest.AssertExistsAndLoadBean(t, &activities_model.Notification{UserID: 1, IssueID: issue.ID}) @@ -117,7 +118,8 @@ func TestUpdateNotificationStatuses(t *testing.T) { &activities_model.Notification{UserID: user.ID, Status: activities_model.NotificationStatusRead}) notfPinned := unittest.AssertExistsAndLoadBean(t, &activities_model.Notification{UserID: user.ID, Status: activities_model.NotificationStatusPinned}) - assert.NoError(t, activities_model.UpdateNotificationStatuses(t.Context(), user, activities_model.NotificationStatusUnread, activities_model.NotificationStatusRead)) + _, err := activities_model.UpdateNotificationStatuses(t.Context(), user, activities_model.NotificationStatusUnread, activities_model.NotificationStatusRead) + assert.NoError(t, err) unittest.AssertExistsAndLoadBean(t, &activities_model.Notification{ID: notfUnread.ID, Status: activities_model.NotificationStatusRead}) unittest.AssertExistsAndLoadBean(t, @@ -131,7 +133,8 @@ func TestSetIssueReadBy(t *testing.T) { user := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 1}) issue := unittest.AssertExistsAndLoadBean(t, &issues_model.Issue{ID: 1}) assert.NoError(t, db.WithTx(t.Context(), func(ctx context.Context) error { - return activities_model.SetIssueReadBy(ctx, issue.ID, user.ID) + _, err := activities_model.SetIssueReadBy(ctx, issue.ID, user.ID) + return err })) nt, err := activities_model.GetIssueNotification(t.Context(), user.ID, issue.ID) diff --git a/modules/eventsource/event.go b/modules/eventsource/event.go deleted file mode 100644 index 1f2372d980..0000000000 --- a/modules/eventsource/event.go +++ /dev/null @@ -1,111 +0,0 @@ -// Copyright 2020 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package eventsource - -import ( - "bytes" - "fmt" - "io" - "time" - - "gitea.dev/modules/json" -) - -func wrapNewlines(w io.Writer, prefix, value []byte) (sum int64, err error) { - if len(value) == 0 { - return 0, nil - } - var n int - last := 0 - for j := bytes.IndexByte(value, '\n'); j > -1; j = bytes.IndexByte(value[last:], '\n') { - n, err = w.Write(prefix) - sum += int64(n) - if err != nil { - return sum, err - } - n, err = w.Write(value[last : last+j+1]) - sum += int64(n) - if err != nil { - return sum, err - } - last += j + 1 - } - n, err = w.Write(prefix) - sum += int64(n) - if err != nil { - return sum, err - } - n, err = w.Write(value[last:]) - sum += int64(n) - if err != nil { - return sum, err - } - n, err = w.Write([]byte("\n")) - sum += int64(n) - return sum, err -} - -// Event is an eventsource event, not all fields need to be set -type Event struct { - // Name represents the value of the event: tag in the stream - Name string - // Data is either JSONified []byte or any that can be JSONd - Data any - // ID represents the ID of an event - ID string - // Retry tells the receiver only to attempt to reconnect to the source after this time - Retry time.Duration -} - -// WriteTo writes data to w until there's no more data to write or when an error occurs. -// The return value n is the number of bytes written. Any error encountered during the write is also returned. -func (e *Event) WriteTo(w io.Writer) (int64, error) { - sum := int64(0) - var nint int - n, err := wrapNewlines(w, []byte("event: "), []byte(e.Name)) - sum += n - if err != nil { - return sum, err - } - - if e.Data != nil { - var data []byte - switch v := e.Data.(type) { - case []byte: - data = v - case string: - data = []byte(v) - default: - var err error - data, err = json.Marshal(e.Data) - if err != nil { - return sum, err - } - } - n, err := wrapNewlines(w, []byte("data: "), data) - sum += n - if err != nil { - return sum, err - } - } - - n, err = wrapNewlines(w, []byte("id: "), []byte(e.ID)) - sum += n - if err != nil { - return sum, err - } - - if e.Retry != 0 { - nint, err = fmt.Fprintf(w, "retry: %d\n", int64(e.Retry/time.Millisecond)) - sum += int64(nint) - if err != nil { - return sum, err - } - } - - nint, err = w.Write([]byte("\n")) - sum += int64(nint) - - return sum, err -} diff --git a/modules/eventsource/event_test.go b/modules/eventsource/event_test.go deleted file mode 100644 index a1c3e5c7a8..0000000000 --- a/modules/eventsource/event_test.go +++ /dev/null @@ -1,50 +0,0 @@ -// Copyright 2020 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package eventsource - -import ( - "bytes" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func Test_wrapNewlines(t *testing.T) { - tests := []struct { - name string - prefix string - value string - output string - }{ - { - "check no new lines", - "prefix: ", - "value", - "prefix: value\n", - }, - { - "check simple newline", - "prefix: ", - "value1\nvalue2", - "prefix: value1\nprefix: value2\n", - }, - { - "check pathological newlines", - "p: ", - "\n1\n\n2\n3\n", - "p: \np: 1\np: \np: 2\np: 3\np: \n", - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - w := &bytes.Buffer{} - gotSum, err := wrapNewlines(w, []byte(tt.prefix), []byte(tt.value)) - require.NoError(t, err) - - assert.EqualValues(t, len(tt.output), gotSum) - assert.Equal(t, tt.output, w.String()) - }) - } -} diff --git a/modules/eventsource/manager.go b/modules/eventsource/manager.go deleted file mode 100644 index 7ed2a82903..0000000000 --- a/modules/eventsource/manager.go +++ /dev/null @@ -1,89 +0,0 @@ -// Copyright 2020 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package eventsource - -import ( - "sync" -) - -// Manager manages the eventsource Messengers -type Manager struct { - mutex sync.Mutex - - messengers map[int64]*Messenger - connection chan struct{} -} - -var manager *Manager - -func init() { - manager = &Manager{ - messengers: make(map[int64]*Messenger), - connection: make(chan struct{}, 1), - } -} - -// GetManager returns a Manager and initializes one as singleton if there's none yet -func GetManager() *Manager { - return manager -} - -// Register message channel -func (m *Manager) Register(uid int64) <-chan *Event { - m.mutex.Lock() - messenger, ok := m.messengers[uid] - if !ok { - messenger = NewMessenger(uid) - m.messengers[uid] = messenger - } - select { - case m.connection <- struct{}{}: - default: - } - m.mutex.Unlock() - return messenger.Register() -} - -// Unregister message channel -func (m *Manager) Unregister(uid int64, channel <-chan *Event) { - m.mutex.Lock() - defer m.mutex.Unlock() - messenger, ok := m.messengers[uid] - if !ok { - return - } - if messenger.Unregister(channel) { - delete(m.messengers, uid) - } -} - -// UnregisterAll message channels -func (m *Manager) UnregisterAll() { - m.mutex.Lock() - defer m.mutex.Unlock() - for _, messenger := range m.messengers { - messenger.UnregisterAll() - } - m.messengers = map[int64]*Messenger{} -} - -// SendMessage sends a message to a particular user -func (m *Manager) SendMessage(uid int64, message *Event) { - m.mutex.Lock() - messenger, ok := m.messengers[uid] - m.mutex.Unlock() - if ok { - messenger.SendMessage(message) - } -} - -// SendMessageBlocking sends a message to a particular user -func (m *Manager) SendMessageBlocking(uid int64, message *Event) { - m.mutex.Lock() - messenger, ok := m.messengers[uid] - m.mutex.Unlock() - if ok { - messenger.SendMessageBlocking(message) - } -} diff --git a/modules/eventsource/manager_run.go b/modules/eventsource/manager_run.go deleted file mode 100644 index b39d400813..0000000000 --- a/modules/eventsource/manager_run.go +++ /dev/null @@ -1,122 +0,0 @@ -// Copyright 2020 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package eventsource - -import ( - "context" - "time" - - activities_model "gitea.dev/models/activities" - issues_model "gitea.dev/models/issues" - user_model "gitea.dev/models/user" - "gitea.dev/modules/graceful" - "gitea.dev/modules/json" - "gitea.dev/modules/log" - "gitea.dev/modules/process" - "gitea.dev/modules/setting" - "gitea.dev/modules/timeutil" - "gitea.dev/services/convert" -) - -// Init starts this eventsource -func (m *Manager) Init() { - if setting.UI.Notification.EventSourceUpdateTime <= 0 { - return - } - go graceful.GetManager().RunWithShutdownContext(m.Run) -} - -// Run runs the manager within a provided context -func (m *Manager) Run(ctx context.Context) { - ctx, _, finished := process.GetManager().AddTypedContext(ctx, "Service: EventSource", process.SystemProcessType, true) - defer finished() - - then := timeutil.TimeStampNow().Add(-2) - timer := time.NewTicker(setting.UI.Notification.EventSourceUpdateTime) -loop: - for { - select { - case <-ctx.Done(): - timer.Stop() - break loop - case <-timer.C: - m.mutex.Lock() - connectionCount := len(m.messengers) - if connectionCount == 0 { - log.Trace("Event source has no listeners") - // empty the connection channel - select { - case <-m.connection: - default: - } - } - m.mutex.Unlock() - if connectionCount == 0 { - // No listeners so the source can be paused - log.Trace("Pausing the eventsource") - select { - case <-ctx.Done(): - break loop - case <-m.connection: - log.Trace("Connection detected - restarting the eventsource") - // OK we're back so lets reset the timer and start again - // We won't change the "then" time because there could be concurrency issues - select { - case <-timer.C: - default: - } - continue - } - } - - now := timeutil.TimeStampNow().Add(-2) - - uidCounts, err := activities_model.GetUIDsAndNotificationCounts(ctx, then, now) - if err != nil { - log.Error("Unable to get UIDcounts: %v", err) - } - for _, uidCount := range uidCounts { - m.SendMessage(uidCount.UserID, &Event{ - Name: "notification-count", - Data: uidCount, - }) - } - then = now - - if setting.Service.EnableTimetracking { - usersStopwatches, err := issues_model.GetUIDsAndStopwatch(ctx) - if err != nil { - log.Error("Unable to get GetUIDsAndStopwatch: %v", err) - return - } - - for _, userStopwatches := range usersStopwatches { - u, err := user_model.GetUserByID(ctx, userStopwatches.UserID) - if err != nil { - log.Error("Unable to get user %d: %v", userStopwatches.UserID, err) - continue - } - - apiSWs, err := convert.ToStopWatches(ctx, u, userStopwatches.StopWatches) - if err != nil { - if !issues_model.IsErrIssueNotExist(err) { - log.Error("Unable to APIFormat stopwatches: %v", err) - } - continue - } - dataBs, err := json.Marshal(apiSWs) - if err != nil { - log.Error("Unable to marshal stopwatches: %v", err) - continue - } - m.SendMessage(userStopwatches.UserID, &Event{ - Name: "stopwatches", - Data: string(dataBs), - }) - } - } - } - } - m.UnregisterAll() -} diff --git a/modules/eventsource/messenger.go b/modules/eventsource/messenger.go deleted file mode 100644 index 6df26716be..0000000000 --- a/modules/eventsource/messenger.go +++ /dev/null @@ -1,77 +0,0 @@ -// Copyright 2020 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package eventsource - -import "sync" - -// Messenger is a per uid message store -type Messenger struct { - mutex sync.Mutex - uid int64 - channels []chan *Event -} - -// NewMessenger creates a messenger for a particular uid -func NewMessenger(uid int64) *Messenger { - return &Messenger{ - uid: uid, - channels: [](chan *Event){}, - } -} - -// Register returns a new chan []byte -func (m *Messenger) Register() <-chan *Event { - m.mutex.Lock() - // TODO: Limit the number of messengers per uid - channel := make(chan *Event, 1) - m.channels = append(m.channels, channel) - m.mutex.Unlock() - return channel -} - -// Unregister removes the provider chan []byte -func (m *Messenger) Unregister(channel <-chan *Event) bool { - m.mutex.Lock() - defer m.mutex.Unlock() - for i, toRemove := range m.channels { - if channel == toRemove { - m.channels = append(m.channels[:i], m.channels[i+1:]...) - close(toRemove) - break - } - } - return len(m.channels) == 0 -} - -// UnregisterAll removes all chan []byte -func (m *Messenger) UnregisterAll() { - m.mutex.Lock() - defer m.mutex.Unlock() - for _, channel := range m.channels { - close(channel) - } - m.channels = nil -} - -// SendMessage sends the message to all registered channels -func (m *Messenger) SendMessage(message *Event) { - m.mutex.Lock() - defer m.mutex.Unlock() - for i := range m.channels { - channel := m.channels[i] - select { - case channel <- message: - default: - } - } -} - -// SendMessageBlocking sends the message to all registered channels and ensures it gets sent -func (m *Messenger) SendMessageBlocking(message *Event) { - m.mutex.Lock() - defer m.mutex.Unlock() - for i := range m.channels { - m.channels[i] <- message - } -} diff --git a/modules/json/json.go b/modules/json/json.go index d053f91cf7..75c5f6ec97 100644 --- a/modules/json/json.go +++ b/modules/json/json.go @@ -23,6 +23,7 @@ type Decoder interface { // Interface represents an interface to handle json data type Interface interface { Marshal(v any) ([]byte, error) + MarshalWrite(w io.Writer, v any) error Unmarshal(data []byte, v any) error NewEncoder(writer io.Writer) Encoder NewDecoder(reader io.Reader) Decoder @@ -36,6 +37,11 @@ func Marshal(v any) ([]byte, error) { return DefaultJSONHandler.Marshal(v) } +// MarshalWrite writes the JSON encoding of v to the given writer +func MarshalWrite(w io.Writer, v any) error { + return DefaultJSONHandler.MarshalWrite(w, v) +} + // Unmarshal decodes object from bytes func Unmarshal(data []byte, v any) error { return DefaultJSONHandler.Unmarshal(data, v) diff --git a/modules/json/jsonv1.go b/modules/json/jsonv1.go index 55ec4736b9..e7e4be5778 100644 --- a/modules/json/jsonv1.go +++ b/modules/json/jsonv1.go @@ -17,6 +17,10 @@ func (jsonV1) Marshal(v any) ([]byte, error) { return json.Marshal(v) } +func (jsonV1) MarshalWrite(w io.Writer, v any) error { + return json.NewEncoder(w).Encode(v) +} + func (jsonV1) Unmarshal(data []byte, v any) error { return json.Unmarshal(data, v) } diff --git a/modules/json/jsonv2.go b/modules/json/jsonv2.go index c4afc9513b..3385327905 100644 --- a/modules/json/jsonv2.go +++ b/modules/json/jsonv2.go @@ -53,6 +53,10 @@ func (j *JSONv2) Marshal(v any) ([]byte, error) { return jsonv2.Marshal(v, j.marshalOptions) } +func (j *JSONv2) MarshalWrite(w io.Writer, v any) error { + return jsonv2.MarshalWrite(w, v, j.marshalOptions) +} + func (j *JSONv2) Unmarshal(data []byte, v any) error { return jsonv2.Unmarshal(data, v, j.unmarshalOptions) } diff --git a/modules/queue/base_redis_test.go b/modules/queue/base_redis_test.go index b03702d951..329a40cfd5 100644 --- a/modules/queue/base_redis_test.go +++ b/modules/queue/base_redis_test.go @@ -4,68 +4,15 @@ package queue import ( - "context" - "os" - "os/exec" "testing" - "time" - "gitea.dev/modules/nosql" "gitea.dev/modules/setting" "gitea.dev/modules/test" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) -func waitRedisReady(conn string, dur time.Duration) (ready bool) { - ctxTimed, cancel := context.WithTimeout(context.Background(), time.Second*5) - defer cancel() - for t := time.Now(); ; time.Sleep(50 * time.Millisecond) { - ret := nosql.GetManager().GetRedisClient(conn).Ping(ctxTimed) - if ret.Err() == nil { - return true - } - if time.Since(t) > dur { - return false - } - } -} - -func redisServerCmd(t *testing.T) *exec.Cmd { - redisServerProg, err := exec.LookPath("redis-server") - if err != nil { - return nil - } - c := &exec.Cmd{ - Path: redisServerProg, - Args: []string{redisServerProg, "--bind", "127.0.0.1", "--port", "6379"}, - Dir: t.TempDir(), - Stdin: os.Stdin, - Stdout: os.Stdout, - Stderr: os.Stderr, - } - return c -} - func TestBaseRedis(t *testing.T) { - var redisServer *exec.Cmd - defer func() { - if redisServer != nil { - _ = redisServer.Process.Signal(os.Interrupt) - _ = redisServer.Wait() - } - }() - if !waitRedisReady("redis://127.0.0.1:6379/0", 0) { - redisServer = redisServerCmd(t) - if redisServer == nil && test.AllowSkipExternalService() { - t.Skip("redis server command not found, skipped") - } - require.NotNil(t, redisServer) - assert.NoError(t, redisServer.Start()) - require.True(t, waitRedisReady("redis://127.0.0.1:6379/0", 5*time.Second), "start redis-server") - } - - testQueueBasic(t, newBaseRedisSimple, toBaseConfig("baseRedis", setting.QueueSettings{Length: 10}), false) - testQueueBasic(t, newBaseRedisUnique, toBaseConfig("baseRedisUnique", setting.QueueSettings{Length: 10}), true) + redisConn := test.PrepareTestRedis(t) + queueSetting := setting.QueueSettings{Length: 10, ConnStr: redisConn} + testQueueBasic(t, newBaseRedisSimple, toBaseConfig("baseRedis", queueSetting), false) + testQueueBasic(t, newBaseRedisUnique, toBaseConfig("baseRedisUnique", queueSetting), true) } diff --git a/modules/setting/setting.go b/modules/setting/setting.go index 790b8fbb7c..a20c202828 100644 --- a/modules/setting/setting.go +++ b/modules/setting/setting.go @@ -221,6 +221,7 @@ func LoadSettings() { loadServiceFrom(CfgProvider) loadOAuth2ClientFrom(CfgProvider) loadCacheFrom(CfgProvider) + loadWebsocketFrom(CfgProvider) loadSessionFrom(CfgProvider) loadCorsFrom(CfgProvider) loadMailsFrom(CfgProvider) diff --git a/modules/setting/ui.go b/modules/setting/ui.go index 5990c7a74d..23204daf87 100644 --- a/modules/setting/ui.go +++ b/modules/setting/ui.go @@ -52,10 +52,9 @@ var UI = struct { DefaultShowFullName bool Notification struct { - MinTimeout time.Duration - TimeoutStep time.Duration - MaxTimeout time.Duration - EventSourceUpdateTime time.Duration + MinTimeout time.Duration + TimeoutStep time.Duration + MaxTimeout time.Duration } `ini:"ui.notification"` SVG struct { @@ -107,15 +106,13 @@ var UI = struct { AmbiguousUnicodeDetection: true, Notification: struct { - MinTimeout time.Duration - TimeoutStep time.Duration - MaxTimeout time.Duration - EventSourceUpdateTime time.Duration + MinTimeout time.Duration + TimeoutStep time.Duration + MaxTimeout time.Duration }{ - MinTimeout: 10 * time.Second, - TimeoutStep: 10 * time.Second, - MaxTimeout: 60 * time.Second, - EventSourceUpdateTime: 10 * time.Second, + MinTimeout: 10 * time.Second, + TimeoutStep: 10 * time.Second, + MaxTimeout: 60 * time.Second, }, SVG: struct { Enabled bool `ini:"ENABLE_RENDER"` diff --git a/modules/setting/websocket.go b/modules/setting/websocket.go new file mode 100644 index 0000000000..89f8e65e36 --- /dev/null +++ b/modules/setting/websocket.go @@ -0,0 +1,34 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package setting + +import "gitea.dev/modules/log" + +const ( + PubsubTypeMemory = "memory" + PubsubTypeRedis = "redis" +) + +// Websocket holds the settings for the websocket event delivery. The pubsub +// backend is scoped to websocket messages only, it is not a general-purpose +// pubsub service. +type WebsocketConfig struct { + PubsubType string + PubsubConnStr string +} + +var Websocket = WebsocketConfig{ + PubsubType: PubsubTypeMemory, +} + +func loadWebsocketFrom(rootCfg ConfigProvider) { + sec := rootCfg.Section("websocket") + Websocket.PubsubType = sec.Key("PUBSUB_TYPE").In(PubsubTypeMemory, []string{PubsubTypeMemory, PubsubTypeRedis}) + if Websocket.PubsubType == PubsubTypeRedis { + Websocket.PubsubConnStr = sec.Key("PUBSUB_CONN_STR").MustString(Redis.ConnStr) + if Websocket.PubsubConnStr == "" { + log.Fatal("[websocket].PUBSUB_CONN_STR is required when PUBSUB_TYPE = redis") + } + } +} diff --git a/modules/setting/websocket_test.go b/modules/setting/websocket_test.go new file mode 100644 index 0000000000..25dc936446 --- /dev/null +++ b/modules/setting/websocket_test.go @@ -0,0 +1,63 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package setting + +import ( + "testing" + + "gitea.dev/modules/test" + + "github.com/stretchr/testify/assert" +) + +func TestLoadWebsocketConfig(t *testing.T) { + cases := []struct { + name string + ini string + wantType string + wantConn string + }{ + { + name: "defaults to memory", + wantType: PubsubTypeMemory, + }, + { + name: "redis with its own conn str", + ini: "[websocket]\nPUBSUB_TYPE = redis\nPUBSUB_CONN_STR = redis://127.0.0.1:6379/0", + wantType: PubsubTypeRedis, + wantConn: "redis://127.0.0.1:6379/0", + }, + { + name: "redis falls back to the shared [redis] section", + ini: "[redis]\nCONN_STR = redis://127.0.0.1:6379/0\n[websocket]\nPUBSUB_TYPE = redis", + wantType: PubsubTypeRedis, + wantConn: "redis://127.0.0.1:6379/0", + }, + { + name: "own conn str wins over the shared one", + ini: "[redis]\nCONN_STR = redis://127.0.0.1:6379/0\n[websocket]\nPUBSUB_TYPE = redis\nPUBSUB_CONN_STR = redis://10.0.0.1:6379/1", + wantType: PubsubTypeRedis, + wantConn: "redis://10.0.0.1:6379/1", + }, + { + name: "memory ignores the shared [redis] section", + ini: "[redis]\nCONN_STR = redis://127.0.0.1:6379/0\n[websocket]\nPUBSUB_TYPE = memory", + wantType: PubsubTypeMemory, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + defer test.MockVariableValue(&Websocket)() + defer test.MockVariableValue(&Redis)() + cfg, err := NewConfigProviderFromData(tc.ini) + assert.NoError(t, err) + + loadRedisFrom(cfg) + loadWebsocketFrom(cfg) + assert.Equal(t, tc.wantType, Websocket.PubsubType) + assert.Equal(t, tc.wantConn, Websocket.PubsubConnStr) + }) + } +} diff --git a/modules/templates/helper.go b/modules/templates/helper.go index 662b89998e..a97b714b78 100644 --- a/modules/templates/helper.go +++ b/modules/templates/helper.go @@ -116,10 +116,9 @@ func newFuncMapWebPage() template.FuncMap { }, "NotificationSettings": func() map[string]any { return map[string]any{ - "MinTimeout": int(setting.UI.Notification.MinTimeout / time.Millisecond), - "TimeoutStep": int(setting.UI.Notification.TimeoutStep / time.Millisecond), - "MaxTimeout": int(setting.UI.Notification.MaxTimeout / time.Millisecond), - "EventSourceUpdateTime": int(setting.UI.Notification.EventSourceUpdateTime / time.Millisecond), + "MinTimeout": int(setting.UI.Notification.MinTimeout / time.Millisecond), + "TimeoutStep": int(setting.UI.Notification.TimeoutStep / time.Millisecond), + "MaxTimeout": int(setting.UI.Notification.MaxTimeout / time.Millisecond), } }, "MermaidMaxSourceCharacters": func() int { diff --git a/modules/test/redis.go b/modules/test/redis.go new file mode 100644 index 0000000000..fbc02814e1 --- /dev/null +++ b/modules/test/redis.go @@ -0,0 +1,78 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package test + +import ( + "net" + "os" + "os/exec" + "time" +) + +const ( + testRedisHost = "127.0.0.1" + testRedisPort = "6379" + testRedisAddr = testRedisHost + ":" + testRedisPort + testRedisConnStr = "redis://" + testRedisAddr + "/0" +) + +// waitRedisReady reports whether redis accepts connections within dur. Redis +// binds its listener last during startup, so a successful dial means it can +// serve. A plain dial, not a redis PING: the client retries its pool on a +// refused connect, which makes the "is one already running" probe take ~1s. +func waitRedisReady(dur time.Duration) bool { + for start := time.Now(); ; time.Sleep(50 * time.Millisecond) { + conn, err := net.DialTimeout("tcp", testRedisAddr, time.Second) + if err == nil { + _ = conn.Close() + return true + } + if time.Since(start) > dur { + return false + } + } +} + +func redisServerCmd(t TestingT) *exec.Cmd { + redisServerProg, err := exec.LookPath("redis-server") + if err != nil { + return nil + } + return &exec.Cmd{ + Path: redisServerProg, + Args: []string{redisServerProg, "--bind", testRedisHost, "--port", testRedisPort}, + Dir: t.TempDir(), + Stdin: os.Stdin, + Stdout: os.Stdout, + Stderr: os.Stderr, + } +} + +// PrepareTestRedis returns a connection string to a running redis, starting one +// for the duration of the test if the port is free. +func PrepareTestRedis(t TestingT) string { + if waitRedisReady(0) { + return testRedisConnStr + } + redisServer := redisServerCmd(t) + if redisServer == nil { + if AllowSkipExternalService() { + t.Skipf("redis-server command not found, skipped") + } else { + t.Fatalf("no redis server or command, but skipping is not allowed") + } + return testRedisConnStr + } + if err := redisServer.Start(); err != nil { + t.Fatalf("failed to start redis-server: %v", err) + } + t.Cleanup(func() { + _ = redisServer.Process.Signal(os.Interrupt) + _ = redisServer.Wait() + }) + if !waitRedisReady(5 * time.Second) { + t.Fatalf("failed to start redis-server") + } + return testRedisConnStr +} diff --git a/modules/test/utils.go b/modules/test/utils.go index 671af5a340..aa2dfb702f 100644 --- a/modules/test/utils.go +++ b/modules/test/utils.go @@ -157,11 +157,13 @@ var AllowSkipExternalService = sync.OnceValue(func() bool { }) type TestingT interface { + Cleanup(func()) Context() context.Context - Helper() - Skipf(format string, args ...any) Errorf(format string, args ...any) Fatalf(format string, args ...any) + Helper() + Skipf(format string, args ...any) + TempDir() string } func ExternalServiceHTTP(t TestingT, envVarName, def string) string { diff --git a/modules/web/routing/logger_manager.go b/modules/web/routing/logger_manager.go index 7e3915013a..e0bf3b3263 100644 --- a/modules/web/routing/logger_manager.go +++ b/modules/web/routing/logger_manager.go @@ -41,7 +41,7 @@ func (manager *loggerRequestManager) startSlowQueryDetector(threshold time.Durat ctx, _, finished := process.GetManager().AddTypedContext(ctx, "Service: SlowQueryDetector", process.SystemProcessType, true) defer finished() // This go-routine checks all active requests every second. - // If a request has been running for a long time (eg: /user/events), we also print a log with "still-executing" message + // If a request has been running for a long time, we also print a log with "still-executing" message // After the "still-executing" log is printed, the record will be removed from the map to prevent from duplicated logs in future // We do not care about accurate duration here. It just does the check periodically, 0.5s or 1.5s are all OK. t := time.NewTicker(time.Second) diff --git a/routers/api/v1/notify/repo.go b/routers/api/v1/notify/repo.go index 7ba4469ff6..e9f0eefe7b 100644 --- a/routers/api/v1/notify/repo.go +++ b/routers/api/v1/notify/repo.go @@ -13,6 +13,7 @@ import ( "gitea.dev/modules/structs" "gitea.dev/services/context" "gitea.dev/services/convert" + "gitea.dev/services/notifications" ) func statusStringToNotificationStatus(status string) activities_model.NotificationStatus { @@ -212,14 +213,13 @@ func ReadRepoNotifications(ctx *context.APIContext) { targetStatus = activities_model.NotificationStatusRead } - changed := make([]*structs.NotificationThread, 0, len(nl)) - - for _, n := range nl { - notif, err := activities_model.SetNotificationStatus(ctx, n.ID, ctx.Doer, targetStatus) - if err != nil { - ctx.APIErrorInternal(err) - return - } + updated, err := notifications.SetManyNotificationStatuses(ctx, nl, ctx.Doer, targetStatus) + if err != nil { + ctx.APIErrorInternal(err) + return + } + changed := make([]*structs.NotificationThread, 0, len(updated)) + for _, notif := range updated { _ = notif.LoadAttributes(ctx) changed = append(changed, convert.ToNotificationThread(ctx, notif)) } diff --git a/routers/api/v1/notify/threads.go b/routers/api/v1/notify/threads.go index c66db13f8b..7fdf7d7846 100644 --- a/routers/api/v1/notify/threads.go +++ b/routers/api/v1/notify/threads.go @@ -12,6 +12,7 @@ import ( issues_model "gitea.dev/models/issues" "gitea.dev/services/context" "gitea.dev/services/convert" + "gitea.dev/services/notifications" ) // GetThread get notification by ID @@ -88,7 +89,7 @@ func ReadThread(ctx *context.APIContext) { targetStatus = activities_model.NotificationStatusRead } - notif, err := activities_model.SetNotificationStatus(ctx, n.ID, ctx.Doer, targetStatus) + notif, err := notifications.SetNotificationStatus(ctx, n.ID, ctx.Doer, targetStatus) if err != nil { ctx.APIErrorInternal(err) return diff --git a/routers/api/v1/notify/user.go b/routers/api/v1/notify/user.go index 9894f577c5..8bcb6fee6c 100644 --- a/routers/api/v1/notify/user.go +++ b/routers/api/v1/notify/user.go @@ -12,6 +12,7 @@ import ( "gitea.dev/modules/structs" "gitea.dev/services/context" "gitea.dev/services/convert" + "gitea.dev/services/notifications" ) // ListNotifications list users's notification threads @@ -160,14 +161,13 @@ func ReadNotifications(ctx *context.APIContext) { targetStatus = activities_model.NotificationStatusRead } - changed := make([]*structs.NotificationThread, 0, len(nl)) - - for _, n := range nl { - notif, err := activities_model.SetNotificationStatus(ctx, n.ID, ctx.Doer, targetStatus) - if err != nil { - ctx.APIErrorInternal(err) - return - } + updated, err := notifications.SetManyNotificationStatuses(ctx, nl, ctx.Doer, targetStatus) + if err != nil { + ctx.APIErrorInternal(err) + return + } + changed := make([]*structs.NotificationThread, 0, len(updated)) + for _, notif := range updated { _ = notif.LoadAttributes(ctx) changed = append(changed, convert.ToNotificationThread(ctx, notif)) } diff --git a/routers/api/v1/repo/issue_stopwatch.go b/routers/api/v1/repo/issue_stopwatch.go index 36f24ed27c..130b76f880 100644 --- a/routers/api/v1/repo/issue_stopwatch.go +++ b/routers/api/v1/repo/issue_stopwatch.go @@ -10,6 +10,7 @@ import ( "gitea.dev/routers/api/v1/utils" "gitea.dev/services/context" "gitea.dev/services/convert" + notify_service "gitea.dev/services/notify" ) // StartIssueStopwatch creates a stopwatch for the given issue. @@ -60,6 +61,7 @@ func StartIssueStopwatch(ctx *context.APIContext) { ctx.APIError(http.StatusConflict, "cannot start a stopwatch again if it already exists") return } + notify_service.StopwatchChanged(ctx, ctx.Doer) ctx.Status(http.StatusCreated) } @@ -112,6 +114,7 @@ func StopIssueStopwatch(ctx *context.APIContext) { ctx.APIError(http.StatusConflict, "cannot stop a non-existent stopwatch") return } + notify_service.StopwatchChanged(ctx, ctx.Doer) ctx.Status(http.StatusCreated) } @@ -163,6 +166,7 @@ func DeleteIssueStopwatch(ctx *context.APIContext) { ctx.APIError(http.StatusConflict, "cannot cancel a non-existent stopwatch") return } + notify_service.StopwatchChanged(ctx, ctx.Doer) ctx.Status(http.StatusNoContent) } diff --git a/routers/api/v1/repo/pull.go b/routers/api/v1/repo/pull.go index 8e8d599df5..8a5a09a65f 100644 --- a/routers/api/v1/repo/pull.go +++ b/routers/api/v1/repo/pull.go @@ -12,7 +12,6 @@ import ( "strings" "time" - activities_model "gitea.dev/models/activities" git_model "gitea.dev/models/git" issues_model "gitea.dev/models/issues" access_model "gitea.dev/models/perm/access" @@ -41,6 +40,7 @@ import ( git_service "gitea.dev/services/git" "gitea.dev/services/gitdiff" issue_service "gitea.dev/services/issue" + "gitea.dev/services/notifications" notify_service "gitea.dev/services/notify" pull_service "gitea.dev/services/pull" repo_service "gitea.dev/services/repository" @@ -943,7 +943,7 @@ func MergePullRequest(ctx *context.APIContext) { if ctx.IsSigned { // Update issue-user. - if err = activities_model.SetIssueReadBy(ctx, pr.Issue.ID, ctx.Doer.ID); err != nil { + if err := notifications.SetIssueReadBy(ctx, pr.Issue.ID, ctx.Doer.ID); err != nil { ctx.APIErrorInternal(err) return } diff --git a/routers/init.go b/routers/init.go index 9385fac96a..483b61726a 100644 --- a/routers/init.go +++ b/routers/init.go @@ -12,7 +12,6 @@ import ( "gitea.dev/models" authmodel "gitea.dev/models/auth" "gitea.dev/modules/cache" - "gitea.dev/modules/eventsource" "gitea.dev/modules/git" "gitea.dev/modules/git/gitcmd" "gitea.dev/modules/log" @@ -55,6 +54,7 @@ import ( "gitea.dev/services/task" "gitea.dev/services/uinotification" "gitea.dev/services/webhook" + websocket_service "gitea.dev/services/websocket" ) func mustInit(fn func() error) { @@ -155,7 +155,7 @@ func InitWebInstalled(ctx context.Context) { mustInit(automerge.Init) 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/auth/auth.go b/routers/web/auth/auth.go index 644eede060..b60592e39b 100644 --- a/routers/web/auth/auth.go +++ b/routers/web/auth/auth.go @@ -16,7 +16,6 @@ import ( "gitea.dev/models/db" user_model "gitea.dev/models/user" "gitea.dev/modules/auth/password" - "gitea.dev/modules/eventsource" "gitea.dev/modules/httplib" "gitea.dev/modules/log" "gitea.dev/modules/optional" @@ -34,6 +33,7 @@ import ( "gitea.dev/services/forms" "gitea.dev/services/mailer" user_service "gitea.dev/services/user" + websocket_service "gitea.dev/services/websocket" "github.com/markbates/goth" ) @@ -460,10 +460,7 @@ func HandleSignOut(ctx *context.Context) { // SignOut sign out from login status func SignOut(ctx *context.Context) { if ctx.Doer != nil { - eventsource.GetManager().SendMessageBlocking(ctx.Doer.ID, &eventsource.Event{ - Name: "logout", - Data: ctx.Session.ID(), - }) + websocket_service.PublishLogout(ctx.Doer.ID, ctx.Session.ID()) } exitedImpersonated, err := auth_service.ExitImpersonatedUser(ctx.Session) diff --git a/routers/web/events/events.go b/routers/web/events/events.go deleted file mode 100644 index 2bbc326f08..0000000000 --- a/routers/web/events/events.go +++ /dev/null @@ -1,124 +0,0 @@ -// Copyright 2020 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package events - -import ( - "net/http" - "time" - - "gitea.dev/modules/eventsource" - "gitea.dev/modules/graceful" - "gitea.dev/modules/log" - "gitea.dev/routers/web/auth" - "gitea.dev/services/context" -) - -// Events listens for events -func Events(ctx *context.Context) { - // FIXME: Need to check if resp is actually a http.Flusher! - how though? - - // Set the headers related to event streaming. - ctx.Resp.Header().Set("Content-Type", "text/event-stream") - ctx.Resp.Header().Set("Cache-Control", "no-cache") - ctx.Resp.Header().Set("Connection", "keep-alive") - ctx.Resp.Header().Set("X-Accel-Buffering", "no") - ctx.Resp.WriteHeader(http.StatusOK) - - if !ctx.IsSigned { - // Return unauthorized status event - event := &eventsource.Event{ - Name: "close", - Data: "unauthorized", - } - _, _ = event.WriteTo(ctx) - ctx.Resp.Flush() - return - } - - // Listen to connection close and un-register messageChan - notify := ctx.Done() - - shutdownCtx := graceful.GetManager().ShutdownContext() - - uid := ctx.Doer.ID - - messageChan := eventsource.GetManager().Register(uid) - - unregister := func() { - eventsource.GetManager().Unregister(uid, messageChan) - // ensure the messageChan is closed - for { - _, ok := <-messageChan - if !ok { - break - } - } - } - - // send the initial response bytes only after registering messageChan, so a client whose - // connection is open can rely on receiving all subsequent events - if _, err := ctx.Resp.Write([]byte("\n")); err != nil { - log.Error("Unable to write to EventStream: %v", err) - unregister() - return - } - ctx.Resp.Flush() - - timer := time.NewTicker(30 * time.Second) - -loop: - for { - select { - case <-timer.C: - event := &eventsource.Event{ - Name: "ping", - } - _, err := event.WriteTo(ctx.Resp) - if err != nil { - log.Error("Unable to write to EventStream for user %s: %v", ctx.Doer.Name, err) - go unregister() - break loop - } - ctx.Resp.Flush() - case <-notify: - go unregister() - break loop - case <-shutdownCtx.Done(): - go unregister() - break loop - case event, ok := <-messageChan: - if !ok { - break loop - } - - // Handle logout - if event.Name == "logout" { - if ctx.Session.ID() == event.Data { - _, _ = (&eventsource.Event{ - Name: "logout", - Data: "here", - }).WriteTo(ctx.Resp) - ctx.Resp.Flush() - go unregister() - auth.HandleSignOut(ctx) - break loop - } - // Replace the event - we don't want to expose the session ID to the user - event = &eventsource.Event{ - Name: "logout", - Data: "elsewhere", - } - } - - _, err := event.WriteTo(ctx.Resp) - if err != nil { - log.Error("Unable to write to EventStream for user %s: %v", ctx.Doer.Name, err) - go unregister() - break loop - } - ctx.Resp.Flush() - } - } - timer.Stop() -} diff --git a/routers/web/repo/issue_stopwatch.go b/routers/web/repo/issue_stopwatch.go index 9b6d83f8c8..14ae07d456 100644 --- a/routers/web/repo/issue_stopwatch.go +++ b/routers/web/repo/issue_stopwatch.go @@ -4,10 +4,9 @@ package repo import ( - "gitea.dev/models/db" issues_model "gitea.dev/models/issues" - "gitea.dev/modules/eventsource" "gitea.dev/services/context" + notify_service "gitea.dev/services/notify" ) // IssueStartStopwatch creates a stopwatch for the given issue. @@ -29,6 +28,7 @@ func IssueStartStopwatch(c *context.Context) { c.Flash.Warning(c.Tr("repo.issues.stopwatch_already_created")) } else { c.Flash.Success(c.Tr("repo.issues.tracker_auto_close")) + notify_service.StopwatchChanged(c, c.Doer) } c.JSONRedirect("") } @@ -50,6 +50,8 @@ func IssueStopStopwatch(c *context.Context) { return } else if !ok { c.Flash.Warning(c.Tr("repo.issues.stopwatch_already_stopped")) + } else { + notify_service.StopwatchChanged(c, c.Doer) } c.JSONRedirect("") } @@ -70,17 +72,6 @@ func CancelStopwatch(c *context.Context) { return } - stopwatches, err := issues_model.GetUserStopwatches(c, c.Doer.ID, db.ListOptions{}) - if err != nil { - c.ServerError("GetUserStopwatches", err) - return - } - if len(stopwatches) == 0 { - eventsource.GetManager().SendMessage(c.Doer.ID, &eventsource.Event{ - Name: "stopwatches", - Data: "{}", - }) - } - + notify_service.StopwatchChanged(c, c.Doer) c.JSONRedirect("") } diff --git a/routers/web/repo/issue_view.go b/routers/web/repo/issue_view.go index f78d04efa9..db9830d450 100644 --- a/routers/web/repo/issue_view.go +++ b/routers/web/repo/issue_view.go @@ -11,7 +11,6 @@ import ( "sort" "strconv" - activities_model "gitea.dev/models/activities" asymkey_model "gitea.dev/models/asymkey" "gitea.dev/models/db" git_model "gitea.dev/models/git" @@ -37,6 +36,7 @@ import ( "gitea.dev/services/context" "gitea.dev/services/context/upload" issue_service "gitea.dev/services/issue" + "gitea.dev/services/notifications" pull_service "gitea.dev/services/pull" user_service "gitea.dev/services/user" ) @@ -355,7 +355,7 @@ func ViewIssue(ctx *context.Context) { if ctx.IsSigned { // Update issue-user. - if err := activities_model.SetIssueReadBy(ctx, issue.ID, ctx.Doer.ID); err != nil { + if err := notifications.SetIssueReadBy(ctx, issue.ID, ctx.Doer.ID); err != nil { ctx.ServerError("ReadBy", err) return } diff --git a/routers/web/repo/pull.go b/routers/web/repo/pull.go index 64674bd63c..12b1eca579 100644 --- a/routers/web/repo/pull.go +++ b/routers/web/repo/pull.go @@ -15,7 +15,6 @@ import ( "strings" "time" - activities_model "gitea.dev/models/activities" "gitea.dev/models/db" git_model "gitea.dev/models/git" issues_model "gitea.dev/models/issues" @@ -49,6 +48,7 @@ import ( "gitea.dev/services/forms" git_service "gitea.dev/services/git" "gitea.dev/services/gitdiff" + "gitea.dev/services/notifications" notify_service "gitea.dev/services/notify" pull_service "gitea.dev/services/pull" repo_service "gitea.dev/services/repository" @@ -151,7 +151,7 @@ func getPullInfo(ctx *context.Context) (issue *issues_model.Issue, ok bool) { if ctx.IsSigned { // Update issue-user. - if err = activities_model.SetIssueReadBy(ctx, issue.ID, ctx.Doer.ID); err != nil { + if err := notifications.SetIssueReadBy(ctx, issue.ID, ctx.Doer.ID); err != nil { ctx.ServerError("ReadBy", err) return nil, false } @@ -1301,8 +1301,12 @@ func CancelAutoMergePullRequest(ctx *context.Context) { } func stopTimerIfAvailable(ctx *context.Context, user *user_model.User, issue *issues_model.Issue) error { - _, err := issues_model.FinishIssueStopwatch(ctx, user, issue) - return err + stopped, err := issues_model.FinishIssueStopwatch(ctx, user, issue) + if err != nil || !stopped { + return err + } + notify_service.StopwatchChanged(ctx, user) + return nil } func PullsNewRedirect(ctx *context.Context) { diff --git a/routers/web/user/notification.go b/routers/web/user/notification.go index b82b32f687..6301e52542 100644 --- a/routers/web/user/notification.go +++ b/routers/web/user/notification.go @@ -27,6 +27,7 @@ import ( "gitea.dev/modules/util" "gitea.dev/services/context" issue_service "gitea.dev/services/issue" + "gitea.dev/services/notifications" pull_service "gitea.dev/services/pull" ) @@ -175,7 +176,7 @@ func NotificationStatusPost(ctx *context.Context) { default: return // ignore user's invalid input } - if _, err := activities_model.SetNotificationStatus(ctx, notificationID, ctx.Doer, newStatus); err != nil { + if _, err := notifications.SetNotificationStatus(ctx, notificationID, ctx.Doer, newStatus); err != nil { ctx.ServerError("SetNotificationStatus", err) return } @@ -189,9 +190,8 @@ func NotificationStatusPost(ctx *context.Context) { // NotificationPurgePost is a route for 'purging' the list of notifications - marking all unread as read func NotificationPurgePost(ctx *context.Context) { - err := activities_model.UpdateNotificationStatuses(ctx, ctx.Doer, activities_model.NotificationStatusUnread, activities_model.NotificationStatusRead) - if err != nil { - ctx.ServerError("UpdateNotificationStatuses", err) + if err := notifications.MarkAllRead(ctx, ctx.Doer); err != nil { + ctx.ServerError("MarkAllRead", err) return } diff --git a/routers/web/web.go b/routers/web/web.go index b3f27c79f9..99ffc06756 100644 --- a/routers/web/web.go +++ b/routers/web/web.go @@ -29,7 +29,6 @@ import ( "gitea.dev/routers/web/admin" "gitea.dev/routers/web/auth" "gitea.dev/routers/web/devtest" - "gitea.dev/routers/web/events" "gitea.dev/routers/web/explore" "gitea.dev/routers/web/feed" "gitea.dev/routers/web/healthcheck" @@ -43,6 +42,7 @@ import ( "gitea.dev/routers/web/user" user_setting "gitea.dev/routers/web/user/setting" "gitea.dev/routers/web/user/setting/security" + gitea_websocket "gitea.dev/routers/web/websocket" auth_service "gitea.dev/services/auth" "gitea.dev/services/context" "gitea.dev/services/forms" @@ -599,7 +599,7 @@ func registerWebRoutes(m *web.Router, webAuth *AuthMiddleware) { }) }, reqSignOut) - m.Any("/user/events", routing.MarkLongPolling(), events.Events) + m.Get("/-/ws", routing.MarkLongPolling(), 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..cc9c75bc3c --- /dev/null +++ b/routers/web/websocket/websocket.go @@ -0,0 +1,117 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +import ( + gocontext "context" + "net/http" + "time" + + "gitea.dev/modules/graceful" + "gitea.dev/modules/json" + "gitea.dev/modules/log" + "gitea.dev/services/context" + "gitea.dev/services/pubsub" + websocket_service "gitea.dev/services/websocket" + + gitea_ws "github.com/coder/websocket" +) + +const ( + pingInterval = 30 * time.Second + pingTimeout = 10 * time.Second + writeTimeout = 10 * time.Second + + // First code in the IANA library/framework reserved range (3000–3999). + // Sentinel for an unauthenticated session so the SharedWorker can tell + // "your cookie is gone" apart from a transient network failure and stop + // reconnecting in a tight loop. + 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. + if ctx.Req.Header.Get("Upgrade") == "" { + ctx.Resp.Header().Set("Connection", "Upgrade") + ctx.Resp.Header().Set("Upgrade", "websocket") + ctx.Resp.WriteHeader(http.StatusUpgradeRequired) + return + } + + conn, err := gitea_ws.Accept(ctx.Resp, ctx.Req, nil) + if err != nil { + log.Error("websocket: accept failed: %v", err) + return + } + 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. + wsCtx := conn.CloseRead(ctx.Req.Context()) + shutdownCtx := graceful.GetManager().ShutdownContext() + pingTicker := time.NewTicker(pingInterval) + defer pingTicker.Stop() + + for { + select { + case <-wsCtx.Done(): + return + case <-shutdownCtx.Done(): + return + case <-pingTicker.C: + pingCtx, cancelPing := gocontext.WithTimeout(wsCtx, pingTimeout) + err := conn.Ping(pingCtx) + cancelPing() + if err != nil { + log.Trace("websocket: ping failed: %v", err) + return + } + case brokerPayload, ok := <-ch: + 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) + return + } + } + } +} diff --git a/routers/web/websocket/websocket_test.go b/routers/web/websocket/websocket_test.go new file mode 100644 index 0000000000..421fb2f763 --- /dev/null +++ b/routers/web/websocket/websocket_test.go @@ -0,0 +1,60 @@ +// 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/issue/status.go b/services/issue/status.go index 8ce44d5cd6..d162990bd8 100644 --- a/services/issue/status.go +++ b/services/issue/status.go @@ -16,6 +16,7 @@ import ( // CloseIssue close an issue. func CloseIssue(ctx context.Context, issue *issues_model.Issue, doer *user_model.User, commitID string) error { var comment *issues_model.Comment + var stopwatchFinished bool if err := db.WithTx(ctx, func(ctx context.Context) error { var err error comment, err = issues_model.CloseIssue(ctx, issue, doer) @@ -28,12 +29,17 @@ func CloseIssue(ctx context.Context, issue *issues_model.Issue, doer *user_model return err } - _, err = issues_model.FinishIssueStopwatch(ctx, doer, issue) + stopwatchFinished, err = issues_model.FinishIssueStopwatch(ctx, doer, issue) return err }); err != nil { return err } + // after the tx: publishing inside it would announce a change a rollback undoes + if stopwatchFinished { + notify_service.StopwatchChanged(ctx, doer) + } + notify_service.IssueChangeStatus(ctx, doer, commitID, issue, comment, true) return nil diff --git a/services/notifications/status.go b/services/notifications/status.go new file mode 100644 index 0000000000..a75417ee97 --- /dev/null +++ b/services/notifications/status.go @@ -0,0 +1,60 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +// Package notifications wraps activities_model notification-status mutations +// with the matching real-time push, so route handlers cannot forget either side. +package notifications + +import ( + "context" + + activities_model "gitea.dev/models/activities" + user_model "gitea.dev/models/user" + notify_service "gitea.dev/services/notify" +) + +func SetIssueReadBy(ctx context.Context, issueID, userID int64) error { + changed, err := activities_model.SetIssueReadBy(ctx, issueID, userID) + if err != nil { + return err + } + if changed { + notify_service.NotificationCountChange(ctx, userID) + } + return nil +} + +func SetNotificationStatus(ctx context.Context, notificationID int64, user *user_model.User, status activities_model.NotificationStatus) (*activities_model.Notification, error) { + notif, err := activities_model.SetNotificationStatus(ctx, notificationID, user, status) + if err != nil { + return notif, err + } + notify_service.NotificationCountChange(ctx, user.ID) + return notif, nil +} + +func SetManyNotificationStatuses(ctx context.Context, ns []*activities_model.Notification, user *user_model.User, status activities_model.NotificationStatus) ([]*activities_model.Notification, error) { + out := make([]*activities_model.Notification, 0, len(ns)) + for _, n := range ns { + notif, err := activities_model.SetNotificationStatus(ctx, n.ID, user, status) + if err != nil { + return out, err + } + out = append(out, notif) + } + if len(out) > 0 { + notify_service.NotificationCountChange(ctx, user.ID) + } + return out, nil +} + +func MarkAllRead(ctx context.Context, user *user_model.User) error { + changed, err := activities_model.UpdateNotificationStatuses(ctx, user, activities_model.NotificationStatusUnread, activities_model.NotificationStatusRead) + if err != nil { + return err + } + if changed > 0 { + notify_service.NotificationCountChange(ctx, user.ID) + } + return nil +} diff --git a/services/notify/notifier.go b/services/notify/notifier.go index 513def7252..9f572163a1 100644 --- a/services/notify/notifier.go +++ b/services/notify/notifier.go @@ -82,4 +82,8 @@ type Notifier interface { WorkflowRunStatusUpdate(ctx context.Context, repo *repo_model.Repository, sender *user_model.User, run *actions_model.ActionRun) WorkflowJobStatusUpdate(ctx context.Context, repo *repo_model.Repository, sender *user_model.User, job *actions_model.ActionRunJob, task *actions_model.ActionTask) + + NotificationCountChange(ctx context.Context, userID int64) + + StopwatchChanged(ctx context.Context, user *user_model.User) } diff --git a/services/notify/notify.go b/services/notify/notify.go index b0de0b6a38..b61588a063 100644 --- a/services/notify/notify.go +++ b/services/notify/notify.go @@ -416,3 +416,17 @@ func WorkflowJobStatusUpdate(ctx context.Context, repo *repo_model.Repository, s notifier.WorkflowJobStatusUpdate(ctx, repo, sender, job, task) } } + +// Callers must invoke this after any DB write affecting the user's unread count. +func NotificationCountChange(ctx context.Context, userID int64) { + for _, notifier := range notifiers { + notifier.NotificationCountChange(ctx, userID) + } +} + +// Callers must invoke this after any stopwatch start/stop/cancel so the user's connected tabs refresh. +func StopwatchChanged(ctx context.Context, user *user_model.User) { + for _, notifier := range notifiers { + notifier.StopwatchChanged(ctx, user) + } +} diff --git a/services/notify/null.go b/services/notify/null.go index 2ff1869cb2..71d8a71af3 100644 --- a/services/notify/null.go +++ b/services/notify/null.go @@ -219,3 +219,9 @@ func (*NullNotifier) WorkflowRunStatusUpdate(ctx context.Context, repo *repo_mod func (*NullNotifier) WorkflowJobStatusUpdate(ctx context.Context, repo *repo_model.Repository, sender *user_model.User, job *actions_model.ActionRunJob, task *actions_model.ActionTask) { } + +func (*NullNotifier) NotificationCountChange(_ context.Context, _ int64) { +} + +func (*NullNotifier) StopwatchChanged(_ context.Context, _ *user_model.User) { +} diff --git a/services/pubsub/broker.go b/services/pubsub/broker.go new file mode 100644 index 0000000000..c7a52a5ff1 --- /dev/null +++ b/services/pubsub/broker.go @@ -0,0 +1,39 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +// Package pubsub fans real-time events out to local WebSocket subscribers. +// Backend is chosen at boot: in-process map (single-instance) or Redis +// (multi-process). DefaultBroker is wired by Init from setting.Websocket. +package pubsub + +import "fmt" + +// subChanBuffer is how many messages a subscriber may fall behind before its +// messages are dropped instead of stalling the publisher. +const subChanBuffer = 8 + +type Broker interface { + // Subscribe returns a buffered channel of messages for topic, and a cancel + // func that closes the channel and removes the subscription. cancel is + // idempotent. + Subscribe(topic string) (<-chan []byte, func()) + + // Publish delivers msg to every subscriber of topic. Non-blocking: a slow + // subscriber drops messages rather than stalling the publisher. + Publish(topic string, msg []byte) + + // HasTopicSubscribers is an optimization hint for publishers that would + // otherwise do a DB lookup just to discover nobody is listening. Backends + // that cannot answer cheaply across processes return true to be safe. + HasTopicSubscribers(topic string) bool +} + +// DefaultBroker is replaced by Init from setting.Websocket. It starts as an +// empty memory broker so non-web entry points (e.g. CLI), which skip Init, can +// publish without nil checks — with no subscribers every publish is a no-op. +// Tests construct a broker explicitly (NewMemoryBroker) instead of relying on this. +var DefaultBroker Broker = NewMemoryBroker() + +func UserTopic(userID int64) string { + return fmt.Sprintf("user-%d", userID) +} diff --git a/services/pubsub/broker_test.go b/services/pubsub/broker_test.go new file mode 100644 index 0000000000..6e6ea3d3f2 --- /dev/null +++ b/services/pubsub/broker_test.go @@ -0,0 +1,170 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package pubsub + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newBrokerFunc builds a fresh Broker bound to the test's lifecycle. +type newBrokerFunc func(t *testing.T) Broker + +// testBrokerBasic runs the behavior every Broker backend must share. Each backend +// invokes it from its own *_test.go (like testQueueBasic in modules/queue) so +// memory and redis prove identical semantics against the same scenarios. +// +// recvTimeout absorbs redis's network round-trip (memory delivers synchronously). +func testBrokerBasic(t *testing.T, newBroker newBrokerFunc, recvTimeout time.Duration) { + t.Run("PublishWithoutSubscribers", func(t *testing.T) { + b := newBroker(t) + b.Publish("nobody", []byte("msg")) // must not block or panic + }) + + t.Run("SubscribeReceivesPublished", func(t *testing.T) { + b := newBroker(t) + ch, cancel := b.Subscribe("topic") + defer cancel() + + b.Publish("topic", []byte("hello")) + assert.Equal(t, []byte("hello"), recvWithin(t, ch, recvTimeout)) + }) + + t.Run("FanOutToAllSubscribers", func(t *testing.T) { + b := newBroker(t) + const n = 3 + channels := make([]<-chan []byte, n) + for i := range n { + ch, cancel := b.Subscribe("topic") + defer cancel() + channels[i] = ch + } + + b.Publish("topic", []byte("broadcast")) + for i, ch := range channels { + assert.Equal(t, []byte("broadcast"), recvWithin(t, ch, recvTimeout), "subscriber %d", i) + } + }) + + t.Run("TopicIsolation", func(t *testing.T) { + b := newBroker(t) + chA, cancelA := b.Subscribe("a") + defer cancelA() + chB, cancelB := b.Subscribe("b") + defer cancelB() + + b.Publish("a", []byte("only-a")) + assert.Equal(t, []byte("only-a"), recvWithin(t, chA, recvTimeout)) + assertQuiet(t, chB, 100*time.Millisecond) // topic b must stay silent + }) + + t.Run("CancelStopsDelivery", func(t *testing.T) { + b := newBroker(t) + ch, cancel := b.Subscribe("topic") + + cancel() + _, ok := <-ch + assert.False(t, ok, "channel must be closed after cancel") + + b.Publish("topic", []byte("after-cancel")) // must not panic or block + }) + + t.Run("CancelIsIdempotent", func(t *testing.T) { + b := newBroker(t) + _, cancel := b.Subscribe("topic") + cancel() + assert.NotPanics(t, cancel, "cancel must be safe to call more than once") + }) + + // What each backend reports with no live subscriber differs, so that stays in + // the backend's own test file. + t.Run("HasTopicSubscribers", func(t *testing.T) { + b := newBroker(t) + _, cancel := b.Subscribe("topic") + defer cancel() + assert.True(t, b.HasTopicSubscribers("topic"), "must report subscribers while one is live") + }) + + t.Run("SlowSubscriberDropsWithoutBlocking", func(t *testing.T) { + b := newBroker(t) + _, cancelSlow := b.Subscribe("topic") // never drained, buffer overflows + defer cancelSlow() + fast, cancelFast := b.Subscribe("topic") + defer cancelFast() + + // Drain fast concurrently so it keeps up while slow's buffer fills. + got := make(chan struct{}, 1) + done := make(chan struct{}) + defer close(done) + go func() { + for { + select { + case _, ok := <-fast: + if !ok { + return + } + select { + case got <- struct{}{}: + default: + } + case <-done: + return + } + } + }() + + // Far more than the 8-slot buffer: Publish must never block on slow. + const n = 50 + published := make(chan struct{}) + go func() { + for i := range n { + b.Publish("topic", []byte{byte(i)}) + } + close(published) + }() + select { + case <-published: + case <-time.After(recvTimeout): + t.Fatal("Publish blocked on slow subscriber") + } + + // The fast subscriber still receives while slow is stuck. + select { + case <-got: + case <-time.After(recvTimeout): + t.Fatal("fast subscriber received nothing while slow subscriber stalled") + } + }) +} + +// 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() + select { + case msg, ok := <-ch: + require.True(t, ok, "channel closed before a message arrived") + return msg + case <-time.After(timeout): + t.Fatalf("timed out after %s waiting for a message", timeout) + return nil + } +} + +// assertQuiet fails if any message arrives on ch within d. +func assertQuiet(t *testing.T, ch <-chan []byte, d time.Duration) { + t.Helper() + select { + case msg := <-ch: + t.Fatalf("unexpected message: %s", msg) + case <-time.After(d): + } +} + +func TestUserTopic(t *testing.T) { + assert.Equal(t, "user-42", UserTopic(42)) + assert.Equal(t, "user-0", UserTopic(0)) +} diff --git a/services/pubsub/init.go b/services/pubsub/init.go new file mode 100644 index 0000000000..90105a194f --- /dev/null +++ b/services/pubsub/init.go @@ -0,0 +1,29 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package pubsub + +import ( + "fmt" + + "gitea.dev/modules/setting" +) + +// Init replaces DefaultBroker according to setting.Websocket. Called from +// websocket.Init before the notifier is registered so subscribers wire up to +// the configured backend. +func Init() error { + switch setting.Websocket.PubsubType { + case setting.PubsubTypeMemory: + DefaultBroker = NewMemoryBroker() + case setting.PubsubTypeRedis: + b, err := NewRedisBroker(setting.Websocket.PubsubConnStr) + if err != nil { + return fmt.Errorf("pubsub: init redis backend: %w", err) + } + DefaultBroker = b + default: + return fmt.Errorf("pubsub: unknown PUBSUB_TYPE %q", setting.Websocket.PubsubType) + } + return nil +} diff --git a/services/pubsub/memory.go b/services/pubsub/memory.go new file mode 100644 index 0000000000..dc6f0f47f2 --- /dev/null +++ b/services/pubsub/memory.go @@ -0,0 +1,71 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package pubsub + +import ( + "sync" + + "gitea.dev/modules/log" + "gitea.dev/modules/util" +) + +// MemoryBroker fans out within a single process. Suitable for single-instance +// deployments; multi-process deployments need a backend that crosses processes. +type MemoryBroker struct { + mu sync.RWMutex + subs map[string][]chan []byte +} + +var _ Broker = (*MemoryBroker)(nil) + +func NewMemoryBroker() *MemoryBroker { + return &MemoryBroker{ + subs: make(map[string][]chan []byte), + } +} + +func (b *MemoryBroker) Subscribe(topic string) (<-chan []byte, func()) { + ch := make(chan []byte, subChanBuffer) + + b.mu.Lock() + b.subs[topic] = append(b.subs[topic], ch) + b.mu.Unlock() + + var once sync.Once + cancel := func() { + once.Do(func() { + b.mu.Lock() + defer b.mu.Unlock() + subs := util.SliceRemoveAll(b.subs[topic], ch) + if len(subs) == 0 { + delete(b.subs, topic) + } else { + b.subs[topic] = subs + } + close(ch) + }) + } + return ch, cancel +} + +func (b *MemoryBroker) HasTopicSubscribers(topic string) bool { + b.mu.RLock() + defer b.mu.RUnlock() + return len(b.subs[topic]) > 0 +} + +// Non-blocking: slow subscribers drop. RLock held across fan-out to block +// cancel() from closing a channel between slice read and send. +func (b *MemoryBroker) Publish(topic string, msg []byte) { + b.mu.RLock() + defer b.mu.RUnlock() + + for _, ch := range b.subs[topic] { + select { + case ch <- msg: + default: + log.Trace("pubsub: dropping message on topic %q — subscriber channel full", topic) + } + } +} diff --git a/services/pubsub/memory_test.go b/services/pubsub/memory_test.go new file mode 100644 index 0000000000..18b023674b --- /dev/null +++ b/services/pubsub/memory_test.go @@ -0,0 +1,99 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package pubsub + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// TestMemoryBroker runs the shared broker scenarios against MemoryBroker. +// Memory delivers synchronously and tracks live subscribers exactly. +func TestMemoryBroker(t *testing.T) { + testBrokerBasic(t, func(t *testing.T) Broker { + return NewMemoryBroker() + }, time.Second) +} + +// MemoryBroker tracks live subscribers exactly, so it reports none once the last cancels. +func TestMemoryBroker_HasNoTopicSubscribers(t *testing.T) { + b := NewMemoryBroker() + assert.False(t, b.HasTopicSubscribers("topic")) + _, cancel := b.Subscribe("topic") + cancel() + assert.False(t, b.HasTopicSubscribers("topic")) +} + +// Backend-specific: MemoryBroker prunes empty topics from its internal map so +// idle-user entries don't accumulate. +func TestMemoryBroker_CancelDeletesEmptyTopic(t *testing.T) { + b := NewMemoryBroker() + _, cancel := b.Subscribe("topic") + cancel() + b.mu.RLock() + defer b.mu.RUnlock() + _, present := b.subs["topic"] + assert.False(t, present, "empty topic must be removed from the map so idle-user entries don't accumulate") +} + +// Backend-specific: stresses MemoryBroker's cancel/Publish mutex interlock that +// prevents send-on-closed-channel panics. +func TestMemoryBroker_ConcurrentPublishSubscribeCancel(t *testing.T) { + b := NewMemoryBroker() + + const writers = 4 + const readers = 8 + const duration = 200 * time.Millisecond + + stop := make(chan struct{}) + var wg sync.WaitGroup + + var published atomic.Int64 + for range writers { + wg.Go(func() { + for { + select { + case <-stop: + return + default: + b.Publish("topic", []byte("x")) + published.Add(1) + } + } + }) + } + + for range readers { + wg.Go(func() { + for { + select { + case <-stop: + return + default: + ch, cancel := b.Subscribe("topic") + // Drain a few messages then cancel - this exercises the + // cancel/Publish interlock that prevents send-on-closed. + for range 3 { + select { + case <-ch: + case <-time.After(10 * time.Millisecond): + } + } + cancel() + } + } + }) + } + + time.Sleep(duration) + close(stop) + wg.Wait() + + // Test passes if no panic (send on closed channel) and no deadlock. + assert.Positive(t, published.Load()) +} diff --git a/services/pubsub/redis.go b/services/pubsub/redis.go new file mode 100644 index 0000000000..ce0c4305cd --- /dev/null +++ b/services/pubsub/redis.go @@ -0,0 +1,190 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package pubsub + +import ( + "context" + "sync" + "time" + + "gitea.dev/modules/graceful" + "gitea.dev/modules/log" + "gitea.dev/modules/nosql" + "gitea.dev/modules/util" + + "github.com/redis/go-redis/v9" +) + +const ( + redisPingTimeout = 3 * time.Second + redisPingRetries = 10 + redisPingRetryDelay = time.Second + redisPublishTimeout = 2 * time.Second +) + +// RedisBroker fans out across processes via Redis pub/sub. Each topic is +// backed by a single Redis SUBSCRIBE shared between local subscribers; the +// last local Unsubscribe tears the Redis subscription down. +type RedisBroker struct { + client redis.UniversalClient + + mu sync.RWMutex + topics map[string]*redisTopic +} + +type redisTopic struct { + ps *redis.PubSub + subs []*redisSub + cancel context.CancelFunc +} + +// redisSub pairs a delivery channel with the once that guards its close, so +// either cancel() or readLoop's error-exit path can safely close it. +type redisSub struct { + ch chan []byte + once sync.Once +} + +func (s *redisSub) close() { s.once.Do(func() { close(s.ch) }) } + +var _ Broker = (*RedisBroker)(nil) + +func redisChannelForTopic(s string) string { + return "gitea-ws-topic:" + s +} + +func NewRedisBroker(connStr string) (*RedisBroker, error) { + client := nosql.GetManager().GetRedisClient(connStr) + // context.Background not graceful.ShutdownContext: shutdown ctx may not be initialized at boot. + // Retry to ride out docker-compose start-order races (matches modules/queue). + var err error + for range redisPingRetries { + pingCtx, cancel := context.WithTimeout(context.Background(), redisPingTimeout) + err = client.Ping(pingCtx).Err() + cancel() + if err == nil { + break + } + log.Warn("pubsub redis: not ready, retrying in 1s: %v", err) + time.Sleep(redisPingRetryDelay) + } + if err != nil { + return nil, err + } + return &RedisBroker{ + client: client, + topics: make(map[string]*redisTopic), + }, nil +} + +func (b *RedisBroker) Subscribe(topic string) (<-chan []byte, func()) { + sub := &redisSub{ch: make(chan []byte, subChanBuffer)} + + // Fast path: topic already has a Redis subscription, just attach locally. + b.mu.Lock() + if state, exists := b.topics[topic]; exists { + state.subs = append(state.subs, sub) + b.mu.Unlock() + return sub.ch, b.makeCancel(topic, sub) + } + b.mu.Unlock() + + // Slow path: create the Redis subscription outside the broker mutex so + // 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, a direct + // ps.Receive blocks for its whole timeout instead of returning on the ack. + ctx, cancelCtx := context.WithCancel(graceful.GetManager().ShutdownContext()) + ps := b.client.Subscribe(ctx, redisChannelForTopic(topic)) + b.mu.Lock() + if existing, exists := b.topics[topic]; exists { + // Another goroutine won the create race; merge into theirs and discard ours. + existing.subs = append(existing.subs, sub) + b.mu.Unlock() + cancelCtx() + _ = ps.Close() + return sub.ch, b.makeCancel(topic, sub) + } + b.topics[topic] = &redisTopic{ps: ps, cancel: cancelCtx, subs: []*redisSub{sub}} + b.mu.Unlock() + go b.readLoop(ctx, topic, ps) + return sub.ch, b.makeCancel(topic, sub) +} + +func (b *RedisBroker) makeCancel(topic string, sub *redisSub) func() { + return func() { + b.mu.Lock() + state, ok := b.topics[topic] + if !ok { + b.mu.Unlock() + sub.close() + return + } + state.subs = util.SliceRemoveAll(state.subs, sub) + if len(state.subs) == 0 { + state.cancel() + _ = state.ps.Close() + delete(b.topics, topic) + } + b.mu.Unlock() + sub.close() + } +} + +func (b *RedisBroker) readLoop(ctx context.Context, topic string, ps *redis.PubSub) { + for { + msg, err := ps.ReceiveMessage(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + // Transport blip: tear the topic down so a fresh Subscribe rebuilds it. + // Closing each subscriber's channel wakes the WebSocket handler, which + // will reconnect and re-Subscribe. + b.mu.Lock() + if cur, ok := b.topics[topic]; ok && cur.ps == ps { + for _, s := range cur.subs { + s.close() + } + cur.cancel() + _ = cur.ps.Close() + delete(b.topics, topic) + } + b.mu.Unlock() + log.Trace("pubsub redis: receive on %q: %v", topic, err) + return + } + payload := []byte(msg.Payload) + b.mu.RLock() + state, ok := b.topics[topic] + if !ok { + b.mu.RUnlock() + return + } + for _, s := range state.subs { + select { + case s.ch <- payload: + default: + log.Trace("pubsub redis: dropping message on topic %q — subscriber channel full", topic) + } + } + b.mu.RUnlock() + } +} + +func (b *RedisBroker) Publish(topic string, msg []byte) { + ctx, cancel := context.WithTimeout(graceful.GetManager().HammerContext(), redisPublishTimeout) + defer cancel() + if err := b.client.Publish(ctx, redisChannelForTopic(topic), msg).Err(); err != nil { + log.Error("pubsub redis: publish to %q: %v", topic, err) + } +} + +// HasTopicSubscribers conservatively returns true: cross-process subscriber +// discovery via PUBSUB NUMSUB is per-node and would silently miss subscribers +// in cluster mode. Publishers do the upstream lookup unconditionally. +func (b *RedisBroker) HasTopicSubscribers(topic string) bool { + return true +} diff --git a/services/pubsub/redis_test.go b/services/pubsub/redis_test.go new file mode 100644 index 0000000000..99ff902a30 --- /dev/null +++ b/services/pubsub/redis_test.go @@ -0,0 +1,71 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package pubsub + +import ( + "testing" + "time" + + "gitea.dev/modules/test" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestRedisBroker runs the shared broker scenarios against a real Redis-backed +// RedisBroker, plus the backend-specific ones. Delivery crosses a Redis +// round-trip (hence the wider timeout) and HasTopicSubscribers answers +// conservatively true, so subscriber tracking is not exact. +// One redis-server is shared by all of them; starting one per test dominated +// the package runtime. +func TestRedisBroker(t *testing.T) { + redisConn := test.PrepareTestRedis(t) + + newBroker := func(t *testing.T) Broker { + broker, err := NewRedisBroker(redisConn) + require.NoError(t, err) + return broker + } + testBrokerBasic(t, newBroker, 2*time.Second) + + // RedisBroker cannot see other processes' subscribers, so it answers true even + // with none locally rather than risk dropping a push. + t.Run("HasTopicSubscribersWithoutAny", func(t *testing.T) { + assert.True(t, newBroker(t).HasTopicSubscribers("topic")) + }) + + // RedisBroker tears down its per-topic Redis subscription and internal + // state once the last local subscriber cancels. + t.Run("CancelCleansTopicState", func(t *testing.T) { + b := newBroker(t).(*RedisBroker) + ch, cancel := b.Subscribe("topic") + cancel() + + _, ok := <-ch + assert.False(t, ok, "channel must be closed after cancel") + + b.mu.RLock() + _, present := b.topics["topic"] + b.mu.RUnlock() + assert.False(t, present, "topic state must be removed after last subscriber cancels") + }) + + // Two RedisBroker instances sharing one Redis simulate two Gitea processes - + // a publish on one must reach a subscriber on the other. + t.Run("CrossBroker", func(t *testing.T) { + publisher, subscriber := newBroker(t), newBroker(t) + + ch, cancel := subscriber.Subscribe("topic") + defer cancel() + + publisher.Publish("topic", []byte("cross-process")) + + select { + case msg := <-ch: + assert.Equal(t, []byte("cross-process"), msg) + case <-time.After(2 * time.Second): + t.Fatal("subscriber on second broker did not receive message") + } + }) +} diff --git a/services/uinotification/notify.go b/services/uinotification/notify.go index 1a6caf6354..072f32d0b5 100644 --- a/services/uinotification/notify.go +++ b/services/uinotification/notify.go @@ -51,9 +51,15 @@ func NewNotifier() notify_service.Notifier { } func handler(items ...issueNotificationOpts) []issueNotificationOpts { + ctx := graceful.GetManager().ShutdownContext() for _, opts := range items { - if err := activities_model.CreateOrUpdateIssueNotifications(graceful.GetManager().ShutdownContext(), opts.IssueID, opts.CommentID, opts.NotificationAuthorID, opts.ReceiverID); err != nil { + 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) + continue + } + for _, userID := range notifiedIDs { + notify_service.NotificationCountChange(ctx, userID) } } return nil diff --git a/services/user/user.go b/services/user/user.go index 8c40102fb4..d03809d882 100644 --- a/services/user/user.go +++ b/services/user/user.go @@ -16,7 +16,6 @@ import ( repo_model "gitea.dev/models/repo" system_model "gitea.dev/models/system" user_model "gitea.dev/models/user" - "gitea.dev/modules/eventsource" "gitea.dev/modules/git/gitrepo" "gitea.dev/modules/log" "gitea.dev/modules/setting" @@ -29,6 +28,7 @@ import ( "gitea.dev/services/packages" container_service "gitea.dev/services/packages/container" repo_service "gitea.dev/services/repository" + websocket_service "gitea.dev/services/websocket" ) // RenameUser renames a user @@ -148,9 +148,7 @@ func DeleteUser(ctx context.Context, u *user_model.User, purge bool) error { // Force any logged in sessions to log out // FIXME: We also need to tell the session manager to log them out too. - eventsource.GetManager().SendMessage(u.ID, &eventsource.Event{ - Name: "logout", - }) + websocket_service.PublishLogout(u.ID, "") // Delete all repos belonging to this user // Now this is not within a transaction because there are internal transactions within the DeleteRepository diff --git a/services/websocket/events.go b/services/websocket/events.go new file mode 100644 index 0000000000..563a368c39 --- /dev/null +++ b/services/websocket/events.go @@ -0,0 +1,53 @@ +// 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.MarshalWrite(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_publisher.go b/services/websocket/logout_publisher.go new file mode 100644 index 0000000000..58c0bb3962 --- /dev/null +++ b/services/websocket/logout_publisher.go @@ -0,0 +1,12 @@ +// 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_notifier.go new file mode 100644 index 0000000000..7013dc2b63 --- /dev/null +++ b/services/websocket/notification_notifier.go @@ -0,0 +1,39 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +import ( + "context" + + 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{} + +func (n *wsNotifier) NotificationCountChange(ctx context.Context, userID int64) { + if !pubsub.DefaultBroker.HasTopicSubscribers(pubsub.UserTopic(userID)) { + return + } + 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 + } + publishUserEvent(userID, EventNotificationCount, notificationCountEventData{Count: count}) +} diff --git a/services/websocket/notifier.go b/services/websocket/notifier.go new file mode 100644 index 0000000000..ec09c4f0d2 --- /dev/null +++ b/services/websocket/notifier.go @@ -0,0 +1,18 @@ +// 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_notifier.go new file mode 100644 index 0000000000..5d81cbd139 --- /dev/null +++ b/services/websocket/stopwatch_notifier.go @@ -0,0 +1,37 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package websocket + +import ( + "context" + + "gitea.dev/models/db" + issues_model "gitea.dev/models/issues" + user_model "gitea.dev/models/user" + "gitea.dev/modules/log" + "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 + } + + sws, err := issues_model.GetUserStopwatches(ctx, user.ID, db.ListOptions{}) + if err != nil { + log.Error("websocket: GetUserStopwatches %d: %v", user.ID, err) + return + } + + apiStopWatches, err := convert.ToStopWatches(ctx, user, sws) + if err != nil { + if !issues_model.IsErrIssueNotExist(err) { + log.Error("websocket: ToStopWatches: %v", err) + } + return + } + publishUserEvent(user.ID, EventStopwatches, util.SliceNilAsEmpty(apiStopWatches)) +} diff --git a/templates/base/head_navbar.tmpl b/templates/base/head_navbar.tmpl index 5d2c4b1ca5..348f236fdc 100644 --- a/templates/base/head_navbar.tmpl +++ b/templates/base/head_navbar.tmpl @@ -152,31 +152,30 @@ {{$activeStopwatch := and .PageGlobalData (call .PageGlobalData.GetActiveStopwatch)}} - {{if $activeStopwatch}} -
-
- - {{svg "octicon-issue-opened" 16}} - {{$activeStopwatch.RepoSlug}}#{{$activeStopwatch.IssueIndex}} - -
-
- -
-
- -
-
+ {{/* always rendered so the Tippy popup has content to clone if the icon appears via a real-time push */}} +
+
+ + {{svg "octicon-issue-opened" 16}} + {{if $activeStopwatch}}{{$activeStopwatch.RepoSlug}}#{{$activeStopwatch.IssueIndex}}{{end}} + +
+
+ +
+
+ +
- {{end}} +
{{template "base/head_banner"}} diff --git a/templates/base/head_navbar_icons.tmpl b/templates/base/head_navbar_icons.tmpl index 019e78cb60..b6a2ef1020 100644 --- a/templates/base/head_navbar_icons.tmpl +++ b/templates/base/head_navbar_icons.tmpl @@ -3,14 +3,13 @@ {{if and $data $data.IsSigned}}{{/* data may not exist, for example: rendering 503 page before the PageGlobalData middleware */}} {{- $activeStopwatch := call $data.GetActiveStopwatch -}} {{- $notificationUnreadCount := call $data.GetNotificationUnreadCount -}} - {{if $activeStopwatch}} - + {{/* always rendered so a real-time push can reveal it without a reload */}} +
{{svg "octicon-stopwatch"}}
- {{end}}
{{svg "octicon-bell"}} diff --git a/templates/base/head_script.tmpl b/templates/base/head_script.tmpl index f8e2798f55..7a918d8bdb 100644 --- a/templates/base/head_script.tmpl +++ b/templates/base/head_script.tmpl @@ -16,7 +16,7 @@ If you introduce mistakes in it, Gitea JavaScript code wouldn't run correctly. notificationSettings: {{NotificationSettings}}, {{/*a map provided by NewFuncMap in helper.go*/}} enableTimeTracking: {{EnableTimetracking}}, mermaidMaxSourceCharacters: {{MermaidMaxSourceCharacters}}, - sharedWorkerUri: '{{AssetURI "web_src/js/eventsource.sharedworker.ts"}}', + sharedWorkerUri: '{{AssetURI "web_src/js/user-events.sharedworker.ts"}}', {{/* this global i18n object should only contain general texts. for specialized texts, it should be provided inside the related modules by: (1) API response (2) HTML data-attribute (3) PageData */}} i18n: { error_occurred: {{ctx.Locale.Tr "error.occurred"}}, diff --git a/templates/repo/editor/upload.tmpl b/templates/repo/editor/upload.tmpl index 0c81802a48..ae062565d3 100644 --- a/templates/repo/editor/upload.tmpl +++ b/templates/repo/editor/upload.tmpl @@ -6,7 +6,7 @@
{{template "repo/view_file_tree" .}}
-
+ {{template "repo/editor/common_top" .}}
{{template "repo/view_file_tree_toggle_button" .}} diff --git a/tests/e2e/events.test.ts b/tests/e2e/events.test.ts index 2276e7e618..2df6ec9f8d 100644 --- a/tests/e2e/events.test.ts +++ b/tests/e2e/events.test.ts @@ -1,16 +1,17 @@ import {test, expect} from '@playwright/test'; -import {loginUser, baseUrl, apiUserHeaders, apiCreateUser, apiCreateRepo, apiCreateIssue, apiStartStopwatch, timeoutFactor, randomString} from './utils.ts'; +import {loginUser, baseUrl, apiUserHeaders, apiCreateUser, apiCreateRepo, apiCreateIssue, apiStartStopwatch, apiCancelStopwatch, apiCloseIssue, randomString} from './utils.ts'; -// These tests rely on a short EVENT_SOURCE_UPDATE_TIME in the e2e server config. +// The /-/ws WebSocket pipeline is push-only: every event is fired by the server +// immediately on the DB write. These tests exercise that each event type +// (notification-count, stopwatches, logout) reaches a connected tab. test.describe('events', () => { - test('notification count', async ({page, request}) => { + test('notification count increases on new notification', async ({page, request}) => { const owner = `ev-notif-owner-${randomString(8)}`; const commenter = `ev-notif-commenter-${randomString(8)}`; const repoName = `ev-notif-${randomString(8)}`; await Promise.all([apiCreateUser(request, owner), apiCreateUser(request, commenter)]); - // Create repo and login in parallel — repo is needed for the issue, login for the event stream await Promise.all([ apiCreateRepo(request, {name: repoName, autoInit: false, headers: apiUserHeaders(owner)}), loginUser(page, owner), @@ -19,33 +20,90 @@ test.describe('events', () => { const badge = page.locator('a.not-mobile .notification_count'); await expect(badge).toBeHidden(); - // Create issue as another user — this generates a notification delivered via server push - await apiCreateIssue(request, {owner, repo: repoName, title: 'events notification test', headers: apiUserHeaders(commenter)}); + await expect(page.locator('html[data-user-events-connected]')).toBeAttached(); - // Wait for the notification badge to appear via server event - await expect(badge).toBeVisible({timeout: 15000 * timeoutFactor}); + await apiCreateIssue(request, {owner, repo: repoName, title: 'events-notif', headers: apiUserHeaders(commenter)}); + await expect(badge).toBeVisible(); }); - test('stopwatch', async ({page, request}) => { - const name = `ev-sw-${randomString(8)}`; + test('stopwatch appears and hides via real-time push', async ({page, request}) => { + const name = `ev-sw-push-${randomString(8)}`; const headers = apiUserHeaders(name); await apiCreateUser(request, name); + await Promise.all([ + loginUser(page, name), + (async () => { + await apiCreateRepo(request, {name, headers}); + await apiCreateIssue(request, {owner: name, repo: name, title: 'events stopwatch push test', headers}); + })(), + ]); + // Page loads before the stopwatch starts — the icon is hidden in the rendered HTML + await page.goto('/'); + const stopwatch = page.locator('.active-stopwatch.not-mobile'); + // Element must exist in the DOM (just hidden); otherwise the push has nothing to reveal. + await expect(stopwatch).toHaveCount(1); + await expect(stopwatch).toBeHidden(); - // Login in parallel with repo+issue+stopwatch setup (all independent after user exists) + await expect(page.locator('html[data-user-events-connected]')).toBeAttached(); + + // Drive both directions from outside this tab; each push must reach it + await apiStartStopwatch(request, name, name, 1, {headers}); + await expect(stopwatch).toBeVisible(); + + await apiCancelStopwatch(request, name, name, 1, {headers}); + await expect(stopwatch).toBeHidden(); + }); + + // Closing an issue stops the timer away from any stopwatch route handler. + test('stopwatch renders when already active and hides when the issue is closed', async ({page, request}) => { + const name = `ev-sw-close-${randomString(8)}`; + const headers = apiUserHeaders(name); + + await apiCreateUser(request, name); await Promise.all([ loginUser(page, name), (async () => { await apiCreateRepo(request, {name, autoInit: false, headers}); - await apiCreateIssue(request, {owner: name, repo: name, title: 'events stopwatch test', headers}); + await apiCreateIssue(request, {owner: name, repo: name, title: 'events stopwatch close test', headers}); await apiStartStopwatch(request, name, name, 1, {headers}); })(), ]); await page.goto('/'); - - // Verify stopwatch is visible and links to the correct issue const stopwatch = page.locator('.active-stopwatch.not-mobile'); await expect(stopwatch).toBeVisible(); + await expect(page.locator('html[data-user-events-connected]')).toBeAttached(); + + await apiCloseIssue(request, name, name, 1, {headers}); + await expect(stopwatch).toBeHidden(); + }); + + // Repro for https://github.com/go-gitea/gitea/pull/36965#issuecomment-4321282667: + // clicking the sidebar "Start timer" button reportedly produced a blank page. + // Drives the actual UI button (not the API) so the link-action → JSONRedirect("") + // → fetchActionDoRedirect("") path is exercised end-to-end. + test('sidebar start timer button starts stopwatch without blanking the page', async ({page, request}) => { + const name = `ev-sw-ui-${randomString(8)}`; + const headers = apiUserHeaders(name); + + await apiCreateUser(request, name); + await Promise.all([ + loginUser(page, name), + (async () => { + await apiCreateRepo(request, {name, headers}); + await apiCreateIssue(request, {owner: name, repo: name, title: 'sidebar start timer test', headers}); + })(), + ]); + await page.goto(`/${name}/${name}/issues/1`); + + await page.getByRole('button', {name: 'Start timer'}).click(); + + // After the click the page reloads; the sidebar should now show the Stop/Discard + // controls and the navbar stopwatch icon should appear. If the page blanked, + // neither of these would be present. + await expect(page.getByRole('button', {name: 'Stop timer'})).toBeVisible(); + await expect(page.getByRole('button', {name: 'Discard timer'})).toBeVisible(); + await expect(page.locator('.active-stopwatch.not-mobile')).toBeVisible(); }); test('logout propagation', async ({browser, request}) => { @@ -65,9 +123,8 @@ test.describe('events', () => { // Verify page2 is logged in await expect(page2.getByRole('link', {name: 'Sign In'})).toBeHidden(); - - // Wait until the server has registered page2's event stream, otherwise the logout - // event can race the connection and be silently dropped. + // Wait until page2's event stream is connected, otherwise the logout event + // can race the connection and be silently dropped. await expect(page2.locator('html[data-user-events-connected]')).toBeAttached(); // Logout from page1 — this sends a logout event to all tabs diff --git a/tests/e2e/utils.ts b/tests/e2e/utils.ts index 5b19793f37..18fe549f2b 100644 --- a/tests/e2e/utils.ts +++ b/tests/e2e/utils.ts @@ -67,6 +67,19 @@ export async function apiStartStopwatch(requestContext: APIRequestContext, owner }), 'apiStartStopwatch'); } +export async function apiCancelStopwatch(requestContext: APIRequestContext, owner: string, repo: string, issueIndex: number, {headers}: {headers?: Record} = {}) { + await apiRetry(() => requestContext.delete(`${baseUrl()}/api/v1/repos/${owner}/${repo}/issues/${issueIndex}/stopwatch/delete`, { + headers: headers || apiHeaders(), + }), 'apiCancelStopwatch'); +} + +export async function apiCloseIssue(requestContext: APIRequestContext, owner: string, repo: string, issueIndex: number, {headers}: {headers?: Record} = {}) { + await apiRetry(() => requestContext.patch(`${baseUrl()}/api/v1/repos/${owner}/${repo}/issues/${issueIndex}`, { + headers: headers || apiHeaders(), + data: {state: 'closed'}, + }), 'apiCloseIssue'); +} + export async function apiCreateFile(requestContext: APIRequestContext, owner: string, repo: string, filepath: string, content: string, {branch, newBranch, message}: {branch?: string; newBranch?: string; message?: string} = {}) { await apiRetry(() => requestContext.post(`${baseUrl()}/api/v1/repos/${owner}/${repo}/contents/${filepath}`, { headers: apiHeaders(), diff --git a/tests/integration/editor_test.go b/tests/integration/editor_test.go index 2a368891d9..54af71295c 100644 --- a/tests/integration/editor_test.go +++ b/tests/integration/editor_test.go @@ -420,7 +420,7 @@ func testForkToEditFile(t *testing.T, session *TestSession, user, owner, repo, b resp := session.MakeRequest(t, req, http.StatusOK) htmlDoc := NewHTMLParser(t, resp.Body) - uploadForm := htmlDoc.doc.Find(".form-fetch-action") + uploadForm := htmlDoc.doc.Find(".repo-file-upload.form-fetch-action") formAction := uploadForm.AttrOr("action", "") assert.Equal(t, fmt.Sprintf("/%s/%s-1/_upload/%s/%s?from_base_branch=%s&foo=bar", user, repo, branch, filePath, branch), formAction) uploadLink := uploadForm.Find(".dropzone").AttrOr("data-link-url", "") diff --git a/tests/integration/eventsource_test.go b/tests/integration/eventsource_test.go deleted file mode 100644 index 5800afb1dc..0000000000 --- a/tests/integration/eventsource_test.go +++ /dev/null @@ -1,84 +0,0 @@ -// Copyright 2020 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package integration - -import ( - "fmt" - "net/http" - "testing" - "time" - - activities_model "gitea.dev/models/activities" - auth_model "gitea.dev/models/auth" - repo_model "gitea.dev/models/repo" - "gitea.dev/models/unittest" - user_model "gitea.dev/models/user" - "gitea.dev/modules/eventsource" - api "gitea.dev/modules/structs" - "gitea.dev/tests" - - "github.com/stretchr/testify/assert" -) - -func TestEventSourceManagerRun(t *testing.T) { - defer tests.PrepareTestEnv(t)() - manager := eventsource.GetManager() - - eventChan := manager.Register(2) - defer func() { - manager.Unregister(2, eventChan) - // ensure the eventChan is closed - for { - _, ok := <-eventChan - if !ok { - break - } - } - }() - expectNotificationCountEvent := func(count int64) func() bool { - return func() bool { - select { - case event, ok := <-eventChan: - if !ok { - return false - } - data, ok := event.Data.(activities_model.UserIDCount) - if !ok { - return false - } - return event.Name == "notification-count" && data.Count == count - default: - return false - } - } - } - - user2 := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2}) - repo1 := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 1}) - thread5 := unittest.AssertExistsAndLoadBean(t, &activities_model.Notification{ID: 5}) - assert.NoError(t, thread5.LoadAttributes(t.Context())) - session := loginUser(t, user2.Name) - token := getTokenForLoggedInUser(t, session, auth_model.AccessTokenScopeWriteNotification, auth_model.AccessTokenScopeWriteRepository) - - // -- mark notifications as read -- - req := NewRequest(t, "GET", "/api/v1/notifications?status-types=unread"). - AddTokenAuth(token) - resp := session.MakeRequest(t, req, http.StatusOK) - - apiNL := DecodeJSON(t, resp, []api.NotificationThread{}) - assert.Len(t, apiNL, 2) - - lastReadAt := "2000-01-01T00%3A50%3A01%2B00%3A00" // 946687801 <- only Notification 4 is in this filter ... - req = NewRequest(t, "PUT", fmt.Sprintf("/api/v1/repos/%s/%s/notifications?last_read_at=%s", user2.Name, repo1.Name, lastReadAt)). - AddTokenAuth(token) - session.MakeRequest(t, req, http.StatusResetContent) - - req = NewRequest(t, "GET", "/api/v1/notifications?status-types=unread"). - AddTokenAuth(token) - resp = session.MakeRequest(t, req, http.StatusOK) - apiNL = DecodeJSON(t, resp, []api.NotificationThread{}) - assert.Len(t, apiNL, 1) - - assert.Eventually(t, expectNotificationCountEvent(1), 30*time.Second, 1*time.Second) -} diff --git a/tests/integration/signin_test.go b/tests/integration/signin_test.go index ab1167ce2c..d703625ea5 100644 --- a/tests/integration/signin_test.go +++ b/tests/integration/signin_test.go @@ -177,8 +177,8 @@ func TestRequireSignInView(t *testing.T) { require.False(t, setting.Service.BlockAnonymousAccessExpensive) req := NewRequest(t, "GET", "/user2/repo1/src/branch/master") MakeRequest(t, req, http.StatusOK) - req = NewRequest(t, "GET", "/user/events") - MakeRequest(t, req, http.StatusOK) + req = NewRequest(t, "GET", "/-/ws") + MakeRequest(t, req, http.StatusUpgradeRequired) }) t.Run("RequireSignInView", func(t *testing.T) { defer test.MockVariableValue(&setting.Service.RequireSignInViewStrict, true)() @@ -194,7 +194,7 @@ func TestRequireSignInView(t *testing.T) { req := NewRequest(t, "GET", "/user2/repo1") MakeRequest(t, req, http.StatusOK) - req = NewRequest(t, "GET", "/user/events") + req = NewRequest(t, "GET", "/-/ws") MakeRequest(t, req, http.StatusSeeOther) req = NewRequest(t, "GET", "/user2/repo1/src/branch/master") diff --git a/vite.config.ts b/vite.config.ts index 8ff0769afd..e128ee0791 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -182,7 +182,7 @@ function reducedSourcemapPlugin(): Plugin { 'js/swagger.', 'js/external-render-frontend.', 'js/external-render-helper.', - 'js/eventsource.sharedworker.', + 'js/user-events.sharedworker.', ]; return { name: 'reduced-sourcemap', @@ -270,7 +270,7 @@ export default defineConfig(commonViteOpts({ index: join(import.meta.dirname, 'web_src/js/index.ts'), swagger: join(import.meta.dirname, 'web_src/js/swagger.ts'), 'external-render-frontend': join(import.meta.dirname, 'web_src/js/external-render-frontend.ts'), - 'eventsource.sharedworker': join(import.meta.dirname, 'web_src/js/eventsource.sharedworker.ts'), + 'user-events.sharedworker': join(import.meta.dirname, 'web_src/js/user-events.sharedworker.ts'), devtest: join(import.meta.dirname, 'web_src/css/devtest.css'), ...themes, }, diff --git a/web_src/js/eventsource.sharedworker.ts b/web_src/js/eventsource.sharedworker.ts deleted file mode 100644 index 64da37c703..0000000000 --- a/web_src/js/eventsource.sharedworker.ts +++ /dev/null @@ -1,150 +0,0 @@ -class Source { - url: string; - eventSource: EventSource | null; - listening: Record; - clients: Array; - - constructor(url: string) { - this.url = url; - this.eventSource = new EventSource(url); - this.listening = {}; - this.clients = []; - this.listen('open'); - this.listen('close'); - this.listen('logout'); - this.listen('notification-count'); - this.listen('stopwatches'); - this.listen('error'); - } - - register(port: MessagePort) { - if (this.clients.includes(port)) return; - - this.clients.push(port); - - port.postMessage({ - type: 'status', - message: `registered to ${this.url}`, - }); - - // replay the "open" event to ports attaching to an already-open source - if (this.eventSource?.readyState === EventSource.OPEN) { - port.postMessage({type: 'open'}); - } - } - - deregister(port: MessagePort) { - const portIdx = this.clients.indexOf(port); - if (portIdx < 0) { - return this.clients.length; - } - this.clients.splice(portIdx, 1); - return this.clients.length; - } - - close() { - if (!this.eventSource) return; - - this.eventSource.close(); - this.eventSource = null; - } - - listen(eventType: string) { - if (this.listening[eventType]) return; - this.listening[eventType] = true; - this.eventSource?.addEventListener(eventType, (event) => { - this.notifyClients({ - type: eventType, - data: event.data, - }); - }); - } - - notifyClients(event: {type: string, data: any}) { - for (const client of this.clients) { - client.postMessage(event); - } - } - - status(port: MessagePort) { - port.postMessage({ - type: 'status', - message: `url: ${this.url} readyState: ${this.eventSource?.readyState}`, - }); - } -} - -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 (!self.EventSource) { - // some browsers (like PaleMoon, Firefox<53) don't support EventSource in SharedWorkerGlobalScope. - // this event handler needs EventSource when doing "new Source(url)", so just post a message back to the caller, - // in case the caller would like to use a fallback method to do its work. - port.postMessage({type: 'no-event-source'}); - return; - } - if (event.data.type === 'start') { - const url = event.data.url; - let source = sourcesByUrl.get(url); - if (source) { - // we have a Source registered to this url - source.register(port); - sourcesByPort.set(port, source); - return; - } - source = sourcesByPort.get(port); - if (source) { - if (source.eventSource && source.url === url) return; - - // How this has happened I don't understand... - // deregister from that source - const count = source.deregister(port); - // Clean-up - if (count === 0) { - source.close(); - sourcesByUrl.set(source.url, null); - } - } - // Create a new Source - source = new Source(url); - source.register(port); - sourcesByUrl.set(url, source); - sourcesByPort.set(port, source); - } else if (event.data.type === 'listen') { - const source = sourcesByPort.get(port)!; - source.listen(event.data.eventType); - } 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; - } - source.status(port); - } else { - // just send it back - port.postMessage({ - type: 'error', - message: `received but don't know how to handle: ${event.data}`, - }); - } - }); - port.start(); - } -}); diff --git a/web_src/js/features/notification.ts b/web_src/js/features/notification.ts index da04e247c3..f18567e62b 100644 --- a/web_src/js/features/notification.ts +++ b/web_src/js/features/notification.ts @@ -1,51 +1,38 @@ import {GET} from '../modules/fetch.ts'; import {toggleElem, createElementFromHTML} from '../utils/dom.ts'; -import {UserEventsSharedWorker} from '../modules/worker.ts'; +import {onUserEvent} from '../modules/worker.ts'; const {appSubUrl, notificationSettings} = window.config; let notificationSequenceNumber = 0; -async function receiveUpdateCount(event: MessageEvent<{type: string, data: string}>) { - 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 = String(data.Count); - } - await updateNotificationTable(); - } catch (error) { - console.error(error, event); +async function receiveUpdateCount(count: number) { + toggleElem('.notification_count', count !== 0); + for (const el of document.querySelectorAll('.notification_count')) { + el.textContent = String(count); } + await updateNotificationTable(); } 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 UserEventsSharedWorker('notification-worker'); - worker.addMessageEventListener((event: MessageEvent) => { - if (event.data.type === 'no-event-source') { - if (!usingPeriodicPoller) startPeriodicPoller(notificationSettings.MinTimeout); - } else if (event.data.type === 'notification-count') { - receiveUpdateCount(event); // no await - } - }); - worker.startPort(); - return; - } - - startPeriodicPoller(notificationSettings.MinTimeout); + 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-unavailable', () => { + if (pollerStarted) return; + pollerStarted = true; + startPeriodicPoller(notificationSettings.MinTimeout); + }); } function getCurrentCount() { diff --git a/web_src/js/features/stopwatch.ts b/web_src/js/features/stopwatch.ts index 6fa8fbbdf3..63dabbd333 100644 --- a/web_src/js/features/stopwatch.ts +++ b/web_src/js/features/stopwatch.ts @@ -1,7 +1,8 @@ import {createTippy} from '../modules/tippy.ts'; import {GET} from '../modules/fetch.ts'; import {hideElem, queryElems, showElem} from '../utils/dom.ts'; -import {UserEventsSharedWorker} from '../modules/worker.ts'; +import {onUserEvent} from '../modules/worker.ts'; +import type {StopwatchData} from '../types.ts'; const {appSubUrl, notificationSettings, enableTimeTracking} = window.config; @@ -17,15 +18,13 @@ export function initStopwatch() { return; } - // global stop watch (in the head_navbar), it should always work in any case either the EventSource or the PeriodicPoller is used. + // Init the icon + popup even when no stopwatch is active so a real-time push has a target to toggle. const seconds = stopwatchEls[0]?.getAttribute('data-seconds'); if (seconds) { updateStopwatchTime(parseInt(seconds)); } for (const stopwatchEl of stopwatchEls) { - stopwatchEl.removeAttribute('href'); // intended for noscript mode only - createTippy(stopwatchEl, { content: stopwatchPopup.cloneNode(true) as Element, placement: 'bottom-end', @@ -34,33 +33,28 @@ export function initStopwatch() { interactive: true, hideOnClick: true, theme: 'default', + onShow(instance) { + // Re-clone on every open so the popup reflects the latest stopwatch state, + // including the case where the icon became visible via a real-time push. + instance.setContent(stopwatchPopup.cloneNode(true) as Element); + }, }); } - let usingPeriodicPoller = false; const startPeriodicPoller = (timeout: number) => { if (timeout <= 0 || !Number.isFinite(timeout)) return; - usingPeriodicPoller = true; setTimeout(() => updateStopwatchWithCallback(startPeriodicPoller, timeout), timeout); }; - // if the browser supports EventSource and SharedWorker, use it instead of the periodic poller - if (notificationSettings.EventSourceUpdateTime > 0 && window.EventSource && window.SharedWorker) { - // Try to connect to the event source via the shared worker first - const worker = new UserEventsSharedWorker('stopwatch-worker'); - worker.addMessageEventListener((event) => { - 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 === 'stopwatches') { - updateStopwatchData(JSON.parse(event.data.data)); - } - }); - worker.startPort(); - return; - } - - startPeriodicPoller(notificationSettings.MinTimeout); + 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-unavailable', () => { + if (pollerStarted) return; + pollerStarted = true; + startPeriodicPoller(notificationSettings.MinTimeout); + }); } async function updateStopwatchWithCallback(callback: (timeout: number) => void, timeout: number) { @@ -76,16 +70,22 @@ async function updateStopwatchWithCallback(callback: (timeout: number) => void, } async function updateStopwatch() { - const response = await GET(`${appSubUrl}/user/stopwatches`); - if (!response.ok) { - console.error('Failed to fetch stopwatch data'); + try { + const response = await GET(`${appSubUrl}/user/stopwatches`); + if (!response.ok) { + console.error('Failed to fetch stopwatch data'); + return false; + } + const data = await response.json(); + return updateStopwatchData(data); + } catch (error) { + console.error(error); return false; } - const data = await response.json(); - return updateStopwatchData(data); } -function updateStopwatchData(data: any) { +function updateStopwatchData(data: Array) { + if (!data) return; const watch = data[0]; const btnEls = document.querySelectorAll('.active-stopwatch'); if (!watch) { @@ -93,6 +93,7 @@ function updateStopwatchData(data: any) { } else { const {repo_owner_name, repo_name, issue_index, seconds} = watch; const issueUrl = `${appSubUrl}/${repo_owner_name}/${repo_name}/issues/${issue_index}`; + for (const btnEl of btnEls) btnEl.setAttribute('href', issueUrl); document.querySelector('.stopwatch-link')?.setAttribute('href', issueUrl); document.querySelector('.stopwatch-commit')?.setAttribute('action', `${issueUrl}/times/stopwatch/stop`); document.querySelector('.stopwatch-cancel')?.setAttribute('action', `${issueUrl}/times/stopwatch/cancel`); diff --git a/web_src/js/globals.d.ts b/web_src/js/globals.d.ts index bb13e1e79f..1fb7165830 100644 --- a/web_src/js/globals.d.ts +++ b/web_src/js/globals.d.ts @@ -44,7 +44,6 @@ interface Window { MinTimeout: number, TimeoutStep: number, MaxTimeout: number, - EventSourceUpdateTime: number, }, enableTimeTracking: boolean, mermaidMaxSourceCharacters: number, diff --git a/web_src/js/modules/worker.test.ts b/web_src/js/modules/worker.test.ts new file mode 100644 index 0000000000..15b5efb5ea --- /dev/null +++ b/web_src/js/modules/worker.test.ts @@ -0,0 +1,69 @@ +import type {UserEventMessage, WorkerInboundMessage} from '../types.ts'; + +// Minimal SharedWorker/MessagePort doubles: worker.ts wires user events onto +// `sharedWorker.port`, so we capture the port to feed it messages and assert dispatch behavior. +type PortListener = (ev: {data: WorkerInboundMessage}) => void; + +class MockMessagePort { + listeners: Record = {}; + posted: unknown[] = []; + addEventListener(type: string, cb: PortListener) { + (this.listeners[type] ||= []).push(cb); + } + removeEventListener() {} + postMessage(msg: unknown) { this.posted.push(msg) } + start() {} + close() {} + // Simulate the underlying worker delivering a message to the page. + deliver(msg: UserEventMessage) { + for (const cb of this.listeners['message'] || []) cb({data: {msgType: 'user-event', msgData: msg}}); + } +} + +let lastWorker: MockSharedWorker; + +class MockSharedWorker { + port = new MockMessagePort(); + // eslint-disable-next-line unicorn/no-this-assignment + constructor() { lastWorker = this } + addEventListener() {} +} + +// worker.ts caches module-scope state (subscribers, initialized), so re-import +// a fresh module per test after stubbing the globals it reads on init. +async function freshWorker() { + vi.resetModules(); + vi.stubGlobal('WebSocket', class {}); + vi.stubGlobal('SharedWorker', MockSharedWorker); + return await import('./worker.ts'); +} + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +// sequential: freshWorker resets the module registry and stubs globals, which is unsafe +// to interleave with the other test under the repo's `sequence.concurrent` vitest config. +// Every push follows a real DB write, so a repeated value still means the +// underlying list changed (e.g. a new comment on an already-unread issue). +test('dispatches every push, including repeated values', {concurrent: false}, async () => { + const {onUserEvent} = await freshWorker(); + const received: number[] = []; + onUserEvent('notification-count', (msg) => { received.push(msg.eventData.count) }); + + lastWorker.port.deliver({eventType: 'notification-count', eventData: {count: 1}}); + lastWorker.port.deliver({eventType: 'notification-count', eventData: {count: 1}}); + + expect(received).toEqual([1, 1]); +}); + +test('worker-connected flags the page and reaches its subscribers', {concurrent: false}, async () => { + const {onUserEvent} = await freshWorker(); + let connects = 0; + onUserEvent('worker-connected', () => { connects++ }); + + lastWorker.port.deliver({eventType: 'worker-connected'}); + + expect(connects).toBe(1); + expect(document.documentElement.getAttribute('data-user-events-connected')).toBe('true'); +}); diff --git a/web_src/js/modules/worker.ts b/web_src/js/modules/worker.ts index 9d97e938ef..e10382782c 100644 --- a/web_src/js/modules/worker.ts +++ b/web_src/js/modules/worker.ts @@ -1,68 +1,130 @@ +import type { + SharedWorkerControlMessage, + UserEventMessage, + UserEventType, + WorkerEventMessage, + WorkerInboundMessage, +} from '../types.ts'; + const {appSubUrl, sharedWorkerUri} = window.config; -export class UserEventsSharedWorker { - sharedWorker: SharedWorker; +type EventOf = Extract; +type Subscriber = (msg: EventOf) => void; +const subscribers = new Map>(); +let fallbackSignalled = false; +let sharedWorker: SharedWorker | null = null; - // options can be either a string (the debug name of the worker) or an object of type WorkerOptions - constructor(options?: string | WorkerOptions) { - const worker = new SharedWorker(sharedWorkerUri, options); - this.sharedWorker = 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('error', (e) => { - console.error('worker port error', e); - }); - window.addEventListener('beforeunload', () => { - // FIXME: this logic is not quite right. - // "beforeunload" can be canceled by some actions like "are-you-sure" and the navigation can be cancelled. - // In this case: the worker port is incorrectly closed while the page is still there. - worker.port.postMessage({type: 'close'}); - worker.port.close(); - }); +function dispatch(msg: UserEventMessage) { + if (msg.eventType === 'worker-connected') { + document.documentElement.setAttribute('data-user-events-connected', 'true'); } + const set = subscribers.get(msg.eventType); + if (!set) return; + for (const cb of set) cb(msg); +} - addMessageEventListener(listener: (event: MessageEvent) => void) { - this.sharedWorker.port.addEventListener('message', (event: MessageEvent) => { - if (!event.data || !event.data.type) { - console.error('unknown worker message event', event); +function signalFallback() { + if (fallbackSignalled) return; + fallbackSignalled = true; + dispatch({eventType: 'worker-unavailable'}); +} + +function init() { + try { + sharedWorker = new SharedWorker(sharedWorkerUri, {type: 'module', name: 'user-events'}); + } catch (err) { + console.warn('SharedWorker unavailable, falling back to periodic polling', err); + queueMicrotask(signalFallback); + return; + } + // Browsers without module-SharedWorker support fail at parse time, before the WebSocket opens. + sharedWorker.addEventListener('error', (event) => { + console.error('worker error', event); + signalFallback(); + }); + sharedWorker.port.addEventListener('messageerror', () => { + console.error('unable to deserialize message'); + }); + sharedWorker.port.addEventListener('error', (e) => { + console.error('worker port error', e); + }); + + const postSharedWorkerControlMessage = (sw: SharedWorker, msg: SharedWorkerControlMessage) => { + sw.port.postMessage(msg); + }; + + const handleWorkerEvent = (msg: WorkerEventMessage) => { + if (msg.workerEvent === 'error') { + console.error('worker port event error', msg); + } else if (msg.workerEvent === 'close') { + postSharedWorkerControlMessage(sharedWorker!, {type: 'close'}); + sharedWorker!.port.close(); + } else { + console.error('unknown worker port event', msg); + } + }; + + const handleLogout = () => { + postSharedWorkerControlMessage(sharedWorker!, {type: 'close'}); + sharedWorker!.port.close(); + // slightly delay our "logout" for a short while, in case there are other logout requests in-flight. + // * if the logout is triggered by a page redirection (e.g.: user clicks "/user/logout") + // * "beforeunload" event is triggered, this code path won't execute + // * if the logout is triggered by a fetch call + // * "beforeunload" event is not triggered until JS does the redirection. + // * in this case, the logout fetch call already completes and has sent the "logout" message to the worker + // * there can be a data-race between the fetch call's redirection and the "logout" message from the worker + // * the fetch call's logout redirection should always win over the worker message, because it might have a custom location + setTimeout(() => { window.location.assign(`${appSubUrl}/`) }, 1000); + }; + + sharedWorker.port.addEventListener('message', (event: MessageEvent) => { + const msg = event.data; + if (msg?.msgType === 'worker-event') { + handleWorkerEvent(msg.msgData); + } else if (msg?.msgType === 'user-event') { + if (msg.msgData?.eventType === 'logout') { + handleLogout(); return; } + dispatch(msg.msgData); + } else { + console.error('unknown inbound message', msg); + } + }); - 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; - this.sharedWorker.port.postMessage({type: 'close'}); - this.sharedWorker.port.close(); - // slightly delay our "logout" for a short while, in case there are other logout requests in-flight. - // * if the logout is triggered by a page redirection (e.g.: user clicks "/user/logout") - // * "beforeunload" event is triggered, this code path won't execute - // * if the logout is triggered by a fetch call - // * "beforeunload" event is not triggered until JS does the redirection. - // * in this case, the logout fetch call already completes and has sent the "logout" message to the worker - // * there can be a data-race between the fetch call's redirection and the "logout" message from the worker - // * the fetch call's logout redirection should always win over the worker message, because it might have a custom location - setTimeout(() => { window.location.assign(`${appSubUrl}/`) }, 1000); - } else if (event.data.type === 'close') { - this.sharedWorker.port.postMessage({type: 'close'}); - this.sharedWorker.port.close(); - } else if (event.data.type === 'open') { - // e2e tests wait for this attribute to know events cannot be missed anymore - document.documentElement.setAttribute('data-user-events-connected', 'true'); - } - listener(event); - }); - } - - startPort() { - this.sharedWorker.port.start(); - } + sharedWorker.port.start(); + const wsProtocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + postSharedWorkerControlMessage(sharedWorker, { + type: 'start', + url: `${wsProtocol}//${window.location.host}${appSubUrl}/-/ws`, + showDebugLog: !window.config.runModeIsProd, + }); + window.addEventListener('beforeunload', () => { + // FIXME: this logic is not quite right. + // "beforeunload" can be canceled by some actions like "are-you-sure" and the navigation can be cancelled. + // In this case: the worker port is incorrectly closed while the page is still there. + postSharedWorkerControlMessage(sharedWorker!, {type: 'close'}); + sharedWorker!.port.close(); + }); +} + +let initialized = false; +export function onUserEvent(type: T, cb: Subscriber): () => void { + if (!initialized) { + initialized = true; + if (window.WebSocket && window.SharedWorker) { + init(); + } else { + queueMicrotask(signalFallback); + } + } + let set = subscribers.get(type); + if (!set) { + set = new Set(); + subscribers.set(type, set); + } + const wrapped: Subscriber = (msg) => cb(msg as EventOf); + set.add(wrapped); + return () => { set.delete(wrapped) }; } diff --git a/web_src/js/types.ts b/web_src/js/types.ts index 1c4be74202..8c2ad5f7b2 100644 --- a/web_src/js/types.ts +++ b/web_src/js/types.ts @@ -63,3 +63,38 @@ export type FomanticInitFunction = { export type GitRefType = 'branch' | 'tag'; export type Promisable = T | Promise; // stricter than type-fest which uses PromiseLike + +export type StopwatchData = { + repo_owner_name: string, + repo_name: string, + issue_index: number, + seconds: number, +}; + +// keep in sync with services/websocket/events.go +export type ServerUserEventMessage = + {eventType: 'notification-count', eventData: {count: number}} | + {eventType: 'stopwatches', eventData: Array} | + {eventType: 'logout'}; + +export const serverUserEventTypes = ['notification-count', 'stopwatches', 'logout'] as const satisfies ReadonlyArray; + +export type UserEventMessage = ServerUserEventMessage | + {eventType: 'worker-unavailable'} | + {eventType: 'worker-connected'}; + +export type UserEventType = UserEventMessage['eventType']; + +export type WorkerEventMessage = + {workerEvent: 'error', message: string} | + {workerEvent: 'close'}; + +export type WorkerInboundMessage = + {msgType: 'user-event', msgData: UserEventMessage} | + {msgType: 'worker-event', msgData: WorkerEventMessage}; + +export type SharedWorkerControlMessage = { + type: 'start', + url: string, + showDebugLog: boolean, +} | {type: 'close'}; diff --git a/web_src/js/user-events.sharedworker.ts b/web_src/js/user-events.sharedworker.ts new file mode 100644 index 0000000000..bcc2b4b0c8 --- /dev/null +++ b/web_src/js/user-events.sharedworker.ts @@ -0,0 +1,243 @@ +import { + serverUserEventTypes, + type SharedWorkerControlMessage, + type WorkerEventMessage, + type WorkerInboundMessage, +} from './types.ts'; +import type {UserEventMessage} from './types.ts'; + +// chrome://inspect/#workers +let showDebugLog = false; + +function logDebug(...args: any[]) { + if (!showDebugLog) return; + console.debug('[user-events.sharedworker]', ...args); +} + +function isServerEventMessage(msg: unknown): msg is UserEventMessage { + if (!msg || typeof msg !== 'object') return false; + const userEvent = msg as UserEventMessage; + return (serverUserEventTypes as ReadonlyArray).includes(userEvent.eventType ?? ''); +} + +function postUserEventMessage(client: MessagePort, msgData: UserEventMessage) { + logDebug('postUserEventMessage', msgData); + const msg: WorkerInboundMessage = {msgType: 'user-event', msgData}; + client.postMessage(msg); +} + +function postWorkerEventMessage(client: MessagePort, msgData: WorkerEventMessage) { + logDebug('postWorkerEventMessage', msgData); + const msg: WorkerInboundMessage = {msgType: 'worker-event', msgData}; + client.postMessage(msg); +} + +class Source { + url: string; + clients: Array; + + constructor(url: string) { + this.url = url; + this.clients = []; + } + + register(port: MessagePort) { + if (this.clients.includes(port)) return; + this.clients.push(port); + } + + deregister(port: MessagePort) { + const portIdx = this.clients.indexOf(port); + if (portIdx < 0) { + return this.clients.length; + } + this.clients.splice(portIdx, 1); + return this.clients.length; + } + + notifyClientsUserEvent(event: UserEventMessage) { + for (const client of this.clients) { + postUserEventMessage(client, event); + } + } + notifyClientsWorkerEvent(event: WorkerEventMessage) { + for (const client of this.clients) { + postWorkerEventMessage(client, event); + } + } +} + +class WsSource { + wsUrl: string; + ws: WebSocket | null; + source: Source; + reconnectTimer: ReturnType | null; + reconnectDelay: number; + failuresWithoutConnect: number; + fallbackSignalled: boolean; + closed: boolean; + + constructor(wsUrl: string, source: Source) { + this.wsUrl = wsUrl; + this.source = source; + this.ws = null; + this.reconnectTimer = null; + this.reconnectDelay = 1000; + this.failuresWithoutConnect = 0; + this.fallbackSignalled = false; + this.closed = false; + this.connect(); + } + + connect() { + if (this.closed) return; + this.ws = new WebSocket(this.wsUrl); + + 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'}); + }); + + this.ws.addEventListener('message', (event: MessageEvent) => { + try { + const msg: unknown = JSON.parse(event.data); + logDebug('websocket message', event.data); + if (!isServerEventMessage(msg)) { + console.error('websocket message is not a valid server user event', msg); + return; + } + this.source.notifyClientsUserEvent(msg); + } catch (err) { + console.error('user-events: dropping malformed WebSocket message', err); + } + }); + + // `error` always fires before `close` on a failed connection, so we count + // failures and schedule reconnects from `close` only — otherwise the + // fallback threshold would trip after two real failures instead of three. + this.ws.addEventListener('error', () => { + this.ws = null; + }); + + this.ws.addEventListener('close', (event: CloseEvent) => { + this.ws = null; + if (this.closed) return; + // Server signals an expired/missing session via the IANA "Unauthorized" close code; reconnecting can't recover that. + if (event.code === 3000) { + this.closed = true; + return; + } + this.failuresWithoutConnect++; + this.maybeSignalFallback(); + this.scheduleReconnect(); + }); + } + + maybeSignalFallback() { + if (this.fallbackSignalled) return; + if (this.failuresWithoutConnect < 3) return; + this.fallbackSignalled = true; + this.source.notifyClientsUserEvent({eventType: 'worker-unavailable'}); + } + + scheduleReconnect() { + if (this.reconnectTimer !== null) return; + // Jitter 50%–150% of base delay to prevent thundering-herd reconnects after a server restart. + const delay = this.reconnectDelay * (0.5 + Math.random()); + logDebug(`scheduling reconnect in ${delay}ms`); + this.reconnectTimer = setTimeout(() => { + logDebug(`reconnecting ...`); + this.reconnectTimer = null; + this.connect(); + }, delay); + this.reconnectDelay = Math.min(this.reconnectDelay * 2, 60000); + } + + close() { + this.closed = true; + 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) { + port.addEventListener('message', (event: MessageEvent) => { + if (event.data.type === 'start') { + logDebug('received control message start', event.data); + showDebugLog = event.data.showDebugLog; + const url = event.data.url; + let source = sourcesByUrl.get(url); + if (source) { + // we have a Source registered to this url + source.register(port); + sourcesByPort.set(port, source); + // A port attaching to an already-open socket won't observe the next + // "open", so replay "worker-connected" to it directly; otherwise a late tab + // never learns the stream is live (mirrors the SSE sharedworker replaying + // its built-in "open" event to late-attaching ports). + const openWs = wsSourcesByUrl.get(url); + if (openWs?.ws?.readyState === WebSocket.OPEN) { + postUserEventMessage(port, {eventType: 'worker-connected'}); + } + return; + } + source = sourcesByPort.get(port); + if (source) { + if (source.url === url) return; + + // How this has happened I don't understand... + // deregister from that source + const count = source.deregister(port); + // Clean-up + if (count === 0) { + sourcesByUrl.delete(source.url); + const ws = wsSourcesByUrl.get(source.url); + if (ws) { + ws.close(); + wsSourcesByUrl.delete(source.url); + } + } + } + // Create a new Source and its WebSocket transport + source = new Source(url); + source.register(port); + sourcesByUrl.set(url, source); + sourcesByPort.set(port, source); + wsSourcesByUrl.set(url, new WsSource(url, source)); + } else if (event.data.type === 'close') { + logDebug('received control message close', event.data); + const source = sourcesByPort.get(port); + if (!source) return; + + const count = source.deregister(port); + sourcesByPort.delete(port); + if (count === 0) { + sourcesByUrl.delete(source.url); + const ws = wsSourcesByUrl.get(source.url); + if (ws) { + ws.close(); + wsSourcesByUrl.delete(source.url); + } + } + } else { + // just send it back + console.error('received control message unknown', event.data); + postWorkerEventMessage(port, {workerEvent: 'error', message: `received but don't know how to handle: ${JSON.stringify(event.data)}`}); + } + }); + port.start(); + } +}); diff --git a/web_src/js/vitest.setup.ts b/web_src/js/vitest.setup.ts index 1c0c27a667..4c1c03a739 100644 --- a/web_src/js/vitest.setup.ts +++ b/web_src/js/vitest.setup.ts @@ -8,7 +8,7 @@ window.config = { runModeIsProd: true, customEmojis: {}, pageData: {}, - notificationSettings: {MinTimeout: 0, TimeoutStep: 0, MaxTimeout: 0, EventSourceUpdateTime: 0}, + notificationSettings: {MinTimeout: 0, TimeoutStep: 0, MaxTimeout: 0}, enableTimeTracking: true, mermaidMaxSourceCharacters: 5000, i18n: {},