apollo-backend/internal/worker/live_activities.go

338 lines
8.9 KiB
Go
Raw Normal View History

2022-10-19 13:37:41 +00:00
package worker
import (
"context"
"encoding/json"
"fmt"
"os"
"sort"
"time"
"github.com/DataDog/datadog-go/statsd"
2022-11-01 23:02:25 +00:00
"github.com/adjust/rmq/v5"
2022-10-19 13:37:41 +00:00
"github.com/go-redis/redis/v8"
"github.com/jackc/pgx/v4/pgxpool"
"github.com/sideshow/apns2"
"github.com/sideshow/apns2/token"
"go.opentelemetry.io/otel/trace"
2022-10-19 13:37:41 +00:00
"go.uber.org/zap"
"github.com/christianselig/apollo-backend/internal/domain"
"github.com/christianselig/apollo-backend/internal/reddit"
"github.com/christianselig/apollo-backend/internal/repository"
)
type DynamicIslandNotification struct {
PostCommentCount int `json:"postTotalComments"`
PostScore int64 `json:"postScore"`
CommentID string `json:"commentId,omitempty"`
CommentAuthor string `json:"commentAuthor,omitempty"`
CommentBody string `json:"commentBody,omitempty"`
CommentAge int64 `json:"commentAge,omitempty"`
CommentScore int64 `json:"commentScore,omitempty"`
2022-10-19 13:37:41 +00:00
}
type liveActivitiesWorker struct {
context.Context
logger *zap.Logger
tracer trace.Tracer
2022-10-19 13:37:41 +00:00
statsd *statsd.Client
db *pgxpool.Pool
redis *redis.Client
queue rmq.Connection
reddit *reddit.Client
apns *token.Token
consumers int
liveActivityRepo domain.LiveActivityRepository
}
func NewLiveActivitiesWorker(ctx context.Context, logger *zap.Logger, tracer trace.Tracer, statsd *statsd.Client, db *pgxpool.Pool, redis *redis.Client, queue rmq.Connection, consumers int) Worker {
2022-10-19 13:37:41 +00:00
reddit := reddit.NewClient(
os.Getenv("REDDIT_CLIENT_ID"),
os.Getenv("REDDIT_CLIENT_SECRET"),
tracer,
2022-10-19 13:37:41 +00:00
statsd,
redis,
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 &liveActivitiesWorker{
ctx,
logger,
tracer,
2022-10-19 13:37:41 +00:00
statsd,
db,
redis,
queue,
reddit,
apns,
consumers,
repository.NewPostgresLiveActivity(db),
}
}
func (law *liveActivitiesWorker) Start() error {
queue, err := law.queue.OpenQueue("live-activities")
if err != nil {
return err
}
law.logger.Info("starting up live activities worker", zap.Int("consumers", law.consumers))
prefetchLimit := int64(law.consumers * 4)
if err := queue.StartConsuming(prefetchLimit, pollDuration); err != nil {
return err
}
host, _ := os.Hostname()
for i := 0; i < law.consumers; i++ {
name := fmt.Sprintf("consumer %s-%d", host, i)
consumer := NewLiveActivitiesConsumer(law, i)
if _, err := queue.AddConsumer(name, consumer); err != nil {
return err
}
}
return nil
}
func (law *liveActivitiesWorker) Stop() {
<-law.queue.StopAllConsuming() // wait for all Consume() calls to finish
}
type liveActivitiesConsumer struct {
*liveActivitiesWorker
tag int
apns *apns2.Client
2022-10-19 13:37:41 +00:00
}
func NewLiveActivitiesConsumer(law *liveActivitiesWorker, tag int) *liveActivitiesConsumer {
return &liveActivitiesConsumer{
law,
tag,
apns2.NewTokenClient(law.apns).Production(),
}
}
func (lac *liveActivitiesConsumer) Consume(delivery rmq.Delivery) {
2022-10-27 00:43:44 +00:00
ctx, cancel := context.WithCancel(lac)
defer cancel()
2022-10-27 19:47:26 +00:00
now := time.Now()
2022-10-19 13:37:41 +00:00
defer func() {
elapsed := time.Now().Sub(now).Milliseconds()
_ = lac.statsd.Histogram("apollo.consumer.runtime", float64(elapsed), []string{"queue:live_activities"}, 0.1)
}()
at := delivery.Payload()
key := fmt.Sprintf("locks:live-activities:%s", at)
// Measure queue latency
2022-10-27 00:43:44 +00:00
ttl := lac.redis.PTTL(ctx, key).Val()
2022-10-19 13:37:41 +00:00
age := (domain.NotificationCheckTimeout - ttl)
_ = lac.statsd.Histogram("apollo.dequeue.latency", float64(age.Milliseconds()), []string{"queue:live_activities"}, 0.1)
defer func() {
2022-10-27 00:43:44 +00:00
if err := lac.redis.Del(ctx, key).Err(); err != nil {
2022-10-19 13:37:41 +00:00
lac.logger.Error("failed to remove account lock", zap.Error(err), zap.String("key", key))
}
}()
lac.logger.Debug("starting job", zap.String("live_activity#apns_token", at))
defer func() {
if err := delivery.Ack(); err != nil {
lac.logger.Error("failed to acknowledge message", zap.Error(err), zap.String("live_activity#apns_token", at))
}
}()
2022-10-27 00:43:44 +00:00
la, err := lac.liveActivityRepo.Get(ctx, at)
2022-10-19 13:37:41 +00:00
if err != nil {
lac.logger.Error("failed to get live activity", zap.Error(err), zap.String("live_activity#apns_token", at))
return
}
rac := lac.reddit.NewAuthenticatedClient(la.RedditAccountID, la.RefreshToken, la.AccessToken)
if la.TokenExpiresAt.Before(now.Add(5 * time.Minute)) {
lac.logger.Debug("refreshing reddit token",
zap.String("live_activity#apns_token", at),
2022-10-26 22:28:11 +00:00
zap.String("reddit#id", la.RedditAccountID),
zap.String("reddit#access_token", rac.ObfuscatedAccessToken()),
zap.String("reddit#refresh_token", rac.ObfuscatedRefreshToken()),
2022-10-19 13:37:41 +00:00
)
2022-10-27 00:43:44 +00:00
tokens, err := rac.RefreshTokens(ctx)
2022-10-19 13:37:41 +00:00
if err != nil {
2022-10-26 22:28:11 +00:00
lac.logger.Error("failed to refresh reddit tokens",
zap.Error(err),
zap.String("live_activity#apns_token", at),
zap.String("reddit#id", la.RedditAccountID),
zap.String("reddit#access_token", rac.ObfuscatedAccessToken()),
zap.String("reddit#refresh_token", rac.ObfuscatedRefreshToken()),
)
if err == reddit.ErrOauthRevoked {
2022-10-27 00:43:44 +00:00
_ = lac.liveActivityRepo.Delete(ctx, at)
}
2022-10-19 13:37:41 +00:00
return
}
// Update account
la.AccessToken = tokens.AccessToken
la.RefreshToken = tokens.RefreshToken
la.TokenExpiresAt = now.Add(tokens.Expiry)
2022-10-27 00:43:44 +00:00
_ = lac.liveActivityRepo.Update(ctx, &la)
2022-10-19 13:37:41 +00:00
// Refresh client
rac = lac.reddit.NewAuthenticatedClient(la.RedditAccountID, tokens.RefreshToken, tokens.AccessToken)
}
lac.logger.Debug("fetching latest comments", zap.String("live_activity#apns_token", at))
2022-10-27 00:43:44 +00:00
tr, err := rac.TopLevelComments(ctx, la.Subreddit, la.ThreadID)
2022-10-19 13:37:41 +00:00
if err != nil {
lac.logger.Error("failed to fetch latest comments",
zap.Error(err),
zap.String("live_activity#apns_token", at),
2022-10-26 22:28:11 +00:00
zap.String("reddit#id", la.RedditAccountID),
zap.String("reddit#access_token", rac.ObfuscatedAccessToken()),
zap.String("reddit#refresh_token", rac.ObfuscatedRefreshToken()),
2022-10-19 13:37:41 +00:00
)
if err == reddit.ErrOauthRevoked {
2022-10-27 00:43:44 +00:00
_ = lac.liveActivityRepo.Delete(ctx, at)
}
2022-10-19 13:37:41 +00:00
return
}
if len(tr.Children) == 0 && la.ExpiresAt.After(now) {
2022-10-19 13:37:41 +00:00
lac.logger.Debug("no comments found", zap.String("live_activity#apns_token", at))
return
}
// Filter out comments in the last minute
candidates := make([]*reddit.Thing, 0)
2022-10-26 23:31:16 +00:00
cutoffs := []time.Time{
now.Add(-domain.LiveActivityCheckInterval),
now.Add(-domain.LiveActivityCheckInterval * 2),
now.Add(-domain.LiveActivityCheckInterval * 4),
}
for _, cutoff := range cutoffs {
for _, t := range tr.Children {
if t.CreatedAt.After(cutoff) {
candidates = append(candidates, t)
}
}
if len(candidates) > 0 {
break
2022-10-19 13:37:41 +00:00
}
}
if len(candidates) == 0 && la.ExpiresAt.After(now) {
2022-10-19 13:37:41 +00:00
lac.logger.Debug("no new comments found", zap.String("live_activity#apns_token", at))
return
}
sort.Slice(candidates, func(i, j int) bool {
return candidates[i].Score > candidates[j].Score
})
din := DynamicIslandNotification{
PostCommentCount: tr.Post.NumComments,
PostScore: tr.Post.Score,
}
if len(candidates) > 0 {
comment := candidates[0]
din.CommentID = comment.ID
din.CommentAuthor = comment.Author
din.CommentBody = comment.Body
din.CommentAge = comment.CreatedAt.Unix()
din.CommentScore = comment.Score
2022-10-19 13:37:41 +00:00
}
ev := "update"
if la.ExpiresAt.Before(now) {
ev = "end"
}
bb, _ := json.Marshal(map[string]interface{}{
2022-10-19 13:37:41 +00:00
"aps": map[string]interface{}{
"content-state": din,
"dismissal-date": la.ExpiresAt.Unix(),
"event": ev,
"timestamp": now.Unix(),
2022-10-19 13:37:41 +00:00
},
})
2022-10-19 13:37:41 +00:00
notification := &apns2.Notification{
DeviceToken: la.APNSToken,
Topic: "com.christianselig.Apollo.push-type.liveactivity",
PushType: "liveactivity",
Payload: bb,
}
res, err := lac.apns.PushWithContext(ctx, notification)
2022-10-19 13:37:41 +00:00
if err != nil {
_ = lac.statsd.Incr("apns.live_activities.errors", []string{}, 1)
lac.logger.Error("failed to send notification",
zap.Error(err),
zap.String("live_activity#apns_token", at),
2022-10-27 16:27:19 +00:00
zap.Bool("live_activity#sandbox", la.Sandbox),
zap.String("notification#type", ev),
2022-10-19 13:37:41 +00:00
)
2022-10-27 00:43:44 +00:00
_ = lac.liveActivityRepo.Delete(ctx, at)
2022-10-19 13:37:41 +00:00
} else if !res.Sent() {
_ = lac.statsd.Incr("apns.live_activities.errors", []string{}, 1)
lac.logger.Error("notification not sent",
zap.String("live_activity#apns_token", at),
2022-10-27 16:27:19 +00:00
zap.Bool("live_activity#sandbox", la.Sandbox),
zap.String("notification#type", ev),
2022-10-19 13:37:41 +00:00
zap.Int("response#status", res.StatusCode),
zap.String("response#reason", res.Reason),
)
2022-10-27 00:43:44 +00:00
_ = lac.liveActivityRepo.Delete(ctx, at)
2022-10-19 13:37:41 +00:00
} else {
_ = lac.statsd.Incr("apns.notification.sent", []string{}, 1)
2022-11-01 14:16:17 +00:00
lac.logger.Debug("sent notification",
2022-10-19 13:37:41 +00:00
zap.String("live_activity#apns_token", at),
2022-10-27 16:27:19 +00:00
zap.Bool("live_activity#sandbox", la.Sandbox),
zap.String("notification#type", ev),
2022-10-19 13:37:41 +00:00
)
}
if la.ExpiresAt.Before(now) {
lac.logger.Debug("live activity expired, deleting", zap.String("live_activity#apns_token", at))
2022-10-27 00:43:44 +00:00
_ = lac.liveActivityRepo.Delete(ctx, at)
2022-10-19 13:37:41 +00:00
}
lac.logger.Debug("finishing job",
zap.String("live_activity#apns_token", at),
)
}