diff --git a/.drone.star b/.drone.star index 951690839d..dcf05705a0 100644 --- a/.drone.star +++ b/.drone.star @@ -78,6 +78,7 @@ config = { "services/search", "services/settings", "services/sharing", + "services/sse", "services/storage-system", "services/storage-publiclink", "services/storage-shares", diff --git a/.gitignore b/.gitignore index 0902237f4a..c4c08a62aa 100644 --- a/.gitignore +++ b/.gitignore @@ -49,6 +49,7 @@ protogen/buf.sha1.lock /third-party-licenses # misc +/tmp go.work go.work.sum .env diff --git a/Makefile b/Makefile index 92d5b88573..8e648ffc23 100644 --- a/Makefile +++ b/Makefile @@ -44,6 +44,7 @@ OCIS_MODULES = \ services/search \ services/settings \ services/sharing \ + services/sse \ services/storage-system \ services/storage-publiclink \ services/storage-shares \ diff --git a/changelog/unreleased/enhancement-sse-messaging.md b/changelog/unreleased/enhancement-sse-messaging.md new file mode 100644 index 0000000000..31c98eb2a0 --- /dev/null +++ b/changelog/unreleased/enhancement-sse-messaging.md @@ -0,0 +1,10 @@ +Enhancement: SSE for messaging + +So far, sse has only been used to exchange messages between the server and the client. +In order to be able to send more content to the client, we have moved the endpoint to a separate service and are now also using it for other notifications like: + +* notify postprocessing state changes. +* notify file locking and unlocking. +* ... @toDo + +https://github.com/owncloud/ocis/pull/6992 diff --git a/docs/services/general-info/port-ranges.md b/docs/services/general-info/port-ranges.md index f2286dadc5..e3baac04d5 100644 --- a/docs/services/general-info/port-ranges.md +++ b/docs/services/general-info/port-ranges.md @@ -30,7 +30,7 @@ We also suggest to use the last port in your extensions' range as a debug/metric | 9120-9124 | [graph]({{< ref "../graph/_index.md" >}}) | | 9125-9129 | [policies]({{< ref "../policies/_index.md" >}}) | | 9130-9134 | [idp]({{< ref "../idp/_index.md" >}}) | -| 9135-9139 | FREE (formerly used by graph-explorer) | +| 9135-9139 | [sse]({{< ref "../sse/_index.md" >}}) | | 9140-9141 | [frontend]({{< ref "../frontend/_index.md" >}}) | | 9142-9143 | [gateway]({{< ref "../gateway/_index.md" >}}) | | 9144-9145 | [users]({{< ref "../users/_index.md" >}}) | diff --git a/docs/services/sse/_index.md b/docs/services/sse/_index.md new file mode 100644 index 0000000000..d839d16de5 --- /dev/null +++ b/docs/services/sse/_index.md @@ -0,0 +1,16 @@ +--- +title: SSE +date: 2022-08-08T00:00:00+00:00 +weight: 20 +geekdocRepo: https://github.com/owncloud/ocis +geekdocEditPath: edit/master/docs/services/sse +geekdocFilePath: _index.md +geekdocCollapseSection: true +--- + +## Abstract + + +## Table of Contents + +{{< toc-tree >}} diff --git a/docs/services/sse/configuration.md b/docs/services/sse/configuration.md new file mode 100644 index 0000000000..902a2b5529 --- /dev/null +++ b/docs/services/sse/configuration.md @@ -0,0 +1,15 @@ +--- +title: Service Configuration +date: 2018-08-08T00:00:00+00:00 +weight: 20 +geekdocRepo: https://github.com/owncloud/ocis +geekdocEditPath: edit/master/docs/services/sse +geekdocFilePath: configuration.md +geekdocCollapseSection: true +--- + +## Example YAML Config + +{{< include file="services/_includes/app-provider-config-example.yaml" language="yaml" >}} + +{{< include file="services/_includes/app-provider_configvars.md" >}} diff --git a/ocis-pkg/config/config.go b/ocis-pkg/config/config.go index 394945fb22..bf16fd6717 100644 --- a/ocis-pkg/config/config.go +++ b/ocis-pkg/config/config.go @@ -27,6 +27,7 @@ import ( search "github.com/owncloud/ocis/v2/services/search/pkg/config" settings "github.com/owncloud/ocis/v2/services/settings/pkg/config" sharing "github.com/owncloud/ocis/v2/services/sharing/pkg/config" + sse "github.com/owncloud/ocis/v2/services/sse/pkg/config" storagepublic "github.com/owncloud/ocis/v2/services/storage-publiclink/pkg/config" storageshares "github.com/owncloud/ocis/v2/services/storage-shares/pkg/config" storagesystem "github.com/owncloud/ocis/v2/services/storage-system/pkg/config" @@ -99,6 +100,7 @@ type Config struct { Proxy *proxy.Config `yaml:"proxy"` Settings *settings.Config `yaml:"settings"` Sharing *sharing.Config `yaml:"sharing"` + SSE *sse.Config `yaml:"sse"` StorageSystem *storagesystem.Config `yaml:"storage_system"` StoragePublicLink *storagepublic.Config `yaml:"storage_public"` StorageShares *storageshares.Config `yaml:"storage_shares"` diff --git a/ocis-pkg/config/defaultconfig.go b/ocis-pkg/config/defaultconfig.go index 3afd0896c8..bfa03aee63 100644 --- a/ocis-pkg/config/defaultconfig.go +++ b/ocis-pkg/config/defaultconfig.go @@ -26,6 +26,7 @@ import ( search "github.com/owncloud/ocis/v2/services/search/pkg/config/defaults" settings "github.com/owncloud/ocis/v2/services/settings/pkg/config/defaults" sharing "github.com/owncloud/ocis/v2/services/sharing/pkg/config/defaults" + sse "github.com/owncloud/ocis/v2/services/sse/pkg/config/defaults" storagepublic "github.com/owncloud/ocis/v2/services/storage-publiclink/pkg/config/defaults" storageshares "github.com/owncloud/ocis/v2/services/storage-shares/pkg/config/defaults" storageSystem "github.com/owncloud/ocis/v2/services/storage-system/pkg/config/defaults" @@ -72,6 +73,7 @@ func DefaultConfig() *Config { Search: search.FullDefaultConfig(), Settings: settings.DefaultConfig(), Sharing: sharing.DefaultConfig(), + SSE: sse.DefaultConfig(), StoragePublicLink: storagepublic.DefaultConfig(), StorageShares: storageshares.DefaultConfig(), StorageSystem: storageSystem.DefaultConfig(), diff --git a/ocis/pkg/command/services.go b/ocis/pkg/command/services.go index 960abd452d..df143b623e 100644 --- a/ocis/pkg/command/services.go +++ b/ocis/pkg/command/services.go @@ -1,13 +1,13 @@ package command import ( + "github.com/urfave/cli/v2" + "github.com/owncloud/ocis/v2/ocis-pkg/config" "github.com/owncloud/ocis/v2/ocis-pkg/config/configlog" "github.com/owncloud/ocis/v2/ocis-pkg/config/parser" "github.com/owncloud/ocis/v2/ocis/pkg/command/helper" "github.com/owncloud/ocis/v2/ocis/pkg/register" - "github.com/urfave/cli/v2" - antivirus "github.com/owncloud/ocis/v2/services/antivirus/pkg/command" appprovider "github.com/owncloud/ocis/v2/services/app-provider/pkg/command" appregistry "github.com/owncloud/ocis/v2/services/app-registry/pkg/command" @@ -33,6 +33,7 @@ import ( search "github.com/owncloud/ocis/v2/services/search/pkg/command" settings "github.com/owncloud/ocis/v2/services/settings/pkg/command" sharing "github.com/owncloud/ocis/v2/services/sharing/pkg/command" + sse "github.com/owncloud/ocis/v2/services/sse/pkg/command" storagepubliclink "github.com/owncloud/ocis/v2/services/storage-publiclink/pkg/command" storageshares "github.com/owncloud/ocis/v2/services/storage-shares/pkg/command" storagesystem "github.com/owncloud/ocis/v2/services/storage-system/pkg/command" @@ -172,6 +173,11 @@ var svccmds = []register.Command{ cfg.Sharing.Commons = cfg.Commons }) }, + func(cfg *config.Config) *cli.Command { + return ServiceCommand(cfg, cfg.SSE.Service.Name, sse.GetCommands(cfg.SSE), func(c *config.Config) { + cfg.SSE.Commons = cfg.Commons + }) + }, func(cfg *config.Config) *cli.Command { return ServiceCommand(cfg, cfg.StoragePublicLink.Service.Name, storagepubliclink.GetCommands(cfg.StoragePublicLink), func(c *config.Config) { cfg.StoragePublicLink.Commons = cfg.Commons diff --git a/ocis/pkg/runtime/service/service.go b/ocis/pkg/runtime/service/service.go index 958d2d1671..027ef7e551 100644 --- a/ocis/pkg/runtime/service/service.go +++ b/ocis/pkg/runtime/service/service.go @@ -15,6 +15,8 @@ import ( "github.com/mohae/deepcopy" "github.com/olekukonko/tablewriter" + "github.com/thejerf/suture/v4" + ociscfg "github.com/owncloud/ocis/v2/ocis-pkg/config" "github.com/owncloud/ocis/v2/ocis-pkg/log" "github.com/owncloud/ocis/v2/ocis-pkg/shared" @@ -42,6 +44,7 @@ import ( search "github.com/owncloud/ocis/v2/services/search/pkg/command" settings "github.com/owncloud/ocis/v2/services/settings/pkg/command" sharing "github.com/owncloud/ocis/v2/services/sharing/pkg/command" + sse "github.com/owncloud/ocis/v2/services/sse/pkg/command" storagepublic "github.com/owncloud/ocis/v2/services/storage-publiclink/pkg/command" storageshares "github.com/owncloud/ocis/v2/services/storage-shares/pkg/command" storageSystem "github.com/owncloud/ocis/v2/services/storage-system/pkg/command" @@ -53,7 +56,6 @@ import ( web "github.com/owncloud/ocis/v2/services/web/pkg/command" webdav "github.com/owncloud/ocis/v2/services/webdav/pkg/command" webfinger "github.com/owncloud/ocis/v2/services/webfinger/pkg/command" - "github.com/thejerf/suture/v4" ) var ( @@ -298,11 +300,16 @@ func NewService(options ...Option) (*Service, error) { cfg.Sharing.Commons = cfg.Commons return sharing.Execute(cfg.Sharing) }) + dreg(opts.Config.SSE.Service.Name, func(ctx context.Context, cfg *ociscfg.Config) error { + cfg.SSE.Context = ctx + cfg.SSE.Commons = cfg.Commons + return sse.Execute(cfg.SSE) + }) return s, nil } -// Start an rpc service. By default the package scope Start will run all default services to provide with a working +// Start a rpc service. By default, the package scope Start will run all default services to provide with a working // oCIS instance. func Start(o ...Option) error { // Start the runtime. Most likely this was called ONLY by the `ocis server` subcommand, but since we cannot protect diff --git a/services/proxy/pkg/config/defaults/defaultconfig.go b/services/proxy/pkg/config/defaults/defaultconfig.go index e8a93e0de5..4c561e66e5 100644 --- a/services/proxy/pkg/config/defaults/defaultconfig.go +++ b/services/proxy/pkg/config/defaults/defaultconfig.go @@ -122,6 +122,11 @@ func DefaultPolicies() []config.Policy { Endpoint: "/archiver", Service: "com.owncloud.web.frontend", }, + { + // reroute oc10 notifications endpoint to userlog service + Endpoint: "/ocs/v2.php/apps/notifications/api/v1/notifications/sse", + Service: "com.owncloud.sse.sse", + }, { // reroute oc10 notifications endpoint to userlog service Endpoint: "/ocs/v2.php/apps/notifications/api/v1/notifications", diff --git a/services/sse/Makefile b/services/sse/Makefile new file mode 100644 index 0000000000..5381f8cfaf --- /dev/null +++ b/services/sse/Makefile @@ -0,0 +1,37 @@ +SHELL := bash +NAME := sse + +include ../../.make/recursion.mk + +############ tooling ############ +ifneq (, $(shell command -v go 2> /dev/null)) # suppress `command not found warnings` for non go targets in CI +include ../../.bingo/Variables.mk +endif + +############ go tooling ############ +include ../../.make/go.mk + +############ release ############ +include ../../.make/release.mk + +############ docs generate ############ +include ../../.make/docs.mk + +.PHONY: docs-generate +docs-generate: config-docs-generate + +############ generate ############ +include ../../.make/generate.mk + +.PHONY: ci-go-generate +ci-go-generate: # CI runs ci-node-generate automatically before this target + +.PHONY: ci-node-generate +ci-node-generate: + +############ licenses ############ +.PHONY: ci-node-check-licenses +ci-node-check-licenses: + +.PHONY: ci-node-save-licenses +ci-node-save-licenses: diff --git a/services/sse/README.md b/services/sse/README.md new file mode 100644 index 0000000000..ac5e43ae41 --- /dev/null +++ b/services/sse/README.md @@ -0,0 +1,7 @@ +# SSE + +The `sse` service is responsible for sending sse (Server-Sent Events) to a user. See [What is Server-Sent Events](https://medium.com/yemeksepeti-teknoloji/what-is-server-sent-events-sse-and-how-to-implement-it-904938bffd73) for a simple introduction and examples of server sent events. + +## Subscribing + +Clients can subscribe to the `/sse` endpoint to be informed by the server when an event happens. The `sse` endpoint will respect language changes of the user without needing to reconnect. Note that SSE has a limitation of six open connections per browser which can be reached if one has opened various tabs of the Web UI pointing to the same Infinite Scale instance. diff --git a/services/sse/cmd/sse/main.go b/services/sse/cmd/sse/main.go new file mode 100644 index 0000000000..8bd438f114 --- /dev/null +++ b/services/sse/cmd/sse/main.go @@ -0,0 +1,14 @@ +package main + +import ( + "os" + + "github.com/owncloud/ocis/v2/services/sse/pkg/command" + "github.com/owncloud/ocis/v2/services/sse/pkg/config/defaults" +) + +func main() { + if err := command.Execute(defaults.DefaultConfig()); err != nil { + os.Exit(1) + } +} diff --git a/services/sse/pkg/command/health.go b/services/sse/pkg/command/health.go new file mode 100644 index 0000000000..195dceade2 --- /dev/null +++ b/services/sse/pkg/command/health.go @@ -0,0 +1,62 @@ +package command + +import ( + "fmt" + "net/http" + + "github.com/owncloud/ocis/v2/ocis-pkg/log" + + "github.com/urfave/cli/v2" + + "github.com/owncloud/ocis/v2/ocis-pkg/config/configlog" + "github.com/owncloud/ocis/v2/services/sse/pkg/config" + "github.com/owncloud/ocis/v2/services/sse/pkg/config/parser" +) + +// Health is the entrypoint for the health command. +func Health(cfg *config.Config) *cli.Command { + return &cli.Command{ + Name: "health", + Usage: "check health status", + Category: "info", + Before: func(c *cli.Context) error { + return configlog.ReturnError(parser.ParseConfig(cfg)) + }, + Action: func(c *cli.Context) error { + logger := log.NewLogger( + log.Name(cfg.Service.Name), + log.Level(cfg.Log.Level), + log.Pretty(cfg.Log.Pretty), + log.Color(cfg.Log.Color), + log.File(cfg.Log.File), + ) + + resp, err := http.Get( + fmt.Sprintf( + "http://%s/healthz", + cfg.Debug.Addr, + ), + ) + + if err != nil { + logger.Fatal(). + Err(err). + Msg("Failed to request health check") + } + + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + logger.Fatal(). + Int("code", resp.StatusCode). + Msg("Health seems to be in bad state") + } + + logger.Debug(). + Int("code", resp.StatusCode). + Msg("Health got a good state") + + return nil + }, + } +} diff --git a/services/sse/pkg/command/root.go b/services/sse/pkg/command/root.go new file mode 100644 index 0000000000..f3a6e4b276 --- /dev/null +++ b/services/sse/pkg/command/root.go @@ -0,0 +1,30 @@ +package command + +import ( + "os" + + "github.com/urfave/cli/v2" + + "github.com/owncloud/ocis/v2/ocis-pkg/clihelper" + "github.com/owncloud/ocis/v2/services/sse/pkg/config" +) + +// GetCommands provides all commands for this service +func GetCommands(cfg *config.Config) cli.Commands { + return []*cli.Command{ + Server(cfg), + Health(cfg), + Version(cfg), + } +} + +// Execute is the entry point for the sse command. +func Execute(cfg *config.Config) error { + app := clihelper.DefaultApp(&cli.App{ + Name: "sse", + Usage: "Serve ownCloud sse for oCIS", + Commands: GetCommands(cfg), + }) + + return app.Run(os.Args) +} diff --git a/services/sse/pkg/command/server.go b/services/sse/pkg/command/server.go new file mode 100644 index 0000000000..b5da4063da --- /dev/null +++ b/services/sse/pkg/command/server.go @@ -0,0 +1,103 @@ +package command + +import ( + "context" + "fmt" + + "github.com/cs3org/reva/v2/pkg/events" + "github.com/cs3org/reva/v2/pkg/events/stream" + "github.com/oklog/run" + "github.com/urfave/cli/v2" + + "github.com/owncloud/ocis/v2/ocis-pkg/config/configlog" + "github.com/owncloud/ocis/v2/ocis-pkg/handlers" + "github.com/owncloud/ocis/v2/ocis-pkg/log" + "github.com/owncloud/ocis/v2/ocis-pkg/service/debug" + "github.com/owncloud/ocis/v2/ocis-pkg/tracing" + "github.com/owncloud/ocis/v2/ocis-pkg/version" + "github.com/owncloud/ocis/v2/services/sse/pkg/config" + "github.com/owncloud/ocis/v2/services/sse/pkg/config/parser" + "github.com/owncloud/ocis/v2/services/sse/pkg/server/http" +) + +// all events we care about +var _registeredEvents = []events.Unmarshaller{ + events.SendSSE{}, +} + +// Server is the entrypoint for the server command. +func Server(cfg *config.Config) *cli.Command { + return &cli.Command{ + Name: "server", + Usage: fmt.Sprintf("start the %s service without runtime (unsupervised mode)", cfg.Service.Name), + Category: "server", + Before: func(c *cli.Context) error { + return configlog.ReturnFatal(parser.ParseConfig(cfg)) + }, + Action: func(c *cli.Context) error { + var ( + gr = run.Group{} + ctx, cancel = func() (context.Context, context.CancelFunc) { + if cfg.Context == nil { + return context.WithCancel(context.Background()) + } + return context.WithCancel(cfg.Context) + }() + logger = log.NewLogger( + log.Name(cfg.Service.Name), + log.Level(cfg.Log.Level), + log.Pretty(cfg.Log.Pretty), + log.Color(cfg.Log.Color), + log.File(cfg.Log.File), + ) + ) + defer cancel() + + tracerProvider, err := tracing.GetServiceTraceProvider(cfg.Tracing, cfg.Service.Name) + if err != nil { + return err + } + + { + natsStream, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events)) + if err != nil { + return err + } + + server, err := http.Server( + http.Logger(logger), + http.Context(ctx), + http.Config(cfg), + http.Consumer(natsStream), + http.RegisteredEvents(_registeredEvents), + http.TracerProvider(tracerProvider), + ) + + gr.Add(server.Run, func(_ error) { + cancel() + }) + } + + { + server := debug.NewService( + debug.Logger(logger), + debug.Name(cfg.Service.Name), + debug.Version(version.GetString()), + debug.Address(cfg.Debug.Addr), + debug.Token(cfg.Debug.Token), + debug.Pprof(cfg.Debug.Pprof), + debug.Zpages(cfg.Debug.Zpages), + debug.Health(handlers.Health), + debug.Ready(handlers.Ready), + ) + + gr.Add(server.ListenAndServe, func(_ error) { + _ = server.Shutdown(ctx) + cancel() + }) + } + + return gr.Run() + }, + } +} diff --git a/services/sse/pkg/command/version.go b/services/sse/pkg/command/version.go new file mode 100644 index 0000000000..fe3b344f0f --- /dev/null +++ b/services/sse/pkg/command/version.go @@ -0,0 +1,27 @@ +package command + +import ( + "fmt" + + "github.com/owncloud/ocis/v2/ocis-pkg/version" + + "github.com/urfave/cli/v2" + + "github.com/owncloud/ocis/v2/services/sse/pkg/config" +) + +// Version prints the service versions of all running instances. +func Version(cfg *config.Config) *cli.Command { + return &cli.Command{ + Name: "version", + Usage: "print the version of this binary and the running service instances", + Category: "info", + Action: func(c *cli.Context) error { + fmt.Println("Version: " + version.GetString()) + fmt.Printf("Compiled: %s\n", version.Compiled()) + fmt.Println("") + + return nil + }, + } +} diff --git a/services/sse/pkg/config/config.go b/services/sse/pkg/config/config.go new file mode 100644 index 0000000000..b7e9509c48 --- /dev/null +++ b/services/sse/pkg/config/config.go @@ -0,0 +1,76 @@ +package config + +import ( + "context" + + "github.com/owncloud/ocis/v2/ocis-pkg/shared" +) + +// Config combines all available configuration parts. +type Config struct { + Commons *shared.Commons `yaml:"-"` // don't use this directly as configuration for a service + Log *Log + + Debug Debug `mask:"struct" yaml:"debug"` + Tracing *Tracing `yaml:"tracing"` + + Service Service `yaml:"-"` + + Events Events + HTTP HTTP `yaml:"http"` + TokenManager *TokenManager `yaml:"token_manager"` + + Context context.Context `yaml:"-" json:"-"` +} + +// Service defines the available service configuration. +type Service struct { + Name string `yaml:"-"` +} + +// Log defines the available log configuration. +type Log struct { + Level string `mapstructure:"level" env:"OCIS_LOG_LEVEL;SSE_LOG_LEVEL" desc:"The log level. Valid values are: 'panic', 'fatal', 'error', 'warn', 'info', 'debug', 'trace'."` + Pretty bool `mapstructure:"pretty" env:"OCIS_LOG_PRETTY;SSE_LOG_PRETTY" desc:"Activates pretty log output."` + Color bool `mapstructure:"color" env:"OCIS_LOG_COLOR;SSE_LOG_COLOR" desc:"Activates colorized log output."` + File string `mapstructure:"file" env:"OCIS_LOG_FILE;SSE_LOG_FILE" desc:"The path to the log file. Activates logging to this file if set."` +} + +// Debug defines the available debug configuration. +type Debug struct { + Addr string `yaml:"addr" env:"SSE_DEBUG_ADDR" desc:"Bind address of the debug server, where metrics, health, config and debug endpoints will be exposed."` + Token string `yaml:"token" env:"SSE_DEBUG_TOKEN" desc:"Token to secure the metrics endpoint."` + Pprof bool `yaml:"pprof" env:"SSE_DEBUG_PPROF" desc:"Enables pprof, which can be used for profiling."` + Zpages bool `yaml:"zpages" env:"SSE_DEBUG_ZPAGES" desc:"Enables zpages, which can be used for collecting and viewing in-memory traces."` +} + +// Events combines the configuration options for the event bus. +type Events struct { + Endpoint string `yaml:"endpoint" env:"OCIS_EVENTS_ENDPOINT;SSE_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."` + Cluster string `yaml:"cluster" env:"OCIS_EVENTS_CLUSTER;SSE_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."` + TLSInsecure bool `yaml:"tls_insecure" env:"OCIS_INSECURE;SSE_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates."` + TLSRootCACertificate string `yaml:"tls_root_ca_certificate" env:"OCIS_EVENTS_TLS_ROOT_CA_CERTIFICATE;SSE_EVENTS_TLS_ROOT_CA_CERTIFICATE" desc:"The root CA certificate used to validate the server's TLS certificate. If provided SSE_EVENTS_TLS_INSECURE will be seen as false."` + EnableTLS bool `yaml:"enable_tls" env:"OCIS_EVENTS_ENABLE_TLS;SSE_EVENTS_ENABLE_TLS" desc:"Enable TLS for the connection to the events broker. The events broker is the ocis service which receives and delivers events between the services."` +} + +// CORS defines the available cors configuration. +type CORS struct { + AllowedOrigins []string `yaml:"allow_origins" env:"OCIS_CORS_ALLOW_ORIGINS;SSE_CORS_ALLOW_ORIGINS" desc:"A comma-separated list of allowed CORS origins. See following chapter for more details: *Access-Control-Allow-Origin* at https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Access-Control-Allow-Origin"` + AllowedMethods []string `yaml:"allow_methods" env:"OCIS_CORS_ALLOW_METHODS;SSE_CORS_ALLOW_METHODS" desc:"A comma-separated list of allowed CORS methods. See following chapter for more details: *Access-Control-Request-Method* at https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Access-Control-Request-Method"` + AllowedHeaders []string `yaml:"allow_headers" env:"OCIS_CORS_ALLOW_HEADERS;SSE_CORS_ALLOW_HEADERS" desc:"A blank or comma-separated list of allowed CORS headers. See following chapter for more details: *Access-Control-Request-Headers* at https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Access-Control-Request-Headers."` + AllowCredentials bool `yaml:"allow_credentials" env:"OCIS_CORS_ALLOW_CREDENTIALS;SSE_CORS_ALLOW_CREDENTIALS" desc:"Allow credentials for CORS.See following chapter for more details: *Access-Control-Allow-Credentials* at https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Access-Control-Allow-Credentials."` +} + +// HTTP defines the available http configuration. +type HTTP struct { + Addr string `yaml:"addr" env:"SSE_HTTP_ADDR" desc:"The bind address of the HTTP service."` + Namespace string `yaml:"-"` + Root string `yaml:"root" env:"SSE_HTTP_ROOT" desc:"Subdirectory that serves as the root for this HTTP service."` + CORS CORS `yaml:"cors"` + TLS shared.HTTPServiceTLS `yaml:"tls"` +} + +// TokenManager is the config for using the reva token manager +type TokenManager struct { + JWTSecret string `yaml:"jwt_secret" env:"OCIS_JWT_SECRET;SSE_JWT_SECRET" desc:"The secret to mint and validate jwt tokens."` +} diff --git a/services/sse/pkg/config/defaults/defaultconfig.go b/services/sse/pkg/config/defaults/defaultconfig.go new file mode 100644 index 0000000000..0aba9557fc --- /dev/null +++ b/services/sse/pkg/config/defaults/defaultconfig.go @@ -0,0 +1,91 @@ +package defaults + +import ( + "strings" + + "github.com/owncloud/ocis/v2/services/sse/pkg/config" +) + +// FullDefaultConfig returns a fully initialized default configuration which is needed for doc generation. +func FullDefaultConfig() *config.Config { + cfg := DefaultConfig() + EnsureDefaults(cfg) + Sanitize(cfg) + return cfg +} + +// DefaultConfig returns the services default config +func DefaultConfig() *config.Config { + return &config.Config{ + Debug: config.Debug{ + Addr: "127.0.0.1:9135", + Token: "", + }, + Service: config.Service{ + Name: "sse", + }, + Events: config.Events{ + Endpoint: "127.0.0.1:9233", + Cluster: "ocis-cluster", + }, + HTTP: config.HTTP{ + Addr: "127.0.0.1:0", + Root: "/", + Namespace: "com.owncloud.sse", + CORS: config.CORS{ + AllowedOrigins: []string{"*"}, + AllowedMethods: []string{"GET"}, + AllowedHeaders: []string{"Authorization", "Origin", "Content-Type", "Accept", "X-Requested-With", "X-Request-Id", "Ocs-Apirequest"}, + AllowCredentials: true, + }, + }, + } +} + +// EnsureDefaults adds default values to the configuration if they are not set yet +func EnsureDefaults(cfg *config.Config) { + // provide with defaults for shared logging, since we need a valid destination address for "envdecode". + if cfg.Log == nil && cfg.Commons != nil && cfg.Commons.Log != nil { + cfg.Log = &config.Log{ + Level: cfg.Commons.Log.Level, + Pretty: cfg.Commons.Log.Pretty, + Color: cfg.Commons.Log.Color, + File: cfg.Commons.Log.File, + } + } else if cfg.Log == nil { + cfg.Log = &config.Log{} + } + + if cfg.TokenManager == nil && cfg.Commons != nil && cfg.Commons.TokenManager != nil { + cfg.TokenManager = &config.TokenManager{ + JWTSecret: cfg.Commons.TokenManager.JWTSecret, + } + } else if cfg.TokenManager == nil { + cfg.TokenManager = &config.TokenManager{} + } + + if cfg.Commons != nil { + cfg.HTTP.TLS = cfg.Commons.HTTPServiceTLS + } + + // provide with defaults for shared tracing, since we need a valid destination address for "envdecode". + if cfg.Tracing == nil && cfg.Commons != nil && cfg.Commons.Tracing != nil { + cfg.Tracing = &config.Tracing{ + Enabled: cfg.Commons.Tracing.Enabled, + Type: cfg.Commons.Tracing.Type, + Endpoint: cfg.Commons.Tracing.Endpoint, + Collector: cfg.Commons.Tracing.Collector, + } + } else if cfg.Tracing == nil { + cfg.Tracing = &config.Tracing{} + } +} + +// Sanitize sanitizes the configuration +func Sanitize(cfg *config.Config) { + // sanitize config + if cfg.HTTP.Root != "/" { + cfg.HTTP.Root = strings.TrimSuffix(cfg.HTTP.Root, "/") + } + +} diff --git a/services/sse/pkg/config/parser/parse.go b/services/sse/pkg/config/parser/parse.go new file mode 100644 index 0000000000..dc9375210d --- /dev/null +++ b/services/sse/pkg/config/parser/parse.go @@ -0,0 +1,38 @@ +package parser + +import ( + "errors" + + ociscfg "github.com/owncloud/ocis/v2/ocis-pkg/config" + "github.com/owncloud/ocis/v2/services/sse/pkg/config" + "github.com/owncloud/ocis/v2/services/sse/pkg/config/defaults" + + "github.com/owncloud/ocis/v2/ocis-pkg/config/envdecode" +) + +// ParseConfig loads configuration from known paths. +func ParseConfig(cfg *config.Config) error { + _, err := ociscfg.BindSourcesToStructs(cfg.Service.Name, cfg) + if err != nil { + return err + } + + defaults.EnsureDefaults(cfg) + + // load all env variables relevant to the config in the current context. + if err := envdecode.Decode(cfg); err != nil { + // no environment variable set for this config is an expected "error" + if !errors.Is(err, envdecode.ErrNoTargetFieldsAreSet) { + return err + } + } + + defaults.Sanitize(cfg) + + return Validate(cfg) +} + +// Validate validates our little config +func Validate(cfg *config.Config) error { + return nil +} diff --git a/services/sse/pkg/config/tracing.go b/services/sse/pkg/config/tracing.go new file mode 100644 index 0000000000..5a177c3f59 --- /dev/null +++ b/services/sse/pkg/config/tracing.go @@ -0,0 +1,21 @@ +package config + +import "github.com/owncloud/ocis/v2/ocis-pkg/tracing" + +// Tracing defines the available tracing configuration. +type Tracing struct { + Enabled bool `yaml:"enabled" env:"OCIS_TRACING_ENABLED;SSE_TRACING_ENABLED" desc:"Activates tracing."` + Type string `yaml:"type" env:"OCIS_TRACING_TYPE;SSE_TRACING_TYPE" desc:"The type of tracing. Defaults to '', which is the same as 'jaeger'. Allowed tracing types are 'jaeger' and '' as of now."` + Endpoint string `yaml:"endpoint" env:"OCIS_TRACING_ENDPOINT;SSE_TRACING_ENDPOINT" desc:"The endpoint of the tracing agent."` + Collector string `yaml:"collector" env:"OCIS_TRACING_COLLECTOR;SSE_TRACING_COLLECTOR" desc:"The HTTP endpoint for sending spans directly to a collector, i.e. http://jaeger-collector:14268/api/traces. Only used if the tracing endpoint is unset."` +} + +// Convert Tracing to the tracing package's Config struct. +func (t Tracing) Convert() tracing.Config { + return tracing.Config{ + Enabled: t.Enabled, + Type: t.Type, + Endpoint: t.Endpoint, + Collector: t.Collector, + } +} diff --git a/services/sse/pkg/server/http/option.go b/services/sse/pkg/server/http/option.go new file mode 100644 index 0000000000..3640ba9ea9 --- /dev/null +++ b/services/sse/pkg/server/http/option.go @@ -0,0 +1,76 @@ +package http + +import ( + "context" + + "github.com/cs3org/reva/v2/pkg/events" + "github.com/owncloud/ocis/v2/ocis-pkg/log" + "github.com/owncloud/ocis/v2/services/sse/pkg/config" + "go.opentelemetry.io/otel/trace" +) + +// 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 + Consumer events.Consumer + RegisteredEvents []events.Unmarshaller + TracerProvider trace.TracerProvider +} + +// newOptions initializes the available default options. +func newOptions(opts ...Option) Options { + opt := Options{} + + for _, o := range opts { + o(&opt) + } + + return opt +} + +// Logger provides a function to set the logger option. +func Logger(val log.Logger) Option { + return func(o *Options) { + o.Logger = val + } +} + +// Context provides a function to set the context option. +func Context(val context.Context) Option { + return func(o *Options) { + o.Context = val + } +} + +// Config provides a function to set the config option. +func Config(val *config.Config) Option { + return func(o *Options) { + o.Config = val + } +} + +// Consumer provides a function to configure the consumer +func Consumer(consumer events.Consumer) Option { + return func(o *Options) { + o.Consumer = consumer + } +} + +// RegisteredEvents provides a function to register events +func RegisteredEvents(evs []events.Unmarshaller) Option { + return func(o *Options) { + o.RegisteredEvents = evs + } +} + +// TracerProvider provides a function to set the TracerProvider option +func TracerProvider(val trace.TracerProvider) Option { + return func(o *Options) { + o.TracerProvider = val + } +} diff --git a/services/sse/pkg/server/http/server.go b/services/sse/pkg/server/http/server.go new file mode 100644 index 0000000000..430ae0db3d --- /dev/null +++ b/services/sse/pkg/server/http/server.go @@ -0,0 +1,96 @@ +package http + +import ( + "fmt" + + stdhttp "net/http" + + "github.com/cs3org/reva/v2/pkg/events" + "github.com/go-chi/chi/v5" + chimiddleware "github.com/go-chi/chi/v5/middleware" + "github.com/owncloud/ocis/v2/ocis-pkg/account" + "github.com/owncloud/ocis/v2/ocis-pkg/cors" + "github.com/owncloud/ocis/v2/ocis-pkg/middleware" + "github.com/owncloud/ocis/v2/ocis-pkg/service/http" + "github.com/owncloud/ocis/v2/ocis-pkg/tracing" + "github.com/owncloud/ocis/v2/ocis-pkg/version" + svc "github.com/owncloud/ocis/v2/services/sse/pkg/service" + "github.com/riandyrn/otelchi" + "go-micro.dev/v4" +) + +// Service is the service interface +type Service interface { +} + +// Server initializes the http service and server. +func Server(opts ...Option) (http.Service, error) { + options := newOptions(opts...) + + service, err := http.NewService( + http.TLSConfig(options.Config.HTTP.TLS), + http.Logger(options.Logger), + http.Namespace(options.Config.HTTP.Namespace), + http.Name(options.Config.Service.Name), + http.Version(version.GetString()), + http.Address(options.Config.HTTP.Addr), + http.Context(options.Context), + ) + if err != nil { + options.Logger.Error(). + Err(err). + Msg("Error initializing http service") + return http.Service{}, fmt.Errorf("could not initialize http service: %w", err) + } + + middlewares := []func(stdhttp.Handler) stdhttp.Handler{ + chimiddleware.RequestID, + middleware.Version( + "userlog", + version.GetString(), + ), + middleware.Logger( + options.Logger, + ), + middleware.ExtractAccountUUID( + account.Logger(options.Logger), + account.JWTSecret(options.Config.TokenManager.JWTSecret), + ), + middleware.Cors( + cors.Logger(options.Logger), + cors.AllowedOrigins(options.Config.HTTP.CORS.AllowedOrigins), + cors.AllowedMethods(options.Config.HTTP.CORS.AllowedMethods), + cors.AllowedHeaders(options.Config.HTTP.CORS.AllowedHeaders), + cors.AllowCredentials(options.Config.HTTP.CORS.AllowCredentials), + ), + middleware.Secure, + } + + mux := chi.NewMux() + mux.Use(middlewares...) + + mux.Use( + otelchi.Middleware( + "sse", + otelchi.WithChiRoutes(mux), + otelchi.WithTracerProvider(options.TracerProvider), + otelchi.WithPropagators(tracing.GetPropagator()), + ), + ) + + ch, err := events.Consume(options.Consumer, "sse", options.RegisteredEvents...) + if err != nil { + return http.Service{}, err + } + + handle, err := svc.NewSSE(options.Config, options.Logger, ch, mux) + if err != nil { + return http.Service{}, err + } + + if err := micro.RegisterHandler(service.Server(), handle); err != nil { + return http.Service{}, err + } + + return service, nil +} diff --git a/services/sse/pkg/service/service.go b/services/sse/pkg/service/service.go new file mode 100644 index 0000000000..e0043f2a86 --- /dev/null +++ b/services/sse/pkg/service/service.go @@ -0,0 +1,89 @@ +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" + + "github.com/owncloud/ocis/v2/ocis-pkg/log" + "github.com/owncloud/ocis/v2/services/sse/pkg/config" +) + +// SSE defines implements the business logic for Service. +type SSE struct { + c *config.Config + l log.Logger + m *chi.Mux + sse *sse.Server + evChannel <-chan events.Event +} + +// NewSSE returns a service implementation for Service. +func NewSSE(c *config.Config, l log.Logger, ch <-chan events.Event, mux *chi.Mux) (SSE, error) { + s := SSE{ + c: c, + l: l, + m: mux, + sse: sse.New(), + evChannel: ch, + } + mux.Route("/ocs/v2.php/apps/notifications/api/v1/notifications", func(r chi.Router) { + r.Get("/sse", s.HandleSSE) + }) + + go s.ListenForEvents() + + return s, nil +} + +// ServeHTTP fulfills Handler interface +func (s SSE) ServeHTTP(w http.ResponseWriter, r *http.Request) { + s.m.ServeHTTP(w, r) +} + +// ListenForEvents listens for events +func (s SSE) ListenForEvents() error { + for e := range s.evChannel { + switch ev := e.Event.(type) { + default: + s.l.Error().Interface("event", ev).Msg("unhandled event") + case events.SendSSE: + s.sse.Publish(ev.UserID, &sse.Event{ + Event: []byte(ev.Type), + Data: ev.Message, + }) + } + } + + return nil +} + +// HandleSSE is the GET handler for events +func (s SSE) HandleSSE(w http.ResponseWriter, r *http.Request) { + u, ok := revactx.ContextGetUser(r.Context()) + if !ok { + s.l.Error().Msg("sse: no user in context") + w.WriteHeader(http.StatusInternalServerError) + return + } + + uid := u.GetId().GetOpaqueId() + if uid == "" { + s.l.Error().Msg("sse: user in context is broken") + w.WriteHeader(http.StatusInternalServerError) + return + } + + stream := s.sse.CreateStream(uid) + stream.AutoReplay = false + + // add stream to URL + q := r.URL.Query() + q.Set("stream", uid) + r.URL.RawQuery = q.Encode() + + s.sse.ServeHTTP(w, r) +} diff --git a/services/userlog/README.md b/services/userlog/README.md index 474fcd07f7..aaaf18f409 100644 --- a/services/userlog/README.md +++ b/services/userlog/README.md @@ -30,10 +30,6 @@ For the time being, the configuration which user related events are of interest The `userlog` service provides an API to retrieve configured events. For now, this API is mostly following the [oc10 notification GET API](https://doc.owncloud.com/server/next/developer_manual/core/apis/ocs-notification-endpoint-v1.html#get-user-notifications). -## Subscribing - -Additionally to the oc10 API, the `userlog` service also provides an `/sse` (Server-Sent Events) endpoint to be informed by the server when an event happens. See [What is Server-Sent Events](https://medium.com/yemeksepeti-teknoloji/what-is-server-sent-events-sse-and-how-to-implement-it-904938bffd73) for a simple introduction and examples of server sent events. The `sse` endpoint will respect language changes of the user without needing to reconnect. Note that SSE has a limitation of six open connections per browser which can be reached if one has opened various tabs of the Web UI pointing to the same Infinite Scale instance. - ## Posting The userlog service is able to store global messages that will be displayed in the Web UI to all users. If a user deletes the message in the Web UI, it reappears on reload. Global messages use the endpoint `/ocs/v2.php/apps/notifications/api/v1/notifications/global` and are activated by sending a `POST` request. Note that sending another `POST` request of the same type overwrites the previous one. For the time being, only the type `deprovision` is supported. diff --git a/services/userlog/pkg/command/server.go b/services/userlog/pkg/command/server.go index 6002bf4f3e..3ec49220cd 100644 --- a/services/userlog/pkg/command/server.go +++ b/services/userlog/pkg/command/server.go @@ -81,7 +81,7 @@ func Server(cfg *config.Config) *cli.Command { defer cancel() - consumer, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events)) + stream, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events)) if err != nil { return err } @@ -121,7 +121,7 @@ func Server(cfg *config.Config) *cli.Command { http.Config(cfg), http.Metrics(mtrcs), http.Store(st), - http.Consumer(consumer), + http.Stream(stream), http.GatewaySelector(gatewaySelector), http.History(hClient), http.Value(vClient), diff --git a/services/userlog/pkg/config/config.go b/services/userlog/pkg/config/config.go index 11e81ccab6..782fb07917 100644 --- a/services/userlog/pkg/config/config.go +++ b/services/userlog/pkg/config/config.go @@ -28,7 +28,7 @@ type Config struct { Events Events `yaml:"events"` Persistence Persistence `yaml:"persistence"` - DisableSSE bool `yaml:"disable_sse" env:"OCIS_DISABLE_SSE,USERLOG_DISABLE_SSE" desc:"Disables server-sent events (sse). When disabled, clients will no longer be able to connect to the sse endpoint."` + DisableSSE bool `yaml:"disable_sse" env:"OCIS_DISABLE_SSE,USERLOG_DISABLE_SSE" desc:"Disables server-sent events (sse). When disabled, clients will no longer receive sse notifications."` GlobalNotificationsSecret string `yaml:"global_notifications_secret" env:"USERLOG_GLOBAL_NOTIFICATIONS_SECRET" desc:"The secret to secure the global notifications endpoint. Only system admins and users knowing that secret can call the global notifications POST/DELETE endpoints."` diff --git a/services/userlog/pkg/server/http/option.go b/services/userlog/pkg/server/http/option.go index 02ffe0d6de..44443d57bd 100644 --- a/services/userlog/pkg/server/http/option.go +++ b/services/userlog/pkg/server/http/option.go @@ -28,7 +28,7 @@ type Options struct { Flags []cli.Flag Namespace string Store store.Store - Consumer events.Consumer + Stream events.Stream GatewaySelector pool.Selectable[gateway.GatewayAPIClient] HistoryClient ehsvc.EventHistoryService ValueClient settingssvc.ValueService @@ -97,10 +97,10 @@ func Store(store store.Store) Option { } } -// Consumer provides a function to configure the consumer -func Consumer(consumer events.Consumer) Option { +// Stream provides a function to configure the stream +func Stream(stream events.Stream) Option { return func(o *Options) { - o.Consumer = consumer + o.Stream = stream } } diff --git a/services/userlog/pkg/server/http/server.go b/services/userlog/pkg/server/http/server.go index 4adcaec018..1766f1d566 100644 --- a/services/userlog/pkg/server/http/server.go +++ b/services/userlog/pkg/server/http/server.go @@ -80,7 +80,7 @@ func Server(opts ...Option) (http.Service, error) { handle, err := svc.NewUserlogService( svc.Logger(options.Logger), - svc.Consumer(options.Consumer), + svc.Stream(options.Stream), svc.Mux(mux), svc.Store(options.Store), svc.Config(options.Config), diff --git a/services/userlog/pkg/service/http.go b/services/userlog/pkg/service/http.go index c689e18eb0..74167e4219 100644 --- a/services/userlog/pkg/service/http.go +++ b/services/userlog/pkg/service/http.go @@ -91,33 +91,6 @@ func (ul *UserlogService) HandleGetEvents(w http.ResponseWriter, r *http.Request w.Write(b) } -// HandleSSE is the GET handler for events -func (ul *UserlogService) HandleSSE(w http.ResponseWriter, r *http.Request) { - u, ok := revactx.ContextGetUser(r.Context()) - if !ok { - ul.log.Error().Msg("sse: no user in context") - w.WriteHeader(http.StatusInternalServerError) - return - } - - uid := u.GetId().GetOpaqueId() - if uid == "" { - ul.log.Error().Msg("sse: user in context is broken") - w.WriteHeader(http.StatusInternalServerError) - return - } - - stream := ul.sse.CreateStream(uid) - stream.AutoReplay = false - - // add stream to URL - q := r.URL.Query() - q.Set("stream", uid) - r.URL.RawQuery = q.Encode() - - ul.sse.ServeHTTP(w, r) -} - // HandlePostGlobaelEvent is the POST handler for global events func (ul *UserlogService) HandlePostGlobalEvent(w http.ResponseWriter, r *http.Request) { var req PostEventsRequest diff --git a/services/userlog/pkg/service/options.go b/services/userlog/pkg/service/options.go index a4422926df..9cd9f97ec0 100644 --- a/services/userlog/pkg/service/options.go +++ b/services/userlog/pkg/service/options.go @@ -19,7 +19,7 @@ type Option func(*Options) // Options for the userlog service type Options struct { Logger log.Logger - Consumer events.Consumer + Stream events.Stream Mux *chi.Mux Store store.Store Config *config.Config @@ -38,10 +38,10 @@ func Logger(log log.Logger) Option { } } -// Consumer configures an event consumer for the userlog service -func Consumer(c events.Consumer) Option { +// Stream configures an event stream for the userlog service +func Stream(s events.Stream) Option { return func(o *Options) { - o.Consumer = c + o.Stream = s } } diff --git a/services/userlog/pkg/service/service.go b/services/userlog/pkg/service/service.go index dd82fae18d..7ebee77768 100644 --- a/services/userlog/pkg/service/service.go +++ b/services/userlog/pkg/service/service.go @@ -26,7 +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" - "github.com/r3labs/sse/v2" micrometadata "go-micro.dev/v4/metadata" "go-micro.dev/v4/store" "go.opentelemetry.io/otel/trace" @@ -42,10 +41,10 @@ type UserlogService struct { historyClient ehsvc.EventHistoryService gatewaySelector pool.Selectable[gateway.GatewayAPIClient] valueClient settingssvc.ValueService - sse *sse.Server registeredEvents map[string]events.Unmarshaller tp trace.TracerProvider tracer trace.Tracer + publisher events.Publisher } // NewUserlogService returns an EventHistory service @@ -55,11 +54,11 @@ func NewUserlogService(opts ...Option) (*UserlogService, error) { opt(o) } - if o.Consumer == nil || o.Store == nil { - return nil, fmt.Errorf("need non nil consumer (%v) and store (%v) to work properly", o.Consumer, o.Store) + if o.Stream == nil || o.Store == nil { + return nil, fmt.Errorf("need non nil stream (%v) and store (%v) to work properly", o.Stream, o.Store) } - ch, err := events.Consume(o.Consumer, "userlog", o.RegisteredEvents...) + ch, err := events.Consume(o.Stream, "userlog", o.RegisteredEvents...) if err != nil { return nil, err } @@ -75,10 +74,7 @@ func NewUserlogService(opts ...Option) (*UserlogService, error) { registeredEvents: make(map[string]events.Unmarshaller), tp: o.TraceProvider, tracer: o.TraceProvider.Tracer("github.com/owncloud/ocis/services/userlog/pkg/service"), - } - - if !ul.cfg.DisableSSE { - ul.sse = sse.New() + publisher: o.Stream, } for _, e := range o.RegisteredEvents { @@ -97,10 +93,6 @@ func NewUserlogService(opts ...Option) (*UserlogService, error) { r.Delete("/", ul.HandleDeleteEvents) r.Post("/global", RequireAdminOrSecret(&m, o.Config.GlobalNotificationsSecret)(ul.HandlePostGlobalEvent)) r.Delete("/global", RequireAdminOrSecret(&m, o.Config.GlobalNotificationsSecret)(ul.HandleDeleteGlobalEvent)) - - if !ul.cfg.DisableSSE { - r.Get("/sse", ul.HandleSSE) - } }) go ul.MemorizeEvents(ch) @@ -348,8 +340,11 @@ func (ul *UserlogService) sendSSE(userid string, event events.Event) error { return err } - ul.sse.Publish(userid, &sse.Event{Data: b}) - return nil + return events.Publish(context.Background(), ul.publisher, events.SendSSE{ + UserID: userid, + Type: "userlog-notification", + Message: b, + }) } func (ul *UserlogService) removeExpiredEvents(userid string, all []string, received []*ehmsg.Event) error { diff --git a/services/userlog/pkg/service/service_test.go b/services/userlog/pkg/service/service_test.go index 6c79b5f6b3..cc34f9c0c4 100644 --- a/services/userlog/pkg/service/service_test.go +++ b/services/userlog/pkg/service/service_test.go @@ -79,7 +79,7 @@ var _ = Describe("UserlogService", func() { ul, err = service.NewUserlogService( service.Config(cfg), - service.Consumer(bus), + service.Stream(bus), service.Store(sto), service.Logger(log.NewLogger()), service.Mux(chi.NewMux()), @@ -96,9 +96,9 @@ var _ = Describe("UserlogService", func() { It("it stores, returns and deletes a couple of events", func() { ids := make(map[string]struct{}) - ids[bus.Publish(events.SpaceDisabled{Executant: &user.UserId{OpaqueId: "executinguserid"}})] = struct{}{} - ids[bus.Publish(events.SpaceDisabled{Executant: &user.UserId{OpaqueId: "executinguserid"}})] = struct{}{} - ids[bus.Publish(events.SpaceDisabled{Executant: &user.UserId{OpaqueId: "executinguserid"}})] = struct{}{} + ids[bus.publish(events.SpaceDisabled{Executant: &user.UserId{OpaqueId: "executinguserid"}})] = struct{}{} + ids[bus.publish(events.SpaceDisabled{Executant: &user.UserId{OpaqueId: "executinguserid"}})] = struct{}{} + ids[bus.publish(events.SpaceDisabled{Executant: &user.UserId{OpaqueId: "executinguserid"}})] = struct{}{} // ids[bus.Publish(events.SpaceMembershipExpired{SpaceOwner: &user.UserId{OpaqueId: "userid"}})] = struct{}{} // ids[bus.Publish(events.ShareCreated{Executant: &user.UserId{OpaqueId: "userid"}})] = struct{}{} @@ -156,7 +156,11 @@ func (tb testBus) Consume(_ string, _ ...microevents.ConsumeOption) (<-chan micr return ch, nil } -func (tb testBus) Publish(e interface{}) string { +func (tb testBus) Publish(_ string, _ interface{}, _ ...microevents.PublishOption) error { + return nil +} + +func (tb testBus) publish(e interface{}) string { ev := events.Event{ ID: uuid.New().String(), Type: reflect.TypeOf(e).String(),