1import type * as Effect from "effect/Effect";
2import type * as Stream from "effect/Stream";
3import {
4 consumeBusEvents,
5 type EventRecord,
6 type EventRouteProps,
7} from "../EventBridge/EventSource.ts";
8
9/**
10 * The `detail` payload AWS DataSync delivers to EventBridge. Task
11 * state-change events describe the task's new state (`AVAILABLE`,
12 * `RUNNING`, `UNAVAILABLE`, `QUEUED`); task-execution state-change events
13 * describe the run's transfer phase (`LAUNCHING`, `PREPARING`,
14 * `TRANSFERRING`, `VERIFYING`, `SUCCESS`, `ERROR`). Fields not shared by
15 * every event kind are optional (the schema grows over time).
16 */
17export interface TaskEventDetail {
18 /** The state the task or task execution transitioned to. */
19 State?: string;
20 /** Additional event fields (the schema grows over time). */
21 [key: string]: unknown;
22}
23
24/** A DataSync EventBridge event delivered to the handler. */
25export type TaskEvent = EventRecord<TaskEventDetail>;
26
27/** Which DataSync notifications to subscribe to. */
28export type TaskEventKind = "task-state-change" | "task-execution-state-change";
29
30const DETAIL_TYPES: Record<TaskEventKind, string> = {
31 "task-state-change": "DataSync Task State Change",
32 "task-execution-state-change": "DataSync Task Execution State Change",
33};
34
35export interface TaskEventSourceProps extends EventRouteProps {
36 /**
37 * Logical id for the backing EventBridge rule.
38 * @default "DataSyncTaskEvents"
39 */
40 id?: string;
41 /**
42 * Which notifications to subscribe to: task state changes (e.g. a task
43 * becoming `UNAVAILABLE`) and/or task-execution state changes (a run
44 * progressing through `LAUNCHING` → … → `SUCCESS`/`ERROR`).
45 * @default ["task-execution-state-change"]
46 */
47 kinds?: readonly TaskEventKind[];
48 /**
49 * Restrict to events about specific tasks. Matched as a prefix against
50 * the event's top-level `resources`, so a task ARN also matches its
51 * execution ARNs (`{taskArn}/execution/{id}`).
52 */
53 taskArns?: readonly string[];
54}
55
56/**
57 * Event source connecting AWS DataSync notifications to the hosting
58 * compute. DataSync publishes task state changes and task-execution state
59 * changes to the account's default EventBridge bus (source `aws.datasync`);
60 * this subscribes the host Function to those events so it can react when a
61 * transfer finishes, fails, or a task drops offline.
62 *
63 * DataSync publishes to EventBridge automatically — no additional resource
64 * is created besides the EventBridge rule targeting the host. Provide the
65 * host-specific implementation layer (e.g. `AWS.Lambda.EventSource`) on the
66 * Function effect.
67 *
68 * ### Consuming Task Events
69 * **Example:** React When A Transfer Finishes
70 * ```typescript
71 * import * as AWS from "alchemy/AWS";
72 *
73 * export default AlertFunction.make(
74 * { main: import.meta.url },
75 * Effect.gen(function* () {
76 * yield* AWS.DataSync.consumeTaskEvents(
77 * { kinds: ["task-execution-state-change"] },
78 * (events) =>
79 * Stream.runForEach(events, (event) =>
80 * event.detail.State === "ERROR"
81 * ? Effect.logError(`transfer failed: ${event.resources[0]}`)
82 * : Effect.log(`transfer ${event.detail.State}`),
83 * ),
84 * );
85 * return {};
86 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
87 * );
88 * ```
89 */
90export const consumeTaskEvents = <StreamReq = never, Req = never>(
91 props: TaskEventSourceProps,
92 process: (
93 events: Stream.Stream<TaskEvent, never, StreamReq>,
94 ) => Effect.Effect<void, never, Req>,
95) =>
96 consumeBusEvents(
97 props.id ?? "DataSyncTaskEvents",
98 {
99 source: ["aws.datasync"],
100 "detail-type": (
101 props.kinds ?? (["task-execution-state-change"] as const)
102 ).map((kind) => DETAIL_TYPES[kind]),
103 ...(props.taskArns !== undefined
104 ? { resources: props.taskArns.map((arn) => ({ prefix: arn })) }
105 : {}),
106 },
107 { description: props.description, state: props.state },
108 process,
109 );