Video Transcode QA and CDN Publish Pipeline

id: media-transcode-cdn-pipeline
namespace: company.team
description: |
  Watch a source video bucket, transcode every new upload into 720p and 1080p variants with ffmpeg, gate the publish on ffprobe style QA of duration, size and resolution, then push the approved variants to the CDN origin bucket and purge the Cloudflare cache.

variables:
  aws_region: eu-west-1
  source_bucket: media-landing-bucket
  cdn_prefix: video
  ffmpeg_image: jrottenberg/ffmpeg:6.1-alpine

triggers:
  - id: new_source_video
    type: io.kestra.plugin.aws.s3.Trigger
    description: Polls the source bucket and starts one execution for each new video
      object that lands under the watched prefix.
    interval: PT1M
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ vars.aws_region }}"
    bucket: "{{ vars.source_bucket }}"
    prefix: incoming/video/
    filter: FILES
    action: NONE
    maxKeys: 1

inputs:
  - id: cdn_bucket
    type: STRING
    defaults: media-cdn-origin
    description: Origin bucket behind the CDN where QA approved variants are published.
  - id: cdn_host
    type: STRING
    defaults: cdn.example.com
    description: Public hostname served by the CDN, used to build the cache purge URLs.
  - id: cloudflare_zone_id
    type: STRING
    defaults: change-me-zone-id
    description: Cloudflare zone that fronts the CDN origin and receives the purge request.

tasks:
  # --- 1. transcode both renditions ---
  - id: transcode
    type: io.kestra.plugin.scripts.shell.Commands
    description: Stage the triggered source video, encode the 720p and 1080p
      variants with ffmpeg, and capture ffprobe metadata plus a size manifest
      for every rendition.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: "{{ vars.ffmpeg_image }}"
    inputFiles:
      source.mp4: "{{ trigger.objects[0].uri }}"
    outputFiles:
      - variant_720p.mp4
      - variant_1080p.mp4
      - probe_source.json
      - probe_720p.json
      - probe_1080p.json
      - sizes.txt
    commands:
      - ffmpeg -y -i {{ workingDir }}/source.mp4 -vf scale=-2:720 -c:v libx264
        -preset veryfast -crf 23 -c:a aac -movflags +faststart {{ workingDir
        }}/variant_720p.mp4
      - ffmpeg -y -i {{ workingDir }}/source.mp4 -vf scale=-2:1080 -c:v libx264
        -preset veryfast -crf 20 -c:a aac -movflags +faststart {{ workingDir
        }}/variant_1080p.mp4
      - ffprobe -v error -select_streams v:0 -show_entries
        stream=width,height:format=duration -of json {{ workingDir }}/source.mp4
        > {{ workingDir }}/probe_source.json
      - ffprobe -v error -select_streams v:0 -show_entries
        stream=width,height:format=duration -of json {{ workingDir
        }}/variant_720p.mp4 > {{ workingDir }}/probe_720p.json
      - ffprobe -v error -select_streams v:0 -show_entries
        stream=width,height:format=duration -of json {{ workingDir
        }}/variant_1080p.mp4 > {{ workingDir }}/probe_1080p.json
      - stat -c '%s %n' {{ workingDir }}/source.mp4 {{ workingDir
        }}/variant_720p.mp4 {{ workingDir }}/variant_1080p.mp4 > {{ workingDir
        }}/sizes.txt

  # --- 2. ffprobe style QA of the variants against the source ---
  - id: qa_check
    type: io.kestra.plugin.scripts.python.Script
    description: Compare ffprobe metadata for both variants with the source and emit
      a pass or fail verdict plus the full list of violations.
    containerImage: python:3.12-slim
    inputFiles:
      probe_source.json: "{{ outputs.transcode.outputFiles['probe_source.json'] }}"
      probe_720p.json: "{{ outputs.transcode.outputFiles['probe_720p.json'] }}"
      probe_1080p.json: "{{ outputs.transcode.outputFiles['probe_1080p.json'] }}"
      sizes.txt: "{{ outputs.transcode.outputFiles['sizes.txt'] }}"
    script: |
      import json
      import os

      DURATION_TOLERANCE = 0.10
      EXPECTED_VARIANTS = {
          "variant_720p.mp4": 720,
          "variant_1080p.mp4": 1080,
      }
      PROBE_FILES = {
          "variant_720p.mp4": "probe_720p.json",
          "variant_1080p.mp4": "probe_1080p.json",
      }


      def probe(path):
          with open(path) as handle:
              data = json.load(handle)
          streams = data.get("streams") or [{}]
          stream = streams[0] or {}
          fmt = data.get("format") or {}
          return {
              "width": int(stream.get("width") or 0),
              "height": int(stream.get("height") or 0),
              "duration": float(fmt.get("duration") or 0),
          }


      def read_sizes(path):
          sizes = {}
          with open(path) as handle:
              for line in handle:
                  parts = line.strip().split(None, 1)
                  if len(parts) != 2:
                      continue
                  sizes[os.path.basename(parts[1])] = int(parts[0])
          return sizes


      def evaluate(source, variants, sizes):
          violations = []
          src_duration = float(source.get("duration") or 0)
          src_width = int(source.get("width") or 0)
          src_height = int(source.get("height") or 0)
          src_size = int(sizes.get("source.mp4") or 0)

          if src_duration <= 0:
              violations.append("source_duration_missing")
          if src_width <= 0 or src_height <= 0:
              violations.append("source_resolution_missing")
          if src_size <= 0:
              violations.append("source_size_missing")

          for name in sorted(EXPECTED_VARIANTS):
              expected_height = EXPECTED_VARIANTS[name]
              size = int(sizes.get(name) or 0)
              if name not in variants:
                  violations.append(name + ":probe_missing")
                  continue
              meta = variants[name]
              if int(meta.get("height") or 0) != expected_height:
                  violations.append(name + ":resolution_mismatch")
              if int(meta.get("width") or 0) <= 0:
                  violations.append(name + ":width_missing")
              if size <= 0:
                  violations.append(name + ":empty_file")
              elif src_size > 0 and size > src_size * 2:
                  violations.append(name + ":larger_than_source")
              duration = float(meta.get("duration") or 0)
              if duration <= 0:
                  violations.append(name + ":duration_missing")
              elif src_duration > 0 and abs(duration - src_duration) / src_duration > DURATION_TOLERANCE:
                  violations.append(name + ":duration_drift")

          return {
              "outputs": {
                  "verdict": "pass" if not violations else "fail",
                  "violations": violations,
                  "checked_variants": len(EXPECTED_VARIANTS),
                  "source_duration": src_duration,
              }
          }


      def main():
          try:
              source = probe("probe_source.json")
              variants = {name: probe(path) for name, path in PROBE_FILES.items()}
              sizes = read_sizes("sizes.txt")
          except Exception as exc:
              payload = {
                  "outputs": {
                      "verdict": "fail",
                      "violations": ["qa_inputs_unreadable: " + str(exc)],
                      "checked_variants": 0,
                      "source_duration": 0.0,
                  }
              }
          else:
              payload = evaluate(source, variants, sizes)
          print("::" + json.dumps(payload) + "::")


      main()

  # --- 3. publish only when QA passes ---
  - id: qa_gate
    type: io.kestra.plugin.core.flow.If
    description: Publish the variants to the CDN only when the QA verdict is pass,
      otherwise stop the run with an explicit failure.
    condition: "{{ outputs.qa_check.vars.verdict == 'pass' }}"
    then:
      - id: push_720p
        type: io.kestra.plugin.aws.s3.Upload
        description: Copy the 720p variant into the CDN origin bucket under the public
          rendition key.
        from: "{{ outputs.transcode.outputFiles['variant_720p.mp4'] }}"
        bucket: "{{ inputs.cdn_bucket }}"
        key: "{{ vars.cdn_prefix }}/variant_720p.mp4"
        region: "{{ vars.aws_region }}"
        accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"

      - id: push_1080p
        type: io.kestra.plugin.aws.s3.Upload
        description: Copy the 1080p variant into the CDN origin bucket under the public
          rendition key.
        from: "{{ outputs.transcode.outputFiles['variant_1080p.mp4'] }}"
        bucket: "{{ inputs.cdn_bucket }}"
        key: "{{ vars.cdn_prefix }}/variant_1080p.mp4"
        region: "{{ vars.aws_region }}"
        accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"

      - id: purge_cache
        type: io.kestra.plugin.cloudflare.cache.Purge
        description: Invalidate both rendition URLs in the Cloudflare zone so the edge
          fetches the new variants from origin.
        apiToken: "{{ secret('CLOUDFLARE_API_TOKEN') }}"
        zoneId: "{{ inputs.cloudflare_zone_id }}"
        purgeAll: false
        files:
          - "https://{{ inputs.cdn_host }}/{{ vars.cdn_prefix
            }}/variant_720p.mp4"
          - "https://{{ inputs.cdn_host }}/{{ vars.cdn_prefix
            }}/variant_1080p.mp4"

      - id: publish_report
        type: io.kestra.plugin.core.log.Log
        description: Record what was published and purged so the execution log doubles
          as a publish receipt.
        message: |
          Published s3://{{ inputs.cdn_bucket }}/{{ vars.cdn_prefix }}/variant_720p.mp4 and s3://{{ inputs.cdn_bucket }}/{{ vars.cdn_prefix }}/variant_1080p.mp4.
          QA verdict {{ outputs.qa_check.vars.verdict }} with {{ outputs.qa_check.vars.violations | length }} violations.
          Purged {{ inputs.cdn_host }} rendition URLs in Cloudflare zone {{ inputs.cloudflare_zone_id }}.
    else:
      - id: fail_qa_gate
        type: io.kestra.plugin.core.execution.Fail
        description: End the run as failed so a broken rendition never reaches the CDN
          origin bucket.
        errorMessage: "QA rejected the transcode output, {{
          outputs.qa_check.vars.violations | join(', ') }}"

outputs:
  - id: verdict
    type: STRING
    value: "{{ outputs.qa_check.vars.verdict }}"
  - id: violations
    type: JSON
    value: "{{ outputs.qa_check.vars.violations | toJson }}"
  - id: checked_variants
    type: INT
    value: "{{ outputs.qa_check.vars.checked_variants }}"
  - id: source_key
    type: STRING
    value: "{{ trigger.objects[0].key }}"
  - id: variant_720p
    type: STRING
    value: "{{ outputs.transcode.outputFiles['variant_720p.mp4'] }}"
  - id: variant_1080p
    type: STRING
    value: "{{ outputs.transcode.outputFiles['variant_1080p.mp4'] }}"
  - id: published_keys
    type: STRING
    value: "{{ vars.cdn_prefix }}/variant_720p.mp4, {{ vars.cdn_prefix
      }}/variant_1080p.mp4"

errors:
  - id: slack_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post a Slack alert when the transcode, the QA gate or the CDN
      publish step fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED {{ flow.id }} run {{ execution.id }}. The video transcode, QA gate or CDN publish step did not complete. Check the execution logs in the Kestra UI."}