Automation Scheduler Worker

The Automation Scheduler is a Cloudflare Worker that handles scheduling and triggering of automation rules. It processes both time-based triggers (cron) and event-based triggers (queue).

Source: workers/automation-scheduler/

Overview

The scheduler has two primary responsibilities:

1. Time-Based Processing: Runs every minute via cron to find and dispatch scheduled automation jobs
2. Event Processing: Consumes events from the automation-events queue and creates executions for matching rules

Configuration

wrangler.toml

name = "podcasterplus-automation-scheduler"
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 executor worker
[[queues.producers]]
queue = "automation-executions"
binding = "AUTOMATION_QUEUE"

# Queue consumer for event triggers
[[queues.consumers]]
queue = "automation-events"
max_batch_size = 50
max_batch_timeout = 30
max_retries = 3
dead_letter_queue = "automation-dlq"

Environment Variables

Variable	Type	Description
HYPERDRIVE	Binding	Hyperdrive connection to Supabase
AUTOMATION_QUEUE	Binding	Queue producer for executions
SUPABASE_SECRET_KEY	Secret	Service role key for database access
PUBLIC_APP_URL	String	Application URL
PUBLIC_SUPABASE_URL	String	Supabase project URL

Required Secrets

Set via wrangler secret put:

wrangler secret put SUPABASE_SECRET_KEY

Entry Points

Cron Handler (scheduled)

Triggered every minute by the cron schedule. Processes due time-based automations.

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

Process:

1. Query automation_scheduled_jobs for pending jobs where scheduled_for <= NOW()
2. Check idempotency to prevent duplicate executions
3. Build execution context from related entities
4. Create automation_executions record
5. Enqueue message to automation-executions queue
6. Update job status to processing

Queue Handler (queue)

Consumes event messages from the automation-events queue.

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

Event Message Format:

interface AutomationEventMessage {
    trigger_type: AutomationTriggerType;  // 'booking.confirmed', etc.
    podcast_id: string;
    episode_id?: string;
    booking_id?: string;
    guest_id?: string;
    timestamp: string;  // ISO 8601
}

Process:

1. Find enabled rules matching trigger_type and podcast_id
2. Evaluate trigger conditions (if any)
3. Generate idempotency key
4. Create automation_executions record
5. Enqueue message to automation-executions queue

HTTP Handler (fetch)

Provides health checks and manual trigger endpoint.

Endpoint	Method	Description
/health	GET	Returns { status: 'ok', service: 'automation-scheduler' }
/_health	GET	Alias for health check
/_trigger/time	POST	Manual trigger for time-based automations (requires auth)

Manual Trigger Authentication:

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

Core Functions

processTimeBasedAutomations(env)

Finds and processes all due scheduled jobs.

async function processTimeBasedAutomations(
    env: Env
): Promise<{ processed: number; skipped: number; errors: number }>

Query for Due Jobs:

SELECT sj.*
FROM automation_scheduled_jobs sj
INNER JOIN automation_rules ar ON sj.rule_id = ar.id
WHERE sj.status = 'pending'
    AND sj.scheduled_for <= NOW()
    AND ar.is_enabled = true
    AND ar.is_paused = false
ORDER BY sj.scheduled_for ASC
LIMIT 100

Returns:

• processed: Number of jobs successfully enqueued
• skipped: Number of jobs skipped (duplicates, missing rules)
• errors: Number of jobs that failed to process

processEventTrigger(env, event)

Processes a single event trigger and creates executions for matching rules.

async function processEventTrigger(
    env: Env,
    event: AutomationEventMessage
): Promise<void>

Query for Matching Rules:

SELECT *
FROM automation_rules
WHERE podcast_id = ${event.podcast_id}
    AND trigger_type = ${event.trigger_type}
    AND is_enabled = true
    AND is_paused = false

buildExecutionContext(sql, job)

Builds the magic tag context by fetching related entity data.

async function buildExecutionContext(
    sql: ReturnType<typeof postgres>,
    job: ScheduledJobRow
): Promise<Record<string, string | number | boolean | undefined>>

Entities Fetched:

• Guest: name, email, phone
• Episode: title, description, episode_number, status, dates
• Booking: start_time, timezone, meeting_url, meeting_platform, guest_notes
• Podcast: title, description, website_url, owner_name

evaluateConditions(conditions, context)

Evaluates trigger conditions against execution context.

function evaluateConditions(
    conditions: Array<{ field: string; operator: string; value?: string | number | boolean }>,
    context: Record<string, string | number | boolean | undefined>
): boolean

Supported Operators:

Operator	Description
equals	Exact match
not_equals	Not equal
contains	String contains
not_contains	String does not contain
is_empty	Value is null, undefined, or empty string
not_empty	Value has content
greater_than	Numeric greater than
less_than	Numeric less than

generateIdempotencyKey(...)

Creates a unique key to prevent duplicate executions.

function generateIdempotencyKey(
    ruleId: string,
    triggerType: string,
    episodeId?: string,
    bookingId?: string,
    guestId?: string,
    timestamp?: string
): string

Format: {rule_id}::{trigger_type}::{episode_id}::{booking_id}::{guest_id}::{date_or_timestamp}

Behavior:

• Time-based triggers: Uses date portion only (YYYY-MM-DD) to allow daily re-execution
• Event-based triggers: Uses full timestamp to prevent re-processing same event

Message Types

AutomationEventMessage (Input)

Received from automation-events queue:

interface AutomationEventMessage {
    trigger_type: AutomationTriggerType;
    podcast_id: string;
    episode_id?: string;
    booking_id?: string;
    guest_id?: string;
    timestamp: string;
}

AutomationExecutionMessage (Output)

Sent to automation-executions queue:

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: Record<string, string | number | boolean | undefined>;
        timestamp: string;
    };
    attempt: number;
    scheduled_at: string;
}

Logging

All logs are JSON-formatted for structured querying:

// Cron start
{ event: 'scheduler_cron_started', scheduled_time: '...', cron: '* * * * *' }

// Jobs found
{ event: 'jobs_found', count: 5 }

// Job enqueued
{ event: 'job_enqueued', job_id: '...', execution_id: '...', trigger_type: 'time.before_recording' }

// Queue batch completed
{ event: 'scheduler_queue_completed', processed: 10, errors: 0, duration_ms: 245 }

// Errors
{ event: 'job_processing_failed', job_id: '...', error: 'Error message' }

Error Handling

Queue Retry Strategy

• Max Retries: 3 attempts
• Retry Behavior: Automatic via queue configuration
• Dead Letter Queue: Failed messages after 3 retries go to automation-dlq

Error Scenarios

Scenario	Handling
Rule not found	Skip job, log warning
Idempotency key exists	Skip job, mark as completed
Rule disabled/paused	Skip execution
Database connection error	Let queue retry
Context building failure	Log error, increment error count

Deployment

Deploy Worker

cd workers/automation-scheduler
npx wrangler deploy

View Logs

npx wrangler tail podcasterplus-automation-scheduler

Create Queues (if not exists)

npx wrangler queues create automation-events
npx wrangler queues create automation-executions
npx wrangler queues create automation-dlq

Testing

Manual Time Trigger

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

Send Test Event

// From main app or another worker
await env.AUTOMATION_EVENTS_QUEUE.send({
    trigger_type: 'booking.confirmed',
    podcast_id: 'podcast-uuid',
    booking_id: 'booking-uuid',
    guest_id: 'guest-uuid',
    timestamp: new Date().toISOString()
});

Performance Considerations

Connection Pooling

Uses Hyperdrive for connection pooling:

• 10-100x faster cold starts than direct connections
• Connection reuse across invocations
• Automatic connection management

Batch Limits

• Time-based: Processes up to 100 jobs per cron invocation
• Event-based: Processes batches of up to 50 messages
• Timeout: 30-second batch timeout for event processing

Database Queries

• Single connection per invocation (max: 1)
• Indexed queries on scheduled_for and status
• Joins with automation_rules for enabled/paused checks