diff --git a/Dockerfile b/Dockerfile index 57796a57..ad23aa99 100644 --- a/Dockerfile +++ b/Dockerfile @@ -9,7 +9,8 @@ RUN go mod download COPY . . ARG TARGETARCH -RUN CGO_ENABLED=0 GOOS=linux GOARCH=$TARGETARCH go build -o bridge main.go +RUN CGO_ENABLED=0 GOOS=linux GOARCH=$TARGETARCH go build -o bridge main.go && \ + CGO_ENABLED=0 GOOS=linux GOARCH=$TARGETARCH go build -o remix-contest-notifications ./cmd/remix_contest_notifications/ FROM node:20-alpine AS plans-builder @@ -28,6 +29,7 @@ RUN apk add --no-cache bash postgresql-client WORKDIR /app COPY --from=builder /app/bridge /bin/bridge +COPY --from=builder /app/remix-contest-notifications /bin/remix-contest-notifications COPY --from=builder /app/ddl ./ddl COPY --from=builder /app/static/swagger-ui ./static/swagger-ui COPY --from=plans-builder /app/static/plans/dist ./static/plans/dist diff --git a/cmd/remix_contest_notifications/main.go b/cmd/remix_contest_notifications/main.go new file mode 100644 index 00000000..18ba755a --- /dev/null +++ b/cmd/remix_contest_notifications/main.go @@ -0,0 +1,37 @@ +package main + +import ( + "context" + "fmt" + "os" + + "api.audius.co/config" + "api.audius.co/jobs" + "github.com/jackc/pgx/v5/pgxpool" +) + +// remix-contest-notifications runs RemixContestNotificationsJob once and exits. +// Deploy as a Kubernetes CronJob on a 15-20 minute schedule; the job is +// idempotent via ON CONFLICT + NOT EXISTS guards so overlapping runs are safe. +// +// Required env vars (same as the main bridge): +// writeDbUrl - postgres connection string +// ENV - "dev", "stage", or "prod" +func main() { + cfg := config.Cfg + ctx := context.Background() + + pool, err := pgxpool.New(ctx, cfg.WriteDbUrl) + if err != nil { + fmt.Fprintf(os.Stderr, "remix-contest-notifications: db connect failed: %v\n", err) + os.Exit(1) + } + defer pool.Close() + + job := jobs.NewRemixContestNotificationsJob(cfg, pool) + if err := job.RunE(ctx); err != nil { + // Individual step errors are already logged by the job via zap; + // a non-zero exit tells the CronJob controller to record a failure. + os.Exit(1) + } +} diff --git a/indexer/indexer.go b/indexer/indexer.go index 4e458296..38938c64 100644 --- a/indexer/indexer.go +++ b/indexer/indexer.go @@ -204,9 +204,6 @@ func (ci *CoreIndexer) startParityJobs(ctx context.Context) { jobs.NewListenStreakReminderJob(ci.Config, ci.pool). ScheduleEvery(ctx, 10*time.Second) - jobs.NewRemixContestNotificationsJob(ci.Config, ci.pool). - ScheduleEvery(ctx, 30*time.Second) - // Backfill missing track bpm / musical_key from content-node audio // analyses. Mirrors apps' repair_audio_analyses celery task, whose beat // schedule ran every 3 minutes. Needs the SDK for content-node discovery. diff --git a/jobs/create_listen_streak_reminder_notifications.go b/jobs/create_listen_streak_reminder_notifications.go index b9b47d0f..daf93a5c 100644 --- a/jobs/create_listen_streak_reminder_notifications.go +++ b/jobs/create_listen_streak_reminder_notifications.go @@ -106,7 +106,11 @@ func (j *ListenStreakReminderJob) run(ctx context.Context) error { @now FROM challenge_listen_streak cls WHERE cls.last_listen_date BETWEEN @window_start AND @window_end - ON CONFLICT (group_id, specifier) DO NOTHING + AND NOT EXISTS ( + SELECT 1 FROM notification + WHERE group_id = 'listen_streak_reminder:' || cls.user_id::text || ':' || to_char(cls.last_listen_date, 'YYYY-MM-DD') + ) +ON CONFLICT (group_id, specifier) DO NOTHING `, pgx.NamedArgs{ "now": now, "window_start": windowStart, diff --git a/jobs/create_remix_contest_notifications.go b/jobs/create_remix_contest_notifications.go index ee4bce97..c574b81a 100644 --- a/jobs/create_remix_contest_notifications.go +++ b/jobs/create_remix_contest_notifications.go @@ -23,6 +23,12 @@ import ( // current/visible, and insert one notification per recipient. uq_notification // (group_id, specifier) — where specifier is the recipient user id — provides // idempotency, so a recipient is notified at most once per (event, type). +// +// Each step also filters out events that already have any notification row for +// their group_id prefix. This prevents Postgres from burning a sequence ID on +// every conflict when the job re-scans the same events each run — without this +// guard, a 72-hour ending-soon window and a 30-second job interval causes +// ~8 640 wasted INSERT attempts per event per day. type RemixContestNotificationsJob struct { pool database.DbPool logger *zap.Logger @@ -71,6 +77,12 @@ func (j *RemixContestNotificationsJob) Run(ctx context.Context) { } } +// RunE executes the job once and returns any error. +// Intended for the cmd/remix_contest_notifications CronJob binary. +func (j *RemixContestNotificationsJob) RunE(ctx context.Context) error { + return j.run(ctx) +} + func (j *RemixContestNotificationsJob) run(ctx context.Context) error { start := time.Now() j.mutex.Lock() @@ -166,7 +178,11 @@ func (j *RemixContestNotificationsJob) fanEnded(ctx context.Context, now time.Ti AND e.end_date IS NOT NULL AND e.end_date BETWEEN @window_start AND @window_end AND aud.user_id <> e.user_id - ON CONFLICT (group_id, specifier) DO NOTHING + AND NOT EXISTS ( + SELECT 1 FROM notification + WHERE group_id = 'fan_remix_contest_ended:' || e.event_id::text + ) +ON CONFLICT (group_id, specifier) DO NOTHING `, pgx.NamedArgs{ "now": now, "window_start": now.Add(-remixContestEndedWindowHours), @@ -210,7 +226,11 @@ func (j *RemixContestNotificationsJob) fanEndingSoon(ctx context.Context, now ti AND e.end_date IS NOT NULL AND e.end_date BETWEEN @window_start AND @window_end AND aud.user_id <> e.user_id - ON CONFLICT (group_id, specifier) DO NOTHING + AND NOT EXISTS ( + SELECT 1 FROM notification + WHERE group_id = 'fan_remix_contest_ending_soon:' || e.event_id::text + ) +ON CONFLICT (group_id, specifier) DO NOTHING `, pgx.NamedArgs{ "now": now, "window_start": now, @@ -240,7 +260,11 @@ func (j *RemixContestNotificationsJob) artistEnded(ctx context.Context, now time WHERE e.event_type = 'remix_contest' AND NOT e.is_deleted AND e.end_date IS NOT NULL AND e.end_date BETWEEN @window_start AND @window_end - ON CONFLICT (group_id, specifier) DO NOTHING + AND NOT EXISTS ( + SELECT 1 FROM notification + WHERE group_id = 'artist_remix_contest_ended:' || e.event_id::text + ) +ON CONFLICT (group_id, specifier) DO NOTHING `, pgx.NamedArgs{ "now": now, "window_start": now.Add(-remixContestEndedWindowHours), @@ -270,7 +294,11 @@ func (j *RemixContestNotificationsJob) artistEndingSoon(ctx context.Context, now WHERE e.event_type = 'remix_contest' AND NOT e.is_deleted AND e.end_date IS NOT NULL AND e.end_date BETWEEN @window_start AND @window_end - ON CONFLICT (group_id, specifier) DO NOTHING + AND NOT EXISTS ( + SELECT 1 FROM notification + WHERE group_id = 'artist_remix_contest_ending_soon:' || e.event_id::text + ) +ON CONFLICT (group_id, specifier) DO NOTHING `, pgx.NamedArgs{ "now": now, "window_start": now, diff --git a/k8s/remix-contest-notifications-cronjob.yaml b/k8s/remix-contest-notifications-cronjob.yaml new file mode 100644 index 00000000..61795e44 --- /dev/null +++ b/k8s/remix-contest-notifications-cronjob.yaml @@ -0,0 +1,47 @@ +# Kubernetes CronJob for remix-contest-notifications. +# +# This job emits four time-based remix-contest notifications: +# fan_remix_contest_ended / fan_remix_contest_ending_soon +# artist_remix_contest_ended / artist_remix_contest_ending_soon +# +# Previously this ran as a 30-second goroutine inside the bridge process, +# which caused Postgres sequence burn (~18M IDs/day) because every INSERT +# attempt against an ON CONFLICT target consumes a sequence ID before the +# conflict check fires. Running it as a CronJob at 20-minute intervals +# cuts the scan frequency from 2880x/day to 72x/day per event window. +# +# The NOT EXISTS guard added in each INSERT provides belt-and-suspenders +# idempotency, so concurrent or overlapping runs are safe. +# +apiVersion: batch/v1 +kind: CronJob +metadata: + name: remix-contest-notifications + namespace: api +spec: + schedule: "*/20 * * * *" # every 20 minutes + concurrencyPolicy: Forbid # skip run if previous pod is still active + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 3 + jobTemplate: + spec: + backoffLimit: 2 + template: + spec: + restartPolicy: OnFailure + containers: + - name: remix-contest-notifications + image: audius/api@sha256:b1bd03d8d3afa7f074a633be135b994be9ec8c8b2ada47f1f225c1e0bfa8efc7 + command: ["/bin/remix-contest-notifications"] + # Mount the same env vars as the main bridge Deployment. + # Minimum required: writeDbUrl, ENV. + envFrom: + - secretRef: + name: bridge-secret-279d7978 + resources: + requests: + cpu: 100m + memory: 128Mi + limits: + cpu: 500m + memory: 256Mi