1---
2title: Kinesis
3description: Add a Kinesis Data Stream, publish records from one Lambda, and consume them in order from another — wired through the same Stream-shaped event source.
4---
5
6**Kinesis Data Streams** are the tool for **ordered**, sharded,
7high-throughput data: delivery in order within a partition,
8replay from arbitrary points in time, and multiple independent
9consumers reading the same records. In Alchemy it's the `Stream`
10resource plus producer bindings (`PutRecord`, `PutRecords`, and
11the sink-shaped `StreamSink`) and a consumer side:
12`Kinesis.consumeStreamRecords` subscribes a Lambda to the stream
13as a typed Effect `Stream`, creating the event source mapping
14and granting the read permissions for you. Reach for
15[SQS](/aws/messaging/sqs) instead when the work is unordered.
16
17The walkthrough below builds on the [Lambda](/aws/compute/lambda),
18[DynamoDB](/aws/data/dynamodb), and [SQS](/aws/messaging/sqs) pages: it adds a
19Kinesis stream, publishes to it from the producer Lambda, and
20consumes it from a new worker.
21
22## Add the stream
23
24Lift the Kinesis stream into a shared module the same way you
25did with the SQS queue, so both producer and consumer can
26import the same handle:
27
28```typescript
29// src/stream.ts
30import * as Kinesis from "alchemy/AWS/Kinesis";
31
32export const Events = Kinesis.Stream("Events", {
33 streamMode: "ON_DEMAND",
34});
35```
36
37`"ON_DEMAND"` lets AWS auto-scale the shards based on load —
38fine for the example. Switch to `"PROVISIONED"` with an explicit
39`shardCount` when you know your throughput.
40
41## Bind PutRecord on the producer
42
43```diff lang="typescript"
44// src/api.ts
45import * as DynamoDB from "alchemy/AWS/DynamoDB";
46+import * as Kinesis from "alchemy/AWS/Kinesis";
47import * as S3 from "alchemy/AWS/S3";
48import * as SQS from "alchemy/AWS/SQS";
49import { Jobs } from "./queue.ts";
50+import { Events } from "./stream.ts";
51// ...
52
53const queue = yield* Jobs;
54+const events = yield* Events;
55const sendMessage = yield* SQS.SendMessage(queue);
56+const putRecord = yield* Kinesis.PutRecord(events);
57```
58
59The shape mirrors every other binding you've seen: `PutRecord(stream)`
60returns a callable Effect, plus an IAM policy with
61`kinesis:PutRecord` scoped to this stream's ARN.
62
63## Publish a record
64
65Use `putRecord` inside the `PUT /items/:id` route — alongside
66the SQS publish — to send an ordered audit record for each
67write:
68
69```diff lang="typescript"
70yield* sendMessage({
71 MessageBody: JSON.stringify({ type: "job.created", id, content }),
72});
73+yield* putRecord({
74+ PartitionKey: id,
75+ Data: new TextEncoder().encode(
76+ JSON.stringify({ type: "job.created", id, content, at: Date.now() }),
77+ ),
78+});
79```
80
81`PartitionKey` controls which shard a record lands in. All
82records with the same partition key are delivered in order to
83the same consumer instance — the entity id is exactly what you
84want here.
85
86`Data` is an arbitrary `Uint8Array`, not a string. JSON-encoded
87UTF-8 bytes are the most common payload but you can ship Avro,
88protobuf, or raw blobs without changing the binding.
89
90## Provide the runtime layer
91
92```diff lang="typescript"
93}).pipe(
94 Effect.provide(
95 Layer.mergeAll(
96 // ... existing layers ...
97+ Kinesis.PutRecordHttp,
98 SQS.SendMessageHttp,
99 ),
100 ),
101),
102```
103
104## Stand up a consumer Lambda
105
106Create `src/audit.ts` with a bare consumer shell — no event
107source yet, just the empty class importing the shared stream
108handle:
109
110```typescript
111// src/audit.ts
112import * as AWS from "alchemy/AWS";
113import * as Effect from "effect/Effect";
114import { Events } from "./stream.ts";
115
116export default class Audit extends AWS.Lambda.Function<Audit>()(
117 "Audit",
118 { main: import.meta.url },
119 Effect.gen(function* () {
120 const events = yield* Events;
121 return {};
122 }),
123) {}
124```
125
126## Subscribe to records
127
128`Kinesis.consumeStreamRecords(stream, ...)` mirrors the
129[SQS `consumeQueueMessages(...)`](/aws/messaging/sqs) shape.
130Add the smallest possible subscription — log each record's
131partition key:
132
133```diff lang="typescript"
134// src/audit.ts
135import * as AWS from "alchemy/AWS";
136+import * as Kinesis from "alchemy/AWS/Kinesis";
137+import * as Console from "effect/Console";
138import * as Effect from "effect/Effect";
139+import * as Stream from "effect/Stream";
140import { Events } from "./stream.ts";
141
142export default class Audit extends AWS.Lambda.Function<Audit>()(
143 "Audit",
144 { main: import.meta.url },
145 Effect.gen(function* () {
146 const events = yield* Events;
147+
148+ yield* Kinesis.consumeStreamRecords(events, {
149+ startingPosition: "LATEST",
150+ batchSize: 100,
151+ }, (stream) =>
152+ stream.pipe(
153+ Stream.runForEach((record) =>
154+ Console.log(`[${record.kinesis.partitionKey}]`),
155+ ),
156+ ),
157+ );
158
159 return {};
160 }),
161) {}
162```
163
164`batchSize: 100` lets Lambda gather up to 100 records per
165invocation; the whole batch is delivered as one stream emission
166sequence, then the stream completes and Lambda checkpoints.
167`startingPosition: "LATEST"` skips history — switch to
168`"TRIM_HORIZON"` to replay the 24-hour retention window.
169
170## Decode the binary payload
171
172Kinesis record data arrives base64-encoded inside the Lambda
173envelope (an artifact of the event shape, not the stream
174itself). Decode it back to bytes and on to a string, and run
175the work as an Effect per record:
176
177```diff lang="typescript"
178yield* Kinesis.consumeStreamRecords(events, {
179 startingPosition: "LATEST",
180 batchSize: 100,
181}, (stream) =>
182 stream.pipe(
183- Stream.runForEach((record) =>
184- Console.log(`[${record.kinesis.partitionKey}]`),
185- ),
186+ Stream.mapEffect((record) =>
187+ Effect.gen(function* () {
188+ const text = new TextDecoder().decode(
189+ Buffer.from(record.kinesis.data, "base64"),
190+ );
191+ yield* Console.log(`[${record.kinesis.partitionKey}] ${text}`);
192+ }),
193+ ),
194+ Stream.runDrain,
195 ),
196);
197```
198
199## Provide the runtime layer
200
201`Lambda.StreamEventSource` is the binding that enables the event
202source mapping and grants `kinesis:GetRecords`,
203`GetShardIterator`, `DescribeStream`, and `ListShards` on the
204stream ARN. Provide it at the bottom of the function:
205
206```diff lang="typescript"
207 return {};
208- }),
209+ }).pipe(Effect.provide(AWS.Lambda.StreamEventSource)),
210) {}
211```
212
213## Wire the consumer into the Stack
214
215```diff lang="typescript"
216// alchemy.run.ts
217import Api from "./src/api.ts";
218+import Audit from "./src/audit.ts";
219import Worker from "./src/worker.ts";
220
221Effect.gen(function* () {
222 const api = yield* Api;
223 yield* Worker;
224+ yield* Audit;
225 return { url: api.functionUrl };
226}),
227```
228
229## Deploy
230
231```sh
232bun alchemy deploy
233```
234
235Three functions now coexist: `Api` produces records, `Worker`
236consumes the SQS queue, and `Audit` consumes the Kinesis stream.
237Each one gets a least-privilege policy that names exactly the
238resources its capability and `consume*` calls touched — no
239`"*"`, no IAM JSON to maintain.
240
241## Verify
242
243```sh
244curl -X PUT --data 'audit me' "$URL/items/k1"
245bun alchemy logs Audit --tail
246```
247
248You'll see a line like:
249
250```
251[k1] {"type":"job.created","id":"k1","content":"audit me","at":1718000000000}
252```
253
254Subsequent writes against the same `id` (`k1`) all land in the
255same shard and arrive in order at the same `Audit` invocation
256context — that's the ordering guarantee Kinesis provides per
257partition key.
258
259## Bonus: in-order side effects
260
261Because each batch is delivered as a `Stream`, **per-shard ordering
262is preserved end-to-end**. If you need strict ordering for side
263effects, drop the concurrency to 1 and let the stream do the work:
264
265```typescript
266stream.pipe(
267 Stream.mapEffect(
268 (record) => persistInOrder(record).pipe(Effect.orDie),
269 { concurrency: 1 },
270 ),
271 Stream.runDrain,
272);
273```
274
275Or fan out per-partition with `Stream.groupByKey(record => record.kinesis.partitionKey)`
276when ordering only matters within a key.
277
278## What you've built
279
280If you followed the whole section, you've gone from a single
281Lambda to a small event-driven system: a public HTTP API backed
282by S3 blobs and a DynamoDB table, table change events on a
283stream, an SQS work queue, and a high-throughput Kinesis
284pipeline — all wired through the same binding pattern, with IAM
285policies that match exactly the operations you call. The same
286`Stream`-shaped consumer surface drives every event source, so
287adding more is a matter of declaring the resource and pointing
288the matching `consume*` helper at it.
289
290## Where next
291
292- [Process DynamoDB Streams](/aws/messaging/dynamodb-streams) —
293 change data capture with the same consumer surface.
294- [React to S3 events](/aws/messaging/s3-events) — bucket
295 notifications as an Effect `Stream`.
296- [`Stream` reference](/providers/aws/kinesis/stream) —
297 provisioned vs on-demand mode, retention, and every other
298 prop.