Objective
Build an automated media processing workflow system that manages processing pipelines, tracks job status, handles errors, and sends notifications on completion.
Step 1: Create Workflow Definitions Table
Define workflow templates.
CREATE TABLE workflow_definitions (
id INTEGER PRIMARY KEY,
workflow_code VARCHAR(50) NOT NULL UNIQUE,
workflow_name VARCHAR(200),
description TEXT,
workflow_type VARCHAR(50),
input_types TEXT,
output_types TEXT,
steps_json TEXT,
default_priority INTEGER DEFAULT 5,
timeout_minutes INTEGER DEFAULT 60,
retry_count INTEGER DEFAULT 3,
is_active BOOLEAN DEFAULT TRUE,
created_by VARCHAR(100),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
Step 2: Create Workflow Steps Table
Define individual steps.
CREATE TABLE workflow_steps (
id INTEGER PRIMARY KEY,
workflow_id INTEGER NOT NULL,
step_number INTEGER,
step_name VARCHAR(100),
step_type VARCHAR(50),
action VARCHAR(100),
parameters TEXT,
required_inputs TEXT,
expected_outputs TEXT,
timeout_seconds INTEGER DEFAULT 300,
on_failure VARCHAR(50) DEFAULT 'stop',
FOREIGN KEY (workflow_id) REFERENCES workflow_definitions(id)
);
Step 3: Create Workflow Jobs Table
Track job executions.
CREATE TABLE workflow_jobs (
id INTEGER PRIMARY KEY,
job_code VARCHAR(50) NOT NULL UNIQUE,
workflow_id INTEGER NOT NULL,
input_file_path TEXT,
input_metadata TEXT,
output_file_path TEXT,
output_metadata TEXT,
priority INTEGER DEFAULT 5,
status VARCHAR(50) DEFAULT 'pending',
current_step INTEGER DEFAULT 0,
progress_percent DECIMAL(5, 2) DEFAULT 0,
started_at TIMESTAMP,
completed_at TIMESTAMP,
error_message TEXT,
retry_count INTEGER DEFAULT 0,
submitted_by VARCHAR(100),
submitted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (workflow_id) REFERENCES workflow_definitions(id)
);
Step 4: Create Step Executions Table
Track step progress.
CREATE TABLE step_executions (
id INTEGER PRIMARY KEY,
job_id INTEGER NOT NULL,
step_id INTEGER NOT NULL,
status VARCHAR(50) DEFAULT 'pending',
input_data TEXT,
output_data TEXT,
started_at TIMESTAMP,
completed_at TIMESTAMP,
duration_seconds INTEGER,
error_message TEXT,
retry_attempt INTEGER DEFAULT 0,
FOREIGN KEY (job_id) REFERENCES workflow_jobs(id),
FOREIGN KEY (step_id) REFERENCES workflow_steps(id)
);
Step 5: Create Media Queue Table
Manage processing queue.
CREATE TABLE media_queue (
id INTEGER PRIMARY KEY,
job_id INTEGER NOT NULL,
queue_name VARCHAR(50),
priority INTEGER DEFAULT 5,
scheduled_time TIMESTAMP,
picked_up_at TIMESTAMP,
worker_id VARCHAR(100),
status VARCHAR(50) DEFAULT 'queued',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (job_id) REFERENCES workflow_jobs(id)
);
Step 6: Create Notifications Table
Track workflow notifications.
CREATE TABLE workflow_notifications (
id INTEGER PRIMARY KEY,
job_id INTEGER NOT NULL,
notification_type VARCHAR(50),
recipient VARCHAR(200),
channel VARCHAR(50),
subject VARCHAR(300),
message TEXT,
status VARCHAR(50) DEFAULT 'pending',
sent_at TIMESTAMP,
error_message TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (job_id) REFERENCES workflow_jobs(id)
);
Step 7: Insert Workflow Definitions
Add workflow templates.
INSERT INTO workflow_definitions (id, workflow_code, workflow_name, description, workflow_type, input_types, output_types, default_priority, timeout_minutes, is_active) VALUES
(1, 'WF-VIDEO-TRANSCODE', 'Video Transcoding Pipeline', 'Transcode video to multiple formats and resolutions', 'video', 'video/mp4,video/avi,video/mov', 'video/mp4', 5, 120, TRUE),
(2, 'WF-IMAGE-PROCESS', 'Image Processing Pipeline', 'Resize, optimize, and generate thumbnails', 'image', 'image/jpeg,image/png', 'image/jpeg,image/png,image/webp', 3, 30, TRUE),
(3, 'WF-AUDIO-TRANSCODE', 'Audio Transcoding Pipeline', 'Convert audio and generate waveforms', 'audio', 'audio/wav,audio/flac', 'audio/mp3,audio/aac', 4, 45, TRUE),
(4, 'WF-DOC-PROCESS', 'Document Processing Pipeline', 'Extract text, generate thumbnails, convert formats', 'document', 'application/pdf,application/docx', 'application/pdf,text/plain', 4, 60, TRUE),
(5, 'WF-VIDEO-ANALYZE', 'Video Analysis Pipeline', 'Extract metadata, generate thumbnails, transcribe audio', 'video_analysis', 'video/mp4', 'application/json', 6, 180, TRUE);
Step 8: Insert Workflow Steps
Define step sequences.
INSERT INTO workflow_steps (id, workflow_id, step_number, step_name, step_type, action, parameters, timeout_seconds, on_failure) VALUES
-- Video Transcoding Pipeline
(1, 1, 1, 'Validate Input', 'validation', 'validate_video', '{"check_codec": true, "check_duration": true}', 30, 'stop'),
(2, 1, 2, 'Extract Metadata', 'extraction', 'extract_video_metadata', '{}', 60, 'continue'),
(3, 1, 3, 'Transcode 1080p', 'transcode', 'transcode_video', '{"resolution": "1080p", "codec": "h264"}', 1800, 'retry'),
(4, 1, 4, 'Transcode 720p', 'transcode', 'transcode_video', '{"resolution": "720p", "codec": "h264"}', 1200, 'retry'),
(5, 1, 5, 'Transcode 480p', 'transcode', 'transcode_video', '{"resolution": "480p", "codec": "h264"}', 600, 'retry'),
(6, 1, 6, 'Generate Thumbnails', 'thumbnail', 'generate_thumbnails', '{"count": 5, "format": "jpeg"}', 120, 'continue'),
(7, 1, 7, 'Finalize Output', 'finalization', 'finalize_job', '{}', 60, 'stop'),
-- Image Processing Pipeline
(8, 2, 1, 'Validate Image', 'validation', 'validate_image', '{}', 30, 'stop'),
(9, 2, 2, 'Resize Large', 'resize', 'resize_image', '{"max_dimension": 2000}', 60, 'retry'),
(10, 2, 3, 'Generate Thumbnail', 'thumbnail', 'create_thumbnail', '{"size": 300}', 30, 'continue'),
(11, 2, 4, 'Optimize', 'optimization', 'optimize_image', '{"quality": 85}', 60, 'retry'),
(12, 2, 5, 'Convert WebP', 'conversion', 'convert_format', '{"format": "webp"}', 45, 'continue');