Merge 1328b2b9fe into 27861d711b

go.mod

@@ -77,9 +77,11 @@ require (
 	github.com/microcosm-cc/bluemonday v1.0.26
 	github.com/microsoft/go-mssqldb v1.7.0
 	github.com/minio/minio-go/v7 v7.0.69
+	github.com/mitchellh/mapstructure v1.5.0
 	github.com/msteinert/pam v1.2.0
 	github.com/nektos/act v0.2.52
 	github.com/niklasfasching/go-org v1.7.0
+	github.com/olahol/melody v1.1.4
 	github.com/olivere/elastic/v7 v7.0.32
 	github.com/opencontainers/go-digest v1.0.0
 	github.com/opencontainers/image-spec v1.1.0
@@ -209,6 +211,7 @@ require (
 	github.com/gorilla/handlers v1.5.2 // indirect
 	github.com/gorilla/mux v1.8.1 // indirect
 	github.com/gorilla/securecookie v1.1.2 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/hashicorp/go-retryablehttp v0.7.5 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
@@ -231,7 +234,6 @@ require (
 	github.com/minio/md5-simd v1.1.2 // indirect
 	github.com/minio/sha256-simd v1.0.1 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
-	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect

go.sum

@@ -430,6 +430,8 @@ github.com/gorilla/securecookie v1.1.2/go.mod h1:NfCASbcHqRSY+3a8tlWJwsQap2VX5pw
 github.com/gorilla/sessions v1.2.0/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
 github.com/gorilla/sessions v1.2.2 h1:lqzMYz6bOfvn2WriPUjNByzeXIlVzURcPmgMczkmTjY=
 github.com/gorilla/sessions v1.2.2/go.mod h1:ePLdVu+jbEgHH+KWw8I1z2wqd0BAdAQh/8LRvBeoNcQ=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ=
 github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48=
 github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
@@ -601,6 +603,8 @@ github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
 github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
 github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
+github.com/olahol/melody v1.1.4 h1:RQHfKZkQmDxI0+SLZRNBCn4LiXdqxLKRGSkT8Dyoe/E=
+github.com/olahol/melody v1.1.4/go.mod h1:GgkTl6Y7yWj/HtfD48Q5vLKPVoZOH+Qqgfa7CvJgJM4=
 github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
 github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY=
 github.com/olivere/elastic/v7 v7.0.32 h1:R7CXvbu8Eq+WlsLgxmKVKPox0oOwAE/2T9Si5BnvK6E=

routers/web/web.go

@@ -40,6 +40,7 @@ import (
 	"code.gitea.io/gitea/routers/web/user"
 	user_setting "code.gitea.io/gitea/routers/web/user/setting"
 	"code.gitea.io/gitea/routers/web/user/setting/security"
+	"code.gitea.io/gitea/routers/web/websocket"
 	auth_service "code.gitea.io/gitea/services/auth"
 	"code.gitea.io/gitea/services/context"
 	"code.gitea.io/gitea/services/forms"
@@ -541,6 +542,8 @@ func registerRoutes(m *web.Route) {
 
 	m.Any("/user/events", routing.MarkLongPolling, events.Events)
 
+	websocket.Init(m)
+
 	m.Group("/login/oauth", func() {
 		m.Get("/authorize", web.Bind(forms.AuthorizationForm{}), auth.AuthorizeOAuth)
 		m.Post("/grant", web.Bind(forms.GrantApplicationForm{}), auth.GrantApplicationOAuth)

routers/web/websocket/websocket.go

@@ -0,0 +1,22 @@
+// Copyright 2023 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"code.gitea.io/gitea/modules/web"
+	"code.gitea.io/gitea/services/context"
+	"code.gitea.io/gitea/services/websocket"
+)
+
+func Init(r *web.Route) {
+	m := websocket.Init()
+
+	r.Any("/-/ws", func(ctx *context.Context) {
+		err := m.HandleRequest(ctx.Resp, ctx.Req)
+		if err != nil {
+			ctx.ServerError("HandleRequest", err)
+			return
+		}
+	})
+}

services/context/response.go

@@ -4,6 +4,9 @@
 package context
 
 import (
+	"bufio"
+	"errors"
+	"net"
 	"net/http"
 
 	web_types "code.gitea.io/gitea/modules/web/types"
@@ -30,6 +33,14 @@ type Response struct {
 	status         int
 	befores        []func(ResponseWriter)
 	beforeExecuted bool
+	hijacker       http.Hijacker
+}
+
+func (r *Response) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	if r.hijacker == nil {
+		return nil, nil, errors.New("http.Hijacker not implemented by underlying http.ResponseWriter")
+	}
+	return r.hijacker.Hijack()
 }
 
 // Write writes bytes to HTTP endpoint
@@ -95,9 +106,11 @@ func WrapResponseWriter(resp http.ResponseWriter) *Response {
 	if v, ok := resp.(*Response); ok {
 		return v
 	}
+	hijacker, _ := resp.(http.Hijacker)
 	return &Response{
 		ResponseWriter: resp,
 		status:         0,
 		befores:        make([]func(ResponseWriter), 0),
+		hijacker:       hijacker,
 	}
 }

services/pubsub/memory.go

@@ -0,0 +1,59 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package pubsub
+
+import (
+	"context"
+	"sync"
+)
+
+type Memory struct {
+	sync.Mutex
+
+	topics map[string]map[*Subscriber]struct{}
+}
+
+// New creates an in-memory publisher.
+func NewMemory() Broker {
+	return &Memory{
+		topics: make(map[string]map[*Subscriber]struct{}),
+	}
+}
+
+func (p *Memory) Publish(_ context.Context, message Message) {
+	p.Lock()
+
+	topic, ok := p.topics[message.Topic]
+	if !ok {
+		p.Unlock()
+		return
+	}
+
+	for s := range topic {
+		go (*s)(message)
+	}
+	p.Unlock()
+}
+
+func (p *Memory) Subscribe(c context.Context, topic string, subscriber Subscriber) {
+	// Subscribe
+	p.Lock()
+	_, ok := p.topics[topic]
+	if !ok {
+		p.topics[topic] = make(map[*Subscriber]struct{})
+	}
+	p.topics[topic][&subscriber] = struct{}{}
+	p.Unlock()
+
+	// Wait for context to be done
+	<-c.Done()
+
+	// Unsubscribe
+	p.Lock()
+	delete(p.topics[topic], &subscriber)
+	if len(p.topics[topic]) == 0 {
+		delete(p.topics, topic)
+	}
+	p.Unlock()
+}

services/pubsub/memory_test.go

@@ -0,0 +1,47 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package pubsub
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPubsub(t *testing.T) {
+	var (
+		wg sync.WaitGroup
+
+		testMessage = Message{
+			Data:  []byte("test"),
+			Topic: "hello-world",
+		}
+	)
+
+	ctx, cancel := context.WithCancelCause(
+		context.Background(),
+	)
+
+	broker := NewMemory()
+	go func() {
+		broker.Subscribe(ctx, "hello-world", func(message Message) { assert.Equal(t, testMessage, message); wg.Done() })
+	}()
+	go func() {
+		broker.Subscribe(ctx, "hello-world", func(_ Message) { wg.Done() })
+	}()
+
+	// Wait a bit for the subscriptions to be registered
+	<-time.After(100 * time.Millisecond)
+
+	wg.Add(2)
+	go func() {
+		broker.Publish(ctx, testMessage)
+	}()
+
+	wg.Wait()
+	cancel(nil)
+}

services/pubsub/types.go

@@ -0,0 +1,23 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package pubsub
+
+import "context"
+
+// Message defines a published message.
+type Message struct {
+	// Data is the actual data in the entry.
+	Data []byte `json:"data"`
+
+	// Topic is the topic of the message.
+	Topic string `json:"topic"`
+}
+
+// Subscriber receives published messages.
+type Subscriber func(Message)
+
+type Broker interface {
+	Publish(c context.Context, message Message)
+	Subscribe(c context.Context, topic string, subscriber Subscriber)
+}

services/websocket/issue_comment_notifier.go

@@ -0,0 +1,48 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"context"
+	"fmt"
+
+	issues_model "code.gitea.io/gitea/models/issues"
+	"code.gitea.io/gitea/models/perm"
+	"code.gitea.io/gitea/models/perm/access"
+	repo_model "code.gitea.io/gitea/models/repo"
+	"code.gitea.io/gitea/models/unit"
+	user_model "code.gitea.io/gitea/models/user"
+	"code.gitea.io/gitea/modules/log"
+	"github.com/olahol/melody"
+)
+
+func (n *websocketNotifier) filterIssueSessions(ctx context.Context, repo *repo_model.Repository, issue *issues_model.Issue) []*melody.Session {
+	return n.filterSessions(func(s *melody.Session, data *sessionData) bool {
+		// if the user is watching the issue, they will get notifications
+		if !data.isOnURL(fmt.Sprintf("/%s/%s/issues/%d", repo.Owner.Name, repo.Name, issue.Index)) {
+			return false
+		}
+
+		// the user will get notifications if they have access to the repos issues
+		hasAccess, err := access.HasAccessUnit(ctx, data.user, repo, unit.TypeIssues, perm.AccessModeRead)
+		if err != nil {
+			log.Error("Failed to check access: %v", err)
+			return false
+		}
+
+		return hasAccess
+	})
+}
+
+func (n *websocketNotifier) DeleteComment(ctx context.Context, doer *user_model.User, c *issues_model.Comment) {
+	sessions := n.filterIssueSessions(ctx, c.Issue.Repo, c.Issue)
+
+	for _, s := range sessions {
+		msg := fmt.Sprintf(htmxRemoveElement, fmt.Sprintf("#%s", c.HashTag()))
+		err := s.Write([]byte(msg))
+		if err != nil {
+			log.Error("Failed to write to session: %v", err)
+		}
+	}
+}

services/websocket/notifier.go

@@ -0,0 +1,53 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/templates"
+	notify_service "code.gitea.io/gitea/services/notify"
+	"code.gitea.io/gitea/services/pubsub"
+	"github.com/olahol/melody"
+)
+
+type websocketNotifier struct {
+	notify_service.NullNotifier
+	m   *melody.Melody
+	rnd *templates.HTMLRender
+}
+
+// NewNotifier create a new webhooksNotifier notifier
+func newNotifier(m *melody.Melody, pubsub pubsub.Broker) notify_service.Notifier {
+	return &websocketNotifier{
+		m:   m,
+		rnd: templates.HTMLRenderer(),
+	}
+}
+
+// htmxAddElementEnd = "<div hx-swap-oob=\"beforebegin:%s\">%s</div>"
+// htmxUpdateElement = "<div hx-swap-oob=\"outerHTML:%s\">%s</div>"
+
+var htmxRemoveElement = "<div hx-swap-oob=\"delete:%s\"></div>"
+
+func (n *websocketNotifier) filterSessions(fn func(*melody.Session, *sessionData) bool) []*melody.Session {
+	sessions, err := n.m.Sessions()
+	if err != nil {
+		log.Error("Failed to get sessions: %v", err)
+		return nil
+	}
+
+	_sessions := make([]*melody.Session, 0, len(sessions))
+	for _, s := range sessions {
+		data, err := getSessionData(s)
+		if err != nil {
+			continue
+		}
+
+		if fn(s, data) {
+			_sessions = append(_sessions, s)
+		}
+	}
+
+	return _sessions
+}

services/websocket/session.go

@@ -0,0 +1,42 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"fmt"
+	"net/url"
+
+	user_model "code.gitea.io/gitea/models/user"
+
+	"github.com/olahol/melody"
+)
+
+type sessionData struct {
+	user  *user_model.User
+	onURL string
+}
+
+func (d *sessionData) isOnURL(_u1 string) bool {
+	if d.onURL == "" {
+		return true
+	}
+
+	u1, _ := url.Parse(d.onURL)
+	u2, _ := url.Parse(_u1)
+	return u1.Path == u2.Path
+}
+
+func getSessionData(s *melody.Session) (*sessionData, error) {
+	_data, ok := s.Get("data")
+	if !ok {
+		return nil, fmt.Errorf("no session data")
+	}
+
+	data, ok := _data.(*sessionData)
+	if !ok {
+		return nil, fmt.Errorf("invalid session data")
+	}
+
+	return data, nil
+}

services/websocket/websocket.go

@@ -0,0 +1,84 @@
+// Copyright 2024 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"code.gitea.io/gitea/modules/json"
+	"code.gitea.io/gitea/services/context"
+	notify_service "code.gitea.io/gitea/services/notify"
+	"code.gitea.io/gitea/services/pubsub"
+
+	"github.com/mitchellh/mapstructure"
+	"github.com/olahol/melody"
+)
+
+var m *melody.Melody
+
+type websocketMessage struct {
+	Action string `json:"action"`
+	Data   any    `json:"data"`
+}
+
+type subscribeMessageData struct {
+	URL string `json:"url"`
+}
+
+func Init() *melody.Melody {
+	m = melody.New()
+	m.HandleConnect(handleConnect)
+	m.HandleMessage(handleMessage)
+	m.HandleDisconnect(handleDisconnect)
+
+	broker := pubsub.NewMemory() // TODO: allow for other pubsub implementations
+	notify_service.RegisterNotifier(newNotifier(m, broker))
+	return m
+}
+
+func handleConnect(s *melody.Session) {
+	ctx := context.GetWebContext(s.Request)
+
+	data := &sessionData{}
+	if ctx.IsSigned {
+		data.user = ctx.Doer
+	}
+
+	s.Set("data", data)
+
+	// TODO: handle logouts
+}
+
+func handleMessage(s *melody.Session, _msg []byte) {
+	data, err := getSessionData(s)
+	if err != nil {
+		return
+	}
+
+	msg := &websocketMessage{}
+	err = json.Unmarshal(_msg, msg)
+	if err != nil {
+		return
+	}
+
+	switch msg.Action {
+	case "subscribe":
+		err := handleSubscribeMessage(data, msg.Data)
+		if err != nil {
+			return
+		}
+	}
+}
+
+func handleSubscribeMessage(data *sessionData, _data any) error {
+	msgData := &subscribeMessageData{}
+	err := mapstructure.Decode(_data, &msgData)
+	if err != nil {
+		return err
+	}
+
+	data.onURL = msgData.URL
+	return nil
+}
+
+func handleDisconnect(s *melody.Session) {
+}

templates/base/head.tmpl

@@ -29,7 +29,7 @@
 	{{template "base/head_style" .}}
 	{{template "custom/header" .}}
 </head>
-<body hx-headers='{"x-csrf-token": "{{.CsrfToken}}"}' hx-swap="outerHTML" hx-ext="morph" hx-push-url="false">
+<body hx-headers='{"x-csrf-token": "{{.CsrfToken}}"}' hx-swap="outerHTML" hx-ext="morph,ws" hx-push-url="false" ws-connect="/-/ws">
 	{{template "custom/body_outer_pre" .}}
 
 	<div class="full height">

web_src/js/htmx.js

@@ -1,5 +1,6 @@
 import * as htmx from 'htmx.org';
 import {showErrorToast} from './modules/toast.js';
+import 'htmx.org/dist/ext/ws.js';
 
 // https://github.com/bigskysoftware/idiomorph#htmx
 import 'idiomorph/dist/idiomorph-ext.js';
@@ -9,13 +10,22 @@ htmx.config.requestClass = 'is-loading';
 htmx.config.scrollIntoViewOnBoost = false;
 
 // https://htmx.org/events/#htmx:sendError
-document.body.addEventListener('htmx:sendError', (event) => {
+document.body.addEventListener('htmx:sendError', (e) => {
   // TODO: add translations
-  showErrorToast(`Network error when calling ${event.detail.requestConfig.path}`);
+  showErrorToast(`Network error when calling ${e.detail.requestConfig.path}`);
 });
 
 // https://htmx.org/events/#htmx:responseError
-document.body.addEventListener('htmx:responseError', (event) => {
+document.body.addEventListener('htmx:responseError', (e) => {
   // TODO: add translations
-  showErrorToast(`Error ${event.detail.xhr.status} when calling ${event.detail.requestConfig.path}`);
+  showErrorToast(`Error ${e.detail.xhr.status} when calling ${e.detail.requestConfig.path}`);
+});
+
+// TODO: move websocket creation to SharedWorker by overriding htmx.createWebSocket
+
+document.body.addEventListener('htmx:wsOpen', (e) => {
+  const socket = e.detail.socketWrapper;
+  socket.send(
+    JSON.stringify({action: 'subscribe', data: {url: window.location.href}})
+  );
 });