apollo-backend/internal/worker/notifications.go

429 lines
11 KiB
Go
Raw Normal View History

package worker
2021-07-08 23:03:46 +00:00
import (
"context"
"fmt"
"os"
"strconv"
"time"
"github.com/DataDog/datadog-go/statsd"
"github.com/adjust/rmq/v4"
"github.com/go-redis/redis/v8"
"github.com/jackc/pgx/v4"
"github.com/jackc/pgx/v4/pgxpool"
"github.com/sideshow/apns2"
"github.com/sideshow/apns2/payload"
"github.com/sideshow/apns2/token"
"github.com/sirupsen/logrus"
"github.com/christianselig/apollo-backend/internal/data"
"github.com/christianselig/apollo-backend/internal/reddit"
)
const (
backoff = 5 // How long we wait in between checking for notifications, in seconds
2021-07-09 00:44:12 +00:00
pollDuration = 10 * time.Millisecond
2021-07-08 23:26:15 +00:00
rate = 0.1
2021-07-08 23:03:46 +00:00
)
type notificationsWorker struct {
logger *logrus.Logger
statsd *statsd.Client
db *pgxpool.Pool
redis *redis.Client
queue rmq.Connection
reddit *reddit.Client
apns *token.Token
}
2021-07-08 23:03:46 +00:00
func NewNotificationsWorker(logger *logrus.Logger, statsd *statsd.Client, db *pgxpool.Pool, redis *redis.Client, queue rmq.Connection) Worker {
reddit := reddit.NewClient(
2021-07-08 23:03:46 +00:00
os.Getenv("REDDIT_CLIENT_ID"),
os.Getenv("REDDIT_CLIENT_SECRET"),
statsd,
)
var apns *token.Token
2021-07-08 23:03:46 +00:00
{
authKey, err := token.AuthKeyFromBytes([]byte(os.Getenv("APPLE_KEY_PKEY")))
if err != nil {
panic(err)
}
apns = &token.Token{
2021-07-08 23:03:46 +00:00
AuthKey: authKey,
KeyID: os.Getenv("APPLE_KEY_ID"),
TeamID: os.Getenv("APPLE_TEAM_ID"),
}
}
return &notificationsWorker{
logger,
statsd,
db,
redis,
queue,
reddit,
apns,
2021-07-08 23:03:46 +00:00
}
}
2021-07-08 23:03:46 +00:00
func (nw *notificationsWorker) Start(consumers int) error {
queue, err := nw.queue.OpenQueue("notifications")
2021-07-08 23:03:46 +00:00
if err != nil {
return err
2021-07-08 23:03:46 +00:00
}
nw.logger.WithFields(logrus.Fields{
"numConsumers": consumers,
2021-07-09 03:50:34 +00:00
}).Info("starting up notifications worker")
2021-07-09 01:28:43 +00:00
prefetchLimit := int64(consumers * 32)
2021-07-08 23:03:46 +00:00
if err := queue.StartConsuming(prefetchLimit, pollDuration); err != nil {
return err
2021-07-08 23:03:46 +00:00
}
2021-07-09 03:50:34 +00:00
host, _ := os.Hostname()
for i := 0; i < consumers; i++ {
2021-07-09 03:50:34 +00:00
name := fmt.Sprintf("consumer %s-%d", host, i)
2021-07-08 23:03:46 +00:00
consumer := NewNotificationsConsumer(nw, i)
2021-07-08 23:03:46 +00:00
if _, err := queue.AddConsumer(name, consumer); err != nil {
return err
2021-07-08 23:03:46 +00:00
}
}
return nil
}
2021-07-08 23:03:46 +00:00
func (nw *notificationsWorker) Stop() {
<-nw.queue.StopAllConsuming() // wait for all Consume() calls to finish
2021-07-08 23:03:46 +00:00
}
type notificationsConsumer struct {
*notificationsWorker
tag int
2021-07-08 23:03:46 +00:00
apnsSandbox *apns2.Client
apnsProduction *apns2.Client
}
func NewNotificationsConsumer(nw *notificationsWorker, tag int) *notificationsConsumer {
return &notificationsConsumer{
nw,
2021-07-08 23:03:46 +00:00
tag,
apns2.NewTokenClient(nw.apns),
apns2.NewTokenClient(nw.apns).Production(),
2021-07-08 23:03:46 +00:00
}
}
func (nc *notificationsConsumer) Consume(delivery rmq.Delivery) {
2021-07-08 23:03:46 +00:00
ctx := context.Background()
2021-07-09 00:37:39 +00:00
defer func() {
2021-07-09 01:44:06 +00:00
lockKey := fmt.Sprintf("locks:accounts:%s", delivery.Payload())
if err := nc.redis.Del(ctx, lockKey).Err(); err != nil {
nc.logger.WithFields(logrus.Fields{
2021-07-09 01:44:06 +00:00
"lockKey": lockKey,
"err": err,
}).Error("failed to remove lock")
}
2021-07-09 00:37:39 +00:00
}()
2021-07-09 00:14:12 +00:00
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": delivery.Payload(),
}).Debug("starting job")
id, err := strconv.ParseInt(delivery.Payload(), 10, 64)
if err != nil {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": delivery.Payload(),
"err": err,
}).Error("failed to parse account ID")
delivery.Reject()
return
}
2021-07-09 03:50:34 +00:00
defer delivery.Ack()
2021-07-08 23:03:46 +00:00
now := float64(time.Now().UnixNano()/int64(time.Millisecond)) / 1000
stmt := `SELECT
id,
2021-07-09 01:44:06 +00:00
username,
2021-07-12 18:36:08 +00:00
account_id,
2021-07-08 23:03:46 +00:00
access_token,
refresh_token,
expires_at,
last_message_id,
last_checked_at
FROM accounts
WHERE id = $1`
account := &data.Account{}
if err := nc.db.QueryRow(ctx, stmt, id).Scan(
2021-07-08 23:03:46 +00:00
&account.ID,
2021-07-09 01:44:06 +00:00
&account.Username,
2021-07-12 18:36:08 +00:00
&account.AccountID,
2021-07-08 23:03:46 +00:00
&account.AccessToken,
&account.RefreshToken,
&account.ExpiresAt,
&account.LastMessageID,
&account.LastCheckedAt,
); err != nil {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
"err": err,
}).Error("failed to fetch account from database")
return
}
2021-07-13 17:58:52 +00:00
if err = nc.db.BeginFunc(ctx, func(tx pgx.Tx) error {
stmt := `
UPDATE accounts
SET last_checked_at = $1
WHERE id = $2`
_, err := tx.Exec(ctx, stmt, now, account.ID)
return err
}); err != nil {
nc.logger.WithFields(logrus.Fields{
"accountID": id,
"err": err,
}).Error("failed to update last_checked_at for account")
return
}
rac := nc.reddit.NewAuthenticatedClient(account.RefreshToken, account.AccessToken)
2021-07-08 23:03:46 +00:00
if account.ExpiresAt < int64(now) {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
}).Debug("refreshing reddit token")
2021-07-08 23:26:15 +00:00
2021-07-08 23:03:46 +00:00
tokens, err := rac.RefreshTokens()
if err != nil {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
"err": err,
}).Error("failed to refresh reddit tokens")
return
}
2021-07-12 21:10:10 +00:00
2021-07-13 18:47:50 +00:00
// Update account
account.AccessToken = tokens.AccessToken
account.RefreshToken = tokens.RefreshToken
account.ExpiresAt = int64(now + 3540)
2021-07-12 21:10:10 +00:00
// Refresh client
rac = nc.reddit.NewAuthenticatedClient(tokens.RefreshToken, tokens.AccessToken)
2021-07-12 21:10:10 +00:00
err = nc.db.BeginFunc(ctx, func(tx pgx.Tx) error {
2021-07-08 23:03:46 +00:00
stmt := `
UPDATE accounts
SET access_token = $1, refresh_token = $2, expires_at = $3 WHERE id = $4`
2021-07-13 19:08:39 +00:00
_, err := tx.Exec(ctx, stmt, account.AccessToken, account.RefreshToken, account.ExpiresAt, account.ID)
2021-07-08 23:03:46 +00:00
return err
})
if err != nil {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
"err": err,
}).Error("failed to update reddit tokens for account")
return
}
}
2021-07-13 19:08:39 +00:00
// Only update delay on accounts we can actually check, otherwise it skews
// the numbers too much.
if account.LastCheckedAt > 0 {
latency := now - account.LastCheckedAt - float64(backoff)
nc.statsd.Histogram("apollo.queue.delay", latency, []string{}, rate)
}
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
}).Debug("fetching message inbox")
msgs, err := rac.MessageUnread(account.LastMessageID)
2021-07-08 23:03:46 +00:00
if err != nil {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
"err": err,
}).Error("failed to fetch message inbox")
return
}
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
"count": len(msgs.MessageListing.Messages),
}).Debug("fetched messages")
if len(msgs.MessageListing.Messages) == 0 {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
}).Debug("no new messages, bailing early")
return
}
// Set latest message we alerted on
latestMsg := msgs.MessageListing.Messages[0]
if err = nc.db.BeginFunc(ctx, func(tx pgx.Tx) error {
stmt := `
UPDATE accounts
SET last_message_id = $1
WHERE id = $2`
_, err := tx.Exec(ctx, stmt, latestMsg.FullName(), account.ID)
return err
}); err != nil {
nc.logger.WithFields(logrus.Fields{
"accountID": id,
"err": err,
}).Error("failed to update last_message_id for account")
return
}
2021-07-08 23:03:46 +00:00
// Let's populate this with the latest message so we don't flood users with stuff
if account.LastMessageID == "" {
nc.logger.WithFields(logrus.Fields{
"accountID": delivery.Payload(),
}).Debug("populating first message ID to prevent spamming")
2021-07-08 23:03:46 +00:00
return
}
devices := []data.Device{}
stmt = `
SELECT apns_token, sandbox
FROM devices
2021-07-12 19:37:03 +00:00
INNER JOIN devices_accounts ON devices.id = devices_accounts.device_id
2021-07-08 23:03:46 +00:00
WHERE devices_accounts.account_id = $1`
rows, err := nc.db.Query(ctx, stmt, account.ID)
2021-07-08 23:03:46 +00:00
if err != nil {
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
"err": err,
}).Error("failed to fetch account devices")
return
}
defer rows.Close()
for rows.Next() {
var device data.Device
rows.Scan(&device.APNSToken, &device.Sandbox)
devices = append(devices, device)
}
for _, msg := range msgs.MessageListing.Messages {
notification := &apns2.Notification{}
notification.Topic = "com.christianselig.Apollo"
2021-07-12 18:15:13 +00:00
notification.Payload = payloadFromMessage(account, &msg, len(msgs.MessageListing.Messages))
2021-07-08 23:03:46 +00:00
for _, device := range devices {
notification.DeviceToken = device.APNSToken
client := nc.apnsProduction
2021-07-08 23:03:46 +00:00
if device.Sandbox {
client = nc.apnsSandbox
2021-07-08 23:03:46 +00:00
}
res, err := client.Push(notification)
if err != nil {
nc.statsd.Incr("apns.notification.errors", []string{}, 1)
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": id,
"err": err,
"status": res.StatusCode,
"reason": res.Reason,
}).Error("failed to send notification")
} else {
nc.statsd.Incr("apns.notification.sent", []string{}, 1)
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": delivery.Payload(),
"token": device.APNSToken,
"redditUser": account.Username,
2021-07-09 03:12:50 +00:00
}).Info("sent notification")
2021-07-08 23:03:46 +00:00
}
}
}
nc.logger.WithFields(logrus.Fields{
2021-07-08 23:03:46 +00:00
"accountID": delivery.Payload(),
}).Debug("finishing job")
}
2021-07-12 18:15:13 +00:00
func payloadFromMessage(acct *data.Account, msg *reddit.MessageData, badgeCount int) *payload.Payload {
2021-07-10 18:51:42 +00:00
postBody := msg.Body
if len(postBody) > 2000 {
postBody = msg.Body[:2000]
}
postTitle := msg.LinkTitle
if postTitle == "" {
postTitle = msg.Subject
}
if len(postTitle) > 75 {
postTitle = fmt.Sprintf("%s…", postTitle[0:75])
}
2021-07-12 18:36:08 +00:00
payload := payload.
NewPayload().
AlertBody(postBody).
AlertSummaryArg(msg.Author).
Badge(badgeCount).
Custom("account_id", acct.AccountID).
Custom("author", msg.Author).
Custom("destination_author", msg.Destination).
Custom("parent_id", msg.ParentID).
Custom("post_title", msg.LinkTitle).
Custom("subreddit", msg.Subreddit).
MutableContent().
Sound("traloop.wav")
2021-07-10 18:51:42 +00:00
switch {
case (msg.Kind == "t1" && msg.Type == "username_mention"):
title := fmt.Sprintf(`Mention in “%s”`, postTitle)
payload = payload.AlertTitle(title).Custom("type", "username")
pType, _ := reddit.SplitID(msg.ParentID)
if pType == "t1" {
payload = payload.Category("inbox-username-mention-context")
} else {
payload = payload.Category("inbox-username-mention-no-context")
}
payload = payload.Custom("subject", "comment").ThreadID("comment")
break
case (msg.Kind == "t1" && msg.Type == "post_reply"):
title := fmt.Sprintf(`%s to “%s”`, msg.Author, postTitle)
2021-07-12 18:36:08 +00:00
payload = payload.
AlertTitle(title).
Category("inbox-post-reply").
Custom("post_id", msg.ID).
Custom("subject", "comment").
Custom("type", "post").
ThreadID("comment")
2021-07-10 18:51:42 +00:00
break
case (msg.Kind == "t1" && msg.Type == "comment_reply"):
title := fmt.Sprintf(`%s in “%s”`, msg.Author, postTitle)
2021-07-12 19:51:02 +00:00
postID := reddit.PostIDFromContext(msg.Context)
2021-07-12 18:36:08 +00:00
payload = payload.
AlertTitle(title).
Category("inbox-comment-reply").
Custom("comment_id", msg.ID).
Custom("post_id", postID).
Custom("subject", "comment").
Custom("type", "comment").
ThreadID("comment")
2021-07-10 18:51:42 +00:00
break
case (msg.Kind == "t4"):
title := fmt.Sprintf(`Message from %s`, msg.Author)
2021-07-12 18:36:08 +00:00
payload = payload.
AlertTitle(title).
AlertSubtitle(postTitle).
Category("inbox-private-message").
Custom("type", "private-message")
2021-07-10 18:51:42 +00:00
break
}
return payload
}