chore(graph): disable HTTP or eventhandlers by configuration

In the scope of the broader issue #1312, this PR deals with performing
those changes for the `graph` service, namely to add the ability to
disable the HTTP API or to disable the events API handler by
configuration.

It also adds metrics for the events processing, and tests for the events
processing.

The previous implementation was combining the HTTP server service and
the events consumption, which is why this PR refactors the composition
of those services:

 * the event consumption has been moved into its own service
 * the identity.Backend is created beforehand, and then injected as a
   collaborator in both the HTTP service as well as the event consumer
   service

It also adds metrics, mainly for the event processing.

To encourage re-use in latter implementations and changes, it also
introduces two top-level package changes:

 * internal/eventstest/events_test_helpers: contains a TestBus
   implementation to unit-test event consumers without NATS
 * internal/metricstest/metrics_test_helpers: contains assertion
   functions to test Prometheus metrics

internal/eventstest/events_test_helpers.go

@@ -0,0 +1,44 @@
+package eventstest
+
+import (
+	"encoding/json"
+	"reflect"
+
+	"github.com/google/uuid"
+
+	rev "github.com/opencloud-eu/reva/v2/pkg/events"
+	microevents "go-micro.dev/v4/events"
+)
+
+func NewTestBus() TestBus {
+	return TestBus(make(chan rev.Event))
+}
+
+type TestBus chan rev.Event
+
+func (tb TestBus) Consume(_ string, _ ...microevents.ConsumeOption) (<-chan microevents.Event, error) {
+	ch := make(chan microevents.Event)
+	go func() {
+		for ev := range tb {
+			b, _ := json.Marshal(ev.Event)
+			ch <- microevents.Event{
+				Payload: b,
+				Metadata: map[string]string{
+					rev.MetadatakeyEventID:   ev.ID,
+					rev.MetadatakeyEventType: ev.Type,
+				},
+			}
+		}
+	}()
+	return ch, nil
+}
+
+func (tb TestBus) Publish(e any) string {
+	ev := rev.Event{
+		ID:    uuid.New().String(),
+		Type:  reflect.TypeOf(e).String(),
+		Event: e,
+	}
+	tb <- ev
+	return ev.ID
+}

internal/metricstest/metrics_test_helpers.go

@@ -0,0 +1,143 @@
+package metricstest
+
+import (
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// copied and adapted from Prometheus testutil.ToFloat64(), since we don't import that package
+func collect(c prometheus.Collector) []prometheus.Metric {
+	result := []prometheus.Metric{}
+	ch := make(chan prometheus.Metric)
+	done := make(chan struct{})
+	go func() {
+		for m := range ch {
+			result = append(result, m)
+		}
+		close(done)
+	}()
+	c.Collect(ch)
+	close(ch)
+	<-done
+	return result
+}
+
+func RequireIsNotSet(t require.TestingT, c prometheus.Collector, msgAndArgs ...any) {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+	if !IsNotSet(t, c, msgAndArgs) {
+		t.FailNow()
+	}
+}
+
+func IsNotSet(t assert.TestingT, c prometheus.Collector, msgAndArgs ...any) bool {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+
+	m := collect(c)
+	if len(m) > 0 {
+		return assert.Fail(t, "Metric exists while expected to not exist", msgAndArgs)
+	} else {
+		return true
+	}
+}
+
+func RequireEqual(t require.TestingT, expected float64, c prometheus.Collector, msgAndArgs ...any) {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+	if !Equal(t, expected, c, msgAndArgs) {
+		t.FailNow()
+	}
+}
+
+// copied and adapted from Prometheus testutil.ToFloat64(), since we don't import that package
+func Equal(t assert.TestingT, expected float64, c prometheus.Collector, msgAndArgs ...any) bool {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+
+	m := collect(c)
+	if !assert.Len(t, m, 1, msgAndArgs...) {
+		return false
+	}
+	pb := &dto.Metric{}
+	err := m[0].Write(pb)
+	if !assert.NoError(t, err, msgAndArgs...) {
+		return false
+	}
+	if pb.Gauge != nil {
+		return assert.Equal(t, expected, pb.Gauge.GetValue(), msgAndArgs...)
+	} else if pb.Counter != nil {
+		return assert.Equal(t, expected, pb.Counter.GetValue(), msgAndArgs...)
+	} else if pb.Untyped != nil {
+		return assert.Equal(t, expected, pb.Untyped.GetValue(), msgAndArgs...)
+	} else {
+		return assert.Fail(t, fmt.Sprintf("collected a non-gauge/counter/untyped metric: %s", pb), msgAndArgs...)
+	}
+}
+
+func RequireEqualWithLabels(t require.TestingT, expectedValue float64, expectedLabels map[string]string, c prometheus.Collector, msgAndArgs ...any) {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+	if !EqualWithLabels(t, expectedValue, expectedLabels, c, msgAndArgs) {
+		t.FailNow()
+	}
+}
+
+func EqualWithLabels(t assert.TestingT, expectedValue float64, expectedLabels map[string]string, c prometheus.Collector, msgAndArgs ...any) bool {
+	if h, ok := t.(interface{ Helper() }); ok {
+		h.Helper()
+	}
+
+	m := collect(c)
+	if !assert.Len(t, m, 1, "collected %d metrics instead of exactly 1", len(m)) {
+		return false
+	}
+	pb := &dto.Metric{}
+	err := m[0].Write(pb)
+	if !assert.NoError(t, err) {
+		return false
+	}
+	if pb.Gauge != nil {
+		if !assert.Equal(t, expectedValue, pb.Gauge.GetValue()) {
+			return false
+		}
+	} else if pb.Counter != nil {
+		if !assert.Equal(t, expectedValue, pb.Counter.GetValue()) {
+			return false
+		}
+	} else if pb.Untyped != nil {
+		if !assert.Equal(t, expectedValue, pb.Untyped.GetValue()) {
+			return false
+		}
+	} else {
+		return assert.Fail(t, "collected a non-gauge/counter/untyped metric: %s", pb)
+	}
+
+	if !assert.NotNil(t, pb.Label) {
+		return false
+	}
+	actualLabels := map[string]string{}
+	for _, label := range pb.Label {
+		if !assert.NotNil(t, label) {
+			return false
+		}
+		if !assert.NotNil(t, label.Name) {
+			return false
+		}
+		if !assert.NotNil(t, label.Value) {
+			return false
+		}
+		actualLabels[*label.Name] = *label.Value
+	}
+	return assert.Equal(t, expectedLabels, actualLabels, msgAndArgs)
+}

services/graph/README.md

@@ -168,7 +168,7 @@ The output of this command includes the following information for each role:
 * `Condition`
 * `Allowed resource actions`
 
-**Example output (shortned)**
+**Example output (shortened)**
 
 ```bash
 +--------------------------------------+----------+--------------------------------+--------------------------------+------------------------------------------+
@@ -184,3 +184,22 @@ The output of this command includes the following information for each role:
 +--------------------------------------+----------+--------------------------------+--------------------------------+------------------------------------------+
 ```
 
+## API Handlers
+
+To specialize `graph` service instances in order to scale them independently, it is possible to disable its API handlers:
+
+* `GRAPH_HTTP_DISABLE`: when set to `true`, the service does not listen on HTTP and only consumes events (defaults to `false`)
+* `GRAPH_EVENTS_DISABLE_CONSUMER`: when set to `true`, the service does not consome events and only listens on HTTP (defaults to `false`)
+
+## Metrics
+
+The `graph` service provides the following metrics:
+
+| Name | Description |
+| ---- | ----------- |
+| `opencloud_graph_build_info{version=...}` | Contains a label `version` that is set to the current version of the service, and always has a value of `1` |
+| `opencloud_graph_events_enabled` | Is set to `1` if the Events API handler is enabled, or `0` if not |
+| `opencloud_graph_http_enabled` | Is set to `1` if the HTTP API handler is enabled, or `0` if not |
+| `opencloud_graph_events{event=...,result=...}` | Counts the number of events that have been consumed, with a `event` label that contains the name of the event, and a `result` label that is set to `success` or `failure` |
+| `opencloud_graph_events_invalid` | Counts the number of invalid events that are malformed or are missing required data |
+| `opencloud_graph_events_unsupported` | Counts the numbef of consumed events that cannot be processes by this service, should always be `0` |

services/graph/pkg/command/server.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
+	"github.com/opencloud-eu/opencloud/pkg/generators"
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	natspkg "github.com/opencloud-eu/opencloud/pkg/nats"
 	"github.com/opencloud-eu/opencloud/pkg/runner"
@@ -14,9 +15,14 @@ import (
 	"github.com/opencloud-eu/opencloud/pkg/version"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/config/parser"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/identity"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/metrics"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/server/debug"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/server/http"
+	evc "github.com/opencloud-eu/opencloud/services/graph/pkg/service/events"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -46,7 +52,7 @@ func Server(cfg *config.Config) *cobra.Command {
 			}
 			ctx := cfg.Context
 
-			mtrcs := metrics.New()
+			mtrcs := metrics.New(prometheus.DefaultRegisterer)
 			mtrcs.BuildInfo.WithLabelValues(version.GetString()).Set(1)
 
 			var kv jetstream.KeyValue
@@ -78,9 +84,37 @@ func Server(cfg *config.Config) *cobra.Command {
 				}
 			}
 
+			identityBackend, eduBackend, err := identity.CreateIdentityBackends(
+				cfg.Identity.Backend,
+				cfg,
+				&logger,
+				traceProvider,
+			)
+			if err != nil {
+				logger.Error().Err(err).Msg("Error initializing the identity backend")
+				return fmt.Errorf("could not initialize identity backend: %w", err)
+			}
+
+			var eventsStream events.Stream
+			if cfg.Events.Endpoint != "" {
+				var err error
+				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
+				eventsStream, err = stream.NatsFromConfig(connName, false, cfg.Events.ToNatsConfig())
+				if err != nil {
+					logger.Error().Err(err).Msg("Error initializing events publisher")
+					return fmt.Errorf("could not initialize events publisher: %w", err)
+				}
+			}
+
 			gr := runner.NewGroup()
-			{
+
+			if !cfg.HTTP.Disabled {
+				mtrcs.HttpEnabled.Set(1)
+
 				server, err := http.Server(
+					identityBackend,
+					eduBackend,
+					eventsStream,
 					http.Logger(logger),
 					http.Context(ctx),
 					http.Config(cfg),
@@ -92,8 +126,37 @@ func Server(cfg *config.Config) *cobra.Command {
 					logger.Error().Err(err).Str("transport", "http").Msg("Failed to initialize server")
 					return err
 				}
-
 				gr.Add(runner.NewGoMicroHttpServerRunner(cfg.Service.Name+".http", server))
+			} else {
+				mtrcs.HttpEnabled.Set(0)
+				logger.Info().Str("transport", "http").Msg("HTTP server is disabled")
+			}
+
+			if !cfg.Events.DisabledConsumer {
+				mtrcs.EventsEnabled.Set(1)
+
+				// even if events are enabled, we still need to differentiate between whether this process
+				// show be consuming events or not (and even when that is disabled, we still need to be
+				// able to produce events), which is why this is a separate setting;
+				// for context, see https://github.com/opencloud-eu/opencloud/issues/1312
+
+				logger := &log.Logger{Logger: logger.With().Str("transport", "events").Logger()}
+				eventConsumer, err := evc.NewService(cfg.Context, eventsStream, identityBackend, mtrcs, logger)
+				if err != nil {
+					return fmt.Errorf("could not initialize events consumer: %w", err)
+				}
+
+				gr.Add(runner.New(cfg.Service.Name+".svc", func() error {
+					return eventConsumer.Start()
+				}, func() {
+					err := eventConsumer.Close()
+					if err != nil {
+						logger.Error().Err(err).Msg("failed to stop event consumer")
+					}
+				}))
+			} else {
+				mtrcs.EventsEnabled.Set(0)
+				logger.Info().Str("transport", "events").Msg("event consumer is disabled")
 			}
 
 			{

services/graph/pkg/config/config.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/opencloud-eu/opencloud/pkg/shared"
+	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
 )
 
 // Config combines all available configuration parts.
@@ -129,6 +130,7 @@ type API struct {
 
 // Events combines the configuration options for the event bus.
 type Events struct {
+	DisabledConsumer     bool   `yaml:"disabled_consumer" env:"GRAPH_EVENTS_DISABLE_CONSUMER" desc:"Disables consuming events. Set this to true if the service should only handle HTTP requests." introductionVersion:"%NEXT%"`
 	Endpoint             string `yaml:"endpoint" env:"OC_EVENTS_ENDPOINT;GRAPH_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. Set to a empty string to disable emitting events." introductionVersion:"1.0.0"`
 	Cluster              string `yaml:"cluster" env:"OC_EVENTS_CLUSTER;GRAPH_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." introductionVersion:"1.0.0"`
 	TLSInsecure          bool   `yaml:"tls_insecure" env:"OC_INSECURE;OC_EVENTS_TLS_INSECURE;GRAPH_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates." introductionVersion:"1.0.0"`
@@ -138,6 +140,18 @@ type Events struct {
 	AuthPassword         string `yaml:"password" env:"OC_EVENTS_AUTH_PASSWORD;GRAPH_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:"1.0.0"`
 }
 
+func (e Events) ToNatsConfig() stream.NatsConfig {
+	return stream.NatsConfig{
+		Endpoint:             e.Endpoint,
+		Cluster:              e.Cluster,
+		TLSInsecure:          e.TLSInsecure,
+		TLSRootCACertificate: e.TLSRootCACertificate,
+		EnableTLS:            e.EnableTLS,
+		AuthUsername:         e.AuthUsername,
+		AuthPassword:         e.AuthPassword,
+	}
+}
+
 // CORS defines the available cors configuration.
 type CORS struct {
 	AllowedOrigins   []string `yaml:"allow_origins" env:"OC_CORS_ALLOW_ORIGINS;GRAPH_CORS_ALLOW_ORIGINS" desc:"A 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. See the Environment Variable Types description for more details." introductionVersion:"1.0.0"`

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

@@ -43,6 +43,7 @@ func DefaultConfig() *config.Config {
 			Token: "",
 		},
 		HTTP: config.HTTP{
+			Disabled:  false,
 			Addr:      "127.0.0.1:9120",
 			Namespace: "eu.opencloud.web",
 			Root:      "/graph",
@@ -118,9 +119,10 @@ func DefaultConfig() *config.Config {
 			TTL:      time.Hour * 24,
 		},
 		Events: config.Events{
-			Endpoint:  "127.0.0.1:9233",
-			Cluster:   "opencloud-cluster",
-			EnableTLS: false,
+			DisabledConsumer: false,
+			Endpoint:         "127.0.0.1:9233",
+			Cluster:          "opencloud-cluster",
+			EnableTLS:        false,
 		},
 		MaxConcurrency: 20,
 		UnifiedRoles: config.UnifiedRoles{

services/graph/pkg/config/http.go

@@ -4,6 +4,7 @@ import "github.com/opencloud-eu/opencloud/pkg/shared"
 
 // HTTP defines the available http configuration.
 type HTTP struct {
+	Disabled  bool                  `yaml:"disabled" env:"GRAPH_HTTP_DISABLE" desc:"Disables the HTTP service. Set this to true if the service should only consume events." introductionVersion:"%NEXT%"`
 	Addr      string                `yaml:"addr" env:"GRAPH_HTTP_ADDR" desc:"The bind address of the HTTP service." introductionVersion:"1.0.0"`
 	Namespace string                `yaml:"-"`
 	Root      string                `yaml:"root" env:"GRAPH_HTTP_ROOT" desc:"Subdirectory that serves as the root for this HTTP service." introductionVersion:"1.0.0"`

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

@@ -39,6 +39,14 @@ func ParseConfig(cfg *config.Config) error {
 }
 
 func Validate(cfg *config.Config) error {
+	if cfg.HTTP.Disabled && cfg.Events.DisabledConsumer {
+		// might be debatable, but this situation should be treated as an error,
+		// as the process wouldn't be able to serve either API and would thus be
+		// completely useless -- in that case, just don't start this service
+		// in the first place (especially since it's optional)
+		return errors.New("both HTTP and events consumption APIs are disabled by configuration; at least one must be enabled")
+	}
+
 	if cfg.TokenManager.JWTSecret == "" {
 		return shared.MissingJWTTokenError(cfg.Service.Name)
 	}

services/graph/pkg/identity/factory.go

@@ -0,0 +1,136 @@
+package identity
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"os"
+
+	ldapv3 "github.com/go-ldap/ldap/v3"
+	ocldap "github.com/opencloud-eu/opencloud/pkg/ldap"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/pkg/registry"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/config"
+	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
+	"github.com/opencloud-eu/reva/v2/pkg/utils/ldap"
+	"go.opentelemetry.io/otel/trace"
+)
+
+func CreateIdentityBackends(name string, cfg *config.Config, logger *log.Logger, traceProvider trace.TracerProvider) (Backend, EducationBackend, error) {
+	switch name {
+	case "cs3":
+		gatewaySelector, err := pool.GatewaySelector(
+			cfg.Reva.Address,
+			append(
+				cfg.Reva.GetRevaOptions(),
+				pool.WithRegistry(registry.GetRegistry()),
+				pool.WithTracerProvider(traceProvider),
+			)...,
+		)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		return &CS3{
+			Config:          cfg.Reva,
+			Logger:          logger,
+			GatewaySelector: gatewaySelector,
+		}, nil, nil
+	case "ldap":
+		var err error
+
+		var tlsConf *tls.Config
+		if cfg.Identity.LDAP.Insecure {
+			// When insecure is set to true then we don't need a certificate.
+			cfg.Identity.LDAP.CACert = ""
+			tlsConf = &tls.Config{
+				MinVersion: tls.VersionTLS12,
+
+				//nolint:gosec // We need the ability to run with "insecure" (dev/testing)
+				InsecureSkipVerify: cfg.Identity.LDAP.Insecure,
+			}
+		}
+
+		if cfg.Identity.LDAP.CACert != "" {
+			if err := ocldap.WaitForCA(*logger,
+				cfg.Identity.LDAP.Insecure,
+				cfg.Identity.LDAP.CACert); err != nil {
+				logger.Fatal().Err(err).Msg("The configured LDAP CA cert does not exist")
+			}
+			if tlsConf == nil {
+				tlsConf = &tls.Config{
+					MinVersion: tls.VersionTLS12,
+				}
+			}
+			certs := x509.NewCertPool()
+			pemData, err := os.ReadFile(cfg.Identity.LDAP.CACert)
+			if err != nil {
+				logger.Error().Err(err).Msg("Error initializing LDAP Backend")
+				return nil, nil, err
+			}
+			if !certs.AppendCertsFromPEM(pemData) {
+				logger.Error().Msg("Error initializing LDAP Backend. Adding CA cert failed")
+				return nil, nil, err
+			}
+			tlsConf.RootCAs = certs
+		}
+
+		conn := ldap.NewLDAPWithReconnect(
+			ldap.Config{
+				URI:          cfg.Identity.LDAP.URI,
+				BindDN:       cfg.Identity.LDAP.BindDN,
+				BindPassword: cfg.Identity.LDAP.BindPassword,
+				TLSConfig:    tlsConf,
+			},
+		)
+		conn.SetLogger(&logger.Logger)
+		lb, err := NewLDAPBackend(conn, cfg.Identity.LDAP, logger)
+		if err != nil {
+			logger.Error().Err(err).Msg("Error initializing LDAP Backend")
+			return nil, nil, err
+		}
+
+		identityBackend := lb
+		var eduBackend EducationBackend = lb
+
+		if !cfg.Identity.LDAP.EducationResourcesEnabled {
+			eduBackend = &ErrEducationBackend{}
+		}
+
+		disableMechanismType, err := ParseDisableMechanismType(cfg.Identity.LDAP.DisableUserMechanism)
+		if err != nil {
+			logger.Error().Err(err).Msg("Error initializing LDAP Backend")
+			return nil, nil, err
+		}
+
+		if disableMechanismType == DisableMechanismGroup {
+			logger.Info().Msg("LocalUserDisable is true, will create group if not exists")
+			err := lb.CreateLDAPGroupByDN(cfg.Identity.LDAP.LdapDisabledUsersGroupDN)
+			if err != nil {
+				isAnError := false
+				var lerr *ldapv3.Error
+				if errors.As(err, &lerr) {
+					if lerr.ResultCode != ldapv3.LDAPResultEntryAlreadyExists {
+						isAnError = true
+					}
+				} else {
+					isAnError = true
+				}
+
+				if isAnError {
+					msg := "error adding group for disabling users"
+					logger.Error().Err(err).Str("local_user_disable", cfg.Identity.LDAP.LdapDisabledUsersGroupDN).Msg(msg)
+					return nil, nil, err
+				}
+			}
+		}
+
+		return identityBackend, eduBackend, nil
+
+	default:
+		err := fmt.Errorf("unknown identity backend: '%s'", name)
+		logger.Err(err)
+		return nil, nil, err
+	}
+}

services/graph/pkg/identity/ldap.go

@@ -703,7 +703,7 @@ func (i *LDAP) usersFromLDAPEntries(entries []*ldap.Entry, exp []string) ([]*lib
 func (i *LDAP) UpdateLastSignInDate(ctx context.Context, userID string, timestamp time.Time) error {
 	if !i.writeEnabled {
 		i.logger.Debug().Str("backend", "ldap").Msg("The LDAP Server is readonly. Skipping update of last sign in date")
-		return nil
+		return nil // TODO: do we really want to just silently do nothing here, rather than returning an error?
 	}
 	e, err := i.getLDAPUserByID(userID)
 	switch {

services/graph/pkg/metrics/metrics.go

@@ -12,12 +12,21 @@ var (
 
 // Metrics defines the available metrics of this service.
 type Metrics struct {
-	// Counter  *prometheus.CounterVec
-	BuildInfo *prometheus.GaugeVec
+	BuildInfo         *prometheus.GaugeVec
+	EventsEnabled     prometheus.Gauge
+	HttpEnabled       prometheus.Gauge
+	EventsProcessed   *prometheus.CounterVec
+	InvalidEvents     prometheus.Counter
+	UnsupportedEvents prometheus.Counter
 }
 
+const (
+	ResultSuccess = "success"
+	ResultFailure = "failure"
+)
+
 // New initializes the available metrics.
-func New() *Metrics {
+func New(registerer prometheus.Registerer) *Metrics {
 	m := &Metrics{
 		BuildInfo: prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: Namespace,
@@ -25,9 +34,44 @@ func New() *Metrics {
 			Name:      "build_info",
 			Help:      "Build information",
 		}, []string{"version"}),
+		EventsEnabled: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_enabled",
+			Help:      "Whether this instance consumes events (1) or not (0)",
+		}),
+		HttpEnabled: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "http_enabled",
+			Help:      "Whether this instance processes HTTP API calls (1) or not (0)",
+		}),
+		EventsProcessed: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events",
+			Help:      "Number of consumed events",
+		}, []string{"event", "result"}),
+		InvalidEvents: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_invalid",
+			Help:      "Number of supported events with invalid data",
+		}),
+		UnsupportedEvents: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "events_unsupported",
+			Help:      "Number of unsupported events that were consumed and ignored",
+		}),
 	}
 
 	_ = prometheus.Register(m.BuildInfo)
+	_ = prometheus.Register(m.EventsEnabled)
+	_ = prometheus.Register(m.HttpEnabled)
+	_ = prometheus.Register(m.EventsProcessed)
+	_ = prometheus.Register(m.UnsupportedEvents)
+	_ = prometheus.Register(m.InvalidEvents)
 	// TODO: implement metrics
 	return m
 }

services/graph/pkg/server/http/server.go

@@ -8,7 +8,6 @@ import (
 
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	chimiddleware "github.com/go-chi/chi/v5/middleware"
-	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 	revaMetadata "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
 	"go-micro.dev/v4"
@@ -16,7 +15,6 @@ import (
 
 	"github.com/opencloud-eu/opencloud/pkg/account"
 	"github.com/opencloud-eu/opencloud/pkg/cors"
-	"github.com/opencloud-eu/opencloud/pkg/generators"
 	"github.com/opencloud-eu/opencloud/pkg/keycloak"
 	"github.com/opencloud-eu/opencloud/pkg/middleware"
 	"github.com/opencloud-eu/opencloud/pkg/registry"
@@ -27,12 +25,13 @@ import (
 	ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0"
 	searchsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/search/v0"
 	settingssvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/identity"
 	graphMiddleware "github.com/opencloud-eu/opencloud/services/graph/pkg/middleware"
 	svc "github.com/opencloud-eu/opencloud/services/graph/pkg/service/v0"
 )
 
 // Server initializes the http service and server.
-func Server(opts ...Option) (http.Service, error) {
+func Server(identityBackend identity.Backend, eduBackend identity.EducationBackend, eventsStream events.Stream, opts ...Option) (http.Service, error) {
 	options := newOptions(opts...)
 
 	service, err := http.NewService(
@@ -53,20 +52,6 @@ func Server(opts ...Option) (http.Service, error) {
 		return http.Service{}, fmt.Errorf("could not initialize http service: %w", err)
 	}
 
-	var eventsStream events.Stream
-
-	if options.Config.Events.Endpoint != "" {
-		var err error
-		connName := generators.GenerateConnectionName(options.Config.Service.Name, generators.NTypeBus)
-		eventsStream, err = stream.NatsFromConfig(connName, false, stream.NatsConfig(options.Config.Events))
-		if err != nil {
-			options.Logger.Error().
-				Err(err).
-				Msg("Error initializing events publisher")
-			return http.Service{}, fmt.Errorf("could not initialize events publisher: %w", err)
-		}
-	}
-
 	middlewares := []func(stdhttp.Handler) stdhttp.Handler{
 		middleware.TraceContext,
 		chimiddleware.RequestID,
@@ -168,8 +153,7 @@ func Server(opts ...Option) (http.Service, error) {
 		svc.Logger(options.Logger),
 		svc.Config(options.Config),
 		svc.Middleware(middlewares...),
-		svc.EventsPublisher(eventsStream),
-		svc.EventsConsumer(eventsStream),
+		svc.EventsPublisher(eventsStream), // is required even when event consumption is disabled
 		svc.WithRoleService(roleService),
 		svc.WithValueService(valueService),
 		svc.WithRequireAdminMiddleware(requireAdminMiddleware),
@@ -179,6 +163,8 @@ func Server(opts ...Option) (http.Service, error) {
 		svc.EventHistoryClient(hClient),
 		svc.TraceProvider(options.TraceProvider),
 		svc.WithNatsKeyValue(options.NatsKeyValue),
+		svc.WithIdentityBackend(identityBackend),
+		svc.WithIdentityEducationBackend(eduBackend),
 	)
 
 	if err != nil {

services/graph/pkg/service/events/service.go

@@ -0,0 +1,124 @@
+package events
+
+import (
+	"context"
+	"io"
+	"sync/atomic"
+
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/utils"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/identity"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/metrics"
+)
+
+func processEvents(ctx context.Context, consumer events.Consumer, stop *atomic.Bool, stopCh chan struct{},
+	backend identity.Backend, m *metrics.Metrics, logger *log.Logger) error {
+	var _registeredEvents = []events.Unmarshaller{
+		events.UserSignedIn{},
+	}
+	evChannel, err := events.Consume(consumer, "graph", _registeredEvents...)
+	if err != nil {
+		logger.Error().Err(err).Msg("cannot consume from nats")
+		return err
+	}
+	logger.Debug().Msg("listening for events")
+	for loop := true; loop; {
+		select {
+		case e := <-evChannel:
+			switch ev := e.Event.(type) {
+			default:
+				// this branch is currently impossible to test and run into because we pick which events we're interested in
+				// through the _registeredEvents above, and the stream won't hand us events we didn't register for
+				m.UnsupportedEvents.Inc()
+				logger.Error().Interface("event", e).Msg("unhandled event")
+			case events.UserSignedIn:
+				name := "UserSignedIn"
+				userId := ""
+				if ev.Executant != nil && ev.Executant.OpaqueId != "" {
+					userId = ev.Executant.OpaqueId
+				} else {
+					m.InvalidEvents.Inc()
+					logger.Error().Err(err).Interface("event", ev).Msg("Received invalid event: executant.opaqueId not set")
+					continue
+				}
+				if err := backend.UpdateLastSignInDate(ctx, userId, utils.TSToTime(ev.Timestamp)); err != nil {
+					m.EventsProcessed.WithLabelValues(name, metrics.ResultFailure).Inc()
+					logger.Error().Err(err).Str("userid", userId).Str("event", name).Msg("Error updating last sign in date")
+				} else {
+					// TODO: UpdateLastSignInDate() currently returns nil instead of an error when the LDAP server is read-only, so those will be accounted for as a success
+					m.EventsProcessed.WithLabelValues(name, metrics.ResultSuccess).Inc()
+					logger.Debug().Str("userid", userId).Str("event", name).Msg("Successfully updated last sign in date")
+				}
+			}
+			if stop.Load() {
+				loop = false
+			}
+		case <-stopCh:
+			logger.Info().Msg("instructed to stop")
+			loop = false
+		case <-ctx.Done():
+			logger.Info().Msg("context cancelled")
+			loop = false
+		}
+	}
+	return nil
+}
+
+type GraphEventConsumer interface {
+	Start() error
+	io.Closer
+}
+
+type GraphEventConsumerImpl struct {
+	ctx      context.Context
+	consumer events.Consumer
+	backend  identity.Backend
+	metrics  *metrics.Metrics
+	logger   *log.Logger
+	stopped  atomic.Bool
+	stopCh   chan struct{}
+}
+
+var _ GraphEventConsumer = &GraphEventConsumerImpl{}
+
+func (g *GraphEventConsumerImpl) Start() error {
+	return processEvents(g.ctx, g.consumer, &g.stopped, g.stopCh, g.backend, g.metrics, g.logger)
+}
+
+func (g *GraphEventConsumerImpl) Close() error {
+	if g.stopped.CompareAndSwap(false, true) {
+		close(g.stopCh)
+	}
+	return nil
+}
+
+type NullGraphEventConsumer struct {
+}
+
+var _ GraphEventConsumer = &NullGraphEventConsumer{}
+
+func (n *NullGraphEventConsumer) Start() error {
+	return nil
+}
+
+func (n *NullGraphEventConsumer) Close() error {
+	return nil
+}
+
+func NewService(ctx context.Context, consumer events.Consumer, backend identity.Backend, metrics *metrics.Metrics, logger *log.Logger) (GraphEventConsumer, error) {
+	if consumer == nil {
+		return &NullGraphEventConsumer{}, nil
+	} else {
+		stopCh := make(chan struct{}, 1)
+		return &GraphEventConsumerImpl{
+			ctx:      ctx,
+			consumer: consumer,
+			backend:  backend,
+			metrics:  metrics,
+			logger:   logger,
+			stopCh:   stopCh,
+		}, nil
+	}
+}

services/graph/pkg/service/events/service_test.go

@@ -0,0 +1,124 @@
+package events_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/rand/v2"
+	"sync"
+	"testing"
+	"time"
+
+	userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/test-go/testify/mock"
+
+	"github.com/opencloud-eu/opencloud/internal/eventstest"
+	"github.com/opencloud-eu/opencloud/internal/metricstest"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/identity/mocks"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/metrics"
+	g "github.com/opencloud-eu/opencloud/services/graph/pkg/service/events"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSuccessfulCall(t *testing.T) {
+	require := require.New(t)
+
+	ctx, cancel := context.WithCancel(t.Context())
+
+	bus := eventstest.NewTestBus()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	userId := fmt.Sprintf("user%d", 1000+rand.IntN(10000))
+
+	backend := mocks.NewBackend(t)
+	backend.EXPECT().UpdateLastSignInDate(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ string, _ time.Time) error {
+		defer wg.Done()
+		return nil
+	})
+
+	reg := prometheus.NewRegistry()
+	m := metrics.New(reg)
+
+	logger := log.NewLogger()
+
+	svc, err := g.NewService(ctx, bus, backend, m, &logger)
+	require.NoError(err)
+	t.Cleanup(func() { svc.Close() })
+	t.Cleanup(cancel)
+	go func() {
+		require.NoError(svc.Start())
+	}()
+
+	metricstest.RequireEqual(t, 0, m.UnsupportedEvents)
+	metricstest.RequireIsNotSet(t, m.EventsProcessed)
+
+	_ = bus.Publish(events.UserSignedIn{
+		Timestamp: nil,
+		Executant: &userv1beta1.UserId{
+			OpaqueId: userId,
+		},
+	})
+
+	wg.Wait()
+	require.Len(backend.Mock.Calls, 1)
+	require.Len(backend.Mock.Calls[0].Arguments, 3)
+	require.Equal(userId, backend.Mock.Calls[0].Arguments[1])
+
+	metricstest.RequireEqual(t, 0, m.UnsupportedEvents)
+	metricstest.RequireEqualWithLabels(t, 1, map[string]string{"event": "UserSignedIn", "result": "success"}, m.EventsProcessed)
+}
+
+func TestBackendReturningAnError(t *testing.T) {
+	require := require.New(t)
+
+	ctx, cancel := context.WithCancel(t.Context())
+
+	bus := eventstest.NewTestBus()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	userId := fmt.Sprintf("user%d", 1000+rand.IntN(10000))
+
+	backend := mocks.NewBackend(t)
+	backend.EXPECT().UpdateLastSignInDate(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ string, _ time.Time) error {
+		defer wg.Done()
+		return errors.New("test")
+	})
+
+	reg := prometheus.NewRegistry()
+	m := metrics.New(reg)
+
+	logger := log.NewLogger()
+
+	svc, err := g.NewService(ctx, bus, backend, m, &logger)
+	require.NoError(err)
+	t.Cleanup(func() { svc.Close() })
+	t.Cleanup(cancel)
+	go func() {
+		require.NoError(svc.Start())
+	}()
+
+	metricstest.RequireEqual(t, 0, m.UnsupportedEvents)
+	metricstest.RequireIsNotSet(t, m.EventsProcessed)
+
+	_ = bus.Publish(events.UserSignedIn{
+		Timestamp: nil,
+		Executant: &userv1beta1.UserId{
+			OpaqueId: userId,
+		},
+	})
+
+	wg.Wait()
+	require.Len(backend.Mock.Calls, 1)
+	require.Len(backend.Mock.Calls[0].Arguments, 3)
+	require.Equal(userId, backend.Mock.Calls[0].Arguments[1])
+
+	metricstest.RequireEqual(t, 0, m.UnsupportedEvents)
+	metricstest.RequireEqualWithLabels(t, 1, map[string]string{"event": "UserSignedIn", "result": "failure"}, m.EventsProcessed)
+}

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

@@ -63,7 +63,6 @@ type Graph struct {
 	valueService             settingssvc.ValueService
 	specialDriveItemsCache   *ttlcache.Cache[string, any]
 	eventsPublisher          events.Publisher
-	eventsConsumer           events.Consumer
 	searchService            searchsvc.SearchProviderService
 	keycloakClient           keycloak.Client
 	historyClient            ehsvc.EventHistoryService

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

@@ -39,7 +39,6 @@ type Options struct {
 	ValueService             settingssvc.ValueService
 	RoleManager              *roles.Manager
 	EventsPublisher          events.Publisher
-	EventsConsumer           events.Consumer
 	SearchService            searchsvc.SearchProviderService
 	KeycloakClient           keycloak.Client
 	EventHistoryClient       ehsvc.EventHistoryService
@@ -163,13 +162,6 @@ func EventsPublisher(val events.Publisher) Option {
 	}
 }
 
-// EventsConsumer provides a function to set the EventsConsumer option.
-func EventsConsumer(val events.Consumer) Option {
-	return func(o *Options) {
-		o.EventsConsumer = val
-	}
-}
-
 // KeycloakClient provides a function to set the KeycloakCient option.
 func KeycloakClient(val keycloak.Client) Option {
 	return func(o *Options) {

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

@@ -1,39 +1,25 @@
 package svc
 
 import (
-	"context"
-	"crypto/tls"
-	"crypto/x509"
-	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
-	"os"
 	"strconv"
 	"time"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/chi/v5/middleware"
-	ldapv3 "github.com/go-ldap/ldap/v3"
 	"github.com/jellydator/ttlcache/v3"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/identity/cache"
 	"github.com/riandyrn/otelchi"
 	microstore "go-micro.dev/v4/store"
 
-	"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/store"
-	"github.com/opencloud-eu/reva/v2/pkg/utils"
-	"github.com/opencloud-eu/reva/v2/pkg/utils/ldap"
 
-	ocldap "github.com/opencloud-eu/opencloud/pkg/ldap"
-	"github.com/opencloud-eu/opencloud/pkg/log"
-	"github.com/opencloud-eu/opencloud/pkg/registry"
 	"github.com/opencloud-eu/opencloud/pkg/roles"
 	"github.com/opencloud-eu/opencloud/pkg/service/grpc"
 	"github.com/opencloud-eu/opencloud/pkg/tracing"
 	settingssvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
-	"github.com/opencloud-eu/opencloud/services/graph/pkg/identity"
 	graphm "github.com/opencloud-eu/opencloud/services/graph/pkg/middleware"
 	"github.com/opencloud-eu/opencloud/services/graph/pkg/unifiedrole"
 )
@@ -199,8 +185,8 @@ func NewService(opts ...Option) (Graph, error) { //nolint:maintidx
 		mux:                      m,
 		specialDriveItemsCache:   spacePropertiesCache,
 		eventsPublisher:          options.EventsPublisher,
-		eventsConsumer:           options.EventsConsumer,
 		searchService:            options.SearchService,
+		identityBackend:          options.IdentityBackend,
 		identityEducationBackend: options.IdentityEducationBackend,
 		keycloakClient:           options.KeycloakClient,
 		historyClient:            options.EventHistoryClient,
@@ -209,10 +195,6 @@ func NewService(opts ...Option) (Graph, error) { //nolint:maintidx
 		natskv:                   options.NatsKeyValue,
 	}
 
-	if err := setIdentityBackends(options, &svc); err != nil {
-		return svc, err
-	}
-
 	if options.PermissionService == nil {
 		grpcClient, err := grpc.NewClient(append(grpc.GetClientOptions(options.Config.GRPCClientTLS), grpc.WithTraceProvider(options.TraceProvider))...)
 		if err != nil {
@@ -450,164 +432,6 @@ func NewService(opts ...Option) (Graph, error) { //nolint:maintidx
 	return svc, nil
 }
 
-func setIdentityBackends(options Options, svc *Graph) error {
-	if options.IdentityBackend == nil {
-		switch options.Config.Identity.Backend {
-		case "cs3":
-			gatewaySelector, err := pool.GatewaySelector(
-				options.Config.Reva.Address,
-				append(
-					options.Config.Reva.GetRevaOptions(),
-					pool.WithRegistry(registry.GetRegistry()),
-					pool.WithTracerProvider(options.TraceProvider),
-				)...,
-			)
-			if err != nil {
-				return err
-			}
-
-			svc.identityBackend = &identity.CS3{
-				Config:          options.Config.Reva,
-				Logger:          &options.Logger,
-				GatewaySelector: gatewaySelector,
-			}
-		case "ldap":
-			var err error
-
-			var tlsConf *tls.Config
-			if options.Config.Identity.LDAP.Insecure {
-
-				// When insecure is set to true then we don't need a certificate.
-				options.Config.Identity.LDAP.CACert = ""
-				tlsConf = &tls.Config{
-					MinVersion: tls.VersionTLS12,
-
-					//nolint:gosec // We need the ability to run with "insecure" (dev/testing)
-					InsecureSkipVerify: options.Config.Identity.LDAP.Insecure,
-				}
-			}
-
-			if options.Config.Identity.LDAP.CACert != "" {
-				if err := ocldap.WaitForCA(options.Logger,
-					options.Config.Identity.LDAP.Insecure,
-					options.Config.Identity.LDAP.CACert); err != nil {
-					options.Logger.Fatal().Err(err).Msg("The configured LDAP CA cert does not exist")
-				}
-				if tlsConf == nil {
-					tlsConf = &tls.Config{
-						MinVersion: tls.VersionTLS12,
-					}
-				}
-				certs := x509.NewCertPool()
-				pemData, err := os.ReadFile(options.Config.Identity.LDAP.CACert)
-				if err != nil {
-					options.Logger.Error().Err(err).Msg("Error initializing LDAP Backend")
-					return err
-				}
-				if !certs.AppendCertsFromPEM(pemData) {
-					options.Logger.Error().Msg("Error initializing LDAP Backend. Adding CA cert failed")
-					return err
-				}
-				tlsConf.RootCAs = certs
-			}
-
-			conn := ldap.NewLDAPWithReconnect(
-				ldap.Config{
-					URI:          options.Config.Identity.LDAP.URI,
-					BindDN:       options.Config.Identity.LDAP.BindDN,
-					BindPassword: options.Config.Identity.LDAP.BindPassword,
-					TLSConfig:    tlsConf,
-				},
-			)
-			conn.SetLogger(&options.Logger.Logger)
-			lb, err := identity.NewLDAPBackend(conn, options.Config.Identity.LDAP, &options.Logger)
-			if err != nil {
-				options.Logger.Error().Err(err).Msg("Error initializing LDAP Backend")
-				return err
-			}
-			svc.identityBackend = lb
-			if options.IdentityEducationBackend == nil {
-				if options.Config.Identity.LDAP.EducationResourcesEnabled {
-					svc.identityEducationBackend = lb
-				} else {
-					errEduBackend := &identity.ErrEducationBackend{}
-					svc.identityEducationBackend = errEduBackend
-				}
-			}
-
-			disableMechanismType, err := identity.ParseDisableMechanismType(options.Config.Identity.LDAP.DisableUserMechanism)
-			if err != nil {
-				options.Logger.Error().Err(err).Msg("Error initializing LDAP Backend")
-				return err
-			}
-
-			if disableMechanismType == identity.DisableMechanismGroup {
-				options.Logger.Info().Msg("LocalUserDisable is true, will create group if not exists")
-				err := lb.CreateLDAPGroupByDN(options.Config.Identity.LDAP.LdapDisabledUsersGroupDN)
-				if err != nil {
-					isAnError := false
-					var lerr *ldapv3.Error
-					if errors.As(err, &lerr) {
-						if lerr.ResultCode != ldapv3.LDAPResultEntryAlreadyExists {
-							isAnError = true
-						}
-					} else {
-						isAnError = true
-					}
-
-					if isAnError {
-						msg := "error adding group for disabling users"
-						options.Logger.Error().Err(err).Str("local_user_disable", options.Config.Identity.LDAP.LdapDisabledUsersGroupDN).Msg(msg)
-						return err
-					}
-				}
-			}
-
-		default:
-			err := fmt.Errorf("unknown identity backend: '%s'", options.Config.Identity.Backend)
-			options.Logger.Err(err)
-			return err
-		}
-	} else {
-		svc.identityBackend = options.IdentityBackend
-	}
-
-	return svc.StartListenForLogonEvents(options.Context, options.Logger)
-}
-
-func (g *Graph) StartListenForLogonEvents(ctx context.Context, l log.Logger) error {
-	if g.eventsConsumer == nil {
-		return nil
-	}
-	var _registeredEvents = []events.Unmarshaller{
-		events.UserSignedIn{},
-	}
-	evChannel, err := events.Consume(g.eventsConsumer, "graph", _registeredEvents...)
-	if err != nil {
-		l.Error().Err(err).Msg("cannot consume from nats")
-		return err
-	}
-	go func() {
-		for loop := true; loop; {
-			select {
-			case e := <-evChannel:
-				switch ev := e.Event.(type) {
-				default:
-					l.Error().Interface("event", e).Msg("unhandled event")
-				case events.UserSignedIn:
-					if err := g.identityBackend.UpdateLastSignInDate(ctx, ev.Executant.OpaqueId, utils.TSToTime(ev.Timestamp)); err != nil {
-						l.Error().Err(err).Str("userid", ev.Executant.OpaqueId).Msg("Error updating last sign in date")
-					}
-				}
-			case <-ctx.Done():
-				l.Info().Msg("context cancelled")
-				loop = false
-			}
-		}
-	}()
-	return nil
-}
-
 // parseHeaderPurge parses the 'Purge' header.
 // '1', 't', 'T', 'TRUE', 'true', 'True' are parsed as true
 // all other values are false.