feat: move the collaboration mentioning api to graph

services/collaboration/pkg/collaboration/collaboration.go

@@ -14,8 +14,7 @@ import (
 type Permission string
 
 const (
-	PermissionCollaborationManageFonts         Permission = "Collaboration.Fonts.Manage"
-	PermissionCollaborationPublishNotification Permission = "Collaboration.Notification.Publish"
+	PermissionCollaborationManageFonts Permission = "Collaboration.Fonts.Manage"
 )
 
 func CheckPermissions(gatewayClient gateway.GatewayAPIClient, ctx context.Context, permission Permission) (*userpb.User, bool, error) {

services/collaboration/pkg/command/server.go

@@ -8,7 +8,6 @@ import (
 	"os/signal"
 	"time"
 
-	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/opencloud-eu/reva/v2/pkg/store"
 	"github.com/spf13/afero"
@@ -17,7 +16,6 @@ import (
 	microstore "go-micro.dev/v4/store"
 
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
-	"github.com/opencloud-eu/opencloud/pkg/generators"
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/pkg/registry"
 	"github.com/opencloud-eu/opencloud/pkg/runner"
@@ -28,7 +26,6 @@ import (
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/connector"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/font"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/helpers"
-	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/notification"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/server/debug"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/server/grpc"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/server/http"
@@ -174,32 +171,8 @@ func Server(cfg *config.Config) *cobra.Command {
 				}
 			}
 
-			var optionalHTTPServerOptions []http.Option
-			var notificationService notification.Service
-			if cfg.Events.Endpoint == "" {
-				logger.Warn().Msg("Events endpoint is not configured, notifications from the collaboration service will not work")
-			} else {
-				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
-				natsStream, err := stream.NatsFromConfig(connName, true, stream.NatsConfig(cfg.Events))
-				if err != nil {
-					return err
-				}
-				notificationService, err = notification.NewService(
-					notification.ServiceOptions{}.
-						WithLogger(logger).
-						WithGatewaySelector(gatewaySelector).
-						WithEventPublisher(natsStream).
-						WithMachineAuthAPIKey(cfg.MachineAuthAPIKey),
-				)
-				if err != nil {
-					return err
-				}
-
-				optionalHTTPServerOptions = append(optionalHTTPServerOptions, http.NotificationService(&notificationService))
-			}
-
 			// start HTTP server
-			httpServer, err := http.Server(append([]http.Option{
+			httpServer, err := http.Server(
 				http.Adapter(connector.NewHttpAdapter(gatewaySelector, cfg, st, selector.NewSelector(selector.Registry(registry.GetRegistry())))),
 				http.Logger(logger),
 				http.Config(cfg),
@@ -207,7 +180,7 @@ func Server(cfg *config.Config) *cobra.Command {
 				http.TracerProvider(traceProvider),
 				http.Store(st),
 				http.FontService(fontService),
-			}, optionalHTTPServerOptions...)...)
+			)
 			if err != nil {
 				logger.Info().Err(err).Str("transport", "http").Msg("Failed to initialize server")
 				return err

services/collaboration/pkg/config/config.go

@@ -14,7 +14,6 @@ type Config struct {
 	App     App     `yaml:"app"`
 	Font    Font    `yaml:"font"`
 	Store   Store   `yaml:"store"`
-	Events  Events  `yaml:"events"`
 
 	TokenManager *TokenManager `yaml:"token_manager"`
 
@@ -28,6 +27,4 @@ type Config struct {
 	Debug    Debug  `yaml:"debug"`
 
 	Context context.Context `yaml:"-"`
-
-	MachineAuthAPIKey string `yaml:"machine_auth_api_key" env:"OC_MACHINE_AUTH_API_KEY;COLLABORATION_MACHINE_AUTH_API_KEY" desc:"The machine auth API key used to validate internal requests necessary to access resources from other services." introductionVersion:"7.3.0"`
 }

services/collaboration/pkg/config/defaults/defaultconfig.go

@@ -39,10 +39,6 @@ func DefaultConfig() *config.Config {
 			AssetPath:   filepath.Join(defaults.BaseDataPath(), "collaboration/fonts"),
 			PreviewText: "OpenCloud",
 		},
-		Events: config.Events{
-			Endpoint: "127.0.0.1:9233",
-			Cluster:  "opencloud-cluster",
-		},
 		Store: config.Store{
 			Store:    "nats-js-kv",
 			Nodes:    []string{"127.0.0.1:9233"},
@@ -96,10 +92,6 @@ func EnsureDefaults(cfg *config.Config) {
 	if cfg.CS3Api.GRPCClientTLS == nil && cfg.Commons != nil {
 		cfg.CS3Api.GRPCClientTLS = structs.CopyOrZeroValue(cfg.Commons.GRPCClientTLS)
 	}
-
-	if cfg.MachineAuthAPIKey == "" && cfg.Commons != nil && cfg.Commons.MachineAuthAPIKey != "" {
-		cfg.MachineAuthAPIKey = cfg.Commons.MachineAuthAPIKey
-	}
 }
 
 // Sanitize sanitized the configuration

services/collaboration/pkg/config/event.go

@@ -1,12 +0,0 @@
-package config
-
-// Events combines the configuration options for the event bus.
-type Events struct {
-	Endpoint             string `yaml:"endpoint" env:"OC_EVENTS_ENDPOINT;COLLABORATION_EVENTS_ENDPOINT" desc:"The address of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture." introductionVersion:"7.3.0"`
-	Cluster              string `yaml:"cluster" env:"OC_EVENTS_CLUSTER;COLLABORATION_EVENTS_CLUSTER" desc:"The clusterID of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture. Mandatory when using NATS as event system." introductionVersion:"7.3.0"`
-	TLSInsecure          bool   `yaml:"tls_insecure" env:"OC_INSECURE;OC_EVENTS_TLS_INSECURE;COLLABORATION_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates." introductionVersion:"7.3.0"`
-	TLSRootCACertificate string `yaml:"tls_root_ca_certificate" env:"OC_EVENTS_TLS_ROOT_CA_CERTIFICATE;COLLABORATION_EVENTS_TLS_ROOT_CA_CERTIFICATE" desc:"The root CA certificate used to validate the server's TLS certificate. If provided COLLABORATION_EVENTS_TLS_INSECURE will be seen as false." introductionVersion:"7.3.0"`
-	EnableTLS            bool   `yaml:"enable_tls" env:"OC_EVENTS_ENABLE_TLS;COLLABORATION_EVENTS_ENABLE_TLS" desc:"Enable TLS for the connection to the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"7.3.0"`
-	AuthUsername         string `yaml:"username" env:"OC_EVENTS_AUTH_USERNAME;COLLABORATION_EVENTS_AUTH_USERNAME" desc:"The username to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"7.3.0"`
-	AuthPassword         string `yaml:"password" env:"OC_EVENTS_AUTH_PASSWORD;COLLABORATION_EVENTS_AUTH_PASSWORD" desc:"The password to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"7.3.0"`
-}

services/collaboration/pkg/notification/notification.go

@@ -1,10 +0,0 @@
-package notification
-
-import (
-	"github.com/go-playground/validator/v10"
-)
-
-var validate = validator.New(
-	validator.WithPrivateFieldValidation(),
-	validator.WithRequiredStructEnabled(),
-)

services/collaboration/pkg/notification/service.go

@@ -1,175 +0,0 @@
-package notification
-
-import (
-	"encoding/json"
-	"io"
-	"net/http"
-	"time"
-
-	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
-	rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
-	storageprovider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
-	revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
-	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
-	"google.golang.org/grpc/metadata"
-
-	ocEvents "github.com/opencloud-eu/opencloud/pkg/events"
-	"github.com/opencloud-eu/opencloud/pkg/log"
-	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/collaboration"
-)
-
-type ServiceOptions struct {
-	logger            log.Logger                                `validate:"required"`
-	eventPublisher    events.Publisher                          `validate:"required"`
-	gatewaySelector   pool.Selectable[gateway.GatewayAPIClient] `validate:"required"`
-	machineAuthAPIKey string                                    `validate:"required,min=1"`
-}
-
-func (o ServiceOptions) WithLogger(logger log.Logger) ServiceOptions {
-	o.logger = logger
-	return o
-}
-
-func (o ServiceOptions) WithEventPublisher(eventPublisher events.Publisher) ServiceOptions {
-	o.eventPublisher = eventPublisher
-	return o
-}
-
-func (o ServiceOptions) WithMachineAuthAPIKey(key string) ServiceOptions {
-	o.machineAuthAPIKey = key
-	return o
-}
-
-func (o ServiceOptions) WithGatewaySelector(gws pool.Selectable[gateway.GatewayAPIClient]) ServiceOptions {
-	o.gatewaySelector = gws
-	return o
-}
-
-type Service struct {
-	log               log.Logger
-	eventPublisher    events.Publisher
-	gatewaySelector   pool.Selectable[gateway.GatewayAPIClient]
-	machineAuthAPIKey string
-}
-
-func NewService(options ServiceOptions) (Service, error) {
-	if err := validate.Struct(options); err != nil {
-		return Service{}, err
-	}
-
-	return Service{
-		log:               options.logger,
-		eventPublisher:    options.eventPublisher,
-		gatewaySelector:   options.gatewaySelector,
-		machineAuthAPIKey: options.machineAuthAPIKey,
-	}, nil
-}
-
-func (s Service) HandleNotification(w http.ResponseWriter, r *http.Request) {
-	ctx := metadata.AppendToOutgoingContext(r.Context(), revactx.TokenHeader, r.Header.Get(revactx.TokenHeader))
-
-	gatewayClient, err := s.gatewaySelector.Next()
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	requestUser, canManage, err := collaboration.CheckPermissions(gatewayClient, ctx, collaboration.PermissionCollaborationPublishNotification)
-	switch {
-	case err != nil:
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	case !canManage:
-		w.WriteHeader(http.StatusForbidden)
-		return
-	}
-
-	defer func() { _ = r.Body.Close() }()
-	body, err := io.ReadAll(r.Body)
-	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
-	var data = struct {
-		UserIDs []string `json:"userIDs" validate:"required,min=1"`
-		FileID  string   `json:"fileID" validate:"required"`
-	}{}
-	if err := json.Unmarshal(body, &data); err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
-	if err := validate.Struct(data); err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
-	resourceID, err := storagespace.ParseID(data.FileID)
-	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
-	ref := &storageprovider.Reference{ResourceId: &resourceID}
-
-	// accept the mention only if the requester can see the file himself
-	statResponse, err := gatewayClient.Stat(ctx, &storageprovider.StatRequest{Ref: ref})
-	switch {
-	case err != nil:
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	case statResponse.GetStatus().GetCode() != rpcv1beta1.Code_CODE_OK:
-		w.WriteHeader(http.StatusNotFound)
-		return
-	}
-
-	event := ocEvents.ResourceMention{
-		Executant: requestUser.GetId(),
-		Ref:       &storageprovider.Reference{ResourceId: statResponse.GetInfo().GetId()},
-		Timestamp: time.Now(),
-	}
-
-	seen := make(map[string]struct{}, len(data.UserIDs))
-	for _, userID := range data.UserIDs {
-		if _, ok := seen[userID]; ok {
-			continue
-		}
-		seen[userID] = struct{}{}
-
-		authResponse, err := gatewayClient.Authenticate(r.Context(), &gateway.AuthenticateRequest{
-			Type:         "machine",
-			ClientId:     "userid:" + userID,
-			ClientSecret: s.machineAuthAPIKey,
-		})
-		if err != nil || authResponse.GetStatus().GetCode() != rpcv1beta1.Code_CODE_OK {
-			s.log.Debug().Str("userID", userID).Msg("could not authenticate the mentioned user, skipping")
-			continue
-		}
-
-		statResponse, err := gatewayClient.Stat(
-			metadata.AppendToOutgoingContext(r.Context(), revactx.TokenHeader, authResponse.GetToken()),
-			&storageprovider.StatRequest{Ref: ref},
-		)
-		if err != nil || statResponse.GetStatus().GetCode() != rpcv1beta1.Code_CODE_OK {
-			s.log.Debug().Str("userID", userID).Msg("mentioned user has no access to the resource, skipping")
-			continue
-		}
-
-		event.UserIDs = append(event.UserIDs, authResponse.GetUser().GetId())
-	}
-
-	if len(event.UserIDs) == 0 {
-		w.WriteHeader(http.StatusOK)
-		return
-	}
-
-	if err := events.Publish(r.Context(), s.eventPublisher, event); err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	w.WriteHeader(http.StatusOK)
-}

services/collaboration/pkg/notification/service_test.go

@@ -1,197 +0,0 @@
-package notification_test
-
-import (
-	"context"
-	"net/http"
-	"net/http/httptest"
-	"strings"
-	"testing"
-
-	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
-	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
-	cs3Permissions "github.com/cs3org/go-cs3apis/cs3/permissions/v1beta1"
-	cs3RPC "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
-	storageprovider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
-	revaCtx "github.com/opencloud-eu/reva/v2/pkg/ctx"
-	revaEvents "github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
-	cs3mocks "github.com/opencloud-eu/reva/v2/tests/cs3mocks/mocks"
-	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
-	"go-micro.dev/v4/events"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/metadata"
-
-	ocEvents "github.com/opencloud-eu/opencloud/pkg/events"
-	"github.com/opencloud-eu/opencloud/pkg/log"
-	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/notification"
-)
-
-const (
-	requesterToken = "requester-token"
-	fileID         = "storage$space!opaque"
-)
-
-type publisher struct {
-	events []any
-}
-
-func (p *publisher) Publish(_ string, event any, _ ...events.PublishOption) error {
-	p.events = append(p.events, event)
-	return nil
-}
-
-func (p *publisher) mentions() []ocEvents.ResourceMention {
-	var mentions []ocEvents.ResourceMention
-	for _, event := range p.events {
-		if mention, ok := event.(ocEvents.ResourceMention); ok {
-			mentions = append(mentions, mention)
-		}
-	}
-
-	return mentions
-}
-
-func token(ctx context.Context) string {
-	md, ok := metadata.FromOutgoingContext(ctx)
-	if !ok {
-		return ""
-	}
-
-	return strings.Join(md.Get(revaCtx.TokenHeader), "")
-}
-
-func statOK() *storageprovider.StatResponse {
-	return &storageprovider.StatResponse{
-		Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_OK},
-		Info: &storageprovider.ResourceInfo{
-			Id: &storageprovider.ResourceId{StorageId: "storage", SpaceId: "space", OpaqueId: "opaque"},
-		},
-	}
-}
-
-func newService(t *testing.T, p revaEvents.Publisher, canStat ...string) notification.Service {
-	t.Helper()
-
-	gatewayAPIClient := &cs3mocks.GatewayAPIClient{}
-	gatewayAPIClient.On("CheckPermission", mock.Anything, mock.Anything).Return(
-		&cs3Permissions.CheckPermissionResponse{Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_OK}}, nil)
-
-	gatewayAPIClient.On("Authenticate", mock.Anything, mock.Anything).Return(
-		func(_ context.Context, req *gateway.AuthenticateRequest, _ ...grpc.CallOption) *gateway.AuthenticateResponse {
-			userID := strings.TrimPrefix(req.GetClientId(), "userid:")
-
-			return &gateway.AuthenticateResponse{
-				Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_OK},
-				Token:  userID + "-token",
-				User:   &userpb.User{Id: &userpb.UserId{OpaqueId: userID}},
-			}
-		}, nil)
-
-	allowed := map[string]struct{}{}
-	for _, name := range canStat {
-		allowed[name+"-token"] = struct{}{}
-	}
-
-	gatewayAPIClient.On("Stat", mock.Anything, mock.Anything).Return(
-		func(ctx context.Context, _ *storageprovider.StatRequest, _ ...grpc.CallOption) *storageprovider.StatResponse {
-			if _, ok := allowed[token(ctx)]; !ok {
-				return &storageprovider.StatResponse{Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_NOT_FOUND}}
-			}
-
-			return statOK()
-		}, nil)
-
-	svc, err := notification.NewService(
-		notification.ServiceOptions{}.
-			WithLogger(log.NopLogger()).
-			WithEventPublisher(p).
-			WithMachineAuthAPIKey("machine-auth-api-key").
-			WithGatewaySelector(pool.GetSelector[gateway.GatewayAPIClient](
-				"GatewaySelector"+t.Name(),
-				"eu.opencloud.api.gateway",
-				func(cc grpc.ClientConnInterface) gateway.GatewayAPIClient {
-					return gatewayAPIClient
-				},
-			)),
-	)
-	require.NoError(t, err)
-
-	return svc
-}
-
-func newRequest(body string) *http.Request {
-	req := httptest.NewRequest(http.MethodPost, "/collaboration/notify", strings.NewReader(body))
-	req.Header.Set(revaCtx.TokenHeader, requesterToken)
-
-	return req.WithContext(revaCtx.ContextSetUser(req.Context(), &userpb.User{
-		Id: &userpb.UserId{OpaqueId: "requester"},
-	}))
-}
-
-func TestHandleNotification_RequesterWithoutAccess(t *testing.T) {
-	p := &publisher{}
-	svc := newService(t, p, "alice")
-
-	resp := httptest.NewRecorder()
-	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice"],"fileID":"`+fileID+`"}`))
-
-	require.Equal(t, http.StatusNotFound, resp.Code)
-	require.Empty(t, p.mentions())
-}
-
-func TestHandleNotification_SkipsRecipientWithoutAccess(t *testing.T) {
-	p := &publisher{}
-	svc := newService(t, p, "requester", "alice")
-
-	resp := httptest.NewRecorder()
-	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice","carol"],"fileID":"`+fileID+`"}`))
-
-	require.Equal(t, http.StatusOK, resp.Code)
-	require.Len(t, p.mentions(), 1)
-	require.Len(t, p.mentions()[0].UserIDs, 1)
-	require.Equal(t, "alice", p.mentions()[0].UserIDs[0].GetOpaqueId())
-}
-
-func TestHandleNotification_NoRecipientLeft(t *testing.T) {
-	p := &publisher{}
-	svc := newService(t, p, "requester")
-
-	resp := httptest.NewRecorder()
-	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice"],"fileID":"`+fileID+`"}`))
-
-	require.Equal(t, http.StatusOK, resp.Code)
-	require.Empty(t, p.mentions())
-}
-
-func TestHandleNotification_DedupsUserIDs(t *testing.T) {
-	p := &publisher{}
-	svc := newService(t, p, "requester", "alice")
-
-	resp := httptest.NewRecorder()
-	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice","alice","alice"],"fileID":"`+fileID+`"}`))
-
-	require.Equal(t, http.StatusOK, resp.Code)
-	require.Len(t, p.mentions(), 1)
-	require.Len(t, p.mentions()[0].UserIDs, 1)
-}
-
-func TestHandleNotification_BadRequest(t *testing.T) {
-	for name, body := range map[string]string{
-		"no json":       `not json`,
-		"no userIDs":    `{"fileID":"` + fileID + `"}`,
-		"empty userIDs": `{"userIDs":[],"fileID":"` + fileID + `"}`,
-		"no fileID":     `{"userIDs":["alice"]}`,
-	} {
-		t.Run(name, func(t *testing.T) {
-			p := &publisher{}
-			svc := newService(t, p, "requester", "alice")
-
-			resp := httptest.NewRecorder()
-			svc.HandleNotification(resp, newRequest(body))
-
-			require.Equal(t, http.StatusBadRequest, resp.Code)
-			require.Empty(t, p.mentions())
-		})
-	}
-}

services/collaboration/pkg/server/http/option.go

@@ -10,7 +10,6 @@ import (
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/connector"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/font"
-	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/notification"
 )
 
 // Option defines a single option function.
@@ -18,14 +17,13 @@ type Option func(o *Options)
 
 // Options define the available options for this package.
 type Options struct {
-	Adapter             *connector.HttpAdapter
-	Logger              log.Logger
-	Context             context.Context
-	Config              *config.Config
-	TracerProvider      trace.TracerProvider
-	Store               microstore.Store
-	FontService         font.Service
-	NotificationService *notification.Service
+	Adapter        *connector.HttpAdapter
+	Logger         log.Logger
+	Context        context.Context
+	Config         *config.Config
+	TracerProvider trace.TracerProvider
+	Store          microstore.Store
+	FontService    font.Service
 }
 
 // newOptions initializes the available default options.
@@ -87,10 +85,3 @@ func FontService(val font.Service) Option {
 		o.FontService = val
 	}
 }
-
-// NotificationService provides a function to set the NotificationService option
-func NotificationService(val *notification.Service) Option {
-	return func(o *Options) {
-		o.NotificationService = val
-	}
-}

services/collaboration/pkg/server/http/server.go

@@ -22,10 +22,6 @@ import (
 func Server(opts ...Option) (http.Service, error) {
 	options := newOptions(opts...)
 
-	if options.NotificationService == nil {
-		options.Logger.Warn().Msg("running without notification service: no notifications will be sent, set the events endpoint to enable them")
-	}
-
 	service, err := http.NewService(
 		http.TLSConfig(options.Config.HTTP.TLS),
 		http.Logger(options.Logger),
@@ -100,7 +96,6 @@ func Server(opts ...Option) (http.Service, error) {
 // prepareRoutes will prepare all the implemented routes
 func prepareRoutes(r *chi.Mux, options Options) {
 	fontService := options.FontService
-	notificationService := options.NotificationService
 	adapter := options.Adapter
 	logger := options.Logger
 	// prepare basic logger for the request
@@ -232,11 +227,5 @@ func prepareRoutes(r *chi.Mux, options Options) {
 				r.Delete("/{id}", fontService.DeleteFont)
 			})
 		})
-
-		if notificationService != nil { // optional
-			r.With(auth).Route("/notify", func(r chi.Router) {
-				r.Post("/", notificationService.HandleNotification)
-			})
-		}
 	})
 }

services/graph/pkg/config/config.go

@@ -36,6 +36,8 @@ type Config struct {
 	Keycloak       Keycloak       `yaml:"keycloak"`
 	ServiceAccount ServiceAccount `yaml:"service_account"`
 
+	MachineAuthAPIKey string `yaml:"machine_auth_api_key" env:"OC_MACHINE_AUTH_API_KEY;GRAPH_MACHINE_AUTH_API_KEY" desc:"Machine auth API key used to validate internal requests necessary for the access to resources from other services." introductionVersion:"7.5.0" mask:"password"`
+
 	Context context.Context `yaml:"-"`
 
 	Metadata Metadata `yaml:"metadata_config"`

services/graph/pkg/config/defaults/defaultconfig.go

@@ -170,6 +170,10 @@ func EnsureDefaults(cfg *config.Config) {
 		cfg.HTTP.TLS = cfg.Commons.HTTPServiceTLS
 	}
 
+	if cfg.MachineAuthAPIKey == "" && cfg.Commons != nil && cfg.Commons.MachineAuthAPIKey != "" {
+		cfg.MachineAuthAPIKey = cfg.Commons.MachineAuthAPIKey
+	}
+
 	if cfg.Identity.LDAP.GroupCreateBaseDN == "" {
 		cfg.Identity.LDAP.GroupCreateBaseDN = cfg.Identity.LDAP.GroupBaseDN
 	}

services/graph/pkg/service/v0/service.go

@@ -341,6 +341,7 @@ func NewService(opts ...Option) (Graph, error) { //nolint:maintidx
 					r.Get("/", svc.GetUser)
 					r.Get("/drive", svc.GetUserDrive)
 					r.Post("/exportPersonalData", svc.ExportPersonalData)
+					r.Post("/teamwork/sendActivityNotification", svc.SendActivityNotification)
 					r.Route("/photo/$value", func(r chi.Router) {
 						r.Get("/", usersUserProfilePhotoApi.GetProfilePhoto(GetSlugValue("userID")))
 					})

services/graph/pkg/service/v0/teamwork.go

@@ -0,0 +1,155 @@
+package svc
+
+import (
+	"net/http"
+	"net/url"
+	"time"
+
+	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
+	storageprovider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/go-chi/chi/v5"
+	revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
+	"google.golang.org/grpc/metadata"
+
+	ocEvents "github.com/opencloud-eu/opencloud/pkg/events"
+	settingssvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/errorcode"
+	settingsServiceExt "github.com/opencloud-eu/opencloud/services/settings/pkg/store/defaults"
+)
+
+const (
+	// WebOfficeAppID identifies the app a notification comes from
+	WebOfficeAppID = "8d1c9c88-9e2c-4d0b-9a1e-6a9de1cb9d3c"
+
+	_activityTypeMentioned = "mentioned"
+	_topicSourceText       = "text"
+)
+
+type activityTopic struct {
+	Source string `json:"source"`
+	Value  string `json:"value"`
+}
+
+type activityNotification struct {
+	Topic        activityTopic `json:"topic"`
+	ActivityType string        `json:"activityType"`
+	TeamsAppID   string        `json:"teamsAppId"`
+}
+
+// SendActivityNotification tells a user that they were named on a resource.
+func (g Graph) SendActivityNotification(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	if g.eventsPublisher == nil {
+		g.logger.Error().Msg("no events publisher configured, activity notifications cannot be delivered")
+		errorcode.ServiceNotAvailable.Render(w, r, http.StatusServiceUnavailable, "activity notifications are not available")
+		return
+	}
+
+	executant, ok := revactx.ContextGetUser(ctx)
+	if !ok {
+		errorcode.GeneralException.Render(w, r, http.StatusUnauthorized, "user not found in context")
+		return
+	}
+
+	if _, err := g.permissionsService.GetPermissionByID(ctx, &settingssvc.GetPermissionByIDRequest{
+		PermissionId: settingsServiceExt.CollaborationPublishNotificationPermission(0).Id,
+	}); err != nil {
+		g.logger.Debug().Err(err).Msg("user is not allowed to publish activity notifications")
+		errorcode.AccessDenied.Render(w, r, http.StatusForbidden, "not allowed to publish activity notifications")
+		return
+	}
+
+	userID, err := url.PathUnescape(chi.URLParam(r, "userID"))
+	if err != nil || userID == "" {
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "invalid user id")
+		return
+	}
+
+	notification := activityNotification{}
+	if err := StrictJSONUnmarshal(r.Body, &notification); err != nil {
+		g.logger.Debug().Err(err).Msg("could not parse the activity notification")
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "invalid request body")
+		return
+	}
+
+	switch {
+	case notification.ActivityType != _activityTypeMentioned:
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "unsupported activityType")
+		return
+	case notification.TeamsAppID != WebOfficeAppID:
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "unknown teamsAppId")
+		return
+	case notification.Topic.Source != _topicSourceText:
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "unsupported topic source")
+		return
+	}
+
+	itemID, err := storagespace.ParseID(notification.Topic.Value)
+	if err != nil || itemID.GetStorageId() == "" || itemID.GetSpaceId() == "" || itemID.GetOpaqueId() == "" {
+		g.logger.Debug().Err(err).Msg("could not parse the topic value")
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "the topic value is no resource id")
+		return
+	}
+
+	gatewayClient, err := g.gatewaySelector.Next()
+	if err != nil {
+		g.logger.Error().Err(err).Msg("could not select next gateway client")
+		errorcode.ServiceNotAvailable.Render(w, r, http.StatusServiceUnavailable, "could not select next gateway client")
+		return
+	}
+
+	ref := &storageprovider.Reference{ResourceId: &itemID}
+
+	// the notification is valid if the caller can see the resource.
+	statResponse, err := gatewayClient.Stat(ctx, &storageprovider.StatRequest{Ref: ref})
+	switch {
+	case err != nil:
+		g.logger.Error().Err(err).Msg("could not stat item")
+		errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, "could not stat item")
+		return
+	case statResponse.GetStatus().GetCode() != rpc.Code_CODE_OK:
+		errorcode.ItemNotFound.Render(w, r, http.StatusNotFound, "item not found")
+		return
+	}
+
+	authResponse, err := gatewayClient.Authenticate(ctx, &gateway.AuthenticateRequest{
+		Type:         "machine",
+		ClientId:     "userid:" + userID,
+		ClientSecret: g.config.MachineAuthAPIKey,
+	})
+	if err != nil || authResponse.GetStatus().GetCode() != rpc.Code_CODE_OK {
+		g.logger.Debug().Str("userID", userID).Msg("could not authenticate the recipient")
+		errorcode.ItemNotFound.Render(w, r, http.StatusNotFound, "recipient not found")
+		return
+	}
+
+	recipientStat, err := gatewayClient.Stat(
+		metadata.AppendToOutgoingContext(ctx, revactx.TokenHeader, authResponse.GetToken()),
+		&storageprovider.StatRequest{Ref: ref},
+	)
+	if err != nil || recipientStat.GetStatus().GetCode() != rpc.Code_CODE_OK {
+		g.logger.Debug().Str("userID", userID).Msg("the recipient has no access to the item")
+		errorcode.AccessDenied.Render(w, r, http.StatusForbidden, "the recipient cannot access the item")
+		return
+	}
+
+	event := ocEvents.ResourceMention{
+		Executant: executant.GetId(),
+		UserIDs:   []*userpb.UserId{authResponse.GetUser().GetId()},
+		Ref:       &storageprovider.Reference{ResourceId: statResponse.GetInfo().GetId()},
+		Timestamp: time.Now(),
+	}
+
+	if err := events.Publish(ctx, g.eventsPublisher, event); err != nil {
+		g.logger.Error().Err(err).Msg("could not publish the activity notification")
+		errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, "could not publish the activity notification")
+		return
+	}
+
+	w.WriteHeader(http.StatusAccepted)
+}

services/graph/pkg/service/v0/teamwork_test.go

@@ -0,0 +1,248 @@
+package svc_test
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+
+	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/go-chi/chi/v5"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
+	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/status"
+	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
+	cs3mocks "github.com/opencloud-eu/reva/v2/tests/cs3mocks/mocks"
+	"github.com/stretchr/testify/mock"
+	"google.golang.org/grpc"
+	grpcmetadata "google.golang.org/grpc/metadata"
+
+	ocEvents "github.com/opencloud-eu/opencloud/pkg/events"
+	"github.com/opencloud-eu/opencloud/pkg/shared"
+	settings "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
+	"github.com/opencloud-eu/opencloud/services/graph/mocks"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/config/defaults"
+	service "github.com/opencloud-eu/opencloud/services/graph/pkg/service/v0"
+)
+
+var _ = Describe("SendActivityNotification", func() {
+	var (
+		svc               service.Graph
+		cfg               *config.Config
+		gatewayClient     *cs3mocks.GatewayAPIClient
+		eventsPublisher   mocks.Publisher
+		permissionService *mocks.Permissions
+
+		currentUser = &userpb.User{Id: &userpb.UserId{OpaqueId: "executant"}}
+
+		mention = `{"topic":{"source":"text","value":"storage$space!item"},` +
+			`"activityType":"mentioned","teamsAppId":"` + service.WebOfficeAppID + `"}`
+	)
+
+	// the recipient is the user in the path, so every request names one
+	request := func(userID, body string) *http.Request {
+		ctx := revactx.ContextSetUser(context.Background(), currentUser)
+
+		rctx := chi.NewRouteContext()
+		rctx.URLParams.Add("userID", userID)
+
+		return httptest.NewRequest(
+			http.MethodPost,
+			"/graph/v1.0/users/"+userID+"/teamwork/sendActivityNotification",
+			strings.NewReader(body),
+		).WithContext(context.WithValue(ctx, chi.RouteCtxKey, rctx))
+	}
+
+	mentions := func() []ocEvents.ResourceMention {
+		var mentions []ocEvents.ResourceMention
+		for _, call := range eventsPublisher.Calls {
+			if mention, ok := call.Arguments[1].(ocEvents.ResourceMention); ok {
+				mentions = append(mentions, mention)
+			}
+		}
+
+		return mentions
+	}
+
+	statAs := func(tokens ...string) {
+		allowed := make(map[string]struct{}, len(tokens))
+		for _, token := range tokens {
+			allowed[token] = struct{}{}
+		}
+
+		gatewayClient.On("Stat", mock.Anything, mock.Anything).Return(
+			func(ctx context.Context, _ *provider.StatRequest, _ ...grpc.CallOption) *provider.StatResponse {
+				var token string
+				if md, ok := grpcmetadata.FromOutgoingContext(ctx); ok {
+					token = strings.Join(md.Get(revactx.TokenHeader), "")
+				}
+
+				if _, ok := allowed[token]; !ok {
+					return &provider.StatResponse{Status: status.NewNotFound(ctx, "not found")}
+				}
+
+				return &provider.StatResponse{
+					Status: status.NewOK(ctx),
+					Info: &provider.ResourceInfo{
+						Id: &provider.ResourceId{StorageId: "storage", SpaceId: "space", OpaqueId: "item"},
+					},
+				}
+			}, nil)
+	}
+
+	BeforeEach(func() {
+		eventsPublisher = mocks.Publisher{}
+		eventsPublisher.On("Publish", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+		permissionService = &mocks.Permissions{}
+		permissionService.On("GetPermissionByID", mock.Anything, mock.Anything).
+			Return(&settings.GetPermissionByIDResponse{}, nil)
+
+		pool.RemoveSelector("GatewaySelector" + "eu.opencloud.api.gateway")
+		gatewayClient = &cs3mocks.GatewayAPIClient{}
+		gatewaySelector := pool.GetSelector[gateway.GatewayAPIClient](
+			"GatewaySelector",
+			"eu.opencloud.api.gateway",
+			func(cc grpc.ClientConnInterface) gateway.GatewayAPIClient {
+				return gatewayClient
+			},
+		)
+
+		gatewayClient.On("Authenticate", mock.Anything, mock.Anything).Return(
+			func(_ context.Context, req *gateway.AuthenticateRequest, _ ...grpc.CallOption) *gateway.AuthenticateResponse {
+				userID := strings.TrimPrefix(req.GetClientId(), "userid:")
+				if userID == "nobody" {
+					return &gateway.AuthenticateResponse{Status: status.NewNotFound(context.Background(), "not found")}
+				}
+
+				return &gateway.AuthenticateResponse{
+					Status: status.NewOK(context.Background()),
+					Token:  userID + "-token",
+					User:   &userpb.User{Id: &userpb.UserId{OpaqueId: userID}},
+				}
+			}, nil)
+
+		cfg = defaults.FullDefaultConfig()
+		cfg.Identity.LDAP.CACert = ""
+		cfg.TokenManager.JWTSecret = "loremipsum"
+		cfg.Commons = &shared.Commons{}
+		cfg.GRPCClientTLS = &shared.GRPCClientTLS{}
+		cfg.MachineAuthAPIKey = "machine-auth-api-key"
+
+		var err error
+		svc, err = service.NewService(
+			service.Config(cfg),
+			service.WithGatewaySelector(gatewaySelector),
+			service.EventsPublisher(&eventsPublisher),
+			service.PermissionService(permissionService),
+		)
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	It("notifies the user from the path", func() {
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusAccepted))
+		Expect(mentions()).To(HaveLen(1))
+		Expect(mentions()[0].Executant.GetOpaqueId()).To(Equal("executant"))
+		Expect(mentions()[0].UserIDs).To(HaveLen(1))
+		Expect(mentions()[0].UserIDs[0].GetOpaqueId()).To(Equal("alice"))
+		Expect(mentions()[0].Ref.GetResourceId().GetOpaqueId()).To(Equal("item"))
+	})
+
+	It("is unavailable without an events publisher", func() {
+		var err error
+		svc, err = service.NewService(
+			service.Config(cfg),
+			service.WithGatewaySelector(pool.GetSelector[gateway.GatewayAPIClient](
+				"GatewaySelector",
+				"eu.opencloud.api.gateway",
+				func(cc grpc.ClientConnInterface) gateway.GatewayAPIClient {
+					return gatewayClient
+				},
+			)),
+			service.PermissionService(permissionService),
+		)
+		Expect(err).ToNot(HaveOccurred())
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusServiceUnavailable))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("denies a caller without the publish permission", func() {
+		permissionService.ExpectedCalls = nil
+		permissionService.On("GetPermissionByID", mock.Anything, mock.Anything).
+			Return(nil, errors.New("not found"))
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusForbidden))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("hides the item from a caller who cannot see it", func() {
+		statAs("alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusNotFound))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("refuses a recipient who cannot see the item", func() {
+		statAs("")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusForbidden))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("refuses a recipient that does not exist", func() {
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("nobody", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusNotFound))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	DescribeTable("rejects a malformed body",
+		func(body string) {
+			statAs("", "alice-token")
+
+			rr := httptest.NewRecorder()
+			svc.SendActivityNotification(rr, request("alice", body))
+
+			Expect(rr.Code).To(Equal(http.StatusBadRequest))
+			Expect(mentions()).To(BeEmpty())
+		},
+		Entry("no json", `not json`),
+		Entry("unknown field", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`","chainId":1}`),
+		// the ms graph api takes them from trusted clients, ours is the browser
+		Entry("template parameters", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`","templateParameters":[{"name":"actor","value":"someone else"}]}`),
+		Entry("no topic", `{"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("topic source entityUrl", `{"topic":{"source":"entityUrl","value":"https://cloud.opencloud.test/f/item"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("topic value is no resource id", `{"topic":{"source":"text","value":"not-an-id"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("unknown activityType", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"reactedTo","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("unknown teamsAppId", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned","teamsAppId":"14a4bd3a-1e0f-4a2e-9f30-1cc1f0d0a1cd"}`),
+		Entry("no teamsAppId", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned"}`),
+	)
+})

services/proxy/pkg/config/defaults/defaultconfig.go

@@ -310,11 +310,6 @@ func DefaultPolicies() []config.Policy {
 					Service:  "eu.opencloud.web.collaboration",
 					// Method: "POST" // toDo: fails with method, WHY???
 				},
-				{
-					Endpoint: "/collaboration/notify",
-					Service:  "eu.opencloud.web.collaboration",
-					// Method: "POST" // toDo: fails with method, WHY???
-				},
 				{
 					Endpoint:    "/collaboration",
 					Service:     "eu.opencloud.web.collaboration",