1import * as Effect from "effect/Effect";
2import * as Stream from "effect/Stream";
3
4import type { Bucket } from "./Bucket.ts";
5import { BucketEventSource } from "./BucketEventSource.ts";
6import type { S3EventType } from "./S3Event.ts";
7
8/**
9 * A normalized S3 event notification record.
10 */
11export type BucketNotification = {
12 /** The S3 event type, including the `s3:` prefix. */
13 type: S3EventType;
14 /** Name of the bucket the event originated from. */
15 bucket: string;
16 /** URL-decoded object key that the event applies to. */
17 key: string;
18 /** Size in bytes, when supplied; removal events may omit it. */
19 size?: number;
20 /** Object ETag, when supplied; removal events may omit it. */
21 eTag?: string;
22 /**
23 * Object or delete-marker version ID. Pass as GetObject's VersionId to read
24 * the exact data version; delete markers are not readable objects.
25 */
26 versionId?: string;
27 /** Ordering token for events affecting the same key, when supplied. */
28 sequencer?: string;
29};
30
31export interface NotificationsProps<Events extends S3EventType[]> {
32 /** S3 event types to subscribe to. Defaults to `s3:ObjectCreated:*`. */
33 events?: Events;
34 /**
35 * Only deliver events for object keys beginning with this prefix.
36 * Maps to an S3 `prefix` filter rule on the notification configuration.
37 */
38 prefix?: string;
39 /**
40 * Only deliver events for object keys ending with this suffix.
41 * Maps to an S3 `suffix` filter rule on the notification configuration.
42 */
43 suffix?: string;
44}
45
46/**
47 * Subscribe to S3 bucket event notifications.
48 *
49 * The handler receives a `Stream<BucketNotification>` for processing events
50 * and is passed as the final positional argument.
51 * ### Subscribing to Events
52 * **Example:** Process all object creation events
53 * ```typescript
54 * import * as S3 from "alchemy/AWS/S3";
55 *
56 * yield* S3.consumeBucketEvents(
57 * bucket,
58 * { events: ["s3:ObjectCreated:*"] },
59 * (stream) =>
60 * stream.pipe(
61 * Stream.runForEach((event) =>
62 * Effect.log(`New object: ${event.key} (${event.size} bytes)`),
63 * ),
64 * ),
65 * );
66 * ```
67 *
68 * **Example:** Process object creation events without a key filter
69 * ```typescript
70 * yield* S3.consumeBucketEvents(bucket, (stream) =>
71 * stream.pipe(
72 * Stream.runForEach((event) =>
73 * Effect.log(`${event.type}: ${event.key}`),
74 * ),
75 * ),
76 * );
77 * ```
78 *
79 * @binding
80 */
81export function consumeBucketEvents<
82 B extends Bucket,
83 Req = never,
84 StreamReq = never,
85 const Events extends S3EventType[] = S3EventType[],
86>(
87 bucket: B,
88 handler: (
89 stream: Stream.Stream<BucketNotification, never, StreamReq>,
90 ) => Effect.Effect<void, never, Req>,
91): Effect.Effect<void, never, Req>;
92export function consumeBucketEvents<
93 B extends Bucket,
94 Req = never,
95 StreamReq = never,
96 const Events extends S3EventType[] = S3EventType[],
97>(
98 bucket: B,
99 props: NotificationsProps<Events>,
100 handler: (
101 stream: Stream.Stream<BucketNotification, never, StreamReq>,
102 ) => Effect.Effect<void, never, Req>,
103): Effect.Effect<void, never, Req>;
104export function consumeBucketEvents<
105 B extends Bucket,
106 Req = never,
107 StreamReq = never,
108 const Events extends S3EventType[] = S3EventType[],
109>(
110 bucket: B,
111 propsOrHandler:
112 | NotificationsProps<Events>
113 | ((
114 stream: Stream.Stream<BucketNotification, never, StreamReq>,
115 ) => Effect.Effect<void, never, Req>),
116 maybeHandler?: (
117 stream: Stream.Stream<BucketNotification, never, StreamReq>,
118 ) => Effect.Effect<void, never, Req>,
119) {
120 const props: NotificationsProps<Events> =
121 typeof propsOrHandler === "function" ? {} : propsOrHandler;
122 const handler =
123 typeof propsOrHandler === "function" ? propsOrHandler : maybeHandler!;
124 return BucketEventSource.use((source) => source(bucket, props, handler));
125}