1// required to avoid this error in consumers: "The inferred type of 'Records' cannot be named without a reference to '../../@distilled.cloud/aws/node_modules/@types/aws-lambda'. This is likely not portable. A type annotation is necessary.ts(2742)"
2export type * as lambda from "aws-lambda";
3
4import * as kinesis from "@distilled.cloud/aws/kinesis";
5import { Unowned } from "alchemy/AdoptPolicy";
6import { isResolved } from "alchemy/Diff";
7import { createPhysicalName } from "alchemy/PhysicalName";
8import * as Provider from "alchemy/Provider";
9import { Resource } from "alchemy/Resource";
10import { createInternalTags, diffTags, hasAlchemyTags, type Tags } from "alchemy/Tags";
11import { toWireHours } from "alchemy/Util/Duration";
12import type * as lambda from "aws-lambda";
13import * as Data from "effect/Data";
14import type * as Duration from "effect/Duration";
15import * as Effect from "effect/Effect";
16import * as Schedule from "effect/Schedule";
17// `Stream` is the resource class name in this file, so alias the Effect
18// `Stream` module to avoid the collision.
19import * as EffectStream from "effect/Stream";
20import { AWSEnvironment, type AccountID } from "../Environment.ts";
21import type { Providers } from "../Providers.ts";
22import type { RegionID } from "../Region.ts";
23
24export class KinesisStreamInvalidProps extends Data.TaggedError("KinesisStreamInvalidProps")<{
25 readonly message: string;
26 readonly cause?: unknown;
27}> {}
28
29export class KinesisStreamUnexpectedResponse extends Data.TaggedError(
30 "KinesisStreamUnexpectedResponse",
31)<{
32 readonly message: string;
33 readonly cause?: unknown;
34}> {}
35
36export type StreamRecord = lambda.KinesisStreamRecord;
37
38export type StreamEvent = lambda.KinesisStreamEvent;
39
40export type StreamName = string;
41export type StreamArn = `arn:aws:kinesis:${RegionID}:${AccountID}:stream/${StreamName}`;
42
43export type StreamStatus = "CREATING" | "DELETING" | "ACTIVE" | "UPDATING";
44
45export type StreamMode = "PROVISIONED" | "ON_DEMAND";
46
47export type EncryptionType = "NONE" | "KMS";
48
49export type WarmThroughput = {
50 /**
51 * Requested warm throughput in MiBps.
52 */
53 targetMiBps?: number;
54 /**
55 * Warm throughput currently provisioned by AWS, in MiBps.
56 */
57 currentMiBps?: number;
58};
59
60export type StreamProps = {
61 /**
62 * Name of the stream.
63 * @default ${app}-${stage}-${id}
64 */
65 streamName?: string;
66 /**
67 * The capacity mode of the data stream.
68 * - PROVISIONED: You specify the number of shards for the data stream.
69 * - ON_DEMAND: AWS manages the shards for the data stream.
70 * @default "ON_DEMAND"
71 */
72 streamMode?: StreamMode;
73 /**
74 * The number of shards that the stream will use when in PROVISIONED mode.
75 * Required when `streamMode` is `"PROVISIONED"`.
76 */
77 shardCount?: number;
78 /**
79 * How long records remain accessible in the stream, e.g. `"48 hours"` or
80 * `Duration.hours(48)`. The API stores whole hours; valid values range
81 * from 24 hours to 8760 hours (365 days).
82 * @default "24 hours"
83 */
84 retentionPeriod?: Duration.Input;
85 /**
86 * If set to true, server-side encryption is enabled on the stream.
87 * Uses the AWS managed CMK for Kinesis (`alias/aws/kinesis`) when `kmsKeyId`
88 * is omitted.
89 * @default false
90 */
91 encryption?: boolean;
92 /**
93 * The AWS KMS key to use when encryption is enabled.
94 */
95 kmsKeyId?: string;
96 /**
97 * A list of shard-level CloudWatch metrics to enable for the stream.
98 */
99 shardLevelMetrics?: ShardLevelMetric[];
100 /**
101 * Pre-provisioned warm throughput for on-demand streams, in MiBps.
102 */
103 warmThroughputMiBps?: number;
104 /**
105 * Maximum size of a single record, in KiB.
106 */
107 maxRecordSizeInKiB?: number;
108 /**
109 * Resource policy attached to the stream.
110 */
111 resourcePolicy?: string;
112 /**
113 * Tags to associate with the stream.
114 */
115 tags?: Record<string, string>;
116};
117
118export interface Stream extends Resource<
119 "AWS.Kinesis.Stream",
120 StreamProps,
121 {
122 /**
123 * The stream's physical name.
124 */
125 streamName: StreamName;
126 /**
127 * ARN of the stream.
128 */
129 streamArn: StreamArn;
130 /**
131 * Provider-assigned unique identifier for the stream, when returned by AWS.
132 */
133 streamId: string | undefined;
134 /**
135 * Current lifecycle status of the stream.
136 */
137 streamStatus: StreamStatus;
138 /**
139 * Current capacity mode of the stream.
140 */
141 streamMode: StreamMode;
142 /**
143 * Number of hours that records remain accessible in the stream.
144 */
145 retentionPeriodHours: number;
146 /**
147 * Current server-side encryption mode.
148 */
149 encryptionType: EncryptionType;
150 /**
151 * KMS key ID backing stream encryption, when encryption is enabled with KMS.
152 */
153 kmsKeyId: string | undefined;
154 /**
155 * Number of open shards currently reported by the stream summary.
156 */
157 openShardCount: number | undefined;
158 /**
159 * Number of registered consumers currently attached to the stream.
160 */
161 consumerCount: number | undefined;
162 /**
163 * Enabled shard-level CloudWatch metrics for the stream.
164 */
165 shardLevelMetrics: ShardLevelMetric[];
166 /**
167 * Current and target warm throughput settings for on-demand streams, when available.
168 */
169 warmThroughput: WarmThroughput | undefined;
170 /**
171 * Maximum record size, in KiB, that the stream accepts.
172 */
173 maxRecordSizeInKiB: number | undefined;
174 /**
175 * Current resource policy attached to the stream, if one is configured.
176 */
177 resourcePolicy: string | undefined;
178 /**
179 * Current tags reported for the stream.
180 */
181 tags: Record<string, string>;
182 },
183 never,
184 Providers
185> {}
186
187/**
188 * An Amazon Kinesis Data Stream.
189 *
190 * `Stream` owns the stream's lifecycle and mutable control-plane configuration,
191 * including retention, encryption, monitoring, warm throughput, record size, tags,
192 * and stream resource policy. A stream name is auto-generated from the app,
193 * stage, and logical ID unless you provide one explicitly.
194 * ### Creating Streams
195 * **Example:** On-Demand Stream
196 * ```typescript
197 * import * as Kinesis from "@alchemy.run/aws/Kinesis";
198 *
199 * const stream = yield* Kinesis.Stream("OrdersStream");
200 * ```
201 *
202 * **Example:** Provisioned Stream
203 * ```typescript
204 * const stream = yield* Kinesis.Stream("AnalyticsStream", {
205 * streamMode: "PROVISIONED",
206 * shardCount: 2,
207 * retentionPeriod: "48 hours",
208 * });
209 * ```
210 *
211 * **Example:** Encrypted Stream
212 * ```typescript
213 * const stream = yield* Kinesis.Stream("SecureStream", {
214 * encryption: true,
215 * kmsKeyId: "alias/my-key",
216 * });
217 * ```
218 *
219 * ### Runtime Producers
220 * Bind producer operations in the init phase and use them in runtime
221 * handlers.
222 *
223 * **Example:** Put a record from a handler
224 * ```typescript
225 * // init
226 * const putRecord = yield* AWS.Kinesis.PutRecord(stream);
227 *
228 * return {
229 * fetch: Effect.gen(function* () {
230 * // runtime
231 * yield* putRecord({
232 * PartitionKey: "order-123",
233 * Data: new TextEncoder().encode(JSON.stringify({ orderId: "123" })),
234 * });
235 * return HttpServerResponse.text("Sent");
236 * }),
237 * };
238 * ```
239 *
240 * ### Event Sources
241 * Process records from a Kinesis stream using a Lambda event source
242 * mapping.
243 *
244 * **Example:** Process stream records
245 * ```typescript
246 * // init
247 * yield* Kinesis.consumeStreamRecords(
248 * stream,
249 * {},
250 * Effect.fn(function* (event) {
251 * for (const record of event.Records) {
252 * const data = Buffer.from(record.kinesis.data, "base64").toString("utf8");
253 * yield* Effect.log(`Received: ${data}`);
254 * }
255 * }),
256 * );
257 * ```
258 *
259 * @resource
260 */
261export const Stream = Resource<Stream>("AWS.Kinesis.Stream");
262
263export type ShardLevelMetric =
264 | "IncomingBytes"
265 | "IncomingRecords"
266 | "OutgoingBytes"
267 | "OutgoingRecords"
268 | "WriteProvisionedThroughputExceeded"
269 | "ReadProvisionedThroughputExceeded"
270 | "IteratorAgeMilliseconds"
271 | "ALL";
272
273const defaultStreamMode = "ON_DEMAND" as const;
274const defaultRetentionPeriodHours = 24;
275const defaultEncryptionType = "NONE" as const;
276
277const createStreamName = (
278 id: string,
279 props: {
280 streamName?: string | undefined;
281 },
282) =>
283 Effect.gen(function* () {
284 if (props.streamName) {
285 return props.streamName;
286 }
287 return yield* createPhysicalName({
288 id,
289 maxLength: 128,
290 });
291 });
292
293const getStreamMode = (props: StreamProps): kinesis.StreamModeDetails => ({
294 StreamMode: props.streamMode ?? defaultStreamMode,
295});
296
297const assertProvisionedProps = (props: StreamProps) =>
298 props.streamMode === "PROVISIONED" && props.shardCount === undefined
299 ? Effect.fail(
300 new KinesisStreamInvalidProps({
301 message: `streamMode "PROVISIONED" requires shardCount to be set`,
302 }),
303 )
304 : Effect.void;
305
306const toTagRecord = (tags: Array<{ Key: string; Value?: string }> | undefined) =>
307 Object.fromEntries(
308 (tags ?? [])
309 .filter((tag): tag is { Key: string; Value: string } => typeof tag.Value === "string")
310 .map((tag) => [tag.Key, tag.Value]),
311 );
312
313const toWarmThroughput = (
314 warmThroughput: kinesis.WarmThroughputObject | undefined,
315): WarmThroughput | undefined =>
316 warmThroughput
317 ? {
318 targetMiBps: warmThroughput.TargetMiBps,
319 currentMiBps: warmThroughput.CurrentMiBps,
320 }
321 : undefined;
322
323const toShardLevelMetrics = (
324 monitoring: kinesis.EnhancedMetrics[] | undefined,
325): ShardLevelMetric[] =>
326 [
327 ...new Set((monitoring ?? []).flatMap((metric) => metric.ShardLevelMetrics ?? [])),
328 ] as ShardLevelMetric[];
329
330const toAttrs = ({
331 summary,
332 tags,
333 resourcePolicy,
334}: {
335 summary: kinesis.StreamDescriptionSummary;
336 tags: Record<string, string>;
337 resourcePolicy?: string;
338}): Stream["Attributes"] => ({
339 streamName: summary.StreamName,
340 streamArn: summary.StreamARN as StreamArn,
341 streamId: summary.StreamId,
342 streamStatus: summary.StreamStatus as StreamStatus,
343 streamMode: (summary.StreamModeDetails?.StreamMode ?? defaultStreamMode) as StreamMode,
344 retentionPeriodHours: summary.RetentionPeriodHours ?? defaultRetentionPeriodHours,
345 encryptionType: (summary.EncryptionType ?? defaultEncryptionType) as EncryptionType,
346 kmsKeyId: summary.KeyId,
347 openShardCount: summary.OpenShardCount,
348 consumerCount: summary.ConsumerCount,
349 shardLevelMetrics: toShardLevelMetrics(summary.EnhancedMonitoring),
350 warmThroughput: toWarmThroughput(summary.WarmThroughput),
351 maxRecordSizeInKiB: summary.MaxRecordSizeInKiB,
352 resourcePolicy,
353 tags,
354});
355
356const readStream = Effect.fn(function* ({
357 streamName,
358 streamArn,
359}: {
360 streamName?: string;
361 streamArn?: string;
362}) {
363 const response = yield* kinesis
364 .describeStreamSummary({
365 StreamName: streamName,
366 StreamARN: streamArn,
367 })
368 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.undefined));
369
370 if (!response) {
371 return undefined;
372 }
373
374 const summary = response.StreamDescriptionSummary;
375 // The stream can vanish between the describe above and these follow-up
376 // calls (e.g. another test tears its stream down while `list` hydrates) —
377 // a `ResourceNotFoundException` here just means it's gone, so report it as
378 // missing rather than failing the whole enumeration.
379 const tagsResponse = yield* kinesis
380 .listTagsForResource({
381 ResourceARN: summary.StreamARN,
382 })
383 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.undefined));
384 if (!tagsResponse) {
385 return undefined;
386 }
387 const policyResponse = yield* kinesis
388 .getResourcePolicy({
389 ResourceARN: summary.StreamARN,
390 })
391 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.undefined));
392
393 return toAttrs({
394 summary,
395 tags: toTagRecord(tagsResponse.Tags),
396 // Treat an empty policy string as "no policy attached" so the
397 // sync step doesn't try to delete a phantom policy.
398 resourcePolicy: policyResponse?.Policy ? policyResponse.Policy : undefined,
399 });
400});
401
402const waitForStreamActive = (streamName: string) =>
403 Effect.gen(function* () {
404 yield* Effect.sleep("2 seconds");
405 const { StreamDescriptionSummary } = yield* kinesis.describeStreamSummary({
406 StreamName: streamName,
407 });
408 if (StreamDescriptionSummary.StreamStatus !== "ACTIVE") {
409 return yield* Effect.fail({ _tag: "StreamNotActive" as const });
410 }
411 return StreamDescriptionSummary;
412 }).pipe(
413 Effect.retry({
414 while: (e: { _tag: string }) => e._tag === "StreamNotActive" || e._tag === "ParseError",
415 schedule: Schedule.max([Schedule.exponential(500), Schedule.recurs(60)]),
416 }),
417 );
418
419const waitForStreamDeleted = (streamName: string) =>
420 Effect.gen(function* () {
421 yield* kinesis.describeStreamSummary({
422 StreamName: streamName,
423 });
424 return yield* Effect.fail({ _tag: "StreamStillExists" as const });
425 }).pipe(
426 Effect.retry({
427 while: (e: { _tag: string }) => e._tag === "StreamStillExists" || e._tag === "ParseError",
428 schedule: Schedule.max([Schedule.exponential(500), Schedule.recurs(60)]),
429 }),
430 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
431 );
432
433export const StreamProvider = () =>
434 Provider.effect(
435 Stream,
436 Effect.gen(function* () {
437 return {
438 stables: ["streamName", "streamArn"],
439 // Enumerate every stream in the ambient account/region. `listStreams`
440 // is paginated; collect every page exhaustively, then hydrate each
441 // stream into the exact `read` Attributes shape (summary + tags +
442 // resource policy) via `readStream` with bounded concurrency. A
443 // stream that disappears between listing and hydration is handled
444 // inside `readStream` (typed `ResourceNotFoundException` -> undefined)
445 // and filtered out.
446 list: () =>
447 Effect.gen(function* () {
448 const streamNames = yield* kinesis.listStreams.pages({}).pipe(
449 EffectStream.runCollect,
450 Effect.map((chunk) => Array.from(chunk).flatMap((page) => page.StreamNames ?? [])),
451 );
452
453 const hydrated = yield* Effect.forEach(
454 streamNames,
455 (streamName) => readStream({ streamName }),
456 { concurrency: 10 },
457 );
458
459 return hydrated.filter((attrs): attrs is Stream["Attributes"] => attrs !== undefined);
460 }),
461 read: Effect.fn(function* ({ id, olds, output }) {
462 const streamName = output?.streamName ?? (yield* createStreamName(id, olds ?? {}));
463 const state = yield* readStream({
464 streamName,
465 streamArn: output?.streamArn,
466 });
467 if (!state) return undefined;
468 return (yield* hasAlchemyTags(id, state.tags as Tags)) ? state : Unowned(state);
469 }),
470 diff: Effect.fn(function* ({ id, news = {}, olds = {} }) {
471 if (!isResolved(news)) return;
472 const oldStreamName = yield* createStreamName(id, olds);
473 const newStreamName = yield* createStreamName(id, news);
474 if (oldStreamName !== newStreamName) {
475 return { action: "replace" } as const;
476 }
477 }),
478 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
479 const { accountId, region } = yield* AWSEnvironment.current;
480 yield* assertProvisionedProps(news);
481
482 const streamName = output?.streamName ?? (yield* createStreamName(id, news));
483 const streamArn = `arn:aws:kinesis:${region}:${accountId}:stream/${streamName}` as const;
484 const internalTags = yield* createInternalTags(id);
485 const desiredTags = { ...internalTags, ...news.tags };
486
487 // Observe — fetch live cloud state. `output` is treated as a cache
488 // for the physical name only; the stream's actual existence and
489 // configuration are fetched fresh so the reconciler converges
490 // regardless of drift, adoption, or a partially-completed prior
491 // run.
492 let state = yield* readStream({ streamName, streamArn });
493
494 // Ensure — create the stream if it's missing. Tolerate
495 // `ResourceInUseException` as a race with a peer reconciler:
496 // re-read and continue with the sync path. Retry transient
497 // `LimitExceededException`.
498 if (state === undefined) {
499 yield* kinesis
500 .createStream({
501 StreamName: streamName,
502 ShardCount: news.streamMode === "PROVISIONED" ? news.shardCount : undefined,
503 StreamModeDetails: getStreamMode(news),
504 Tags: desiredTags,
505 WarmThroughputMiBps: news.warmThroughputMiBps,
506 MaxRecordSizeInKiB: news.maxRecordSizeInKiB,
507 })
508 .pipe(
509 Effect.catchTag("ResourceInUseException", () => Effect.void),
510 Effect.retry({
511 while: (e: any) => e._tag === "LimitExceededException",
512 schedule: Schedule.exponential(1000),
513 }),
514 );
515
516 yield* session.note(`Creating stream ${streamName}...`);
517 yield* waitForStreamActive(streamName);
518
519 state = yield* readStream({ streamName, streamArn });
520 if (state === undefined) {
521 return yield* new KinesisStreamUnexpectedResponse({
522 message: `failed to read created stream ${streamName}`,
523 });
524 }
525 }
526
527 // Sync stream mode — observed ↔ desired.
528 const desiredMode = news.streamMode ?? defaultStreamMode;
529 if (state.streamMode !== desiredMode) {
530 yield* kinesis.updateStreamMode({
531 StreamARN: streamArn,
532 StreamModeDetails: getStreamMode(news),
533 WarmThroughputMiBps:
534 desiredMode === "ON_DEMAND" ? news.warmThroughputMiBps : undefined,
535 });
536 yield* waitForStreamActive(streamName);
537 yield* session.note(`Updated stream mode to ${desiredMode}`);
538 state = yield* readStream({ streamName, streamArn });
539 if (state === undefined) {
540 return yield* new KinesisStreamUnexpectedResponse({
541 message: `failed to re-read stream ${streamName}`,
542 });
543 }
544 }
545
546 // Sync shard count — only meaningful in PROVISIONED mode.
547 if (
548 desiredMode === "PROVISIONED" &&
549 news.shardCount !== undefined &&
550 state.openShardCount !== news.shardCount
551 ) {
552 yield* kinesis.updateShardCount({
553 StreamName: streamName,
554 TargetShardCount: news.shardCount,
555 ScalingType: "UNIFORM_SCALING",
556 });
557 yield* waitForStreamActive(streamName);
558 yield* session.note(`Updated shard count to ${news.shardCount}`);
559 }
560
561 // Sync retention period — observed ↔ desired.
562 const desiredRetention = toWireHours(news.retentionPeriod) ?? defaultRetentionPeriodHours;
563 if (state.retentionPeriodHours !== desiredRetention) {
564 if (desiredRetention > state.retentionPeriodHours) {
565 yield* kinesis.increaseStreamRetentionPeriod({
566 StreamName: streamName,
567 RetentionPeriodHours: desiredRetention,
568 });
569 } else {
570 yield* kinesis.decreaseStreamRetentionPeriod({
571 StreamName: streamName,
572 RetentionPeriodHours: desiredRetention,
573 });
574 }
575 yield* waitForStreamActive(streamName);
576 yield* session.note(`Updated retention period to ${desiredRetention} hours`);
577 }
578
579 // Sync encryption — observed ↔ desired.
580 const desiredEncryption = news.encryption ?? false;
581 const desiredKmsKey = news.kmsKeyId ?? "alias/aws/kinesis";
582 const observedEncryption = state.encryptionType === "KMS";
583 if (!observedEncryption && desiredEncryption) {
584 yield* kinesis.startStreamEncryption({
585 StreamName: streamName,
586 EncryptionType: "KMS",
587 KeyId: desiredKmsKey,
588 });
589 yield* waitForStreamActive(streamName);
590 yield* session.note("Enabled encryption");
591 } else if (observedEncryption && !desiredEncryption) {
592 yield* kinesis.stopStreamEncryption({
593 StreamName: streamName,
594 EncryptionType: "KMS",
595 KeyId: state.kmsKeyId ?? desiredKmsKey,
596 });
597 yield* waitForStreamActive(streamName);
598 yield* session.note("Disabled encryption");
599 } else if (
600 observedEncryption &&
601 desiredEncryption &&
602 news.kmsKeyId !== undefined &&
603 state.kmsKeyId !== news.kmsKeyId
604 ) {
605 yield* kinesis.startStreamEncryption({
606 StreamName: streamName,
607 EncryptionType: "KMS",
608 KeyId: news.kmsKeyId,
609 });
610 yield* waitForStreamActive(streamName);
611 yield* session.note("Updated KMS key");
612 }
613
614 // Sync shard-level metrics — observed ↔ desired.
615 const observedMetrics = new Set(state.shardLevelMetrics);
616 const desiredMetrics = new Set(news.shardLevelMetrics ?? []);
617 const metricsToEnable = (news.shardLevelMetrics ?? []).filter(
618 (metric) => !observedMetrics.has(metric),
619 );
620 const metricsToDisable = state.shardLevelMetrics.filter(
621 (metric) => !desiredMetrics.has(metric),
622 );
623
624 if (metricsToDisable.length > 0) {
625 yield* kinesis.disableEnhancedMonitoring({
626 StreamName: streamName,
627 ShardLevelMetrics: metricsToDisable,
628 });
629 yield* waitForStreamActive(streamName);
630 yield* session.note(`Disabled metrics: ${metricsToDisable.join(", ")}`);
631 }
632
633 if (metricsToEnable.length > 0) {
634 yield* kinesis.enableEnhancedMonitoring({
635 StreamName: streamName,
636 ShardLevelMetrics: metricsToEnable,
637 });
638 yield* waitForStreamActive(streamName);
639 yield* session.note(`Enabled metrics: ${metricsToEnable.join(", ")}`);
640 }
641
642 // Sync warm throughput — only meaningful in ON_DEMAND mode.
643 if (
644 desiredMode === "ON_DEMAND" &&
645 news.warmThroughputMiBps !== undefined &&
646 state.warmThroughput?.targetMiBps !== news.warmThroughputMiBps
647 ) {
648 yield* kinesis.updateStreamWarmThroughput({
649 StreamARN: streamArn,
650 WarmThroughputMiBps: news.warmThroughputMiBps,
651 });
652 yield* waitForStreamActive(streamName);
653 yield* session.note(`Updated warm throughput to ${news.warmThroughputMiBps} MiBps`);
654 }
655
656 // Sync max record size — observed ↔ desired.
657 if (
658 news.maxRecordSizeInKiB !== undefined &&
659 state.maxRecordSizeInKiB !== news.maxRecordSizeInKiB
660 ) {
661 yield* kinesis.updateMaxRecordSize({
662 StreamARN: streamArn,
663 MaxRecordSizeInKiB: news.maxRecordSizeInKiB,
664 });
665 yield* waitForStreamActive(streamName);
666 yield* session.note(`Updated max record size to ${news.maxRecordSizeInKiB} KiB`);
667 }
668
669 // Sync tags — diff observed cloud tags against desired. Adoption
670 // may bring us a stream that already has its own tag set; diffing
671 // against `state.tags` (fetched fresh) lets the reconciler
672 // converge ownership without fighting whatever was there before.
673 const { removed, upsert } = diffTags(state.tags, desiredTags);
674
675 if (removed.length > 0) {
676 yield* kinesis.removeTagsFromStream({
677 StreamName: streamName,
678 TagKeys: removed,
679 });
680 }
681
682 if (upsert.length > 0) {
683 const tagsToAdd: Record<string, string> = {};
684 for (const { Key, Value } of upsert) {
685 tagsToAdd[Key] = Value;
686 }
687 yield* kinesis.addTagsToStream({
688 StreamName: streamName,
689 Tags: tagsToAdd,
690 });
691 }
692
693 // Sync resource policy — observed ↔ desired. Tolerate the
694 // race where the cloud's policy was already removed (delete
695 // fails with `ResourceNotFoundException`) or where another
696 // reconcile already wrote the same policy.
697 if (state.resourcePolicy !== news.resourcePolicy) {
698 if (news.resourcePolicy) {
699 yield* kinesis.putResourcePolicy({
700 ResourceARN: streamArn,
701 Policy: news.resourcePolicy,
702 });
703 } else if (state.resourcePolicy) {
704 yield* kinesis
705 .deleteResourcePolicy({
706 ResourceARN: streamArn,
707 })
708 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
709 }
710 }
711
712 yield* session.note(streamArn);
713
714 // Re-read final state so the returned attributes reflect what's
715 // actually in the cloud after all sync steps.
716 const final = yield* readStream({ streamName, streamArn });
717 if (!final) {
718 return yield* new KinesisStreamUnexpectedResponse({
719 message: `failed to read reconciled stream ${streamName}`,
720 });
721 }
722 return final;
723 }),
724 delete: Effect.fn(function* ({ output }) {
725 yield* kinesis
726 .deleteStream({
727 StreamName: output.streamName,
728 EnforceConsumerDeletion: true,
729 })
730 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
731
732 yield* waitForStreamDeleted(output.streamName);
733 }),
734 };
735 }),
736 );