1import * as omics from "@distilled.cloud/aws/omics";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12import { fetchOmicsTags, syncOmicsTags } from "./internal.ts";
13
14/**
15 * Coerce a definition zip that survived engine plan/state serialization back
16 * into a `Uint8Array`. The engine round-trips a `Uint8Array` prop as an
17 * index-keyed plain object (`{"0":80,"1":75,...}`), so a provider that carries
18 * binary members must rebuild the array before encoding the request.
19 */
20const coerceBytes = (value: unknown): Uint8Array | undefined => {
21 if (value === undefined) return undefined;
22 if (value instanceof Uint8Array) return value;
23 if (Array.isArray(value)) return Uint8Array.from(value as number[]);
24 if (typeof value === "object" && value !== null) {
25 const entries = Object.entries(value as Record<string, number>)
26 .map(([k, v]) => [Number(k), v] as const)
27 .sort((a, b) => a[0] - b[0]);
28 return Uint8Array.from(entries.map(([, v]) => v));
29 }
30 return undefined;
31};
32
33export interface WorkflowProps {
34 /**
35 * A name for the workflow. If omitted, a unique name is generated from the
36 * app, stage, and logical ID. Mutable.
37 */
38 name?: string;
39 /**
40 * A description for the workflow. Mutable.
41 */
42 description?: string;
43 /**
44 * The workflow engine. Immutable — changing it replaces the workflow.
45 * @default "WDL"
46 */
47 engine?: "WDL" | "NEXTFLOW" | "CWL";
48 /**
49 * The definition of the workflow, as a zip archive of the workflow source
50 * files. Provided as raw bytes. Mutually exclusive with `definitionUri`.
51 * Immutable — changing it replaces the workflow.
52 */
53 definitionZip?: Uint8Array;
54 /**
55 * The S3 URI of a zip archive containing the workflow definition. Mutually
56 * exclusive with `definitionZip`. Immutable — changing it replaces the
57 * workflow.
58 */
59 definitionUri?: string;
60 /**
61 * The path of the main definition file for the workflow within the zip
62 * archive. Immutable.
63 */
64 main?: string;
65 /**
66 * A parameter template describing the workflow's input parameters.
67 * Immutable.
68 */
69 parameterTemplate?: Record<
70 string,
71 { description?: string; optional?: boolean }
72 >;
73 /**
74 * The default static storage capacity (in gibibytes) for runs that use this
75 * workflow. Mutable.
76 */
77 storageCapacity?: number;
78 /**
79 * The storage type for runs that use this workflow. Mutable.
80 * @default "DYNAMIC"
81 */
82 storageType?: "STATIC" | "DYNAMIC";
83 /**
84 * The computational accelerators used by the workflow. Immutable.
85 */
86 accelerators?: "GPU";
87 /**
88 * Tags to apply to the workflow. Merged with internal Alchemy tags.
89 */
90 tags?: Record<string, string>;
91}
92
93export interface Workflow extends Resource<
94 "AWS.Omics.Workflow",
95 WorkflowProps,
96 {
97 /**
98 * ID of the workflow.
99 */
100 workflowId: string;
101 /**
102 * ARN of the workflow.
103 */
104 workflowArn: string;
105 /**
106 * Name of the workflow.
107 */
108 name: string;
109 /**
110 * Workflow status (e.g. `ACTIVE`, `CREATING`, `UPDATING`, `FAILED`).
111 */
112 status: string;
113 },
114 never,
115 Providers
116> {}
117
118/**
119 * An Amazon HealthOmics private workflow — a bioinformatics workflow (WDL,
120 * Nextflow, or CWL) that is executed as one or more runs.
121 *
122 * A workflow name is auto-generated from the app, stage, and logical ID
123 * unless you provide one. The workflow definition (`engine`, `definitionZip`,
124 * `definitionUri`, `main`, `parameterTemplate`, `accelerators`) is immutable —
125 * changing any of it replaces the workflow. `name`, `description`,
126 * `storageCapacity`, and `storageType` are updated in place.
127 * ### Creating a Workflow
128 * **Example:** Workflow from an inline definition zip
129 * ```typescript
130 * import * as Omics from "alchemy/AWS/Omics";
131 *
132 * const workflow = yield* Omics.Workflow("Hello", {
133 * engine: "WDL",
134 * main: "main.wdl",
135 * definitionZip: myZipBytes,
136 * });
137 * ```
138 *
139 * **Example:** Workflow from an S3-hosted definition
140 * ```typescript
141 * const workflow = yield* Omics.Workflow("Hello", {
142 * engine: "NEXTFLOW",
143 * definitionUri: "s3://my-bucket/workflows/hello.zip",
144 * });
145 * ```
146 *
147 * @resource
148 */
149export const Workflow = Resource<Workflow>("AWS.Omics.Workflow");
150
151export const WorkflowProvider = () =>
152 Provider.effect(
153 Workflow,
154 Effect.gen(function* () {
155 const createName = Effect.fn(function* (
156 id: string,
157 props: { name?: string | undefined },
158 ) {
159 return props.name ?? (yield* createPhysicalName({ id, maxLength: 96 }));
160 });
161
162 // A freshly created workflow is CREATING until its definition is
163 // validated; wait until it reaches a terminal state before returning so
164 // downstream runs (and deletes) don't race the async provisioning.
165 const waitUntilReady = Effect.fn(function* (workflowId: string) {
166 const final = yield* omics.getWorkflow({ id: workflowId }).pipe(
167 Effect.repeat({
168 schedule: Schedule.max([
169 Schedule.fixed("5 seconds"),
170 Schedule.recurs(11),
171 ]),
172 until: (w) => w.status === "ACTIVE" || w.status === "FAILED",
173 }),
174 );
175 if (final.status === "FAILED") {
176 return yield* Effect.fail(
177 new omics.ValidationException({
178 message: `Workflow ${workflowId} failed: ${final.statusMessage ?? "unknown"}`,
179 }),
180 );
181 }
182 return final;
183 });
184
185 return Workflow.Provider.of({
186 stables: ["workflowId", "workflowArn"],
187 list: () =>
188 omics.listWorkflows.items({}).pipe(
189 Stream.map((item) => ({
190 workflowId: item.id!,
191 workflowArn: item.arn!,
192 name: item.name ?? "",
193 status: item.status ?? "",
194 })),
195 Stream.runCollect,
196 Effect.map((chunk) => Array.from(chunk)),
197 ),
198 read: Effect.fn(function* ({ id, output }) {
199 if (output?.workflowId === undefined) return undefined;
200 const found = yield* omics
201 .getWorkflow({ id: output.workflowId })
202 .pipe(
203 Effect.catchTag("ResourceNotFoundException", () =>
204 Effect.succeed(undefined),
205 ),
206 );
207 if (found === undefined || found.id === undefined) return undefined;
208 const attrs = {
209 workflowId: found.id,
210 workflowArn: found.arn!,
211 name: found.name ?? "",
212 status: found.status ?? "",
213 };
214 const tags = yield* fetchOmicsTags(found.arn!);
215 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
216 }),
217 diff: Effect.fn(function* ({ news, olds }) {
218 if (!isResolved(news)) return undefined;
219 const prev = olds ?? {};
220 if (
221 (prev.engine ?? "WDL") !== (news.engine ?? "WDL") ||
222 (prev.definitionUri ?? "") !== (news.definitionUri ?? "") ||
223 (prev.main ?? "") !== (news.main ?? "") ||
224 (prev.accelerators ?? "") !== (news.accelerators ?? "")
225 ) {
226 return { action: "replace" } as const;
227 }
228 }),
229 reconcile: Effect.fn(function* ({ id, news, output, session }) {
230 const name = output?.name ?? (yield* createName(id, news));
231 const internalTags = yield* createInternalTags(id);
232 const desiredTags = { ...news.tags, ...internalTags };
233
234 let workflow =
235 output?.workflowId === undefined
236 ? undefined
237 : yield* omics
238 .getWorkflow({ id: output.workflowId })
239 .pipe(
240 Effect.catchTag("ResourceNotFoundException", () =>
241 Effect.succeed(undefined),
242 ),
243 );
244
245 if (workflow === undefined || workflow.id === undefined) {
246 const created = yield* omics.createWorkflow({
247 name,
248 description: news.description,
249 engine: news.engine ?? "WDL",
250 definitionZip: coerceBytes(news.definitionZip),
251 definitionUri: news.definitionUri,
252 main: news.main,
253 parameterTemplate: news.parameterTemplate,
254 storageCapacity: news.storageCapacity,
255 storageType: news.storageType,
256 accelerators: news.accelerators,
257 requestId: `alchemy-${id}`,
258 tags: desiredTags,
259 });
260 yield* waitUntilReady(created.id!);
261 workflow = yield* omics.getWorkflow({ id: created.id! });
262 } else {
263 const patch: {
264 name?: string;
265 description?: string;
266 storageType?: string;
267 storageCapacity?: number;
268 } = {};
269 if (news.name !== undefined && news.name !== workflow.name) {
270 patch.name = news.name;
271 }
272 if (
273 news.description !== undefined &&
274 news.description !== workflow.description
275 ) {
276 patch.description = news.description;
277 }
278 if (
279 news.storageType !== undefined &&
280 news.storageType !== workflow.storageType
281 ) {
282 patch.storageType = news.storageType;
283 }
284 if (
285 news.storageCapacity !== undefined &&
286 news.storageCapacity !== workflow.storageCapacity
287 ) {
288 patch.storageCapacity = news.storageCapacity;
289 }
290 if (Object.keys(patch).length > 0) {
291 yield* omics.updateWorkflow({ id: workflow.id, ...patch });
292 workflow = yield* omics.getWorkflow({ id: workflow.id });
293 }
294 }
295
296 yield* syncOmicsTags(workflow.arn!, desiredTags);
297
298 yield* session.note(workflow.id!);
299 return {
300 workflowId: workflow.id!,
301 workflowArn: workflow.arn!,
302 name: workflow.name ?? name,
303 status: workflow.status ?? "",
304 };
305 }),
306 delete: Effect.fn(function* ({ output }) {
307 yield* omics
308 .deleteWorkflow({ id: output.workflowId })
309 .pipe(
310 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
311 );
312 }),
313 });
314 }),
315 );