1import * as iam from "@distilled.cloud/aws/iam";
2import * as sfn from "@distilled.cloud/aws/sfn";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Redacted from "effect/Redacted";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import * as Provider from "../../Provider.ts";
12import { Resource, type ResourceBinding } from "../../Resource.ts";
13import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
14import { AWSEnvironment } from "../Environment.ts";
15import type { PolicyStatement } from "../IAM/Policy.ts";
16import type { Providers } from "../Providers.ts";
17import { compileProgram, SfnCompileError } from "./Asl/compile.ts";
18import type { SfnEffect } from "./Asl/Program.ts";
19
20/**
21 * The workflow type. `STANDARD` workflows are durable (up to one year),
22 * exactly-once, and support execution history; `EXPRESS` workflows are
23 * high-throughput, at-least-once, cheaper, and support synchronous
24 * invocation via `StartSyncExecution`.
25 */
26export type StateMachineType = "STANDARD" | "EXPRESS";
27
28/**
29 * Log level for a state machine's CloudWatch Logs configuration.
30 */
31export type StateMachineLogLevel = "ALL" | "ERROR" | "FATAL" | "OFF";
32
33export interface StateMachineLogging {
34 /**
35 * Which execution history events are logged.
36 */
37 level: StateMachineLogLevel;
38 /**
39 * Whether execution input, data passed between states, and output are
40 * included in the log events.
41 * @default false
42 */
43 includeExecutionData?: boolean;
44 /**
45 * CloudWatch log group ARNs to deliver logs to. The `:*` suffix is
46 * appended automatically when missing (the API requires it).
47 */
48 destinations?: string[];
49}
50
51export interface StateMachineProps {
52 /**
53 * Name of the state machine (1-80 characters; letters, digits, dashes and
54 * underscores). If omitted, a deterministic physical name is generated
55 * from the app, stage, and logical ID. Changing the name triggers a
56 * replacement.
57 */
58 stateMachineName?: string;
59 /**
60 * The Amazon States Language (ASL) definition of the workflow. Accepts a
61 * plain object (recommended — `Output` values such as function ARNs are
62 * resolved before serialization) or a pre-serialized JSON string.
63 */
64 definition: Record<string, unknown> | string;
65 /**
66 * Literal substitutions applied to the serialized definition: every
67 * occurrence of `${key}` is replaced with the mapped value. Useful when
68 * the definition is authored as a static JSON string.
69 */
70 definitionSubstitutions?: Record<string, string>;
71 /**
72 * The ARN of an existing IAM role for the state machine to use. When
73 * omitted, an execution role is created automatically with
74 * `states.amazonaws.com` trust; `lambda:InvokeFunction` is granted for
75 * every Lambda function ARN referenced in the definition, and
76 * {@link StateMachineProps.policyStatements} are attached as an inline
77 * policy.
78 */
79 roleArn?: string;
80 /**
81 * Additional IAM policy statements attached to the auto-created
82 * execution role (ignored when {@link StateMachineProps.roleArn} is
83 * provided). Use this to authorize service integrations such as
84 * `sqs:SendMessage` or `dynamodb:PutItem` task states.
85 */
86 policyStatements?: PolicyStatement[];
87 /**
88 * The workflow type. Changing the type triggers a replacement.
89 * @default "STANDARD"
90 */
91 type?: StateMachineType;
92 /**
93 * CloudWatch Logs configuration. Required (with level `ALL`/`ERROR`/
94 * `FATAL`) to observe `EXPRESS` workflow executions. The auto-created
95 * execution role is granted the CloudWatch Logs delivery permissions
96 * automatically.
97 */
98 logging?: StateMachineLogging;
99 /**
100 * Whether AWS X-Ray tracing is enabled.
101 * @default false
102 */
103 tracingEnabled?: boolean;
104 /**
105 * Tags to apply to the state machine. Merged with internal Alchemy tags.
106 */
107 tags?: Record<string, string>;
108}
109
110export interface StateMachine extends Resource<
111 "AWS.StepFunctions.StateMachine",
112 StateMachineProps,
113 {
114 /**
115 * Physical name of the state machine.
116 */
117 stateMachineName: string;
118 /**
119 * ARN of the state machine — pass it to `StartExecution` or reference it
120 * from other workflows.
121 */
122 stateMachineArn: string;
123 /**
124 * Workflow type (`STANDARD` or `EXPRESS`).
125 */
126 type: StateMachineType;
127 /**
128 * ARN of the IAM execution role the workflow assumes.
129 */
130 roleArn: string;
131 /**
132 * Name of the auto-created execution role. `undefined` when an explicit
133 * {@link StateMachineProps.roleArn} is used.
134 */
135 roleName: string | undefined;
136 },
137 {
138 /**
139 * IAM policy statements bindings attach to the auto-created execution
140 * role (e.g. service-integration permissions collected by `fromProgram`).
141 */
142 policyStatements: PolicyStatement[];
143 },
144 Providers
145> {}
146
147/**
148 * The definition failed AWS's `validateStateMachineDefinition` pre-flight
149 * check (run in reconcile before create/update). Carries the validator's
150 * ERROR-severity diagnostics.
151 */
152export class InvalidStateMachineDefinition extends Data.TaggedError(
153 "InvalidStateMachineDefinition",
154)<{
155 readonly diagnostics: readonly {
156 readonly severity: string;
157 readonly code: string;
158 readonly message: string;
159 readonly location: string | undefined;
160 }[];
161}> {}
162
163const StateMachineResource = Resource<StateMachine>(
164 "AWS.StepFunctions.StateMachine",
165);
166
167/**
168 * Props for {@link StateMachine.fromProgram} — everything a raw
169 * `StateMachine` accepts except the `definition` (which is compiled from
170 * the program). User `policyStatements` are merged with the statements the
171 * compiler collects from `Sfn.invoke`/`Sfn.integrate` task states.
172 */
173export interface FromProgramProps extends Omit<
174 StateMachineProps,
175 "definition" | "definitionSubstitutions"
176> {
177 /** The typed Step Functions program to compile (built with `Sfn.gen`). */
178 program: SfnEffect<any, any>;
179}
180
181/**
182 * Sugar over the raw `StateMachine` resource: compile a typed `Sfn` program
183 * to a plain ASL definition object plus collected IAM policy statements,
184 * then register the same `StateMachine` resource the raw path uses. Nothing
185 * engine-level — the raw `definition` escape hatch stays first-class.
186 */
187const fromProgram = (id: string, props: FromProgramProps) =>
188 Effect.gen(function* () {
189 const { program, policyStatements, ...rest } = props;
190 const compiled = yield* Effect.try({
191 try: () => compileProgram(program),
192 catch: (error) =>
193 error instanceof SfnCompileError
194 ? error
195 : new SfnCompileError({ message: String(error) }),
196 });
197 return yield* StateMachineResource(id, {
198 ...rest,
199 definition: compiled.definition,
200 policyStatements: [
201 ...compiled.policyStatements,
202 ...(policyStatements ?? []),
203 ],
204 });
205 });
206
207/**
208 * An AWS Step Functions state machine (workflow).
209 *
210 * `StateMachine` owns the lifecycle of a `STANDARD` or `EXPRESS` workflow.
211 * The Amazon States Language definition may be provided as a plain object —
212 * `Output` values (like Lambda function ARNs) inside it are resolved before
213 * serialization — and an execution role is created automatically unless an
214 * explicit `roleArn` is given. Lambda functions referenced in the definition
215 * are granted `lambda:InvokeFunction` on the auto-created role.
216 * ### Creating State Machines
217 * **Example:** Standard Workflow with a Pass State
218 * ```typescript
219 * import * as StepFunctions from "alchemy/AWS/StepFunctions";
220 *
221 * const machine = yield* StepFunctions.StateMachine("OrderWorkflow", {
222 * definition: {
223 * StartAt: "Done",
224 * States: {
225 * Done: { Type: "Pass", End: true },
226 * },
227 * },
228 * });
229 * ```
230 *
231 * **Example:** Express Workflow
232 * ```typescript
233 * const machine = yield* StepFunctions.StateMachine("FastWorkflow", {
234 * type: "EXPRESS",
235 * definition: {
236 * StartAt: "Echo",
237 * States: {
238 * Echo: { Type: "Pass", End: true },
239 * },
240 * },
241 * });
242 * ```
243 *
244 * ### Orchestrating Lambda Functions
245 * Reference a function ARN in a Task state — `lambda:InvokeFunction` is
246 * granted on the auto-created execution role automatically.
247 *
248 * **Example:** Invoke a Lambda Function
249 * ```typescript
250 * const machine = yield* StepFunctions.StateMachine("Pipeline", {
251 * definition: {
252 * StartAt: "Process",
253 * States: {
254 * Process: {
255 * Type: "Task",
256 * Resource: fn.functionArn,
257 * End: true,
258 * },
259 * },
260 * },
261 * });
262 * ```
263 *
264 * ### Service Integrations
265 * **Example:** Send a Task Token to SQS (callback pattern)
266 * ```typescript
267 * const machine = yield* StepFunctions.StateMachine("Callback", {
268 * definition: {
269 * StartAt: "WaitForApproval",
270 * States: {
271 * WaitForApproval: {
272 * Type: "Task",
273 * Resource: "arn:aws:states:::sqs:sendMessage.waitForTaskToken",
274 * Parameters: {
275 * QueueUrl: queue.queueUrl,
276 * MessageBody: { "token.$": "$$.Task.Token" },
277 * },
278 * End: true,
279 * },
280 * },
281 * },
282 * policyStatements: [
283 * {
284 * Effect: "Allow",
285 * Action: ["sqs:SendMessage"],
286 * Resource: [queue.queueArn],
287 * },
288 * ],
289 * });
290 * ```
291 *
292 * ### Starting Executions at Runtime
293 * Bind execution operations in the init phase and use them in runtime
294 * handlers.
295 *
296 * **Example:** Start a workflow from a handler
297 * ```typescript
298 * // init
299 * const startExecution = yield* StepFunctions.StartExecution(machine);
300 *
301 * return {
302 * fetch: Effect.gen(function* () {
303 * // runtime
304 * const execution = yield* startExecution({
305 * input: JSON.stringify({ orderId: "123" }),
306 * });
307 * return HttpServerResponse.json({ executionArn: execution.executionArn });
308 * }),
309 * };
310 * ```
311 *
312 * **Example:** Run an EXPRESS workflow synchronously
313 * ```typescript
314 * // init
315 * const startSyncExecution = yield* StepFunctions.StartSyncExecution(machine);
316 *
317 * // runtime
318 * const result = yield* startSyncExecution({
319 * input: JSON.stringify({ value: 21 }),
320 * });
321 * // result.status === "SUCCEEDED", result.output is the workflow output
322 * ```
323 *
324 * ### Typed Programs
325 * Author the workflow as a typed `Sfn` program (mirroring Effect's names —
326 * `Sfn.gen`, `Sfn.invoke`, `Sfn.when`, `Sfn.forEach`, `Sfn.catchTag`, ...)
327 * and compile it with `StateMachine.fromProgram`. The compiler emits a plain
328 * ASL definition plus the IAM policy statements its task states need; the
329 * raw `definition` path above stays fully usable underneath.
330 *
331 * **Example:** Compile a typed program
332 * ```typescript
333 * import { Sfn, StateMachine } from "alchemy/AWS/StepFunctions";
334 *
335 * const machine = yield* StateMachine.fromProgram("OrderWorkflow", {
336 * type: "EXPRESS",
337 * program: Sfn.gen(function* (input: Sfn.Expr<{ value: number }>) {
338 * const result = yield* Sfn.invoke<{ doubled: number }>(doubler, {
339 * value: input.value,
340 * });
341 * const size = yield* Sfn.when(
342 * Sfn.gt(result.doubled, 10),
343 * Sfn.succeed("big"),
344 * Sfn.succeed("small"),
345 * );
346 * return { doubled: result.doubled, size };
347 * }),
348 * });
349 * ```
350 *
351 * @resource
352 */
353export const StateMachine: typeof StateMachineResource & {
354 fromProgram: typeof fromProgram;
355} = Object.assign(StateMachineResource, { fromProgram });
356
357/** Normalize a plain or redacted string to its plain value. */
358const plain = (value: string | Redacted.Redacted<string>): string =>
359 typeof value === "string" ? value : Redacted.value(value);
360
361/** Convert a tag record to the SFN wire shape (lowercase key/value). */
362const toSfnTags = (tags: Record<string, string>): sfn.Tag[] =>
363 Object.entries(tags).map(([key, value]) => ({ key, value }));
364
365/** Convert an SFN wire tag list back to a record. */
366const fromSfnTags = (
367 tags: ReadonlyArray<sfn.Tag> | undefined,
368): Record<string, string> =>
369 Object.fromEntries(
370 (tags ?? [])
371 .filter(
372 (tag): tag is { key: string; value: string } =>
373 typeof tag.key === "string" && typeof tag.value === "string",
374 )
375 .map((tag) => [tag.key, tag.value]),
376 );
377
378/**
379 * Serialize the desired definition to its ASL JSON string, applying
380 * `${key}` substitutions.
381 */
382const serializeDefinition = (props: {
383 definition: Record<string, unknown> | string;
384 definitionSubstitutions?: Record<string, string>;
385}): string => {
386 let definition =
387 typeof props.definition === "string"
388 ? props.definition
389 : JSON.stringify(props.definition, null, 2);
390 for (const [key, value] of Object.entries(
391 props.definitionSubstitutions ?? {},
392 )) {
393 definition = definition.replaceAll(`\${${key}}`, value);
394 }
395 return definition;
396};
397
398/**
399 * Canonicalize an ASL JSON string for drift comparison (whitespace/key-order
400 * insensitive). Falls back to the raw string when unparsable.
401 */
402const normalizeDefinition = (definition: string): string => {
403 try {
404 const stable = (value: unknown): unknown =>
405 Array.isArray(value)
406 ? value.map(stable)
407 : value !== null && typeof value === "object"
408 ? Object.fromEntries(
409 Object.entries(value as Record<string, unknown>)
410 .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
411 .map(([k, v]) => [k, stable(v)]),
412 )
413 : value;
414 return JSON.stringify(stable(JSON.parse(definition)));
415 } catch {
416 return definition;
417 }
418};
419
420/**
421 * Extract every Lambda function ARN referenced in a serialized definition so
422 * the auto-created execution role can be granted `lambda:InvokeFunction`.
423 */
424const scanLambdaArns = (definition: string): string[] => {
425 const matches = definition.match(
426 /arn:aws[a-z0-9-]*:lambda:[a-z0-9-]+:\d{12}:function:[A-Za-z0-9_.-]+(?::[A-Za-z0-9_$-]+)?/g,
427 );
428 return [...new Set(matches ?? [])];
429};
430
431/**
432 * CloudWatch Logs delivery permissions required by Step Functions logging.
433 * These actions do not support resource-level scoping.
434 */
435const LOG_DELIVERY_ACTIONS = [
436 "logs:CreateLogDelivery",
437 "logs:CreateLogStream",
438 "logs:GetLogDelivery",
439 "logs:UpdateLogDelivery",
440 "logs:DeleteLogDelivery",
441 "logs:ListLogDeliveries",
442 "logs:PutLogEvents",
443 "logs:PutResourcePolicy",
444 "logs:DescribeResourcePolicies",
445 "logs:DescribeLogGroups",
446];
447
448/** Map the `logging` prop to the wire `LoggingConfiguration`. */
449const toWireLogging = (
450 logging: StateMachineLogging | undefined,
451): sfn.LoggingConfiguration | undefined =>
452 logging === undefined
453 ? undefined
454 : {
455 level: logging.level,
456 includeExecutionData: logging.includeExecutionData ?? false,
457 destinations: (logging.destinations ?? []).map((arn) => ({
458 cloudWatchLogsLogGroup: {
459 logGroupArn: arn.endsWith(":*") ? arn : `${arn}:*`,
460 },
461 })),
462 };
463
464/** Compare observed logging configuration against the desired one. */
465const loggingDrifted = (
466 observed: sfn.LoggingConfiguration | undefined,
467 desired: sfn.LoggingConfiguration | undefined,
468): boolean => {
469 const observedLevel = observed?.level ?? "OFF";
470 const desiredLevel = desired?.level ?? "OFF";
471 if (observedLevel !== desiredLevel) return true;
472 if (desiredLevel === "OFF") return false;
473 if (
474 (observed?.includeExecutionData ?? false) !==
475 (desired?.includeExecutionData ?? false)
476 ) {
477 return true;
478 }
479 const arnsOf = (config: sfn.LoggingConfiguration | undefined) =>
480 (config?.destinations ?? [])
481 .map((d) => d.cloudWatchLogsLogGroup?.logGroupArn ?? "")
482 .sort()
483 .join(",");
484 return arnsOf(observed) !== arnsOf(desired);
485};
486
487/**
488 * IAM changes (new roles, fresh trust policies) propagate to Step Functions
489 * eventually; a create right after role creation can transiently fail with
490 * `AccessDeniedException`. Bounded retry, explicitly typed so declaration
491 * emit never widens the provider layer (see PATTERNS §7).
492 */
493const retryWhileRolePropagates = <A, E extends { _tag: string }, R>(
494 self: Effect.Effect<A, E, R>,
495): Effect.Effect<A, E, R> =>
496 Effect.retry(self, {
497 while: (e) => e._tag === "AccessDeniedException",
498 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(10)]),
499 });
500
501/**
502 * A state machine of the same name may still be `DELETING` from a prior
503 * destroy; creation returns the typed `StateMachineDeleting` until the
504 * deletion completes. Bounded retry (~60s).
505 */
506const retryWhileDeleting = <A, E extends { _tag: string }, R>(
507 self: Effect.Effect<A, E, R>,
508): Effect.Effect<A, E, R> =>
509 Effect.retry(self, {
510 while: (e) => e._tag === "StateMachineDeleting",
511 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
512 });
513
514export const StateMachineProvider = () =>
515 Provider.effect(
516 StateMachine,
517 Effect.gen(function* () {
518 const createName = Effect.fn(function* (
519 id: string,
520 props: Pick<StateMachineProps, "stateMachineName">,
521 ) {
522 return (
523 props.stateMachineName ??
524 (yield* createPhysicalName({ id, maxLength: 80 }))
525 );
526 });
527
528 const createRoleName = (id: string) =>
529 createPhysicalName({ id, maxLength: 64 });
530
531 const createPolicyName = (id: string) =>
532 createPhysicalName({ id, maxLength: 128 });
533
534 const stateMachineArnOf = (
535 region: string,
536 accountId: string,
537 name: string,
538 ) => `arn:aws:states:${region}:${accountId}:stateMachine:${name}`;
539
540 const describeOrUndefined = Effect.fn(function* (
541 stateMachineArn: string,
542 ) {
543 return yield* sfn
544 .describeStateMachine({ stateMachineArn })
545 .pipe(
546 Effect.catchTag("StateMachineDoesNotExist", () =>
547 Effect.succeed(undefined),
548 ),
549 );
550 });
551
552 /**
553 * Pre-flight the serialized definition through the vendor validator
554 * so a definitively invalid document fails with typed diagnostics
555 * before any create/update call. Best-effort: if the validate call
556 * itself fails (missing IAM permission, throttling), reconcile
557 * proceeds and create/updateStateMachine remains the authority.
558 */
559 const preflightValidateDefinition = Effect.fn(function* (
560 definition: string,
561 type: StateMachineType,
562 ) {
563 const report = yield* sfn
564 .validateStateMachineDefinition({
565 definition,
566 type,
567 severity: "ERROR",
568 })
569 .pipe(Effect.catch(() => Effect.succeed(undefined)));
570 if (report !== undefined && report.result === "FAIL") {
571 return yield* Effect.fail(
572 new InvalidStateMachineDefinition({
573 diagnostics: report.diagnostics.map((diagnostic) => ({
574 severity: diagnostic.severity,
575 code: plain(diagnostic.code),
576 message: plain(diagnostic.message),
577 location:
578 diagnostic.location === undefined
579 ? undefined
580 : plain(diagnostic.location),
581 })),
582 }),
583 );
584 }
585 });
586
587 const fetchObservedTags = Effect.fn(function* (resourceArn: string) {
588 return yield* sfn.listTagsForResource({ resourceArn }).pipe(
589 Effect.map((r) => fromSfnTags(r.tags)),
590 Effect.catchTag("ResourceNotFound", () =>
591 Effect.succeed({} as Record<string, string>),
592 ),
593 );
594 });
595
596 /**
597 * Compute the full inline-policy statement list for the auto-created
598 * execution role: Lambda invoke grants scanned from the definition,
599 * user statements, binding-contributed statements, and CloudWatch
600 * Logs delivery permissions when logging is enabled.
601 */
602 const buildRolePolicyStatements = (
603 definition: string,
604 news: StateMachineProps,
605 bindings: ResourceBinding<StateMachine["Binding"]>[],
606 ): PolicyStatement[] => {
607 const statements: PolicyStatement[] = [];
608 const lambdaArns = scanLambdaArns(definition);
609 if (lambdaArns.length > 0) {
610 statements.push({
611 Effect: "Allow",
612 Action: ["lambda:InvokeFunction"],
613 Resource: [
614 ...lambdaArns,
615 // cover qualified invocations (versions/aliases)
616 ...lambdaArns
617 .filter((arn) => !/:function:[^:]+:/.test(arn))
618 .map((arn) => `${arn}:*`),
619 ],
620 });
621 }
622 if (news.logging && news.logging.level !== "OFF") {
623 statements.push({
624 Effect: "Allow",
625 Action: LOG_DELIVERY_ACTIONS,
626 Resource: ["*"],
627 });
628 }
629 if (news.tracingEnabled) {
630 statements.push({
631 Effect: "Allow",
632 Action: [
633 "xray:PutTraceSegments",
634 "xray:PutTelemetryRecords",
635 "xray:GetSamplingRules",
636 "xray:GetSamplingTargets",
637 ],
638 Resource: ["*"],
639 });
640 }
641 statements.push(...(news.policyStatements ?? []));
642 statements.push(...bindings.flatMap((b) => b.data.policyStatements));
643 return statements;
644 };
645
646 /**
647 * Ensure the auto-created execution role exists with the
648 * `states.amazonaws.com` trust policy and the desired inline policy.
649 * Idempotent: tolerates the role already existing and converges the
650 * inline policy on every reconcile.
651 */
652 const ensureExecutionRole = Effect.fn(function* ({
653 id,
654 roleName,
655 policyName,
656 statements,
657 }: {
658 id: string;
659 roleName: string;
660 policyName: string;
661 statements: PolicyStatement[];
662 }) {
663 const tags = yield* createInternalTags(id);
664 const role = yield* iam
665 .createRole({
666 RoleName: roleName,
667 AssumeRolePolicyDocument: JSON.stringify({
668 Version: "2012-10-17",
669 Statement: [
670 {
671 Effect: "Allow",
672 Principal: { Service: "states.amazonaws.com" },
673 Action: "sts:AssumeRole",
674 },
675 ],
676 }),
677 Tags: Object.entries(tags).map(([Key, Value]) => ({ Key, Value })),
678 })
679 .pipe(
680 Effect.catchTag("EntityAlreadyExistsException", () =>
681 iam.getRole({ RoleName: roleName }),
682 ),
683 );
684
685 if (statements.length > 0) {
686 yield* iam.putRolePolicy({
687 RoleName: roleName,
688 PolicyName: policyName,
689 PolicyDocument: JSON.stringify({
690 Version: "2012-10-17",
691 Statement: statements,
692 }),
693 });
694 } else {
695 yield* iam
696 .deleteRolePolicy({ RoleName: roleName, PolicyName: policyName })
697 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
698 }
699
700 return role.Role.Arn;
701 });
702
703 return StateMachine.Provider.of({
704 stables: ["stateMachineName", "stateMachineArn", "type"],
705
706 list: () =>
707 Effect.gen(function* () {
708 const pages = yield* sfn.listStateMachines
709 .pages({})
710 .pipe(Stream.runCollect);
711 const items = Array.from(pages).flatMap(
712 (page) => page.stateMachines ?? [],
713 );
714 const results = yield* Effect.forEach(
715 items,
716 (item) =>
717 sfn
718 .describeStateMachine({
719 stateMachineArn: item.stateMachineArn,
720 })
721 .pipe(
722 // name/roleArn are optional on the wire; a machine
723 // missing either cannot be expressed as Attributes.
724 Effect.map((machine) =>
725 machine.name != null &&
726 machine.stateMachineArn != null &&
727 machine.roleArn != null
728 ? {
729 stateMachineName: machine.name,
730 stateMachineArn: machine.stateMachineArn,
731 type: machine.type as StateMachineType,
732 roleArn: machine.roleArn,
733 roleName: undefined as string | undefined,
734 }
735 : undefined,
736 ),
737 Effect.catchTag("StateMachineDoesNotExist", () =>
738 Effect.succeed(undefined),
739 ),
740 ),
741 { concurrency: 5 },
742 );
743 return results.filter(
744 (item): item is StateMachine["Attributes"] => item !== undefined,
745 );
746 }),
747
748 read: Effect.fn(function* ({ id, olds, output }) {
749 const { accountId, region } = yield* AWSEnvironment.current;
750 const name =
751 output?.stateMachineName ?? (yield* createName(id, olds ?? {}));
752 const stateMachineArn =
753 output?.stateMachineArn ??
754 stateMachineArnOf(region, accountId, name);
755 const found = yield* describeOrUndefined(stateMachineArn);
756 if (!found || found.status === "DELETING") return undefined;
757 const attrs = {
758 stateMachineName: found.name,
759 stateMachineArn: found.stateMachineArn,
760 type: found.type as StateMachineType,
761 roleArn: found.roleArn,
762 roleName: output?.roleName,
763 };
764 const tags = yield* fetchObservedTags(stateMachineArn);
765 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
766 }),
767
768 diff: Effect.fn(function* ({ id, news, olds }) {
769 if (!isResolved(news)) return undefined;
770 const oldName = yield* createName(id, olds ?? {});
771 const newName = yield* createName(id, news ?? {});
772 if (oldName !== newName) {
773 return { action: "replace" } as const;
774 }
775 if ((olds?.type ?? "STANDARD") !== (news?.type ?? "STANDARD")) {
776 return { action: "replace" } as const;
777 }
778 // definition/role/logging/tracing/tags converge via update
779 }),
780
781 reconcile: Effect.fn(function* ({
782 id,
783 news,
784 output,
785 session,
786 bindings,
787 }) {
788 const { accountId, region } = yield* AWSEnvironment.current;
789 const name =
790 output?.stateMachineName ?? (yield* createName(id, news));
791 const stateMachineArn = stateMachineArnOf(region, accountId, name);
792 const type = (news.type ?? "STANDARD") as StateMachineType;
793 const definition = serializeDefinition(news);
794 const desiredLogging = toWireLogging(news.logging);
795 const desiredTracing = { enabled: news.tracingEnabled ?? false };
796 const internalTags = yield* createInternalTags(id);
797 const desiredTags = { ...news.tags, ...internalTags };
798
799 // Pre-flight: fail with typed diagnostics before touching IAM or
800 // the machine when the definition is definitively invalid.
801 yield* preflightValidateDefinition(definition, type);
802
803 // Ensure the execution role first — the machine cannot exist
804 // without one. Managed unless an explicit roleArn is provided.
805 let roleArn = news.roleArn;
806 let roleName: string | undefined;
807 if (roleArn === undefined) {
808 roleName = yield* createRoleName(id);
809 const policyName = yield* createPolicyName(id);
810 roleArn = yield* ensureExecutionRole({
811 id,
812 roleName,
813 policyName,
814 statements: buildRolePolicyStatements(definition, news, bindings),
815 });
816 }
817
818 // 1. OBSERVE — cloud state is authoritative; `output` is only a
819 // name cache. A machine still DELETING from a prior destroy
820 // must finish deleting before we can recreate it.
821 const observed = yield* describeOrUndefined(stateMachineArn).pipe(
822 Effect.map((machine) =>
823 machine?.status === "DELETING" ? undefined : machine,
824 ),
825 );
826
827 if (observed === undefined) {
828 // 2. ENSURE — create; tolerate the concurrent-create race and
829 // wait out a same-name deletion or IAM propagation delay.
830 yield* retryWhileRolePropagates(
831 retryWhileDeleting(
832 sfn
833 .createStateMachine({
834 name,
835 definition,
836 roleArn,
837 type,
838 loggingConfiguration: desiredLogging,
839 tracingConfiguration: desiredTracing,
840 tags: toSfnTags(desiredTags),
841 })
842 .pipe(
843 Effect.catchTag("StateMachineAlreadyExists", () =>
844 Effect.succeed(undefined),
845 ),
846 ),
847 ),
848 );
849 } else {
850 // 3. SYNC — diff OBSERVED cloud state against desired and issue
851 // a single update only when something actually drifted.
852 const drifted =
853 normalizeDefinition(plain(observed.definition)) !==
854 normalizeDefinition(definition) ||
855 observed.roleArn !== roleArn ||
856 loggingDrifted(observed.loggingConfiguration, desiredLogging) ||
857 (observed.tracingConfiguration?.enabled ?? false) !==
858 desiredTracing.enabled;
859 if (drifted) {
860 yield* retryWhileRolePropagates(
861 sfn.updateStateMachine({
862 stateMachineArn,
863 definition,
864 roleArn,
865 loggingConfiguration: desiredLogging ?? { level: "OFF" },
866 tracingConfiguration: desiredTracing,
867 }),
868 );
869 // Updates propagate eventually (typically seconds). Poll the
870 // describe endpoint (bounded) so subsequent reads observe the
871 // new revision; proceed best-effort if it hasn't settled.
872 yield* describeOrUndefined(stateMachineArn).pipe(
873 Effect.repeat({
874 schedule: Schedule.fixed("1 second"),
875 until: (machine) =>
876 machine === undefined ||
877 normalizeDefinition(plain(machine.definition)) ===
878 normalizeDefinition(definition),
879 times: 20,
880 }),
881 );
882 }
883 }
884
885 // 3b. SYNC TAGS — against OBSERVED cloud tags so adoption
886 // converges (create-time tags only apply on first create).
887 const observedTags = yield* fetchObservedTags(stateMachineArn);
888 const { upsert, removed } = diffTags(observedTags, desiredTags);
889 if (upsert.length > 0) {
890 yield* sfn.tagResource({
891 resourceArn: stateMachineArn,
892 tags: upsert.map(({ Key, Value }) => ({
893 key: Key,
894 value: Value,
895 })),
896 });
897 }
898 if (removed.length > 0) {
899 yield* sfn.untagResource({
900 resourceArn: stateMachineArn,
901 tagKeys: removed,
902 });
903 }
904
905 yield* session.note(stateMachineArn);
906 return {
907 stateMachineName: name,
908 stateMachineArn,
909 type,
910 roleArn,
911 roleName,
912 };
913 }),
914
915 delete: Effect.fn(function* ({ output }) {
916 // deleteStateMachine is idempotent (no error when already gone)
917 // and asynchronous — the machine transitions through DELETING.
918 yield* sfn.deleteStateMachine({
919 stateMachineArn: output.stateMachineArn,
920 });
921
922 // Tear down the managed execution role (absent when the user
923 // supplied an explicit roleArn). Every step tolerates the role
924 // being partially or fully gone already.
925 if (output.roleName !== undefined) {
926 const roleName = output.roleName;
927 yield* iam.listRolePolicies.items({ RoleName: roleName }).pipe(
928 Stream.mapEffect((policyName) =>
929 iam
930 .deleteRolePolicy({
931 RoleName: roleName,
932 PolicyName: policyName,
933 })
934 .pipe(
935 Effect.catchTag("NoSuchEntityException", () => Effect.void),
936 ),
937 ),
938 Stream.runDrain,
939 Effect.catchTag("NoSuchEntityException", () => Effect.void),
940 );
941 yield* iam
942 .deleteRole({ RoleName: roleName })
943 .pipe(
944 Effect.catchTag("NoSuchEntityException", () => Effect.void),
945 );
946 }
947 }),
948 });
949 }),
950 );