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

Notify (#211)

Browse Source 
* add notify service with no destinations

* change to request for destinations

* add telegram notify and wire to main

* lint: missed err checks

* fix notify factory

* externalize timeout for notify

* fix race in notifier mock

* lint: check notify send status in tests

* lint: write warn

* lint: wrong assignment

* lint: two more writes in tests

* missing notif timeout default
This commit is contained in:
Umputun
2018-10-24 22:28:56 -05:00
committed by GitHub
parent be41daae73
commit 8c5531b178
8 changed files with 487 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+44 -39
View File
@@ -82,45 +82,50 @@ _this is the recommended way to run remark42_
#### Parameters
| Command line | Environment | Default | Description |
| ------------------ | ------------------ | --------------------- | ------------------------------------------------ |
| url | REMARK_URL | | url to remark42 server, _required_ |
| secret | SECRET | | secret key, _required_ |
| site | SITE | `remark` | site name(s), _multi_ |
| store.type | STORE_TYPE | `bolt` | type of storage, `bolt` or `mongo` |
| store.bolt.path | STORE_BOLT_PATH | `./var` | path to data directory |
| store.bolt.timeout | STORE_BOLT_TIMEOUT | `30s` | boltdb access timeout |
| mongo.url | MONGO_URL | | mongo url for all stores using mongodb |
| mongo.db | MONGO_DB | | mongo database |
| admin.shared.id | ADMIN_SHARED_ID | | admin names (list of user ids), _multi_ |
| admin.shared.email | ADMIN_SHARED_EMAIL | `admin@${REMARK_URL}` | admin email |
| backup | BACKUP_PATH | `./var/backup` | backups location |
| max-back | MAX_BACKUP_FILES | `10` | max backup files to keep |
| cache.max.items | CACHE_MAX_ITEMS | `1000` | max number of cached items, `0` - unlimited |
| cache.max.value | CACHE_MAX_VALUE | `65536` | max size of cached value, `0` - unlimited |
| cache.max.size | CACHE_MAX_SIZE | `50000000` | max size of all cached values, `0` - unlimited |
| avatar.type | AVATAR_TYPE | `fs` | type of avatar storage, `fs`, 'bolt`, or `mongo` |
| avatar.fs.path | AVATAR_FS_PATH | `./var/avatars` | avatars location for `fs` store |
| avatar.bolt.file | AVATAR_BOLT_FILE | `./var/avatars.db` | file name for `bolt` store |
| avatar.rsz-lmt | AVATAR_RSZ_LMT | 0 | max image size for resizing avatars on save |
| auth.ttl.jwt | AUTH_TTL_JWT | 5m | jwt TTL |
| auth.ttl.cookie | AUTH_TTL_COOKIE | 200h | cookie TTL |
| auth.google.cid | AUTH_GOOGLE_CID | | Google OAuth client ID |
| auth.google.csec | AUTH_GOOGLE_CSEC | | Google OAuth client secret |
| auth.facebook.cid | AUTH_FACEBOOK_CID | | Facebook OAuth client ID |
| auth.facebook.csec | AUTH_FACEBOOK_CSEC | | Facebook OAuth client secret |
| auth.github.cid | AUTH_GITHUB_CID | | Github OAuth client ID |
| auth.github.csec | AUTH_GITHUB_CSEC | | Github OAuth client secret |
| auth.yandex.cid | AUTH_YANDEX_CID | | Yandex OAuth client ID |
| auth.yandex.csec | AUTH_YANDEX_CSEC | | Yandex OAuth client secret |
| auth.dev | AUTH_DEV | false | local oauth2 server, development mode only |
| max-comment | MAX_COMMENT_SIZE | 2048 | comment's size limit |
| low-score | LOW_SCORE | `-5` | low score threshold |
| critical-score | CRITICAL_SCORE | `-10` | critical score threshold |
| edit-time | EDIT_TIME | `5m` | edit window |
| img-proxy | IMG_PROXY | `false` | enable http->https proxy for images |
| dbg | DEBUG | `false` | debug mode |
| dev-passwd | DEV_PASSWD | | password for `dev` user |
| Command line | Environment | Default | Description |
| ----------------------- | ----------------------- | --------------------- | ------------------------------------------------ |
| url | REMARK_URL | | url to remark42 server, _required_ |
| secret | SECRET | | secret key, _required_ |
| site | SITE | `remark` | site name(s), _multi_ |
| store.type | STORE_TYPE | `bolt` | type of storage, `bolt` or `mongo` |
| store.bolt.path | STORE_BOLT_PATH | `./var` | path to data directory |
| store.bolt.timeout | STORE_BOLT_TIMEOUT | `30s` | boltdb access timeout |
| mongo.url | MONGO_URL | | mongo url for all stores using mongodb |
| mongo.db | MONGO_DB | | mongo database |
| admin.shared.id | ADMIN_SHARED_ID | | admin names (list of user ids), _multi_ |
| admin.shared.email | ADMIN_SHARED_EMAIL | `admin@${REMARK_URL}` | admin email |
| backup | BACKUP_PATH | `./var/backup` | backups location |
| max-back | MAX_BACKUP_FILES | `10` | max backup files to keep |
| cache.max.items | CACHE_MAX_ITEMS | `1000` | max number of cached items, `0` - unlimited |
| cache.max.value | CACHE_MAX_VALUE | `65536` | max size of cached value, `0` - unlimited |
| cache.max.size | CACHE_MAX_SIZE | `50000000` | max size of all cached values, `0` - unlimited |
| avatar.type | AVATAR_TYPE | `fs` | type of avatar storage, `fs`, 'bolt`, or `mongo` |
| avatar.fs.path | AVATAR_FS_PATH | `./var/avatars` | avatars location for `fs` store |
| avatar.bolt.file | AVATAR_BOLT_FILE | `./var/avatars.db` | file name for `bolt` store |
| avatar.rsz-lmt | AVATAR_RSZ_LMT | 0 | max image size for resizing avatars on save |
| auth.ttl.jwt | AUTH_TTL_JWT | 5m | jwt TTL |
| auth.ttl.cookie | AUTH_TTL_COOKIE | 200h | cookie TTL |
| auth.google.cid | AUTH_GOOGLE_CID | | Google OAuth client ID |
| auth.google.csec | AUTH_GOOGLE_CSEC | | Google OAuth client secret |
| auth.facebook.cid | AUTH_FACEBOOK_CID | | Facebook OAuth client ID |
| auth.facebook.csec | AUTH_FACEBOOK_CSEC | | Facebook OAuth client secret |
| auth.github.cid | AUTH_GITHUB_CID | | Github OAuth client ID |
| auth.github.csec | AUTH_GITHUB_CSEC | | Github OAuth client secret |
| auth.yandex.cid | AUTH_YANDEX_CID | | Yandex OAuth client ID |
| auth.yandex.csec | AUTH_YANDEX_CSEC | | Yandex OAuth client secret |
| auth.dev | AUTH_DEV | false | local oauth2 server, development mode only |
| notify.type | NOTIFY_TYPE | none | type of notification (none or telegram) |
| notify.queue | NOTIFY_QUEUE | 100 | size of notification queue |
| notify.telegram.token | NOTIFY_TELEGRAM_TOKEN | | telegram token |
| notify.telegram.chan | NOTIFY_TELEGRAM_CHAN | | telegram channel |
| notify.telegram.timeout | NOTIFY_TELEGRAM_TIMEOUT | | telegram timeout |
| max-comment | MAX_COMMENT_SIZE | 2048 | comment's size limit |
| low-score | LOW_SCORE | `-5` | low score threshold |
| critical-score | CRITICAL_SCORE | `-10` | critical score threshold |
| edit-time | EDIT_TIME | `5m` | edit window |
| img-proxy | IMG_PROXY | `false` | enable http->https proxy for images |
| dbg | DEBUG | `false` | debug mode |
| dev-passwd | DEV_PASSWD | | password for `dev` user |
* command line parameters are long form `--<key>=value`, i.e. `--site=https://demo.remark42.com`
* _multi_ parameters separated by `,` in the environment or repeated with command line key, like `--site=s1 --site=s2 ...`
backend/app/cmd/server.go
+35 -1
View File
@@ -15,6 +15,7 @@ import (
"github.com/coreos/bbolt"
"github.com/go-pkgz/mongo"
"github.com/pkg/errors"
"github.com/umputun/remark/backend/app/notify"
"github.com/umputun/remark/backend/app/migrator"
"github.com/umputun/remark/backend/app/rest/api"
@@ -35,6 +36,7 @@ type ServerCommand struct {
Cache CacheGroup `group:"cache" namespace:"cache" env-namespace:"CACHE"`
Mongo MongoGroup `group:"mongo" namespace:"mongo" env-namespace:"MONGO"`
Admin AdminGroup `group:"admin" namespace:"admin" env-namespace:"ADMIN"`
Notify NotifyGroup `group:"notify" namespace:"notify" env-namespace:"NOTIFY"`
Sites []string `long:"site" env:"SITE" default:"remark" description:"site names" env-delim:","`
DevPasswd string `long:"dev-passwd" env:"DEV_PASSWD" default:"" description:"development mode password"`
@@ -116,6 +118,17 @@ type AdminGroup struct {
} `group:"shared" namespace:"shared" env-namespace:"SHARED"`
}
// NotifyGroup defines options for notification
type NotifyGroup struct {
Type string `long:"type" env:"TYPE" description:"type of notification" choice:"none" choice:"telegram" default:"none"`
QueueSize int `long:"queue" env:"QUEUE" description:"size of notification queue" default:"100"`
Telegram struct {
Token string `long:"token" env:"TOKEN" description:"telegram token"`
Channel string `long:"chan" env:"CHAN" description:"telegram channel"`
Timeout time.Duration `long:"timeout" env:"TIMEOUT" default:"5s" description:"telegram timeout"`
} `group:"telegram" namespace:"telegram" env-namespace:"TELEGRAM"`
}
// serverApp holds all active objects
type serverApp struct {
*ServerCommand
@@ -213,6 +226,11 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) {
KeyStore: adminStore,
}
notifyService, err := s.makeNotify(dataService)
if err != nil {
return nil, errors.Wrap(err, "failed to make notify service")
}
authProviders := s.makeAuthProviders(jwtService, avatarProxy, dataService)
imgProxy := &proxy.Image{Enabled: s.ImageProxy, RoutePath: "/api/v1/img", RemarkURL: s.RemarkURL}
commentFormatter := store.NewCommentFormatter(imgProxy)
@@ -235,7 +253,8 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) {
DevPasswd: s.DevPasswd,
PermissionChecker: dataService,
},
Cache: loadingCache,
Cache: loadingCache,
NotifyService: notifyService,
}
srv.ScoreThresholds.Low, srv.ScoreThresholds.Critical = s.LowScore, s.CriticalScore
@@ -443,3 +462,18 @@ func (s *ServerCommand) makeAuthProviders(jwt *auth.JWT, ap *proxy.Avatar, ds *s
}
return providers
}
func (s *ServerCommand) makeNotify(dataStore *service.DataStore) (*notify.Service, error) {
log.Printf("[INFO] make notify, type=%s", s.Notify.Type)
switch s.Notify.Type {
case "telegram":
tg, err := notify.NewTelegram(s.Notify.Telegram.Token, s.Notify.Telegram.Channel, s.Notify.Telegram.Timeout, "")
if err != nil {
return nil, errors.Wrap(err, "failed to create telegram notification destination")
}
return notify.NewService(dataStore, s.Notify.QueueSize, tg), nil
case "none":
return notify.NewService(dataStore, s.Notify.QueueSize), nil
}
return nil, errors.Errorf("unsupported notification type %q", s.Notify.Type)
}
backend/app/notify/notify.go
+99
View File
@@ -0,0 +1,99 @@
// Package notify provides notification functionality.
package notify
import (
"context"
"fmt"
"log"
"sync"
"github.com/umputun/remark/backend/app/store"
"github.com/umputun/remark/backend/app/store/service"
)
type request struct {
comment store.Comment
parent store.Comment
}
// 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
}
// Service delivers notifications to multiple destinations
type Service struct {
dataService *service.DataStore
destinations []Destination
queue chan request
closed bool
ctx context.Context
cancel context.CancelFunc
}
const defaultQueueSize = 100
const uiNav = "#remark42__comment-"
// NewService makes notification service routing comments to all destinations.
func NewService(dataService *service.DataStore, size int, destinations ...Destination) *Service {
if size <= 0 {
size = defaultQueueSize
}
ctx, cancel := context.WithCancel(context.Background())
res := Service{
dataService: dataService,
queue: make(chan request, size),
destinations: destinations,
ctx: ctx,
cancel: cancel,
}
if len(destinations) > 0 {
go res.do()
}
log.Printf("[INFO] create notifier service, queue size=%d", size)
return &res
}
// Submit comment to internal channel if not busy, drop if can't send
func (s *Service) Submit(comment store.Comment) {
if len(s.destinations) == 0 || s.closed {
return
}
parentComment := store.Comment{}
if s.dataService != nil {
if p, err := s.dataService.Get(comment.Locator, comment.ParentID); err == nil {
parentComment = p
}
}
select {
case s.queue <- request{comment: comment, parent: parentComment}:
default:
log.Printf("[WARN] can't send comment notification to queue, %+v", comment)
}
}
// Close queue channel and wait for completion
func (s *Service) Close() {
close(s.queue)
s.cancel()
<-s.ctx.Done()
s.closed = true
}
func (s *Service) do() {
for c := range s.queue {
var wg sync.WaitGroup
for _, dest := range s.destinations {
wg.Add(1)
go func(d Destination) {
if err := d.Send(s.ctx, c); err != nil {
log.Printf("[WARN] failed to send to %s", d)
}
wg.Done()
}(dest)
}
wg.Wait()
}
}
backend/app/notify/notify_test.go
+111
View File
@@ -0,0 +1,111 @@
package notify
import (
"context"
"fmt"
"log"
"math/rand"
"sync"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/umputun/remark/backend/app/store"
)
func TestService_NoDestinations(t *testing.T) {
s := NewService(nil, 1)
assert.NotNil(t, s)
s.Submit(store.Comment{ID: "123"})
s.Submit(store.Comment{ID: "123"})
s.Submit(store.Comment{ID: "123"})
s.Close()
}
func TestService_WithDestinations(t *testing.T) {
d1, d2 := &mockDest{id: 1}, &mockDest{id: 2}
s := NewService(nil, 1, d1, d2)
assert.NotNil(t, s)
s.Submit(store.Comment{ID: "100"})
time.Sleep(time.Millisecond * 110)
s.Submit(store.Comment{ID: "101"})
time.Sleep(time.Millisecond * 110)
s.Submit(store.Comment{ID: "102"})
time.Sleep(time.Millisecond * 110)
s.Close()
assert.Equal(t, 3, len(d1.get()), "got all comments to d1")
assert.Equal(t, 3, len(d2.get()), "got all comments to d2")
assert.Equal(t, "100", d1.get()[0].ID)
assert.Equal(t, "101", d1.get()[1].ID)
assert.Equal(t, "102", d1.get()[2].ID)
}
func TestService_WithDrops(t *testing.T) {
d1, d2 := &mockDest{id: 1}, &mockDest{id: 2}
s := NewService(nil, 1, d1, d2)
assert.NotNil(t, s)
s.Submit(store.Comment{ID: "100"})
s.Submit(store.Comment{ID: "101"})
time.Sleep(time.Millisecond * 110)
s.Submit(store.Comment{ID: "102"})
time.Sleep(time.Millisecond * 110)
s.Close()
s.Submit(store.Comment{ID: "111"}) // safe to send after close
assert.Equal(t, 2, len(d1.get()), "one comment dropped from d1")
assert.Equal(t, 2, len(d2.get()), "one comment dropped from d2")
}
func TestService_Many(t *testing.T) {
d1, d2 := &mockDest{id: 1}, &mockDest{id: 2}
s := NewService(nil, 5, d1, d2)
assert.NotNil(t, s)
for i := 0; i < 10; i++ {
s.Submit(store.Comment{ID: fmt.Sprintf("%d", 100+i)})
time.Sleep(time.Millisecond * time.Duration(rand.Int31n(200)))
}
s.Close()
time.Sleep(time.Millisecond * 10)
assert.NotEqual(t, 10, len(d1.get()), "some comments dropped from d1")
assert.NotEqual(t, 10, len(d2.get()), "some comments dropped from d2")
assert.True(t, d1.closed)
assert.True(t, d2.closed)
}
type mockDest struct {
data []store.Comment
id int
closed bool
lock sync.Mutex
}
func (m *mockDest) Send(ctx context.Context, r request) error {
m.lock.Lock()
defer m.lock.Unlock()
select {
case <-time.After(100 * time.Millisecond):
m.data = append(m.data, r.comment)
log.Printf("sent %s -> %d", r.comment.ID, m.id)
case <-ctx.Done():
log.Printf("ctx closed %d", m.id)
m.closed = true
}
return nil
}
func (m *mockDest) get() []store.Comment {
m.lock.Lock()
defer m.lock.Unlock()
res := make([]store.Comment, len(m.data))
copy(res, m.data)
return res
}
func (m *mockDest) String() string { return fmt.Sprintf("mock id=%d, closed=%v", m.id, m.closed) }
backend/app/notify/telegram.go
+119
View File
@@ -0,0 +1,119 @@
package notify
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"net/url"
"strings"
"time"
"github.com/pkg/errors"
)
// Telegram implements notify.Destination for telegram
type Telegram struct {
channelName string
token string
apiPrefix string
timeout time.Duration
}
const telegramTimeOut = 2500 * time.Millisecond
const telegramAPIPrefix = "https://api.telegram.org/bot"
// NewTelegram makes telegram bot for notifications
func NewTelegram(token string, channelName string, timeout time.Duration, api string) (*Telegram, error) {
log.Printf("[DEBUG] create new telegram notifier for cham %s, timeout=%s, api=%s", channelName, timeout, api)
res := Telegram{channelName: channelName, token: token, apiPrefix: api, timeout: timeout}
res.channelName = strings.TrimPrefix(res.channelName, "@")
if res.apiPrefix == "" {
res.apiPrefix = telegramAPIPrefix
}
if res.timeout == 0 {
res.timeout = telegramTimeOut
}
client := http.Client{Timeout: telegramTimeOut}
resp, err := client.Get(fmt.Sprintf("%s%s/getMe", res.apiPrefix, token))
if err != nil {
return nil, errors.Wrap(err, "can't initialize telegram notifications")
}
defer func() {
if err = resp.Body.Close(); err != nil {
log.Printf("[WARN] can't close request body, %s", err)
}
}()
if resp.StatusCode != http.StatusOK {
return nil, errors.Errorf("unexpected telegram status code %d", resp.StatusCode)
}
tgResp := struct {
OK bool `json:"ok"`
Result struct {
FirstName string `json:"first_name"`
ID uint64 `json:"id"`
IsBot bool `json:"is_bot"`
UserName string `json:"username"`
}
}{}
if err = json.NewDecoder(resp.Body).Decode(&tgResp); err != nil {
return nil, errors.Wrap(err, "can't authorize to telegram")
}
if !tgResp.OK || !tgResp.Result.IsBot {
return nil, errors.Errorf("unexpected telegram response %+v", tgResp)
}
return &res, nil
}
// Send to telegram channel
func (t *Telegram) Send(ctx context.Context, req request) error {
client := http.Client{Timeout: telegramTimeOut}
from := req.comment.User.Name
if req.comment.ParentID != "" {
from += " -> " + req.parent.User.Name
}
from = "*" + from + "*"
link := fmt.Sprintf("[comment](%s)", req.comment.Locator.URL+uiNav+req.comment.ID)
msg := fmt.Sprintf("%s\n\n%s\n\n%s", from, req.comment.Text, link)
r, err := http.NewRequest("GET", fmt.Sprintf("%s%s/sendMessage?chat_id=@%s&text=%s&parse_mode=Markdown",
t.apiPrefix, t.token, t.channelName, url.QueryEscape(msg)), nil)
if err != nil {
return errors.Wrap(err, "failed to make telegram request")
}
r = r.WithContext(ctx)
resp, err := client.Do(r)
if err != nil {
return errors.Wrap(err, "failed to get telegram response")
}
defer func() {
if err = resp.Body.Close(); err != nil {
log.Printf("[WARN] can't close request body, %s", err)
}
}()
if resp.StatusCode != http.StatusOK {
return errors.Errorf("unexpected telegram status code %d", resp.StatusCode)
}
tgResp := struct {
OK bool `json:"ok"`
}{}
if err = json.NewDecoder(resp.Body).Decode(&tgResp); err != nil {
return errors.Wrap(err, "can't decode telegram response")
}
return nil
}
func (t *Telegram) String() string {
return "telegram-" + t.channelName
}
backend/app/notify/telegram_test.go
+73
View File
@@ -0,0 +1,73 @@
package notify
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/go-chi/chi"
"github.com/stretchr/testify/assert"
"github.com/umputun/remark/backend/app/store"
)
func TestTelegram_New(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 = NewTelegram("bad-resp", "remark_test", 2*time.Second, ts.URL+"/")
assert.NotNil(t, err)
_, err = NewTelegram("404", "remark_test", 2*time.Second, ts.URL+"/")
assert.NotNil(t, err)
}
func TestTelegram_Send(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.Send(context.TODO(), request{comment: store.Comment{Text: "some text"}})
assert.NoError(t, err)
}
func mockTelegramServer() *httptest.Server {
router := chi.NewRouter()
router.Get("/good-token/getMe", func(w http.ResponseWriter, r *http.Request) {
s := `{"ok": true,
"result": {
"first_name": "comments_test",
"id": 707381019,
"is_bot": true,
"username": "remark42_test_bot"
}}`
_, _ = w.Write([]byte(s))
})
router.Get("/bad-resp/getMe", func(w http.ResponseWriter, r *http.Request) {
s := `{"ok": false,
"result": {
"first_name": "comments_test",
"id": 707381019,
"is_bot": false,
"username": "remark42_test_bot"
}}`
_, _ = w.Write([]byte(s))
})
router.Get("/404/getMe", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(404)
})
router.Get("/good-token/sendMessage", func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write([]byte(`{"ok": true}`))
})
return httptest.NewServer(router)
}
backend/app/rest/api/rest.go
+2
View File
@@ -21,6 +21,7 @@ import (
"github.com/go-chi/render"
"github.com/pkg/errors"
"github.com/rakyll/statik/fs"
"github.com/umputun/remark/backend/app/notify"
"github.com/umputun/remark/backend/app/rest"
"github.com/umputun/remark/backend/app/rest/auth"
@@ -41,6 +42,7 @@ type Rest struct {
ImageProxy *proxy.Image
CommentFormatter *store.CommentFormatter
Migrator *Migrator
NotifyService *notify.Service
WebRoot string
RemarkURL string
backend/app/rest/api/rest_private.go
+4
View File
@@ -73,6 +73,10 @@ func (s *Rest) createCommentCtrl(w http.ResponseWriter, r *http.Request) {
s.Cache.Flush(cache.Flusher(comment.Locator.SiteID).
Scopes(comment.Locator.URL, lastCommentsScope, comment.User.ID, comment.Locator.SiteID))
if s.NotifyService != nil {
s.NotifyService.Submit(finalComment)
}
render.Status(r, http.StatusCreated)
render.JSON(w, r, &finalComment)
}