> ## Documentation Index
> Fetch the complete documentation index at: https://docs.goldsky.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Turbo Pipelines (turbo)

## Manage Turbo Pipelines with "turbo"

`ctx.turbo` lets a task create and manage [Turbo pipelines](/turbo-pipelines/introduction) and
[jobs](/turbo-pipelines/job-mode) from inside your Compose app. Use it to spin up a pipeline in
response to an on-chain event, kick off a one-time backfill and poll it to completion, or pause and
clean up pipelines your app owns.

A Turbo **job** is a pipeline whose definition has `job: true` — it runs to completion and then exits,
rather than streaming forever. `ctx.turbo.createJob` is a shortcut for `createPipeline` with that flag
set.

### How pipelines are named and cleaned up

Pipelines you create through `ctx.turbo` are namespaced under your Compose app. You pass a plain name
like `token-transfers`, and the pipeline is registered internally with your app's prefix. Every method
handles the prefix for you, so you always refer to pipelines by the plain name you gave them, and
`listPipelines` only returns pipelines that belong to your app.

The upshot is that when your Compose app is deleted, the pipelines it created are deleted with it — you
don't have to track and tear them down yourself.

<Note>
  The namespaced name (app prefix included) has a length limit. If your app name plus the pipeline
  name is too long, `createPipeline` throws. Keep pipeline names short and the error message tells you
  the maximum.
</Note>

### Methods

```typescript theme={null}
turbo: {
  createPipeline(config: CreateTurboPipelineConfig, retryConfig?): Promise<TurboPipeline>;
  createJob(config: CreateTurboJobConfig, retryConfig?): Promise<TurboPipeline>;
  getPipeline(name: string, retryConfig?): Promise<TurboPipeline>;
  getStatus(name: string, retryConfig?): Promise<TurboPipelineStatus>;
  listPipelines(retryConfig?): Promise<TurboPipeline[]>;
  pausePipeline(name: string, retryConfig?): Promise<PipelineActionResult>;
  resumePipeline(name: string, retryConfig?): Promise<PipelineActionResult>;
  restartPipeline(name: string, options?: RestartOptions, retryConfig?): Promise<PipelineActionResult>;
  deletePipeline(name: string, retryConfig?): Promise<PipelineDeleteResult>;
  getLogs(name: string, options?: GetLogsOptions, retryConfig?): Promise<PipelineLogsResult>;
  getState(name: string, retryConfig?): Promise<unknown>;
  validate(definition: TurboPipelineDefinition, retryConfig?): Promise<ValidateResult>;
}
```

### Config types

```typescript theme={null}
export type TurboPipelineDefinition = {
  sources: Record<string, unknown>;
  transforms: Record<string, unknown>;
  sinks: Record<string, unknown>;
  job?: boolean;
};

export type CreateTurboPipelineConfig = {
  name: string; // lowercase letters, numbers, and hyphens
  resourceSize?: string;
  description?: string;
  useDedicatedIp?: boolean;
  definition: TurboPipelineDefinition;
};

// createJob takes the same config — the client sets job: true for you.
export type CreateTurboJobConfig = Omit<CreateTurboPipelineConfig, "definition"> & {
  definition: Omit<TurboPipelineDefinition, "job">;
};

export type RestartOptions = { clearState?: boolean };

export type GetLogsOptions = {
  logLevels?: string; // comma-separated, e.g. "error,warn"
  cursor?: number;
  after?: number;
  search?: string;
  direction?: "asc" | "desc";
};
```

### Response types

```typescript theme={null}
export type TurboPipeline = {
  name: string;
  status?: string;
  version?: number;
  definition?: TurboPipelineDefinition & {
    name?: string;
    resource_size?: string;
    description?: string;
    use_dedicated_ip?: boolean;
  };
  created_at: string;
  updated_at?: string;
};

export type TurboPipelineStatus = {
  name: string;
  status: string;
};

export type ValidateResult = {
  valid: boolean;
  errors?: Array<{ message: string } | string>;
  warnings?: Array<{ message: string } | string>;
};

export type PipelineActionResult = {
  name: string;
  action: string;
};

export type PipelineDeleteResult = {
  name: string;
  deleted: boolean;
};

export type PipelineLogsResult = {
  logs: unknown[];
};
```

The `definition` is the same [pipeline configuration](/turbo-pipelines/pipeline-config) you'd write in
YAML, expressed as an object. Reads (`getPipeline`, `getStatus`, `listPipelines`, `getLogs`,
`getState`, `validate`) retry on transient failures by default; the mutating methods do not. Every call
is recorded as a span on the task run, like [`ctx.fetch`](./fetch).

### Examples

#### Kick off a one-time backfill (job)

```typescript theme={null}
import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const job = await turbo.createJob({
    name: "solana-backfill",
    resourceSize: "m",
    definition: {
      sources: {
        token_transfers: {
          type: "dataset",
          dataset_name: "solana.token_transfers",
          version: "1.0.0",
          start_block: 250000000,
          end_block: 250000002,
        },
      },
      transforms: {
        filtered: { type: "sql", primary_key: "id", sql: "SELECT * FROM token_transfers" },
      },
      sinks: {
        out: { type: "postgres", from: "filtered", schema: "public", table: "token_transfers", secret_name: "MY_PG", primary_key: "id" },
      },
    },
  });

  return job;
}
```

#### Create a streaming pipeline in response to an event

```typescript theme={null}
import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext, params: { contract: string }) {
  // Sanitize the contract address for use in the pipeline name
  const slug = params.contract.slice(2, 10).toLowerCase();

  return turbo.createPipeline({
    name: `transfers-${slug}`,
    definition: {
      sources: {
        logs: { type: "dataset", dataset_name: "ethereum.raw_logs", version: "1.0.0", start_at: "latest" },
      },
      transforms: {
        filtered: {
          type: "sql",
          primary_key: "id",
          sql: `SELECT * FROM logs WHERE address = lower('${params.contract}')`,
        },
      },
      sinks: {
        out: { type: "postgres", from: "filtered", schema: "public", table: "transfers", secret_name: "MY_PG", primary_key: "id" },
      },
    },
  });
}
```

#### Poll a job until it finishes

```typescript theme={null}
import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const { status } = await turbo.getStatus("solana-backfill");

  if (status === "STOPPED" || status === "FAILED") {
    const logs = await turbo.getLogs("solana-backfill", { logLevels: "error" });
    return { done: true, status, logs };
  }

  return { done: false, status };
}
```

<Tip>
  Pair this with a [cron trigger](../task-triggers) to check on a long-running job on a schedule
  instead of blocking a single task run.
</Tip>

#### Validate before deploying

```typescript theme={null}
import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const result = await turbo.validate({
    sources: { blocks: { type: "dataset", dataset_name: "ethereum.raw_blocks", version: "1.0.0" } },
    transforms: { t: { type: "sql", primary_key: "id", sql: "SELECT * FROM blocks" } },
    sinks: { out: { type: "blackhole", from: "t" } },
  });

  if (!result.valid) {
    const messages = (result.errors ?? []).map((e) =>
      typeof e === "string" ? e : e.message
    );
    throw new Error(messages.join("; "));
  }
}
```

#### List, pause, and delete the pipelines your app owns

```typescript theme={null}
import { TaskContext } from "compose";

export async function main({ turbo }: TaskContext) {
  const pipelines = await turbo.listPipelines();

  for (const p of pipelines) {
    await turbo.pausePipeline(p.name);
  }

  await turbo.deletePipeline("solana-backfill");
}
```

## Next Steps

<CardGroup cols={2}>
  <Card title="Turbo Pipelines" icon="bolt" href="/turbo-pipelines/introduction">
    Learn how Turbo pipelines, sources, transforms, and sinks work.
  </Card>

  <Card title="Job mode" icon="list-check" href="/turbo-pipelines/job-mode">
    Run a pipeline as a one-time job that exits when it's done.
  </Card>
</CardGroup>


## Related topics

- [Deploy a Turbo pipeline](/turbo-pipelines/quickstart.md)
- [Turbo Pipeline Configuration](/turbo-pipelines/pipeline-config.md)
- [Mirror vs. Turbo pipelines](/mirror-vs-turbo.md)
- [Migrate from Mirror to Turbo](/turbo-pipelines/migrate-from-mirror.md)
- [Turbo CLI Reference](/turbo-pipelines/cli-reference.md)