BucketEventSource

Source: src/AWS/S3/BucketEventSource.ts

Event source that streams a bucket’s notifications (object created, object removed, …) into the host Lambda Function. Usually consumed through the consumeBucketEvents helper, which provisions the bucket-notification subscription at deploy time and registers the stream handler at runtime. Provide the implementation with Effect.provide(Lambda.BucketEventSource).

Consuming Bucket Events

export default MyFunction.make(
  { main: import.meta.url },
  Effect.gen(function* () {
    const bucket = yield* AWS.S3.Bucket("UploadsBucket");
    const putObject = yield* AWS.S3.PutObject(bucket);

    // filter to `incoming/` so the derived `processed/` write does not
    // re-trigger the subscription
    yield* AWS.S3.consumeBucketEvents(
      bucket,
      { events: ["s3:ObjectCreated:*"], prefix: "incoming/" },
      (stream) =>
        stream.pipe(
          Stream.runForEach((event) =>
            putObject({
              Key: `processed/${event.key.slice("incoming/".length)}`,
              Body: JSON.stringify({ key: event.key, size: event.size }),
            }).pipe(Effect.orDie),
          ),
        ),
    );

    return {
      fetch: Effect.succeed(HttpServerResponse.text("ok")),
    };
  }).pipe(
    Effect.provide(
      Layer.mergeAll(Lambda.BucketEventSource, AWS.S3.PutObjectHttp),
    ),
  ),
);