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