Pipelines

A pipeline is a graph of typed nodes. One input can fan out to several outputs; each output is its own job.

Pipeline structure

Under the hood that graph is a DAG (directed acyclic graph): nodes wired by edges, no loops. You write it as { nodes, edges }. Nodes reference a type and hold config; edges wire them together. Every graph starts at a pipeline.input and ends at one or more pipeline.output nodes. From here on we'll say graph.

lib/pipelines.ts

import type { Pipeline } from '@transform-kit/sdk';

// Resize to fit 1600px, encode as WebP, emit one output.
export const webPipeline: Pipeline = {
  nodes: [
    { id: 'in', type: 'pipeline.input' },
    {
      id: 'resize',
      type: 'image.resize',
      config: {
        mode: { value: 'pixels' },
        width: { value: 1600 },
        height: { value: 1600 },
        fit: { value: 'contain' },
      },
    },
    {
      id: 'convert',
      type: 'image.convert',
      config: {
        format: { value: 'webp' },
        quality: { value: 82 },
      },
    },
    {
      id: 'out',
      type: 'pipeline.output',
      config: { suffix: { value: 'web' } },
    },
  ],
  edges: [
    { source: 'in', target: 'resize' },
    { source: 'resize', target: 'convert' },
    { source: 'convert', target: 'out' },
  ],
};

Config is { value, editable }

Each config field is an object rather than a bare value. Use mergePipelineNodeConfig to fill omitted fields from the node defaults.

Multiple outputs

The server splits your graph into a self-contained job per pipeline.output, so a multi-output pipeline fans out in a single submit. Each output runs independently. Over a batch, M files × N outputs produces M × N results.

lib/pipelines.ts

import type { Pipeline } from '@transform-kit/sdk';

// One input → two outputs (a thumbnail and a hero). Two jobs per file.
export const responsivePipeline: Pipeline = {
  nodes: [
    { id: 'in', type: 'pipeline.input' },

    // Thumbnail branch, square crop, small WebP.
    { id: 'thumb-resize', type: 'image.resize', config: {
      mode: { value: 'pixels' },
      width: { value: 320 },
      height: { value: 320 },
      fit: { value: 'cover' },
    } },
    { id: 'thumb-convert', type: 'image.convert', config: {
      format: { value: 'webp' },
      quality: { value: 80 },
    } },
    { id: 'thumb-out', type: 'pipeline.output', config: {
      suffix: { value: 'thumb' },
    } },

    // Hero branch, larger WebP, fit inside 1600px.
    { id: 'hero-resize', type: 'image.resize', config: {
      mode: { value: 'pixels' },
      width: { value: 1600 },
      height: { value: 1600 },
      fit: { value: 'contain' },
    } },
    { id: 'hero-convert', type: 'image.convert', config: {
      format: { value: 'webp' },
      quality: { value: 82 },
    } },
    { id: 'hero-out', type: 'pipeline.output', config: {
      suffix: { value: 'hero' },
    } },
  ],
  edges: [
    { source: 'in', target: 'thumb-resize' },
    { source: 'thumb-resize', target: 'thumb-convert' },
    { source: 'thumb-convert', target: 'thumb-out' },

    { source: 'in', target: 'hero-resize' },
    { source: 'hero-resize', target: 'hero-convert' },
    { source: 'hero-convert', target: 'hero-out' },
  ],
};

Default node configuration

mergePipelineNodeConfigfills a node's config from the type defaults, so you only pass what you want to change.

lib/pipelines.ts

import { mergePipelineNodeConfig, type Pipeline } from '@transform-kit/sdk';

const convert = {
  id: 'convert',
  type: 'image.convert',
  config: mergePipelineNodeConfig('image.convert', { format: 'webp', quality: 82 }),
};

Run a pipeline

Give runPipeline files and a graph. It validates, uploads, submits, and awaits every output. Results remain in input order, and one failed file does not stop the batch.

batch.ts

const results = await tk.runPipeline(
  [file1, file2, file3], // bytes, { bytes, filename }, or a lazy loader
  responsivePipeline,
  {
    concurrency: 6, // files in flight at once (default 6)
    onProgress: (e) => console.log(e.stage, e.filename ?? e.index),
  },
);

for (const r of results) {
  if (!r.ok) {
    console.error(r.filename, r.error);
    continue;
  }
  for (const out of r.outputs) {
    console.log(r.filename, out.output, out.media.url);
  }
}

Each result is one input's outcome:

type PipelineRunResult = {
  index: number;        // position in the input array
  filename?: string;
  ok: boolean;          // false isolates this file's failure
  outputs: { job_id: string; output?: string; media: { url: string; /* … */ } }[];
  error?: string;
};

Lazy file loading

A file can be a function that returns its bytes, it is only called when the file reaches an open slot, so a 500-file run never holds 500 files in memory at once. This is where a server reads from disk without pulling node:fs into the SDK.

loaders.ts

import { readFile } from 'node:fs/promises';
import { basename } from 'node:path';

const paths = ['a.png', 'b.jpg', 'c.webp'];

await tk.runPipeline(
  paths.map((p) => async () => ({ bytes: await readFile(p), filename: basename(p) })),
  webPipeline,
  { concurrency: 8 },
);

Pipeline validation

runPipeline validates locally before any upload and throws invalid_pipeline if the graph is malformed. To check a graph you built yourself (missing input/output, a cycle, a duplicate id, an unknown node type), call validatePipeline directly. Pass ALLOWED_NODE_TYPES so any node outside the supported set is flagged.

import { validatePipeline, ALLOWED_NODE_TYPES } from '@transform-kit/sdk';

const { valid, errors } = validatePipeline(myPipeline, ALLOWED_NODE_TYPES);
if (!valid) console.error(errors); // [{ code: 'NO_OUTPUT', message: '…' }]

Node reference

Every graph starts at one pipeline.input and ends at one or more pipeline.output nodes. This table comes from the SDK node registry; each type links to its full reference page.

Node	Config defaults and options
`pipeline.input`	None
`pipeline.output`	`suffix` =
`image.resize`	`mode` = `percentage` (percentage \| pixels \| maxSize); `percent` = `50`; `width` = `1920`; `height` = `1080`; `fit` = `contain` (contain \| cover \| fill \| inside \| outside); `maxSize` = `2048`
`image.filter`	`format` = `jpg` (png \| jpg \| webp \| gif \| avif \| tiff)
`image.convert`	`format` = `jpg` (png \| jpg \| webp \| gif \| avif \| tiff); `quality` = `90`
`video.resize`	`mode` = `percentage` (percentage \| pixels); `percent` = `50`; `width` = `1280`; `height` = `720`; `fit` = `contain` (contain \| cover \| fill \| inside \| outside); `padColor` = `black`
`video.filter`	`format` = `mp4` (mp4 \| webm \| mov \| mkv)
`video.convert`	`format` = `mp4` (mp4 \| webm \| mov \| mkv); `quality` = `75`
`audio.filter`	`format` = `mp3` (mp3 \| m4a \| opus \| flac \| wav)
`audio.convert`	`format` = `mp3` (mp3 \| m4a \| opus \| flac \| wav); `bitrate` = `192`
`document.filter`	`format` = `txt` (txt \| md)
`document.convert`	`format` = `txt` (txt \| md)

A pipeline must use a single media type, don't mix node families in the same graph. Filter nodes compare the input container; a mismatch passes through unchanged.