feat: Replace SSE with WebSocket for UI notifications (#36965)

* Closes #36942
* Fixes #19265 

Replaces the SSE-based push channel (`/user/events`) with a WebSocket
endpoint (`/-/ws`).

### What changes

- **New `/-/ws` endpoint** (authenticated). One WebSocket per origin,
shared across tabs via a single `SharedWorker`.
- **Pubsub broker** (`services/pubsub`) for fan-out by topic, behind a
`Broker` interface. `MemoryBroker` is the default (single process); a
Redis backend is available for multi-process setups, configured via
`[websocket].PUBSUB_TYPE` / `PUBSUB_CONN_STR`. The internal Gitea queue
was not usable here because it has FIFO/single-consumer semantics.
- **Push-only event production.** Events are emitted by write-triggered
notifiers — `NotificationCountChange`, `PublishStopwatchesForUser`, and
the logout publisher — wired into the existing `notify.Notifier`
interface. No server-side pollers.
- **Typed pub/sub on the client.** `web_src/js/modules/worker.ts` is a
singleton transport; features subscribe per event type via
`onUserEvent('notification-count', cb)` instead of branching on
`event.data.type`.
- **Wire contract** (`UserEventType` union) is shared between the worker
and consumers via `web_src/js/types.ts`, kept in sync with
`services/websocket/events.go`.
- **Client-side periodic polling fallback** kicks in only when the
WebSocket cannot be established (e.g. proxy blocks WS, browser lacks
module-SharedWorker support).

### What's removed

- `modules/eventsource` (SSE manager, run loop, messenger).
- `/user/events` route and `tests/integration/eventsource_test.go`.
- All server-side polling for stopwatches and notification counts.

### Stopwatch multi-tab fix

The navbar stopwatch icon was previously rendered conditionally on `{{if
$activeStopwatch}}`, so tabs loaded before the timer started had no DOM
element to update. The icon and popup are now always rendered (toggled
with `tw-hidden`), and the start/stop/cancel handlers POST silently so
all open tabs reflect the change in real time.

### Deployment note

WebSocket needs the upgrade headers to pass through a reverse proxy,
e.g. for nginx:

```nginx
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
```

Without them the WebSocket cannot be established, and after 3
consecutive failed opens the shared worker signals `push-unavailable`:
the notification count and stopwatch fall back to periodic polling on
the existing `[ui.notification]` timeouts. Real-time push is lost, the
features keep working. The reverse-proxy docs need the same note (see
the `docs-update-needed` label).

---------

Co-authored-by: silverwind <me@silverwind.io>
Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>
Co-authored-by: Epid <rexmrj@gmail.com>
This commit is contained in:
mohammad rahimi
2026-07-27 07:46:00 +00:00
committed by GitHub
co-authored by silverwind wxiaoguang Epid
parent e15a7e9066
commit 13d0f24423
77 changed files with 2070 additions and 1151 deletions
+5
View File
@@ -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",
+15 -4
View File
@@ -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
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+1
View File
@@ -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
+2
View File
@@ -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=
+16 -12
View File
@@ -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
}
+26 -21
View File
@@ -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
+6 -3
View File
@@ -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)
-111
View File
@@ -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
}
-50
View File
@@ -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())
})
}
}
-89
View File
@@ -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)
}
}
-122
View File
@@ -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()
}
-77
View File
@@ -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
}
}
+6
View File
@@ -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)
+4
View File
@@ -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)
}
+4
View File
@@ -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)
}
+4 -57
View File
@@ -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)
}
+1
View File
@@ -221,6 +221,7 @@ func LoadSettings() {
loadServiceFrom(CfgProvider)
loadOAuth2ClientFrom(CfgProvider)
loadCacheFrom(CfgProvider)
loadWebsocketFrom(CfgProvider)
loadSessionFrom(CfgProvider)
loadCorsFrom(CfgProvider)
loadMailsFrom(CfgProvider)
+9 -12
View File
@@ -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"`
+34
View File
@@ -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")
}
}
}
+63
View File
@@ -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)
})
}
}
+3 -4
View File
@@ -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 {
+78
View File
@@ -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
}
+4 -2
View File
@@ -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 {
+1 -1
View File
@@ -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)
+8 -8
View File
@@ -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))
}
+2 -1
View File
@@ -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
+8 -8
View File
@@ -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))
}
+4
View File
@@ -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)
}
+2 -2
View File
@@ -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
}
+2 -2
View File
@@ -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)
+2 -5
View File
@@ -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)
-124
View File
@@ -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()
}
+5 -14
View File
@@ -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("")
}
+2 -2
View File
@@ -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
}
+8 -4
View File
@@ -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) {
+4 -4
View File
@@ -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
}
+2 -2
View File
@@ -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() {
+117
View File
@@ -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 (30003999).
// 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
}
}
}
}
+60
View File
@@ -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)
})
}
}
+7 -1
View File
@@ -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
+60
View File
@@ -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
}
+4
View File
@@ -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)
}
+14
View File
@@ -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)
}
}
+6
View File
@@ -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) {
}
+39
View File
@@ -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)
}
+170
View File
@@ -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))
}
+29
View File
@@ -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
}
+71
View File
@@ -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)
}
}
}
+99
View File
@@ -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())
}
+190
View File
@@ -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
}
+71
View File
@@ -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")
}
})
}
+7 -1
View File
@@ -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
+2 -4
View File
@@ -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
+53
View File
@@ -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
}
+12
View File
@@ -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})
}
@@ -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})
}
+18
View File
@@ -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
}
+37
View File
@@ -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))
}
+23 -24
View File
@@ -152,31 +152,30 @@
</div><!-- end full right menu -->
{{$activeStopwatch := and .PageGlobalData (call .PageGlobalData.GetActiveStopwatch)}}
{{if $activeStopwatch}}
<div class="active-stopwatch-popup tippy-target">
<div class="flex-text-block tw-p-3">
<a class="stopwatch-link flex-text-block muted" href="{{$activeStopwatch.IssueLink}}">
{{svg "octicon-issue-opened" 16}}
<span class="stopwatch-issue">{{$activeStopwatch.RepoSlug}}#{{$activeStopwatch.IssueIndex}}</span>
</a>
<div class="tw-flex tw-gap-1">
<form class="stopwatch-commit form-fetch-action" method="post" action="{{$activeStopwatch.IssueLink}}/times/stopwatch/stop">
<button
type="submit"
class="ui button mini compact basic icon tw-mr-0"
data-tooltip-content="{{ctx.Locale.Tr "repo.issues.stop_tracking"}}"
>{{svg "octicon-square-fill"}}</button>
</form>
<form class="stopwatch-cancel form-fetch-action" method="post" action="{{$activeStopwatch.IssueLink}}/times/stopwatch/cancel">
<button
type="submit"
class="ui button mini compact basic icon tw-mr-0"
data-tooltip-content="{{ctx.Locale.Tr "repo.issues.cancel_tracking"}}"
>{{svg "octicon-trash"}}</button>
</form>
</div>
{{/* always rendered so the Tippy popup has content to clone if the icon appears via a real-time push */}}
<div class="active-stopwatch-popup tippy-target">
<div class="flex-text-block tw-p-3">
<a class="stopwatch-link flex-text-block muted" {{if $activeStopwatch}}href="{{$activeStopwatch.IssueLink}}"{{end}}>
{{svg "octicon-issue-opened" 16}}
<span class="stopwatch-issue">{{if $activeStopwatch}}{{$activeStopwatch.RepoSlug}}#{{$activeStopwatch.IssueIndex}}{{end}}</span>
</a>
<div class="tw-flex tw-gap-1">
<form class="stopwatch-commit form-fetch-action" method="post" {{if $activeStopwatch}}action="{{$activeStopwatch.IssueLink}}/times/stopwatch/stop"{{end}}>
<button
type="submit"
class="ui button mini compact basic icon tw-mr-0"
data-tooltip-content="{{ctx.Locale.Tr "repo.issues.stop_tracking"}}"
>{{svg "octicon-square-fill"}}</button>
</form>
<form class="stopwatch-cancel form-fetch-action" method="post" {{if $activeStopwatch}}action="{{$activeStopwatch.IssueLink}}/times/stopwatch/cancel"{{end}}>
<button
type="submit"
class="ui button mini compact basic icon tw-mr-0"
data-tooltip-content="{{ctx.Locale.Tr "repo.issues.cancel_tracking"}}"
>{{svg "octicon-trash"}}</button>
</form>
</div>
</div>
{{end}}
</div>
</nav>
{{template "base/head_banner"}}
+2 -3
View File
@@ -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}}
<a class="item active-stopwatch {{$itemExtraClass}}" href="{{$activeStopwatch.IssueLink}}" title="{{ctx.Locale.Tr "active_stopwatch"}}" data-seconds="{{$activeStopwatch.Seconds}}">
{{/* always rendered so a real-time push can reveal it without a reload */}}
<a class="item active-stopwatch{{if not $activeStopwatch}} tw-hidden{{end}} {{$itemExtraClass}}" {{if $activeStopwatch}}href="{{$activeStopwatch.IssueLink}}" data-seconds="{{$activeStopwatch.Seconds}}"{{end}} title="{{ctx.Locale.Tr "active_stopwatch"}}">
<div class="tw-relative flex-text-block">
{{svg "octicon-stopwatch"}}
<span class="header-stopwatch-dot"></span>
</div>
</a>
{{end}}
<a class="item {{$itemExtraClass}}" href="{{AppSubUrl}}/notifications" data-tooltip-content="{{ctx.Locale.Tr "notifications"}}">
<div class="tw-relative flex-text-block">
{{svg "octicon-bell"}}
+1 -1
View File
@@ -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"}},
+1 -1
View File
@@ -6,7 +6,7 @@
<div class="repo-view-container">
{{template "repo/view_file_tree" .}}
<div class="repo-view-content">
<form class="ui comment form form-fetch-action" method="post" action="{{.CommitFormOptions.TargetFormAction}}">
<form class="ui form form-fetch-action repo-file-upload" method="post" action="{{.CommitFormOptions.TargetFormAction}}">
{{template "repo/editor/common_top" .}}
<div class="repo-editor-header">
{{template "repo/view_file_tree_toggle_button" .}}
+74 -17
View File
@@ -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
+13
View File
@@ -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<string, string>} = {}) {
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<string, string>} = {}) {
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(),
+1 -1
View File
@@ -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", "")
-84
View File
@@ -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)
}
+3 -3
View File
@@ -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")
+2 -2
View File
@@ -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,
},
-150
View File
@@ -1,150 +0,0 @@
class Source {
url: string;
eventSource: EventSource | null;
listening: Record<string, boolean>;
clients: Array<MessagePort>;
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<string, Source | null>();
const sourcesByPort = new Map<MessagePort, Source | null>();
(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();
}
});
+15 -28
View File
@@ -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() {
+30 -29
View File
@@ -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<StopwatchData>) {
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`);
-1
View File
@@ -44,7 +44,6 @@ interface Window {
MinTimeout: number,
TimeoutStep: number,
MaxTimeout: number,
EventSourceUpdateTime: number,
},
enableTimeTracking: boolean,
mermaidMaxSourceCharacters: number,
+69
View File
@@ -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<string, PortListener[]> = {};
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');
});
+121 -59
View File
@@ -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<T extends UserEventType> = Extract<UserEventMessage, {eventType: T}>;
type Subscriber<T extends UserEventType = UserEventType> = (msg: EventOf<T>) => void;
const subscribers = new Map<UserEventType, Set<Subscriber>>();
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<WorkerInboundMessage>) => {
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<T extends UserEventType>(type: T, cb: Subscriber<T>): () => 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<T>);
set.add(wrapped);
return () => { set.delete(wrapped) };
}
+35
View File
@@ -63,3 +63,38 @@ export type FomanticInitFunction = {
export type GitRefType = 'branch' | 'tag';
export type Promisable<T> = T | Promise<T>; // 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<StopwatchData>} |
{eventType: 'logout'};
export const serverUserEventTypes = ['notification-count', 'stopwatches', 'logout'] as const satisfies ReadonlyArray<ServerUserEventMessage['eventType']>;
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'};
+243
View File
@@ -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<string>).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<MessagePort>;
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<typeof setTimeout> | 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<string>) => {
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<string, Source>();
const sourcesByPort = new Map<MessagePort, Source>();
const wsSourcesByUrl = new Map<string, WsSource>();
(self as unknown as SharedWorkerGlobalScope).addEventListener('connect', (e: MessageEvent) => {
for (const port of e.ports) {
port.addEventListener('message', (event: MessageEvent<SharedWorkerControlMessage>) => {
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();
}
});
+1 -1
View File
@@ -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: {},