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 messages arrive as an Effect Stream:

import * as AWS from "alchemy/AWS";
import * as Effect from "effect/Effect";
import * as Stream from "effect/Stream";

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, (records) =>
      records.pipe(
        Stream.map((record) => record.body),
        Stream.runForEach((body) => Effect.log(`order: ${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 turns each invocation’s batch into records.

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).

It’s a real Effect Stream

records is the same Stream from any Effect program, so every combinator composes. 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" },
  (stream) =>
    stream.pipe(
      Stream.map((record) =>
        JSON.stringify({
          eventName: record.eventName,
          keys: record.dynamodb.Keys,
        }),
      ),
      Stream.run(sink),
    ),
);

Stream.throttle or Stream.groupedWithin drop in anywhere along the chain. 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/Cloudflare";
import * as Effect from "effect/Effect";
import * as Stream from "effect/Stream";
import * as HttpServerResponse from "effect/unstable/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" },
  (records) => Stream.runForEach(records, (record) => 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,
  },
  (stream) => Stream.runForEach(stream, (record) => 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:

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

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

Not every source is a Stream

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 a Stream per batch
yield* AWS.SQS.consumeQueueMessages(queue, fn);
yield* AWS.Kinesis.consumeStreamRecords(stream, props, fn);
yield* AWS.DynamoDB.consumeTableChanges(table, props, fn);
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.