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 Transfer Family delivers to EventBridge for
11 * server file-transfer events. Fields not shared by every event kind are
12 * optional (the schema grows over time).
13 */
14export interface FileTransferEventDetail {
15 /** The user that performed the transfer. */
16 username?: string;
17 /** The ID of the server the transfer went through (e.g. `s-…`). */
18 "server-id"?: string;
19 /** Session identifier of the client connection. */
20 "session-id"?: string;
21 /** Additional event fields (the schema grows over time). */
22 [key: string]: unknown;
23}
24
25/** A Transfer Family EventBridge event delivered to the handler. */
26export type FileTransferEvent = EventRecord<FileTransferEventDetail>;
27
28/** Which Transfer Family server notifications to subscribe to. */
29export type FileTransferEventKind =
30 | "file-upload-completed"
31 | "file-upload-failed"
32 | "file-download-completed"
33 | "file-download-failed";
34
35/** File-transfer protocols whose server events to subscribe to. */
36export type FileTransferEventProtocol = "SFTP" | "FTP" | "FTPS";
37
38const KIND_SUFFIXES: Record<FileTransferEventKind, string> = {
39 "file-upload-completed": "Server File Upload Completed",
40 "file-upload-failed": "Server File Upload Failed",
41 "file-download-completed": "Server File Download Completed",
42 "file-download-failed": "Server File Download Failed",
43};
44
45export interface FileTransferEventSourceProps extends EventRouteProps {
46 /**
47 * Logical id for the backing EventBridge rule.
48 * @default "TransferFileTransferEvents"
49 */
50 id?: string;
51 /**
52 * Which notifications to subscribe to: upload/download completions and
53 * failures on the account's Transfer Family servers.
54 * @default ["file-upload-completed"]
55 */
56 kinds?: readonly FileTransferEventKind[];
57 /**
58 * Which protocols' server events to subscribe to.
59 * @default ["SFTP", "FTP", "FTPS"]
60 */
61 protocols?: readonly FileTransferEventProtocol[];
62 /**
63 * Restrict to events about specific servers. Matched as a prefix against
64 * the event's top-level `resources` (the server ARN).
65 */
66 serverArns?: readonly string[];
67}
68
69/**
70 * Event source connecting AWS Transfer Family notifications to the hosting
71 * compute. Transfer Family publishes file upload/download completions and
72 * failures on its servers to the account's default EventBridge bus (source
73 * `aws.transfer`); this subscribes the host Function to those events so it
74 * can react the moment a partner drops a file on the SFTP endpoint.
75 *
76 * Transfer Family publishes to EventBridge automatically — no additional
77 * resource is created besides the EventBridge rule targeting the host.
78 * Provide the host-specific implementation layer (e.g.
79 * `AWS.Lambda.EventSource`) on the Function effect.
80 *
81 * ### Consuming File-Transfer Events
82 * **Example:** Process Every Uploaded File
83 * ```typescript
84 * import * as AWS from "alchemy/AWS";
85 *
86 * export default IngestFunction.make(
87 * { main: import.meta.url },
88 * Effect.gen(function* () {
89 * yield* AWS.Transfer.consumeFileTransferEvents(
90 * { kinds: ["file-upload-completed"] },
91 * (events) =>
92 * Stream.runForEach(events, (event) =>
93 * Effect.log(
94 * `${event.detail.username} uploaded a file on ${event.detail["server-id"]}`,
95 * ),
96 * ),
97 * );
98 * return {};
99 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
100 * );
101 * ```
102 */
103export const consumeFileTransferEvents = <StreamReq = never, Req = never>(
104 props: FileTransferEventSourceProps,
105 process: (
106 events: Stream.Stream<FileTransferEvent, never, StreamReq>,
107 ) => Effect.Effect<void, never, Req>,
108) => {
109 const kinds = props.kinds ?? (["file-upload-completed"] as const);
110 const protocols = props.protocols ?? (["SFTP", "FTP", "FTPS"] as const);
111 return consumeBusEvents(
112 props.id ?? "TransferFileTransferEvents",
113 {
114 source: ["aws.transfer"],
115 "detail-type": protocols.flatMap((protocol) =>
116 kinds.map((kind) => `${protocol} ${KIND_SUFFIXES[kind]}`),
117 ),
118 ...(props.serverArns !== undefined
119 ? { resources: props.serverArns.map((arn) => ({ prefix: arn })) }
120 : {}),
121 },
122 { description: props.description, state: props.state },
123 process,
124 );
125};