Queues

Handle time-consuming tasks asynchronously without blocking user requests. Pixflare uses three queues for variant generation, S3 backup, and custom domain verification.

Setup

Enable queues with:

ENABLE_QUEUES='true'

Queues are provisioned by Terraform and bound in wrangler.toml. They cost $5/month on Cloudflare (includes 1 million operations).

Why They're Needed

Some operations take too long to run synchronously:

• Variant generation - Creating 8 different image sizes can take 5-15 seconds
• S3 backup - Uploading to external storage adds latency
• Domain verification - DNS propagation requires polling over 20 minutes

Running these in-request would timeout Workers (max 30s CPU time) and frustrate users waiting for uploads to finish.

How It Works

After an upload completes, the API returns immediately while queues process in the background:

1. Upload endpoint stores the image in R2 and enqueues a variant generation job
2. Queue consumer fetches the job, generates variants, stores them, and triggers AI if enabled
3. Frontend polls the API to check when variants are ready (typically 10-30 seconds)

Messages are processed in batches (configurable per queue) with automatic retries on failure.

The Three Queues

Variant Queue

• Purpose: Generate image variants (w128, w256, etc.) and run AI classification
• Batch size: 5 messages
• Timeout: 30 seconds per batch
• Retries: 3 attempts with exponential backoff

Backup Queue

• Purpose: Sync images to S3-compatible storage
• Batch size: 10 messages
• Timeout: 60 seconds per batch
• Retries: 3 attempts

Custom Domain Queue

• Purpose: Poll Cloudflare API for DNS verification status
• Batch size: 5 messages
• Timeout: 30 seconds per batch
• Retries: Up to 40 checks (every 30 seconds for 20 minutes)

Disabling

To disable queues (development only):

ENABLE_QUEUES='false'

Uploads will still work but variants won't be generated, backups won't sync, and custom domains won't verify. Images will only have the original variant.

Queue Processing Flow

flowchart TB
    subgraph VQ["Variant Generation Queue"]
        direction LR
        VTrigger[Image Upload Complete] --> VEnqueue["Enqueue to VARIANT_QUEUE<br/>Message: imageId, owner"]
        VEnqueue --> VConsumer["Queue Consumer<br/>Batch: 5, Timeout: 30s"]
        VConsumer --> VFetch[Fetch Original from R2]
        VFetch --> VGen["Generate 8 Variants:<br/>w128, w256, w512, w1024,<br/>w1536, w2048, thumb, og-image"]
        VGen --> VStore[Store to R2<br/>owner/album/file/variant]
        VStore --> VUpdate[Update D1<br/>variants_ready array]
        VUpdate --> VCheck{AI Enabled?}
        VCheck -->|Yes| VAI[Trigger AI Classification]
        VCheck -->|No| VDone[Complete]
        VAI --> VDone
    end

    subgraph BQ["Backup Queue"]
        direction LR
        BTrigger[Upload/Delete Event] --> BEnqueue["Enqueue to BACKUP_QUEUE<br/>Message: imageId, action"]
        BEnqueue --> BConsumer["Queue Consumer<br/>Batch: 10, Timeout: 60s"]
        BConsumer --> BCheck{Action Type}
        BCheck -->|upload| BFetch[Fetch from R2]
        BCheck -->|delete| BDelete[Delete from S3]
        BFetch --> BEncrypt[Encrypt if Configured]
        BEncrypt --> BS3[Upload to S3-compatible<br/>AWS/MinIO/Backblaze]
        BS3 --> BUpdate[Update backup_status in D1]
        BDelete --> BUpdate
        BUpdate --> BDone[Complete]
    end

    subgraph DQ["Custom Domain Queue"]
        direction LR
        DTrigger[Domain Verification Request] --> DEnqueue["Enqueue to CUSTOM_DOMAIN_QUEUE<br/>Message: domainId, hostname"]
        DEnqueue --> DConsumer["Queue Consumer<br/>Batch: 5, Timeout: 30s"]
        DConsumer --> DDNSCheck[Check DNS Records<br/>CNAME + TXT]
        DDNSCheck --> DVerified{DNS Valid?}
        DVerified -->|Yes| DActivate[Activate Custom Hostname<br/>via Cloudflare for SaaS API]
        DVerified -->|No| DRetry{Retry Count < 3?}
        DRetry -->|Yes| DRequeue[Re-enqueue with Delay]
        DRetry -->|No| DFail[Mark as Failed]
        DActivate --> DUpdate[Update status in D1]
        DFail --> DUpdate
        DRequeue --> DConsumer
        DUpdate --> DDone[Complete]
    end

    subgraph CRON["Scheduled Jobs - Cron Triggers"]
        direction LR
        Cron1[Daily 1 AM UTC:<br/>Cleanup Job]
        Cron2[Daily 2 AM UTC:<br/>Analytics Aggregation]
        Cron3[Daily 3 AM UTC:<br/>Backup Sync]

        Cron1 --> Clean1[Delete Expired Soft-deletes]
        Cron1 --> Clean2[Remove Orphaned Variants]

        Cron2 --> Agg1[Roll up View Counts]
        Cron2 --> Agg2[Generate Daily/Monthly Stats]

        Cron3 --> Sync1[Queue Pending Images]
        Cron3 --> Sync2[Verify Backup Status]
    end

    VQ ~~~ BQ
    BQ ~~~ DQ
    DQ ~~~ CRON

    style VTrigger fill:#90E24A
    style BTrigger fill:#4A90E2
    style DTrigger fill:#E2904A
    style Cron1 fill:#7B42BC
    style Cron2 fill:#7B42BC
    style Cron3 fill:#7B42BC