Event Sources

An Event Source is a Binding that runs your Function when something happens on a resource. A message lands on a queue, an object lands in a bucket, a row changes in a table. You name the resource and a handler, and the handler runs once per batch:

import * as AWS from "@alchemy.run/aws";
import * as Effect from "effect/Effect";

export const Orders = AWS.SQS.Queue("Orders");

export default AWS.Lambda.Function(
  "OrderProcessor",
  { main: import.meta.url },
  Effect.gen(function* () {
    yield* AWS.SQS.consumeQueueMessages(
      Orders,
      Effect.fn(function* (event) {
        for (const record of event.Records) {
          yield* Effect.log(`order: ${record.body}`);
        }
      }),
    );
  }).pipe(Effect.provide(AWS.Lambda.QueueEventSource)),
);

A pure consumer returns nothing from its constructor. It only declares what it listens to.

What one call wires

In traditional IaC, running code when a message lands takes three pieces: permission to receive and delete messages, a mapping that tells the queue to invoke the Function, and a handler that unpacks the raw event:

// infrastructure: grant access and wire the trigger
policy: { Action: ["sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes"], Resource: [queue.arn] },
eventSourceMapping: { eventSourceArn: queue.arn, functionName: fn.name },

// handler: unpack the batch by hand
export const handler = async (event) => {
  for (const record of event.Records) {
    console.log(record.body);
  }
};

consumeQueueMessages does all three. At deploy time it attaches that statement to the Function’s role, scoped to the queue’s ARN, and creates the event source mapping that points the queue at this Function. At runtime it registers the listener that hands each invocation’s batch to your handler, inside a process Orders consumer span.

consumeQueueMessages is the contract and AWS.Lambda.QueueEventSource is the Layer that satisfies it, the same split every binding has (a contract and a Layer).

Report the messages that failed

The handler takes the batch Lambda delivers and returns Lambda’s batch response. Name the messages that failed in batchItemFailures, and only those are retried:

yield* AWS.SQS.consumeQueueMessages(
  Orders,
  Effect.fn(function* (event) {
    const failed = yield* Effect.filter(
      event.Records,
      (record) =>
        processOrder(record.body).pipe(
          Effect.as(false),
          Effect.catchCause(() => Effect.succeed(true)),
        ),
      { concurrency: "unbounded" },
    );
    return {
      batchItemFailures: failed.map((record) => ({ itemIdentifier: record.messageId })),
    };
  }),
);

Return nothing when every message succeeded. If the handler fails, the whole batch is retried. Kinesis and DynamoDB Streams take the same response, naming the first record that failed so Lambda checkpoints before it.

Compose with a Stream

The batch is plain data, so any Effect works on it. Map table changes to JSON and run them straight into a queue:

const sink = yield* AWS.SQS.QueueSink(Outbound);

yield* AWS.DynamoDB.consumeTableChanges(
  Jobs,
  { streamViewType: "NEW_AND_OLD_IMAGES", startingPosition: "TRIM_HORIZON" },
  Effect.fn(function* (event) {
    yield* Stream.fromIterable(event.Records).pipe(
      Stream.map((record) =>
        JSON.stringify({
          eventName: record.eventName,
          keys: record.dynamodb.Keys,
        }),
      ),
      Stream.run(sink),
    );
  }),
);

That sink is a Sink, the write-side dual of an event source. Its permissions are generated the same way.

The same shape on Cloudflare

Consume a Cloudflare queue the same way. The consumer settings go on the call, and the handler receives a Stream of typed messages:

import * as Cloudflare from "@alchemy.run/cloudflare";
import * as Effect from "effect/Effect";
import * as Stream from "effect/Stream";
import * as HttpServerResponse from "effect/http/HttpServerResponse";

export const Orders = Cloudflare.Queues.Queue("Orders");

export default Cloudflare.Worker(
  "OrderProcessor",
  { main: import.meta.url },
  Effect.gen(function* () {
    yield* Cloudflare.Queues.consumeQueueMessages<{ text: string }>(
      Orders,
      { batchSize: 10, maxRetries: 3, retryDelay: "1 second" },
      (stream) =>
        Stream.runForEach(stream, (msg) => Effect.log(msg.body.text)),
    );

    return {
      fetch: Effect.gen(function* () {
        return HttpServerResponse.text("ok");
      }),
    };
  }).pipe(Effect.provide(Cloudflare.Queues.EventSourceLive)),
);

There is no IAM here. At deploy time the call creates the queue consumer that dispatches batches to this Worker. At runtime it registers the queue listener.

Batching and retries

Set batchSize and maximumBatchingWindow on the call to shape each invocation’s batch:

yield* AWS.SQS.consumeQueueMessages(
  Orders,
  { batchSize: 10, maximumBatchingWindow: "5 seconds" },
  Effect.fn(function* (event) {
    for (const record of event.Records) {
      yield* Effect.log(record.body);
    }
  }),
);

If your handler fails, the queue redelivers the batch. Stream sources like Kinesis and DynamoDB add settings that bound a poison record’s blast radius:

yield* AWS.DynamoDB.consumeTableChanges(
  Jobs,
  {
    streamViewType: "NEW_AND_OLD_IMAGES",
    startingPosition: "TRIM_HORIZON",
    bisectBatchOnFunctionError: true,
    maximumRetryAttempts: 3,
  },
  Effect.fn(function* (event) {
    for (const record of event.Records) {
      yield* Effect.log(record.eventName);
    }
  }),
);

On Cloudflare, set the retry policy on the call. A batch is acked when your handler succeeds, retried maxRetries times when it fails, then dead-lettered:

const deadLetters = yield* Cloudflare.Queues.Queue("OrdersDeadLetters");

yield* Cloudflare.Queues.consumeQueueMessages<{ text: string }>(
  Orders,
  { maxRetries: 3, retryDelay: "1 second", deadLetterQueue: deadLetters },
  (stream) => Stream.runForEach(stream, (msg) => Effect.log(msg.body.text)),
);

Nothing reads the dead-letter queue unless you consume it, so a failed message expires there silently. Subscribe to it the same way (example).

Call msg.ack() or msg.retry() inside the handler to decide per message.

Sources that deliver one event at a time

A source that delivers one event at a time hands your handler the event directly. A cron trigger passes one controller per fire:

export default Cloudflare.Worker(
  "Nightly",
  { main: import.meta.url },
  Effect.gen(function* () {
    yield* Cloudflare.Workers.cron("0 12 * * *", (controller) =>
      Effect.log(`scheduled at ${controller.scheduledTime}`),
    );

    return {
      fetch: Effect.gen(function* () {
        return HttpServerResponse.text("ok");
      }),
    };
  }).pipe(Effect.provide(Cloudflare.Workers.CronEventSourceLive)),
);

At deploy time the cron expression is attached to the Worker. At runtime the scheduled listener is registered. GitHub works the same way: consumeRepositoryEvents creates a repository webhook pointing at the Worker’s URL, and the runtime listener verifies each delivery’s HMAC signature before calling your handler.

Every event source

The shape never changes. Only the consume* callable, the record type, and the Layer you provide differ:

// AWS: handlers receive each batch and return Lambda's batch response
yield* AWS.SQS.consumeQueueMessages(queue, fn);
yield* AWS.Kinesis.consumeStreamRecords(stream, props, fn);
yield* AWS.DynamoDB.consumeTableChanges(table, props, fn);

// AWS: handlers receive a Stream of notifications
yield* AWS.S3.consumeBucketEvents(bucket, fn);
yield* AWS.SNS.consumeTopicNotifications(topic, fn);
yield* AWS.EventBridge.consumeBusEvents(bus, pattern, fn);

// Cloudflare
yield* Cloudflare.Queues.consumeQueueMessages(queue, fn);
yield* Cloudflare.Workers.cron(expression, fn);
yield* GitHub.consumeRepositoryEvents(props, fn);

EventBridge can also route matching events to another resource instead of consuming them locally:

yield* AWS.EventBridge.events(bus, { source: ["my.app"] }).toQueue(queue);

Each source has its own guide. On AWS: SQS, Kinesis, DynamoDB Streams, S3 Events, SNS, and EventBridge. On Cloudflare: Queues, Cron, and GitHub Events.

Where next

• Sinks — the write-side dual: resources as Effect Sinks, with batching and IAM generated the same way.
• Bindings — the deploy-time mechanics every event source is built on.
• Runtime — the Effectful Constructor these examples live inside.