S3 events

A Bucket can tell you when something happens to its objects. Each time an object is created, deleted or restored, S3 sends a notification naming the Bucket, the key and the event. A Function that receives them can make a thumbnail of every uploaded image, index every new document or scan every file for viruses, without the uploader doing anything but upload.

S3 invokes the Function

Notifications are pushed. S3 invokes the Function directly, with a batch of notifications, shortly after each change.

S3 answers the upload first and invokes the Function shortly after. A notification can arrive twice, so writing the thumbnail again must be safe.

S3 delivers each notification at least once, and occasionally more than once, so processing the same key twice should be safe. Write the thumbnail under a key derived from the original’s, and a second run just replaces it.

Subscribe a Function

Subscribe from the Function’s Construction phase. Notifications arrive as a Stream:

src/thumbnails.ts

export default class Thumbnails extends AWS.Lambda.Function<Thumbnails>()(
  "Thumbnails",
  { main: import.meta.url, functionUrl: false },
  Effect.gen(function* () {
    const uploads = yield* Uploads;
    yield* AWS.S3.consumeBucketEvents(uploads, (stream) =>
      stream.pipe(Stream.runForEach((event) => Console.log(`${event.type}: ${event.key}`))),
    );
    return {};
  }).pipe(Effect.provide(AWS.Lambda.BucketEventSource)),
) {}

Deploy adds the notification to the Bucket and the Lambda permission that lets S3 invoke the Function. Each notification carries its type, bucket, key, size and eTag.

Choose which events

Without a filter, the Function receives every kind of notification: created, removed, restored, replicated. Name the ones you want:

src/thumbnails.ts

yield* AWS.S3.consumeBucketEvents(uploads, { events: ["s3:ObjectCreated:*"] }, (stream) =>
  stream.pipe(Stream.runForEach((event) => Console.log(event.key))),
);

The event names are checked against S3’s own list, so a typo fails to compile.

Process each object

To read each new object, bind S3.GetObject to the same Bucket and run an Effect per notification:

src/thumbnails.ts

const getObject = yield* AWS.S3.GetObject(uploads);

yield* AWS.S3.consumeBucketEvents(uploads, { events: ["s3:ObjectCreated:*"] }, (stream) =>
  stream.pipe(
    Stream.mapEffect(
      (event) =>
        getObject({ Key: event.key }).pipe(
          Effect.flatMap((object) => makeThumbnail(event.key, object.Body!)),
          Effect.orDie,
        ),
      { concurrency: 4 },
    ),
    Stream.runDrain,
  ),
);

Provide AWS.S3.GetObjectHttp next to the event source. Here four objects are processed at a time. Leave out concurrency to take them one by one.

A batch is one invocation. The stream ends when the batch is done, and the next invocation starts a new one.

Don’t trigger yourself

A Function that writes its output into the Bucket it watches will be notified about its own output. A thumbnail of an upload produces a new object, which produces a notification, which produces a thumbnail of the thumbnail.

Write outputs to a second Bucket, or keep inputs and outputs under separate prefixes and ignore the output prefix:

src/thumbnails.ts

stream.pipe(
  Stream.filter((event) => !event.key.startsWith("thumbnails/")),
  Stream.runForEach((event) => Console.log(event.key)),
)