Skip to content

Commit

Permalink
Changes to schema
Browse files Browse the repository at this point in the history
  • Loading branch information
@andremedeiros
andremedeiros committed Mar 28, 2022
1 parent 59e435e commit dbcda74
Show file tree
Hide file tree
Showing 20 changed files with 226 additions and 192 deletions.
9 changes: 6 additions & 3 deletions internal/api/accounts.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ func (a *api) upsertAccountsHandler(w http.ResponseWriter, r *http.Request) {
}

// Reset expiration timer
acc.ExpiresAt = time.Now().Unix() + 3540
acc.TokenExpiresAt = time.Now().Add(1 * time.Hour)
acc.RefreshToken = tokens.RefreshToken
acc.AccessToken = tokens.AccessToken

Expand All @@ -175,7 +175,10 @@ func (a *api) upsertAccountsHandler(w http.ResponseWriter, r *http.Request) {
return
}

_ = a.accountRepo.Associate(ctx, &acc, &dev)
if err := a.accountRepo.Associate(ctx, &acc, &dev); err != nil {
a.errorResponse(w, r, 422, err.Error())
return
}
}

for _, acc := range accsMap {
Expand Down Expand Up @@ -212,7 +215,7 @@ func (a *api) upsertAccountHandler(w http.ResponseWriter, r *http.Request) {
}

// Reset expiration timer
acct.ExpiresAt = time.Now().Unix() + 3540
acct.TokenExpiresAt = time.Now().Add(1 * time.Hour)
acct.RefreshToken = tokens.RefreshToken
acct.AccessToken = tokens.AccessToken

Expand Down
4 changes: 2 additions & 2 deletions internal/api/devices.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@ func (a *api) upsertDeviceHandler(w http.ResponseWriter, r *http.Request) {
return
}

d.ActiveUntil = time.Now().Unix() + domain.DeviceGracePeriodDuration
d.GracePeriodUntil = d.ActiveUntil + domain.DeviceGracePeriodAfterReceiptExpiry
d.ExpiresAt = time.Now().Add(domain.DeviceReceiptCheckPeriodDuration)
d.GracePeriodExpiresAt = d.ExpiresAt.Add(domain.DeviceGracePeriodAfterReceiptExpiry)

if err := a.deviceRepo.CreateOrUpdate(ctx, d); err != nil {
a.errorResponse(w, r, 500, err.Error())
Expand Down
9 changes: 7 additions & 2 deletions internal/api/receipt.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,11 @@ func (a *api) checkReceiptHandler(w http.ResponseWriter, r *http.Request) {
}

if iapr.DeleteDevice {
if dev.GracePeriodExpiresAt.After(time.Now()) {
w.WriteHeader(http.StatusOK)
return
}

accs, err := a.accountRepo.GetByAPNSToken(ctx, apns)
if err != nil {
a.errorResponse(w, r, 500, err.Error())
Expand All @@ -51,8 +56,8 @@ func (a *api) checkReceiptHandler(w http.ResponseWriter, r *http.Request) {

_ = a.deviceRepo.Delete(ctx, apns)
} else {
dev.ActiveUntil = time.Now().Unix() + domain.DeviceActiveAfterReceitCheckDuration
dev.GracePeriodUntil = dev.ActiveUntil + domain.DeviceGracePeriodAfterReceiptExpiry
dev.ExpiresAt = time.Now().Add(domain.DeviceActiveAfterReceitCheckDuration)
dev.GracePeriodExpiresAt = dev.ExpiresAt.Add(domain.DeviceGracePeriodAfterReceiptExpiry)
_ = a.deviceRepo.Update(ctx, &dev)
}
}
Expand Down
23 changes: 12 additions & 11 deletions internal/api/watcher.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"net/http"
"strconv"
"strings"
"time"

validation "github.com/go-ozzo/ozzo-validation/v4"
"github.com/gorilla/mux"
Expand Down Expand Up @@ -230,17 +231,17 @@ func (a *api) deleteWatcherHandler(w http.ResponseWriter, r *http.Request) {
}

type watcherItem struct {
ID int64 `json:"id"`
CreatedAt float64 `json:"created_at"`
Type string `json:"type"`
Label string `json:"label"`
SourceLabel string `json:"source_label"`
Upvotes *int64 `json:"upvotes,omitempty"`
Keyword string `json:"keyword,omitempty"`
Flair string `json:"flair,omitempty"`
Domain string `json:"domain,omitempty"`
Hits int64 `json:"hits"`
Author string `json:"author,omitempty"`
ID int64 `json:"id"`
CreatedAt time.Time `json:"created_at"`
Type string `json:"type"`
Label string `json:"label"`
SourceLabel string `json:"source_label"`
Upvotes *int64 `json:"upvotes,omitempty"`
Keyword string `json:"keyword,omitempty"`
Flair string `json:"flair,omitempty"`
Domain string `json:"domain,omitempty"`
Hits int64 `json:"hits"`
Author string `json:"author,omitempty"`
}

func (a *api) listWatchersHandler(w http.ResponseWriter, r *http.Request) {
Expand Down
104 changes: 47 additions & 57 deletions internal/cmd/scheduler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,28 +9,20 @@ import (

"github.com/DataDog/datadog-go/statsd"
"github.com/adjust/rmq/v4"
"github.com/christianselig/apollo-backend/internal/cmdutil"
"github.com/christianselig/apollo-backend/internal/repository"
"github.com/go-co-op/gocron"
"github.com/go-redis/redis/v8"
"github.com/jackc/pgx/v4"
"github.com/jackc/pgx/v4/pgxpool"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
)

const (
batchSize = 250
checkTimeout = 60 // how long until we force a check

accountEnqueueInterval = 5 // how frequently we want to check (seconds)
subredditEnqueueInterval = 2 * 60 // how frequently we want to check (seconds)
userEnqueueInterval = 2 * 60 // how frequently we want to check (seconds)
stuckAccountEnqueueInterval = 2 * 60 // how frequently we want to check (seconds)

staleAccountThreshold = 7200 // 2 hours
"github.com/christianselig/apollo-backend/internal/cmdutil"
"github.com/christianselig/apollo-backend/internal/domain"
"github.com/christianselig/apollo-backend/internal/repository"
)

const batchSize = 250

func SchedulerCmd(ctx context.Context) *cobra.Command {
cmd := &cobra.Command{
Use: "scheduler",
Expand Down Expand Up @@ -129,16 +121,16 @@ func evalScript(ctx context.Context, redis *redis.Client) (string, error) {
end
return retv
`, checkTimeout)
`, int64(domain.NotificationCheckTimeout.Seconds()))

return redis.ScriptLoad(ctx, lua).Result()
}

func pruneAccounts(ctx context.Context, logger *logrus.Logger, pool *pgxpool.Pool) {
before := time.Now().Unix() - staleAccountThreshold
expiry := time.Now().Add(-domain.StaleTokenThreshold)
ar := repository.NewPostgresAccount(pool)

stale, err := ar.PruneStale(ctx, before)
stale, err := ar.PruneStale(ctx, expiry)
if err != nil {
logger.WithFields(logrus.Fields{
"err": err,
Expand All @@ -158,16 +150,17 @@ func pruneAccounts(ctx context.Context, logger *logrus.Logger, pool *pgxpool.Poo

if count > 0 {
logger.WithFields(logrus.Fields{
"count": count,
"stale": stale,
"orphaned": orphaned,
}).Info("pruned accounts")
}
}

func pruneDevices(ctx context.Context, logger *logrus.Logger, pool *pgxpool.Pool) {
threshold := time.Now().Unix()
now := time.Now()
dr := repository.NewPostgresDevice(pool)

count, err := dr.PruneStale(ctx, threshold)
count, err := dr.PruneStale(ctx, now)
if err != nil {
logger.WithFields(logrus.Fields{
"err": err,
Expand Down Expand Up @@ -227,6 +220,8 @@ func reportStats(ctx context.Context, logger *logrus.Logger, statsd *statsd.Clie

func enqueueUsers(ctx context.Context, logger *logrus.Logger, statsd *statsd.Client, pool *pgxpool.Pool, queue rmq.Queue) {
now := time.Now()
next := now.Add(domain.NotificationCheckInterval)

ids := []int64{}

defer func() {
Expand All @@ -235,21 +230,20 @@ func enqueueUsers(ctx context.Context, logger *logrus.Logger, statsd *statsd.Cli
_ = statsd.Histogram("apollo.queue.runtime", float64(time.Since(now).Milliseconds()), tags, 1)
}()

ready := now.Unix() - userEnqueueInterval
err := pool.BeginFunc(ctx, func(tx pgx.Tx) error {
stmt := `
WITH userb AS (
WITH batch AS (
SELECT id
FROM users
WHERE last_checked_at < $1
ORDER BY last_checked_at
WHERE next_check_at < $1
ORDER BY next_check_at
LIMIT 100
)
UPDATE users
SET last_checked_at = $2
WHERE users.id IN(SELECT id FROM userb)
SET next_check_at = $2
WHERE users.id IN(SELECT id FROM batch)
RETURNING users.id`
rows, err := tx.Query(ctx, stmt, ready, now.Unix())
rows, err := tx.Query(ctx, stmt, now, next)
if err != nil {
return err
}
Expand All @@ -275,7 +269,7 @@ func enqueueUsers(ctx context.Context, logger *logrus.Logger, statsd *statsd.Cli

logger.WithFields(logrus.Fields{
"count": len(ids),
"start": ready,
"start": now,
}).Debug("enqueueing user batch")

batchIds := make([]string, len(ids))
Expand All @@ -292,6 +286,8 @@ func enqueueUsers(ctx context.Context, logger *logrus.Logger, statsd *statsd.Cli

func enqueueSubreddits(ctx context.Context, logger *logrus.Logger, statsd *statsd.Client, pool *pgxpool.Pool, queues []rmq.Queue) {
now := time.Now()
next := now.Add(domain.SubredditCheckInterval)

ids := []int64{}

defer func() {
Expand All @@ -300,21 +296,20 @@ func enqueueSubreddits(ctx context.Context, logger *logrus.Logger, statsd *stats
_ = statsd.Histogram("apollo.queue.runtime", float64(time.Since(now).Milliseconds()), tags, 1)
}()

ready := now.Unix() - subredditEnqueueInterval
err := pool.BeginFunc(ctx, func(tx pgx.Tx) error {
stmt := `
WITH subreddit AS (
WITH batch AS (
SELECT id
FROM subreddits
WHERE last_checked_at < $1
ORDER BY last_checked_at
WHERE next_check_at < $1
ORDER BY next_check_at
LIMIT 100
)
UPDATE subreddits
SET last_checked_at = $2
WHERE subreddits.id IN(SELECT id FROM subreddit)
SET next_check_at = $2
WHERE subreddits.id IN(SELECT id FROM batch)
RETURNING subreddits.id`
rows, err := tx.Query(ctx, stmt, ready, now.Unix())
rows, err := tx.Query(ctx, stmt, now, next)
if err != nil {
return err
}
Expand All @@ -340,7 +335,7 @@ func enqueueSubreddits(ctx context.Context, logger *logrus.Logger, statsd *stats

logger.WithFields(logrus.Fields{
"count": len(ids),
"start": ready,
"start": now,
}).Debug("enqueueing subreddit batch")

batchIds := make([]string, len(ids))
Expand All @@ -361,6 +356,8 @@ func enqueueSubreddits(ctx context.Context, logger *logrus.Logger, statsd *stats

func enqueueStuckAccounts(ctx context.Context, logger *logrus.Logger, statsd *statsd.Client, pool *pgxpool.Pool, queue rmq.Queue) {
now := time.Now()
next := now.Add(domain.StuckNotificationCheckInterval)

ids := []int64{}

defer func() {
Expand All @@ -369,22 +366,21 @@ func enqueueStuckAccounts(ctx context.Context, logger *logrus.Logger, statsd *st
_ = statsd.Histogram("apollo.queue.runtime", float64(time.Since(now).Milliseconds()), tags, 1)
}()

ready := now.Unix() - stuckAccountEnqueueInterval
err := pool.BeginFunc(ctx, func(tx pgx.Tx) error {
stmt := `
WITH account AS (
WITH batch AS (
SELECT id
FROM accounts
WHERE
last_unstuck_at < $1
ORDER BY last_unstuck_at
next_stuck_notification_check_at < $1
ORDER BY next_stuck_notification_check_at
LIMIT 500
)
UPDATE accounts
SET last_unstuck_at = $2
WHERE accounts.id IN(SELECT id FROM account)
SET next_stuck_notification_check_at = $2
WHERE accounts.id IN(SELECT id FROM batch)
RETURNING accounts.id`
rows, err := tx.Query(ctx, stmt, ready, now.Unix())
rows, err := tx.Query(ctx, stmt, now, next)
if err != nil {
return err
}
Expand All @@ -410,7 +406,7 @@ func enqueueStuckAccounts(ctx context.Context, logger *logrus.Logger, statsd *st

logger.WithFields(logrus.Fields{
"count": len(ids),
"start": ready,
"start": now,
}).Debug("enqueueing stuck account batch")

batchIds := make([]string, len(ids))
Expand All @@ -428,6 +424,8 @@ func enqueueStuckAccounts(ctx context.Context, logger *logrus.Logger, statsd *st

func enqueueAccounts(ctx context.Context, logger *logrus.Logger, statsd *statsd.Client, pool *pgxpool.Pool, redisConn *redis.Client, luaSha string, queue rmq.Queue) {
now := time.Now()
next := now.Add(domain.NotificationCheckInterval)

ids := []int64{}
enqueued := 0
skipped := 0
Expand All @@ -439,29 +437,21 @@ func enqueueAccounts(ctx context.Context, logger *logrus.Logger, statsd *statsd.
_ = statsd.Histogram("apollo.queue.runtime", float64(time.Since(now).Milliseconds()), tags, 1)
}()

// Start looking for accounts that were last checked at least 5 seconds ago
// and at most 6 seconds ago. Also look for accounts that haven't been checked
// in over a minute.
ts := now.Unix()
ready := ts - accountEnqueueInterval
expired := ts - checkTimeout

err := pool.BeginFunc(ctx, func(tx pgx.Tx) error {
stmt := `
WITH account AS (
SELECT id
FROM accounts
WHERE
last_enqueued_at < $1
OR last_checked_at < $2
ORDER BY last_checked_at
next_notification_check_at < $1
ORDER BY next_notification_check_at
LIMIT 2500
)
UPDATE accounts
SET last_enqueued_at = $3
SET next_notification_check_at = $2
WHERE accounts.id IN(SELECT id FROM account)
RETURNING accounts.id`
rows, err := tx.Query(ctx, stmt, ready, expired, ts)
rows, err := tx.Query(ctx, stmt, now, next)
if err != nil {
return err
}
Expand All @@ -487,7 +477,7 @@ func enqueueAccounts(ctx context.Context, logger *logrus.Logger, statsd *statsd.

logger.WithFields(logrus.Fields{
"count": len(ids),
"start": ready,
"start": now,
}).Debug("enqueueing account batch")
// Split ids in batches
for i := 0; i < len(ids); i += batchSize {
Expand Down Expand Up @@ -532,7 +522,7 @@ func enqueueAccounts(ctx context.Context, logger *logrus.Logger, statsd *statsd.
logger.WithFields(logrus.Fields{
"count": enqueued,
"skipped": skipped,
"start": ready,
"start": now,
}).Debug("done enqueueing account batch")
}

Expand Down
Loading

0 comments on commit dbcda74

Please sign in to comment.