ability to disable http and/or event consumer for user log service

pkg/shared/errors.go

@@ -102,10 +102,10 @@ func MissingURLSigningSecret(service string) error {
 		service, defaults.BaseConfigPath())
 }
 
-func AllComponentsDisabledError(service string) error {
+func AllApiHandlersDisabledError(service string) error {
 	return fmt.Errorf("All request handlers and event consumers are disabled for %s; at least one component must be enabled."+
 		"Make sure your %s config contains the proper values "+
 		"(e.g. by using 'opencloud init --diff' and applying the patch or setting a value manually in "+
-		"the config/corresponding environment variable).", 
+		"the config/corresponding environment variable).",
 		service, defaults.BaseConfigPath())
 }

services/eventhistory/pkg/config/parser/parse.go

@@ -36,7 +36,7 @@ func ParseConfig(cfg *config.Config) error {
 // Validate validates the config
 func Validate(cfg *config.Config) error {
 	if cfg.Events.Disabled && cfg.GRPC.Disabled {
-		return shared.AllComponentsDisabledError(cfg.Service.Name)
+		return shared.AllApiHandlersDisabledError(cfg.Service.Name)
 	}
 
 	return nil

services/userlog/pkg/command/server.go

@@ -21,6 +21,9 @@ import (
 	"github.com/opencloud-eu/opencloud/services/userlog/pkg/metrics"
 	"github.com/opencloud-eu/opencloud/services/userlog/pkg/server/debug"
 	"github.com/opencloud-eu/opencloud/services/userlog/pkg/server/http"
+	"github.com/opencloud-eu/opencloud/services/userlog/pkg/service"
+	consumerSvc "github.com/opencloud-eu/opencloud/services/userlog/pkg/service/consumer"
+	httpSvc "github.com/opencloud-eu/opencloud/services/userlog/pkg/service/http"
 
 	"github.com/opencloud-eu/reva/v2/pkg/events"
 	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
@@ -84,10 +87,6 @@ func Server(cfg *config.Config) *cobra.Command {
 			mtrcs.BuildInfo.WithLabelValues(version.GetString()).Set(1)
 
 			connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
-			stream, err := stream.NatsFromConfig(connName, false, stream.NatsConfig(cfg.Events))
-			if err != nil {
-				return err
-			}
 
 			st := store.Create(
 				store.Store(cfg.Persistence.Store),
@@ -120,20 +119,40 @@ func Server(cfg *config.Config) *cobra.Command {
 			vClient := settingssvc.NewValueService("eu.opencloud.api.settings", grpcClient)
 			rClient := settingssvc.NewRoleService("eu.opencloud.api.settings", grpcClient)
 
+			handle, err := service.NewUserlogService(
+				service.Logger(logger),
+				service.Store(st),
+				service.HistoryClient(hClient),
+				service.TraceProvider(tracerProvider),
+			)
+			if err != nil {
+				return err
+			}
+
 			gr := runner.NewGroup()
-			{
+
+			if !cfg.HTTP.Disabled {
+				httpService, err := httpSvc.New(
+					handle,
+					httpSvc.Logger(logger),
+					httpSvc.Config(cfg),
+					httpSvc.GatewaySelector(gatewaySelector),
+					httpSvc.ValueClient(vClient),
+					httpSvc.RegisteredEvents(_registeredEvents),
+					httpSvc.TraceProvider(tracerProvider),
+				)
+				if err != nil {
+					logger.Info().Err(err).Str("transport", "http").Msg("Failed to initialize server")
+					return err
+				}
+
 				server, err := http.Server(
 					http.Logger(logger),
 					http.Context(ctx),
 					http.Config(cfg),
 					http.Metrics(mtrcs),
-					http.Store(st),
-					http.Stream(stream),
-					http.GatewaySelector(gatewaySelector),
-					http.History(hClient),
-					http.Value(vClient),
 					http.Role(rClient),
-					http.RegisteredEvents(_registeredEvents),
+					http.Service(httpService),
 					http.TracerProvider(tracerProvider),
 				)
 
@@ -143,6 +162,41 @@ func Server(cfg *config.Config) *cobra.Command {
 				}
 
 				gr.Add(runner.NewGoMicroHttpServerRunner(cfg.Service.Name+".http", server))
+			} else {
+				logger.Info().Msg("HTTP server disabled, not starting HTTP service")
+			}
+
+			if !cfg.Events.Disabled {
+				evStream, err := stream.NatsFromConfig(connName, false, stream.NatsConfig{
+					Endpoint:             cfg.Events.Endpoint,
+					Cluster:              cfg.Events.Cluster,
+					TLSInsecure:          cfg.Events.TLSInsecure,
+					TLSRootCACertificate: cfg.Events.TLSRootCACertificate,
+					EnableTLS:            cfg.Events.EnableTLS,
+					AuthUsername:         cfg.Events.AuthUsername,
+					AuthPassword:         cfg.Events.AuthPassword,
+				})
+				if err != nil {
+					return err
+				}
+
+				consumerService, err := consumerSvc.New(
+					handle,
+					evStream,
+					consumerSvc.Context(ctx),
+					consumerSvc.Logger(logger),
+					consumerSvc.Config(cfg),
+					consumerSvc.GatewaySelector(gatewaySelector),
+					consumerSvc.ValueClient(vClient),
+					consumerSvc.RegisteredEvents(_registeredEvents),
+				)
+				if err != nil {
+					return err
+				}
+
+				gr.Add(runner.New(cfg.Service.Name+".consumer", consumerService.Run, consumerService.Close))
+			} else {
+				logger.Info().Msg("event listening disabled, not starting event consumer")
 			}
 
 			{

services/userlog/pkg/config/config.go

@@ -53,6 +53,7 @@ type Persistence struct {
 
 // Events combines the configuration options for the event bus.
 type Events struct {
+	Disabled             bool   `yaml:"disabled" env:"USERLOG_EVENTS_DISABLED" desc:"Disables listening for events. Set this to true if the service should only handle HTTP requests." introductionVersion:"%NEXT%"`
 	Endpoint             string `yaml:"endpoint" env:"OC_EVENTS_ENDPOINT;USERLOG_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:"1.0.0"`
 	Cluster              string `yaml:"cluster" env:"OC_EVENTS_CLUSTER;USERLOG_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:"1.0.0"`
 	TLSInsecure          bool   `yaml:"tls_insecure" env:"OC_INSECURE;OC_EVENTS_TLS_INSECURE;USERLOG_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates." introductionVersion:"1.0.0"`
@@ -72,6 +73,7 @@ type CORS struct {
 
 // HTTP defines the available http configuration.
 type HTTP struct {
+	Disabled  bool                  `yaml:"disabled" env:"USERLOG_HTTP_DISABLED" desc:"Disables the HTTP service. Set this to true if the service should only handle events." introductionVersion:"%NEXT%"`
 	Addr      string                `yaml:"addr" env:"USERLOG_HTTP_ADDR" desc:"The bind address of the HTTP service." introductionVersion:"1.0.0"`
 	Namespace string                `yaml:"-"`
 	Root      string                `yaml:"root" env:"USERLOG_HTTP_ROOT" desc:"Subdirectory that serves as the root for this HTTP service." introductionVersion:"1.0.0"`

services/userlog/pkg/config/parser/parse.go

@@ -46,5 +46,9 @@ func Validate(cfg *config.Config) error {
 		return shared.MissingServiceAccountSecret(cfg.Service.Name)
 	}
 
+	if cfg.Events.Disabled && cfg.HTTP.Disabled {
+		return shared.AllApiHandlersDisabledError(cfg.Service.Name)
+	}
+
 	return nil
 }

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

@@ -2,40 +2,39 @@ package http
 
 import (
 	"context"
+	"net/http"
 
-	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	"github.com/opencloud-eu/opencloud/pkg/log"
-	ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0"
 	settingssvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
 	"github.com/opencloud-eu/opencloud/services/userlog/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/userlog/pkg/metrics"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 
 	"github.com/spf13/pflag"
-	"go-micro.dev/v4/store"
 	"go.opentelemetry.io/otel/trace"
 )
 
+// UserlogService is the http service interface the server needs
+type UserlogService interface {
+	HandleGetEvents(w http.ResponseWriter, r *http.Request)
+	HandleDeleteEvents(w http.ResponseWriter, r *http.Request)
+	HandlePostGlobalEvent(w http.ResponseWriter, r *http.Request)
+	HandleDeleteGlobalEvent(w http.ResponseWriter, r *http.Request)
+}
+
 // Option defines a single option function.
 type Option func(o *Options)
 
 // Options defines the available options for this package.
 type Options struct {
-	Logger           log.Logger
-	Context          context.Context
-	Config           *config.Config
-	Metrics          *metrics.Metrics
-	Flags            []pflag.Flag
-	Namespace        string
-	Store            store.Store
-	Stream           events.Stream
-	GatewaySelector  pool.Selectable[gateway.GatewayAPIClient]
-	HistoryClient    ehsvc.EventHistoryService
-	ValueClient      settingssvc.ValueService
-	RoleClient       settingssvc.RoleService
-	RegisteredEvents []events.Unmarshaller
-	TracerProvider   trace.TracerProvider
+	Logger         log.Logger
+	Context        context.Context
+	Config         *config.Config
+	Metrics        *metrics.Metrics
+	Flags          []pflag.Flag
+	Namespace      string
+	RoleClient     settingssvc.RoleService
+	UserlogService        UserlogService
+	TracerProvider trace.TracerProvider
 }
 
 // newOptions initializes the available default options.
@@ -91,45 +90,10 @@ func Namespace(val string) Option {
 	}
 }
 
-// Store provides a function to configure the store
-func Store(store store.Store) Option {
-	return func(o *Options) {
-		o.Store = store
-	}
-}
-
-// Stream provides a function to configure the stream
-func Stream(stream events.Stream) Option {
-	return func(o *Options) {
-		o.Stream = stream
-	}
-}
-
-// GatewaySelector provides a function to configure the gateway client selector
-func GatewaySelector(gatewaySelector pool.Selectable[gateway.GatewayAPIClient]) Option {
-	return func(o *Options) {
-		o.GatewaySelector = gatewaySelector
-	}
-}
-
-// History provides a function to configure the event history client
-func History(h ehsvc.EventHistoryService) Option {
-	return func(o *Options) {
-		o.HistoryClient = h
-	}
-}
-
-// RegisteredEvents provides a function to register events
-func RegisteredEvents(evs []events.Unmarshaller) Option {
-	return func(o *Options) {
-		o.RegisteredEvents = evs
-	}
-}
-
-// Value provides a function to configure the value service client
-func Value(vs settingssvc.ValueService) Option {
+// Service provides a function to set the userlog http service
+func Service(s UserlogService) Option {
 	return func(o *Options) {
-		o.ValueClient = vs
+		o.UserlogService = s
 	}
 }
 

services/userlog/pkg/server/http/server.go

@@ -1,6 +1,7 @@
 package http
 
 import (
+	"errors"
 	"fmt"
 
 	stdhttp "net/http"
@@ -10,17 +11,15 @@ import (
 	"github.com/opencloud-eu/opencloud/pkg/account"
 	"github.com/opencloud-eu/opencloud/pkg/cors"
 	"github.com/opencloud-eu/opencloud/pkg/middleware"
+	"github.com/opencloud-eu/opencloud/pkg/roles"
 	"github.com/opencloud-eu/opencloud/pkg/service/http"
 	"github.com/opencloud-eu/opencloud/pkg/tracing"
 	"github.com/opencloud-eu/opencloud/pkg/version"
-	svc "github.com/opencloud-eu/opencloud/services/userlog/pkg/service"
+	httpSvc "github.com/opencloud-eu/opencloud/services/userlog/pkg/service/http"
 	"github.com/riandyrn/otelchi"
 	"go-micro.dev/v4"
 )
 
-// Service is the service interface
-type Service any
-
 // Server initializes the http service and server.
 func Server(opts ...Option) (http.Service, error) {
 	options := newOptions(opts...)
@@ -77,24 +76,24 @@ func Server(opts ...Option) (http.Service, error) {
 		),
 	)
 
-	handle, err := svc.NewUserlogService(
-		svc.Logger(options.Logger),
-		svc.Stream(options.Stream),
-		svc.Mux(mux),
-		svc.Store(options.Store),
-		svc.Config(options.Config),
-		svc.HistoryClient(options.HistoryClient),
-		svc.GatewaySelector(options.GatewaySelector),
-		svc.ValueClient(options.ValueClient),
-		svc.RoleClient(options.RoleClient),
-		svc.RegisteredEvents(options.RegisteredEvents),
-		svc.TraceProvider(options.TracerProvider),
-	)
-	if err != nil {
-		return http.Service{}, err
+	if options.UserlogService == nil {
+		return http.Service{}, errors.New("need non nil userlog http service to serve http requests")
 	}
 
-	if err := micro.RegisterHandler(service.Server(), handle); err != nil {
+	m := roles.NewManager(
+		// TODO: caching?
+		roles.Logger(options.Logger),
+		roles.RoleService(options.RoleClient),
+	)
+
+	mux.Route("/ocs/v2.php/apps/notifications/api/v1/notifications", func(r chi.Router) {
+		r.Get("/", options.UserlogService.HandleGetEvents)
+		r.Delete("/", options.UserlogService.HandleDeleteEvents)
+		r.Post("/global", httpSvc.RequireAdminOrSecret(&m, options.Config.GlobalNotificationsSecret)(options.UserlogService.HandlePostGlobalEvent))
+		r.Delete("/global", httpSvc.RequireAdminOrSecret(&m, options.Config.GlobalNotificationsSecret)(options.UserlogService.HandleDeleteGlobalEvent))
+	})
+
+	if err := micro.RegisterHandler(service.Server(), mux); err != nil {
 		return http.Service{}, err
 	}
 

services/userlog/pkg/service/consumer/consumer_suite_test.go

@@ -0,0 +1,27 @@
+package consumer
+
+import (
+	"testing"
+
+	mRegistry "go-micro.dev/v4/registry"
+
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+
+	"github.com/opencloud-eu/opencloud/pkg/registry"
+)
+
+func init() {
+	r := registry.GetRegistry(registry.Inmemory())
+	service := registry.BuildGRPCService("eu.opencloud.api.gateway", "", "", "")
+	service.Nodes = []*mRegistry.Node{{
+		Address: "any",
+	}}
+
+	_ = r.Register(service)
+}
+
+func TestConsumer(t *testing.T) {
+	gomega.RegisterFailHandler(ginkgo.Fail)
+	ginkgo.RunSpecs(t, "Userlog consumer Suite")
+}

services/userlog/pkg/service/filter.go → services/userlog/pkg/service/consumer/filter.go

@@ -1,4 +1,4 @@
-package service
+package consumer
 
 import (
 	"context"