1import * as firehose from "@distilled.cloud/aws/firehose";
2import * as iam from "@distilled.cloud/aws/iam";
3import * as Data from "effect/Data";
4import type * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import {
13 createInternalTags,
14 createTagsList,
15 diffTags,
16 hasAlchemyTags,
17 type Tags,
18} from "../../Tags.ts";
19import { AWSEnvironment, type AccountID } from "../Environment.ts";
20import type { PolicyStatement } from "../IAM/Policy.ts";
21import { toWireSeconds } from "../../Util/Duration.ts";
22import type { Providers } from "../Providers.ts";
23import type { RegionID } from "../Region.ts";
24
25export type DeliveryStreamName = string;
26export type DeliveryStreamArn =
27 `arn:aws:firehose:${RegionID}:${AccountID}:deliverystream/${DeliveryStreamName}`;
28
29export type DeliveryStreamStatus =
30 | "CREATING"
31 | "CREATING_FAILED"
32 | "DELETING"
33 | "DELETING_FAILED"
34 | "ACTIVE";
35
36export type DeliveryStreamSourceType = "DirectPut" | "KinesisStreamAsSource";
37
38export type CompressionFormat =
39 | "UNCOMPRESSED"
40 | "GZIP"
41 | "ZIP"
42 | "Snappy"
43 | "HADOOP_SNAPPY";
44
45export interface KinesisStreamSourceProps {
46 /**
47 * ARN of the Kinesis Data Stream that feeds the delivery stream.
48 * Changing the source stream replaces the delivery stream.
49 */
50 kinesisStreamArn: string;
51 /**
52 * ARN of the IAM role Firehose assumes to read from the Kinesis stream.
53 * @default a role is auto-created granting kinesis:DescribeStream,
54 * kinesis:GetShardIterator, kinesis:GetRecords and kinesis:ListShards on
55 * the source stream.
56 */
57 roleArn?: string;
58}
59
60export interface S3DestinationProps {
61 /**
62 * ARN of the destination S3 bucket.
63 */
64 bucketArn: string;
65 /**
66 * ARN of the IAM role Firehose assumes to write to the bucket.
67 * @default a role is auto-created granting s3:PutObject, s3:GetObject,
68 * s3:ListBucket, s3:GetBucketLocation, s3:AbortMultipartUpload and
69 * s3:ListBucketMultipartUploads on the bucket.
70 */
71 roleArn?: string;
72 /**
73 * Prefix prepended to delivered S3 object keys.
74 */
75 prefix?: string;
76 /**
77 * Prefix for objects that failed delivery or transformation.
78 */
79 errorOutputPrefix?: string;
80 /**
81 * Buffering interval before flushing to S3 — e.g. `"1 minute"` or
82 * `Duration.seconds(60)`. Sent to AWS as whole seconds; valid values range
83 * from 0 (zero buffering) to 900 seconds.
84 * @default "300 seconds"
85 */
86 bufferingInterval?: Duration.Input;
87 /**
88 * Buffering size in MiB before flushing to S3.
89 * Valid values range from 1 to 128.
90 * @default 5
91 */
92 bufferingSizeInMBs?: number;
93 /**
94 * Compression format applied to delivered objects.
95 * @default "UNCOMPRESSED"
96 */
97 compressionFormat?: CompressionFormat;
98}
99
100export type DeliveryStreamEncryptionKeyType =
101 | "AWS_OWNED_CMK"
102 | "CUSTOMER_MANAGED_CMK";
103
104export type DeliveryStreamEncryptionStatus =
105 | "ENABLED"
106 | "ENABLING"
107 | "ENABLING_FAILED"
108 | "DISABLED"
109 | "DISABLING"
110 | "DISABLING_FAILED";
111
112export interface DeliveryStreamEncryptionProps {
113 /**
114 * Which key to use for server-side encryption: the AWS-owned CMK that
115 * Firehose manages for you, or a customer-managed KMS key (`keyArn`
116 * required).
117 */
118 keyType: DeliveryStreamEncryptionKeyType;
119 /**
120 * ARN of the customer-managed KMS key. Required when `keyType` is
121 * `CUSTOMER_MANAGED_CMK`; must be omitted for `AWS_OWNED_CMK`.
122 */
123 keyArn?: string;
124}
125
126export interface DeliveryStreamProps {
127 /**
128 * Name of the delivery stream. Changing the name replaces the stream.
129 * @default ${app}-${id}-${stage}-${instanceId}
130 */
131 deliveryStreamName?: string;
132 /**
133 * Kinesis Data Stream source. When omitted the stream is `DirectPut` and
134 * producers write via `PutRecord` / `PutRecordBatch`. Adding, removing or
135 * changing the source replaces the delivery stream.
136 */
137 source?: KinesisStreamSourceProps;
138 /**
139 * S3 destination configuration (Firehose "extended S3" destination).
140 * Destination settings update in place via `UpdateDestination`.
141 */
142 destination: S3DestinationProps;
143 /**
144 * Server-side encryption (SSE) for records at rest inside the delivery
145 * stream. Only supported for `DirectPut` streams — streams with a Kinesis
146 * source inherit encryption from the source stream. Adding, changing or
147 * removing encryption updates the stream in place via
148 * `StartDeliveryStreamEncryption` / `StopDeliveryStreamEncryption`.
149 * @default no server-side encryption
150 */
151 encryption?: DeliveryStreamEncryptionProps;
152 /**
153 * Tags to associate with the delivery stream.
154 */
155 tags?: Record<string, string>;
156}
157
158export interface DeliveryStream extends Resource<
159 "AWS.Firehose.DeliveryStream",
160 DeliveryStreamProps,
161 {
162 /**
163 * The delivery stream's physical name.
164 */
165 deliveryStreamName: DeliveryStreamName;
166 /**
167 * ARN of the delivery stream.
168 */
169 deliveryStreamArn: DeliveryStreamArn;
170 /**
171 * Current lifecycle status of the delivery stream.
172 */
173 deliveryStreamStatus: DeliveryStreamStatus;
174 /**
175 * Source type of the delivery stream.
176 */
177 deliveryStreamType: DeliveryStreamSourceType;
178 /**
179 * Current version ID of the delivery stream configuration.
180 */
181 versionId: string;
182 /**
183 * ID of the (single) destination attached to the stream.
184 */
185 destinationId: string | undefined;
186 /**
187 * ARN of the destination S3 bucket.
188 */
189 bucketArn: string;
190 /**
191 * ARN of the IAM role Firehose assumes to write to the destination.
192 */
193 roleArn: string;
194 /**
195 * Name of the auto-created IAM role, when one was synthesized for this
196 * stream. `undefined` when the caller supplied every role ARN.
197 */
198 roleName: string | undefined;
199 /**
200 * ARN of the source Kinesis stream, for `KinesisStreamAsSource` streams.
201 */
202 kinesisStreamArn: string | undefined;
203 /**
204 * Prefix prepended to delivered S3 object keys.
205 */
206 prefix: string | undefined;
207 /**
208 * Prefix for objects that failed delivery.
209 */
210 errorOutputPrefix: string | undefined;
211 /**
212 * Buffering interval in seconds currently configured on the destination.
213 */
214 bufferingIntervalInSeconds: number | undefined;
215 /**
216 * Buffering size in MiB currently configured on the destination.
217 */
218 bufferingSizeInMBs: number | undefined;
219 /**
220 * Compression format currently configured on the destination.
221 */
222 compressionFormat: CompressionFormat | undefined;
223 /**
224 * Server-side encryption status reported by the stream — `ENABLED` when
225 * SSE is active, `DISABLED` when it is not.
226 */
227 encryptionStatus: DeliveryStreamEncryptionStatus | undefined;
228 /**
229 * Key type used for server-side encryption, when enabled.
230 */
231 encryptionKeyType: DeliveryStreamEncryptionKeyType | undefined;
232 /**
233 * ARN of the customer-managed KMS key used for server-side encryption,
234 * when `encryptionKeyType` is `CUSTOMER_MANAGED_CMK`.
235 */
236 encryptionKeyArn: string | undefined;
237 /**
238 * Current tags reported for the delivery stream.
239 */
240 tags: Record<string, string>;
241 },
242 never,
243 Providers
244> {}
245
246/**
247 * An Amazon Data Firehose delivery stream that buffers records and delivers
248 * them to an S3 bucket.
249 *
250 * `DeliveryStream` owns the stream's lifecycle and mutable destination
251 * configuration (buffering hints, compression, prefixes, tags). The stream is
252 * `DirectPut` by default — producers write with `PutRecord` /
253 * `PutRecordBatch` — or it can drain an existing Kinesis Data Stream via the
254 * `source` prop. Unless you supply role ARNs, an IAM role is auto-created
255 * granting Firehose write access to the destination bucket (and read access
256 * to the source stream when one is configured).
257 * ### Creating Delivery Streams
258 * **Example:** DirectPut stream delivering to S3
259 * ```typescript
260 * import * as AWS from "alchemy/AWS";
261 *
262 * const bucket = yield* AWS.S3.Bucket("DataLake");
263 * const stream = yield* AWS.Firehose.DeliveryStream("Events", {
264 * destination: {
265 * bucketArn: bucket.bucketArn,
266 * },
267 * });
268 * ```
269 *
270 * **Example:** Tuned buffering and compression
271 * ```typescript
272 * const stream = yield* AWS.Firehose.DeliveryStream("Events", {
273 * destination: {
274 * bucketArn: bucket.bucketArn,
275 * prefix: "events/",
276 * errorOutputPrefix: "errors/",
277 * bufferingInterval: "1 minute",
278 * bufferingSizeInMBs: 1,
279 * compressionFormat: "GZIP",
280 * },
281 * });
282 * ```
283 *
284 * **Example:** Server-side encryption at rest
285 * ```typescript
286 * const stream = yield* AWS.Firehose.DeliveryStream("Events", {
287 * destination: { bucketArn: bucket.bucketArn },
288 * encryption: { keyType: "AWS_OWNED_CMK" },
289 * });
290 * ```
291 *
292 * **Example:** Kinesis Data Stream as source
293 * ```typescript
294 * const source = yield* AWS.Kinesis.Stream("Clickstream");
295 * const stream = yield* AWS.Firehose.DeliveryStream("ClickstreamArchive", {
296 * source: { kinesisStreamArn: source.streamArn },
297 * destination: { bucketArn: bucket.bucketArn },
298 * });
299 * ```
300 *
301 * ### Runtime Producers
302 * Bind producer operations in the init phase and use them in runtime
303 * handlers. Records are buffered by Firehose and appear in S3 after the
304 * buffering interval elapses.
305 *
306 * **Example:** Put a record from a handler
307 * ```typescript
308 * // init
309 * const putRecord = yield* AWS.Firehose.PutRecord(stream);
310 *
311 * return {
312 * fetch: Effect.gen(function* () {
313 * // runtime
314 * const response = yield* putRecord({
315 * Record: { Data: new TextEncoder().encode("hello\n") },
316 * });
317 * return HttpServerResponse.json({ recordId: response.RecordId });
318 * }),
319 * };
320 * ```
321 *
322 * **Example:** Put a batch of records
323 * ```typescript
324 * // init
325 * const putRecordBatch = yield* AWS.Firehose.PutRecordBatch(stream);
326 *
327 * // runtime
328 * const response = yield* putRecordBatch({
329 * Records: lines.map((line) => ({
330 * Data: new TextEncoder().encode(`${line}\n`),
331 * })),
332 * });
333 * ```
334 *
335 * @resource
336 */
337export const DeliveryStream = Resource<DeliveryStream>(
338 "AWS.Firehose.DeliveryStream",
339);
340
341/**
342 * The delivery stream entered `CREATING_FAILED` — AWS never recovers this
343 * state; the stream must be deleted and recreated.
344 */
345export class DeliveryStreamCreateFailed extends Data.TaggedError(
346 "DeliveryStreamCreateFailed",
347)<{
348 readonly deliveryStreamName: string;
349 readonly details: string | undefined;
350}> {}
351
352/**
353 * Validation error raised before any AWS call when the props are invalid.
354 */
355export class DeliveryStreamValidationError extends Data.TaggedError(
356 "DeliveryStreamValidationError",
357)<{
358 readonly message: string;
359}> {}
360
361/**
362 * Server-side encryption reached a terminal failure state
363 * (`ENABLING_FAILED` / `DISABLING_FAILED`) while converging to the desired
364 * encryption configuration.
365 */
366export class DeliveryStreamEncryptionFailed extends Data.TaggedError(
367 "DeliveryStreamEncryptionFailed",
368)<{
369 readonly deliveryStreamName: string;
370 readonly status: string;
371 readonly details: string | undefined;
372}> {}
373
374class DeliveryStreamNotActive extends Data.TaggedError(
375 "DeliveryStreamNotActive",
376)<{
377 readonly status: string;
378}> {}
379
380class DeliveryStreamEncryptionPending extends Data.TaggedError(
381 "DeliveryStreamEncryptionPending",
382) {}
383
384class DeliveryStreamStillExists extends Data.TaggedError(
385 "DeliveryStreamStillExists",
386) {}
387
388const defaultBufferingIntervalInSeconds = 300;
389const defaultBufferingSizeInMBs = 5;
390const defaultCompressionFormat = "UNCOMPRESSED" as const;
391
392const createDeliveryStreamName = (
393 id: string,
394 props: { deliveryStreamName?: string | undefined },
395) =>
396 Effect.gen(function* () {
397 if (props.deliveryStreamName) {
398 return props.deliveryStreamName;
399 }
400 return yield* createPhysicalName({ id, maxLength: 64 });
401 });
402
403/**
404 * Data-FIRST `Effect.retry(self, options)` wrapped with an explicit return
405 * annotation — inlining `Effect.retry` with an options object in provider
406 * lifecycle code leaves `Retry.Return`'s conditional type unresolved in the
407 * provider's inferred layer type, which TypeScript's declaration emit widens
408 * to an `unknown` R — poisoning `AWS.providers()` for every downstream
409 * consumer (see `retryThroughDeletionWindow` in SecretsManager/Secret.ts).
410 *
411 * Retries `InvalidArgumentException`s that mention the IAM role — a freshly
412 * created role takes a few seconds to become assumable by the Firehose
413 * service principal, and `CreateDeliveryStream` surfaces that propagation
414 * window as a role-related `InvalidArgumentException`.
415 */
416const retryThroughRolePropagation = <A, E extends { _tag: string }, R>(
417 self: Effect.Effect<A, E, R>,
418): Effect.Effect<A, E, R> =>
419 Effect.retry(self, {
420 while: (e) =>
421 e._tag === "InvalidArgumentException" &&
422 /role|assume|trust/i.test((e as { message?: string }).message ?? ""),
423 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(15)]),
424 });
425
426/**
427 * Retries `ResourceInUseException` (stream still `CREATING`, or a concurrent
428 * operation in flight) on a bounded schedule. Explicitly annotated for the
429 * same declaration-emit reason as {@link retryThroughRolePropagation}.
430 */
431const retryWhileInUse = <A, E extends { _tag: string }, R>(
432 self: Effect.Effect<A, E, R>,
433): Effect.Effect<A, E, R> =>
434 Effect.retry(self, {
435 while: (e) => e._tag === "ResourceInUseException",
436 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(30)]),
437 });
438
439/**
440 * Retries `ConcurrentModificationException` from `UpdateDestination` — the
441 * read-modify-write cycle re-reads the fresh `VersionId` on every attempt.
442 * Explicitly annotated for the same declaration-emit reason as
443 * {@link retryThroughRolePropagation}.
444 */
445const retryConcurrentModification = <A, E extends { _tag: string }, R>(
446 self: Effect.Effect<A, E, R>,
447): Effect.Effect<A, E, R> =>
448 Effect.retry(self, {
449 while: (e) => e._tag === "ConcurrentModificationException",
450 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(5)]),
451 });
452
453const toTagRecord = (
454 tags: ReadonlyArray<{ Key: string; Value?: string }> | undefined,
455) =>
456 Object.fromEntries(
457 (tags ?? [])
458 .filter(
459 (tag): tag is { Key: string; Value: string } =>
460 typeof tag.Value === "string",
461 )
462 .map((tag) => [tag.Key, tag.Value]),
463 );
464
465const toAttrs = ({
466 description,
467 tags,
468 roleName,
469}: {
470 description: firehose.DeliveryStreamDescription;
471 tags: Record<string, string>;
472 roleName: string | undefined;
473}): DeliveryStream["Attributes"] => {
474 const destination = description.Destinations[0];
475 const s3 = destination?.ExtendedS3DestinationDescription;
476 const encryption = description.DeliveryStreamEncryptionConfiguration;
477 return {
478 deliveryStreamName: description.DeliveryStreamName,
479 deliveryStreamArn: description.DeliveryStreamARN as DeliveryStreamArn,
480 deliveryStreamStatus:
481 description.DeliveryStreamStatus as DeliveryStreamStatus,
482 deliveryStreamType:
483 description.DeliveryStreamType as DeliveryStreamSourceType,
484 versionId: description.VersionId,
485 destinationId: destination?.DestinationId,
486 bucketArn: s3?.BucketARN ?? "",
487 roleArn: s3?.RoleARN ?? "",
488 roleName,
489 kinesisStreamArn:
490 description.Source?.KinesisStreamSourceDescription?.KinesisStreamARN,
491 prefix: s3?.Prefix || undefined,
492 errorOutputPrefix: s3?.ErrorOutputPrefix || undefined,
493 bufferingIntervalInSeconds: s3?.BufferingHints?.IntervalInSeconds,
494 bufferingSizeInMBs: s3?.BufferingHints?.SizeInMBs,
495 compressionFormat: s3?.CompressionFormat as CompressionFormat | undefined,
496 encryptionStatus: encryption?.Status as
497 | DeliveryStreamEncryptionStatus
498 | undefined,
499 encryptionKeyType:
500 encryption?.Status === "ENABLED"
501 ? (encryption.KeyType as DeliveryStreamEncryptionKeyType | undefined)
502 : undefined,
503 encryptionKeyArn:
504 encryption?.Status === "ENABLED" ? encryption.KeyARN : undefined,
505 tags,
506 };
507};
508
509const describeDeliveryStream = Effect.fn(function* (
510 deliveryStreamName: string,
511) {
512 const response = yield* firehose
513 .describeDeliveryStream({ DeliveryStreamName: deliveryStreamName })
514 .pipe(
515 Effect.catchTag("ResourceNotFoundException", () =>
516 Effect.succeed(undefined),
517 ),
518 );
519 return response?.DeliveryStreamDescription;
520});
521
522const readDeliveryStream = Effect.fn(function* ({
523 deliveryStreamName,
524 roleName,
525}: {
526 deliveryStreamName: string;
527 roleName: string | undefined;
528}) {
529 const description = yield* describeDeliveryStream(deliveryStreamName);
530 if (!description) {
531 return undefined;
532 }
533 // The stream can vanish between the describe above and the tag read (e.g.
534 // a concurrent destroy) — a typed `ResourceNotFoundException` here just
535 // means it's gone.
536 const tagsResponse = yield* firehose
537 .listTagsForDeliveryStream({ DeliveryStreamName: deliveryStreamName })
538 .pipe(
539 Effect.catchTag("ResourceNotFoundException", () =>
540 Effect.succeed(undefined),
541 ),
542 );
543 if (!tagsResponse) {
544 return undefined;
545 }
546 return toAttrs({
547 description,
548 tags: toTagRecord(tagsResponse.Tags),
549 roleName,
550 });
551});
552
553// Firehose activates fast (usually well under a minute); 2s × 45 bounds the
554// wait at 90s.
555const waitForDeliveryStreamActive = (deliveryStreamName: string) =>
556 Effect.gen(function* () {
557 const description = yield* describeDeliveryStream(deliveryStreamName);
558 if (!description) {
559 return yield* Effect.fail(
560 new DeliveryStreamNotActive({ status: "MISSING" }),
561 );
562 }
563 if (description.DeliveryStreamStatus === "CREATING_FAILED") {
564 return yield* Effect.fail(
565 new DeliveryStreamCreateFailed({
566 deliveryStreamName,
567 details: description.FailureDescription?.Details,
568 }),
569 );
570 }
571 if (description.DeliveryStreamStatus !== "ACTIVE") {
572 return yield* Effect.fail(
573 new DeliveryStreamNotActive({
574 status: description.DeliveryStreamStatus,
575 }),
576 );
577 }
578 return description;
579 }).pipe(
580 Effect.retry({
581 while: (e: { _tag: string }) => e._tag === "DeliveryStreamNotActive",
582 schedule: Schedule.max([
583 Schedule.fixed("2 seconds"),
584 Schedule.recurs(45),
585 ]),
586 }),
587 );
588
589// SSE transitions (ENABLING → ENABLED, DISABLING → DISABLED) settle in
590// seconds for the AWS-owned CMK; 2s × 45 bounds the wait at 90s. Returns the
591// terminal encryption configuration (or `undefined` when the stream has
592// never been encrypted).
593const waitForEncryptionSettled = (deliveryStreamName: string) =>
594 Effect.gen(function* () {
595 const description = yield* describeDeliveryStream(deliveryStreamName);
596 const encryption = description?.DeliveryStreamEncryptionConfiguration;
597 if (
598 encryption?.Status === "ENABLING" ||
599 encryption?.Status === "DISABLING"
600 ) {
601 return yield* Effect.fail(new DeliveryStreamEncryptionPending());
602 }
603 return encryption;
604 }).pipe(
605 Effect.retry({
606 while: (e: { _tag: string }) =>
607 e._tag === "DeliveryStreamEncryptionPending",
608 schedule: Schedule.max([
609 Schedule.fixed("2 seconds"),
610 Schedule.recurs(45),
611 ]),
612 }),
613 );
614
615// Deletion is usually quick but AWS documents that DELETING can linger;
616// 3s × 50 bounds the wait at 150s.
617const waitForDeliveryStreamDeleted = (deliveryStreamName: string) =>
618 Effect.gen(function* () {
619 const description = yield* describeDeliveryStream(deliveryStreamName);
620 if (description !== undefined) {
621 return yield* Effect.fail(new DeliveryStreamStillExists());
622 }
623 }).pipe(
624 Effect.retry({
625 while: (e: { _tag: string }) => e._tag === "DeliveryStreamStillExists",
626 schedule: Schedule.max([
627 Schedule.fixed("3 seconds"),
628 Schedule.recurs(50),
629 ]),
630 }),
631 );
632
633const deleteRoleIfExists = Effect.fn(function* (roleName: string) {
634 yield* iam
635 .deleteRolePolicy({ RoleName: roleName, PolicyName: roleName })
636 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
637 yield* iam
638 .deleteRole({ RoleName: roleName })
639 .pipe(Effect.catchTag("NoSuchEntityException", () => Effect.void));
640});
641
642export const DeliveryStreamProvider = () =>
643 Provider.effect(
644 DeliveryStream,
645 Effect.gen(function* () {
646 const createRoleName = (id: string) =>
647 createPhysicalName({ id, maxLength: 64 });
648
649 // Ensure the synthesized IAM role exists and its inline policy matches
650 // the desired destination/source access. `createRole` tolerates the
651 // already-exists race; `putRolePolicy` is an idempotent upsert, so the
652 // policy re-converges on every reconcile (e.g. after a bucket change).
653 const ensureRole = Effect.fn(function* ({
654 id,
655 roleName,
656 bucketArn,
657 kinesisStreamArn,
658 }: {
659 id: string;
660 roleName: string;
661 bucketArn: string;
662 kinesisStreamArn: string | undefined;
663 }) {
664 const tags = yield* createInternalTags(id);
665 yield* iam
666 .createRole({
667 RoleName: roleName,
668 AssumeRolePolicyDocument: JSON.stringify({
669 Version: "2012-10-17",
670 Statement: [
671 {
672 Effect: "Allow",
673 Principal: { Service: "firehose.amazonaws.com" },
674 Action: "sts:AssumeRole",
675 },
676 ],
677 }),
678 Tags: createTagsList(tags),
679 })
680 .pipe(
681 Effect.catchTag("EntityAlreadyExistsException", () =>
682 iam.getRole({ RoleName: roleName }),
683 ),
684 );
685
686 const statements: PolicyStatement[] = [
687 {
688 Effect: "Allow",
689 Action: [
690 "s3:AbortMultipartUpload",
691 "s3:GetBucketLocation",
692 "s3:GetObject",
693 "s3:ListBucket",
694 "s3:ListBucketMultipartUploads",
695 "s3:PutObject",
696 ],
697 Resource: [bucketArn, `${bucketArn}/*`],
698 },
699 ];
700 if (kinesisStreamArn) {
701 statements.push({
702 Effect: "Allow",
703 Action: [
704 "kinesis:DescribeStream",
705 "kinesis:GetShardIterator",
706 "kinesis:GetRecords",
707 "kinesis:ListShards",
708 ],
709 Resource: [kinesisStreamArn],
710 });
711 }
712
713 yield* iam.putRolePolicy({
714 RoleName: roleName,
715 PolicyName: roleName,
716 PolicyDocument: JSON.stringify({
717 Version: "2012-10-17",
718 Statement: statements,
719 }),
720 });
721 });
722
723 return DeliveryStream.Provider.of({
724 stables: [
725 "deliveryStreamName",
726 "deliveryStreamArn",
727 "deliveryStreamType",
728 "roleName",
729 ],
730
731 // Enumerate every delivery stream in the ambient account/region.
732 // `listDeliveryStreams` pages via ExclusiveStartDeliveryStreamName +
733 // HasMoreDeliveryStreams; hydrate each name into the `read` shape
734 // with bounded concurrency, dropping streams that vanish mid-flight.
735 list: () =>
736 Effect.gen(function* () {
737 const names: string[] = [];
738 let exclusiveStart: string | undefined;
739 while (true) {
740 const page = yield* firehose.listDeliveryStreams({
741 Limit: 100,
742 ExclusiveStartDeliveryStreamName: exclusiveStart,
743 });
744 names.push(...page.DeliveryStreamNames);
745 if (
746 !page.HasMoreDeliveryStreams ||
747 page.DeliveryStreamNames.length === 0
748 ) {
749 break;
750 }
751 exclusiveStart = names[names.length - 1];
752 }
753
754 const hydrated = yield* Effect.forEach(
755 names,
756 (deliveryStreamName) =>
757 readDeliveryStream({ deliveryStreamName, roleName: undefined }),
758 { concurrency: 10 },
759 );
760
761 return hydrated.filter(
762 (attrs): attrs is DeliveryStream["Attributes"] =>
763 attrs !== undefined,
764 );
765 }),
766
767 read: Effect.fn(function* ({ id, olds, output }) {
768 const deliveryStreamName =
769 output?.deliveryStreamName ??
770 (yield* createDeliveryStreamName(id, olds ?? {}));
771 const state = yield* readDeliveryStream({
772 deliveryStreamName,
773 roleName: output?.roleName,
774 });
775 if (!state) return undefined;
776 return (yield* hasAlchemyTags(id, state.tags as Tags))
777 ? state
778 : Unowned(state);
779 }),
780
781 diff: Effect.fn(function* ({ id, news, olds }) {
782 if (!isResolved(news)) return undefined;
783 const oldName = yield* createDeliveryStreamName(id, olds ?? {});
784 const newName = yield* createDeliveryStreamName(id, news ?? {});
785 if (oldName !== newName) {
786 return { action: "replace" } as const;
787 }
788 // The source (DirectPut vs KinesisStreamAsSource, and which stream)
789 // is immutable — any change replaces the delivery stream.
790 const oldSource = olds?.source;
791 const newSource = news?.source;
792 if (
793 (oldSource === undefined) !== (newSource === undefined) ||
794 oldSource?.kinesisStreamArn !== newSource?.kinesisStreamArn
795 ) {
796 return { action: "replace" } as const;
797 }
798 }),
799
800 reconcile: Effect.fn(function* ({ id, news, output, session }) {
801 const { accountId } = yield* AWSEnvironment.current;
802
803 if (!news?.destination?.bucketArn) {
804 return yield* Effect.fail(
805 new DeliveryStreamValidationError({
806 message: `DeliveryStream "${id}" requires destination.bucketArn`,
807 }),
808 );
809 }
810 if (news.encryption && news.source) {
811 return yield* Effect.fail(
812 new DeliveryStreamValidationError({
813 message: `DeliveryStream "${id}" cannot enable server-side encryption on a KinesisStreamAsSource stream — encryption is inherited from the source Kinesis stream`,
814 }),
815 );
816 }
817 if (
818 news.encryption?.keyType === "CUSTOMER_MANAGED_CMK" &&
819 !news.encryption.keyArn
820 ) {
821 return yield* Effect.fail(
822 new DeliveryStreamValidationError({
823 message: `DeliveryStream "${id}" requires encryption.keyArn when keyType is CUSTOMER_MANAGED_CMK`,
824 }),
825 );
826 }
827
828 const deliveryStreamName =
829 output?.deliveryStreamName ??
830 (yield* createDeliveryStreamName(id, news));
831 const internalTags = yield* createInternalTags(id);
832 const desiredTags = { ...news.tags, ...internalTags };
833
834 // Synthesize the IAM role when the caller didn't supply one for
835 // the destination (and/or the Kinesis source).
836 const needsRole =
837 !news.destination.roleArn ||
838 (news.source !== undefined && !news.source.roleArn);
839 const roleName = needsRole
840 ? (output?.roleName ?? (yield* createRoleName(id)))
841 : undefined;
842 const synthesizedRoleArn = roleName
843 ? `arn:aws:iam::${accountId}:role/${roleName}`
844 : undefined;
845 const destinationRoleArn =
846 news.destination.roleArn ?? synthesizedRoleArn!;
847 const sourceRoleArn = news.source
848 ? (news.source.roleArn ?? synthesizedRoleArn!)
849 : undefined;
850
851 if (roleName) {
852 yield* ensureRole({
853 id,
854 roleName,
855 bucketArn: news.destination.bucketArn,
856 kinesisStreamArn: news.source?.kinesisStreamArn,
857 });
858 }
859
860 // Observe — cloud state is authoritative; `output` is only a cache
861 // for the physical name.
862 let observed = yield* describeDeliveryStream(deliveryStreamName);
863
864 // A stream stuck in CREATING_FAILED / DELETING_FAILED can never
865 // recover — converge by deleting it and recreating below.
866 if (
867 observed &&
868 (observed.DeliveryStreamStatus === "CREATING_FAILED" ||
869 observed.DeliveryStreamStatus === "DELETING_FAILED")
870 ) {
871 yield* session.note(
872 `Delivery stream ${deliveryStreamName} is ${observed.DeliveryStreamStatus}; deleting and recreating`,
873 );
874 yield* firehose
875 .deleteDeliveryStream({
876 DeliveryStreamName: deliveryStreamName,
877 AllowForceDelete: true,
878 })
879 .pipe(
880 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
881 );
882 yield* waitForDeliveryStreamDeleted(deliveryStreamName);
883 observed = undefined;
884 }
885
886 const desiredBufferingHints = {
887 IntervalInSeconds:
888 toWireSeconds(news.destination.bufferingInterval) ??
889 defaultBufferingIntervalInSeconds,
890 SizeInMBs:
891 news.destination.bufferingSizeInMBs ?? defaultBufferingSizeInMBs,
892 };
893 const desiredCompression =
894 news.destination.compressionFormat ?? defaultCompressionFormat;
895
896 // Ensure — create the stream if it's missing. Tolerate the
897 // already-exists race (`ResourceInUseException`) and retry through
898 // the fresh role's IAM propagation window.
899 if (observed === undefined) {
900 yield* retryThroughRolePropagation(
901 firehose
902 .createDeliveryStream({
903 DeliveryStreamName: deliveryStreamName,
904 DeliveryStreamType: news.source
905 ? "KinesisStreamAsSource"
906 : "DirectPut",
907 KinesisStreamSourceConfiguration: news.source
908 ? {
909 KinesisStreamARN: news.source.kinesisStreamArn,
910 RoleARN: sourceRoleArn!,
911 }
912 : undefined,
913 DeliveryStreamEncryptionConfigurationInput: news.encryption
914 ? {
915 KeyType: news.encryption.keyType,
916 KeyARN: news.encryption.keyArn,
917 }
918 : undefined,
919 ExtendedS3DestinationConfiguration: {
920 RoleARN: destinationRoleArn,
921 BucketARN: news.destination.bucketArn,
922 Prefix: news.destination.prefix,
923 ErrorOutputPrefix: news.destination.errorOutputPrefix,
924 BufferingHints: desiredBufferingHints,
925 CompressionFormat: desiredCompression,
926 },
927 Tags: createTagsList(desiredTags),
928 })
929 .pipe(
930 Effect.catchTag("ResourceInUseException", () => Effect.void),
931 ),
932 );
933 yield* session.note(
934 `Creating delivery stream ${deliveryStreamName}...`,
935 );
936 }
937
938 // Both a fresh create and a crashed prior run land here in
939 // `CREATING` — wait for `ACTIVE` before syncing (bounded).
940 const active = yield* waitForDeliveryStreamActive(deliveryStreamName);
941
942 // Sync destination settings — diff observed against desired and
943 // apply only the delta via UpdateDestination (read-modify-write on
944 // VersionId, retried through ConcurrentModificationException).
945 const syncDestination = Effect.gen(function* () {
946 const description =
947 yield* describeDeliveryStream(deliveryStreamName);
948 if (!description) return;
949 const destination = description.Destinations[0];
950 const s3 = destination?.ExtendedS3DestinationDescription;
951 if (!destination || !s3) return;
952
953 const delta =
954 s3.RoleARN !== destinationRoleArn ||
955 s3.BucketARN !== news.destination.bucketArn ||
956 (s3.Prefix ?? "") !== (news.destination.prefix ?? "") ||
957 (s3.ErrorOutputPrefix ?? "") !==
958 (news.destination.errorOutputPrefix ?? "") ||
959 s3.BufferingHints.IntervalInSeconds !==
960 desiredBufferingHints.IntervalInSeconds ||
961 s3.BufferingHints.SizeInMBs !== desiredBufferingHints.SizeInMBs ||
962 s3.CompressionFormat !== desiredCompression;
963 if (!delta) return;
964
965 yield* firehose.updateDestination({
966 DeliveryStreamName: deliveryStreamName,
967 CurrentDeliveryStreamVersionId: description.VersionId,
968 DestinationId: destination.DestinationId,
969 ExtendedS3DestinationUpdate: {
970 RoleARN: destinationRoleArn,
971 BucketARN: news.destination.bucketArn,
972 Prefix: news.destination.prefix ?? "",
973 ErrorOutputPrefix: news.destination.errorOutputPrefix ?? "",
974 BufferingHints: desiredBufferingHints,
975 CompressionFormat: desiredCompression,
976 },
977 });
978 yield* session.note(
979 `Updated destination settings for ${deliveryStreamName}`,
980 );
981 });
982 yield* retryConcurrentModification(syncDestination);
983
984 // Sync server-side encryption — settle any in-flight transition
985 // (a crashed prior run may have left ENABLING/DISABLING), diff the
986 // observed terminal state against the desired config, and apply
987 // Start/Stop only on a delta. Both operations are async; the
988 // bounded settle-wait converges them. A terminal *_FAILED status
989 // also mismatches the desired state, so the op is re-applied.
990 const observedEncryption =
991 yield* waitForEncryptionSettled(deliveryStreamName);
992 const encryptionEnabled = observedEncryption?.Status === "ENABLED";
993 if (news.encryption) {
994 const keyMatches =
995 encryptionEnabled &&
996 observedEncryption?.KeyType === news.encryption.keyType &&
997 (news.encryption.keyType === "AWS_OWNED_CMK" ||
998 observedEncryption?.KeyARN === news.encryption.keyArn);
999 if (!keyMatches) {
1000 yield* retryWhileInUse(
1001 firehose.startDeliveryStreamEncryption({
1002 DeliveryStreamName: deliveryStreamName,
1003 DeliveryStreamEncryptionConfigurationInput: {
1004 KeyType: news.encryption.keyType,
1005 KeyARN: news.encryption.keyArn,
1006 },
1007 }),
1008 );
1009 const settled =
1010 yield* waitForEncryptionSettled(deliveryStreamName);
1011 if (settled?.Status !== "ENABLED") {
1012 return yield* Effect.fail(
1013 new DeliveryStreamEncryptionFailed({
1014 deliveryStreamName,
1015 status: settled?.Status ?? "UNKNOWN",
1016 details: settled?.FailureDescription?.Details,
1017 }),
1018 );
1019 }
1020 yield* session.note(
1021 `Enabled SSE (${news.encryption.keyType}) on ${deliveryStreamName}`,
1022 );
1023 }
1024 } else if (encryptionEnabled) {
1025 yield* retryWhileInUse(
1026 firehose.stopDeliveryStreamEncryption({
1027 DeliveryStreamName: deliveryStreamName,
1028 }),
1029 );
1030 const settled = yield* waitForEncryptionSettled(deliveryStreamName);
1031 if (settled !== undefined && settled.Status !== "DISABLED") {
1032 return yield* Effect.fail(
1033 new DeliveryStreamEncryptionFailed({
1034 deliveryStreamName,
1035 status: settled.Status ?? "UNKNOWN",
1036 details: settled.FailureDescription?.Details,
1037 }),
1038 );
1039 }
1040 yield* session.note(`Disabled SSE on ${deliveryStreamName}`);
1041 }
1042
1043 // Sync tags — diff against OBSERVED cloud tags (adoption may bring
1044 // a stream with foreign tags), never olds/output.
1045 const observedTagsResponse = yield* firehose
1046 .listTagsForDeliveryStream({
1047 DeliveryStreamName: deliveryStreamName,
1048 })
1049 .pipe(
1050 Effect.catchTag("ResourceNotFoundException", () =>
1051 Effect.succeed(undefined),
1052 ),
1053 );
1054 const observedTags = toTagRecord(observedTagsResponse?.Tags);
1055 const { removed, upsert } = diffTags(observedTags, desiredTags);
1056 if (removed.length > 0) {
1057 yield* firehose.untagDeliveryStream({
1058 DeliveryStreamName: deliveryStreamName,
1059 TagKeys: removed,
1060 });
1061 }
1062 if (upsert.length > 0) {
1063 yield* firehose.tagDeliveryStream({
1064 DeliveryStreamName: deliveryStreamName,
1065 Tags: upsert,
1066 });
1067 }
1068
1069 yield* session.note(active.DeliveryStreamARN);
1070
1071 // Return fresh attributes reflecting post-sync cloud state.
1072 const final = yield* readDeliveryStream({
1073 deliveryStreamName,
1074 roleName,
1075 });
1076 if (!final) {
1077 return yield* Effect.fail(
1078 new DeliveryStreamValidationError({
1079 message: `failed to read reconciled delivery stream ${deliveryStreamName}`,
1080 }),
1081 );
1082 }
1083 return final;
1084 }),
1085
1086 delete: Effect.fn(function* ({ output }) {
1087 // A stream in CREATING can't be deleted — retry through the window
1088 // (bounded). Already-gone is success.
1089 yield* retryWhileInUse(
1090 firehose
1091 .deleteDeliveryStream({
1092 DeliveryStreamName: output.deliveryStreamName,
1093 AllowForceDelete: true,
1094 })
1095 .pipe(
1096 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
1097 ),
1098 );
1099 yield* waitForDeliveryStreamDeleted(output.deliveryStreamName);
1100
1101 // Remove the synthesized IAM role (idempotent — it may already be
1102 // gone from a previous partial delete).
1103 if (output.roleName) {
1104 yield* deleteRoleIfExists(output.roleName);
1105 }
1106 }),
1107 });
1108 }),
1109 );