Queue Producers

This document describes the producer-side implementation of the automation system - how the main SvelteKit application emits events and schedules time-based jobs for worker consumption.

Architecture Pattern

The automation system uses a database-backed queue pattern:

Key Insight: The producer does NOT send queue messages directly. Instead, it writes records to database tables. Workers poll these tables and send queue messages for execution.

Source Files

File	Purpose
src/lib/automation/events.ts	Event-based automation emission
src/lib/automation/scheduler.ts	Time-based job scheduling
src/lib/automation/index.ts	Module exports

Event-Based Producers

Core Function: emitAutomationEvent()

The main producer function for event-based automations.

import { emitAutomationEvent } from '$lib/automation/events';

const result = await emitAutomationEvent(supabase, {
	trigger_type: 'booking.confirmed',
	podcast_id: booking.podcast_id,
	booking_id: booking.id,
	episode_id: booking.episode_id,
	guest_id: booking.guest_id
});

What it does:

1. Queries automation_rules for enabled rules matching the trigger type
2. Builds full context via buildFullContext() (fetches podcast, booking, episode, guest data)
3. Creates automation_executions records with status pending
4. Returns count of executions created

AutomationEventPayload Interface

interface AutomationEventPayload {
	trigger_type: AutomationTriggerType;
	podcast_id: string;
	episode_id?: string | null;
	booking_id?: string | null;
	guest_id?: string | null;
	context?: Partial<MagicTagContext>; // Additional context data
}

EmitEventResult Interface

interface EmitEventResult {
	success: boolean;
	executions_created: number; // How many execution records were created
	rule_ids: string[]; // Which rules matched and were triggered
	error?: string; // Error message if failed
}

Helper Functions

Convenience wrappers for common event types:

// Booking events
await emitBookingEvent(supabase, 'booking.confirmed', bookingId, podcastId, episodeId, guestId);
await emitBookingEvent(supabase, 'booking.declined', bookingId, podcastId);
await emitBookingEvent(supabase, 'booking.canceled', bookingId, podcastId);
await emitBookingEvent(supabase, 'booking.rescheduled', bookingId, podcastId, episodeId, guestId);

// Episode events
await emitEpisodeEvent(supabase, 'episode.published', episodeId, podcastId);
await emitEpisodeEvent(supabase, 'episode.scheduled', episodeId, podcastId);
await emitEpisodeEvent(supabase, 'episode.draft_created', episodeId, podcastId);

// Guest events
await emitGuestEvent(supabase, 'guest.responded', guestId, episodeId, podcastId);
await emitGuestEvent(supabase, 'guest.reminder_sent', guestId, episodeId, podcastId);

Time-Based Producers

Core Function: scheduleTimeBasedJobs()

Creates scheduled job records for time-based automation rules.

import { scheduleTimeBasedJobs } from '$lib/automation/scheduler';

const result = await scheduleTimeBasedJobs(supabase, {
	podcast_id: podcastId,
	episode_id: episodeId,
	booking_id: bookingId,
	guest_id: guestId,
	recording_date: new Date(booking.start_time),
	publish_date: episode.scheduled_for ? new Date(episode.scheduled_for) : null,
	booking_date: new Date(),
	timezone: booking.timezone
});

What it does:

1. Queries automation_rules for enabled time-based rules (trigger types starting with time.)
2. Calculates scheduled_for timestamp based on reference date and offset
3. Creates automation_scheduled_jobs records with status pending
4. Handles idempotency via unique constraint on idempotency_key

Time-Based Trigger Types

Trigger Type	Reference Field	Example
time.before_recording	recording_date	24h before start_time
time.after_recording	recording_date	1h after start_time
time.before_publish	publish_date	48h before scheduled_for
time.after_publish	publish_date	1h after published_at
time.after_booking	booking_date	5min after confirmation

TimeBasedSchedulingContext Interface

interface TimeBasedSchedulingContext {
	podcast_id: string;
	episode_id?: string | null;
	booking_id?: string | null;
	guest_id?: string | null;
	recording_date?: Date | null; // For recording-related triggers
	publish_date?: Date | null; // For publish-related triggers
	booking_date?: Date | null; // For booking-related triggers
	timezone?: string; // For time calculations
}

ScheduleResult Interface

interface ScheduleResult {
	success: boolean;
	jobs_created: number; // New scheduled jobs created
	jobs_skipped: number; // Skipped (no reference date, already exists)
	rule_ids: string[]; // Rules that got scheduled
	error?: string;
}

Specialized Schedulers

// For booking confirmations
await scheduleTimeBasedJobsForBooking(
	supabase,
	bookingId,
	podcastId,
	episodeId, // May be null
	guestId, // May be null
	recordingDate, // booking.start_time
	bookingDate, // new Date() - when booking was confirmed
	timezone
);

// For episode scheduling/publishing
await scheduleTimeBasedJobsForEpisode(
	supabase,
	episodeId,
	podcastId,
	publishDate, // episode.scheduled_for or published_at
	timezone
);

Cancellation and Rescheduling

// Cancel pending jobs when booking/episode is cancelled
await cancelPendingTimeBasedJobs(supabase, {
	episode_id: episodeId,
	// OR
	booking_id: bookingId,
	// OR
	podcast_id: podcastId // Cancels ALL for podcast
});

// Reschedule when dates change (cancels + creates new)
await rescheduleTimeBasedJobsForEpisode(
	supabase,
	episodeId,
	podcastId,
	newRecordingDate, // null to skip recording triggers
	newPublishDate, // null to skip publish triggers
	bookingId,
	guestId,
	timezone
);

Context Building

Both producers automatically enrich events with database data before storing:

Data Fetched

Source	Fields Used
podcasts	title, slug
bookings	guest_name, guest_email, guest_phone, start_time, timezone, meeting_url
booking_links	name
episodes	title, episode_number, description, status, scheduled_for, published_at
episode_guests	name, email, access_token

Context Snapshot

The enriched context is stored as JSON in:

• automation_executions.context_data (event-based)
• automation_scheduled_jobs.context_snapshot (time-based)

This ensures workers have all necessary data without additional DB queries.

Idempotency

Event-Based Idempotency Key

Format: {rule_id}:{trigger_type}:{podcast_id}:{episode_id}:{booking_id}:{guest_id}:{date}

• Includes date (YYYY-MM-DD) to allow same event daily (for digests)
• Database UNIQUE constraint prevents duplicates
• Returns success with 0 executions if duplicate detected

Time-Based Idempotency Key

Format: time_trigger:{rule_id}:{trigger_type}:{entity_id}:{reference_datetime_ISO}

• Uses full ISO timestamp to support same-day reschedules
• If user reschedules from 14:00 to 15:45, gets different keys
• Entity ID is booking_id || episode_id || podcast_id

Real-World Usage

Booking Confirmation Flow

// src/api/routes/bookings/index.ts (simplified)
async function confirmBooking(bookingId: string) {
	// ... booking logic ...

	// 1. Emit event-based automations (immediate)
	const eventResult = await emitBookingEvent(
		supabase,
		'booking.confirmed',
		bookingId,
		booking.podcast_id,
		booking.episode_id,
		booking.guest_id
	);

	// 2. Schedule time-based automations (delayed)
	const scheduleResult = await scheduleTimeBasedJobsForBooking(
		supabase,
		bookingId,
		booking.podcast_id,
		booking.episode_id,
		booking.guest_id,
		new Date(booking.start_time),
		new Date(),
		booking.timezone
	);

	console.log(
		`Automation: ${eventResult.executions_created} immediate, ${scheduleResult.jobs_created} scheduled`
	);
}

Episode Rescheduling Flow

// When episode recording or publish date changes
async function updateEpisodeDates(episodeId: string, newDates: Partial<Episode>) {
	// ... update episode ...

	// Reschedule all time-based automations
	const result = await rescheduleTimeBasedJobsForEpisode(
		supabase,
		episodeId,
		episode.podcast_id,
		newDates.recording_scheduled_at ? new Date(newDates.recording_scheduled_at) : null,
		newDates.scheduled_for ? new Date(newDates.scheduled_for) : null,
		episode.booking_id,
		episode.guest_id,
		episode.timezone
	);

	console.log(`Rescheduled: ${result.jobs_cancelled} cancelled, ${result.jobs_created} created`);
}

Database Tables Used

Table	Written By	Read By
automation_rules	UI/API	Producers (for rule matching)
automation_executions	Event producers	Executor worker
automation_scheduled_jobs	Time producers	Scheduler worker

Design Patterns

Pattern	Implementation
Database-Backed Queue	Records in DB, workers poll and process
Idempotency	UNIQUE constraint on idempotency_key columns
Context Capture	All data captured upfront, no worker DB queries for context
Fire-and-Forget	Producers don't wait for execution results
Two-Phase Dispatch	Event → immediate; Time → scheduled for later
Admin Client	Producers use service role key to bypass RLS

Error Handling

Producers return structured results instead of throwing:

const result = await emitAutomationEvent(supabase, payload);

if (!result.success) {
	console.error('Automation failed:', result.error);
	// Non-blocking: don't fail the main operation
}

Related Documentation

• Automation Architecture - System-wide design
• Automation Scheduler Worker - Consumer for scheduled jobs
• Automation Executor Worker - Consumer for executions
• Cloudflare Services - Queue configuration