diff --git a/compose/service/module.go b/compose/service/module.go index d62613b43..70e268323 100644 --- a/compose/service/module.go +++ b/compose/service/module.go @@ -22,7 +22,8 @@ type ( ctx context.Context logger *zap.Logger - ac moduleAccessController + ac moduleAccessController + eventbus eventDispatcher moduleRepo repository.ModuleRepository recordRepo repository.RecordRepository @@ -56,8 +57,9 @@ type ( func Module() ModuleService { return (&module{ - logger: DefaultLogger.Named("module"), - ac: DefaultAccessControl, + logger: DefaultLogger.Named("module"), + ac: DefaultAccessControl, + eventbus: eventbus.Service(), }).With(context.Background()) } @@ -68,7 +70,8 @@ func (svc module) With(ctx context.Context) ModuleService { ctx: ctx, logger: svc.logger, - ac: svc.ac, + ac: svc.ac, + eventbus: svc.eventbus, moduleRepo: repository.Module(ctx, db), recordRepo: repository.Record(ctx, db), @@ -166,7 +169,7 @@ func (svc module) Create(new *types.Module) (m *types.Module, err error) { } // Calling before-create scripts - if err = eventbus.WaitFor(svc.ctx, event.ModuleBeforeCreate(new, nil, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.ModuleBeforeCreate(new, nil, ns)); err != nil { return } @@ -183,7 +186,7 @@ func (svc module) Create(new *types.Module) (m *types.Module, err error) { return nil, err } - defer eventbus.Dispatch(svc.ctx, event.ModuleAfterCreate(m, nil, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.ModuleAfterCreate(m, nil, ns)) return } @@ -216,7 +219,7 @@ func (svc module) Update(upd *types.Module) (m *types.Module, err error) { return nil, ErrNoUpdatePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.ModuleBeforeUpdate(upd, m, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.ModuleBeforeUpdate(upd, m, ns)); err != nil { return } @@ -243,7 +246,7 @@ func (svc module) Update(upd *types.Module) (m *types.Module, err error) { return nil, err } - defer eventbus.Dispatch(svc.ctx, event.ModuleAfterUpdate(upd, m, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.ModuleAfterUpdate(upd, m, ns)) return } @@ -271,7 +274,7 @@ func (svc module) DeleteByID(namespaceID, moduleID uint64) (err error) { return ErrNoDeletePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.ModuleBeforeDelete(nil, del, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.ModuleBeforeDelete(nil, del, ns)); err != nil { return } @@ -279,7 +282,7 @@ func (svc module) DeleteByID(namespaceID, moduleID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.ModuleAfterDelete(nil, del, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.ModuleAfterDelete(nil, del, ns)) return } diff --git a/compose/service/namespace.go b/compose/service/namespace.go index bb5809204..44f0009b9 100644 --- a/compose/service/namespace.go +++ b/compose/service/namespace.go @@ -22,7 +22,8 @@ type ( ctx context.Context logger *zap.Logger - ac namespaceAccessController + ac namespaceAccessController + eventbus eventDispatcher namespaceRepo repository.NamespaceRepository } @@ -53,8 +54,9 @@ type ( func Namespace() NamespaceService { return (&namespace{ - logger: DefaultLogger.Named("namespace"), - ac: DefaultAccessControl, + logger: DefaultLogger.Named("namespace"), + ac: DefaultAccessControl, + eventbus: eventbus.Service(), }).With(context.Background()) } @@ -65,7 +67,8 @@ func (svc namespace) With(ctx context.Context) NamespaceService { ctx: ctx, logger: svc.logger, - ac: svc.ac, + ac: svc.ac, + eventbus: svc.eventbus, namespaceRepo: repository.Namespace(ctx, db), } @@ -119,7 +122,7 @@ func (svc namespace) Create(new *types.Namespace) (ns *types.Namespace, err erro return nil, ErrNoCreatePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.NamespaceBeforeCreate(new, nil)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.NamespaceBeforeCreate(new, nil)); err != nil { return } @@ -131,7 +134,7 @@ func (svc namespace) Create(new *types.Namespace) (ns *types.Namespace, err erro return nil, err } - defer eventbus.Dispatch(svc.ctx, event.NamespaceAfterCreate(ns, nil)) + defer svc.eventbus.Dispatch(svc.ctx, event.NamespaceAfterCreate(ns, nil)) return } @@ -156,7 +159,7 @@ func (svc namespace) Update(upd *types.Namespace) (ns *types.Namespace, err erro return nil, ErrNoUpdatePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.NamespaceBeforeUpdate(upd, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.NamespaceBeforeUpdate(upd, ns)); err != nil { return } @@ -174,7 +177,7 @@ func (svc namespace) Update(upd *types.Namespace) (ns *types.Namespace, err erro return nil, err } - defer eventbus.Dispatch(svc.ctx, event.NamespaceAfterUpdate(upd, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.NamespaceAfterUpdate(upd, ns)) return } @@ -193,7 +196,7 @@ func (svc namespace) DeleteByID(namespaceID uint64) (err error) { return ErrNoDeletePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.NamespaceBeforeDelete(nil, del)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.NamespaceBeforeDelete(nil, del)); err != nil { return } @@ -201,7 +204,7 @@ func (svc namespace) DeleteByID(namespaceID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.NamespaceAfterDelete(nil, del)) + defer svc.eventbus.Dispatch(svc.ctx, event.NamespaceAfterDelete(nil, del)) return } diff --git a/compose/service/page.go b/compose/service/page.go index 73d151573..9e23d9475 100644 --- a/compose/service/page.go +++ b/compose/service/page.go @@ -22,7 +22,8 @@ type ( ctx context.Context logger *zap.Logger - ac pageAccessController + ac pageAccessController + eventbus eventDispatcher pageRepo repository.PageRepository moduleRepo repository.ModuleRepository @@ -59,8 +60,9 @@ type ( func Page() PageService { return (&page{ - logger: DefaultLogger.Named("page"), - ac: DefaultAccessControl, + logger: DefaultLogger.Named("page"), + ac: DefaultAccessControl, + eventbus: eventbus.Service(), }).With(context.Background()) } @@ -71,7 +73,8 @@ func (svc page) With(ctx context.Context) PageService { ctx: ctx, logger: svc.logger, - ac: svc.ac, + ac: svc.ac, + eventbus: svc.eventbus, pageRepo: repository.Page(ctx, db), moduleRepo: repository.Module(ctx, db), @@ -221,7 +224,7 @@ func (svc page) Create(new *types.Page) (p *types.Page, err error) { return nil, ErrNoCreatePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.PageBeforeCreate(new, nil, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.PageBeforeCreate(new, nil, ns)); err != nil { return } @@ -233,7 +236,7 @@ func (svc page) Create(new *types.Page) (p *types.Page, err error) { return } - defer eventbus.Dispatch(svc.ctx, event.PageAfterCreate(new, nil, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.PageAfterCreate(new, nil, ns)) return } @@ -266,7 +269,7 @@ func (svc page) Update(upd *types.Page) (p *types.Page, err error) { return nil, ErrNoUpdatePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.PageBeforeUpdate(upd, p, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.PageBeforeUpdate(upd, p, ns)); err != nil { return } @@ -287,7 +290,7 @@ func (svc page) Update(upd *types.Page) (p *types.Page, err error) { return } - defer eventbus.Dispatch(svc.ctx, event.PageAfterUpdate(upd, p, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.PageAfterUpdate(upd, p, ns)) return } @@ -327,7 +330,7 @@ func (svc page) DeleteByID(namespaceID, pageID uint64) (err error) { return ErrNoDeletePermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.PageBeforeDelete(nil, del, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.PageBeforeDelete(nil, del, ns)); err != nil { return } @@ -335,7 +338,7 @@ func (svc page) DeleteByID(namespaceID, pageID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.PageAfterDelete(nil, del, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.PageAfterDelete(nil, del, ns)) return } diff --git a/compose/service/record.go b/compose/service/record.go index 3aa528f43..675b223ea 100644 --- a/compose/service/record.go +++ b/compose/service/record.go @@ -32,7 +32,8 @@ type ( ctx context.Context logger *zap.Logger - ac recordAccessController + ac recordAccessController + eventbus eventDispatcher recordRepo repository.RecordRepository moduleRepo repository.ModuleRepository @@ -103,8 +104,9 @@ type ( func Record() RecordService { return (&record{ - logger: DefaultLogger.Named("record"), - ac: DefaultAccessControl, + logger: DefaultLogger.Named("record"), + ac: DefaultAccessControl, + eventbus: eventbus.Service(), }).With(context.Background()) } @@ -116,7 +118,8 @@ func (svc record) With(ctx context.Context) RecordService { ctx: ctx, logger: svc.logger, - ac: svc.ac, + ac: svc.ac, + eventbus: svc.eventbus, recordRepo: repository.Record(ctx, db), moduleRepo: repository.Module(ctx, db), @@ -300,7 +303,7 @@ func (svc record) Create(new *types.Record) (r *types.Record, err error) { CreatedAt: time.Now(), } - if err = eventbus.WaitFor(svc.ctx, event.RecordBeforeCreate(new, nil, m, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RecordBeforeCreate(new, nil, m, ns)); err != nil { return } @@ -327,7 +330,7 @@ func (svc record) Create(new *types.Record) (r *types.Record, err error) { return } - defer eventbus.Dispatch(svc.ctx, event.RecordAfterCreate(r, nil, m, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.RecordAfterCreate(r, nil, m, ns)) return }) } @@ -351,7 +354,7 @@ func (svc record) Update(upd *types.Record) (r *types.Record, err error) { return nil, ErrStaleData.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.RecordBeforeUpdate(upd, r, m, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RecordBeforeUpdate(upd, r, m, ns)); err != nil { return } @@ -376,7 +379,7 @@ func (svc record) Update(upd *types.Record) (r *types.Record, err error) { return } - defer eventbus.Dispatch(svc.ctx, event.RecordAfterUpdate(upd, r, m, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.RecordAfterUpdate(upd, r, m, ns)) return }) } @@ -407,7 +410,7 @@ func (svc record) DeleteByID(namespaceID, recordID uint64) (err error) { } // Calling before-record-delete scripts - if err = eventbus.WaitFor(svc.ctx, event.RecordBeforeDelete(nil, del, m, ns)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RecordBeforeDelete(nil, del, m, ns)); err != nil { return } @@ -424,7 +427,7 @@ func (svc record) DeleteByID(namespaceID, recordID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.RecordAfterDelete(nil, del, m, ns)) + defer svc.eventbus.Dispatch(svc.ctx, event.RecordAfterDelete(nil, del, m, ns)) return }) diff --git a/compose/service/service.go b/compose/service/service.go index f3601fad4..6e444a4f9 100644 --- a/compose/service/service.go +++ b/compose/service/service.go @@ -10,6 +10,7 @@ import ( "github.com/cortezaproject/corteza-server/compose/types" "github.com/cortezaproject/corteza-server/pkg/app/options" "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/pkg/eventbus" "github.com/cortezaproject/corteza-server/pkg/permissions" "github.com/cortezaproject/corteza-server/pkg/settings" "github.com/cortezaproject/corteza-server/pkg/store" @@ -29,6 +30,11 @@ type ( Corredor options.CorredorOpt GRPCClientSystem options.GRPCServerOpt } + + eventDispatcher interface { + WaitFor(ctx context.Context, ev eventbus.Event) (err error) + Dispatch(ctx context.Context, ev eventbus.Event) + } ) var ( diff --git a/pkg/eventbus/eventbus.go b/pkg/eventbus/eventbus.go index 848893ab5..301178a0a 100644 --- a/pkg/eventbus/eventbus.go +++ b/pkg/eventbus/eventbus.go @@ -37,26 +37,6 @@ func init() { gEventBus = New() } -// WaitFor is a package level wrapper for global instance of eventbus -func WaitFor(ctx context.Context, ev Event) (err error) { - return gEventBus.WaitFor(ctx, ev) -} - -// Dispatch is a package level wrapper for global instance of eventbus -func Dispatch(ctx context.Context, ev Event) { - gEventBus.Dispatch(ctx, ev) -} - -// Register is a package level wrapper for global instance of eventbus -func Register(handler Handler, ops ...TriggerRegOp) uintptr { - return gEventBus.Register(handler, ops...) -} - -// Unregister is a package level wrapper for global instance of eventbus -func Unregister(ptrs ...uintptr) { - gEventBus.Unregister(ptrs...) -} - // Returns func Default() *eventbus { return gEventBus diff --git a/system/service/application.go b/system/service/application.go index e4e96550a..02fee1513 100644 --- a/system/service/application.go +++ b/system/service/application.go @@ -18,7 +18,8 @@ type ( db *factory.DB ctx context.Context - ac applicationAccessController + ac applicationAccessController + eventbus eventDispatcher application repository.ApplicationRepository } @@ -48,7 +49,8 @@ type ( func Application(ctx context.Context) ApplicationService { return (&application{ - ac: DefaultAccessControl, + ac: DefaultAccessControl, + eventbus: eventbus.Service(), }).With(ctx) } @@ -56,9 +58,12 @@ func Application(ctx context.Context) ApplicationService { func (svc *application) With(ctx context.Context) ApplicationService { db := repository.DB(ctx) return &application{ - db: db, - ctx: ctx, - ac: svc.ac, + db: db, + ctx: ctx, + + ac: svc.ac, + eventbus: svc.eventbus, + application: repository.Application(ctx, db), } } @@ -101,7 +106,7 @@ func (svc *application) Create(new *types.Application) (app *types.Application, return nil, ErrNoPermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.ApplicationBeforeCreate(new, nil)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.ApplicationBeforeCreate(new, nil)); err != nil { return } @@ -109,7 +114,7 @@ func (svc *application) Create(new *types.Application) (app *types.Application, return } - defer eventbus.Dispatch(svc.ctx, event.ApplicationAfterCreate(new, nil)) + defer svc.eventbus.Dispatch(svc.ctx, event.ApplicationAfterCreate(new, nil)) return } @@ -124,7 +129,7 @@ func (svc *application) Update(upd *types.Application) (app *types.Application, return } - if err = eventbus.WaitFor(svc.ctx, event.ApplicationBeforeUpdate(upd, app)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.ApplicationBeforeUpdate(upd, app)); err != nil { return } @@ -137,7 +142,7 @@ func (svc *application) Update(upd *types.Application) (app *types.Application, return err } - defer eventbus.Dispatch(svc.ctx, event.ApplicationAfterUpdate(upd, app)) + defer svc.eventbus.Dispatch(svc.ctx, event.ApplicationAfterUpdate(upd, app)) return nil }) } @@ -154,7 +159,7 @@ func (svc *application) Delete(ID uint64) (err error) { if !svc.ac.CanDeleteApplication(svc.ctx, app) { return ErrNoPermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.ApplicationBeforeDelete(nil, app)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.ApplicationBeforeDelete(nil, app)); err != nil { return } @@ -162,7 +167,7 @@ func (svc *application) Delete(ID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.ApplicationAfterDelete(nil, app)) + defer svc.eventbus.Dispatch(svc.ctx, event.ApplicationAfterDelete(nil, app)) return } diff --git a/system/service/auth.go b/system/service/auth.go index 35d7ef830..8c2f7f69f 100644 --- a/system/service/auth.go +++ b/system/service/auth.go @@ -28,6 +28,8 @@ type ( ctx context.Context logger *zap.Logger + eventbus eventDispatcher + subscription authSubscriptionChecker credentials repository.CredentialsRepository users repository.UserRepository @@ -93,6 +95,18 @@ func defaultProviderValidator(provider string) error { func Auth(ctx context.Context) AuthService { return (&auth{ logger: DefaultLogger.Named("auth"), + + eventbus: eventbus.Service(), + subscription: CurrentSubscription, + settings: CurrentSettings, + notifications: DefaultAuthNotification, + + providerValidator: defaultProviderValidator, + + now: func() *time.Time { + var now = time.Now() + return &now + }, }).With(ctx) } @@ -107,15 +121,13 @@ func (svc auth) With(ctx context.Context) AuthService { users: repository.User(ctx, db), roles: repository.Role(ctx, db), - subscription: CurrentSubscription, - settings: CurrentSettings, - notifications: DefaultAuthNotification, + subscription: svc.subscription, + settings: svc.settings, + notifications: svc.notifications, + eventbus: svc.eventbus, + providerValidator: svc.providerValidator, - providerValidator: defaultProviderValidator, - now: func() *time.Time { - var now = time.Now() - return &now - }, + now: svc.now, } } @@ -182,7 +194,7 @@ func (svc auth) External(profile goth.User) (u *types.User, err error) { return err } - if err = eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(u, authProvider)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(u, authProvider)); err != nil { return err } @@ -199,7 +211,7 @@ func (svc auth) External(profile goth.User) (u *types.User, err error) { zap.String("email", u.Email), ) - defer eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(u, authProvider)) + defer svc.eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(u, authProvider)) return nil } else { @@ -233,7 +245,7 @@ func (svc auth) External(profile goth.User) (u *types.User, err error) { return err } - if err = eventbus.WaitFor(svc.ctx, event.AuthBeforeSignup(u, authProvider)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.AuthBeforeSignup(u, authProvider)); err != nil { return err } @@ -246,7 +258,7 @@ func (svc auth) External(profile goth.User) (u *types.User, err error) { zap.String("email", u.Email), ) - defer eventbus.Dispatch(svc.ctx, event.AuthAfterSignup(u, authProvider)) + defer svc.eventbus.Dispatch(svc.ctx, event.AuthAfterSignup(u, authProvider)) _ = svc.autoPromote(u) } else if err != nil { @@ -259,11 +271,11 @@ func (svc auth) External(profile goth.User) (u *types.User, err error) { zap.String("email", u.Email), ) - if err = eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(u, authProvider)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(u, authProvider)); err != nil { return err } - defer eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(u, authProvider)) + defer svc.eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(u, authProvider)) } @@ -342,11 +354,11 @@ func (svc auth) InternalSignUp(input *types.User, password string) (u *types.Use // We're not actually doing sign-up here - user exists, // password is a match, so lets trigger before/after user login events - if err = eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(existing, &types.AuthProvider{})); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(existing, &types.AuthProvider{})); err != nil { return nil, err } - defer eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(existing, &types.AuthProvider{})) + defer svc.eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(existing, &types.AuthProvider{})) return existing, nil @@ -379,7 +391,7 @@ func (svc auth) InternalSignUp(input *types.User, password string) (u *types.Use EmailConfirmed: !svc.settings.Auth.Internal.Signup.EmailConfirmationRequired, } - if err = eventbus.WaitFor(svc.ctx, event.AuthBeforeSignup(new, &types.AuthProvider{})); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.AuthBeforeSignup(new, &types.AuthProvider{})); err != nil { return } @@ -410,7 +422,7 @@ func (svc auth) InternalSignUp(input *types.User, password string) (u *types.Use return nil, err } - defer eventbus.Dispatch(svc.ctx, event.AuthAfterSignup(u, &types.AuthProvider{})) + defer svc.eventbus.Dispatch(svc.ctx, event.AuthAfterSignup(u, &types.AuthProvider{})) return u, nil } @@ -467,7 +479,7 @@ func (svc auth) InternalLogin(email string, password string) (u *types.User, err return } - if err = eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(u, &types.AuthProvider{})); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.AuthBeforeLogin(u, &types.AuthProvider{})); err != nil { return nil, err } @@ -492,7 +504,7 @@ func (svc auth) InternalLogin(email string, password string) (u *types.User, err return nil, errors.New("user email pending confirmation") } - defer eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(u, &types.AuthProvider{})) + defer svc.eventbus.Dispatch(svc.ctx, event.AuthAfterLogin(u, &types.AuthProvider{})) return u, err } diff --git a/system/service/auth_test.go b/system/service/auth_test.go index 6f6f3d95e..33399eba5 100644 --- a/system/service/auth_test.go +++ b/system/service/auth_test.go @@ -10,6 +10,7 @@ import ( "go.uber.org/zap" "golang.org/x/crypto/bcrypt" + "github.com/cortezaproject/corteza-server/pkg/eventbus" "github.com/cortezaproject/corteza-server/system/repository" repomock "github.com/cortezaproject/corteza-server/system/repository/mocks" "github.com/cortezaproject/corteza-server/system/types" @@ -36,6 +37,8 @@ func makeMockAuthService(u repository.UserRepository, c repository.CredentialsRe settings: &types.Settings{}, + eventbus: eventbus.New(), + now: func() *time.Time { return nil }, diff --git a/system/service/role.go b/system/service/role.go index 130f92180..c46f2aa2c 100644 --- a/system/service/role.go +++ b/system/service/role.go @@ -28,7 +28,9 @@ type ( ctx context.Context logger *zap.Logger - ac roleAccessController + ac roleAccessController + eventbus eventDispatcher + user UserService role repository.RoleRepository @@ -73,21 +75,25 @@ type ( func Role(ctx context.Context) RoleService { return (&role{ - ac: DefaultAccessControl, - logger: DefaultLogger.Named("role"), + ac: DefaultAccessControl, + eventbus: eventbus.Service(), + logger: DefaultLogger.Named("role"), + user: DefaultUser.With(ctx), }).With(ctx) } func (svc role) With(ctx context.Context) RoleService { db := repository.DB(ctx) return &role{ - db: db, - ctx: ctx, - logger: svc.logger, - ac: svc.ac, + db: db, + ctx: ctx, + + logger: svc.logger, + ac: svc.ac, + eventbus: svc.eventbus, + user: svc.user, role: repository.Role(ctx, db), - user: DefaultUser.With(ctx), } } @@ -151,7 +157,7 @@ func (svc role) Create(new *types.Role) (r *types.Role, err error) { } return r, svc.db.Transaction(func() (err error) { - if err = eventbus.WaitFor(svc.ctx, event.RoleBeforeCreate(new, r)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RoleBeforeCreate(new, r)); err != nil { return } @@ -163,7 +169,7 @@ func (svc role) Create(new *types.Role) (r *types.Role, err error) { return } - defer eventbus.Dispatch(svc.ctx, event.RoleAfterCreate(new, r)) + defer svc.eventbus.Dispatch(svc.ctx, event.RoleAfterCreate(new, r)) return }) } @@ -186,7 +192,7 @@ func (svc role) Update(upd *types.Role) (r *types.Role, err error) { return } - if err = eventbus.WaitFor(svc.ctx, event.RoleBeforeUpdate(upd, r)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RoleBeforeUpdate(upd, r)); err != nil { return } @@ -202,7 +208,7 @@ func (svc role) Update(upd *types.Role) (r *types.Role, err error) { return err } - defer eventbus.Dispatch(svc.ctx, event.RoleAfterUpdate(upd, r)) + defer svc.eventbus.Dispatch(svc.ctx, event.RoleAfterUpdate(upd, r)) return nil }) @@ -237,7 +243,7 @@ func (svc role) Delete(roleID uint64) (err error) { return ErrNoPermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.RoleBeforeDelete(nil, role)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RoleBeforeDelete(nil, role)); err != nil { return } @@ -245,7 +251,7 @@ func (svc role) Delete(roleID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.RoleAfterDelete(nil, role)) + defer svc.eventbus.Dispatch(svc.ctx, event.RoleAfterDelete(nil, role)) return } @@ -349,7 +355,7 @@ func (svc role) MemberAdd(roleID, userID uint64) (err error) { return } - if err = eventbus.WaitFor(svc.ctx, event.RoleMemberBeforeAdd(user, role)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RoleMemberBeforeAdd(user, role)); err != nil { return } @@ -361,7 +367,7 @@ func (svc role) MemberAdd(roleID, userID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.RoleMemberAfterAdd(user, role)) + defer svc.eventbus.Dispatch(svc.ctx, event.RoleMemberAfterAdd(user, role)) return nil } @@ -379,7 +385,7 @@ func (svc role) MemberRemove(roleID, userID uint64) (err error) { return } - if err = eventbus.WaitFor(svc.ctx, event.RoleMemberBeforeRemove(user, role)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.RoleMemberBeforeRemove(user, role)); err != nil { return } @@ -391,7 +397,7 @@ func (svc role) MemberRemove(roleID, userID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.RoleMemberAfterRemove(user, role)) + defer svc.eventbus.Dispatch(svc.ctx, event.RoleMemberAfterRemove(user, role)) return nil } diff --git a/system/service/service.go b/system/service/service.go index 1d242a5e5..1209d9dbd 100644 --- a/system/service/service.go +++ b/system/service/service.go @@ -7,6 +7,7 @@ import ( "github.com/cortezaproject/corteza-server/pkg/app/options" intAuth "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/pkg/eventbus" "github.com/cortezaproject/corteza-server/pkg/permissions" "github.com/cortezaproject/corteza-server/pkg/settings" "github.com/cortezaproject/corteza-server/system/repository" @@ -34,6 +35,11 @@ type ( CanCreateUser(uint) error CanRegister(uint) error } + + eventDispatcher interface { + WaitFor(ctx context.Context, ev eventbus.Event) (err error) + Dispatch(ctx context.Context, ev eventbus.Event) + } ) var ( diff --git a/system/service/user.go b/system/service/user.go index 996deb72e..b7052a19c 100644 --- a/system/service/user.go +++ b/system/service/user.go @@ -42,7 +42,9 @@ type ( auth userAuth subscription userSubscriptionChecker - ac userAccessController + ac userAccessController + eventbus eventDispatcher + user repository.UserRepository credentials repository.CredentialsRepository @@ -103,7 +105,21 @@ type ( func User(ctx context.Context) UserService { return (&user{ - logger: DefaultLogger.Named("user"), + logger: DefaultLogger.Named("user"), + eventbus: eventbus.Service(), + ac: DefaultAccessControl, + settings: CurrentSettings, + auth: DefaultAuth, + + subscription: CurrentSubscription, + + // @todo wire this with settings (privacy.mask.email) + // new default value will be true! + privacyMaskEmail: false, + + // @todo wire this with settings (privacy.mask.name) + // new default value will be true! + privacyMaskName: false, }).With(ctx) } @@ -120,22 +136,17 @@ func (svc user) With(ctx context.Context) UserService { db: db, logger: svc.logger, - ac: DefaultAccessControl, - settings: CurrentSettings, - auth: DefaultAuth, - - subscription: CurrentSubscription, + ac: svc.ac, + eventbus: svc.eventbus, + settings: svc.settings, + auth: svc.auth, + subscription: svc.subscription, user: repository.User(ctx, db), credentials: repository.Credentials(ctx, db), - // @todo wire this with settings (privacy.mask.email) - // new default value will be true! - privacyMaskEmail: false, - - // @todo wire this with settings (privacy.mask.name) - // new default value will be true! - privacyMaskName: false, + privacyMaskEmail: svc.privacyMaskEmail, + privacyMaskName: svc.privacyMaskName, } } @@ -239,7 +250,7 @@ func (svc user) Create(new *types.User) (u *types.User, err error) { } } - if err = eventbus.WaitFor(svc.ctx, event.UserBeforeCreate(new, u)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.UserBeforeCreate(new, u)); err != nil { return } @@ -253,7 +264,7 @@ func (svc user) Create(new *types.User) (u *types.User, err error) { return } - defer eventbus.Dispatch(svc.ctx, event.UserAfterCreate(new, u)) + defer svc.eventbus.Dispatch(svc.ctx, event.UserAfterCreate(new, u)) return }) } @@ -285,7 +296,7 @@ func (svc user) Update(upd *types.User) (u *types.User, err error) { u.Handle = upd.Handle u.Kind = upd.Kind - if err = eventbus.WaitFor(svc.ctx, event.UserBeforeUpdate(upd, u)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.UserBeforeUpdate(upd, u)); err != nil { return } @@ -298,7 +309,7 @@ func (svc user) Update(upd *types.User) (u *types.User, err error) { return } - defer eventbus.Dispatch(svc.ctx, event.UserAfterUpdate(upd, u)) + defer svc.eventbus.Dispatch(svc.ctx, event.UserAfterUpdate(upd, u)) return }) } @@ -347,7 +358,7 @@ func (svc user) Delete(ID uint64) (err error) { return ErrNoPermissions.withStack() } - if err = eventbus.WaitFor(svc.ctx, event.UserBeforeUpdate(nil, del)); err != nil { + if err = svc.eventbus.WaitFor(svc.ctx, event.UserBeforeUpdate(nil, del)); err != nil { return } @@ -355,7 +366,7 @@ func (svc user) Delete(ID uint64) (err error) { return } - defer eventbus.Dispatch(svc.ctx, event.UserAfterDelete(nil, del)) + defer svc.eventbus.Dispatch(svc.ctx, event.UserAfterDelete(nil, del)) return }