apollo-backend/internal/worker/trending.go

344 lines
8.6 KiB
Go
Raw Normal View History

2021-10-10 15:51:42 +00:00
package worker
import (
"context"
"fmt"
"os"
2022-05-26 22:17:49 +00:00
"sort"
2021-10-10 15:51:42 +00:00
"strconv"
"time"
"github.com/DataDog/datadog-go/statsd"
"github.com/adjust/rmq/v4"
"github.com/go-redis/redis/v8"
"github.com/jackc/pgx/v4/pgxpool"
"github.com/sideshow/apns2"
"github.com/sideshow/apns2/payload"
"github.com/sideshow/apns2/token"
2022-05-23 18:17:25 +00:00
"go.uber.org/zap"
2021-10-10 15:51:42 +00:00
"github.com/christianselig/apollo-backend/internal/domain"
"github.com/christianselig/apollo-backend/internal/reddit"
"github.com/christianselig/apollo-backend/internal/repository"
)
type trendingWorker struct {
context.Context
2022-05-23 18:17:25 +00:00
logger *zap.Logger
2021-10-10 15:51:42 +00:00
statsd *statsd.Client
redis *redis.Client
queue rmq.Connection
reddit *reddit.Client
apns *token.Token
consumers int
accountRepo domain.AccountRepository
deviceRepo domain.DeviceRepository
subredditRepo domain.SubredditRepository
watcherRepo domain.WatcherRepository
}
2022-05-01 17:04:05 +00:00
const trendingNotificationTitleFormat = "🔥 r/%s Trending"
2021-10-10 15:51:42 +00:00
2022-05-23 18:17:25 +00:00
func NewTrendingWorker(ctx context.Context, logger *zap.Logger, statsd *statsd.Client, db *pgxpool.Pool, redis *redis.Client, queue rmq.Connection, consumers int) Worker {
2021-10-10 15:51:42 +00:00
reddit := reddit.NewClient(
os.Getenv("REDDIT_CLIENT_ID"),
os.Getenv("REDDIT_CLIENT_SECRET"),
statsd,
2022-03-12 17:50:05 +00:00
redis,
2021-10-10 15:51:42 +00:00
consumers,
)
var apns *token.Token
{
authKey, err := token.AuthKeyFromFile(os.Getenv("APPLE_KEY_PATH"))
if err != nil {
panic(err)
}
apns = &token.Token{
AuthKey: authKey,
KeyID: os.Getenv("APPLE_KEY_ID"),
TeamID: os.Getenv("APPLE_TEAM_ID"),
}
}
return &trendingWorker{
ctx,
2021-10-10 15:51:42 +00:00
logger,
statsd,
redis,
queue,
reddit,
apns,
consumers,
repository.NewPostgresAccount(db),
repository.NewPostgresDevice(db),
repository.NewPostgresSubreddit(db),
repository.NewPostgresWatcher(db),
}
}
func (tw *trendingWorker) Start() error {
queue, err := tw.queue.OpenQueue("trending")
if err != nil {
return err
}
2022-05-23 18:17:25 +00:00
tw.logger.Info("starting up trending subreddits worker", zap.Int("consumers", tw.consumers))
2021-10-10 15:51:42 +00:00
prefetchLimit := int64(tw.consumers * 2)
if err := queue.StartConsuming(prefetchLimit, pollDuration); err != nil {
return err
}
host, _ := os.Hostname()
for i := 0; i < tw.consumers; i++ {
name := fmt.Sprintf("consumer %s-%d", host, i)
consumer := NewTrendingConsumer(tw, i)
if _, err := queue.AddConsumer(name, consumer); err != nil {
return err
}
}
return nil
}
func (tw *trendingWorker) Stop() {
<-tw.queue.StopAllConsuming() // wait for all Consume() calls to finish
}
type trendingConsumer struct {
*trendingWorker
tag int
apnsSandbox *apns2.Client
apnsProduction *apns2.Client
}
func NewTrendingConsumer(tw *trendingWorker, tag int) *trendingConsumer {
return &trendingConsumer{
tw,
tag,
apns2.NewTokenClient(tw.apns),
apns2.NewTokenClient(tw.apns).Production(),
}
}
func (tc *trendingConsumer) Consume(delivery rmq.Delivery) {
id, err := strconv.ParseInt(delivery.Payload(), 10, 64)
if err != nil {
2022-05-23 18:17:25 +00:00
tc.logger.Error("failed to parse subreddit id from payload", zap.Error(err), zap.String("payload", delivery.Payload()))
2021-10-10 15:51:42 +00:00
_ = delivery.Reject()
return
}
2022-05-23 18:17:25 +00:00
tc.logger.Debug("starting job", zap.Int64("subreddit#id", id))
2021-10-10 15:51:42 +00:00
defer func() { _ = delivery.Ack() }()
subreddit, err := tc.subredditRepo.GetByID(tc, id)
2021-10-10 15:51:42 +00:00
if err != nil {
2022-05-23 18:17:25 +00:00
tc.logger.Error("failed to fetch subreddit from database", zap.Error(err), zap.Int64("subreddit#id", id))
2021-10-10 15:51:42 +00:00
return
}
watchers, err := tc.watcherRepo.GetByTrendingSubredditID(tc, subreddit.ID)
2021-10-10 15:51:42 +00:00
if err != nil {
2022-05-23 18:17:25 +00:00
tc.logger.Error("failed to fetch watchers from database",
zap.Error(err),
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
)
2021-10-10 15:51:42 +00:00
return
}
if len(watchers) == 0 {
2022-05-23 18:17:25 +00:00
tc.logger.Debug("no watchers for subreddit, bailing early",
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
)
2021-10-10 15:51:42 +00:00
return
}
tps, err := tc.reddit.SubredditTop(tc, subreddit.Name, reddit.WithQuery("t", "week"))
2021-10-10 15:51:42 +00:00
if err != nil {
2022-05-25 23:55:51 +00:00
tc.logger.Error("failed to fetch weeks's top posts",
2022-05-23 18:17:25 +00:00
zap.Error(err),
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
)
2021-10-10 15:51:42 +00:00
return
}
2022-05-23 18:17:25 +00:00
2022-05-25 23:55:51 +00:00
tc.logger.Debug("loaded weeks's top posts",
2022-05-23 18:17:25 +00:00
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
zap.Int("count", tps.Count),
)
2021-10-10 15:51:42 +00:00
if tps.Count == 0 {
2022-05-23 18:17:25 +00:00
tc.logger.Debug("no top posts, bailing early",
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
)
2021-10-10 15:51:42 +00:00
return
}
if tps.Count < 20 {
2022-05-23 18:17:25 +00:00
tc.logger.Debug("no top posts, bailing early",
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
zap.Int("count", tps.Count),
)
2021-10-10 15:51:42 +00:00
return
}
2022-05-26 22:17:49 +00:00
sort.SliceStable(tps.Children, func(i, j int) bool {
return tps.Children[i].Score > tps.Children[j].Score
})
2022-05-25 23:55:51 +00:00
posts := make([]string, tps.Count)
for i, post := range tps.Children {
posts[i] = fmt.Sprintf("%s: %d", post.Title, post.Score)
}
2021-10-10 15:51:42 +00:00
middlePost := tps.Count / 2
medianScore := tps.Children[middlePost].Score
2022-05-25 19:57:55 +00:00
tc.logger.Info("calculated median score",
2022-05-23 18:17:25 +00:00
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
2022-05-25 23:55:51 +00:00
zap.Strings("subreddit#posts", posts),
2022-05-23 18:17:25 +00:00
zap.Int64("score", medianScore),
)
2021-10-10 15:51:42 +00:00
hps, err := tc.reddit.SubredditHot(tc, subreddit.Name)
2021-10-10 15:51:42 +00:00
if err != nil {
2022-05-23 18:17:25 +00:00
tc.logger.Error("failed to fetch hot posts",
zap.Error(err),
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
)
2021-10-10 15:51:42 +00:00
return
}
2022-05-23 18:17:25 +00:00
tc.logger.Debug("loaded hot posts",
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
zap.Int("count", hps.Count),
)
2021-10-10 15:51:42 +00:00
// Trending only counts for posts less than 2 days old
2022-03-28 21:05:01 +00:00
threshold := time.Now().Add(-24 * time.Hour * 2)
2021-10-10 15:51:42 +00:00
for _, post := range hps.Children {
if post.Score < medianScore {
continue
}
2022-03-28 21:05:01 +00:00
if post.CreatedAt.Before(threshold) {
2021-10-10 15:51:42 +00:00
break
}
notification := &apns2.Notification{}
notification.Topic = "com.christianselig.Apollo"
notification.Payload = payloadFromTrendingPost(post)
for _, watcher := range watchers {
2022-03-28 21:05:01 +00:00
if watcher.CreatedAt.After(post.CreatedAt) {
2021-10-10 15:51:42 +00:00
continue
}
lockKey := fmt.Sprintf("watcher:trending:%d:%s", watcher.DeviceID, post.ID)
notified, _ := tc.redis.Get(tc, lockKey).Bool()
2021-10-10 15:51:42 +00:00
if notified {
2022-05-23 18:17:25 +00:00
tc.logger.Debug("already notified, skipping",
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
zap.Int64("watcher#id", watcher.ID),
zap.String("post#id", post.ID),
)
2021-10-10 15:51:42 +00:00
continue
}
tc.redis.SetEX(tc, lockKey, true, 48*time.Hour)
2021-10-10 15:51:42 +00:00
if err := tc.watcherRepo.IncrementHits(tc, watcher.ID); err != nil {
2022-05-23 18:17:25 +00:00
tc.logger.Error("could not increment hits",
zap.Error(err),
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
zap.Int64("watcher#id", watcher.ID),
)
2021-10-10 15:51:42 +00:00
return
}
notification.DeviceToken = watcher.Device.APNSToken
client := tc.apnsProduction
if watcher.Device.Sandbox {
client = tc.apnsSandbox
}
res, err := client.Push(notification)
2022-01-14 20:29:56 +00:00
if err != nil || !res.Sent() {
2021-10-10 15:51:42 +00:00
_ = tc.statsd.Incr("apns.notification.errors", []string{}, 1)
2022-05-23 18:17:25 +00:00
tc.logger.Error("failed to send notification",
zap.Error(err),
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
zap.String("post#id", post.ID),
zap.String("apns", watcher.Device.APNSToken),
zap.Int("response#status", res.StatusCode),
zap.String("response#reason", res.Reason),
)
2021-10-10 15:51:42 +00:00
} else {
_ = tc.statsd.Incr("apns.notification.sent", []string{}, 1)
2022-05-23 18:17:25 +00:00
tc.logger.Info("sent notification",
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
zap.String("post#id", post.ID),
2022-05-25 20:05:49 +00:00
zap.Int64("post#score", post.Score),
2022-05-23 18:17:25 +00:00
zap.String("device#token", watcher.Device.APNSToken),
)
2021-10-10 15:51:42 +00:00
}
}
}
2022-05-23 18:17:25 +00:00
tc.logger.Debug("finishing job",
zap.Int64("subreddit#id", id),
zap.String("subreddit#name", subreddit.NormalizedName()),
)
2021-10-10 15:51:42 +00:00
}
func payloadFromTrendingPost(post *reddit.Thing) *payload.Payload {
title := fmt.Sprintf(trendingNotificationTitleFormat, post.Subreddit)
payload := payload.
NewPayload().
AlertTitle(title).
AlertBody(post.Title).
AlertSummaryArg(post.Subreddit).
Category("trending-post").
2021-10-10 15:51:42 +00:00
Custom("post_title", post.Title).
Custom("post_id", post.ID).
Custom("subreddit", post.Subreddit).
Custom("author", post.Author).
Custom("post_age", post.CreatedAt).
2022-05-01 17:36:35 +00:00
ThreadID("trending-post").
2021-10-10 15:51:42 +00:00
MutableContent().
Sound("traloop.wav")
if post.Thumbnail != "" && !post.Over18 {
2022-05-01 17:33:09 +00:00
payload.Custom("thumbnail", post.Thumbnail)
}
2021-10-10 15:51:42 +00:00
return payload
}