fix: add NOT EXISTS guard to stop notification_id_seq burn on repeated job runs

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

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)
+	}
+}

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.

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,

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,

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