1import * as Effect from "effect/Effect";
2import * as Stream from "effect/Stream";
3import * as Binding from "../../Binding.ts";
4import type { Bucket } from "./Bucket.ts";
5import type {
6 BucketNotification,
7 NotificationsProps,
8} from "./BucketNotifications.ts";
9import type { S3EventType } from "./S3Event.ts";
10
11/**
12 * Event source that streams a bucket's notifications (object created, object
13 * removed, ...) into the host Lambda Function. Usually consumed through the
14 * `consumeBucketEvents` helper, which provisions the bucket-notification
15 * subscription at deploy time and registers the stream handler at runtime.
16 * Provide the implementation with `Effect.provide(Lambda.BucketEventSource)`.
17 * ### Consuming Bucket Events
18 * **Example:** Process Object-Created Events
19 * ```typescript
20 * export default MyFunction.make(
21 * { main: import.meta.url },
22 * Effect.gen(function* () {
23 * const bucket = yield* AWS.S3.Bucket("UploadsBucket");
24 * const putObject = yield* AWS.S3.PutObject(bucket);
25 *
26 * // filter to `incoming/` so the derived `processed/` write does not
27 * // re-trigger the subscription
28 * yield* AWS.S3.consumeBucketEvents(
29 * bucket,
30 * { events: ["s3:ObjectCreated:*"], prefix: "incoming/" },
31 * (stream) =>
32 * stream.pipe(
33 * Stream.runForEach((event) =>
34 * putObject({
35 * Key: `processed/${event.key.slice("incoming/".length)}`,
36 * Body: JSON.stringify({ key: event.key, size: event.size }),
37 * }).pipe(Effect.orDie),
38 * ),
39 * ),
40 * );
41 *
42 * return {
43 * fetch: Effect.succeed(HttpServerResponse.text("ok")),
44 * };
45 * }).pipe(
46 * Effect.provide(
47 * Layer.mergeAll(Lambda.BucketEventSource, AWS.S3.PutObjectHttp),
48 * ),
49 * ),
50 * );
51 * ```
52 *
53 * ### Reading the Event's Object Version
54 * **Example:** Read the version that triggered a creation event
55 * ```typescript
56 * const getObject = yield* AWS.S3.GetObject(bucket);
57 * yield* AWS.S3.consumeBucketEvents(
58 * bucket,
59 * { events: ["s3:ObjectCreated:*"], prefix: "incoming/" },
60 * (events) => events.pipe(
61 * Stream.runForEach((event) =>
62 * getObject({ Key: event.key, VersionId: event.versionId }).pipe(
63 * Effect.flatMap(({ Body }) => Stream.runDrain(Body!)),
64 * Effect.orDie,
65 * ),
66 * ),
67 * ),
68 * );
69 * // Provide Lambda.BucketEventSource and AWS.S3.GetObjectHttp on the function.
70 * ```
71 *
72 * Passing `event.versionId` reads the triggering data version even if the key
73 * has since been overwritten. Unversioned events omit it and read the current
74 * object. Removal events can identify a delete marker or an already-deleted
75 * version; they may omit `size` and `eTag` and are not object-read signals.
76 * `sequencer`, when present, orders events only for the same object key.
77 *
78 * @binding
79 */
80export interface BucketEventSource extends Binding.Service<
81 BucketEventSource,
82 "BucketNotificationStream",
83 BucketEventSourceService
84> {}
85
86export const BucketEventSource = Binding.Service<BucketEventSource>(
87 "BucketNotificationStream",
88);
89
90export type BucketEventSourceService = <
91 Events extends S3EventType[],
92 StreamReq = never,
93 Req = never,
94>(
95 bucket: Bucket,
96 props: NotificationsProps<Events>,
97 process: (
98 stream: Stream.Stream<BucketNotification, never, StreamReq>,
99 ) => Effect.Effect<void, never, Req>,
100) => Effect.Effect<void, never, never>;