AWS.Kinesis reference

DescribeAccountSettings

Source: src/AWS/Kinesis/DescribeAccountSettings.ts

Runtime binding for kinesis:DescribeAccountSettings.

An account-level operation (no stream argument) that reports the account’s Kinesis settings, such as on-demand stream count quotas. Provide the implementation with Effect.provide(AWS.Kinesis.DescribeAccountSettingsHttp).

DescribeAccountSettings: Account Settings

// init — account-level binding takes no resource
const describeAccountSettings = yield* AWS.Kinesis.DescribeAccountSettings();

// runtime
const settings = yield* describeAccountSettings();

DescribeLimits

Source: src/AWS/Kinesis/DescribeLimits.ts

Runtime binding for kinesis:DescribeLimits.

An account-level operation (no stream argument) that reports shard quotas and current usage for the region. Provide the implementation with Effect.provide(AWS.Kinesis.DescribeLimitsHttp).

DescribeLimits: Account Settings

// init — account-level binding takes no resource
const describeLimits = yield* AWS.Kinesis.DescribeLimits();

// runtime
const limits = yield* describeLimits();
const headroom = (limits.ShardLimit ?? 0) - (limits.OpenShardCount ?? 0);

DescribeStream

Source: src/AWS/Kinesis/DescribeStream.ts

Runtime binding for kinesis:DescribeStream.

Bind this operation to a Stream to read its full description, including the shard map — the stream name is injected automatically. For status and counts without the shard list, prefer AWS.Kinesis.DescribeStreamSummary. Provide the implementation with Effect.provide(AWS.Kinesis.DescribeStreamHttp).

DescribeStream: Inspecting Streams

// init
const describeStream = yield* AWS.Kinesis.DescribeStream(stream);

// runtime
const result = yield* describeStream();
const status = result.StreamDescription.StreamStatus;
const shards = result.StreamDescription.Shards;

DescribeStreamConsumer

Source: src/AWS/Kinesis/DescribeStreamConsumer.ts

Runtime binding for kinesis:DescribeStreamConsumer.

Bind this operation to a StreamConsumer to read the enhanced fan-out consumer’s status and ARN — the consumer ARN is injected automatically. Provide the implementation with Effect.provide(AWS.Kinesis.DescribeStreamConsumerHttp).

DescribeStreamConsumer: Enhanced Fan-Out

const consumer = yield* AWS.Kinesis.StreamConsumer("Analytics", {
  streamArn: stream.streamArn,
});
// init
const describeStreamConsumer =
  yield* AWS.Kinesis.DescribeStreamConsumer(consumer);

// runtime
const result = yield* describeStreamConsumer();
const status = result.ConsumerDescription.ConsumerStatus;

DescribeStreamSummary

Source: src/AWS/Kinesis/DescribeStreamSummary.ts

Runtime binding for kinesis:DescribeStreamSummary.

Bind this operation to a Stream to read its status, mode, retention, encryption, and open shard count without paginating the full shard map. Provide the implementation with Effect.provide(AWS.Kinesis.DescribeStreamSummaryHttp).

DescribeStreamSummary: Inspecting Streams

// init
const describeStreamSummary = yield* AWS.Kinesis.DescribeStreamSummary(stream);

// runtime
const result = yield* describeStreamSummary();
const summary = result.StreamDescriptionSummary;
yield* Effect.log(`${summary.StreamStatus}: ${summary.OpenShardCount} shards`);

GetRecords

Source: src/AWS/Kinesis/GetRecords.ts

Runtime binding for kinesis:GetRecords.

Bind this operation to a Stream to read records from a shard using an iterator obtained via AWS.Kinesis.GetShardIterator. Provide the implementation with Effect.provide(AWS.Kinesis.GetRecordsHttp). For push-based processing, prefer consumeStreamRecords (a Lambda event source) over manual polling.

GetRecords: Reading Records

// init — bind the operations to the stream
const getShardIterator = yield* AWS.Kinesis.GetShardIterator(stream);
const getRecords = yield* AWS.Kinesis.GetRecords(stream);

// runtime — obtain an iterator, then read
const iterator = yield* getShardIterator({
  ShardId: shardId,
  ShardIteratorType: "TRIM_HORIZON",
});
const result = yield* getRecords({
  ShardIterator: iterator.ShardIterator!,
});
for (const record of result.Records ?? []) {
  yield* Effect.log(record.PartitionKey);
}

GetResourcePolicy

Source: src/AWS/Kinesis/GetResourcePolicy.ts

Runtime binding for kinesis:GetResourcePolicy.

Bind this operation to a Stream to read the resource policy attached to it (set via the stream’s resourcePolicy prop) — the stream ARN is injected automatically. Provide the implementation with Effect.provide(AWS.Kinesis.GetResourcePolicyHttp).

GetResourcePolicy: Inspecting Streams

// init
const getResourcePolicy = yield* AWS.Kinesis.GetResourcePolicy(stream);

// runtime
const result = yield* getResourcePolicy();
const policy = JSON.parse(result.Policy);

GetShardIterator

Source: src/AWS/Kinesis/GetShardIterator.ts

Runtime binding for kinesis:GetShardIterator.

Bind this operation to a Stream to obtain a shard iterator — the starting position for reading records with AWS.Kinesis.GetRecords. The stream name is injected automatically. Provide the implementation with Effect.provide(AWS.Kinesis.GetShardIteratorHttp).

GetShardIterator: Reading Records

// init
const getShardIterator = yield* AWS.Kinesis.GetShardIterator(stream);

// runtime
const iterator = yield* getShardIterator({
  ShardId: shardId,
  ShardIteratorType: "LATEST",
});
// pass iterator.ShardIterator to getRecords

ListShards

Source: src/AWS/Kinesis/ListShards.ts

Runtime binding for kinesis:ListShards.

Bind this operation to a Stream to enumerate its shards — typically the first step before obtaining a shard iterator and reading records. Provide the implementation with Effect.provide(AWS.Kinesis.ListShardsHttp).

ListShards: Inspecting Streams

// init
const listShards = yield* AWS.Kinesis.ListShards(stream);

// runtime
const result = yield* listShards();
const shardIds = (result.Shards ?? []).map((shard) => shard.ShardId);

ListStreamConsumers

Source: src/AWS/Kinesis/ListStreamConsumers.ts

Runtime binding for kinesis:ListStreamConsumers.

Bind this operation to a Stream to enumerate the enhanced fan-out consumers registered on it — the stream ARN is injected automatically. Provide the implementation with Effect.provide(AWS.Kinesis.ListStreamConsumersHttp).

ListStreamConsumers: Enhanced Fan-Out

// init
const listStreamConsumers = yield* AWS.Kinesis.ListStreamConsumers(stream);

// runtime
const result = yield* listStreamConsumers();
const names = (result.Consumers ?? []).map((c) => c.ConsumerName);

ListStreams

Source: src/AWS/Kinesis/ListStreams.ts

Runtime binding for kinesis:ListStreams.

An account-level operation (no stream argument) that enumerates all Kinesis streams in the region. Provide the implementation with Effect.provide(AWS.Kinesis.ListStreamsHttp).

ListStreams: Inspecting Streams

// init — account-level binding takes no resource
const listStreams = yield* AWS.Kinesis.ListStreams();

// runtime
const result = yield* listStreams();
yield* Effect.log(result.StreamNames);

ListTagsForResource

Source: src/AWS/Kinesis/ListTagsForResource.ts

Runtime binding for kinesis:ListTagsForResource.

Bind this operation to a Stream or StreamConsumer to read its tags — the resource ARN is injected automatically. Provide the implementation with Effect.provide(AWS.Kinesis.ListTagsForResourceHttp).

ListTagsForResource: Inspecting Streams

// init — works for a Stream or a StreamConsumer
const listTagsForResource = yield* AWS.Kinesis.ListTagsForResource(stream);

// runtime
const result = yield* listTagsForResource();
const tags = Object.fromEntries(
  (result.Tags ?? []).map((tag) => [tag.Key, tag.Value]),
);

MergeShards

Source: src/AWS/Kinesis/MergeShards.ts

Runtime binding for kinesis:MergeShards.

Bind this operation to a Stream to merge two adjacent shards of a PROVISIONED-mode stream into one — the stream ARN is injected automatically. Useful for building custom shard-scaling logic (the Stream resource’s shardCount prop covers uniform scaling via UpdateShardCount; merge/split give per-shard control). Provide the implementation with Effect.provide(AWS.Kinesis.MergeShardsHttp).

MergeShards: Managing Shards

// init — bind the operation to the stream
const mergeShards = yield* AWS.Kinesis.MergeShards(stream);

// runtime — merge a shard with its adjacent neighbor
yield* mergeShards({
  ShardToMerge: "shardId-000000000000",
  AdjacentShardToMerge: "shardId-000000000001",
});

PutRecord

Source: src/AWS/Kinesis/PutRecord.ts

Runtime binding for kinesis:PutRecord.

Bind this operation to a Stream in the function’s init phase to get a callable that writes single records — the stream name is injected automatically and kinesis:PutRecord is granted on the stream. Provide the implementation with Effect.provide(AWS.Kinesis.PutRecordHttp).

PutRecord: Writing Records

export default MyFunction.make(
  { main: import.meta.url, functionUrl: true },
  Effect.gen(function* () {
    const stream = yield* AWS.Kinesis.Stream("OrdersStream");
    // init — bind the operation to the stream
    const putRecord = yield* AWS.Kinesis.PutRecord(stream);

    return {
      fetch: Effect.gen(function* () {
        // runtime — write a record
        yield* putRecord({
          PartitionKey: "order-123",
          Data: new TextEncoder().encode(JSON.stringify({ orderId: "123" })),
        });
        return HttpServerResponse.text("sent");
      }).pipe(Effect.orDie),
    };
  }).pipe(Effect.provide(AWS.Kinesis.PutRecordHttp)),
);

PutRecords

Source: src/AWS/Kinesis/PutRecords.ts

Runtime binding for kinesis:PutRecords.

Bind this operation to a Stream to write up to 500 records per call — the stream name is injected automatically and kinesis:PutRecords is granted on the stream. Provide the implementation with Effect.provide(AWS.Kinesis.PutRecordsHttp). For unbounded batching with automatic partial-failure retry, use AWS.Kinesis.StreamSink instead.

PutRecords: Writing Batches

// init — bind the operation to the stream
const putRecords = yield* AWS.Kinesis.PutRecords(stream);

// runtime — write a batch from a handler
const result = yield* putRecords({
  Records: orders.map((order) => ({
    PartitionKey: order.id,
    Data: new TextEncoder().encode(JSON.stringify(order)),
  })),
});
// result.FailedRecordCount > 0 means some entries need re-submission

SplitShard

Source: src/AWS/Kinesis/SplitShard.ts

Runtime binding for kinesis:SplitShard.

Bind this operation to a Stream to split one shard of a PROVISIONED-mode stream into two — the stream ARN is injected automatically. Useful for building custom shard-scaling logic that targets a hot shard directly (the Stream resource’s shardCount prop covers uniform scaling via UpdateShardCount). Provide the implementation with Effect.provide(AWS.Kinesis.SplitShardHttp).

SplitShard: Managing Shards

// init — bind the operation to the stream
const splitShard = yield* AWS.Kinesis.SplitShard(stream);

// runtime — split at the midpoint of the shard's hash-key range
yield* splitShard({
  ShardToSplit: "shardId-000000000000",
  NewStartingHashKey: "170141183460469231731687303715884105728",
});

Stream

Source: src/AWS/Kinesis/Stream.ts

An Amazon Kinesis Data Stream.

Stream owns the stream’s lifecycle and mutable control-plane configuration, including retention, encryption, monitoring, warm throughput, record size, tags, and stream resource policy. A stream name is auto-generated from the app, stage, and logical ID unless you provide one explicitly.

Stream: Creating Streams

On-Demand Stream

import * as Kinesis from "alchemy/AWS/Kinesis";

const stream = yield* Kinesis.Stream("OrdersStream");

Provisioned Stream

const stream = yield* Kinesis.Stream("AnalyticsStream", {
  streamMode: "PROVISIONED",
  shardCount: 2,
  retentionPeriod: "48 hours",
});

Encrypted Stream

const stream = yield* Kinesis.Stream("SecureStream", {
  encryption: true,
  kmsKeyId: "alias/my-key",
});

Stream: Runtime Producers

Bind producer operations in the init phase and use them in runtime handlers.

// init
const putRecord = yield* AWS.Kinesis.PutRecord(stream);

return {
  fetch: Effect.gen(function* () {
    // runtime
    yield* putRecord({
      PartitionKey: "order-123",
      Data: new TextEncoder().encode(JSON.stringify({ orderId: "123" })),
    });
    return HttpServerResponse.text("Sent");
  }),
};

Stream: Event Sources

Process records from a Kinesis stream using a Lambda event source mapping.

// init
yield* Kinesis.consumeStreamRecords(
  stream,
  {},
  Effect.fn(function* (record) {
    const data = new TextDecoder().decode(record.data);
    yield* Effect.log(`Received: ${data}`);
  }),
);

StreamConsumer

Source: src/AWS/Kinesis/StreamConsumer.ts

A registered Kinesis enhanced fan-out consumer.

StreamConsumer is the canonical lifecycle resource for RegisterStreamConsumer / DeregisterStreamConsumer.

StreamConsumer: Creating Consumers

const consumer = yield* StreamConsumer("AnalyticsConsumer", {
  streamArn: stream.streamArn,
});

StreamEventSource

Source: src/AWS/Kinesis/StreamEventSource.ts

Event source connecting a Kinesis Stream to the hosting compute.

The contract is a Binding.Service; the Lambda implementation layer (AWS.Lambda.StreamEventSource) creates an event source mapping on the stream, grants the read IAM actions, and forwards aws:kinesis records into the handler’s Stream. Use the consumeStreamRecords helper rather than calling the service directly.

StreamEventSource: Consuming Records

export default MyFunction.make(
  { main: import.meta.url },
  Effect.gen(function* () {
    const stream = yield* AWS.Kinesis.Stream("OrdersStream");

    // init — registers the event source mapping and the record handler
    yield* AWS.Kinesis.consumeStreamRecords(
      stream,
      { startingPosition: "LATEST", batchSize: 10 },
      (records) =>
        records.pipe(
          Stream.runForEach((record) =>
            Effect.log(
              Buffer.from(record.kinesis.data, "base64").toString("utf8"),
            ),
          ),
        ),
    );

    return {};
  }).pipe(Effect.provide(AWS.Lambda.StreamEventSource)),
);

StreamSink

Source: src/AWS/Kinesis/StreamSink.ts

A partition-aware sink for batching PutRecords requests into a stream (500 records / 5 MiB per call).

Each input element is a raw PutRecordsRequestEntry, so callers stay in control of PartitionKey and optional ExplicitHashKey.

Records the API reports as failed (FailedRecordCount > 0, per-record ErrorCode — throughput exceeded or internal failure) are re-submitted on a bounded schedule; exhausting retries fails the sink with a typed BatchRetryExhaustedError carrying the stranded records.

Provide the implementation with Effect.provide(AWS.Kinesis.StreamSinkHttp).

StreamSink: Writing Streams of Records

// init — bind the sink to the stream
const sink = yield* AWS.Kinesis.StreamSink(stream);

// runtime — batches into PutRecords calls of up to 500 records / 5 MiB
yield* Stream.fromIterable(
  orders.map((order) => ({
    PartitionKey: order.id,
    Data: new TextEncoder().encode(JSON.stringify(order)),
  })),
).pipe(Stream.run(sink));

SubscribeToShard

Source: src/AWS/Kinesis/SubscribeToShard.ts

Runtime binding for kinesis:SubscribeToShard (enhanced fan-out).

Bind this operation to a StreamConsumer (a registered enhanced fan-out consumer) to open a push-based subscription to a shard — the consumer ARN is injected automatically. Provide the implementation with Effect.provide(AWS.Kinesis.SubscribeToShardHttp).

SubscribeToShard: Enhanced Fan-Out

const consumer = yield* AWS.Kinesis.StreamConsumer("Analytics", {
  streamArn: stream.streamArn,
});
// init — bind the operation to the registered consumer
const subscribeToShard = yield* AWS.Kinesis.SubscribeToShard(consumer);

// runtime — open the subscription
const result = yield* subscribeToShard({
  ShardId: shardId,
  StartingPosition: { Type: "LATEST" },
});
// result.EventStream delivers records for up to 5 minutes