1import * as appflow from "@distilled.cloud/aws/appflow";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import { readAppFlowTags, syncAppFlowTags } from "./internal.ts";
12
13export interface FlowProps {
14 /**
15 * The name of the flow. Must be unique in the account/region, may contain
16 * only letters, digits, hyphens, underscores, `!`, `@`, `#`, `.`, `$`.
17 * If omitted, a deterministic physical name is generated. Changing the
18 * name replaces the flow.
19 */
20 flowName?: string;
21 /**
22 * A description of the flow.
23 */
24 description?: string;
25 /**
26 * The ARN of a KMS key AppFlow uses to encrypt data. If omitted, the
27 * AWS-managed AppFlow key is used.
28 */
29 kmsArn?: string;
30 /**
31 * How the flow is triggered — on demand, on a schedule, or by an event.
32 * @default { triggerType: "OnDemand" }
33 */
34 triggerConfig?: appflow.TriggerConfig;
35 /**
36 * Configuration of the source connector (e.g. S3, Salesforce) the flow
37 * pulls data from.
38 */
39 sourceFlowConfig: appflow.SourceFlowConfig;
40 /**
41 * Configuration of the destination connector(s) the flow pushes data to.
42 */
43 destinationFlowConfigList: appflow.DestinationFlowConfig[];
44 /**
45 * The field mapping tasks applied to records as they move from source to
46 * destination.
47 */
48 tasks: appflow.Task[];
49 /**
50 * Glue Data Catalog configuration for cataloging flow output.
51 */
52 metadataCatalogConfig?: appflow.MetadataCatalogConfig;
53 /**
54 * User-defined tags for the flow.
55 */
56 tags?: Record<string, string>;
57}
58
59export interface Flow extends Resource<
60 "AWS.AppFlow.Flow",
61 FlowProps,
62 {
63 flowName: string;
64 flowArn: string;
65 flowStatus: string | undefined;
66 },
67 never,
68 Providers
69> {}
70
71/**
72 * An Amazon AppFlow flow. A flow transfers data between a source connector
73 * and one or more destination connectors, applying field-mapping tasks.
74 * The credential-free S3-to-S3 path is directly testable; other connectors
75 * require a {@link ConnectorProfile} with vendor credentials.
76 *
77 * For an S3 source, the bucket policy must authorize the
78 * `appflow.amazonaws.com` service principal (`s3:GetObject` + `s3:ListBucket`
79 * on the source; `s3:PutObject` and the multipart/ACL actions on the
80 * destination), and the source prefix must contain at least one object when
81 * the flow is created — AppFlow validates connectivity by listing the source
82 * at `CreateFlow` time and rejects an empty prefix with
83 * `ConnectorServerException`.
84 * ### Creating a Flow
85 * **Example:** S3 to S3 On-Demand Flow
86 * ```typescript
87 * const flow = yield* AppFlow.Flow("Copy", {
88 * triggerConfig: { triggerType: "OnDemand" },
89 * sourceFlowConfig: {
90 * connectorType: "S3",
91 * sourceConnectorProperties: {
92 * S3: { bucketName: srcBucket, bucketPrefix: "input" },
93 * },
94 * },
95 * destinationFlowConfigList: [
96 * {
97 * connectorType: "S3",
98 * destinationConnectorProperties: {
99 * S3: { bucketName: dstBucket, bucketPrefix: "output" },
100 * },
101 * },
102 * ],
103 * tasks: [
104 * {
105 * taskType: "Map_all",
106 * sourceFields: [],
107 * connectorOperator: { S3: "NO_OP" },
108 * taskProperties: {},
109 * },
110 * ],
111 * });
112 * ```
113 *
114 * @resource
115 */
116export const Flow = Resource<Flow>("AWS.AppFlow.Flow");
117
118const ON_DEMAND: appflow.TriggerConfig = { triggerType: "OnDemand" };
119
120export const FlowProvider = () =>
121 Provider.effect(
122 Flow,
123 Effect.gen(function* () {
124 const toName = (id: string, props: FlowProps) =>
125 props.flowName
126 ? Effect.succeed(props.flowName)
127 : createPhysicalName({ id, maxLength: 100 });
128
129 return {
130 stables: ["flowName", "flowArn"],
131
132 diff: Effect.fn(function* ({ id, olds, news }) {
133 if (!isResolved(news)) return undefined;
134 if ((yield* toName(id, olds)) !== (yield* toName(id, news))) {
135 return { action: "replace" } as const;
136 }
137 }),
138
139 read: Effect.fn(function* ({ id, olds, output }) {
140 const name = output?.flowName ?? (yield* toName(id, olds ?? {}));
141 const flow = yield* appflow
142 .describeFlow({ flowName: name })
143 .pipe(
144 Effect.catchTag("ResourceNotFoundException", () =>
145 Effect.succeed(undefined),
146 ),
147 );
148 if (flow === undefined || flow.flowArn === undefined)
149 return undefined;
150 const attrs = {
151 flowName: name,
152 flowArn: flow.flowArn,
153 flowStatus: flow.flowStatus,
154 };
155 const tags = yield* readAppFlowTags(flow.flowArn);
156 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
157 }),
158
159 reconcile: Effect.fn(function* ({ id, news, output, session }) {
160 const name = output?.flowName ?? (yield* toName(id, news));
161 const internalTags = yield* createInternalTags(id);
162 const desiredTags = { ...internalTags, ...news.tags };
163 const trigger = news.triggerConfig ?? ON_DEMAND;
164
165 // 1. Observe.
166 let live = yield* appflow
167 .describeFlow({ flowName: name })
168 .pipe(
169 Effect.catchTag("ResourceNotFoundException", () =>
170 Effect.succeed(undefined),
171 ),
172 );
173
174 // 2. Ensure — create if missing. Tolerate an AlreadyExists race by
175 // falling back to describe.
176 if (live === undefined) {
177 yield* appflow
178 .createFlow({
179 flowName: name,
180 description: news.description,
181 kmsArn: news.kmsArn,
182 triggerConfig: trigger,
183 sourceFlowConfig: news.sourceFlowConfig,
184 destinationFlowConfigList: news.destinationFlowConfigList,
185 tasks: news.tasks,
186 metadataCatalogConfig: news.metadataCatalogConfig,
187 tags: desiredTags,
188 })
189 .pipe(Effect.catchTag("ConflictException", () => Effect.void));
190 live = yield* appflow.describeFlow({ flowName: name });
191 } else {
192 // 3. Sync — updateFlow converges the mutable configuration
193 // (trigger, source/destination, tasks). It does not accept tags.
194 yield* appflow.updateFlow({
195 flowName: name,
196 description: news.description,
197 triggerConfig: trigger,
198 sourceFlowConfig: news.sourceFlowConfig,
199 destinationFlowConfigList: news.destinationFlowConfigList,
200 tasks: news.tasks,
201 metadataCatalogConfig: news.metadataCatalogConfig,
202 });
203 live = yield* appflow.describeFlow({ flowName: name });
204 }
205
206 // 3b. Sync tags against observed cloud tags.
207 if (live.flowArn) {
208 yield* syncAppFlowTags(live.flowArn, desiredTags);
209 }
210
211 yield* session.note(name);
212 return {
213 flowName: name,
214 flowArn: live.flowArn!,
215 flowStatus: live.flowStatus,
216 };
217 }),
218
219 delete: Effect.fn(function* ({ output }) {
220 yield* appflow
221 .deleteFlow({ flowName: output.flowName, forceDelete: true })
222 .pipe(
223 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
224 );
225 }),
226
227 list: () =>
228 appflow.listFlows.pages({}).pipe(
229 Stream.runCollect,
230 Effect.map((chunk) =>
231 Array.from(chunk).flatMap((page) =>
232 (page.flows ?? [])
233 .filter(
234 (
235 f,
236 ): f is appflow.FlowDefinition & {
237 flowName: string;
238 flowArn: string;
239 } => f.flowName !== undefined && f.flowArn !== undefined,
240 )
241 .map((f) => ({
242 flowName: f.flowName,
243 flowArn: f.flowArn,
244 flowStatus: f.flowStatus,
245 })),
246 ),
247 ),
248 ),
249 };
250 }),
251 );