Automation Executor Worker

The Automation Executor is a Cloudflare Worker that consumes execution messages from the queue and performs automation actions like sending emails, making webhook calls, and updating database fields.

Source: workers/automation-executor/

Overview

The executor is a pure queue consumer that receives pre-validated execution messages and performs the configured actions.

Configuration

wrangler.toml

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

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

# Queue consumer for automation executions
[[queues.consumers]]
queue = "automation-executions"
max_batch_size = 25
max_batch_timeout = 30
max_retries = 5
dead_letter_queue = "automation-dlq"

# CPU time limit (cost protection against runaway execution)
[limits]
cpu_ms = 10000

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

Environment Variables

Variable	Type	Description
HYPERDRIVE	Binding	Hyperdrive connection to Supabase
RESEND_API_KEY	Secret	API key for Resend email service
RESEND_FROM_EMAIL	Secret	Sender email address
SUPABASE_SECRET_KEY	Secret	Service role key for database access
PUBLIC_APP_URL	String	Application URL (https://app.podcasterplus.com)
PUBLIC_BOOK_URL	String	Booking page URL (https://book.podcasterplus.com)
PUBLIC_SUPABASE_URL	String	Supabase project URL

Required Secrets

wrangler secret put SUPABASE_SECRET_KEY
wrangler secret put RESEND_API_KEY
wrangler secret put RESEND_FROM_EMAIL

Entry Points

Queue Handler (queue)

Processes batches of execution messages from the queue.

async queue(batch: MessageBatch<AutomationExecutionMessage>, env: Env): Promise<void>

Message Format:

interface AutomationExecutionMessage {
    type: 'execute_rule';
    rule_id: string;
    execution_id: string;
    payload: {
        trigger_type: AutomationTriggerType;
        podcast_id: string;
        episode_id?: string;
        booking_id?: string;
        guest_id?: string;
        context: MagicTagContext;
        timestamp: string;
    };
    attempt: number;
    scheduled_at: string;
}

HTTP Handler (fetch)

Health check endpoint only.

Endpoint	Method	Description
/health	GET	Returns { status: 'ok', service: 'automation-executor' }
/_health	GET	Alias for health check

Core Processing

processExecution(env, message)

Main execution orchestrator.

async function processExecution(
    env: Env,
    message: AutomationExecutionMessage
): Promise<void>

Process Flow:

1. Idempotency Check: Only process if status is pending or waiting
2. Atomic Claim: UPDATE ... SET status = 'processing' WHERE status IN ('pending', 'waiting') RETURNING id
3. Fetch Rule: Get rule and verify is_enabled and is_paused
4. Fetch Actions: Get ordered actions for the rule
5. Build Context: Query database for full magic tag context (including profile fallback)
6. Execute Actions: Run each action in order, starting from resume_from_index (for delay continuations)
7. Record Results: Update execution with action results and final status

Execution States:

Status	Description
pending	Created, waiting in queue
processing	Currently executing actions
completed	All actions succeeded
failed	One or more actions failed
waiting	Paused for a delay action, will resume later
skipped	Rule disabled/paused or no actions

Action Executors

Email Executor

Source: src/executors/email.ts

Sends emails via Resend API with magic tag replacement.

async function executeEmailAction(
    env: Env,
    sql: ReturnType<typeof postgres>,
    config: SendEmailConfig,
    context: MagicTagContext,
    podcastId: string,
    guestId?: string,
    bookingId?: string
): Promise<ActionExecutionResult>

Configuration:

interface SendEmailConfig {
    template_id: string;      // notification_templates.id
    to: 'guest' | 'host' | 'podcast_owner' | 'custom';
    custom_email?: string;    // When to='custom', supports magic tags
    cc?: string[];
    bcc?: string[];
    reply_to?: string;
}

Process:

1. Validate Resend API configuration
2. Fetch template from database
3. Resolve recipient email:
   • guest: From guests table or context
   • host/podcast_owner: From podcasts + auth.users join
   • custom: Use custom_email with magic tag replacement
4. Replace magic tags in subject and body
5. Send via Resend API
6. Return result with email_id

Recipient Resolution:

To Value	Resolution
guest	Lookup episode_guests table by guest_id, fallback to context.guest_email
host / podcast_owner	Join podcasts → podcast_members (owner) → auth.users to get owner email
custom	Use custom_email field, magic tags replaced

Webhook Executor

Source: src/executors/webhook.ts

Makes HTTP requests to external URLs.

async function executeWebhookAction(
    env: Env,
    config: SendWebhookConfig,
    context: MagicTagContext
): Promise<ActionExecutionResult>

Configuration:

interface SendWebhookConfig {
    url: string;              // Supports magic tags
    method: 'GET' | 'POST' | 'PUT' | 'PATCH' | 'DELETE';
    headers?: Record<string, string>;  // Supports magic tags in values
    payload_template?: string;  // JSON template with magic tags
    timeout_ms?: number;        // Default: 30000, max: 60000
}

Process:

1. Replace magic tags in URL
2. Validate URL format
3. Build headers with magic tag replacement
4. Build request body:
   • If payload_template: Parse as JSON with tags replaced
   • If invalid JSON: Use full context as payload
   • Default: { event: 'automation_trigger', timestamp, context }
5. Execute request with timeout
6. Return response status and body

Default Headers:

{
    'Content-Type': 'application/json',
    'User-Agent': 'show.fm-automation/1.0'
}

Field Update Executor

Source: src/executors/field-update.ts

Updates database fields with security restrictions.

async function executeFieldUpdateAction(
    env: Env,
    sql: ReturnType<typeof postgres>,
    config: UpdateFieldConfig,
    context: MagicTagContext,
    podcastId: string,
    episodeId?: string,
    bookingId?: string,
    guestId?: string
): Promise<ActionExecutionResult>

Configuration:

interface UpdateFieldConfig {
    entity: 'episode' | 'booking' | 'guest';
    field: string;
    value: string | number | boolean;  // Supports magic tags in strings
}

Security: Allowed Fields:

Entity	Allowed Fields
episode	title, description, status, episode_number, season_number, scheduled_for, recording_scheduled_at, notes
booking	status, guest_notes, host_notes, meeting_url, meeting_platform
guest	name, bio, phone, notes

Process:

1. Validate entity type
2. Validate field against allowlist
3. Determine target ID from context
4. Replace magic tags in value (if string)
5. Execute parameterized UPDATE query
6. Return updated record info

Delay Executor

Source: src/index.ts (inline)

Pauses execution and schedules a continuation job.

async function executeDelayAction(
    env: Env,
    sql: ReturnType<typeof postgres>,
    action: AutomationActionRow,
    message: AutomationExecutionMessage,
    currentActionIndex: number
): Promise<ActionExecutionResult>

Configuration:

interface DelayConfig {
    delay_value: number;    // Amount of time
    delay_unit: 'minutes' | 'hours' | 'days';
}

Process:

1. Calculate scheduledFor from delay config
2. Generate idempotency key: delay:{execution_id}:{action_index}
3. Insert into automation_scheduled_jobs with ON CONFLICT DO NOTHING
4. Store resume_from_index on the execution record
5. Return completed status (main loop transitions to waiting)

The automation-scheduler worker picks up the job when it's due and re-queues the execution message. The executor then resumes from resume_from_index, skipping previously-executed actions.

Magic Tag Context Building

The executor builds a complete magic tag context by querying multiple database tables. This is the most important function in the worker, as it determines what data is available for tag replacement.

Query Sequence

Step Details

Step	Table Queried	Fields Populated	Notes
1	podcasts	podcast_name, podcast_host_name, podcast_description, podcast_website, podcast_listen_link (guarded — see below)	Always runs. Also captures slug for URL building
2	bookings + booking_links + booking_sessions	Guest name/email/phone, recording dates, booking_link_name, meeting_url, meeting_platform, calendar_link, booking_page_url	Resolves episode_id and guest_id if not provided
2b	bookings (by episode_id)	Same as step 2	Only runs if no booking_id but episode_id exists
3	episode_guests (by episode_id)	guest_first_name, guest_full_name, guest_email, guest_portal_link	Only runs if no guest resolved yet. Captures user_id
4	episodes	episode_title, episode_number, episode_description, episode_status, season_number, publish dates, episode_url	Runs if episode_id provided or resolved from booking
5	episode_guests (by guest_id)	Guest name/email, guest_portal_link	Captures user_id for profile lookup. Only overrides name if not already set from booking
6	user_profiles	guest_full_name (override), guest_first_name (override), guest_bio, guest_website, guest_twitter, guest_avatar	Profile Fallback - only runs if user_id was discovered
7	-	All undefined tags set to ""	Ensures no undefined values leak into templates

Profile Fallback (Step 6)

When a guest has linked their account (has a user_id in episode_guests pointing to user_profiles):

• display_name overrides guest_full_name and guest_first_name
• bio, website_url, twitter_handle, avatar_url populate profile-only tags
• Twitter handles are automatically @-prefixed if not already

If no profile exists, profile-only tags remain empty strings. See the Magic Tags Reference for the full resolution table.

Listen-link guard (podcast_listen_link)

podcast_listen_link is only populated when the listen page would actually resolve, so automation email never renders a link that 404s. The guard mirrors the listen page's full filter set (src/routes/(listen)/[podcastSlug]/+page.server.ts), all four conditions (workers/automation-executor/src/index.ts:436-442):

hosting_type = 'podcasterplus'
AND is_active = true
AND status IN ('active', 'paused')
AND overage_suspended_at IS NULL

• overage_suspended_at IS NULL is the hosted-content suspension gate (downgrade-overages §7.4, docs/planning/plans/2026-07-12-downgrade-overages.md): the stamp is platform-owned and independent of both is_active and status, so a suspended show's link is withheld even though the owner's Active switch is on.
• Checking status alone is not enough — is_active can diverge from status (the Active switch writes is_active directly and the sync trigger only fires ON UPDATE OF status).
• Any condition failing renders the tag as an empty string.
• The app-side context builder (src/lib/automation/tag-parser.ts:444-454, dev-mode sync path) applies the same guard — keep the two in lockstep with each other and with the listen page's filters, or suspended shows get emailed links to their own 404.

Timezone Resolution

While building the context, the executor also captures the timezone used to render date tags:

• If a booking is in scope, booking.timezone wins for all recording_* tags.
• Otherwise, podcasts.default_timezone is used for episode-derived tags.
• If neither is present, tags render against 'UTC'.

Formatting happens via Intl.DateTimeFormat({ timeZone }) inside tag-parser.ts. Cron ticks fire on UTC; anything that needs a wall-clock offset in a user's timezone must be converted upstream before the scheduler enqueues the job. See Timezone Management.

Logging

The context builder logs each query start/complete for observability:

{ "event": "query_start", "query": "user_profile", "user_id": "..." }
{ "event": "query_complete", "query": "user_profile", "found": true }
{ "event": "magic_tag_context_built", "podcast_id": "...", "user_id": "...", "has_profile": true, "populated_count": 25, "total_tags": 33 }

Action Results

Each action returns a standardized result:

interface ActionExecutionResult {
    node_id?: string;      // Workflow node ID
    action_type: string;   // 'send_email', 'send_webhook', 'update_field'
    status: 'completed' | 'failed' | 'skipped';
    started_at: string;    // ISO 8601
    completed_at: string;  // ISO 8601
    duration_ms?: number;
    result?: {
        // Action-specific success data
        email_id?: string;
        webhook_response?: { status: number; body: unknown };
        updated_record?: { entity: string; id: string; field: string };
    };
    error?: string;
    error_details?: Record<string, unknown>;
}

Magic Tag Replacement

All executors use the same replacement pattern. The context is pre-built by buildMagicTagContext() which includes profile fallback data when available.

function replaceMagicTags(content: string, context: MagicTagContext): string {
    return content.replace(/\{([a-zA-Z_][a-zA-Z0-9_]*)\}/g, (match, tagName) => {
        const value = context[tagName];
        if (value === undefined || value === null || value === '') {
            return '';  // Empty string for missing values
        }
        return String(value);
    });
}

Supported in:

• Email subject and body
• Webhook URL, headers, and payload template
• Field update string values
• Custom email recipient

Profile-aware tags: When {guest_bio}, {guest_website}, {guest_twitter}, or {guest_avatar} appear in templates, they resolve to profile data if the guest has an account, or empty strings if not. The {guest_full_name} and {guest_first_name} tags use the profile display_name when available, falling back to the booking/guest record name.

Error Handling

Retry Strategy

• Max Retries: 5 attempts (queue configuration)
• Retry Behavior: Automatic exponential backoff via queue
• Dead Letter Queue: After 5 failures, message goes to automation-dlq

Action Failure Handling

• Individual action failures don't stop the execution
• All actions are attempted regardless of previous failures
• Final status is failed if any action failed
• Detailed error information stored in action_results

Failure Scenarios

Scenario	Action	Result Status
Resend API error	Email	failed with HTTP status
Template not found	Email	failed
Recipient unresolvable	Email	failed
Webhook timeout	Webhook	failed
HTTP 4xx/5xx	Webhook	failed with response
Field not in allowlist	Field Update	failed
Entity not found	Field Update	failed
Rule disabled mid-execution	Skip remaining	skipped

Logging

JSON-formatted logs for structured analysis:

// Execution started
{ event: 'action_started', execution_id: '...', action_id: '...', action_type: 'send_email' }

// Action completed
{ event: 'action_completed', execution_id: '...', action_type: 'send_email', status: 'completed', duration_ms: 150 }

// Email sent
{ event: 'email_sent', template_id: '...', to: '[email protected]', email_id: 'resend-id' }

// Webhook sent
{ event: 'webhook_sent', url: 'https://...', method: 'POST', status: 200 }

// Field updated
{ event: 'field_updated', entity: 'episode', id: '...', field: 'status' }

// Errors
{ event: 'execution_failed', execution_id: '...', attempt: 3, error: 'Error message' }

Deployment

Deploy Worker

cd workers/automation-executor
npx wrangler deploy

View Logs

npx wrangler tail podcasterplus-automation-executor

Check Queue Status

npx wrangler queues list
npx wrangler queues describe automation-executions

Process Dead Letter Queue

# View DLQ messages
npx wrangler queues describe automation-dlq

# Reprocess failed messages (manual intervention required)

Testing

Local Development

cd workers/automation-executor
npx wrangler dev

Send Test Message

// Simulate queue message for testing
const testMessage: AutomationExecutionMessage = {
    type: 'execute_rule',
    rule_id: 'rule-uuid',
    execution_id: 'exec-uuid',
    payload: {
        trigger_type: 'booking.confirmed',
        podcast_id: 'podcast-uuid',
        episode_id: 'episode-uuid',
        booking_id: 'booking-uuid',
        guest_id: 'guest-uuid',
        context: {
            guest_first_name: 'John',
            guest_email: '[email protected]',
            episode_title: 'Test Episode'
        },
        timestamp: new Date().toISOString()
    },
    attempt: 1,
    scheduled_at: new Date().toISOString()
};

Performance

Batch Processing

• Batch Size: Up to 25 messages per invocation
• Batch Timeout: 30 seconds
• Sequential Processing: Actions within an execution run in order
• Parallel Batches: Multiple messages in a batch can be processed concurrently

Database Connections

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

External API Timeouts

• Resend: Default fetch timeout
• Webhooks: Configurable up to 60 seconds (default 30s)
• Database: Hyperdrive manages timeouts

Monitoring

Key Metrics to Track

1. Execution Duration: Total time per execution
2. Action Success Rate: Percentage of successful actions
3. Queue Depth: Messages waiting in automation-executions
4. DLQ Size: Failed messages requiring attention
5. Resend Delivery Rate: Email delivery success

Alerts to Configure

• DLQ message count > 0
• Execution failure rate > 5%
• Queue depth growing consistently
• Average execution time > 10s