mirror of
https://github.com/go-gitea/gitea.git
synced 2026-08-01 10:40:38 +00:00
13d0f24423
* 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>
423 lines
13 KiB
Go
423 lines
13 KiB
Go
// Copyright 2016 The Gitea Authors. All rights reserved.
|
|
// SPDX-License-Identifier: MIT
|
|
|
|
package activities
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/url"
|
|
"strconv"
|
|
|
|
"gitea.dev/models/db"
|
|
issues_model "gitea.dev/models/issues"
|
|
"gitea.dev/models/organization"
|
|
repo_model "gitea.dev/models/repo"
|
|
user_model "gitea.dev/models/user"
|
|
"gitea.dev/modules/setting"
|
|
"gitea.dev/modules/timeutil"
|
|
|
|
"xorm.io/builder"
|
|
"xorm.io/xorm/schemas"
|
|
)
|
|
|
|
type (
|
|
// NotificationStatus is the status of the notification (read or unread)
|
|
NotificationStatus uint8
|
|
// NotificationSource is the source of the notification (issue, PR, commit, etc)
|
|
NotificationSource uint8
|
|
)
|
|
|
|
const (
|
|
// NotificationStatusUnread represents an unread notification
|
|
NotificationStatusUnread NotificationStatus = iota + 1
|
|
// NotificationStatusRead represents a read notification
|
|
NotificationStatusRead
|
|
// NotificationStatusPinned represents a pinned notification
|
|
NotificationStatusPinned
|
|
)
|
|
|
|
const (
|
|
// NotificationSourceIssue is a notification of an issue
|
|
NotificationSourceIssue NotificationSource = iota + 1
|
|
// NotificationSourcePullRequest is a notification of a pull request
|
|
NotificationSourcePullRequest
|
|
// NotificationSourceCommit is a notification of a commit
|
|
NotificationSourceCommit
|
|
// NotificationSourceRepository is a notification for a repository
|
|
NotificationSourceRepository
|
|
)
|
|
|
|
// Notification represents a notification
|
|
type Notification struct {
|
|
ID int64 `xorm:"pk autoincr"`
|
|
UserID int64 `xorm:"NOT NULL"`
|
|
RepoID int64 `xorm:"NOT NULL"`
|
|
|
|
Status NotificationStatus `xorm:"SMALLINT NOT NULL"`
|
|
Source NotificationSource `xorm:"SMALLINT NOT NULL"`
|
|
|
|
IssueID int64 `xorm:"NOT NULL"`
|
|
CommitID string
|
|
CommentID int64
|
|
|
|
UpdatedBy int64 `xorm:"NOT NULL"`
|
|
|
|
Issue *issues_model.Issue `xorm:"-"`
|
|
Repository *repo_model.Repository `xorm:"-"`
|
|
Comment *issues_model.Comment `xorm:"-"`
|
|
User *user_model.User `xorm:"-"`
|
|
|
|
CreatedUnix timeutil.TimeStamp `xorm:"created NOT NULL"`
|
|
UpdatedUnix timeutil.TimeStamp `xorm:"updated NOT NULL"`
|
|
}
|
|
|
|
// TableIndices implements xorm's TableIndices interface
|
|
func (n *Notification) TableIndices() []*schemas.Index {
|
|
indices := make([]*schemas.Index, 0, 8)
|
|
usuuIndex := schemas.NewIndex("u_s_uu", schemas.IndexType)
|
|
usuuIndex.AddColumn("user_id", "status", "updated_unix")
|
|
indices = append(indices, usuuIndex)
|
|
|
|
// Add the individual indices that were previously defined in struct tags
|
|
userIDIndex := schemas.NewIndex("idx_notification_user_id", schemas.IndexType)
|
|
userIDIndex.AddColumn("user_id")
|
|
indices = append(indices, userIDIndex)
|
|
|
|
repoIDIndex := schemas.NewIndex("idx_notification_repo_id", schemas.IndexType)
|
|
repoIDIndex.AddColumn("repo_id")
|
|
indices = append(indices, repoIDIndex)
|
|
|
|
statusIndex := schemas.NewIndex("idx_notification_status", schemas.IndexType)
|
|
statusIndex.AddColumn("status")
|
|
indices = append(indices, statusIndex)
|
|
|
|
sourceIndex := schemas.NewIndex("idx_notification_source", schemas.IndexType)
|
|
sourceIndex.AddColumn("source")
|
|
indices = append(indices, sourceIndex)
|
|
|
|
issueIDIndex := schemas.NewIndex("idx_notification_issue_id", schemas.IndexType)
|
|
issueIDIndex.AddColumn("issue_id")
|
|
indices = append(indices, issueIDIndex)
|
|
|
|
commitIDIndex := schemas.NewIndex("idx_notification_commit_id", schemas.IndexType)
|
|
commitIDIndex.AddColumn("commit_id")
|
|
indices = append(indices, commitIDIndex)
|
|
|
|
updatedByIndex := schemas.NewIndex("idx_notification_updated_by", schemas.IndexType)
|
|
updatedByIndex.AddColumn("updated_by")
|
|
indices = append(indices, updatedByIndex)
|
|
|
|
return indices
|
|
}
|
|
|
|
func init() {
|
|
db.RegisterModel(new(Notification))
|
|
}
|
|
|
|
// CreateRepoTransferNotification creates notification for the user a repository was transferred to
|
|
func CreateRepoTransferNotification(ctx context.Context, doer, newOwner *user_model.User, repo *repo_model.Repository) error {
|
|
return db.WithTx(ctx, func(ctx context.Context) error {
|
|
var notify []*Notification
|
|
|
|
if newOwner.IsOrganization() {
|
|
users, err := organization.GetUsersWhoCanCreateOrgRepo(ctx, newOwner.ID)
|
|
if err != nil || len(users) == 0 {
|
|
return err
|
|
}
|
|
for i := range users {
|
|
notify = append(notify, &Notification{
|
|
UserID: i,
|
|
RepoID: repo.ID,
|
|
Status: NotificationStatusUnread,
|
|
UpdatedBy: doer.ID,
|
|
Source: NotificationSourceRepository,
|
|
})
|
|
}
|
|
} else {
|
|
notify = []*Notification{{
|
|
UserID: newOwner.ID,
|
|
RepoID: repo.ID,
|
|
Status: NotificationStatusUnread,
|
|
UpdatedBy: doer.ID,
|
|
Source: NotificationSourceRepository,
|
|
}}
|
|
}
|
|
|
|
return db.Insert(ctx, notify)
|
|
})
|
|
}
|
|
|
|
func createIssueNotification(ctx context.Context, userID int64, issue *issues_model.Issue, commentID, updatedByID int64) error {
|
|
notification := &Notification{
|
|
UserID: userID,
|
|
RepoID: issue.RepoID,
|
|
Status: NotificationStatusUnread,
|
|
IssueID: issue.ID,
|
|
CommentID: commentID,
|
|
UpdatedBy: updatedByID,
|
|
}
|
|
|
|
if issue.IsPull {
|
|
notification.Source = NotificationSourcePullRequest
|
|
} else {
|
|
notification.Source = NotificationSourceIssue
|
|
}
|
|
|
|
return db.Insert(ctx, notification)
|
|
}
|
|
|
|
func updateIssueNotification(ctx context.Context, userID, issueID, commentID, updatedByID int64) error {
|
|
notification, err := GetIssueNotification(ctx, userID, issueID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// NOTICE: Only update comment id when the before notification on this issue is read, otherwise you may miss some old comments.
|
|
// But we need update update_by so that the notification will be reorder
|
|
var cols []string
|
|
if notification.Status == NotificationStatusRead {
|
|
notification.Status = NotificationStatusUnread
|
|
notification.CommentID = commentID
|
|
cols = []string{"status", "update_by", "comment_id"}
|
|
} else {
|
|
notification.UpdatedBy = updatedByID
|
|
cols = []string{"update_by"}
|
|
}
|
|
|
|
_, err = db.GetEngine(ctx).ID(notification.ID).Cols(cols...).Update(notification)
|
|
return err
|
|
}
|
|
|
|
// GetIssueNotification return the notification about an issue
|
|
func GetIssueNotification(ctx context.Context, userID, issueID int64) (*Notification, error) {
|
|
notification := new(Notification)
|
|
_, err := db.GetEngine(ctx).
|
|
Where("user_id = ?", userID).
|
|
And("issue_id = ?", issueID).
|
|
Get(notification)
|
|
return notification, err
|
|
}
|
|
|
|
// LoadAttributes load Repo Issue User and Comment if not loaded
|
|
func (n *Notification) LoadAttributes(ctx context.Context) (err error) {
|
|
if err = n.loadRepo(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err = n.loadIssue(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err = n.loadUser(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err = n.loadComment(ctx); err != nil {
|
|
return err
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (n *Notification) loadRepo(ctx context.Context) (err error) {
|
|
if n.Repository == nil {
|
|
n.Repository, err = repo_model.GetRepositoryByID(ctx, n.RepoID)
|
|
if err != nil {
|
|
return fmt.Errorf("getRepositoryByID [%d]: %w", n.RepoID, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (n *Notification) loadIssue(ctx context.Context) (err error) {
|
|
if n.Issue == nil && n.IssueID != 0 {
|
|
n.Issue, err = issues_model.GetIssueByID(ctx, n.IssueID)
|
|
if err != nil {
|
|
return fmt.Errorf("getIssueByID [%d]: %w", n.IssueID, err)
|
|
}
|
|
return n.Issue.LoadAttributes(ctx)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (n *Notification) loadComment(ctx context.Context) (err error) {
|
|
if n.Comment == nil && n.CommentID != 0 {
|
|
n.Comment, err = issues_model.GetCommentByID(ctx, n.CommentID)
|
|
if err != nil {
|
|
if issues_model.IsErrCommentNotExist(err) {
|
|
return issues_model.ErrCommentNotExist{
|
|
ID: n.CommentID,
|
|
IssueID: n.IssueID,
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (n *Notification) loadUser(ctx context.Context) (err error) {
|
|
if n.User == nil {
|
|
n.User, err = user_model.GetUserByID(ctx, n.UserID)
|
|
if err != nil {
|
|
return fmt.Errorf("getUserByID [%d]: %w", n.UserID, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// GetRepo returns the repo of the notification
|
|
func (n *Notification) GetRepo(ctx context.Context) (*repo_model.Repository, error) {
|
|
return n.Repository, n.loadRepo(ctx)
|
|
}
|
|
|
|
// GetIssue returns the issue of the notification
|
|
func (n *Notification) GetIssue(ctx context.Context) (*issues_model.Issue, error) {
|
|
return n.Issue, n.loadIssue(ctx)
|
|
}
|
|
|
|
// HTMLURL formats a URL-string to the notification
|
|
func (n *Notification) HTMLURL(ctx context.Context) string {
|
|
switch n.Source {
|
|
case NotificationSourceIssue, NotificationSourcePullRequest:
|
|
if n.Comment != nil {
|
|
return n.Comment.HTMLURL(ctx)
|
|
}
|
|
return n.Issue.HTMLURL(ctx)
|
|
case NotificationSourceCommit:
|
|
return n.Repository.HTMLURL(ctx) + "/commit/" + url.PathEscape(n.CommitID)
|
|
case NotificationSourceRepository:
|
|
return n.Repository.HTMLURL(ctx)
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// Link formats a relative URL-string to the notification
|
|
func (n *Notification) Link(ctx context.Context) string {
|
|
switch n.Source {
|
|
case NotificationSourceIssue, NotificationSourcePullRequest:
|
|
if n.Comment != nil {
|
|
return n.Comment.Link(ctx)
|
|
}
|
|
return n.Issue.Link()
|
|
case NotificationSourceCommit:
|
|
return n.Repository.Link() + "/commit/" + url.PathEscape(n.CommitID)
|
|
case NotificationSourceRepository:
|
|
return n.Repository.Link()
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// APIURL formats a URL-string to the notification
|
|
func (n *Notification) APIURL() string {
|
|
return setting.AppURL + "api/v1/notifications/threads/" + strconv.FormatInt(n.ID, 10)
|
|
}
|
|
|
|
func notificationExists(notifications []*Notification, issueID, userID int64) bool {
|
|
for _, notification := range notifications {
|
|
if notification.IssueID == issueID && notification.UserID == userID {
|
|
return true
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
// UserIDCount is a simple coalition of UserID and Count
|
|
type UserIDCount struct {
|
|
UserID int64
|
|
Count int64
|
|
}
|
|
|
|
// GetUIDsAndNotificationCounts returns the unread counts for every user between the two provided times.
|
|
// It must return all user IDs which appear during the period, including count=0 for users who have read all.
|
|
func GetUIDsAndNotificationCounts(ctx context.Context, since, until timeutil.TimeStamp) ([]UserIDCount, error) {
|
|
sql := `SELECT user_id, sum(case when status= ? then 1 else 0 end) AS count FROM notification ` +
|
|
`WHERE user_id IN (SELECT user_id FROM notification WHERE updated_unix >= ? AND ` +
|
|
`updated_unix < ?) GROUP BY user_id`
|
|
var res []UserIDCount
|
|
return res, db.GetEngine(ctx).SQL(sql, NotificationStatusUnread, since, until).Find(&res)
|
|
}
|
|
|
|
// 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 false, err
|
|
}
|
|
|
|
return setIssueNotificationStatusReadIfUnread(ctx, userID, issueID)
|
|
}
|
|
|
|
func setIssueNotificationStatusReadIfUnread(ctx context.Context, userID, issueID int64) (bool, error) {
|
|
notification, err := GetIssueNotification(ctx, userID, issueID)
|
|
// ignore if not exists
|
|
if err != nil {
|
|
return false, nil
|
|
}
|
|
|
|
if notification.Status != NotificationStatusUnread {
|
|
return false, nil
|
|
}
|
|
|
|
notification.Status = NotificationStatusRead
|
|
|
|
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.
|
|
func SetRepoReadBy(ctx context.Context, userID, repoID int64) error {
|
|
_, err := db.GetEngine(ctx).Where(builder.Eq{
|
|
"user_id": userID,
|
|
"status": NotificationStatusUnread,
|
|
"source": NotificationSourceRepository,
|
|
"repo_id": repoID,
|
|
}).Cols("status").Update(&Notification{Status: NotificationStatusRead})
|
|
return err
|
|
}
|
|
|
|
// SetNotificationStatus change the notification status
|
|
func SetNotificationStatus(ctx context.Context, notificationID int64, user *user_model.User, status NotificationStatus) (*Notification, error) {
|
|
notification, err := GetNotificationByID(ctx, notificationID)
|
|
if err != nil {
|
|
return notification, err
|
|
}
|
|
|
|
if notification.UserID != user.ID {
|
|
return nil, fmt.Errorf("Can't change notification of another user: %d, %d", notification.UserID, user.ID)
|
|
}
|
|
|
|
notification.Status = status
|
|
|
|
_, err = db.GetEngine(ctx).ID(notificationID).Cols("status").Update(notification)
|
|
return notification, err
|
|
}
|
|
|
|
// GetNotificationByID return notification by ID
|
|
func GetNotificationByID(ctx context.Context, notificationID int64) (*Notification, error) {
|
|
notification := new(Notification)
|
|
ok, err := db.GetEngine(ctx).
|
|
Where("id = ?", notificationID).
|
|
Get(notification)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if !ok {
|
|
return nil, db.ErrNotExist{Resource: "notification", ID: notificationID}
|
|
}
|
|
|
|
return notification, nil
|
|
}
|
|
|
|
// 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}
|
|
return db.GetEngine(ctx).
|
|
Where("user_id = ? AND status = ?", user.ID, currentStatus).
|
|
Cols("status", "updated_by", "updated_unix").
|
|
Update(n)
|
|
}
|