feat: notifications (#109)

* implement notification providers

* email provider

* notification handler

* notify users

* implement code sent on user eventstore

* send email implementation

* send init code

* handle codes

* fix project member handler

* add some logs for debug

* send emails

* text changes

* send sms

* notification process

* send password code

* format phone number

* test format phone

* remove fmts

* remove unused code

* rename files

* add mocks

* merge master

* Update internal/notification/providers/email/message.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update internal/notification/repository/eventsourcing/handler/notification.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update internal/notification/repository/eventsourcing/handler/notification.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update internal/notification/providers/email/provider.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* requested changes of mr

* move locker to eventstore pkg

* Update internal/notification/providers/chat/message.go

Co-authored-by: Livio Amstutz <livio.a@gmail.com>

* move locker to eventstore pkg

* linebreak

* Update internal/notification/providers/email/provider.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update internal/notification/repository/eventsourcing/handler/notification.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

* Update internal/notification/repository/eventsourcing/handler/notification.go

Co-authored-by: Silvan <silvan.reusser@gmail.com>

Co-authored-by: Silvan <silvan.reusser@gmail.com>
Co-authored-by: Livio Amstutz <livio.a@gmail.com>
This commit is contained in:
Fabi
2020-05-20 14:28:08 +02:00
committed by GitHub
parent c365a98cc8
commit e318139b37
67 changed files with 3278 additions and 119 deletions

View File

@@ -0,0 +1,49 @@
package handler
import (
"github.com/caos/logging"
sd "github.com/caos/zitadel/internal/config/systemdefaults"
"github.com/caos/zitadel/internal/config/types"
"github.com/caos/zitadel/internal/crypto"
"github.com/caos/zitadel/internal/eventstore"
"github.com/caos/zitadel/internal/eventstore/spooler"
"github.com/caos/zitadel/internal/notification/repository/eventsourcing/view"
usr_event "github.com/caos/zitadel/internal/user/repository/eventsourcing"
"time"
)
type Configs map[string]*Config
type Config struct {
MinimumCycleDuration types.Duration
}
type handler struct {
view *view.View
bulkLimit uint64
cycleDuration time.Duration
errorCountUntilSkip uint64
}
type EventstoreRepos struct {
UserEvents *usr_event.UserEventstore
}
func Register(configs Configs, bulkLimit, errorCount uint64, view *view.View, eventstore eventstore.Eventstore, repos EventstoreRepos, systemDefaults sd.SystemDefaults) []spooler.Handler {
aesCrypto, err := crypto.NewAESCrypto(systemDefaults.UserVerificationKey)
if err != nil {
logging.Log("HANDL-s90ew").WithError(err).Debug("error create new aes crypto")
}
return []spooler.Handler{
&NotifyUser{handler: handler{view, bulkLimit, configs.cycleDuration("User"), errorCount}},
&Notification{handler: handler{view, bulkLimit, configs.cycleDuration("Notification"), errorCount}, eventstore: eventstore, userEvents: repos.UserEvents, systemDefaults: systemDefaults, AesCrypto: aesCrypto},
}
}
func (configs Configs) cycleDuration(viewModel string) time.Duration {
c, ok := configs[viewModel]
if !ok {
return 1 * time.Second
}
return c.MinimumCycleDuration.Duration
}

View File

@@ -0,0 +1,168 @@
package handler
import (
"context"
"github.com/caos/zitadel/internal/api/auth"
sd "github.com/caos/zitadel/internal/config/systemdefaults"
"github.com/caos/zitadel/internal/crypto"
"github.com/caos/zitadel/internal/eventstore"
"github.com/caos/zitadel/internal/notification/types"
usr_event "github.com/caos/zitadel/internal/user/repository/eventsourcing"
es_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
"time"
"github.com/caos/logging"
"github.com/caos/zitadel/internal/eventstore/models"
"github.com/caos/zitadel/internal/eventstore/spooler"
"github.com/caos/zitadel/internal/user/repository/eventsourcing"
)
type Notification struct {
handler
eventstore eventstore.Eventstore
userEvents *usr_event.UserEventstore
systemDefaults sd.SystemDefaults
AesCrypto crypto.EncryptionAlgorithm
}
const (
notificationTable = "notification.notifications"
NOTIFY_USER = "NOTIFICATION"
)
func (n *Notification) MinimumCycleDuration() time.Duration { return n.cycleDuration }
func (n *Notification) ViewModel() string {
return notificationTable
}
func (n *Notification) EventQuery() (*models.SearchQuery, error) {
sequence, err := n.view.GetLatestNotificationSequence()
if err != nil {
return nil, err
}
return eventsourcing.UserQuery(sequence), nil
}
func (n *Notification) Process(event *models.Event) (err error) {
switch event.Type {
case es_model.InitializedUserCodeAdded:
err = n.handleInitUserCode(event)
case es_model.UserEmailCodeAdded:
err = n.handleEmailVerificationCode(event)
case es_model.UserPhoneCodeAdded:
err = n.handlePhoneVerificationCode(event)
case es_model.UserPasswordCodeAdded:
err = n.handlePasswordCode(event)
default:
return n.view.ProcessedNotificationSequence(event.Sequence)
}
if err != nil {
return err
}
return n.view.ProcessedNotificationSequence(event.Sequence)
}
func (n *Notification) handleInitUserCode(event *models.Event) (err error) {
alreadyHandled, err := n.checkIfCodeAlreadyHandled(event.AggregateID, event.Sequence, es_model.InitializedUserCodeAdded, es_model.InitializedUserCodeSent)
if err != nil || alreadyHandled {
return err
}
initCode := new(es_model.InitUserCode)
initCode.SetData(event)
user, err := n.view.NotifyUserByID(event.AggregateID)
if err != nil {
return err
}
err = types.SendUserInitCode(user, initCode, n.systemDefaults, n.AesCrypto)
if err != nil {
return err
}
return n.userEvents.InitCodeSent(getSetNotifyContextData(event.ResourceOwner), event.AggregateID)
}
func (n *Notification) handlePasswordCode(event *models.Event) (err error) {
alreadyHandled, err := n.checkIfCodeAlreadyHandled(event.AggregateID, event.Sequence, es_model.UserPasswordCodeAdded, es_model.UserPasswordCodeSent)
if err != nil || alreadyHandled {
return err
}
pwCode := new(es_model.PasswordCode)
pwCode.SetData(event)
user, err := n.view.NotifyUserByID(event.AggregateID)
if err != nil {
return err
}
err = types.SendPasswordCode(user, pwCode, n.systemDefaults, n.AesCrypto)
if err != nil {
return err
}
return n.userEvents.PasswordCodeSent(getSetNotifyContextData(event.ResourceOwner), event.AggregateID)
}
func (n *Notification) handleEmailVerificationCode(event *models.Event) (err error) {
alreadyHandled, err := n.checkIfCodeAlreadyHandled(event.AggregateID, event.Sequence, es_model.UserEmailCodeAdded, es_model.UserEmailCodeSent)
if err != nil || alreadyHandled {
return nil
}
emailCode := new(es_model.EmailCode)
emailCode.SetData(event)
user, err := n.view.NotifyUserByID(event.AggregateID)
if err != nil {
return err
}
err = types.SendEmailVerificationCode(user, emailCode, n.systemDefaults, n.AesCrypto)
if err != nil {
return err
}
return n.userEvents.EmailVerificationCodeSent(getSetNotifyContextData(event.ResourceOwner), event.AggregateID)
}
func (n *Notification) handlePhoneVerificationCode(event *models.Event) (err error) {
alreadyHandled, err := n.checkIfCodeAlreadyHandled(event.AggregateID, event.Sequence, es_model.UserPhoneCodeAdded, es_model.UserPhoneCodeSent)
if err != nil || alreadyHandled {
return nil
}
phoneCode := new(es_model.PhoneCode)
phoneCode.SetData(event)
user, err := n.view.NotifyUserByID(event.AggregateID)
if err != nil {
return err
}
err = types.SendPhoneVerificationCode(user, phoneCode, n.systemDefaults, n.AesCrypto)
if err != nil {
return err
}
return n.userEvents.PhoneVerificationCodeSent(getSetNotifyContextData(event.ResourceOwner), event.AggregateID)
}
func (n *Notification) checkIfCodeAlreadyHandled(userID string, sequence uint64, addedType, sentType models.EventType) (bool, error) {
events, err := n.getUserEvents(userID, sequence)
if err != nil {
return false, err
}
for _, event := range events {
if event.Type == addedType || event.Type == sentType {
return true, nil
}
}
return false, nil
}
func (n *Notification) getUserEvents(userID string, sequence uint64) ([]*models.Event, error) {
query, err := eventsourcing.UserByIDQuery(userID, sequence)
if err != nil {
return nil, err
}
return n.eventstore.FilterEvents(context.Background(), query)
}
func (n *Notification) OnError(event *models.Event, err error) error {
logging.LogWithFields("SPOOL-s9opc", "id", event.AggregateID, "sequence", event.Sequence).WithError(err).Warn("something went wrong in notification handler")
return spooler.HandleError(event, err, n.view.GetLatestNotificationFailedEvent, n.view.ProcessedNotificationFailedEvent, n.view.ProcessedNotificationSequence, n.errorCountUntilSkip)
}
func getSetNotifyContextData(orgID string) context.Context {
return auth.SetCtxData(context.Background(), auth.CtxData{UserID: NOTIFY_USER, OrgID: orgID})
}

View File

@@ -0,0 +1,69 @@
package handler
import (
es_model "github.com/caos/zitadel/internal/user/repository/eventsourcing/model"
"time"
"github.com/caos/logging"
"github.com/caos/zitadel/internal/eventstore"
"github.com/caos/zitadel/internal/eventstore/models"
"github.com/caos/zitadel/internal/eventstore/spooler"
"github.com/caos/zitadel/internal/user/repository/eventsourcing"
view_model "github.com/caos/zitadel/internal/user/repository/view/model"
)
type NotifyUser struct {
handler
eventstore eventstore.Eventstore
}
const (
userTable = "notification.notify_users"
)
func (p *NotifyUser) MinimumCycleDuration() time.Duration { return p.cycleDuration }
func (p *NotifyUser) ViewModel() string {
return userTable
}
func (p *NotifyUser) EventQuery() (*models.SearchQuery, error) {
sequence, err := p.view.GetLatestNotifyUserSequence()
if err != nil {
return nil, err
}
return eventsourcing.UserQuery(sequence), nil
}
func (p *NotifyUser) Process(event *models.Event) (err error) {
user := new(view_model.NotifyUser)
switch event.Type {
case es_model.UserAdded,
es_model.UserRegistered:
user.AppendEvent(event)
case es_model.UserProfileChanged,
es_model.UserEmailChanged,
es_model.UserEmailVerified,
es_model.UserPhoneChanged,
es_model.UserPhoneVerified:
user, err = p.view.NotifyUserByID(event.AggregateID)
if err != nil {
return err
}
err = user.AppendEvent(event)
case es_model.UserDeleted:
err = p.view.DeleteNotifyUser(event.AggregateID, event.Sequence)
default:
return p.view.ProcessedNotifyUserSequence(event.Sequence)
}
if err != nil {
return err
}
return p.view.PutNotifyUser(user)
}
func (p *NotifyUser) OnError(event *models.Event, err error) error {
logging.LogWithFields("SPOOL-9spwf", "id", event.AggregateID).WithError(err).Warn("something went wrong in notify user handler")
return spooler.HandleError(event, err, p.view.GetLatestNotifyUserFailedEvent, p.view.ProcessedNotifyUserFailedEvent, p.view.ProcessedNotifyUserSequence, p.errorCountUntilSkip)
}

View File

@@ -0,0 +1,56 @@
package eventsourcing
import (
sd "github.com/caos/zitadel/internal/config/systemdefaults"
"github.com/caos/zitadel/internal/config/types"
es_int "github.com/caos/zitadel/internal/eventstore"
es_spol "github.com/caos/zitadel/internal/eventstore/spooler"
"github.com/caos/zitadel/internal/notification/repository/eventsourcing/handler"
"github.com/caos/zitadel/internal/notification/repository/eventsourcing/spooler"
noti_view "github.com/caos/zitadel/internal/notification/repository/eventsourcing/view"
es_usr "github.com/caos/zitadel/internal/user/repository/eventsourcing"
)
type Config struct {
Eventstore es_int.Config
View types.SQL
Spooler spooler.SpoolerConfig
}
type EsRepository struct {
spooler *es_spol.Spooler
}
func Start(conf Config, systemDefaults sd.SystemDefaults) (*EsRepository, error) {
es, err := es_int.Start(conf.Eventstore)
if err != nil {
return nil, err
}
sqlClient, err := conf.View.Start()
if err != nil {
return nil, err
}
view, err := noti_view.StartView(sqlClient)
if err != nil {
return nil, err
}
user, err := es_usr.StartUser(es_usr.UserConfig{
Eventstore: es,
Cache: conf.Eventstore.Cache,
}, systemDefaults)
if err != nil {
return nil, err
}
eventstoreRepos := handler.EventstoreRepos{UserEvents: user}
spool := spooler.StartSpooler(conf.Spooler, es, view, sqlClient, eventstoreRepos, systemDefaults)
return &EsRepository{
spool,
}, nil
}
func (repo *EsRepository) Health() error {
return nil
}

View File

@@ -0,0 +1,19 @@
package spooler
import (
"database/sql"
es_locker "github.com/caos/zitadel/internal/eventstore/locker"
"time"
)
const (
lockTable = "notification.locks"
)
type locker struct {
dbClient *sql.DB
}
func (l *locker) Renew(lockerID, viewModel string, waitTime time.Duration) error {
return es_locker.Renew(l.dbClient, lockTable, lockerID, viewModel, waitTime)
}

View File

@@ -0,0 +1,127 @@
package spooler
import (
"database/sql"
"testing"
"time"
"github.com/DATA-DOG/go-sqlmock"
)
type dbMock struct {
db *sql.DB
mock sqlmock.Sqlmock
}
func mockDB(t *testing.T) *dbMock {
mockDB := dbMock{}
var err error
mockDB.db, mockDB.mock, err = sqlmock.New()
if err != nil {
t.Fatalf("error occured while creating stub db %v", err)
}
mockDB.mock.MatchExpectationsInOrder(true)
return &mockDB
}
func (db *dbMock) expectCommit() *dbMock {
db.mock.ExpectCommit()
return db
}
func (db *dbMock) expectRollback() *dbMock {
db.mock.ExpectRollback()
return db
}
func (db *dbMock) expectBegin() *dbMock {
db.mock.ExpectBegin()
return db
}
func (db *dbMock) expectSavepoint() *dbMock {
db.mock.ExpectExec("SAVEPOINT").WillReturnResult(sqlmock.NewResult(1, 1))
return db
}
func (db *dbMock) expectReleaseSavepoint() *dbMock {
db.mock.ExpectExec("RELEASE SAVEPOINT").WillReturnResult(sqlmock.NewResult(1, 1))
return db
}
func (db *dbMock) expectRenew(lockerID, view string, affectedRows int64) *dbMock {
query := db.mock.
ExpectExec(`INSERT INTO notification\.locks \(object_type, locker_id, locked_until\) VALUES \(\$1, \$2, now\(\)\+\$3\) ON CONFLICT \(object_type\) DO UPDATE SET locked_until = now\(\)\+\$4, locker_id = \$5 WHERE \(locks\.locked_until < now\(\) OR locks\.locker_id = \$6\) AND locks\.object_type = \$7`).
WithArgs(view, lockerID, sqlmock.AnyArg(), sqlmock.AnyArg(), lockerID, lockerID, view).
WillReturnResult(sqlmock.NewResult(1, 1))
if affectedRows == 0 {
query.WillReturnResult(sqlmock.NewResult(0, 0))
} else {
query.WillReturnResult(sqlmock.NewResult(1, affectedRows))
}
return db
}
func Test_locker_Renew(t *testing.T) {
type fields struct {
db *dbMock
}
type args struct {
lockerID string
viewModel string
waitTime time.Duration
}
tests := []struct {
name string
fields fields
args args
wantErr bool
}{
{
name: "renew succeeded",
fields: fields{
db: mockDB(t).
expectBegin().
expectSavepoint().
expectRenew("locker", "view", 1).
expectReleaseSavepoint().
expectCommit(),
},
args: args{lockerID: "locker", viewModel: "view", waitTime: 1 * time.Second},
wantErr: false,
},
{
name: "renew now rows updated",
fields: fields{
db: mockDB(t).
expectBegin().
expectSavepoint().
expectRenew("locker", "view", 0).
expectRollback(),
},
args: args{lockerID: "locker", viewModel: "view", waitTime: 1 * time.Second},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
l := &locker{
dbClient: tt.fields.db.db,
}
if err := l.Renew(tt.args.lockerID, tt.args.viewModel, tt.args.waitTime); (err != nil) != tt.wantErr {
t.Errorf("locker.Renew() error = %v, wantErr %v", err, tt.wantErr)
}
if err := tt.fields.db.mock.ExpectationsWereMet(); err != nil {
t.Errorf("not all database expectations met: %v", err)
}
})
}
}

View File

@@ -0,0 +1,34 @@
package spooler
import (
"database/sql"
sd "github.com/caos/zitadel/internal/config/systemdefaults"
"github.com/caos/zitadel/internal/eventstore"
"github.com/caos/zitadel/internal/eventstore/spooler"
"github.com/caos/zitadel/internal/notification/repository/eventsourcing/handler"
"github.com/caos/zitadel/internal/notification/repository/eventsourcing/view"
usr_event "github.com/caos/zitadel/internal/user/repository/eventsourcing"
)
type SpoolerConfig struct {
BulkLimit uint64
FailureCountUntilSkip uint64
ConcurrentTasks int
Handlers handler.Configs
}
type EventstoreRepos struct {
UserEvents *usr_event.UserEventstore
}
func StartSpooler(c SpoolerConfig, es eventstore.Eventstore, view *view.View, sql *sql.DB, eventstoreRepos handler.EventstoreRepos, systemDefaults sd.SystemDefaults) *spooler.Spooler {
spoolerConfig := spooler.Config{
Eventstore: es,
Locker: &locker{dbClient: sql},
ConcurrentTasks: c.ConcurrentTasks,
ViewHandlers: handler.Register(c.Handlers, c.BulkLimit, c.FailureCountUntilSkip, view, es, eventstoreRepos, systemDefaults),
}
spool := spoolerConfig.New()
spool.Start()
return spool
}

View File

@@ -0,0 +1,17 @@
package view
import (
"github.com/caos/zitadel/internal/view"
)
const (
errTable = "notification.failed_event"
)
func (v *View) saveFailedEvent(failedEvent *view.FailedEvent) error {
return view.SaveFailedEvent(v.Db, errTable, failedEvent)
}
func (v *View) latestFailedEvent(viewName string, sequence uint64) (*view.FailedEvent, error) {
return view.LatestFailedEvent(v.Db, errTable, viewName, sequence)
}

View File

@@ -0,0 +1,25 @@
package view
import (
global_view "github.com/caos/zitadel/internal/view"
)
const (
notificationTable = "notification.notifications"
)
func (v *View) GetLatestNotificationSequence() (uint64, error) {
return v.latestSequence(notificationTable)
}
func (v *View) ProcessedNotificationSequence(eventSequence uint64) error {
return v.saveCurrentSequence(notificationTable, eventSequence)
}
func (v *View) GetLatestNotificationFailedEvent(sequence uint64) (*global_view.FailedEvent, error) {
return v.latestFailedEvent(notificationTable, sequence)
}
func (v *View) ProcessedNotificationFailedEvent(failedEvent *global_view.FailedEvent) error {
return v.saveFailedEvent(failedEvent)
}

View File

@@ -0,0 +1,47 @@
package view
import (
"github.com/caos/zitadel/internal/user/repository/view"
"github.com/caos/zitadel/internal/user/repository/view/model"
global_view "github.com/caos/zitadel/internal/view"
)
const (
notifyUserTable = "notification.notify_users"
)
func (v *View) NotifyUserByID(userID string) (*model.NotifyUser, error) {
return view.NotifyUserByID(v.Db, notifyUserTable, userID)
}
func (v *View) PutNotifyUser(user *model.NotifyUser) error {
err := view.PutNotifyUser(v.Db, notifyUserTable, user)
if err != nil {
return err
}
return v.ProcessedNotifyUserSequence(user.Sequence)
}
func (v *View) DeleteNotifyUser(userID string, eventSequence uint64) error {
err := view.DeleteNotifyUser(v.Db, notifyUserTable, userID)
if err != nil {
return nil
}
return v.ProcessedNotifyUserSequence(eventSequence)
}
func (v *View) GetLatestNotifyUserSequence() (uint64, error) {
return v.latestSequence(notifyUserTable)
}
func (v *View) ProcessedNotifyUserSequence(eventSequence uint64) error {
return v.saveCurrentSequence(notifyUserTable, eventSequence)
}
func (v *View) GetLatestNotifyUserFailedEvent(sequence uint64) (*global_view.FailedEvent, error) {
return v.latestFailedEvent(notifyUserTable, sequence)
}
func (v *View) ProcessedNotifyUserFailedEvent(failedEvent *global_view.FailedEvent) error {
return v.saveFailedEvent(failedEvent)
}

View File

@@ -0,0 +1,17 @@
package view
import (
"github.com/caos/zitadel/internal/view"
)
const (
sequencesTable = "notification.current_sequences"
)
func (v *View) saveCurrentSequence(viewName string, sequence uint64) error {
return view.SaveCurrentSequence(v.Db, sequencesTable, viewName, sequence)
}
func (v *View) latestSequence(viewName string) (uint64, error) {
return view.LatestSequence(v.Db, sequencesTable, viewName)
}

View File

@@ -0,0 +1,25 @@
package view
import (
"database/sql"
"github.com/jinzhu/gorm"
)
type View struct {
Db *gorm.DB
}
func StartView(sqlClient *sql.DB) (*View, error) {
gorm, err := gorm.Open("postgres", sqlClient)
if err != nil {
return nil, err
}
return &View{
Db: gorm,
}, nil
}
func (v *View) Health() (err error) {
return v.Db.DB().Ping()
}