yesBad/proton-bridge
1
0
Fork 0
forked from Silverfish/proton-bridge
Code Actions Activity

chore(GODT-2848): Simplify User Event Service

Browse Source 
Rather than having the services subscribe to each individual event type,
have only one subscriber for the whole event object. This spreads the
logic to the service through the `EventHandler` type.

This is important for the next stage where the IMAP service will be
required to apply events from the past.
This commit is contained in:
Leander Beernaert
2023-08-09 16:09:21 +02:00
parent e8ce8942be
commit 4ee6da4baa
16 changed files with 597 additions and 878 deletions
Download Patch File Download Diff File Expand all files Collapse all files

222
internal/services/useridentity/service.go
View File

@ -47,10 +47,7 @@ type Service struct {
log *logrus.Entry
identity State
userSubscriber *userevents.UserChanneledSubscriber
addressSubscriber *userevents.AddressChanneledSubscriber
usedSpaceSubscriber *userevents.UserUsedSpaceChanneledSubscriber
refreshSubscriber *userevents.RefreshChanneledSubscriber
subscription *userevents.EventChanneledSubscriber
bridgePassProvider BridgePassProvider
telemetry Telemetry
@ -74,12 +71,9 @@ func NewService(
"service": "user-identity",
"user": state.User.ID,
}),
userSubscriber: userevents.NewUserSubscriber(subscriberName),
refreshSubscriber: userevents.NewRefreshSubscriber(subscriberName),
addressSubscriber: userevents.NewAddressSubscriber(subscriberName),
usedSpaceSubscriber: userevents.NewUserUsedSpaceSubscriber(subscriberName),
bridgePassProvider: bridgePassProvider,
telemetry: telemetry,
subscription: userevents.NewEventSubscriber(subscriberName),
bridgePassProvider: bridgePassProvider,
telemetry: telemetry,
}
}
@ -108,134 +102,30 @@ func (s *Service) CheckAuth(ctx context.Context, email string, password []byte)
})
}
func (s *Service) run(ctx context.Context) {
s.log.WithFields(logrus.Fields{
"numAddr": len(s.identity.Addresses),
}).Info("Starting user identity service")
defer s.log.Info("Exiting Service")
func (s *Service) HandleUsedSpaceEvent(ctx context.Context, newSpace int) error {
s.log.Info("Handling User Space Changed event")
s.registerSubscription()
defer s.unregisterSubscription()
defer s.cpc.Close()
for {
select {
case <-ctx.Done():
return
case r, ok := <-s.cpc.ReceiveCh():
if !ok {
continue
}
switch req := r.Value().(type) {
case *resyncReq:
err := s.identity.OnRefreshEvent(ctx)
r.Reply(ctx, nil, err)
case *getUserReq:
r.Reply(ctx, s.identity.User, nil)
case *getAddressesReq:
r.Reply(ctx, maps.Clone(s.identity.Addresses), nil)
case *checkAuthReq:
id, err := s.identity.CheckAuth(req.email, req.password, s.bridgePassProvider, s.telemetry)
r.Reply(ctx, id, err)
default:
s.log.Error("Invalid request")
}
case evt, ok := <-s.userSubscriber.OnEventCh():
if !ok {
continue
}
evt.Consume(func(user proton.User) error {
s.onUserEvent(ctx, user)
return nil
})
case evt, ok := <-s.refreshSubscriber.OnEventCh():
if !ok {
continue
}
evt.Consume(func(_ proton.RefreshFlag) error {
return s.onRefreshEvent(ctx)
})
case evt, ok := <-s.usedSpaceSubscriber.OnEventCh():
if !ok {
continue
}
evt.Consume(func(usedSpace int) error {
s.onUserSpaceChanged(ctx, usedSpace)
return nil
})
case evt, ok := <-s.addressSubscriber.OnEventCh():
if !ok {
continue
}
evt.Consume(func(events []proton.AddressEvent) error {
return s.onAddressEvent(ctx, events)
})
}
if s.identity.OnUserSpaceChanged(newSpace) {
s.eventPublisher.PublishEvent(ctx, events.UsedSpaceChanged{
UserID: s.identity.User.ID,
UsedSpace: newSpace,
})
}
return nil
}
func (s *Service) registerSubscription() {
s.eventService.Subscribe(userevents.Subscription{
Refresh: s.refreshSubscriber,
User: s.userSubscriber,
Address: s.addressSubscriber,
UserUsedSpace: s.usedSpaceSubscriber,
})
}
func (s *Service) unregisterSubscription() {
s.eventService.Unsubscribe(userevents.Subscription{
Refresh: s.refreshSubscriber,
User: s.userSubscriber,
Address: s.addressSubscriber,
UserUsedSpace: s.usedSpaceSubscriber,
})
}
func (s *Service) onUserEvent(ctx context.Context, user proton.User) {
func (s *Service) HandleUserEvent(ctx context.Context, user *proton.User) error {
s.log.WithField("username", logging.Sensitive(user.Name)).Info("Handling user event")
s.identity.OnUserEvent(user)
s.identity.OnUserEvent(*user)
s.eventPublisher.PublishEvent(ctx, events.UserChanged{
UserID: user.ID,
})
}
func (s *Service) onRefreshEvent(ctx context.Context) error {
s.log.Info("Handling refresh event")
if err := s.identity.OnRefreshEvent(ctx); err != nil {
s.log.WithError(err).Error("Failed to handle refresh event")
return err
}
s.eventPublisher.PublishEvent(ctx, events.UserRefreshed{
UserID: s.identity.User.ID,
CancelEventPool: false,
})
return nil
}
func (s *Service) onUserSpaceChanged(ctx context.Context, value int) {
s.log.Info("Handling User Space Changed event")
if !s.identity.OnUserSpaceChanged(value) {
return
}
s.eventPublisher.PublishEvent(ctx, events.UsedSpaceChanged{
UserID: s.identity.User.ID,
UsedSpace: value,
})
}
func (s *Service) onAddressEvent(ctx context.Context, addressEvents []proton.AddressEvent) error {
func (s *Service) HandleAddressEvents(ctx context.Context, addressEvents []proton.AddressEvent) error {
s.log.Infof("Handling Address Events (%v)", len(addressEvents))
for idx, event := range addressEvents {
@ -305,6 +195,86 @@ func (s *Service) onAddressEvent(ctx context.Context, addressEvents []proton.Add
return nil
}
func (s *Service) HandleRefreshEvent(ctx context.Context, _ proton.RefreshFlag) error {
s.log.Info("Handling refresh event")
if err := s.identity.OnRefreshEvent(ctx); err != nil {
s.log.WithError(err).Error("Failed to handle refresh event")
return err
}
s.eventPublisher.PublishEvent(ctx, events.UserRefreshed{
UserID: s.identity.User.ID,
CancelEventPool: false,
})
return nil
}
func (s *Service) run(ctx context.Context) {
s.log.WithFields(logrus.Fields{
"numAddr": len(s.identity.Addresses),
}).Info("Starting user identity service")
defer s.log.Info("Exiting Service")
eventHandler := userevents.EventHandler{
UserHandler: s,
AddressHandler: s,
UsedSpaceHandler: s,
RefreshHandler: s,
}
s.registerSubscription()
defer s.unregisterSubscription()
defer s.cpc.Close()
for {
select {
case <-ctx.Done():
return
case r, ok := <-s.cpc.ReceiveCh():
if !ok {
continue
}
switch req := r.Value().(type) {
case *resyncReq:
err := s.identity.OnRefreshEvent(ctx)
r.Reply(ctx, nil, err)
case *getUserReq:
r.Reply(ctx, s.identity.User, nil)
case *getAddressesReq:
r.Reply(ctx, maps.Clone(s.identity.Addresses), nil)
case *checkAuthReq:
id, err := s.identity.CheckAuth(req.email, req.password, s.bridgePassProvider, s.telemetry)
r.Reply(ctx, id, err)
default:
s.log.Error("Invalid request")
}
case evt, ok := <-s.subscription.OnEventCh():
if !ok {
continue
}
evt.Consume(func(event proton.Event) error {
return eventHandler.OnEvent(ctx, event)
})
}
}
}
func (s *Service) registerSubscription() {
s.eventService.Subscribe(s.subscription)
}
func (s *Service) unregisterSubscription() {
s.eventService.Unsubscribe(s.subscription)
}
func sortAddresses(addr []proton.Address) []proton.Address {
slices.SortFunc(addr, func(a, b proton.Address) bool {
return a.Order < b.Order

42
internal/services/useridentity/service_test.go
View File

@ -39,7 +39,7 @@ func TestService_OnUserEvent(t *testing.T) {
eventPublisher.EXPECT().PublishEvent(gomock.Any(), gomock.Eq(events.UserChanged{UserID: TestUserID})).Times(1)
service.onUserEvent(context.Background(), newTestUser())
require.NoError(t, service.HandleUserEvent(context.Background(), newTestUser()))
}
func TestService_OnUserSpaceChanged(t *testing.T) {
@ -50,10 +50,10 @@ func TestService_OnUserSpaceChanged(t *testing.T) {
eventPublisher.EXPECT().PublishEvent(gomock.Any(), gomock.Eq(events.UsedSpaceChanged{UserID: TestUserID, UsedSpace: 1024})).Times(1)
// Original value, no changes.
service.onUserSpaceChanged(context.Background(), 0)
require.NoError(t, service.HandleUsedSpaceEvent(context.Background(), 0))
// New value, event should be published.
service.onUserSpaceChanged(context.Background(), 1024)
require.NoError(t, service.HandleUsedSpaceEvent(context.Background(), 1024))
require.Equal(t, 1024, service.identity.User.UsedSpace)
}
@ -68,14 +68,14 @@ func TestService_OnRefreshEvent(t *testing.T) {
newAddresses := newTestAddressesRefreshed()
{
getUserCall := provider.EXPECT().GetUser(gomock.Any()).Times(1).Return(newUser, nil)
getUserCall := provider.EXPECT().GetUser(gomock.Any()).Times(1).Return(*newUser, nil)
provider.EXPECT().GetAddresses(gomock.Any()).After(getUserCall).Times(1).Return(newAddresses, nil)
}
// Original value, no changes.
require.NoError(t, service.onRefreshEvent(context.Background()))
require.NoError(t, service.HandleRefreshEvent(context.Background(), 0))
require.Equal(t, newUser, service.identity.User)
require.Equal(t, *newUser, service.identity.User)
require.Equal(t, newAddresses, service.identity.AddressesSorted)
}
@ -96,7 +96,7 @@ func TestService_OnAddressCreated(t *testing.T) {
Email: newAddress.Email,
})).Times(1)
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: "",
@ -121,7 +121,7 @@ func TestService_OnAddressCreatedDisabledDoesNotProduceEvent(t *testing.T) {
Status: proton.AddressStatusEnabled,
}
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: "",
@ -146,7 +146,7 @@ func TestService_OnAddressCreatedDuplicateDoesNotProduceEvent(t *testing.T) {
Status: proton.AddressStatusDisabled,
}
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: "",
@ -177,7 +177,7 @@ func TestService_OnAddressUpdated(t *testing.T) {
Email: newAddress.Email,
})).Times(1)
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: "",
@ -221,7 +221,7 @@ func TestService_OnAddressUpdatedDisableFollowedByEnable(t *testing.T) {
})).Times(1).After(disabledCall)
}
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: "",
@ -234,7 +234,7 @@ func TestService_OnAddressUpdatedDisableFollowedByEnable(t *testing.T) {
require.Equal(t, newAddressDisabled, service.identity.Addresses[newAddressEnabled.ID])
err = service.onAddressEvent(context.Background(), []proton.AddressEvent{
err = service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: "",
@ -265,7 +265,7 @@ func TestService_OnAddressUpdateCreatedIfNotExists(t *testing.T) {
Email: newAddress.Email,
})).Times(1)
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: "",
@ -296,7 +296,7 @@ func TestService_OnAddressDeleted(t *testing.T) {
Email: address.Email,
})).Times(1)
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: address.ID,
@ -320,7 +320,7 @@ func TestService_OnAddressDeleteDisabledDoesNotProduceEvent(t *testing.T) {
Status: proton.AddressStatusDisabled,
}
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: address.ID,
@ -344,7 +344,7 @@ func TestService_OnAddressDeletedUnknownDoesNotProduceEvent(t *testing.T) {
Status: proton.AddressStatusEnabled,
}
err := service.onAddressEvent(context.Background(), []proton.AddressEvent{
err := service.HandleAddressEvents(context.Background(), []proton.AddressEvent{
{
EventItem: proton.EventItem{
ID: address.ID,
@ -364,12 +364,12 @@ func newTestService(_ *testing.T, mockCtrl *gomock.Controller) (*Service, *mocks
telemetry := mocks.NewMockTelemetry(mockCtrl)
bridgePassProvider := NewFixedBridgePassProvider([]byte("hello"))
service := NewService(subscribable, eventPublisher, NewState(user, newTestAddresses(), provider), bridgePassProvider, telemetry)
service := NewService(subscribable, eventPublisher, NewState(*user, newTestAddresses(), provider), bridgePassProvider, telemetry)
return service, eventPublisher, provider
}
func newTestUser() proton.User {
return proton.User{
func newTestUser() *proton.User {
return &proton.User{
ID: TestUserID,
Name: "Foo",
DisplayName: "Foo",
@ -383,8 +383,8 @@ func newTestUser() proton.User {
}
}
func newTestUserRefreshed() proton.User {
return proton.User{
func newTestUserRefreshed() *proton.User {
return &proton.User{
ID: TestUserID,
Name: "Alternate",
DisplayName: "Universe",