1import type * as pipes from "@distilled.cloud/aws/pipes";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import { toWireSeconds } from "../../Util/Duration.ts";
5import type { Table } from "../DynamoDB/Table.ts";
6import * as IAM from "../IAM/index.ts";
7import type { Stream } from "../Kinesis/Stream.ts";
8import type { Function } from "../Lambda/Function.ts";
9import type { Queue } from "../SQS/Queue.ts";
10import { Pipe, type PipeDesiredState } from "./Pipe.ts";
11
12/**
13 * Resources the {@link from} builder accepts as a pipe source. DynamoDB
14 * tables must have streams enabled (`latestStreamArn` defined).
15 */
16export type PipeSource = Queue | Stream | Table;
17
18export interface PipeSourceOptions {
19 /**
20 * Maximum number of source records per delivered batch.
21 */
22 batchSize?: number;
23 /**
24 * Maximum time to gather records before delivering a batch (e.g.
25 * `"30 seconds"` or `Duration.seconds(30)`). Sent to the API in whole
26 * seconds.
27 */
28 maximumBatchingWindow?: Duration.Input;
29 /**
30 * Where to start reading a Kinesis or DynamoDB stream source. Ignored
31 * for SQS sources. Changing it triggers a replacement.
32 * @default "LATEST"
33 */
34 startingPosition?: "TRIM_HORIZON" | "LATEST";
35}
36
37export interface PipeTargetOptions {
38 /**
39 * Explicit physical pipe name. Also used as the logical ID, so it must
40 * be deterministic. If omitted, a logical ID is derived from the source
41 * and target logical IDs and the physical name is auto-generated.
42 */
43 name?: string;
44 /**
45 * A description of the pipe.
46 */
47 description?: string;
48 /**
49 * The state the pipe should be in.
50 * @default "RUNNING"
51 */
52 desiredState?: PipeDesiredState;
53 /**
54 * Input template applied to each record before delivery to the target.
55 */
56 inputTemplate?: string;
57}
58
59export interface LambdaTargetOptions extends PipeTargetOptions {
60 /**
61 * How the target Lambda function is invoked.
62 * @default "REQUEST_RESPONSE"
63 */
64 invocationType?: "REQUEST_RESPONSE" | "FIRE_AND_FORGET";
65}
66
67export interface QueueTargetOptions extends PipeTargetOptions {
68 /**
69 * Message group ID for FIFO target queues.
70 */
71 messageGroupId?: string;
72 /**
73 * Message deduplication ID for FIFO target queues.
74 */
75 messageDeduplicationId?: string;
76}
77
78interface PipeBuilderState {
79 source: PipeSource;
80 options: PipeSourceOptions;
81 filters: string[];
82 enrichment?: {
83 fn: Function;
84 parameters?: pipes.PipeEnrichmentParameters;
85 };
86}
87
88/**
89 * Start building an EventBridge Pipe from a source resource. The terminal
90 * `.toLambda(...)` / `.toQueue(...)` call synthesizes the
91 * `pipes.amazonaws.com` execution role — source-read plus target-invoke
92 * (plus enrichment-invoke) policies scoped to the exact resource ARNs —
93 * and yields the {@link Pipe}.
94 *
95 * ```typescript
96 * const pipe = yield* AWS.Pipes.from(queue, { batchSize: 1 })
97 * .filter(JSON.stringify({ body: { type: ["order.created"] } }))
98 * .toLambda(fn);
99 * ```
100 */
101export const from = (source: PipeSource, options: PipeSourceOptions = {}) =>
102 makePipeBuilder({ source, options, filters: [] });
103
104const makePipeBuilder = (state: PipeBuilderState) => ({
105 /**
106 * Add EventBridge event-pattern filters. Only source events matching at
107 * least one pattern reach the enrichment/target. Accepts pattern strings
108 * or plain objects (JSON-stringified).
109 */
110 filter: (...patterns: (string | Record<string, unknown>)[]) =>
111 makePipeBuilder({
112 ...state,
113 filters: [
114 ...state.filters,
115 ...patterns.map((pattern) =>
116 typeof pattern === "string" ? pattern : JSON.stringify(pattern),
117 ),
118 ],
119 }),
120
121 /**
122 * Enrich each batch with a Lambda function before delivery to the
123 * target. The synthesized role is granted `lambda:InvokeFunction` on the
124 * enrichment function.
125 */
126 enrich: (fn: Function, parameters?: pipes.PipeEnrichmentParameters) =>
127 makePipeBuilder({ ...state, enrichment: { fn, parameters } }),
128
129 /**
130 * Deliver batches to a Lambda function target.
131 */
132 toLambda: (fn: Function, options: LambdaTargetOptions = {}) =>
133 materializePipe(
134 state,
135 fn.LogicalId,
136 [
137 {
138 Effect: "Allow",
139 Action: ["lambda:InvokeFunction"],
140 Resource: [fn.functionArn],
141 },
142 ],
143 fn.functionArn,
144 {
145 InputTemplate: options.inputTemplate,
146 LambdaFunctionParameters: options.invocationType
147 ? { InvocationType: options.invocationType }
148 : undefined,
149 },
150 options,
151 ),
152
153 /**
154 * Deliver batches to an SQS queue target.
155 */
156 toQueue: (queue: Queue, options: QueueTargetOptions = {}) =>
157 materializePipe(
158 state,
159 queue.LogicalId,
160 [
161 {
162 Effect: "Allow",
163 Action: ["sqs:SendMessage"],
164 Resource: [queue.queueArn],
165 },
166 ],
167 queue.queueArn,
168 {
169 InputTemplate: options.inputTemplate,
170 SqsQueueParameters:
171 options.messageGroupId !== undefined ||
172 options.messageDeduplicationId !== undefined
173 ? {
174 MessageGroupId: options.messageGroupId,
175 MessageDeduplicationId: options.messageDeduplicationId,
176 }
177 : undefined,
178 },
179 options,
180 ),
181});
182
183interface SourceSpec {
184 arn: unknown;
185 statements: unknown[];
186 parameters: pipes.PipeSourceParameters;
187}
188
189const sourceSpec = (state: PipeBuilderState): SourceSpec => {
190 const { source, options, filters } = state;
191 // The wire field is whole seconds.
192 const maximumBatchingWindowInSeconds = toWireSeconds(
193 options.maximumBatchingWindow,
194 );
195 const filterCriteria: Pick<pipes.PipeSourceParameters, "FilterCriteria"> =
196 filters.length > 0
197 ? { FilterCriteria: { Filters: filters.map((Pattern) => ({ Pattern })) } }
198 : {};
199 switch (source.Type) {
200 case "AWS.SQS.Queue":
201 return {
202 arn: source.queueArn,
203 statements: [
204 {
205 Effect: "Allow",
206 Action: [
207 "sqs:ReceiveMessage",
208 "sqs:DeleteMessage",
209 "sqs:GetQueueAttributes",
210 ],
211 Resource: [source.queueArn],
212 },
213 ],
214 parameters: {
215 ...filterCriteria,
216 SqsQueueParameters: {
217 BatchSize: options.batchSize,
218 MaximumBatchingWindowInSeconds: maximumBatchingWindowInSeconds,
219 },
220 },
221 };
222 case "AWS.Kinesis.Stream":
223 return {
224 arn: source.streamArn,
225 statements: [
226 {
227 Effect: "Allow",
228 Action: [
229 "kinesis:DescribeStream",
230 "kinesis:DescribeStreamSummary",
231 "kinesis:GetRecords",
232 "kinesis:GetShardIterator",
233 "kinesis:ListShards",
234 "kinesis:ListStreams",
235 ],
236 Resource: [source.streamArn],
237 },
238 ],
239 parameters: {
240 ...filterCriteria,
241 KinesisStreamParameters: {
242 StartingPosition: options.startingPosition ?? "LATEST",
243 BatchSize: options.batchSize,
244 MaximumBatchingWindowInSeconds: maximumBatchingWindowInSeconds,
245 },
246 },
247 };
248 case "AWS.DynamoDB.Table":
249 return {
250 arn: source.latestStreamArn,
251 statements: [
252 {
253 Effect: "Allow",
254 Action: [
255 "dynamodb:DescribeStream",
256 "dynamodb:GetRecords",
257 "dynamodb:GetShardIterator",
258 "dynamodb:ListStreams",
259 ],
260 Resource: [source.latestStreamArn],
261 },
262 ],
263 parameters: {
264 ...filterCriteria,
265 DynamoDBStreamParameters: {
266 StartingPosition: options.startingPosition ?? "LATEST",
267 BatchSize: options.batchSize,
268 MaximumBatchingWindowInSeconds: maximumBatchingWindowInSeconds,
269 },
270 },
271 };
272 }
273};
274
275const materializePipe = (
276 state: PipeBuilderState,
277 targetId: string,
278 targetStatements: unknown[],
279 targetArn: unknown,
280 targetParameters: pipes.PipeTargetParameters,
281 options: PipeTargetOptions,
282) =>
283 Effect.gen(function* () {
284 const src = sourceSpec(state);
285 const pipeId = options.name ?? `${state.source.LogicalId}To${targetId}Pipe`;
286
287 const enrichmentStatements = state.enrichment
288 ? [
289 {
290 Effect: "Allow",
291 Action: ["lambda:InvokeFunction"],
292 Resource: [state.enrichment.fn.functionArn],
293 },
294 ]
295 : [];
296
297 const role = yield* IAM.Role(`${pipeId}Role`, {
298 assumeRolePolicyDocument: {
299 Version: "2012-10-17",
300 Statement: [
301 {
302 Effect: "Allow",
303 Principal: { Service: "pipes.amazonaws.com" },
304 Action: ["sts:AssumeRole"],
305 },
306 ],
307 },
308 inlinePolicies: {
309 PipeSource: {
310 Version: "2012-10-17",
311 Statement: src.statements as any,
312 },
313 PipeTarget: {
314 Version: "2012-10-17",
315 Statement: [...targetStatements, ...enrichmentStatements] as any,
316 },
317 },
318 });
319
320 return yield* Pipe(pipeId, {
321 pipeName: options.name,
322 description: options.description,
323 desiredState: options.desiredState,
324 source: src.arn as any,
325 sourceParameters: src.parameters,
326 enrichment: state.enrichment
327 ? (state.enrichment.fn.functionArn as any)
328 : undefined,
329 enrichmentParameters: state.enrichment?.parameters,
330 target: targetArn as any,
331 targetParameters,
332 roleArn: role.roleArn,
333 });
334 });