1import * as kinesis from "@distilled.cloud/aws/kinesis";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import type { Input } from "../../Input.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import {
13 createInternalTags,
14 diffTags,
15 hasAlchemyTags,
16 type Tags,
17} from "../../Tags.ts";
18import type { Providers } from "../Providers.ts";
19import type { StreamArn } from "./Stream.ts";
20
21export type ConsumerName = string;
22
23export type ConsumerArn = string;
24
25export type ConsumerStatus = "CREATING" | "DELETING" | "ACTIVE";
26
27export interface StreamConsumerProps {
28 /**
29 * ARN of the stream that owns the consumer.
30 */
31 streamArn: Input<StreamArn>;
32 /**
33 * Name of the stream consumer.
34 * @default ${app}-${stage}-${id}
35 */
36 consumerName?: string;
37 /**
38 * Tags to associate with the consumer.
39 */
40 tags?: Record<string, string>;
41}
42
43export interface StreamConsumer extends Resource<
44 "AWS.Kinesis.StreamConsumer",
45 StreamConsumerProps,
46 {
47 /**
48 * The consumer's physical name.
49 */
50 consumerName: ConsumerName;
51 /**
52 * ARN of the registered consumer.
53 */
54 consumerArn: ConsumerArn;
55 /**
56 * Current lifecycle status of the consumer.
57 */
58 consumerStatus: ConsumerStatus;
59 /**
60 * ARN of the stream the consumer is registered on.
61 */
62 streamArn: StreamArn;
63 /**
64 * When the consumer was registered.
65 */
66 consumerCreationTimestamp: Date;
67 /**
68 * Current tags reported for the consumer.
69 */
70 tags: Record<string, string>;
71 },
72 never,
73 Providers
74> {}
75
76/**
77 * A registered Kinesis enhanced fan-out consumer.
78 *
79 * `StreamConsumer` is the canonical lifecycle resource for
80 * `RegisterStreamConsumer` / `DeregisterStreamConsumer`.
81 * ### Creating Consumers
82 * **Example:** Register a Consumer
83 * ```typescript
84 * const consumer = yield* StreamConsumer("AnalyticsConsumer", {
85 * streamArn: stream.streamArn,
86 * });
87 * ```
88 *
89 * @resource
90 */
91export const StreamConsumer = Resource<StreamConsumer>(
92 "AWS.Kinesis.StreamConsumer",
93);
94
95const createConsumerName = (
96 id: string,
97 props: {
98 consumerName?: string | undefined;
99 },
100) =>
101 Effect.gen(function* () {
102 if (props.consumerName) {
103 return props.consumerName;
104 }
105 return yield* createPhysicalName({
106 id,
107 maxLength: 128,
108 });
109 });
110
111const toTagRecord = (
112 tags: Array<{ Key: string; Value?: string }> | undefined,
113) =>
114 Object.fromEntries(
115 (tags ?? [])
116 .filter(
117 (tag): tag is { Key: string; Value: string } =>
118 typeof tag.Value === "string",
119 )
120 .map((tag) => [tag.Key, tag.Value]),
121 );
122
123const toAttrs = ({
124 description,
125 tags,
126}: {
127 description: kinesis.ConsumerDescription;
128 tags: Record<string, string>;
129}): StreamConsumer["Attributes"] => ({
130 consumerName: description.ConsumerName,
131 consumerArn: description.ConsumerARN,
132 consumerStatus: description.ConsumerStatus as ConsumerStatus,
133 streamArn: description.StreamARN as StreamArn,
134 consumerCreationTimestamp: description.ConsumerCreationTimestamp,
135 tags,
136});
137
138const readConsumer = Effect.fn(function* ({
139 streamArn,
140 consumerName,
141 consumerArn,
142}: {
143 streamArn?: string;
144 consumerName?: string;
145 consumerArn?: string;
146}) {
147 const response = yield* kinesis
148 .describeStreamConsumer({
149 StreamARN: streamArn,
150 ConsumerName: consumerName,
151 ConsumerARN: consumerArn,
152 })
153 .pipe(
154 Effect.catchTag("ResourceNotFoundException", () =>
155 Effect.succeed(undefined),
156 ),
157 );
158
159 if (!response) {
160 return undefined;
161 }
162
163 const description = response.ConsumerDescription;
164 const tagsResponse = yield* kinesis.listTagsForResource({
165 ResourceARN: description.ConsumerARN,
166 });
167
168 return toAttrs({
169 description,
170 tags: toTagRecord(tagsResponse.Tags),
171 });
172});
173
174// Kinesis tells us "in use" via ResourceInUseException, but the consumer
175// registry is eventually consistent — describeStreamConsumer can briefly
176// return ResourceNotFoundException for a consumer the registry just
177// confirmed exists. Poll up to ~10s before giving up.
178class ConsumerRegistryNotConsistent extends Data.TaggedError(
179 "ConsumerRegistryNotConsistent",
180)<{ consumerName: string }> {}
181
182const adoptExistingConsumer = Effect.fn(function* (
183 streamArn: string,
184 consumerName: string,
185) {
186 return yield* Effect.gen(function* () {
187 const state = yield* readConsumer({
188 streamArn,
189 consumerName,
190 });
191
192 if (!state) {
193 return yield* new ConsumerRegistryNotConsistent({ consumerName });
194 }
195
196 return state;
197 }).pipe(
198 Effect.retry({
199 while: (e) => e._tag === "ConsumerRegistryNotConsistent",
200 schedule: Schedule.max([Schedule.exponential(250), Schedule.recurs(8)]),
201 }),
202 Effect.catchTag("ConsumerRegistryNotConsistent", () =>
203 Effect.fail(
204 new Error(
205 `consumer ${consumerName} exists but could not be read after ` +
206 `~10s of registry-consistency retries`,
207 ),
208 ),
209 ),
210 );
211});
212
213const waitForConsumerStatus = (
214 consumerArn: string,
215 expectedStatus: ConsumerStatus,
216) =>
217 Effect.gen(function* () {
218 yield* Effect.sleep("2 seconds");
219 const response = yield* kinesis.describeStreamConsumer({
220 ConsumerARN: consumerArn,
221 });
222 if (response.ConsumerDescription.ConsumerStatus !== expectedStatus) {
223 return yield* Effect.fail({ _tag: "ConsumerStatusNotReady" as const });
224 }
225 return response.ConsumerDescription;
226 }).pipe(
227 Effect.retry({
228 while: (e: { _tag: string }) =>
229 e._tag === "ConsumerStatusNotReady" || e._tag === "ParseError",
230 schedule: Schedule.max([Schedule.exponential(500), Schedule.recurs(60)]),
231 }),
232 );
233
234const waitForConsumerDeleted = (consumerArn: string) =>
235 Effect.gen(function* () {
236 yield* kinesis.describeStreamConsumer({
237 ConsumerARN: consumerArn,
238 });
239 return yield* Effect.fail({ _tag: "ConsumerStillExists" as const });
240 }).pipe(
241 Effect.retry({
242 while: (e: { _tag: string }) =>
243 e._tag === "ConsumerStillExists" || e._tag === "ParseError",
244 schedule: Schedule.max([Schedule.exponential(500), Schedule.recurs(60)]),
245 }),
246 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
247 );
248
249export const StreamConsumerProvider = () =>
250 Provider.succeed(StreamConsumer, {
251 stables: ["consumerArn", "consumerName"],
252 read: Effect.fn(function* ({ id, olds, output }) {
253 const consumerName =
254 output?.consumerName ?? (yield* createConsumerName(id, olds ?? {}));
255 const streamArn = output?.streamArn ?? olds?.streamArn;
256 // describeStreamConsumer rejects with InvalidArgumentException unless
257 // either consumerARN, or both streamARN + consumerName, are provided.
258 // If the engine probed for adoption before our upstream Stream was
259 // created we'll have neither — treat that as "doesn't exist".
260 if (typeof streamArn !== "string" && !output?.consumerArn) {
261 return undefined;
262 }
263 const state = yield* readConsumer({
264 streamArn: typeof streamArn === "string" ? streamArn : undefined,
265 consumerName,
266 consumerArn: output?.consumerArn,
267 });
268 if (!state) return undefined;
269 return (yield* hasAlchemyTags(id, state.tags as Tags))
270 ? state
271 : Unowned(state);
272 }),
273 diff: Effect.fn(function* ({ id, news, olds }) {
274 if (!isResolved(news)) return;
275 const oldConsumerName = yield* createConsumerName(id, olds);
276 const newConsumerName = yield* createConsumerName(id, news);
277 if (oldConsumerName !== newConsumerName) {
278 return { action: "replace" } as const;
279 }
280 if (
281 typeof news.streamArn === "string" &&
282 typeof olds.streamArn === "string" &&
283 news.streamArn !== olds.streamArn
284 ) {
285 return { action: "replace" } as const;
286 }
287 }),
288 reconcile: Effect.fn(function* ({ id, news, output, session }) {
289 const consumerName =
290 output?.consumerName ?? (yield* createConsumerName(id, news));
291 const streamArn = news.streamArn as string;
292 const internalTags = yield* createInternalTags(id);
293 const desiredTags = { ...internalTags, ...news.tags };
294
295 // Observe — fetch live cloud state. `output` is treated as a cache
296 // for the consumer ARN; the consumer's actual existence and tags are
297 // fetched fresh so the reconciler converges regardless of drift,
298 // adoption, or a partially-completed prior run.
299 let state = yield* readConsumer({
300 streamArn,
301 consumerName,
302 consumerArn: output?.consumerArn,
303 });
304
305 // Ensure — register the consumer if it's missing. Tolerate
306 // `ResourceInUseException` as a race with a peer reconciler: a
307 // brief registry-consistency wait, then re-read and continue the
308 // sync path.
309 if (state === undefined) {
310 yield* kinesis
311 .registerStreamConsumer({
312 StreamARN: streamArn,
313 ConsumerName: consumerName,
314 Tags: desiredTags,
315 })
316 .pipe(
317 // The upstream Stream provider waits for ACTIVE via
318 // describeStreamSummary, but the consumer-registry control plane
319 // is eventually consistent against that: registerStreamConsumer
320 // can briefly 404 ("Stream ... not found") for a stream that
321 // describeStreamSummary already reports ACTIVE. Ride out that
322 // propagation window before giving up.
323 Effect.retry({
324 while: (e) => e._tag === "ResourceNotFoundException",
325 schedule: Schedule.max([
326 Schedule.exponential(500),
327 Schedule.recurs(8),
328 ]),
329 }),
330 Effect.asVoid,
331 Effect.catchTag("ResourceInUseException", () =>
332 adoptExistingConsumer(streamArn, consumerName).pipe(
333 Effect.asVoid,
334 ),
335 ),
336 );
337
338 state = yield* adoptExistingConsumer(streamArn, consumerName);
339 yield* waitForConsumerStatus(state.consumerArn, "ACTIVE");
340
341 state = yield* readConsumer({
342 consumerArn: state.consumerArn,
343 streamArn,
344 consumerName,
345 });
346 if (state === undefined) {
347 return yield* Effect.fail(
348 new Error(`failed to read created consumer ${consumerName}`),
349 );
350 }
351 }
352
353 // Sync tags — diff observed cloud tags against desired. Adoption
354 // may bring us a consumer that already has its own tag set; diffing
355 // against `state.tags` (fetched fresh) lets the reconciler converge
356 // ownership without fighting whatever was there before.
357 const { removed, upsert } = diffTags(state.tags, desiredTags);
358
359 if (removed.length > 0) {
360 yield* kinesis.untagResource({
361 ResourceARN: state.consumerArn,
362 TagKeys: removed,
363 });
364 }
365
366 if (upsert.length > 0) {
367 const tagsToAdd: Record<string, string> = {};
368 for (const { Key, Value } of upsert) {
369 tagsToAdd[Key] = Value;
370 }
371 yield* kinesis.tagResource({
372 ResourceARN: state.consumerArn,
373 Tags: tagsToAdd,
374 });
375 }
376
377 // Re-read final state so the returned attributes reflect what's
378 // actually in the cloud after all sync steps.
379 const final = yield* readConsumer({
380 consumerArn: state.consumerArn,
381 });
382 if (!final) {
383 return yield* Effect.fail(
384 new Error(`failed to read reconciled consumer ${consumerName}`),
385 );
386 }
387
388 yield* session.note(final.consumerArn);
389 return final;
390 }),
391 delete: Effect.fn(function* ({ output }) {
392 yield* kinesis
393 .deregisterStreamConsumer({
394 ConsumerARN: output.consumerArn,
395 })
396 .pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
397
398 yield* waitForConsumerDeleted(output.consumerArn);
399 }),
400 // Consumers are stream-scoped: listStreamConsumers requires a StreamARN.
401 // Fan out — enumerate every stream in the account/region, then list the
402 // consumers registered on each stream, then hydrate tags per consumer.
403 list: Effect.fn(function* () {
404 const streamArns = yield* kinesis.listStreams.pages({}).pipe(
405 Stream.runCollect,
406 Effect.map((chunk) =>
407 Array.from(chunk).flatMap((page) =>
408 (page.StreamSummaries ?? [])
409 .map((summary) => summary.StreamARN)
410 .filter((arn): arn is string => typeof arn === "string"),
411 ),
412 ),
413 );
414
415 const consumers = yield* Effect.forEach(
416 streamArns,
417 (streamArn) =>
418 kinesis.listStreamConsumers.pages({ StreamARN: streamArn }).pipe(
419 Stream.runCollect,
420 Effect.map((chunk) =>
421 Array.from(chunk).flatMap((page) =>
422 (page.Consumers ?? []).map((consumer) => ({
423 streamArn,
424 consumer,
425 })),
426 ),
427 ),
428 // A stream may be deleted mid-enumeration; skip it.
429 Effect.catchTag("ResourceNotFoundException", () =>
430 Effect.succeed(
431 [] as { streamArn: string; consumer: kinesis.Consumer }[],
432 ),
433 ),
434 ),
435 { concurrency: 10 },
436 ).pipe(Effect.map((rows) => rows.flat()));
437
438 const hydrated = yield* Effect.forEach(
439 consumers,
440 ({ streamArn, consumer }) =>
441 kinesis
442 .listTagsForResource({ ResourceARN: consumer.ConsumerARN })
443 .pipe(
444 Effect.map((tagsResponse): StreamConsumer["Attributes"] => ({
445 consumerName: consumer.ConsumerName,
446 consumerArn: consumer.ConsumerARN,
447 consumerStatus: consumer.ConsumerStatus as ConsumerStatus,
448 streamArn: streamArn as StreamArn,
449 consumerCreationTimestamp: consumer.ConsumerCreationTimestamp,
450 tags: toTagRecord(tagsResponse.Tags),
451 })),
452 // A consumer may be deregistered between list and tag fetch.
453 Effect.catchTag("ResourceNotFoundException", () =>
454 Effect.succeed(undefined),
455 ),
456 ),
457 { concurrency: 10 },
458 );
459
460 return hydrated.filter(
461 (row): row is StreamConsumer["Attributes"] => row !== undefined,
462 );
463 }),
464 });