Sinks

A Sink is the dual of an Event Source: a Binding for writing that exposes a resource as an Effect Sink. Where an Event Source hands you records as a Stream, a Sink accepts a Stream and drains it into the resource’s batch API (SendMessageBatch, PublishBatch, PutRecords) — emitting the minimal IAM to go with it.

A Sink in a constructor

Like every Binding, a Sink is declared in the Effectful Constructor and used in the interface it returns:

import * as AWS from "alchemy/AWS";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Stream from "effect/Stream";
import { HttpServerRequest } from "effect/http/HttpServerRequest";
import * as HttpServerResponse from "effect/http/HttpServerResponse";
import { OutboundQueue } from "./queue.ts";

export default class Publisher extends AWS.Lambda.Function<Publisher>()(
  "Publisher",
  { main: import.meta.url, functionUrl: true },
  Effect.gen(function* () {
    const sink = yield* AWS.SQS.QueueSink(OutboundQueue);
    // → Sink<void, string, readonly string[], never>
    // → policy: sqs:SendMessage + sqs:SendMessageBatch on OutboundQueue's ARN

    return {
      fetch: Effect.gen(function* () {
        const request = yield* HttpServerRequest;
        const body = (yield* request.json) as { messages: string[] };

        yield* Stream.fromIterable(body.messages).pipe(Stream.run(sink));

        return yield* HttpServerResponse.json({
          ok: true,
          count: body.messages.length,
        });
      }),
    };
  }).pipe(
    Effect.provide(
      AWS.SQS.QueueSinkHttp.pipe(
        Layer.provideMerge(AWS.SQS.SendMessageBatchHttp),
      ),
    ),
  ),
) {}

QueueSink is the contract; the provided AWS.SQS.QueueSinkHttp Layer implements it over SendMessageBatch — itself a contract, satisfied by SendMessageBatchHttp. That is why the provide chains a Layer.provideMerge: implementation Layers can depend on other contracts. The Layer attaches the least-privilege policy statements to the Function’s role; the yield* returns a plain Effect Sink — no QUEUE_URL env var to read, no SendMessageBatchCommand chunking by hand. The contract-and-Layer split is the same rule every Binding follows — see Bindings › A contract and a Layer.

Element types

Sinks are typed by the element they accept, so the type system stops you from feeding the wrong shape in:

Sink	Element	Batch API	Grants
SQS.QueueSink(queue)	string	SendMessageBatch	sqs:SendMessage, sqs:SendMessageBatch
SNS.TopicSink(topic)	string	PublishBatch	sns:Publish
Kinesis.StreamSink(stream)	PutRecordsRequestEntry	PutRecords	kinesis:PutRecords
Cloudflare.Queues.QueueSink(queue)	message body (any JSON value)	sendBatch	queue binding on the Worker

Every AWS policy is scoped to the exact resource ARN; the Cloudflare sink binds only the one queue into the Worker.

QueueSink and TopicSink take message bodies as string. StreamSink takes raw PutRecordsRequestEntry values, so you keep control of PartitionKey:

export default class Ingest extends AWS.Lambda.Function<Ingest>()(
  "Ingest",
  { main: import.meta.url, functionUrl: true },
  Effect.gen(function* () {
    const sink = yield* AWS.Kinesis.StreamSink(ClickStream);

    return {
      fetch: Effect.gen(function* () {
        const request = yield* HttpServerRequest;
        const body = (yield* request.json) as {
          records: Array<{ partitionKey: string; data: string }>;
        };

        yield* Stream.fromIterable(
          body.records.map((record) => ({
            PartitionKey: record.partitionKey,
            Data: new TextEncoder().encode(record.data),
          })),
        ).pipe(Stream.run(sink));

        return yield* HttpServerResponse.json({ ok: true });
      }),
    };
  }),
  // (layers elided — same shape as Publisher above:
  //  AWS.Kinesis.StreamSinkHttp over AWS.Kinesis.PutRecordsHttp)
) {}

Batching

A Sink drains its input chunk by chunk — each chunk of the stream becomes one call to the underlying batch API. To control batch size, rechunk the stream before running it into the sink:

yield* Stream.fromIterable(messages).pipe(
  Stream.rechunk(10), // one SendMessageBatch call per 10 messages
  Stream.run(sink),
);

Source → transform → sink

Where the model pays off is when you compose a Sink with an Event Source. Because every step is just Stream / Sink, the whole pipe-and-transform pipeline is one expression inside the constructor:

export default class OrderRelay extends AWS.Lambda.Function<OrderRelay>()(
  "OrderRelay",
  { main: import.meta.url },
  Effect.gen(function* () {
    const sink = yield* AWS.SQS.QueueSink(OutboundQueue);

    yield* AWS.DynamoDB.consumeTableChanges<Order>(
      OrdersTable,
      {
        streamViewType: "NEW_AND_OLD_IMAGES",
        startingPosition: "LATEST",
      },
      (records) =>
        records.pipe(
          Stream.filterMap((r) => Option.fromNullable(r.dynamodb.NewImage)),
          Stream.filter((order) => order.status === "PAID"),
          Stream.map((order) => JSON.stringify({ orderId: order.id })),
          Stream.run(sink),
        ),
    );
  }).pipe(
    Effect.provide(
      Layer.mergeAll(AWS.Lambda.TableEventSource, AWS.SQS.QueueSinkHttp).pipe(
        Layer.provideMerge(AWS.SQS.SendMessageBatchHttp),
      ),
    ),
  ),
) {}

Two binding calls, one pipeline. The provided Layers attach both sides of the policy to the Function’s role — dynamodb:DescribeStream / GetRecords / GetShardIterator on OrdersTable’s stream, sqs:SendMessage / SendMessageBatch on OutboundQueue — and AWS.Lambda.TableEventSource wires the event-source mapping.

Drop in Effect.retry, Stream.throttle, Stream.groupedWithin, or Stream.mapEffect anywhere along the chain — they’re all the same Stream you’d write in a plain Effect program.

Cloudflare Queues

Cloudflare.Queues.QueueSink(queue) is the same shape on a Worker. Each element is one message body, and each stream chunk becomes one sendBatch call. Chunks over Cloudflare’s batch limits (100 messages / 256 KB) are split into consecutive batches, in order:

import * as Cloudflare from "alchemy/Cloudflare";
import * as Effect from "effect/Effect";
import * as Stream from "effect/Stream";
import * as HttpServerResponse from "effect/http/HttpServerResponse";
import { Clicks } from "./queues.ts";

export default Cloudflare.Worker(
  "Api",
  { main: import.meta.url },
  Effect.gen(function* () {
    const sink = yield* Cloudflare.Queues.QueueSink(Clicks);
    // → Sink<void, unknown, never, SendError, RuntimeContext>

    return {
      fetch: Effect.gen(function* () {
        yield* Stream.fromIterable(events).pipe(Stream.run(sink));
        return HttpServerResponse.empty({ status: 202 });
      }),
    };
  }).pipe(Effect.provide(Cloudflare.Queues.QueueSinkBinding)),
);

QueueSinkBinding layers the sink over the native producer binding (WriteQueueBinding), so there is nothing else to provide. It also works under alchemy dev, where the binding targets the local queue broker. QueueSinkHttp (scoped API token) and QueueSinkLocal (current credentials, for Actions) layer it over the matching WriteQueue implementations.

Paired with consumeQueueMessages, a queue-to-queue pipeline is one expression:

Effect.gen(function* () {
  const enriched = yield* Cloudflare.Queues.QueueSink(EnrichedClicks);

  yield* Cloudflare.Queues.consumeQueueMessages<Click>(Clicks, (messages) =>
    messages.pipe(
      Stream.map((message) => message.body),
      Stream.filter((click) => click.button === "buy"),
      Stream.map((click) => ({ ...click, receivedAt: Date.now() })),
      Stream.run(enriched),
    ),
  );
}).pipe(
  Effect.provide(
    Layer.mergeAll(
      Cloudflare.Queues.EventSourceLive,
      Cloudflare.Queues.QueueSinkBinding,
    ),
  ),
);

A failed sendBatch fails the handler, so Cloudflare retries the whole consumed batch. Messages already forwarded before the failure are sent again, which fits the queue’s at-least-once delivery.

Where next

• Event Sources — the read-side dual: resources as Streams.
• Bindings — the deploy-time mechanics every Sink shares.
• Layers — hide Sinks behind a service interface.