enhancement: allow sending multiple userIDs in one SSE event

changelog/unreleased/enhancement-allow-multiple-event-user-ids.md

@@ -0,0 +1,6 @@
+Enhancement: allow sending multiple user ids in one sse event
+
+Sending multiple user ids in one sse event is now possible which reduces the number of sent events.
+
+https://github.com/owncloud/ocis/pull/8379
+https://github.com/cs3org/reva/pull/4501

services/clientlog/pkg/service/service.go

@@ -8,13 +8,15 @@ import (
 	"reflect"
 
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	"go.opentelemetry.io/otel/trace"
+
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
 	"github.com/cs3org/reva/v2/pkg/utils"
+
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/services/clientlog/pkg/config"
-	"go.opentelemetry.io/otel/trace"
 )
 
 // ClientlogService is the service responsible for user activities
@@ -116,22 +118,20 @@ func (cl *ClientlogService) processEvent(event events.Event) {
 	}
 
 	// II) instruct sse service to send the information
-	for _, id := range users {
+	if err := cl.sendSSE(users, evType, data); err != nil {
-		if err := cl.sendSSE(id, evType, data); err != nil {
+		cl.log.Error().Err(err).Interface("userIDs", users).Str("eventid", event.ID).Msg("failed to store event for user")
-			cl.log.Error().Err(err).Str("userID", id).Str("eventid", event.ID).Msg("failed to store event for user")
+		return
-			return
-		}
 	}
 }
 
-func (cl *ClientlogService) sendSSE(userid string, evType string, data interface{}) error {
+func (cl *ClientlogService) sendSSE(userIDs []string, evType string, data interface{}) error {
 	b, err := json.Marshal(data)
 	if err != nil {
 		return err
 	}
 
 	return events.Publish(context.Background(), cl.publisher, events.SendSSE{
-		UserID:  userid,
+		UserIDs: userIDs,
 		Type:    evType,
 		Message: b,
 	})

services/sse/pkg/service/service.go

@@ -3,11 +3,12 @@ package service
 import (
 	"net/http"
 
-	revactx "github.com/cs3org/reva/v2/pkg/ctx"
-	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/go-chi/chi/v5"
 	"github.com/r3labs/sse/v2"
 
+	revactx "github.com/cs3org/reva/v2/pkg/ctx"
+	"github.com/cs3org/reva/v2/pkg/events"
+
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/services/sse/pkg/config"
 )
@@ -51,10 +52,12 @@ func (s SSE) ListenForEvents() {
 		default:
 			s.l.Error().Interface("event", ev).Msg("unhandled event")
 		case events.SendSSE:
-			s.sse.Publish(ev.UserID, &sse.Event{
+			for _, uid := range ev.UserIDs {
-				Event: []byte(ev.Type),
+				s.sse.Publish(uid, &sse.Event{
-				Data:  ev.Message,
+					Event: []byte(ev.Type),
-			})
+					Data:  ev.Message,
+				})
+			}
 		}
 	}
 }

services/userlog/pkg/service/service.go

@@ -10,10 +10,14 @@ import (
 
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	"github.com/go-chi/chi/v5"
+	micrometadata "go-micro.dev/v4/metadata"
+	"go-micro.dev/v4/store"
+	"go.opentelemetry.io/otel/trace"
+
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/cs3org/reva/v2/pkg/utils"
-	"github.com/go-chi/chi/v5"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
 	"github.com/owncloud/ocis/v2/ocis-pkg/middleware"
 	"github.com/owncloud/ocis/v2/ocis-pkg/roles"
@@ -22,9 +26,6 @@ import (
 	settingssvc "github.com/owncloud/ocis/v2/protogen/gen/ocis/services/settings/v0"
 	"github.com/owncloud/ocis/v2/services/settings/pkg/store/defaults"
 	"github.com/owncloud/ocis/v2/services/userlog/pkg/config"
-	micrometadata "go-micro.dev/v4/metadata"
-	"go-micro.dev/v4/store"
-	"go.opentelemetry.io/otel/trace"
 )
 
 // UserlogService is the service responsible for user activities
@@ -348,7 +349,7 @@ func (ul *UserlogService) sendSSE(ctx context.Context, userid string, event even
 	}
 
 	return events.Publish(context.Background(), ul.publisher, events.SendSSE{
-		UserID:  userid,
+		UserIDs: []string{userid},
 		Type:    "userlog-notification",
 		Message: b,
 	})