Notification Scheduler Worker

The notification scheduler runs every minute via cron trigger, queries the database for due notification deliveries, and enqueues them to the notification-deliveries queue for the notification-executor to process.

Source: workers/notification-scheduler/

Worker Name: podcasterplus-notification-scheduler

Architecture

Processing Flow

1. Claim Due Deliveries

The scheduler queries notification_deliveries for records that are ready to process:

SELECT ... FROM notification_deliveries
WHERE (
    status IN ('pending', 'scheduled') AND scheduled_for <= NOW()
) OR (
    status = 'failed' AND next_retry_at <= NOW() AND attempts < max_attempts
)
ORDER BY
    CASE priority WHEN 'immediate' THEN 0 WHEN 'normal' THEN 1 WHEN 'low' THEN 2 END,
    scheduled_for ASC
FOR UPDATE SKIP LOCKED
LIMIT {page_size}

Key details:

• FOR UPDATE SKIP LOCKED prevents race conditions between concurrent scheduler runs
• Priority ordering ensures immediate notifications are processed first
• Pagination processes in pages of 50 (configurable) up to 200 total per run
• Status update to queued with queued_at timestamp upon claim

2. Enqueue Messages

Each claimed delivery is converted to a queue message:

interface NotificationDeliveryQueueMessage {
  version: 1;
  type: 'notification_delivery.execute';
  deliveryId: string;
  notificationId: string;
  channel: 'email' | 'in_app' | 'push';
  priority: 'low' | 'normal' | 'immediate';
  reason: 'due' | 'retry';
  queuedAt: string;
}

If enqueue fails, the delivery is released back to its previous status to be retried on the next cron run.

3. Cleanup

After processing deliveries, the scheduler runs cleanup tasks:

Task	Retention	Batch Size
Expired tracking tokens	30 days	100
Old dead letters	90 days	100

Configuration

Wrangler Bindings

name = "podcasterplus-notification-scheduler"
compatibility_date = "2024-12-30"
compatibility_flags = ["nodejs_compat"]

[triggers]
crons = ["* * * * *"]

[[hyperdrive]]
binding = "HYPERDRIVE"
id = "a81d477ff9264805989f5a72f0354ee8"

[[queues.producers]]
queue = "notification-deliveries"
binding = "NOTIFICATION_DELIVERY_QUEUE"

Environment Variables

Variable	Default	Purpose
SCHEDULER_PAGE_SIZE	50	Records claimed per database page
SCHEDULER_MAX_RECORDS_PER_RUN	200	Maximum records per cron invocation
NOTIFICATION_CLEANUP_BATCH_SIZE	100	Cleanup operation batch size
NOTIFICATION_DEAD_LETTER_RETENTION_DAYS	90	Dead letter retention period
NOTIFICATION_TRACKING_TOKEN_RETENTION_DAYS	30	Tracking token retention period
SUPABASE_SECRET_KEY	--	Required for manual trigger auth

Endpoints

Method	Path	Auth	Purpose
GET	/health, /_health	None	Health check
POST	/_trigger	Bearer (SUPABASE_SECRET_KEY)	Manual trigger

Manual Trigger

For debugging or catch-up processing:

curl -X POST https://worker-url/_trigger \
  -H "Authorization: Bearer $SUPABASE_SECRET_KEY"

Returns JSON with claim/enqueue counts.

Observability

[observability.logs]
enabled = true
head_sampling_rate = 1
invocation_logs = true
persist = true

Structured Logging

{
  "event": "notification_scheduler_run_completed",
  "claimed": 45,
  "enqueued": 45,
  "pages": 1,
  "cleanup": { "tokens_deleted": 12, "dead_letters_deleted": 0 },
  "duration_ms": 230
}

Deployment

cd workers/notification-scheduler
npx wrangler deploy

Secrets

npx wrangler secret put SUPABASE_SECRET_KEY

Related Documentation

• Notification System -- System overview
• Delivery Pipeline -- Full pipeline architecture
• Notification Executor -- Queue consumer worker
• Workers Overview -- All workers