show.fm Docs (internal)

Appearance

Podcast Import Executor ​

Queue-consumer Worker that drives two independent flows behind one deployment:

  1. podcast-imports consumer — per-episode pipeline for both back-catalogue imports (is_hosting_migration = false, create new episodes) and hosting migrations (is_hosting_migration = true, merge audio into existing episodes by external_guid, then cut over hosting_type once the parent import lands completed).
  2. external-episode-link consumer (Epic 11) — lightweight feed-fetch + DB-write loop that resolves a newly-published external episode to its matching RSS <guid> by pubDate proximity (with title-similarity disambiguation), or marks it unmatched after exhausting the backoff schedule.

Worker name: podcasterplus-podcast-import-executorTrigger: Queue consumer on podcast-imports AND external-episode-linkSource: workers/podcast-import-executor/

Message routing ​

src/index.ts discriminates on message.body.type so both consumers share one worker. Cloudflare delivers each queue's messages to this worker; the entry handler dispatches by type:

typeQueueHandler
import_itempodcast-importsprocessItem (back-catalogue or hosting-migration depending on podcast_imports.is_hosting_migration)
attempt_linkexternal-episode-linkprocessLinkAttempt

Unknown types are logged (unknown_message_type) and ack'd so a future schema mismatch cannot DLQ-flood the worker.

podcast-imports consumer ​

Every terminal branch calls finaliseIfLastItem, which increments counters and — only when the last outstanding item closes — fires the side effects. Finalisation is wrapped in a swallow-all try/catch so a transient DB or Resend failure never rolls back the item's terminal state.

import_item message schema ​

typescript
interface PodcastImportMessage {
  type: 'import_item';
  import_id: string;        // podcast_imports.id
  import_item_id: string;   // import_items.id
  podcast_id: string;
}

Produced by POST /api/imports/podcast, /api/imports/:id/confirm, /:id/resume, /:id/retry-failed, /:id/items/:itemId/retry, and POST /api/podcasts/:id/upgrade-hosting/commit. See Imports API and External Link API.

Hosting-migration merge mode (Epic 11) ​

When podcast_imports.is_hosting_migration = true, processItem first probes episodes by (podcast_id, external_guid = payload.guid). If a match is found, it branches into processMergeMode:

  1. Stream the source enclosure into podcasts/{podcastId}/e/{existingEpisodeId}/audio/migrated.{ext} (R2 key keyed on the existing episode id so per-podcast cleanup still finds it on deletion).
  2. Mirror the source episode artwork into cover-migrated.{ext} only if the existing episode has no cover_image_url (fill-if-missing).
  3. Call mergeAudioIntoExistingEpisode (workers/podcast-import-executor/src/db.ts) to:
    • Unconditionally rewrite audio_url, audio_content_type, audio_file_size_bytes, audio_duration_seconds, episode_guid, external_rss_pub_date, published_at, is_explicit.
    • Conditionally rewrite title / description per migration_config.use_rss_titles / use_rss_descriptions.
    • Fill-if-missing: cover_image_url, episode_number, season_number.
    • show_notes and show_note_sections are never touched.

When the GUID does not match any existing episode, the migration falls through to the standard create path — but the new episode is inserted with external_guid = payload.guid so the migration's bookkeeping is complete and subscriber-facing <guid> continuity is preserved.

Hosting-migration cutover (last-item side effect) ​

sendSideEffects (called from finaliseIfLastItem when the parent enters a terminal state) handles the cutover:

  • Only on outcome === 'completed' — a partial migration would publish an incomplete feed to subscribers; the user retries failed items, and once they land the parent re-enters completed and this block fires again.
  • Calls flipHostingTypeToInternal(podcastId) to flip podcasts.hosting_type from external to podcasterplus. The UPDATE is guarded by WHERE hosting_type = 'external' so it is idempotent on retry.
  • On cutover failure: flips the import to failed with a specific last_error so the wizard surfaces the issue, and skips the post-completion notifications (RSS invalidation is moot — hosting is still external).
  • On cutover success: emits one rss-invalidation message (type: 'import.completed') and one completion email.

Queue configuration — podcast-imports ​

toml
[[queues.consumers]]
queue = "podcast-imports"
max_batch_size = 1
max_batch_timeout = 5
max_retries = 5
dead_letter_queue = "podcast-imports-dlq"
max_concurrency = 4
  • max_batch_size = 1 — each message is a long-running stream (median ~5 s wall-time). Batching would block the whole batch on the slowest download.
  • max_concurrency = 4 — keeps the per-host request rate polite for small podcast hosts.
  • max_retries = 5 — on transient failure the item goes back on the queue with exponential-backoff jitter; final attempt terminalises.

Retry & terminalisation ​

The entry handler wraps processItem in a try/catch. Error classification is explicit:

Error classBehaviour
FatalImportErrormarkItemFailed(reason, detail) inside processItem; finalise.
RetryableImportErrorThrow; entry handler calls message.retry({ delaySeconds }).
Unclassified throwRelease the item (resetItemForRetry) and bubble so the queue retries.

On the final attempt (message.attempts >= 5) any remaining retry is converted to a terminalise call so the row never stays stuck:

typescript
// workers/podcast-import-executor/src/index.ts
if (onFinalAttempt || isFatal) {
  await terminaliseItem(env, message.body.import_item_id, reason, errMessage);
  message.ack();
}

Backoff formula:

typescript
base = min(900, 5 * 4^(attempts - 1))
jitter = random(0, min(30, base / 4))
delaySeconds = floor(base + jitter)

Caps at 900 s (Cloudflare Queues per-message delay limit).

Per-item pipeline (create path) ​

src/processItem.ts handles one message. Key guarantees:

  1. Cancel-safe. If the parent import's status is canceled or preview at claim time, the message is ack'd and the item is left at pending. Resume / commit-activation re-dispatches it.
  2. Exactly-once episode insert. claimedInsertAndComplete re-verifies the processing claim under a row lock before touching episodes, so a stale worker whose claim was rescued by the lifecycle-manager watchdog cannot durably create an episode that a fresh worker's terminal write would then orphan. On ON CONFLICT (parallel insert of same GUID), the helper marks the item as a race duplicate and links it to the winning episode_id in the same transaction.
  3. All assets mirrored. Cover image, chapters JSON, and every <podcast:transcript> the feed advertises are copied to R2 via mirrorAssetBestEffort (failure is logged and ignored — the episode still inserts). The preferred transcript (VTT > SRT > JSON > first) populates episodes.transcript_url.
  4. Status from pubDate. Future-dated items become scheduled; past or undated items become published. Slug is derived from title + stable 6-char hash of the GUID.

R2 keys written ​

Create path:

podcasts/{podcastId}/e/{episodeId}/audio/original.{mp3|m4a|aac|wav|ogg|webm}
podcasts/{podcastId}/e/{episodeId}/cover.{jpg|png|webp|gif}
podcasts/{podcastId}/e/{episodeId}/chapters.json
podcasts/{podcastId}/e/{episodeId}/transcript.{vtt|srt|txt|json}
podcasts/{podcastId}/e/{episodeId}/transcript.1.{…}   // 2nd format, 3rd, …

Merge path (hosting migration):

podcasts/{podcastId}/e/{existingEpisodeId}/audio/migrated.{mp3|m4a|aac|wav|ogg|webm}
podcasts/{podcastId}/e/{existingEpisodeId}/cover-migrated.{jpg|png|webp|gif}  # fill-if-missing only

Both paths live under podcasts/{podcastId}/e/{episodeId}/... so the lifecycle-manager's per-podcast R2 cleanup removes migrated assets when a podcast is deleted.

Finalisation side effects ​

When finaliseIfLastItem detects the last outstanding row, sendSideEffects fires:

  1. Hosting migration cutover — only when is_hosting_migration and outcome === 'completed'. Flips hosting_type to podcasterplus; on failure, marks the import failed with a specific last_error and short-circuits the rest of the block.
  2. RSS invalidation — one message to the rss-invalidation queue:
    typescript
    {
  type: 'import.completed',
  podcast_id: string,
  podcast_slug: string,
  timestamp: string
}
    Consumed by the RSS Feed Worker which deletes the KV cache entry for the slug.
  3. Completion email via Resend, with per-outcome copy (completed, partial, failed) including counts of published vs scheduled episodes.

RSS invalidation and email are best-effort — failures are logged and swallowed to protect the item's already-written terminal state.

typescript
interface ExternalEpisodeLinkMessage {
  type: 'attempt_link';
  episode_id: string;
  podcast_id: string;
  target_published_at: string;   // ISO 8601
  attempt: number;               // starts at 0, incremented by consumer on retry
}

Producers (see src/lib/external-link/enqueue.ts):

  • SvelteKit publish-handoff actions (initial enqueue on confirmPublished / confirmScheduled).
  • POST /api/episodes/:id/external-link/reattempt (manual re-attempt).
  • workers/lifecycle-manager (stalled-link watchdog re-enqueue).
  • This worker (consumer self-retry on backoff).

Backoff schedule ​

Driven by Cloudflare Queues' native delaySeconds; the worker uses a fresh queue.send for self-retries rather than message.retry() so the domain backoff is not constrained by max_retries = 2.

attemptDelay
05 minutes
115 minutes
21 hour
36 hours
424 hours
5+(schedule exhausted → mark unmatched)

Shared between worker and main app via the parallel definitions in workers/podcast-import-executor/src/external-link/types.ts and src/lib/external-link/queue-message.ts.

Disambiguation rules ​

  • ±24 h pubDate window around the user-entered handoff time (PROXIMITY_WINDOW_MS). Wide enough to absorb host timezone shifts and platforms that backdate pubDate to upload time.
  • Already-linked GUIDs excluded from the candidate set so an item already linked to another PP episode in the same podcast cannot win.
  • Title similarity (Sørensen–Dice bigram, src/external-link/title-similarity.ts): when >1 candidate falls in the pubDate window, the picker requires top_score ≥ 0.8 AND top_score − runner_up ≥ 0.15 to declare a unique winner. Otherwise: retry or mark unmatched.
  • Tenant guard: refuses to act if the message's podcast_id does not match the episode's actual podcast_id (defence-in-depth against a future producer writing one tenant's RSS identity onto another tenant's row).
toml
[[queues.consumers]]
queue = "external-episode-link"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 2
dead_letter_queue = "external-episode-link-dlq"
max_concurrency = 4

[[queues.producers]]
queue = "external-episode-link"
binding = "EXTERNAL_EPISODE_LINK_QUEUE"
  • Higher batch + concurrency than the import consumer because each message is a single feed fetch + single DB write, not a long-running audio stream.
  • max_retries = 2 keeps Cloudflare's automatic redelivery short — the domain backoff (5m → 24h) lives entirely in the consumer.

DB writes ​

The consumer writes via direct Postgres through Hyperdrive (not PostgREST), so auth.role() is NULL. The episode-update trigger added in 20260508093849_harden_external_link_authorization.sql bypasses on auth.role() = 'service_role' OR auth.role() IS NULL, leaving worker writes unblocked while PostgREST anon callers still hit the has_podcast_role gate.

Bindings ​

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

[[r2_buckets]]
binding = "MEDIA_BUCKET"
bucket_name = "podcasterplus-media"

[[queues.producers]]
queue = "rss-invalidation"
binding = "RSS_INVALIDATION_QUEUE"

[[queues.producers]]
queue = "external-episode-link"
binding = "EXTERNAL_EPISODE_LINK_QUEUE"

[vars]
PUBLIC_APP_URL = "https://app.podcasterplus.com"
PUBLIC_MEDIA_URL = "https://media.podcasterplus.com"
IMPORTER_USER_AGENT = "show.fm-importer/1.0 (+https://show.fm)"
PER_FILE_MAX_BYTES = "524288000"     # 500 MB cap on audio stream

Secrets: RESEND_API_KEY, RESEND_FROM_EMAIL (set via wrangler secret put).

Failure modes ​

podcast-imports consumer ​

ConditionClassificationEffect
Parsed payload missing audioUrlFatalImportErrorItem → failed, reason parse_error.
Source HEAD / GET returns >= 400VariesFatal for 404/410, retryable for 5xx.
Audio exceeds PER_FILE_MAX_BYTESFatalImportErrorItem → failed, reason skipped_oversize.
Duplicate GUID detected at processing time(not an error)Item → skipped_duplicate.
Hosting-migration cutover throws(terminal-import write)Import → failed with Hosting cutover failed: …; no RSS invalidation, no email. User re-runs migration to retry (merge is idempotent).
Unclassified throwresetItemForRetry + rethrowQueue retry with backoff.
All retries exhaustedTerminaliseItem → failed, reason retries_exhausted.
Finalisation throws after terminal writeLogged, swallowedNo state rollback.
Queue message lost in transit (never delivered)Watchdog re-dispatchPer-minute stalled-import sweep in lifecycle-manager detects import_items rows past their per-attempt threshold and re-sends the queue message. Safe because claimPendingItem is atomic — a late-arriving original ack-skips.
ConditionOutcome
Episode no longer pendingidempotent_ack
Tenant mismatch (podcast_id ≠ episode's)idempotent_ack + structured error log
Podcast missingMark unmatched
Podcast hosting_type no longer externalidempotent_ack (cutover happened)
Podcast missing external_rss_urlMark unmatched
Feed fetch / parse failureretry (treats as host not yet propagated)
Malformed target_published_atMark unmatched
0 or >1 candidates with no title winnerretry until schedule exhausted, then unmatched
Exactly 1 candidate, or unique title winnerlinked (external_link_method = 'auto')

Observability ​

Structured JSON logs with a leading event field. Filter by event in wrangler tail.

podcast-imports consumer ​

text
import_item_ack                       # item successfully processed
import_item_retry                     # transient failure, scheduling retry
import_item_terminalised              # final attempt failed, terminalised
import_item_threw                     # unexpected throw; released back to pending
import_item_reset_failed              # couldn't release item (DB error)
import_item_skipped_non_running_import # parent canceled/preview
item_already_claimed                  # race lost to another invocation
import_item_late_duplicate            # late ON CONFLICT match on episode_guid
import_item_claim_lost                # claim token mismatched at terminal write
import_gone                           # orphaned message; ack
finalise_if_last_item_failed          # logged, swallowed
terminalise_failed                    # catastrophic; logged
rss_invalidation_send_failed          # queue send threw; logged
side_effects_skipped                  # missing podcast/importer refs
hosting_migration_cutover             # hosting_type flipped to internal
hosting_migration_cutover_failed      # cutover UPDATE threw; import marked failed
hosting_migration_mark_failed_after_cutover_failed # last-ditch fallback
text
external_link_attempt                       # outcome=linked|retry|unmatched|idempotent_ack|tenant_mismatch|podcast_internal_skip
external_link_attempt_error                 # unexpected throw; Cloudflare retry then DLQ
external_link_title_disambiguation          # outcome=matched|still_ambiguous, top_score, runner_up_score

Enqueue-side (emitted from src/api/routes/imports/index.ts::enqueueImportItems) ​

text
import_items_enqueued                 # sent=N, expected=M — diff N≠M signals send failure
import_items_enqueue_failed           # loop aborted mid-way; includes sent count and error

Watchdog-side (emitted from workers/lifecycle-manager/src/index.ts, cron * * * * *) ​

text
import_watchdog_completed             # combined: items_requeued, links_requeued, previews_canceled
import_watchdog_found_stalled         # per-import summary
import_watchdog_resend_failed         # per-item send failure
external_link_watchdog_resent         # per-tick count
hosting_migration_preview_watchdog_canceled # per-tick count

See the RSS Import cost baseline for a known observability gap — not all per-phase events currently surface through wrangler tail, which is a remediation item rather than a cost concern.

Deployment ​

bash
cd workers/podcast-import-executor
npx wrangler deploy

# Set secrets
npx wrangler secret put RESEND_API_KEY
npx wrangler secret put RESEND_FROM_EMAIL

# Tail logs
npx wrangler tail podcasterplus-podcast-import-executor --format json

Internal documentation - Not for public distribution

show.fm Team