1import * as sfn from "@distilled.cloud/aws/sfn";
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, diffTags, hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12
13export interface ActivityProps {
14 /**
15 * Name of the activity (1-80 characters; letters, digits, dashes and
16 * underscores). If omitted, a deterministic physical name is generated
17 * from the app, stage, and logical ID. Changing the name triggers a
18 * replacement.
19 */
20 activityName?: string;
21 /**
22 * Tags to apply to the activity. Merged with internal Alchemy tags.
23 */
24 tags?: Record<string, string>;
25}
26
27export interface Activity extends Resource<
28 "AWS.StepFunctions.Activity",
29 ActivityProps,
30 {
31 /**
32 * Physical name of the activity.
33 */
34 activityName: string;
35 /**
36 * ARN of the activity — reference it from a state machine Task state's
37 * `Resource` field.
38 */
39 activityArn: string;
40 },
41 never,
42 Providers
43> {}
44
45/**
46 * An AWS Step Functions activity — a named endpoint that external workers
47 * poll for tasks (`GetActivityTask`) and complete with the `SendTask*`
48 * callback operations.
49 *
50 * Activities support worker-hosted task processing outside Lambda. For
51 * most callback flows the `.waitForTaskToken` service-integration pattern
52 * on a {@link StateMachine} Task state is preferred.
53 * ### Creating Activities
54 * **Example:** Basic Activity
55 * ```typescript
56 * import * as StepFunctions from "alchemy/AWS/StepFunctions";
57 *
58 * const activity = yield* StepFunctions.Activity("ApprovalActivity");
59 * ```
60 *
61 * **Example:** Reference an Activity from a State Machine
62 * ```typescript
63 * const machine = yield* StepFunctions.StateMachine("ApprovalFlow", {
64 * definition: {
65 * StartAt: "WaitForWorker",
66 * States: {
67 * WaitForWorker: {
68 * Type: "Task",
69 * Resource: activity.activityArn,
70 * End: true,
71 * },
72 * },
73 * },
74 * });
75 * ```
76 *
77 * ### Completing Tasks at Runtime
78 * **Example:** Send a task result scoped to the activity
79 * ```typescript
80 * // init
81 * const sendTaskSuccess = yield* StepFunctions.SendTaskSuccess(activity);
82 *
83 * // runtime
84 * yield* sendTaskSuccess({
85 * taskToken: token,
86 * output: JSON.stringify({ approved: true }),
87 * });
88 * ```
89 *
90 * @resource
91 */
92export const Activity = Resource<Activity>("AWS.StepFunctions.Activity");
93
94const toSfnTags = (tags: Record<string, string>): sfn.Tag[] =>
95 Object.entries(tags).map(([key, value]) => ({ key, value }));
96
97const fromSfnTags = (
98 tags: ReadonlyArray<sfn.Tag> | undefined,
99): Record<string, string> =>
100 Object.fromEntries(
101 (tags ?? [])
102 .filter(
103 (tag): tag is { key: string; value: string } =>
104 typeof tag.key === "string" && typeof tag.value === "string",
105 )
106 .map((tag) => [tag.key, tag.value]),
107 );
108
109export const ActivityProvider = () =>
110 Provider.effect(
111 Activity,
112 Effect.gen(function* () {
113 const createName = Effect.fn(function* (
114 id: string,
115 props: Pick<ActivityProps, "activityName">,
116 ) {
117 return (
118 props.activityName ??
119 (yield* createPhysicalName({ id, maxLength: 80 }))
120 );
121 });
122
123 const activityArnOf = (region: string, accountId: string, name: string) =>
124 `arn:aws:states:${region}:${accountId}:activity:${name}`;
125
126 const describeOrUndefined = Effect.fn(function* (activityArn: string) {
127 return yield* sfn
128 .describeActivity({ activityArn })
129 .pipe(
130 Effect.catchTag("ActivityDoesNotExist", () =>
131 Effect.succeed(undefined),
132 ),
133 );
134 });
135
136 const fetchObservedTags = Effect.fn(function* (resourceArn: string) {
137 return yield* sfn.listTagsForResource({ resourceArn }).pipe(
138 Effect.map((r) => fromSfnTags(r.tags)),
139 Effect.catchTag("ResourceNotFound", () =>
140 Effect.succeed({} as Record<string, string>),
141 ),
142 );
143 });
144
145 return Activity.Provider.of({
146 stables: ["activityName", "activityArn"],
147
148 list: () =>
149 Effect.gen(function* () {
150 const pages = yield* sfn.listActivities
151 .pages({})
152 .pipe(Stream.runCollect);
153 return Array.from(pages)
154 .flatMap((page) => page.activities ?? [])
155 .map((activity) => ({
156 activityName: activity.name,
157 activityArn: activity.activityArn,
158 }));
159 }),
160
161 read: Effect.fn(function* ({ id, olds, output }) {
162 const { accountId, region } = yield* AWSEnvironment.current;
163 const name =
164 output?.activityName ?? (yield* createName(id, olds ?? {}));
165 const activityArn =
166 output?.activityArn ?? activityArnOf(region, accountId, name);
167 const found = yield* describeOrUndefined(activityArn);
168 if (!found) return undefined;
169 const attrs = {
170 activityName: found.name,
171 activityArn: found.activityArn,
172 };
173 const tags = yield* fetchObservedTags(activityArn);
174 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
175 }),
176
177 diff: Effect.fn(function* ({ id, news, olds }) {
178 if (!isResolved(news)) return undefined;
179 const oldName = yield* createName(id, olds ?? {});
180 const newName = yield* createName(id, news ?? {});
181 if (oldName !== newName) {
182 return { action: "replace" } as const;
183 }
184 // tags converge via update
185 }),
186
187 reconcile: Effect.fn(function* ({ id, news, output, session }) {
188 const { accountId, region } = yield* AWSEnvironment.current;
189 // `news` is undefined when the resource is declared without a
190 // props argument (`Activity("id")`).
191 const props = news ?? {};
192 const name = output?.activityName ?? (yield* createName(id, props));
193 const activityArn = activityArnOf(region, accountId, name);
194 const internalTags = yield* createInternalTags(id);
195 const desiredTags = { ...props.tags, ...internalTags };
196
197 // 1. OBSERVE — cloud state is authoritative.
198 const observed = yield* describeOrUndefined(activityArn);
199
200 // 2. ENSURE — createActivity is idempotent for an identical
201 // configuration; ActivityAlreadyExists only fires when the
202 // existing activity differs (encryption config), in which case
203 // we keep the existing one (the ARN is deterministic).
204 if (observed === undefined) {
205 yield* sfn
206 .createActivity({ name, tags: toSfnTags(desiredTags) })
207 .pipe(
208 Effect.catchTag("ActivityAlreadyExists", () => Effect.void),
209 );
210 }
211
212 // 3. SYNC TAGS — against OBSERVED cloud tags (create-time tags
213 // only apply on first create; adoption may carry foreign tags).
214 const observedTags = yield* fetchObservedTags(activityArn);
215 const { upsert, removed } = diffTags(observedTags, desiredTags);
216 if (upsert.length > 0) {
217 yield* sfn.tagResource({
218 resourceArn: activityArn,
219 tags: upsert.map(({ Key, Value }) => ({
220 key: Key,
221 value: Value,
222 })),
223 });
224 }
225 if (removed.length > 0) {
226 yield* sfn.untagResource({
227 resourceArn: activityArn,
228 tagKeys: removed,
229 });
230 }
231
232 yield* session.note(activityArn);
233 return { activityName: name, activityArn };
234 }),
235
236 delete: Effect.fn(function* ({ output }) {
237 // deleteActivity is idempotent — deleting a non-existent activity
238 // succeeds.
239 yield* sfn.deleteActivity({ activityArn: output.activityArn });
240 }),
241 });
242 }),
243 );