1---
2title: Sinks
3description: A Sink is the write-side dual of an Event Source — a Binding that exposes a resource as an Effect Sink, batching writes into the batch API with least-privilege IAM.
4---
5
6A **Sink** is the dual of an [Event Source](/infrastructure-as-effects/event-sources):
7a [Binding](/infrastructure-as-effects/binding) for *writing* that
8exposes a resource as an Effect
9[`Sink`](https://effect.website/docs/sink/introduction/). Where an
10Event Source hands you records as a `Stream`, a Sink accepts a
11`Stream` and drains it into the resource's batch API
12(`SendMessageBatch`, `PublishBatch`, `PutRecords`) — emitting the
13minimal IAM to go with it.
14
15## A Sink in a constructor
16
17Like every Binding, a Sink is declared in the
18[Effectful Constructor](/infrastructure-as-effects/runtime#the-effectful-constructor)
19and used in the interface it returns:
20
21```typescript
22import * as AWS from "alchemy/AWS";
23import * as Effect from "effect/Effect";
24import * as Layer from "effect/Layer";
25import * as Stream from "effect/Stream";
26import { HttpServerRequest } from "effect/http/HttpServerRequest";
27import * as HttpServerResponse from "effect/http/HttpServerResponse";
28import { OutboundQueue } from "./queue.ts";
29
30export default class Publisher extends AWS.Lambda.Function<Publisher>()(
31 "Publisher",
32 { main: import.meta.url, functionUrl: true },
33 Effect.gen(function* () {
34 const sink = yield* AWS.SQS.QueueSink(OutboundQueue);
35 // → Sink<void, string, readonly string[], never>
36 // → policy: sqs:SendMessage + sqs:SendMessageBatch on OutboundQueue's ARN
37
38 return {
39 fetch: Effect.gen(function* () {
40 const request = yield* HttpServerRequest;
41 const body = (yield* request.json) as { messages: string[] };
42
43 yield* Stream.fromIterable(body.messages).pipe(Stream.run(sink));
44
45 return yield* HttpServerResponse.json({
46 ok: true,
47 count: body.messages.length,
48 });
49 }),
50 };
51 }).pipe(
52 Effect.provide(
53 AWS.SQS.QueueSinkHttp.pipe(
54 Layer.provideMerge(AWS.SQS.SendMessageBatchHttp),
55 ),
56 ),
57 ),
58) {}
59```
60
61`QueueSink` is the contract; the provided `AWS.SQS.QueueSinkHttp`
62Layer implements it over `SendMessageBatch` — itself a contract,
63satisfied by `SendMessageBatchHttp`. That is why the provide chains a
64`Layer.provideMerge`: implementation Layers can depend on other
65contracts. The Layer attaches the least-privilege policy statements
66to the Function's role; the `yield*` returns a plain Effect `Sink` —
67no `QUEUE_URL` env var to read, no `SendMessageBatchCommand`
68chunking by hand. The contract-and-Layer split is the same rule
69every Binding follows — see
70[Bindings › A contract and a Layer](/infrastructure-as-effects/binding#a-contract-and-a-layer).
71
72## Element types
73
74Sinks are typed by the element they accept, so the type system stops
75you from feeding the wrong shape in:
76
77| Sink | Element | Batch API | IAM emitted |
78| ----------------------------- | ------------------------ | ------------------ | ---------------------------------------------- |
79| `SQS.QueueSink(queue)` | `string` | `SendMessageBatch` | `sqs:SendMessage`, `sqs:SendMessageBatch` |
80| `SNS.TopicSink(topic)` | `string` | `PublishBatch` | `sns:Publish` |
81| `Kinesis.StreamSink(stream)` | `PutRecordsRequestEntry` | `PutRecords` | `kinesis:PutRecords` |
82
83Every policy is scoped to the exact resource ARN.
84
85`QueueSink` and `TopicSink` take message bodies as `string`.
86`StreamSink` takes raw `PutRecordsRequestEntry` values, so you keep
87control of `PartitionKey`:
88
89```typescript
90export default class Ingest extends AWS.Lambda.Function<Ingest>()(
91 "Ingest",
92 { main: import.meta.url, functionUrl: true },
93 Effect.gen(function* () {
94 const sink = yield* AWS.Kinesis.StreamSink(ClickStream);
95
96 return {
97 fetch: Effect.gen(function* () {
98 const request = yield* HttpServerRequest;
99 const body = (yield* request.json) as {
100 records: Array<{ partitionKey: string; data: string }>;
101 };
102
103 yield* Stream.fromIterable(
104 body.records.map((record) => ({
105 PartitionKey: record.partitionKey,
106 Data: new TextEncoder().encode(record.data),
107 })),
108 ).pipe(Stream.run(sink));
109
110 return yield* HttpServerResponse.json({ ok: true });
111 }),
112 };
113 }),
114 // (layers elided — same shape as Publisher above:
115 // AWS.Kinesis.StreamSinkHttp over AWS.Kinesis.PutRecordsHttp)
116) {}
117```
118
119## Batching
120
121A Sink drains its input chunk by chunk — each chunk of the stream
122becomes **one** call to the underlying batch API. To control batch
123size, rechunk the stream before running it into the sink:
124
125```typescript
126yield* Stream.fromIterable(messages).pipe(
127 Stream.rechunk(10), // one SendMessageBatch call per 10 messages
128 Stream.run(sink),
129);
130```
131
132## Source → transform → sink
133
134Where the model pays off is when you compose a Sink with an
135[Event Source](/infrastructure-as-effects/event-sources). Because
136every step is just `Stream` / `Sink`, the whole
137pipe-and-transform pipeline is one expression inside the
138constructor:
139
140```typescript
141export default class OrderRelay extends AWS.Lambda.Function<OrderRelay>()(
142 "OrderRelay",
143 { main: import.meta.url },
144 Effect.gen(function* () {
145 const sink = yield* AWS.SQS.QueueSink(OutboundQueue);
146
147 yield* AWS.DynamoDB.consumeTableChanges<Order>(
148 OrdersTable,
149 {
150 streamViewType: "NEW_AND_OLD_IMAGES",
151 startingPosition: "LATEST",
152 },
153 (records) =>
154 records.pipe(
155 Stream.filterMap((r) => Option.fromNullable(r.dynamodb.NewImage)),
156 Stream.filter((order) => order.status === "PAID"),
157 Stream.map((order) => JSON.stringify({ orderId: order.id })),
158 Stream.run(sink),
159 ),
160 );
161 }).pipe(
162 Effect.provide(
163 Layer.mergeAll(AWS.Lambda.TableEventSource, AWS.SQS.QueueSinkHttp).pipe(
164 Layer.provideMerge(AWS.SQS.SendMessageBatchHttp),
165 ),
166 ),
167 ),
168) {}
169```
170
171Two binding calls, one pipeline. The provided Layers attach **both**
172sides of the policy to the Function's role —
173`dynamodb:DescribeStream` / `GetRecords` / `GetShardIterator` on
174`OrdersTable`'s stream, `sqs:SendMessage` / `SendMessageBatch` on
175`OutboundQueue` — and `AWS.Lambda.TableEventSource` wires the
176event-source mapping.
177
178Drop in `Effect.retry`, `Stream.throttle`, `Stream.groupedWithin`,
179or `Stream.mapEffect` anywhere along the chain — they're all the
180same `Stream` you'd write in a plain Effect program.
181
182## Where next
183
184- [Event Sources](/infrastructure-as-effects/event-sources) — the read-side dual: resources as `Stream`s.
185- [Bindings](/infrastructure-as-effects/binding) — the deploy-time mechanics every Sink shares.
186- [Layers](/infrastructure-as-effects/layers) — hide Sinks behind a service interface.