From c9b395f60da59d6403e279d368a93912bd0548ae Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Mon, 12 Oct 2020 01:19:35 +0200 Subject: [PATCH] separate notify.SubmitVerification from notify.Submit --- backend/app/notify/email.go | 51 +++++++++++++-------- backend/app/notify/email_test.go | 33 +++++++++----- backend/app/notify/notify.go | 55 +++++++++++++++++------ backend/app/notify/notify_mock.go | 34 ++++++++++++-- backend/app/notify/notify_test.go | 33 ++++++++++++++ backend/app/notify/telegram.go | 10 ++--- backend/app/notify/telegram_test.go | 13 +++++- backend/app/rest/api/rest_private.go | 14 +++--- backend/app/rest/api/rest_private_test.go | 15 ++++--- 9 files changed, 191 insertions(+), 67 deletions(-) diff --git a/backend/app/notify/email.go b/backend/app/notify/email.go index f01fc9f3..bef4e41d 100644 --- a/backend/app/notify/email.go +++ b/backend/app/notify/email.go @@ -167,7 +167,7 @@ func (e *Email) setTemplates() error { // Send email about comment reply to Request.Email if it's set, // also sends email to site administrator if appropriate option is set. // Thread safe -func (e *Email) Send(ctx context.Context, req Request) (err error) { +func (e *Email) Send(ctx context.Context, req Request) error { if req.Email == "" { // this means we can't send this request via Email return nil @@ -177,26 +177,41 @@ func (e *Email) Send(ctx context.Context, req Request) (err error) { return errors.Errorf("sending message to %q aborted due to canceled context", req.Email) default: } - var msg string - if req.Verification.Token != "" { - log.Printf("[DEBUG] send verification via %s, user %s", e, req.Verification.User) - msg, err = e.buildVerificationMessage(req.Verification.User, req.Email, req.Verification.Token, req.Verification.SiteID) - if err != nil { - return err - } + if req.parent.User.ID == req.Comment.User.ID && !req.ForAdmin { + // don't send anything if if user replied to their own comment + return nil + } + log.Printf("[DEBUG] send notification via %s, comment id %s", e, req.Comment.ID) + msg, err := e.buildMessageFromRequest(req, req.ForAdmin) + if err != nil { + return err } - if req.Comment.ID != "" { - if req.parent.User.ID == req.Comment.User.ID && !req.ForAdmin { - // don't send anything if if user replied to their own comment - return nil - } - log.Printf("[DEBUG] send notification via %s, comment id %s", e, req.Comment.ID) - msg, err = e.buildMessageFromRequest(req, req.ForAdmin) - if err != nil { - return err - } + return repeater.NewDefault(5, time.Millisecond*250).Do( + ctx, + func() error { + return e.sendMessage(emailMessage{from: e.From, to: req.Email, message: msg}) + }) +} + +// SendVerification email verification VerificationRequest.Email if it's set. +// Thread safe +func (e *Email) SendVerification(ctx context.Context, req VerificationRequest) error { + if req.Email == "" { + // this means we can't send this request via Email + return nil + } + select { + case <-ctx.Done(): + return errors.Errorf("sending message to %q aborted due to canceled context", req.Email) + default: + } + + log.Printf("[DEBUG] send verification via %s, user %s", e, req.User) + msg, err := e.buildVerificationMessage(req.User, req.Email, req.Token, req.SiteID) + if err != nil { + return err } return repeater.NewDefault(5, time.Millisecond*250).Do( diff --git a/backend/app/notify/email_test.go b/backend/app/notify/email_test.go index 133238f9..0e0c6993 100644 --- a/backend/app/notify/email_test.go +++ b/backend/app/notify/email_test.go @@ -108,7 +108,7 @@ func TestEmailSendErrors(t *testing.T) { e.verifyTmpl, err = template.New("test").Parse("{{.Test}}") assert.NoError(t, err) - assert.EqualError(t, e.Send(context.Background(), Request{Email: "bad@example.org", Verification: VerificationMetadata{Token: "some"}}), + assert.EqualError(t, e.SendVerification(context.Background(), VerificationRequest{Email: "bad@example.org", Token: "some"}), "error executing template to build verification message: template: test:1:2: executing \"test\" at <.Test>: can't evaluate field Test in type notify.verifyTmplData") e.msgTmpl, err = template.New("test").Parse("{{.Test}}") @@ -247,20 +247,31 @@ func TestEmail_SendVerification(t *testing.T) { fakeSMTP := fakeTestSMTP{} email.smtp = &fakeSMTP email.TokenGenFn = TokenGenFn - req := Request{ - Email: "test@example.org", - Verification: VerificationMetadata{ - SiteID: "remark", - User: "test_username", - Token: "secret_", - }, + // proper VerificationRequest without email + req := VerificationRequest{ + SiteID: "remark", + User: "test_username", + Token: "secret_", } - assert.NoError(t, email.Send(context.TODO(), req)) + assert.NoError(t, email.SendVerification(context.TODO(), req)) + assert.Equal(t, "", fakeSMTP.readMail()) + assert.Equal(t, 0, fakeSMTP.readQuitCount()) + assert.Equal(t, "", fakeSMTP.readRcpt()) + + // proper VerificationRequest with email + req.Email = "test@example.org" + assert.NoError(t, email.SendVerification(context.TODO(), req)) assert.Equal(t, "from@example.org", fakeSMTP.readMail()) assert.Equal(t, 1, fakeSMTP.readQuitCount()) assert.Equal(t, "test@example.org", fakeSMTP.readRcpt()) + + // VerificationRequest with canceled context + ctx, cancel := context.WithCancel(context.TODO()) + cancel() + assert.EqualError(t, email.SendVerification(ctx, req), "sending message to \"test@example.org\" aborted due to canceled context") + // test buildVerificationMessage separately for message text - res, err := email.buildVerificationMessage(req.Verification.User, req.Email, req.Verification.Token, req.Verification.SiteID) + res, err := email.buildVerificationMessage(req.User, req.Email, req.Token, req.SiteID) assert.NoError(t, err) assert.Contains(t, res, `From: from@example.org To: test@example.org @@ -272,7 +283,7 @@ Date: `) assert.Contains(t, res, `secret_`) assert.NotContains(t, res, `https://example.org/`) email.SubscribeURL = "https://example.org/subscribe.html?token=" - res, err = email.buildVerificationMessage(req.Verification.User, req.Email, req.Verification.Token, req.Verification.SiteID) + res, err = email.buildVerificationMessage(req.User, req.Email, req.Token, req.SiteID) assert.NoError(t, err) assert.Contains(t, res, `From: from@example.org To: test@example.org diff --git a/backend/app/notify/notify.go b/backend/app/notify/notify.go index d6e3c0a1..d64363f7 100644 --- a/backend/app/notify/notify.go +++ b/backend/app/notify/notify.go @@ -14,9 +14,10 @@ import ( // Service delivers notifications to multiple destinations type Service struct { - dataService Store - destinations []Destination - queue chan Request + dataService Store + destinations []Destination + queue chan Request + verificationQueue chan VerificationRequest closed uint32 // non-zero means closed. uses uint instead of bool for atomic ctx context.Context @@ -26,7 +27,8 @@ type Service struct { // Destination defines interface for a given destination service, like telegram, email and so on type Destination interface { fmt.Stringer - Send(ctx context.Context, req Request) error + Send(context.Context, Request) error + SendVerification(context.Context, VerificationRequest) error } // Store defines the minimal interface accessing stored comments used by notifier @@ -41,14 +43,13 @@ type Request struct { parent store.Comment // fetched only in case Comment is set Email string // if set (also) send email ForAdmin bool // if set, message supposed to be sent to administrator - - Verification VerificationMetadata // if set sent verification notification } -// VerificationMetadata required to send notify method verification message -type VerificationMetadata struct { +// VerificationRequest notification for user +type VerificationRequest struct { SiteID string User string + Email string // if set, send email only Token string } @@ -62,11 +63,12 @@ func NewService(dataService Store, size int, destinations ...Destination) *Servi } ctx, cancel := context.WithCancel(context.Background()) res := Service{ - dataService: dataService, - queue: make(chan Request, size), - destinations: destinations, - ctx: ctx, - cancel: cancel, + dataService: dataService, + queue: make(chan Request, size), + verificationQueue: make(chan VerificationRequest, size), + destinations: destinations, + ctx: ctx, + cancel: cancel, } if len(destinations) > 0 { go res.do() @@ -101,11 +103,24 @@ func (s *Service) Submit(req Request) { } } +// SubmitVerification to internal channel if not busy, drop if can't send +func (s *Service) SubmitVerification(req VerificationRequest) { + if len(s.destinations) == 0 || atomic.LoadUint32(&s.closed) != 0 { + return + } + select { + case s.verificationQueue <- req: + default: + log.Printf("[WARN] can't send verification to queue, %s for %s", req.User, req.Email) + } +} + // Close queue channel and wait for completion func (s *Service) Close() { if s.queue != nil { log.Print("[DEBUG] close notifier") close(s.queue) + close(s.verificationQueue) s.cancel() <-s.ctx.Done() } @@ -131,6 +146,20 @@ func (s *Service) do() { }(dest) } wg.Wait() + case v, ok := <-s.verificationQueue: + if !ok { + return + } + wg.Add(len(s.destinations)) + for _, dest := range s.destinations { + go func(d Destination) { + if err := d.SendVerification(s.ctx, v); err != nil { + log.Printf("[WARN] failed to send to %s, %s", d, err) + } + wg.Done() + }(dest) + } + wg.Wait() case <-s.ctx.Done(): return } diff --git a/backend/app/notify/notify_mock.go b/backend/app/notify/notify_mock.go index f934e24c..381467a0 100644 --- a/backend/app/notify/notify_mock.go +++ b/backend/app/notify/notify_mock.go @@ -11,10 +11,11 @@ import ( // MockDest is a destination mock type MockDest struct { - data []Request - id int - closed bool - lock sync.Mutex + data []Request + verificationData []VerificationRequest + id int + closed bool + lock sync.Mutex } // Send mock @@ -32,6 +33,21 @@ func (m *MockDest) Send(ctx context.Context, r Request) error { return nil } +// SendVerification mock +func (m *MockDest) SendVerification(ctx context.Context, v VerificationRequest) error { + m.lock.Lock() + defer m.lock.Unlock() + select { + case <-time.After(10 * time.Millisecond): + m.verificationData = append(m.verificationData, v) + log.Printf("sent %s -> %d", v.User, m.id) + case <-ctx.Done(): + log.Printf("ctx closed %d", m.id) + m.closed = true + } + return nil +} + // Get mock func (m *MockDest) Get() []Request { m.lock.Lock() @@ -40,4 +56,14 @@ func (m *MockDest) Get() []Request { copy(res, m.data) return res } + +// GetVerify mock +func (m *MockDest) GetVerify() []VerificationRequest { + m.lock.Lock() + defer m.lock.Unlock() + res := make([]VerificationRequest, len(m.verificationData)) + copy(res, m.verificationData) + return res +} + func (m *MockDest) String() string { return fmt.Sprintf("mock id=%d, closed=%v", m.id, m.closed) } diff --git a/backend/app/notify/notify_test.go b/backend/app/notify/notify_test.go index 10c097fb..32ed8672 100644 --- a/backend/app/notify/notify_test.go +++ b/backend/app/notify/notify_test.go @@ -63,6 +63,35 @@ func TestService_WithDrops(t *testing.T) { assert.Equal(t, 2, len(d2.Get()), "one comment from three dropped from d2, got: %v", d2.Get()) } +func TestService_SubmitVerificationWithDrops(t *testing.T) { + d1, d2 := &MockDest{id: 1}, &MockDest{id: 2} + s := NewService(nil, 1, d1, d2) + assert.NotNil(t, s) + + s.SubmitVerification(VerificationRequest{ + SiteID: "remark", + User: "testUser", + Email: "test@example.org", + Token: "testToken", + }) + s.SubmitVerification(VerificationRequest{}) + time.Sleep(time.Millisecond * 11) + s.SubmitVerification(VerificationRequest{}) + time.Sleep(time.Millisecond * 11) + s.Close() + + s.SubmitVerification(VerificationRequest{}) // safe to send after close + + assert.Equal(t, 2, len(d2.GetVerify()), "one request from three dropped from d2, got: %v", d2.GetVerify()) + + verifyDest := d1.GetVerify() + require.Equal(t, 2, len(verifyDest), "one request from three dropped from d1, got: %v", verifyDest) + assert.Equal(t, "remark", verifyDest[0].SiteID) + assert.Equal(t, "testUser", verifyDest[0].User) + assert.Equal(t, "test@example.org", verifyDest[0].Email) + assert.Equal(t, "testToken", verifyDest[0].Token) +} + func TestService_Many(t *testing.T) { d1, d2 := &MockDest{id: 1}, &MockDest{id: 2} s := NewService(nil, 5, d1, d2) @@ -70,16 +99,20 @@ func TestService_Many(t *testing.T) { for i := 0; i < 10; i++ { s.Submit(Request{Comment: store.Comment{ID: fmt.Sprintf("%d", 100+i)}}) + s.SubmitVerification(VerificationRequest{User: fmt.Sprintf("%d", 100+i)}) time.Sleep(time.Millisecond * time.Duration(rand.Int31n(20))) } s.Close() time.Sleep(time.Millisecond * 10) assert.NotEqual(t, 10, len(d1.Get()), "some comments dropped from d1") + assert.NotEqual(t, 10, len(d1.GetVerify()), "some verifications dropped from d1") assert.NotEqual(t, 10, len(d2.Get()), "some comments dropped from d2") + assert.NotEqual(t, 10, len(d2.GetVerify()), "some verifications dropped from d2") assert.True(t, d1.closed) assert.True(t, d2.closed) + assert.Equal(t, "mock id=1, closed=true", d1.String()) } func TestService_WithParent(t *testing.T) { diff --git a/backend/app/notify/telegram.go b/backend/app/notify/telegram.go index 092dc810..6a5ad467 100644 --- a/backend/app/notify/telegram.go +++ b/backend/app/notify/telegram.go @@ -28,7 +28,6 @@ const telegramAPIPrefix = "https://api.telegram.org/bot" // NewTelegram makes telegram bot for notifications func NewTelegram(token, channelID string, timeout time.Duration, api string) (*Telegram, error) { - if _, err := strconv.ParseInt(channelID, 10, 64); err != nil { channelID = "@" + channelID // if channelID not a number enforce @ prefix } @@ -86,10 +85,6 @@ func NewTelegram(token, channelID string, timeout time.Duration, api string) (*T // Send to telegram channel func (t *Telegram) Send(ctx context.Context, req Request) error { - if req.Comment.ID == "" { - // verification request received, send nothing - return nil - } if req.ForAdmin { // request for administrator received, do nothing with it // as we already sent message on request without this flag set @@ -152,6 +147,11 @@ func (t *Telegram) Send(ctx context.Context, req Request) error { return nil } +// SendVerification is not implemented for telegram +func (t *Telegram) SendVerification(_ context.Context, _ VerificationRequest) error { + return nil +} + func (t *Telegram) String() string { return "telegram: " + t.channelID } diff --git a/backend/app/notify/telegram_test.go b/backend/app/notify/telegram_test.go index 8660ab57..9f86e380 100644 --- a/backend/app/notify/telegram_test.go +++ b/backend/app/notify/telegram_test.go @@ -78,7 +78,18 @@ func TestTelegram_Send(t *testing.T) { assert.Contains(t, err.Error(), "unexpected telegram status code 404", "send on broken tg") assert.Equal(t, "telegram: @remark_test", tb.String()) - require.NoError(t, tb.Send(context.TODO(), Request{}), "Empty Comment doesn't send anything") +} + +func TestTelegram_SendVerification(t *testing.T) { + ts := mockTelegramServer() + defer ts.Close() + + tb, err := NewTelegram("good-token", "remark_test", 2*time.Second, ts.URL+"/") + assert.NoError(t, err) + assert.NotNil(t, tb) + + err = tb.SendVerification(context.TODO(), VerificationRequest{}) + assert.NoError(t, err) } func mockTelegramServer() *httptest.Server { diff --git a/backend/app/rest/api/rest_private.go b/backend/app/rest/api/rest_private.go index b21199ec..1dee5cc2 100644 --- a/backend/app/rest/api/rest_private.go +++ b/backend/app/rest/api/rest_private.go @@ -296,14 +296,12 @@ func (s *private) sendEmailConfirmationCtrl(w http.ResponseWriter, r *http.Reque return } - s.notifyService.Submit( - notify.Request{ - Email: address, - Verification: notify.VerificationMetadata{ - SiteID: siteID, - User: user.Name, - Token: tkn, - }, + s.notifyService.SubmitVerification( + notify.VerificationRequest{ + SiteID: siteID, + User: user.Name, + Email: address, + Token: tkn, }, ) diff --git a/backend/app/rest/api/rest_private_test.go b/backend/app/rest/api/rest_private_test.go index 76ac28d6..7619afe1 100644 --- a/backend/app/rest/api/rest_private_test.go +++ b/backend/app/rest/api/rest_private_test.go @@ -655,9 +655,9 @@ func TestRest_EmailNotification(t *testing.T) { require.Equal(t, http.StatusOK, resp.StatusCode, string(body)) // wait for mock notification Submit to kick off time.Sleep(time.Millisecond * 30) - require.Equal(t, 5, len(mockDestination.Get())) - require.NotEmpty(t, mockDestination.Get()[4].Verification) - verificationToken := mockDestination.Get()[4].Verification.Token + require.Equal(t, 1, len(mockDestination.GetVerify())) + assert.Equal(t, "good@example.com", mockDestination.GetVerify()[0].Email) + verificationToken := mockDestination.GetVerify()[0].Token // verify email req, err = http.NewRequest(http.MethodPost, ts.URL+fmt.Sprintf("/api/v1/email/confirm?site=remark42&tkn=%s", verificationToken), nil) @@ -703,8 +703,9 @@ func TestRest_EmailNotification(t *testing.T) { require.Equal(t, http.StatusCreated, resp.StatusCode, string(body)) // wait for mock notification Submit to kick off time.Sleep(time.Millisecond * 30) - require.Equal(t, 7, len(mockDestination.Get())) - assert.Equal(t, "good@example.com", mockDestination.Get()[5].Email) + require.Equal(t, 6, len(mockDestination.Get())) + assert.Equal(t, "good@example.com", mockDestination.Get()[4].Email) + assert.Equal(t, "admin@example.org", mockDestination.Get()[5].Email) // delete user's email req, err = http.NewRequest(http.MethodDelete, ts.URL+"/api/v1/email?site=remark42", nil) @@ -733,8 +734,8 @@ func TestRest_EmailNotification(t *testing.T) { require.Equal(t, http.StatusCreated, resp.StatusCode, string(body)) // wait for mock notification Submit to kick off time.Sleep(time.Millisecond * 30) - require.Equal(t, 9, len(mockDestination.Get())) - assert.Empty(t, mockDestination.Get()[7].Email) + require.Equal(t, 8, len(mockDestination.Get())) + assert.Empty(t, mockDestination.Get()[6].Email) } func TestRest_UserAllData(t *testing.T) {