Media processing with Cloudflare Workers, R2, and Ittybit

View Markdown

R2 holds your files, Ittybit processes them, and Workers wire it all together. When a file lands in R2, a Worker dispatches an Ittybit task using an s3:// input URL. When processing finishes, a webhook hits a second Worker that writes the output metadata to D1.

Connect R2 to Ittybit

Create a connection so Ittybit can read directly from your R2 bucket. R2 is S3-compatible, so use the S3 connection type with your Cloudflare account endpoint.

const res = await fetch("https://api.ittybit.com/connections", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${process.env.ITTYBIT_API_KEY}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    kind: "s3",
    name: "my-r2-bucket",
    endpoint: `https://${CLOUDFLARE_ACCOUNT_ID}.r2.cloudflarestorage.com`,
    region: "auto",
    access_key_id: process.env.R2_ACCESS_KEY_ID,
    secret_access_key: process.env.R2_SECRET_ACCESS_KEY,
  }),
});
const connection = await res.json();
// { id: "conn_abc123", name: "my-r2-bucket", ... }
curl -X POST https://api.ittybit.com/connections \
  -H "Authorization: Bearer $ITTYBIT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "kind": "s3",
    "name": "my-r2-bucket",
    "endpoint": "https://<account-id>.r2.cloudflarestorage.com",
    "region": "auto",
    "access_key_id": "'$R2_ACCESS_KEY_ID'",
    "secret_access_key": "'$R2_SECRET_ACCESS_KEY'"
  }'

Set up the D1 database

Create a table to track processed media. This is where the webhook Worker will write results.

CREATE TABLE media (
  id TEXT PRIMARY KEY,
  r2_key TEXT NOT NULL,
  task_id TEXT,
  status TEXT DEFAULT 'pending',
  output_url TEXT,
  created_at TEXT DEFAULT (datetime('now'))
);

Worker: dispatch tasks on upload

This Worker fires when a new object arrives in R2. It creates a record in D1 and dispatches an Ittybit task with the s3:// input pointing to the uploaded file.

export interface Env {
  ITTYBIT_API_KEY: string;
  MEDIA_BUCKET: R2Bucket;
  DB: D1Database;
}

export default {
  async queue(batch: MessageBatch, env: Env) {
    for (const message of batch.messages) {
      const event = message.body as { key: string };
      const r2Key = event.key;

      // Insert a pending record
      await env.DB.prepare(
        "INSERT INTO media (id, r2_key, status) VALUES (?, ?, 'pending')"
      )
        .bind(crypto.randomUUID(), r2Key)
        .run();

      // Dispatch the Ittybit task
      const task = {
        kind: "video",
        input: `s3://my-r2-bucket/${r2Key}`,
        connection_id: "conn_abc123",
        webhook_url: "https://webhook.example.com/ittybit",
        options: {
          format: "mp4",
          width: 1920,
          quality: "high",
        },
      };

      await fetch("https://api.ittybit.com/jobs", {
        method: "POST",
        headers: {
          Authorization: `Bearer ${env.ITTYBIT_API_KEY}`,
          "Content-Type": "application/json",
        },
        body: JSON.stringify(task),
      });

      message.ack();
    }

},
} satisfies ExportedHandler<Env>;

# What the Worker sends to Ittybit for each upload:

curl -X POST https://api.ittybit.com/jobs \
  -H "Authorization: Bearer $ITTYBIT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "kind": "video",
    "input": "s3://my-r2-bucket/uploads/video.mov",
    "connection_id": "conn_abc123",
    "webhook_url": "https://webhook.example.com/ittybit",
    "options": {
      "format": "mp4",
      "width": 1920,
      "quality": "high"
    }
  }'

Worker: handle the webhook callback

When Ittybit finishes processing, it POSTs to your webhook URL. This second Worker receives the payload and updates D1 with the output.

export interface Env {
  ITTYBIT_WEBHOOK_SECRET: string;
  DB: D1Database;
}

export default {
  async fetch(request: Request, env: Env) {
    if (request.method !== "POST") {
      return new Response("Method not allowed", { status: 405 });
    }

    const body = await request.json<{
      id: string;
      object: string;
      status: string;
      input: string;
      output: { url: string };
    }>();

    // Extract the R2 key from the s3:// input URL
    const r2Key = body.input.replace("s3://my-r2-bucket/", "");

    if (body.status === "succeeded") {
      await env.DB.prepare(
        "UPDATE media SET task_id = ?, status = 'completed', output_url = ? WHERE r2_key = ?"
      )
        .bind(body.id, body.output.url, r2Key)
        .run();
    } else {
      await env.DB.prepare(
        "UPDATE media SET task_id = ?, status = 'failed' WHERE r2_key = ?"
      )
        .bind(body.id, r2Key)
        .run();
    }

    return new Response("OK", { status: 200 });

},
} satisfies ExportedHandler<Env>;

# Example webhook payload from Ittybit on task completion:

# POST https://webhook.example.com/ittybit
# Content-Type: application/json
#
# {
#   "id": "task_abc123",
#   "object": "task",
#   "kind": "video",
#   "status": "succeeded",
#   "input": "s3://my-r2-bucket/uploads/video.mov",
#   "output": {
#     "url": "https://cdn.ittybit.com/abc123/video.mp4"
#   }
# }

Wrangler config

Bind both Workers to R2, D1, and secrets in your wrangler.toml:

name = "media-pipeline"

[[r2_buckets]]
binding = "MEDIA_BUCKET"
bucket_name = "my-r2-bucket"

[[d1_databases]]
binding = "DB"
database_name = "media-db"
database_id = "<your-d1-database-id>"

[[queues.consumers]]
queue = "r2-upload-notifications"

[vars]
# Set secrets via: wrangler secret put ITTYBIT_API_KEY

Multiple output formats

Dispatch several tasks per upload for different renditions — web, mobile, thumbnail:

const tasks = [
  {
    kind: "video",
    input: `s3://my-r2-bucket/${r2Key}`,
    connection_id: "conn_abc123",
    webhook_url: "https://webhook.example.com/ittybit",
    options: { format: "mp4", width: 1920, quality: "high" },
  },
  {
    kind: "video",
    input: `s3://my-r2-bucket/${r2Key}`,
    connection_id: "conn_abc123",
    webhook_url: "https://webhook.example.com/ittybit",
    options: { format: "mp4", width: 480, quality: "medium" },
  },
  {
    kind: "image",
    input: `s3://my-r2-bucket/${r2Key}`,
    connection_id: "conn_abc123",
    webhook_url: "https://webhook.example.com/ittybit",
    options: { format: "webp", width: 640, start: 2 },
  },
];

await Promise.all(
tasks.map((task) =>
fetch("https://api.ittybit.com/jobs", {
method: "POST",
headers: {
Authorization: `Bearer ${env.ITTYBIT_API_KEY}`,
"Content-Type": "application/json",
},
body: JSON.stringify(task),
})
)
);

# Web version
curl -X POST https://api.ittybit.com/jobs \
  -H "Authorization: Bearer $ITTYBIT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "kind": "video",
    "input": "s3://my-r2-bucket/uploads/video.mov",
    "connection_id": "conn_abc123",
    "webhook_url": "https://webhook.example.com/ittybit",
    "options": { "format": "mp4", "width": 1920, "quality": "high" }
  }'

# Mobile version
curl -X POST https://api.ittybit.com/jobs \
  -H "Authorization: Bearer $ITTYBIT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "kind": "video",
    "input": "s3://my-r2-bucket/uploads/video.mov",
    "connection_id": "conn_abc123",
    "webhook_url": "https://webhook.example.com/ittybit",
    "options": { "format": "mp4", "width": 480, "quality": "medium" }
  }'

# Thumbnail
curl -X POST https://api.ittybit.com/jobs \
  -H "Authorization: Bearer $ITTYBIT_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "kind": "image",
    "input": "s3://my-r2-bucket/uploads/video.mov",
    "connection_id": "conn_abc123",
    "webhook_url": "https://webhook.example.com/ittybit",
    "options": { "format": "webp", "width": 640, "start": 2 }
  }'

See also