Pipe

Source: src/AWS/Pipes/Pipe.ts

An Amazon EventBridge Pipe — point-to-point source→(filter)→(enrich)→target plumbing between AWS services without glue code.

Pipe owns the lifecycle of an EventBridge Pipe. Reconcile waits (bounded) for the pipe to leave its CREATING/UPDATING transitional states, and a pipe that lands in a *_FAILED state surfaces as a typed PipeFailed error rather than hanging. Prefer the from builder for the common pairs — it synthesizes the pipes.amazonaws.com execution role with source-read and target-invoke policies for you.

Creating Pipes

SQS to Lambda (builder — role synthesized automatically)

import * as AWS from "alchemy/AWS";

const queue = yield* AWS.SQS.Queue("OrdersQueue");
const pipe = yield* AWS.Pipes.from(queue, { batchSize: 1 }).toLambda(fn);

SQS to SQS (canonical resource with an explicit role)

const pipe = yield* AWS.Pipes.Pipe("OrdersPipe", {
  source: source.queueArn,
  target: target.queueArn,
  roleArn: role.roleArn,
  sourceParameters: {
    SqsQueueParameters: { BatchSize: 1 },
  },
});

Filtering

const pipe = yield* AWS.Pipes.from(queue)
  .filter(JSON.stringify({ body: { type: ["order.created"] } }))
  .toLambda(fn);

Enrichment

const pipe = yield* AWS.Pipes.from(queue)
  .enrich(enricherFn)
  .toQueue(target);

Stream Sources

Kinesis stream source

const pipe = yield* AWS.Pipes.from(stream, {
  startingPosition: "TRIM_HORIZON",
  batchSize: 10,
}).toLambda(fn);

Stop a pipe without deleting it

const pipe = yield* AWS.Pipes.Pipe("OrdersPipe", {
  source: source.queueArn,
  target: target.queueArn,
  roleArn: role.roleArn,
  desiredState: "STOPPED",
});