Scheduled Publisher Worker

The Scheduled Publisher is a Cloudflare Worker that automatically publishes episodes at their scheduled time. It runs every minute via cron, finding episodes ready for publication and triggering downstream automation events.

Source: workers/scheduled-publisher/

Overview

The publisher handles the critical transition from "scheduled" to "published" status, ensuring episodes go live at the exact time specified by podcast hosts.

Configuration

wrangler.toml

name = "podcasterplus-scheduled-publisher"
main = "src/index.ts"
compatibility_date = "2024-12-30"
compatibility_flags = ["nodejs_compat"]

# Cron trigger - runs every minute
[triggers]
crons = ["* * * * *"]

# Hyperdrive for database connection pooling
[[hyperdrive]]
binding = "HYPERDRIVE"
id = "a81d477ff9264805989f5a72f0354ee8"

# Queue producer for RSS cache invalidation
[[queues.producers]]
queue = "rss-invalidation"
binding = "RSS_INVALIDATION_QUEUE"

Environment Variables

Variable	Type	Description
HYPERDRIVE	Binding	Hyperdrive connection to Supabase
RSS_INVALIDATION_QUEUE	Binding	Queue producer for RSS cache invalidation
SUPABASE_SECRET_KEY	Secret	Service role key for database access
RSS_INVALIDATION_SECRET	Secret	Auth token for HTTP fallback invalidation
PUBLIC_SUPABASE_URL	String	Supabase project URL

Required Secrets

Set via wrangler secret put:

wrangler secret put SUPABASE_SECRET_KEY
wrangler secret put RSS_INVALIDATION_SECRET

Entry Points

Cron Handler (scheduled)

Triggered every minute by the cron schedule. Finds and publishes due episodes.

async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext): Promise<void>

Process:

1. Query episodes where status = 'scheduled' AND scheduled_for <= NOW() AND audio_url IS NOT NULL
2. Update each episode to status = 'published' with published_at = scheduled_for
3. Queue RSS cache invalidation for the podcast
4. Emit episode.published automation event for matching rules
5. Log results with metrics

HTTP Handler (fetch)

Provides health checks and manual trigger endpoint.

Endpoint	Method	Description
/health	GET	Returns { status: 'ok', service: 'scheduled-publisher' }
/_health	GET	Alias for health check
/_trigger	POST	Manual trigger for scheduled publishing (requires auth)

Manual Trigger Authentication:

curl -X POST https://podcasterplus-scheduled-publisher.<account>.workers.dev/_trigger \
  -H "Authorization: Bearer ${SUPABASE_SECRET_KEY}"

Core Functions

publishScheduledEpisodes(env)

Main orchestrator that finds and publishes all due episodes.

async function publishScheduledEpisodes(
    env: Env
): Promise<{ published: number; errors: number }>

Query for Due Episodes:

SELECT e.id, e.podcast_id, e.title, e.scheduled_for, p.slug
FROM episodes e
INNER JOIN podcasts p ON e.podcast_id = p.id
WHERE e.status = 'scheduled'
    AND e.scheduled_for <= NOW()
    AND e.audio_url IS NOT NULL
ORDER BY e.scheduled_for ASC
LIMIT 100

Safety Check: Episodes without audio_url are skipped to prevent publishing empty episodes.

Returns:

• published: Number of episodes successfully published
• errors: Number of episodes that failed to process

publishEpisode(sql, episode)

Publishes a single episode with idempotent update.

async function publishEpisode(
    sql: ReturnType<typeof postgres>,
    episode: DueEpisode
): Promise<boolean>

Update Query (Idempotent):

UPDATE episodes
SET status = 'published',
    published_at = ${episode.scheduled_for},
    updated_at = NOW()
WHERE id = ${episode.id}
    AND status = 'scheduled'  -- Double-check prevents race conditions
RETURNING id

queueInvalidation(env, podcastSlug, podcastId, episodeId)

Queues RSS cache invalidation with HTTP fallback.

async function queueInvalidation(
    env: Env,
    podcastSlug: string,
    podcastId: string,
    episodeId: string
): Promise<void>

Primary: Queue message to rss-invalidation

await env.RSS_INVALIDATION_QUEUE.send({
    type: 'episode.published',
    podcast_id: podcastId,
    podcast_slug: podcastSlug,
    episode_id: episodeId,
    timestamp: new Date().toISOString()
});

Fallback: HTTP POST to RSS Worker

await fetch('https://rss.cdn.media/_internal/invalidate', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${env.RSS_INVALIDATION_SECRET}`
    },
    body: JSON.stringify({ slug: podcastSlug, reason: 'episode.published' })
});

emitEpisodePublishedEvent(sql, episode)

Creates automation executions for rules triggered by episode publication.

async function emitEpisodePublishedEvent(
    sql: ReturnType<typeof postgres>,
    episode: PublishedEpisode
): Promise<void>

Query for Matching Rules:

SELECT id, name
FROM automation_rules
WHERE podcast_id = ${episode.podcast_id}
    AND trigger_type = 'episode.published'
    AND is_enabled = true
    AND is_paused = false

Create Execution Records (Idempotent):

INSERT INTO automation_executions (
    id, rule_id, trigger_type, podcast_id, episode_id,
    idempotency_key, context_data, status, created_at
)
VALUES (...)
ON CONFLICT (idempotency_key) DO NOTHING

Idempotency Key Format: {rule_id}::episode.published::{podcast_id}::{episode_id}::{date}

Execution Flow

Logging

All logs are JSON-formatted for structured querying:

// Cron start
{ event: 'publisher_cron_started', scheduled_time: '2025-01-13T12:00:00Z' }

// Episodes found
{ event: 'due_episodes_found', count: 5 }

// Episode published
{ event: 'episode_published', episode_id: '...', podcast_slug: 'show-name', title: 'Episode 1' }

// RSS invalidation queued
{ event: 'rss_invalidation_queued', podcast_slug: 'show-name', method: 'queue' }

// Automation events created
{ event: 'automation_events_created', episode_id: '...', rules_matched: 3 }

// Cron complete
{ event: 'publisher_cron_completed', published: 5, errors: 0, duration_ms: 245 }

// Errors
{ event: 'episode_publish_failed', episode_id: '...', error: 'Error message' }

Error Handling

Failure Isolation

Each episode is processed independently. A failure for one episode does not block others:

for (const episode of dueEpisodes) {
    try {
        await publishEpisode(sql, episode);
        await queueInvalidation(env, ...);
        await emitEpisodePublishedEvent(sql, episode);
        published++;
    } catch (error) {
        console.error(JSON.stringify({ event: 'episode_publish_failed', episode_id: episode.id, error: String(error) }));
        errors++;
    }
}

Error Scenarios

Scenario	Handling
Episode already published (race)	Idempotent update returns no rows, silently succeeds
Database connection error	Episode skipped, error logged
Queue unavailable	Falls back to HTTP invalidation
HTTP fallback fails	Error logged, RSS cache stale until next invalidation
Automation rule query fails	Error logged, automations not triggered

Deployment

Deploy Worker

cd workers/scheduled-publisher
npx wrangler deploy

View Logs

npx wrangler tail podcasterplus-scheduled-publisher

Manual Testing

# Trigger scheduled publishing manually
curl -X POST https://podcasterplus-scheduled-publisher.<account>.workers.dev/_trigger \
  -H "Authorization: Bearer ${SUPABASE_SECRET_KEY}"

Performance

Processing Limits

• Batch Size: Up to 100 episodes per cron invocation
• Cron Frequency: Every minute
• Max Throughput: ~6,000 episodes per hour

Database Connections

• Single connection per invocation (max: 1)
• Connection pooling via Hyperdrive
• Connections properly closed in finally block

Query Performance

• Episodes table indexed on (status, scheduled_for)
• Automation rules indexed on (podcast_id, trigger_type)
• Idempotency keys have unique constraint for fast duplicate detection

Monitoring

Key Metrics to Track

1. Episodes Published: Count per hour/day
2. Error Rate: Failed vs successful publications
3. Latency: Time from scheduled_for to actual published_at
4. Queue Backlog: Due episodes waiting > 1 minute

Alerts to Configure

• Cron invocation failures
• Error rate > 1%
• Unpublished scheduled episodes older than 5 minutes
• Queue message failures

Related Documentation

• RSS Feed Worker - Cache invalidation consumer
• Automation Scheduler - Event processing
• Automation Executor - Action execution
• Cloudflare Services - Infrastructure overview