mirrors/remark42
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

separate notify.SubmitVerification from notify.Submit

Browse Source
This commit is contained in:
Dmitry Verkhoturov
2020-10-11 23:13:24 -05:00
committed by Umputun
parent 1980f30666
commit c9b395f60d
9 changed files with 191 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/notify/email.go
+33 -18
View File
@@ -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(
backend/app/notify/email_test.go
+22 -11
View File
@@ -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
backend/app/notify/notify.go
+42 -13
View File
@@ -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
}
backend/app/notify/notify_mock.go
+30 -4
View File
@@ -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) }
backend/app/notify/notify_test.go
+33
View File
@@ -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) {
backend/app/notify/telegram.go
+5 -5
View File
@@ -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
}
backend/app/notify/telegram_test.go
+12 -1
View File
@@ -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 {
backend/app/rest/api/rest_private.go
+6 -8
View File
@@ -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,
},
)
backend/app/rest/api/rest_private_test.go
+8 -7
View File
@@ -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) {