eventhistory service: fix consumer isolation and conditional initialization

services/eventhistory/pkg/command/server.go

@@ -6,19 +6,19 @@ import (
 	"os/signal"
 
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
-	"github.com/opencloud-eu/opencloud/pkg/generators"
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/pkg/runner"
 	ogrpc "github.com/opencloud-eu/opencloud/pkg/service/grpc"
 	"github.com/opencloud-eu/opencloud/pkg/tracing"
 	"github.com/opencloud-eu/opencloud/pkg/version"
+	ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config/parser"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/metrics"
+	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/consumer"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/debug"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/grpc"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
+	svc "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service"
 	"github.com/opencloud-eu/reva/v2/pkg/store"
 
 	"github.com/spf13/cobra"
@@ -59,25 +59,6 @@ func Server(cfg *config.Config) *cobra.Command {
 
 			gr := runner.NewGroup()
 
-			var consumer events.Stream
-			if !cfg.Events.Disabled {
-				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
-				consumer, 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
-				}
-			} else {
-				logger.Info().Msg("event listening disabled, not starting event consumer")
-			}
-
 			st := store.Create(
 				store.Store(cfg.Store.Store),
 				store.TTL(cfg.Store.TTL),
@@ -90,7 +71,27 @@ func Server(cfg *config.Config) *cobra.Command {
 				store.TLSRootCA(cfg.Store.TLSRootCACertificate),
 			)
 
+			if !cfg.Events.Disabled {
+				evConsumer, err := consumer.NewConsumer(
+					consumer.Logger(logger),
+					consumer.Config(cfg),
+					consumer.Persistence(st),
+				)
+				if err != nil {
+					return err
+				}
+
+				go evConsumer.StoreEvents()
+			} else {
+				logger.Info().Msg("event listening disabled, not starting event consumer")
+			}
+
 			if !cfg.GRPC.Disabled {
+				eh, err := svc.NewEventHistoryService(cfg, st, logger)
+				if err != nil {
+					return err
+				}
+
 				service := grpc.NewService(
 					grpc.Logger(logger),
 					grpc.Context(ctx),
@@ -99,11 +100,13 @@ func Server(cfg *config.Config) *cobra.Command {
 					grpc.Namespace(cfg.GRPC.Namespace),
 					grpc.Address(cfg.GRPC.Addr),
 					grpc.Metrics(m),
-					grpc.Consumer(consumer),
-					grpc.Persistence(st),
 					grpc.TraceProvider(traceProvider),
 				)
 
+				if err := ehsvc.RegisterEventHistoryServiceHandler(service.Server(), eh); err != nil {
+					return err
+				}
+
 				gr.Add(runner.NewGoMicroGrpcServerRunner(cfg.Service.Name+".grpc", service))
 			} else {
 				logger.Info().Msg("gRPC server disabled, not starting gRPC service")

services/eventhistory/pkg/server/consumer/option.go

@@ -0,0 +1,58 @@
+package consumer
+
+import (
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"go-micro.dev/v4/store"
+)
+
+// 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
+	Config      *config.Config
+	Persistence store.Store
+	Consumer    events.Consumer
+}
+
+// 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
+	}
+}
+
+// Config provides a function to set the config option.
+func Config(val *config.Config) Option {
+	return func(o *Options) {
+		o.Config = val
+	}
+}
+
+// Persistence provides a function to configure the store
+func Persistence(store store.Store) Option {
+	return func(o *Options) {
+		o.Persistence = store
+	}
+}
+
+// Stream provides a function to set the event source to consume from.
+func Stream(consumer events.Consumer) Option {
+	return func(o *Options) {
+		o.Consumer = consumer
+	}
+}

services/eventhistory/pkg/server/consumer/server.go

@@ -0,0 +1,78 @@
+package consumer
+
+import (
+	"encoding/json"
+
+	"github.com/opencloud-eu/opencloud/pkg/generators"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config"
+	svc "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
+	"go-micro.dev/v4/store"
+)
+
+// Consumer consumes all events and stores them in the store
+type Consumer struct {
+	ch    <-chan events.Event
+	store store.Store
+	cfg   *config.Config
+	log   log.Logger
+}
+
+// NewConsumer initializes the event consumer
+func NewConsumer(opts ...Option) (*Consumer, error) {
+	options := newOptions(opts...)
+
+	cons := options.Consumer
+	if cons == nil {
+		connName := generators.GenerateConnectionName(options.Config.Service.Name, generators.NTypeBus)
+		var err error
+		cons, err = stream.NatsFromConfig(connName, false, stream.NatsConfig{
+			Endpoint:             options.Config.Events.Endpoint,
+			Cluster:              options.Config.Events.Cluster,
+			TLSInsecure:          options.Config.Events.TLSInsecure,
+			TLSRootCACertificate: options.Config.Events.TLSRootCACertificate,
+			EnableTLS:            options.Config.Events.EnableTLS,
+			AuthUsername:         options.Config.Events.AuthUsername,
+			AuthPassword:         options.Config.Events.AuthPassword,
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	ch, err := events.ConsumeAll(cons, "evhistory")
+	if err != nil {
+		return nil, err
+	}
+
+	return &Consumer{ch: ch, store: options.Persistence, cfg: options.Config, log: options.Logger}, nil
+}
+
+// StoreEvents consumes all events and stores them in the store. Will block
+func (c *Consumer) StoreEvents() {
+	for event := range c.ch {
+		ev, err := json.Marshal(svc.StoreEvent{
+			ID:    event.ID,
+			Type:  event.Type,
+			Event: event.Event.([]byte),
+		})
+		if err != nil {
+			c.log.Error().Err(err).Str("eventid", event.ID).Msg("could not marshal event")
+			continue
+		}
+		if err := c.store.Write(&store.Record{
+			Key:    event.ID,
+			Value:  ev,
+			Expiry: c.cfg.Store.TTL,
+			Metadata: map[string]any{
+				"type": event.Type,
+			},
+		}); err != nil {
+			// we can't store. That's it for us.
+			c.log.Error().Err(err).Str("eventid", event.ID).Msg("could not store event")
+			continue
+		}
+	}
+}

services/eventhistory/pkg/server/grpc/option.go

@@ -6,8 +6,6 @@ import (
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/metrics"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"go-micro.dev/v4/store"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -23,8 +21,6 @@ type Options struct {
 	Config        *config.Config
 	Metrics       *metrics.Metrics
 	Namespace     string
-	Persistence   store.Store
-	Consumer      events.Consumer
 	TraceProvider trace.TracerProvider
 }
 
@@ -88,20 +84,6 @@ func Namespace(val string) Option {
 	}
 }
 
-// Persistence provides a function to configure the store
-func Persistence(store store.Store) Option {
-	return func(o *Options) {
-		o.Persistence = store
-	}
-}
-
-// Consumer provides a function to configure the consumer
-func Consumer(consumer events.Consumer) Option {
-	return func(o *Options) {
-		o.Consumer = consumer
-	}
-}
-
 // TraceProvider provides a function to configure the trace provider
 func TraceProvider(traceProvider trace.TracerProvider) Option {
 	return func(o *Options) {

services/eventhistory/pkg/server/grpc/server.go

@@ -3,8 +3,6 @@ package grpc
 import (
 	"github.com/opencloud-eu/opencloud/pkg/service/grpc"
 	"github.com/opencloud-eu/opencloud/pkg/version"
-	ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0"
-	svc "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service"
 )
 
 // NewService initializes the grpc service and server.
@@ -32,16 +30,5 @@ func NewService(opts ...Option) grpc.Service {
 		return grpc.Service{}
 	}
 
-	eh, err := svc.NewEventHistoryService(options.Config, options.Consumer, options.Persistence, options.Logger)
-	if err != nil {
-		options.Logger.Fatal().Err(err).Msg("Error creating event history service")
-		return grpc.Service{}
-	}
-
-	_ = ehsvc.RegisterEventHistoryServiceHandler(
-		service.Server(),
-		eh,
-	)
-
 	return service
 }

services/eventhistory/pkg/service/service.go

@@ -11,7 +11,6 @@ import (
 	ehmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/eventhistory/v0"
 	ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
 	"go-micro.dev/v4/store"
 )
 
@@ -24,54 +23,14 @@ type StoreEvent struct {
 
 // EventHistoryService is the service responsible for event history
 type EventHistoryService struct {
-	ch    <-chan events.Event
 	store store.Store
 	cfg   *config.Config
 	log   log.Logger
 }
 
 // NewEventHistoryService returns an EventHistory service
-func NewEventHistoryService(cfg *config.Config, consumer events.Consumer, store store.Store, log log.Logger) (*EventHistoryService, error) {
-	eh := &EventHistoryService{store: store, cfg: cfg, log: log}
-
-	if consumer != nil {
-		ch, err := events.ConsumeAll(consumer, "evhistory")
-		if err != nil {
-			return nil, err
-		}
-
-		eh.ch = ch
-		go eh.StoreEvents()
-	}
-
-	return eh, nil
-}
-
-// StoreEvents consumes all events and stores them in the store. Will block
-func (eh *EventHistoryService) StoreEvents() {
-	for event := range eh.ch {
-		ev, err := json.Marshal(StoreEvent{
-			ID:    event.ID,
-			Type:  event.Type,
-			Event: event.Event.([]byte),
-		})
-		if err != nil {
-			eh.log.Error().Err(err).Str("eventid", event.ID).Msg("could not marshal event")
-			continue
-		}
-		if err := eh.store.Write(&store.Record{
-			Key:    event.ID,
-			Value:  ev,
-			Expiry: eh.cfg.Store.TTL,
-			Metadata: map[string]any{
-				"type": event.Type,
-			},
-		}); err != nil {
-			// we can't store. That's it for us.
-			eh.log.Error().Err(err).Str("eventid", event.ID).Msg("could not store event")
-			continue
-		}
-	}
+func NewEventHistoryService(cfg *config.Config, store store.Store, log log.Logger) (*EventHistoryService, error) {
+	return &EventHistoryService{store: store, cfg: cfg, log: log}, nil
 }
 
 // GetEvents allows retrieving events from the eventstore by id

services/eventhistory/pkg/service/service_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/consumer"
 	"github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service"
 	"github.com/opencloud-eu/reva/v2/pkg/events"
 	"github.com/opencloud-eu/reva/v2/pkg/store"
@@ -34,103 +35,123 @@ var _ = Describe("EventHistoryService", func() {
 	BeforeEach(func() {
 		var err error
 		sto = store.Create()
-		bus = testBus(make(chan events.Event))
-		eh, err = service.NewEventHistoryService(cfg, bus, sto, log.Logger{})
+		eh, err = service.NewEventHistoryService(cfg, sto, log.Logger{})
 		Expect(err).ToNot(HaveOccurred())
 	})
 
-	AfterEach(func() {
-		close(bus)
-	})
+	Context("with event consumer", func() {
+		BeforeEach(func() {
+			bus = testBus(make(chan events.Event))
+			evConsumer, err := consumer.NewConsumer(
+				consumer.Logger(log.Logger{}),
+				consumer.Config(cfg),
+				consumer.Persistence(sto),
+				consumer.Stream(bus),
+			)
+			Expect(err).ToNot(HaveOccurred())
+			go evConsumer.StoreEvents()
+		})
 
-	It("Records events, stores them and allows them to be retrieved", func() {
-		id := bus.Publish(events.UploadReady{})
+		AfterEach(func() {
+			close(bus)
+		})
 
-		// service will store eventually
-		time.Sleep(500 * time.Millisecond)
+		It("Records events, stores them and allows them to be retrieved", func() {
+			id := bus.Publish(events.UploadReady{})
 
-		resp := &ehsvc.GetEventsResponse{}
-		err := eh.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp)
-		Expect(err).ToNot(HaveOccurred())
-		Expect(resp).ToNot(BeNil())
+			// service will store eventually
+			time.Sleep(500 * time.Millisecond)
 
-		Expect(len(resp.Events)).To(Equal(1))
-		Expect(resp.Events[0].Id).To(Equal(id))
-	})
+			resp := &ehsvc.GetEventsResponse{}
+			err := eh.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(resp).ToNot(BeNil())
 
-	It("Gets all events", func() {
-		ids := make([]string, 3)
-		ids[0] = bus.Publish(events.UploadReady{
-			ExecutingUser: &userv1beta1.User{
-				Id: &userv1beta1.UserId{
-					OpaqueId: "test-id",
+			Expect(len(resp.Events)).To(Equal(1))
+			Expect(resp.Events[0].Id).To(Equal(id))
+		})
+
+		It("Gets all events", func() {
+			ids := make([]string, 3)
+			ids[0] = bus.Publish(events.UploadReady{
+				ExecutingUser: &userv1beta1.User{
+					Id: &userv1beta1.UserId{
+						OpaqueId: "test-id",
+					},
 				},
-			},
-			Failed:    false,
-			Timestamp: utils.TimeToTS(time.Time{}),
-		})
-		ids[1] = bus.Publish(events.UserCreated{
-			UserID: "another-id",
-		})
-		ids[2] = bus.Publish(events.UserDeleted{
-			Executant: &userv1beta1.UserId{
-				OpaqueId: "another-id",
-			},
-			UserID: "test-id",
+				Failed:    false,
+				Timestamp: utils.TimeToTS(time.Time{}),
+			})
+			ids[1] = bus.Publish(events.UserCreated{
+				UserID: "another-id",
+			})
+			ids[2] = bus.Publish(events.UserDeleted{
+				Executant: &userv1beta1.UserId{
+					OpaqueId: "another-id",
+				},
+				UserID: "test-id",
+			})
+
+			time.Sleep(500 * time.Millisecond)
+
+			resp := &ehsvc.GetEventsResponse{}
+			err := eh.GetEventsForUser(context.Background(), &ehsvc.GetEventsForUserRequest{UserID: "test-id"}, resp)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(resp).ToNot(BeNil())
+
+			// Events don't always come back in the same order as they were sent, so we need to sort them and
+			// do the same for the expected IDs as well.
+			expectedIDs := []string{ids[0], ids[2]}
+			sort.Strings(expectedIDs)
+			var gotIDs []string
+			for _, ev := range resp.Events {
+				gotIDs = append(gotIDs, ev.Id)
+			}
+			sort.Strings(gotIDs)
+
+			Expect(len(gotIDs)).To(Equal(len(expectedIDs)))
+			Expect(gotIDs[0]).To(Equal(expectedIDs[0]))
+			Expect(gotIDs[1]).To(Equal(expectedIDs[1]))
 		})
 
-		time.Sleep(500 * time.Millisecond)
+		It("Stores events and verifies them in the store directly (consumer-only mode)", func() {
+			id := bus.Publish(events.UploadReady{})
+			time.Sleep(500 * time.Millisecond)
 
-		resp := &ehsvc.GetEventsResponse{}
-		err := eh.GetEventsForUser(context.Background(), &ehsvc.GetEventsForUserRequest{UserID: "test-id"}, resp)
-		Expect(err).ToNot(HaveOccurred())
-		Expect(resp).ToNot(BeNil())
+			records, err := sto.Read(id)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(len(records)).To(Equal(1))
 
-		// Events don't always come back in the same order as they were sent, so we need to sort them and
-		// do the same for the expected IDs as well.
-		expectedIDs := []string{ids[0], ids[2]}
-		sort.Strings(expectedIDs)
-		var gotIDs []string
-		for _, ev := range resp.Events {
-			gotIDs = append(gotIDs, ev.Id)
-		}
-		sort.Strings(gotIDs)
-
-		Expect(len(gotIDs)).To(Equal(len(expectedIDs)))
-		Expect(gotIDs[0]).To(Equal(expectedIDs[0]))
-		Expect(gotIDs[1]).To(Equal(expectedIDs[1]))
+			var stored service.StoreEvent
+			err = json.Unmarshal(records[0].Value, &stored)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(stored.ID).To(Equal(id))
+		})
 	})
 
 	It("Starts without consumer and reads events from shared store", func() {
-		ehNilConsumer, err := service.NewEventHistoryService(cfg, nil, sto, log.Logger{})
+		// A separate consumer instance stores events into the store shared with the
+		// consumer-less instance (gRPC-only mode), which can still find them.
+		bus = testBus(make(chan events.Event))
+		defer close(bus)
+		evConsumer, err := consumer.NewConsumer(
+			consumer.Logger(log.Logger{}),
+			consumer.Config(cfg),
+			consumer.Persistence(sto),
+			consumer.Stream(bus),
+		)
 		Expect(err).ToNot(HaveOccurred())
-		Expect(ehNilConsumer).ToNot(BeNil())
+		go evConsumer.StoreEvents()
 
-		// We assume that the event was consumed and stored by the original instance, so the nil-consumer
-		// instance (gRPC-only mode) can still find it in the shared store.
 		id := bus.Publish(events.UploadReady{})
 		time.Sleep(500 * time.Millisecond)
 
 		resp := &ehsvc.GetEventsResponse{}
-		err = ehNilConsumer.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp)
+		err = eh.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(len(resp.Events)).To(Equal(1))
 		Expect(resp.Events[0].Id).To(Equal(id))
 	})
-
-	It("Stores events and verifies them in the store directly (consumer-only mode)", func() {
-		id := bus.Publish(events.UploadReady{})
-		time.Sleep(500 * time.Millisecond)
-
-		records, err := sto.Read(id)
-		Expect(err).ToNot(HaveOccurred())
-		Expect(len(records)).To(Equal(1))
-
-		var stored service.StoreEvent
-		err = json.Unmarshal(records[0].Value, &stored)
-		Expect(err).ToNot(HaveOccurred())
-		Expect(stored.ID).To(Equal(id))
-	})
 })
 
 type testBus chan events.Event