Media library on Cloud Run with Ittybit

View Markdown

A Cloud Run service accepts file uploads, stores them in Cloud Storage, dispatches transcoding and thumbnail tasks to Ittybit, and receives webhook callbacks that update a Cloud SQL catalog. A separate endpoint serves the media library to clients.

Architecture

  1. Client uploads a file to the Cloud Run service
  2. The service streams it to a Cloud Storage bucket and inserts a pending row into Cloud SQL
  3. Two Ittybit tasks are created: a video transcode and a thumbnail extraction
  4. Ittybit processes both asynchronously and POSTs webhooks on completion
  5. The webhook handler updates Cloud SQL with the output URLs
  6. Clients query the /media endpoint to browse the catalog

Database schema

Create a media table in your Cloud SQL Postgres instance:

CREATE TABLE media (
  id TEXT PRIMARY KEY DEFAULT gen_random_uuid()::text,
  filename TEXT NOT NULL,
  source_url TEXT NOT NULL,
  video_url TEXT,
  thumbnail_url TEXT,
  width INTEGER,
  height INTEGER,
  duration REAL,
  status TEXT NOT NULL DEFAULT 'pending',
  created_at TIMESTAMPTZ DEFAULT now(),
  updated_at TIMESTAMPTZ DEFAULT now()
);

Upload route

Accept a file, stream it to Cloud Storage, insert a database row, and dispatch both Ittybit tasks.

import express from "express";
import { Storage } from "@google-cloud/storage";
import pg from "pg";

const app = express();
app.use(express.json());

const storage = new Storage();
const bucket = storage.bucket(process.env.GCS_BUCKET!);
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const ITTYBIT_API_KEY = process.env.ITTYBIT_API_KEY!;
const WEBHOOK_URL = process.env.WEBHOOK_URL!;

app.post("/upload", express.raw({ type: "video/\*", limit: "500mb" }), async (req, res) => {
const filename = req.headers["x-filename"] as string;
if (!filename) return res.status(400).json({ error: "x-filename header required" });

// 1. Write to Cloud Storage
const blob = bucket.file(`uploads/${Date.now()}-${filename}`);
await blob.save(req.body, { contentType: req.headers["content-type"] });
const sourceUrl = `gs://${bucket.name}/${blob.name}`;

// 2. Generate a signed URL for Ittybit to read
const [signedUrl] = await blob.getSignedUrl({
version: "v4",
action: "read",
expires: Date.now() + 2 _ 60 _ 60 \* 1000,
});

// 3. Insert pending record
const { rows } = await pool.query(
`INSERT INTO media (filename, source_url, status)
     VALUES ($1, $2, 'processing') RETURNING id`,
[filename, sourceUrl]
);
const mediaId = rows[0].id;

// 4. Dispatch transcode + thumbnail tasks
const tasks = [
{
input: signedUrl,
kind: "video",
options: { width: 1920, format: "mp4", quality: "high" },
metadata: { media_id: mediaId, type: "transcode" },
webhook_url: WEBHOOK_URL,
},
{
input: signedUrl,
kind: "image",
options: { width: 640, format: "webp", start: 2 },
metadata: { media_id: mediaId, type: "thumbnail" },
webhook_url: WEBHOOK_URL,
},
];

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

res.status(201).json({ id: mediaId, status: "processing" });
});

import os
import time
import uuid
from datetime import timedelta
from flask import Flask, request, jsonify
from google.cloud import storage
import psycopg2
import requests

app = Flask(__name__)

storage_client = storage.Client()
bucket = storage_client.bucket(os.environ["GCS_BUCKET"])
ITTYBIT_API_KEY = os.environ["ITTYBIT_API_KEY"]
WEBHOOK_URL = os.environ["WEBHOOK_URL"]

def get_db():
    return psycopg2.connect(os.environ["DATABASE_URL"])

@app.route("/upload", methods=["POST"])
def upload():
    filename = request.headers.get("X-Filename")
    if not filename:
        return jsonify(error="X-Filename header required"), 400

    # 1. Write to Cloud Storage
    blob_name = f"uploads/{int(time.time())}-{filename}"
    blob = bucket.blob(blob_name)
    blob.upload_from_string(request.data, content_type=request.content_type)
    source_url = f"gs://{bucket.name}/{blob_name}"

    # 2. Generate a signed URL for Ittybit to read
    signed_url = blob.generate_signed_url(
        version="v4",
        expiration=timedelta(hours=2),
        method="GET",
    )

    # 3. Insert pending record
    media_id = str(uuid.uuid4())
    conn = get_db()
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO media (id, filename, source_url, status) VALUES (%s, %s, %s, 'processing')",
            (media_id, filename, source_url),
        )
    conn.commit()
    conn.close()

    # 4. Dispatch transcode + thumbnail tasks
    headers = {
        "Authorization": f"Bearer {ITTYBIT_API_KEY}",
        "Content-Type": "application/json",
    }
    tasks = [
        {
            "input": signed_url,
            "kind": "video",
            "options": {"width": 1920, "format": "mp4", "quality": "high"},
            "metadata": {"media_id": media_id, "type": "transcode"},
            "webhook_url": WEBHOOK_URL,
        },
        {
            "input": signed_url,
            "kind": "image",
            "options": {"width": 640, "format": "webp", "start": 2},
            "metadata": {"media_id": media_id, "type": "thumbnail"},
            "webhook_url": WEBHOOK_URL,
        },
    ]
    for task in tasks:
        requests.post("https://api.ittybit.com/jobs", headers=headers, json=task).raise_for_status()

    return jsonify(id=media_id, status="processing"), 201

Webhook handler

Ittybit POSTs to this endpoint when each task finishes. The metadata.type field determines which column to update.

app.post("/webhooks/ittybit", async (req, res) => {
  const { id, status, output, metadata } = req.body;
  const mediaId = metadata?.media_id;

if (!mediaId) return res.sendStatus(200);

if (status === "completed" && metadata.type === "transcode") {
await pool.query(
`UPDATE media
       SET video_url = $1, width = $2, height = $3, duration = $4, updated_at = now()
       WHERE id = $5`,
[output.url, output.width, output.height, output.duration, mediaId]
);
}

if (status === "completed" && metadata.type === "thumbnail") {
await pool.query(
`UPDATE media SET thumbnail_url = $1, updated_at = now() WHERE id = $2`,
[output.url, mediaId]
);
}

if (status === "failed") {
await pool.query(
`UPDATE media SET status = 'failed', updated_at = now() WHERE id = $1`,
[mediaId]
);
return res.sendStatus(200);
}

// Mark as ready once both outputs are present
await pool.query(
`UPDATE media SET status = 'ready', updated_at = now()
     WHERE id = $1 AND video_url IS NOT NULL AND thumbnail_url IS NOT NULL`,
[mediaId]
);

res.sendStatus(200);
});

@app.route("/webhooks/ittybit", methods=["POST"])
def webhook():
    body = request.get_json(silent=True)
    if not body:
        return "", 200

    status = body.get("status")
    output = body.get("output", {})
    metadata = body.get("metadata", {})
    media_id = metadata.get("media_id")

    if not media_id:
        return "", 200

    conn = get_db()
    with conn.cursor() as cur:
        if status == "completed" and metadata.get("type") == "transcode":
            cur.execute(
                """UPDATE media
                   SET video_url = %s, width = %s, height = %s, duration = %s, updated_at = now()
                   WHERE id = %s""",
                (output["url"], output["width"], output["height"], output["duration"], media_id),
            )

        if status == "completed" and metadata.get("type") == "thumbnail":
            cur.execute(
                "UPDATE media SET thumbnail_url = %s, updated_at = now() WHERE id = %s",
                (output["url"], media_id),
            )

        if status == "failed":
            cur.execute(
                "UPDATE media SET status = 'failed', updated_at = now() WHERE id = %s",
                (media_id,),
            )
            conn.commit()
            conn.close()
            return "", 200

        # Mark as ready once both outputs are present
        cur.execute(
            """UPDATE media SET status = 'ready', updated_at = now()
               WHERE id = %s AND video_url IS NOT NULL AND thumbnail_url IS NOT NULL""",
            (media_id,),
        )

    conn.commit()
    conn.close()
    return "", 200

Media catalog endpoint

Serve the library with pagination so clients can browse processed media.

app.get("/media", async (req, res) => {
  const limit = Math.min(parseInt(req.query.limit as string) || 20, 100);
  const offset = parseInt(req.query.offset as string) || 0;

const { rows } = await pool.query(
`SELECT id, filename, video_url, thumbnail_url, width, height, duration, status, created_at
     FROM media
     ORDER BY created_at DESC
     LIMIT $1 OFFSET $2`,
[limit, offset]
);

res.json(rows);
});

app.get("/media/:id", async (req, res) => {
const { rows } = await pool.query("SELECT \* FROM media WHERE id = $1", [req.params.id]);
if (!rows.length) return res.status(404).json({ error: "not found" });
res.json(rows[0]);
});

app.listen(parseInt(process.env.PORT || "8080"));

@app.route("/media")
def list_media():
    limit = min(int(request.args.get("limit", 20)), 100)
    offset = int(request.args.get("offset", 0))

    conn = get_db()
    with conn.cursor() as cur:
        cur.execute(
            """SELECT id, filename, video_url, thumbnail_url, width, height,
                      duration, status, created_at
               FROM media ORDER BY created_at DESC LIMIT %s OFFSET %s""",
            (limit, offset),
        )
        columns = [desc[0] for desc in cur.description]
        rows = [dict(zip(columns, row)) for row in cur.fetchall()]
    conn.close()
    return jsonify(rows)

@app.route("/media/<media_id>")
def get_media(media_id):
    conn = get_db()
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM media WHERE id = %s", (media_id,))
        row = cur.fetchone()
        if not row:
            conn.close()
            return jsonify(error="not found"), 404
        columns = [desc[0] for desc in cur.description]
        result = dict(zip(columns, row))
    conn.close()
    return jsonify(result)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))

Deploy

Build a container and deploy to Cloud Run. Set environment variables for your Cloud SQL connection, GCS bucket, and Ittybit API key.

gcloud run deploy media-library \
  --source . \
  --region us-central1 \
  --allow-unauthenticated \
  --add-cloudsql-instances $CLOUD_SQL_CONNECTION \
  --set-env-vars "DATABASE_URL=postgresql://user:pass@localhost/media?host=/cloudsql/$CLOUD_SQL_CONNECTION" \
  --set-env-vars "GCS_BUCKET=my-media-uploads" \
  --set-env-vars "ITTYBIT_API_KEY=$ITTYBIT_API_KEY" \
  --set-env-vars "WEBHOOK_URL=https://media-library-xxxxx-uc.a.run.app/webhooks/ittybit"
gcloud run deploy media-library \
  --source . \
  --region us-central1 \
  --allow-unauthenticated \
  --add-cloudsql-instances $CLOUD_SQL_CONNECTION \
  --set-env-vars "DATABASE_URL=postgresql://user:pass@localhost/media?host=/cloudsql/$CLOUD_SQL_CONNECTION" \
  --set-env-vars "GCS_BUCKET=my-media-uploads" \
  --set-env-vars "ITTYBIT_API_KEY=$ITTYBIT_API_KEY" \
  --set-env-vars "WEBHOOK_URL=https://media-library-xxxxx-uc.a.run.app/webhooks/ittybit"

After deploying, update the WEBHOOK_URL env var with the actual Cloud Run service URL, then redeploy.

See also