1import * as mq from "@distilled.cloud/aws/mq";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import type * as Redacted from "effect/Redacted";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
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 { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
13import type { Providers } from "../Providers.ts";
14import { syncMqTags, toTagRecord } from "./internal.ts";
15
16/**
17 * An ActiveMQ or RabbitMQ user provisioned on the broker at creation.
18 */
19export interface BrokerUser {
20 /**
21 * Username (2-100 chars, `a-z A-Z 0-9 _ ~ . -`). Must not contain commas,
22 * colons, equals signs, or the string `tag:`.
23 */
24 username: string;
25 /**
26 * Password (12-250 printable chars, no commas, colons, or equals signs).
27 */
28 password: Redacted.Redacted<string>;
29 /**
30 * Whether the user can access the ActiveMQ Web Console. Ignored for
31 * RabbitMQ.
32 * @default false
33 */
34 consoleAccess?: boolean;
35 /**
36 * ActiveMQ groups the user belongs to. Ignored for RabbitMQ.
37 */
38 groups?: string[];
39}
40
41/**
42 * Reference to a specific {@link Configuration} revision to apply to the
43 * broker.
44 */
45export interface BrokerConfigurationRef {
46 /** The configuration id (`configurationId` attribute of a Configuration). */
47 id: string;
48 /** The revision number to apply. */
49 revision?: number;
50}
51
52/**
53 * CloudWatch log exports to enable for the broker.
54 */
55export interface BrokerLogs {
56 /** Enable audit logging (ActiveMQ only). */
57 audit?: boolean;
58 /** Enable general logging. */
59 general?: boolean;
60}
61
62/**
63 * Weekly maintenance window during which the broker applies pending changes.
64 */
65export interface BrokerMaintenanceWindow {
66 /** Day of week, e.g. `"SUNDAY"`. */
67 dayOfWeek:
68 | "MONDAY"
69 | "TUESDAY"
70 | "WEDNESDAY"
71 | "THURSDAY"
72 | "FRIDAY"
73 | "SATURDAY"
74 | "SUNDAY";
75 /** Start time in 24h `HH:mm` format, e.g. `"03:00"`. */
76 timeOfDay: string;
77 /** IANA time zone, e.g. `"UTC"`. */
78 timeZone?: string;
79}
80
81/**
82 * At-rest encryption options for the broker.
83 */
84export interface BrokerEncryptionOptions {
85 /** Customer-managed KMS key id/ARN. Omit to use an AWS-owned key. */
86 kmsKeyId?: string;
87 /** Use an AWS-owned key rather than a customer-managed one. */
88 useAwsOwnedKey?: boolean;
89}
90
91export interface BrokerProps {
92 /**
93 * Name of the broker (1-50 chars, `a-z A-Z 0-9 _ ~ -`). If omitted a
94 * deterministic physical name is generated. Immutable — changing it
95 * replaces the broker.
96 */
97 brokerName?: string;
98 /**
99 * Broker engine. Immutable — changing it replaces the broker.
100 */
101 engineType: "ACTIVEMQ" | "RABBITMQ";
102 /**
103 * Broker engine version (e.g. `"5.18"` for ActiveMQ, `"3.13"` for
104 * RabbitMQ). Mutable — a change is applied as a pending broker update.
105 */
106 engineVersion?: string;
107 /**
108 * Broker instance type, e.g. `"mq.t3.micro"` (the cheapest) or
109 * `"mq.m5.large"`. Mutable — a change is applied as a pending broker
110 * update.
111 */
112 hostInstanceType: string;
113 /**
114 * Deployment topology. `SINGLE_INSTANCE` is a single broker (cheapest,
115 * no HA); `ACTIVE_STANDBY_MULTI_AZ` (ActiveMQ) and `CLUSTER_MULTI_AZ`
116 * (RabbitMQ) provide HA. Immutable — changing it replaces the broker.
117 * @default "SINGLE_INSTANCE"
118 */
119 deploymentMode?:
120 | "SINGLE_INSTANCE"
121 | "ACTIVE_STANDBY_MULTI_AZ"
122 | "CLUSTER_MULTI_AZ";
123 /**
124 * Broker users. ActiveMQ requires at least one; RabbitMQ requires exactly
125 * one. Users are provisioned at creation and are not reconciled on update.
126 */
127 users: BrokerUser[];
128 /**
129 * Whether the broker is reachable over the public internet. Immutable —
130 * changing it replaces the broker.
131 * @default false
132 */
133 publiclyAccessible?: boolean;
134 /**
135 * Subnet ids to deploy the broker into. `SINGLE_INSTANCE` needs one
136 * subnet; multi-AZ needs two (ActiveMQ) or three (RabbitMQ cluster). If
137 * omitted, a default-VPC subnet is chosen. Immutable — changing it
138 * replaces the broker.
139 */
140 subnetIds?: string[];
141 /**
142 * Security group ids controlling access to the broker. Mutable — a change
143 * is applied as a pending broker update.
144 */
145 securityGroups?: string[];
146 /**
147 * Authentication strategy. Mutable — a change is applied as a pending
148 * broker update.
149 * @default "SIMPLE"
150 */
151 authenticationStrategy?: "SIMPLE" | "LDAP";
152 /**
153 * Whether to automatically apply minor engine version upgrades during the
154 * maintenance window. RabbitMQ and newer ActiveMQ versions require `true`.
155 * Mutable.
156 * @default true
157 */
158 autoMinorVersionUpgrade?: boolean;
159 /**
160 * Storage type. `EBS` (durable, default) or `EFS` (ActiveMQ multi-AZ
161 * only). Immutable — changing it replaces the broker.
162 */
163 storageType?: "EBS" | "EFS";
164 /**
165 * A {@link Configuration} revision to apply. Mutable — a change is applied
166 * as a pending broker update.
167 */
168 configuration?: BrokerConfigurationRef;
169 /**
170 * CloudWatch log exports. Mutable.
171 */
172 logs?: BrokerLogs;
173 /**
174 * Weekly maintenance window. Mutable.
175 */
176 maintenanceWindow?: BrokerMaintenanceWindow;
177 /**
178 * At-rest encryption options. Immutable — changing it replaces the broker.
179 */
180 encryptionOptions?: BrokerEncryptionOptions;
181 /**
182 * User-defined tags for the broker.
183 */
184 tags?: Record<string, string>;
185}
186
187export interface Broker extends Resource<
188 "AWS.MQ.Broker",
189 BrokerProps,
190 {
191 /** Server-assigned unique id of the broker (e.g. `b-1234...`). */
192 brokerId: string;
193 /** ARN of the broker. */
194 brokerArn: string;
195 /** Name of the broker. */
196 brokerName: string;
197 /** Current lifecycle state (e.g. `CREATION_IN_PROGRESS`, `RUNNING`). */
198 brokerState: string;
199 /** Wire-level connection endpoints (protocol URIs) for the broker. */
200 endpoints: string[] | undefined;
201 /** ActiveMQ Web Console URL (undefined for RabbitMQ). */
202 consoleUrl: string | undefined;
203 },
204 never,
205 Providers
206> {}
207
208/**
209 * An Amazon MQ broker — a managed message broker running Apache ActiveMQ or
210 * RabbitMQ. Amazon MQ handles provisioning, patching, and (for multi-AZ
211 * deployments) failover, exposing standard wire protocols (OpenWire, AMQP,
212 * MQTT, STOMP, WSS) so existing clients connect unchanged.
213 *
214 * Broker creation and deletion are asynchronous and take several minutes;
215 * the provider waits (bounded) for the broker to reach `RUNNING` before
216 * returning, and waits for it to disappear on delete.
217 *
218 * ### Creating a Broker
219 * **Example:** Single-instance ActiveMQ (cheapest)
220 * ```typescript
221 * const broker = yield* MQ.Broker("Orders", {
222 * engineType: "ACTIVEMQ",
223 * engineVersion: "5.18",
224 * hostInstanceType: "mq.t3.micro",
225 * deploymentMode: "SINGLE_INSTANCE",
226 * publiclyAccessible: true,
227 * users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
228 * });
229 * // broker.endpoints -> ["ssl://b-xxxx-1.mq.us-west-2.amazonaws.com:61617", ...]
230 * ```
231 *
232 * **Example:** Single-instance RabbitMQ
233 * ```typescript
234 * const broker = yield* MQ.Broker("Events", {
235 * engineType: "RABBITMQ",
236 * engineVersion: "3.13",
237 * hostInstanceType: "mq.t3.micro",
238 * publiclyAccessible: true,
239 * users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
240 * });
241 * ```
242 *
243 * ### Networking and Encryption
244 * **Example:** Private broker in specific subnets with a customer KMS key
245 * ```typescript
246 * const broker = yield* MQ.Broker("Orders", {
247 * engineType: "ACTIVEMQ",
248 * engineVersion: "5.18",
249 * hostInstanceType: "mq.m5.large",
250 * deploymentMode: "ACTIVE_STANDBY_MULTI_AZ",
251 * publiclyAccessible: false,
252 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
253 * securityGroups: [group.groupId],
254 * encryptionOptions: { kmsKeyId: key.keyArn },
255 * users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
256 * });
257 * ```
258 *
259 * ### Logging and Maintenance
260 * **Example:** Enable CloudWatch logs and pin a maintenance window
261 * ```typescript
262 * const broker = yield* MQ.Broker("Orders", {
263 * engineType: "ACTIVEMQ",
264 * engineVersion: "5.18",
265 * hostInstanceType: "mq.t3.micro",
266 * users: [{ username: "admin", password: Redacted.make("SuperSecretPassw0rd") }],
267 * logs: { general: true, audit: true },
268 * maintenanceWindow: {
269 * dayOfWeek: "SUNDAY",
270 * timeOfDay: "03:00",
271 * timeZone: "UTC",
272 * },
273 * });
274 * ```
275 *
276 * ### Consuming Messages
277 * Subscribe a Lambda function to broker queues from the init phase via
278 * {@link consumeBrokerMessages}. The event-source mapping, IAM grants, and
279 * runtime dispatch are created automatically (provide
280 * `Lambda.BrokerEventSource` on the function).
281 *
282 * **Example:** Process queue messages in a Lambda function
283 * ```typescript
284 * // init
285 * yield* MQ.consumeBrokerMessages(
286 * broker,
287 * {
288 * queues: ["orders"],
289 * credentialsSecretArn: secret.secretArn,
290 * },
291 * (messages) =>
292 * messages.pipe(
293 * Stream.runForEach((message) =>
294 * Effect.log(`received: ${message.data}`),
295 * ),
296 * ),
297 * );
298 * ```
299 *
300 * @resource
301 */
302export const Broker = Resource<Broker>("AWS.MQ.Broker");
303
304class BrokerNotSettled extends Data.TaggedError("BrokerNotSettled")<{
305 readonly brokerId: string;
306 readonly state: string;
307}> {}
308
309const sameArray = (
310 a: readonly string[] | undefined,
311 b: readonly string[] | undefined,
312): boolean => {
313 const as = [...(a ?? [])].sort();
314 const bs = [...(b ?? [])].sort();
315 return as.length === bs.length && as.every((v, i) => v === bs[i]);
316};
317
318const toWireUsers = (users: BrokerUser[]): mq.User[] =>
319 users.map((u) => ({
320 Username: u.username,
321 // distilled marks Password sensitive — pass the Redacted value through so
322 // it stays redacted until wire encoding.
323 Password: u.password,
324 ConsoleAccess: u.consoleAccess,
325 Groups: u.groups,
326 }));
327
328export const BrokerProvider = () =>
329 Provider.effect(
330 Broker,
331 Effect.gen(function* () {
332 const toName = (id: string, props: Partial<BrokerProps>) =>
333 props.brokerName
334 ? Effect.succeed(props.brokerName)
335 : createPhysicalName({ id, maxLength: 50 });
336
337 /** Describe a broker by id; a missing broker reads as absent. */
338 const readBroker = Effect.fn(function* (brokerId: string) {
339 return yield* mq
340 .describeBroker({ BrokerId: brokerId })
341 .pipe(
342 Effect.catchTag("NotFoundException", () =>
343 Effect.succeed(undefined),
344 ),
345 );
346 });
347
348 /** Find a live broker by name (listBrokers has no name filter). */
349 const findByName = Effect.fn(function* (name: string) {
350 const summary = yield* mq.listBrokers.pages({}).pipe(
351 Stream.runCollect,
352 Effect.map((chunk) =>
353 Array.from(chunk)
354 .flatMap((page) => page.BrokerSummaries ?? [])
355 .find((s) => s.BrokerName === name && s.BrokerId),
356 ),
357 );
358 if (!summary?.BrokerId) return undefined;
359 return yield* readBroker(summary.BrokerId);
360 });
361
362 // Broker creation is asynchronous and typically takes 5-10 minutes;
363 // budget ~12 min (72 * 10s). Wait until the broker leaves a transient
364 // state (CREATION_IN_PROGRESS / REBOOT_IN_PROGRESS).
365 const waitForSettled = Effect.fn(function* (brokerId: string) {
366 return yield* readBroker(brokerId).pipe(
367 Effect.flatMap((broker) => {
368 const state = broker?.BrokerState;
369 if (
370 state === "CREATION_IN_PROGRESS" ||
371 state === "REBOOT_IN_PROGRESS"
372 ) {
373 return Effect.fail(new BrokerNotSettled({ brokerId, state }));
374 }
375 return Effect.succeed(broker);
376 }),
377 Effect.retry({
378 while: (e) => e instanceof BrokerNotSettled,
379 schedule: Schedule.max([
380 Schedule.fixed("10 seconds"),
381 Schedule.recurs(72),
382 ]),
383 }),
384 );
385 });
386
387 // Deletion is asynchronous too; wait until the broker is gone so
388 // networking dependencies (subnets, security groups) can be removed.
389 const waitUntilGone = Effect.fn(function* (brokerId: string) {
390 yield* readBroker(brokerId).pipe(
391 Effect.flatMap((broker) => {
392 if (broker === undefined) return Effect.void;
393 return Effect.fail(
394 new BrokerNotSettled({
395 brokerId,
396 state: broker.BrokerState ?? "UNKNOWN",
397 }),
398 );
399 }),
400 Effect.retry({
401 while: (e) => e instanceof BrokerNotSettled,
402 schedule: Schedule.max([
403 Schedule.fixed("10 seconds"),
404 Schedule.recurs(72),
405 ]),
406 }),
407 );
408 });
409
410 const toAttrs = (broker: mq.DescribeBrokerResponse) => ({
411 brokerId: broker.BrokerId!,
412 brokerArn: broker.BrokerArn!,
413 brokerName: broker.BrokerName!,
414 brokerState: broker.BrokerState!,
415 endpoints: broker.BrokerInstances?.[0]?.Endpoints,
416 consoleUrl: broker.BrokerInstances?.[0]?.ConsoleURL,
417 });
418
419 return {
420 stables: ["brokerId", "brokerArn", "brokerName"],
421
422 diff: Effect.fn(function* ({ id, olds, news }) {
423 if (!isResolved(news)) return undefined;
424 const replaced =
425 (yield* toName(id, olds ?? {})) !== (yield* toName(id, news ?? {}));
426 if (replaced) return { action: "replace" } as const;
427 // Create-only properties.
428 if (
429 (news.engineType ?? undefined) !== (olds?.engineType ?? undefined)
430 )
431 return { action: "replace" } as const;
432 if (
433 (news.deploymentMode ?? "SINGLE_INSTANCE") !==
434 (olds?.deploymentMode ?? "SINGLE_INSTANCE")
435 )
436 return { action: "replace" } as const;
437 if (
438 (news.publiclyAccessible ?? false) !==
439 (olds?.publiclyAccessible ?? false)
440 )
441 return { action: "replace" } as const;
442 if (
443 (news.storageType ?? undefined) !== (olds?.storageType ?? undefined)
444 )
445 return { action: "replace" } as const;
446 if (
447 news.subnetIds !== undefined &&
448 !sameArray(news.subnetIds, olds?.subnetIds)
449 )
450 return { action: "replace" } as const;
451 if (
452 (news.encryptionOptions?.kmsKeyId ?? undefined) !==
453 (olds?.encryptionOptions?.kmsKeyId ?? undefined) ||
454 (news.encryptionOptions?.useAwsOwnedKey ?? undefined) !==
455 (olds?.encryptionOptions?.useAwsOwnedKey ?? undefined)
456 )
457 return { action: "replace" } as const;
458 }),
459
460 read: Effect.fn(function* ({ id, olds, output }) {
461 const broker = output?.brokerId
462 ? yield* readBroker(output.brokerId)
463 : yield* findByName(yield* toName(id, olds ?? {}));
464 if (broker === undefined) return undefined;
465 const attrs = toAttrs(broker);
466 const tags = toTagRecord(broker.Tags);
467 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
468 }),
469
470 reconcile: Effect.fn(function* ({ id, news, output, session }) {
471 const name = output?.brokerName ?? (yield* toName(id, news));
472 const internalTags = yield* createInternalTags(id);
473 const desiredTags = { ...internalTags, ...news.tags };
474
475 // 1. Observe — cloud state is authoritative.
476 let observed = output?.brokerId
477 ? yield* readBroker(output.brokerId)
478 : undefined;
479 if (observed === undefined) {
480 observed = yield* findByName(name);
481 }
482
483 // 2. Ensure — create if missing. The deterministic CreatorRequestId
484 // makes a retried create idempotent. Tolerate a ConflictException
485 // race by re-finding the broker by name.
486 if (observed === undefined) {
487 const created = yield* mq
488 .createBroker({
489 BrokerName: name,
490 EngineType: news.engineType,
491 EngineVersion: news.engineVersion,
492 HostInstanceType: news.hostInstanceType,
493 DeploymentMode: news.deploymentMode ?? "SINGLE_INSTANCE",
494 PubliclyAccessible: news.publiclyAccessible ?? false,
495 AutoMinorVersionUpgrade: news.autoMinorVersionUpgrade ?? true,
496 AuthenticationStrategy: news.authenticationStrategy,
497 StorageType: news.storageType,
498 SubnetIds: news.subnetIds,
499 SecurityGroups: news.securityGroups,
500 Users: toWireUsers(news.users),
501 Configuration: news.configuration
502 ? {
503 Id: news.configuration.id,
504 Revision: news.configuration.revision,
505 }
506 : undefined,
507 Logs: news.logs
508 ? { Audit: news.logs.audit, General: news.logs.general }
509 : undefined,
510 MaintenanceWindowStartTime: news.maintenanceWindow
511 ? {
512 DayOfWeek: news.maintenanceWindow.dayOfWeek,
513 TimeOfDay: news.maintenanceWindow.timeOfDay,
514 TimeZone: news.maintenanceWindow.timeZone,
515 }
516 : undefined,
517 EncryptionOptions: news.encryptionOptions
518 ? {
519 KmsKeyId: news.encryptionOptions.kmsKeyId,
520 UseAwsOwnedKey: news.encryptionOptions.useAwsOwnedKey,
521 }
522 : undefined,
523 CreatorRequestId: name,
524 Tags: desiredTags,
525 })
526 .pipe(
527 Effect.catchTag("ConflictException", () =>
528 Effect.gen(function* () {
529 const existing = yield* findByName(name);
530 if (existing?.BrokerId) {
531 return { BrokerId: existing.BrokerId };
532 }
533 return yield* Effect.fail(
534 new Error(
535 `MQ broker '${name}' create conflicted but no broker was found`,
536 ),
537 );
538 }),
539 ),
540 );
541 observed = yield* readBroker(created.BrokerId!);
542 }
543
544 if (observed?.BrokerId === undefined) {
545 return yield* Effect.fail(
546 new Error(`MQ broker '${name}' could not be reconciled`),
547 );
548 }
549 const brokerId = observed.BrokerId;
550
551 // Wait (bounded) for the broker to settle before mutating it.
552 const settled = yield* waitForSettled(brokerId);
553 if (settled === undefined) {
554 return yield* Effect.fail(
555 new Error(`MQ broker '${name}' disappeared while reconciling`),
556 );
557 }
558 observed = settled;
559 if (observed.BrokerState === "CREATION_FAILED") {
560 return yield* Effect.fail(
561 new Error(
562 `MQ broker '${name}' failed to create (state: CREATION_FAILED)`,
563 ),
564 );
565 }
566
567 // 3. Sync mutable settings — compute an updateBroker request with
568 // only the drifted fields (each mutation becomes a pending change).
569 const update: Omit<mq.UpdateBrokerRequest, "BrokerId"> = {};
570 if (
571 news.engineVersion !== undefined &&
572 news.engineVersion !== observed.EngineVersion &&
573 news.engineVersion !== observed.PendingEngineVersion
574 ) {
575 update.EngineVersion = news.engineVersion;
576 }
577 if (
578 news.hostInstanceType !== observed.HostInstanceType &&
579 news.hostInstanceType !== observed.PendingHostInstanceType
580 ) {
581 update.HostInstanceType = news.hostInstanceType;
582 }
583 if (
584 news.autoMinorVersionUpgrade !== undefined &&
585 news.autoMinorVersionUpgrade !== observed.AutoMinorVersionUpgrade
586 ) {
587 update.AutoMinorVersionUpgrade = news.autoMinorVersionUpgrade;
588 }
589 if (
590 news.securityGroups !== undefined &&
591 !sameArray(news.securityGroups, observed.SecurityGroups)
592 ) {
593 update.SecurityGroups = news.securityGroups;
594 }
595 if (
596 news.logs !== undefined &&
597 (news.logs.audit !== observed.Logs?.Audit ||
598 news.logs.general !== observed.Logs?.General)
599 ) {
600 update.Logs = {
601 Audit: news.logs.audit,
602 General: news.logs.general,
603 };
604 }
605 if (
606 news.configuration !== undefined &&
607 (news.configuration.id !== observed.Configurations?.Current?.Id ||
608 news.configuration.revision !==
609 observed.Configurations?.Current?.Revision)
610 ) {
611 update.Configuration = {
612 Id: news.configuration.id,
613 Revision: news.configuration.revision,
614 };
615 }
616 if (
617 news.maintenanceWindow !== undefined &&
618 (news.maintenanceWindow.dayOfWeek !==
619 observed.MaintenanceWindowStartTime?.DayOfWeek ||
620 news.maintenanceWindow.timeOfDay !==
621 observed.MaintenanceWindowStartTime?.TimeOfDay ||
622 news.maintenanceWindow.timeZone !==
623 observed.MaintenanceWindowStartTime?.TimeZone)
624 ) {
625 update.MaintenanceWindowStartTime = {
626 DayOfWeek: news.maintenanceWindow.dayOfWeek,
627 TimeOfDay: news.maintenanceWindow.timeOfDay,
628 TimeZone: news.maintenanceWindow.timeZone,
629 };
630 }
631 if (
632 news.authenticationStrategy !== undefined &&
633 news.authenticationStrategy !== observed.AuthenticationStrategy &&
634 news.authenticationStrategy !==
635 observed.PendingAuthenticationStrategy
636 ) {
637 update.AuthenticationStrategy = news.authenticationStrategy;
638 }
639 if (Object.keys(update).length > 0) {
640 yield* mq.updateBroker({ BrokerId: brokerId, ...update });
641 }
642
643 // 3b. Sync tags — diff against OBSERVED cloud tags.
644 yield* syncMqTags(
645 observed.BrokerArn!,
646 toTagRecord(observed.Tags),
647 desiredTags,
648 );
649
650 // 4. Re-read for fresh attributes (endpoints appear once RUNNING).
651 const final = (yield* readBroker(brokerId)) ?? observed;
652 yield* session.note(name);
653 return toAttrs(final);
654 }),
655
656 delete: Effect.fn(function* ({ output }) {
657 // A broker mid-operation rejects deletion — wait for it to settle.
658 yield* waitForSettled(output.brokerId).pipe(
659 Effect.catch(() => Effect.void),
660 );
661 yield* mq
662 .deleteBroker({ BrokerId: output.brokerId })
663 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
664 yield* waitUntilGone(output.brokerId);
665 }),
666
667 list: () =>
668 mq.listBrokers.pages({}).pipe(
669 Stream.runCollect,
670 Effect.map((chunk) =>
671 Array.from(chunk)
672 .flatMap((page) => page.BrokerSummaries ?? [])
673 .flatMap((s) =>
674 s.BrokerId !== undefined &&
675 s.BrokerArn !== undefined &&
676 s.BrokerName !== undefined &&
677 s.BrokerState !== undefined
678 ? [
679 {
680 brokerId: s.BrokerId,
681 brokerArn: s.BrokerArn,
682 brokerName: s.BrokerName,
683 brokerState: s.BrokerState,
684 endpoints: undefined,
685 consoleUrl: undefined,
686 },
687 ]
688 : [],
689 ),
690 ),
691 ),
692 };
693 }),
694 );