1import * as eventbridge from "@distilled.cloud/aws/eventbridge";
2import * as Effect from "effect/Effect";
3import { Unowned } from "../../AdoptPolicy.ts";
4import { isResolved } from "../../Diff.ts";
5import type { Input } from "../../Input.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import {
10 createInternalTags,
11 createTagsList,
12 diffTags,
13 hasAlchemyTags,
14} from "../../Tags.ts";
15import { AWSEnvironment, type AccountID } from "../Environment.ts";
16import type { Providers } from "../Providers.ts";
17import type { RegionID } from "../Region.ts";
18
19export type {
20 AppSyncParameters,
21 AssignPublicIp,
22 AwsVpcConfiguration,
23 CapacityProviderStrategyItem,
24 HttpParameters,
25 InputTransformer,
26 KinesisParameters,
27 LaunchType,
28 NetworkConfiguration,
29 PlacementConstraint,
30 PlacementConstraintType,
31 PlacementStrategy,
32 PlacementStrategyType,
33 PropagateTags,
34 RedshiftDataParameters,
35 RetryPolicy,
36 RuleState,
37 RunCommandParameters,
38 RunCommandTarget,
39 SageMakerPipelineParameter,
40 SageMakerPipelineParameters,
41 SqsParameters,
42} from "@distilled.cloud/aws/eventbridge";
43
44export type RuleName = string;
45
46export type RuleArn = `arn:aws:events:${RegionID}:${AccountID}:rule/${string}`;
47
48export interface RuleTarget {
49 /** Unique identifier for this target within the rule. */
50 Id: string;
51 /** ARN of the target resource. */
52 Arn: string;
53 /** ARN of the IAM role to use for this target when the rule is triggered. */
54 RoleArn?: string;
55 /** Valid JSON text passed to the target. Mutually exclusive with InputPath and InputTransformer. */
56 Input?: string;
57 /** JSONPath expression to extract from the event and send to the target. Mutually exclusive with Input and InputTransformer. */
58 InputPath?: string;
59 /** Settings to transform input before sending to the target. Mutually exclusive with Input and InputPath. */
60 InputTransformer?: eventbridge.InputTransformer;
61 /** Settings for a Kinesis Data Stream target. */
62 KinesisParameters?: eventbridge.KinesisParameters;
63 /** Parameters for Systems Manager Run Command targets. */
64 RunCommandParameters?: eventbridge.RunCommandParameters;
65 /** Parameters for ECS task targets. */
66 EcsParameters?: RuleTargetEcsParameters;
67 /** Parameters for AWS Batch job targets. */
68 BatchParameters?: RuleTargetBatchParameters;
69 /** Parameters for SQS queue targets (e.g., FIFO message group ID). */
70 SqsParameters?: eventbridge.SqsParameters;
71 /** Parameters for HTTP endpoint targets (API Gateway, API Destinations). */
72 HttpParameters?: eventbridge.HttpParameters;
73 /** Parameters for Amazon Redshift Data API targets. */
74 RedshiftDataParameters?: eventbridge.RedshiftDataParameters;
75 /** Parameters for SageMaker Pipeline targets. */
76 SageMakerPipelineParameters?: eventbridge.SageMakerPipelineParameters;
77 /** Dead-letter queue configuration for failed event delivery. */
78 DeadLetterConfig?: RuleTargetDeadLetterConfig;
79 /** Retry policy settings for the target. */
80 RetryPolicy?: eventbridge.RetryPolicy;
81 /** Parameters for AWS AppSync GraphQL API targets. */
82 AppSyncParameters?: eventbridge.AppSyncParameters;
83}
84
85/** ECS parameters for a rule target with Input-wrapped ARN fields. */
86export interface RuleTargetEcsParameters extends Omit<
87 eventbridge.EcsParameters,
88 "TaskDefinitionArn"
89> {
90 /** ARN of the ECS task definition to run. */
91 TaskDefinitionArn: string;
92}
93
94/** Batch parameters for a rule target with Input-wrapped ARN fields. */
95export interface RuleTargetBatchParameters extends Omit<
96 eventbridge.BatchParameters,
97 "JobDefinition"
98> {
99 /** ARN or name of the Batch job definition. */
100 JobDefinition: string;
101}
102
103/** Dead-letter config for a rule target with Input-wrapped ARN. */
104export interface RuleTargetDeadLetterConfig {
105 /** ARN of the SQS queue used as the dead-letter queue. */
106 Arn?: string;
107}
108
109export interface RuleProps {
110 /**
111 * Name of the rule. Must match [\.\-_A-Za-z0-9]+, 1-64 characters.
112 * If omitted, a unique name will be generated.
113 */
114 name?: string;
115
116 /**
117 * Description of the rule. Max 512 characters.
118 */
119 description?: string;
120
121 /**
122 * The name or ARN of the event bus to associate with this rule.
123 * If omitted, the default event bus is used.
124 */
125 eventBusName?: string;
126
127 /**
128 * The event pattern that triggers this rule. Specified as a JSON-compatible object.
129 * A rule must contain at least an eventPattern or scheduleExpression.
130 */
131 eventPattern?: Record<string, any>;
132
133 /**
134 * The scheduling expression (e.g. "rate(5 minutes)", "cron(0 20 * * ? *)").
135 * A rule must contain at least an eventPattern or scheduleExpression.
136 */
137 scheduleExpression?: string;
138
139 /**
140 * Whether the rule is enabled or disabled.
141 * @default "ENABLED"
142 */
143 state?: eventbridge.RuleState;
144
145 /**
146 * ARN of the IAM role associated with the rule. Required for targets that need
147 * IAM roles (e.g. Kinesis, Step Functions, ECS, API Gateway).
148 */
149 roleArn?: string;
150
151 /**
152 * The targets to invoke when this rule is triggered. Maximum 5 targets per rule.
153 */
154 targets?: RuleTarget[];
155
156 /**
157 * Tags to assign to the rule.
158 */
159 tags?: Record<string, string>;
160}
161
162/**
163 * An Amazon EventBridge rule that matches events and routes them to targets.
164 * ### Creating Rules
165 * **Example:** Event Pattern Rule
166 * ```typescript
167 * const rule = yield* Rule("S3Events", {
168 * eventPattern: {
169 * source: ["aws.s3"],
170 * "detail-type": ["Object Created"],
171 * },
172 * targets: [{
173 * Id: "MyTarget",
174 * Arn: yield* queue.queueArn,
175 * }],
176 * });
177 * ```
178 *
179 * **Example:** Scheduled Rule
180 * ```typescript
181 * const rule = yield* Rule("EveryFiveMinutes", {
182 * scheduleExpression: "rate(5 minutes)",
183 * targets: [{
184 * Id: "LambdaTarget",
185 * Arn: yield* fn.functionArn(),
186 * }],
187 * });
188 * ```
189 *
190 * ### Targeting
191 * **Example:** Rule with Input Transformer
192 * ```typescript
193 * const rule = yield* Rule("TransformedEvents", {
194 * eventPattern: {
195 * source: ["aws.ec2"],
196 * "detail-type": ["EC2 Instance State-change Notification"],
197 * },
198 * targets: [{
199 * Id: "SqsTarget",
200 * Arn: yield* queue.queueArn,
201 * InputTransformer: {
202 * InputPathsMap: {
203 * instance: "$.detail.instance-id",
204 * state: "$.detail.state",
205 * },
206 * InputTemplate: '{"instanceId": <instance>, "newState": <state>}',
207 * },
208 * }],
209 * });
210 * ```
211 *
212 * **Example:** Rule with Dead Letter Queue
213 * ```typescript
214 * const rule = yield* Rule("ReliableEvents", {
215 * eventPattern: { source: ["my.app"] },
216 * targets: [{
217 * Id: "Target",
218 * Arn: yield* fn.functionArn(),
219 * DeadLetterConfig: {
220 * Arn: yield* dlq.queueArn,
221 * },
222 * RetryPolicy: {
223 * MaximumRetryAttempts: 3,
224 * MaximumEventAgeInSeconds: 3600,
225 * },
226 * }],
227 * });
228 * ```
229 *
230 * **Example:** Rule with ECS Target
231 * ```typescript
232 * const rule = yield* Rule("EcsSchedule", {
233 * scheduleExpression: "rate(1 hour)",
234 * roleArn: yield* role.roleArn(),
235 * targets: [{
236 * Id: "EcsTask",
237 * Arn: yield* cluster.clusterArn(),
238 * RoleArn: yield* ecsRole.roleArn(),
239 * EcsParameters: {
240 * TaskDefinitionArn: yield* taskDef.taskDefinitionArn(),
241 * TaskCount: 1,
242 * LaunchType: "FARGATE",
243 * NetworkConfiguration: {
244 * awsvpcConfiguration: {
245 * Subnets: ["subnet-abc123"],
246 * AssignPublicIp: "ENABLED",
247 * },
248 * },
249 * },
250 * }],
251 * });
252 * ```
253 *
254 * @resource
255 */
256export interface Rule extends Resource<
257 "AWS.EventBridge.Rule",
258 RuleProps,
259 {
260 /** The name of the rule. */
261 ruleName: RuleName;
262 /** The ARN of the rule. */
263 ruleArn: RuleArn;
264 /** The event bus associated with the rule. */
265 eventBusName: string;
266 },
267 never,
268 Providers
269> {}
270export const Rule = Resource<Rule>("AWS.EventBridge.Rule");
271
272export const RuleProvider = () =>
273 Provider.effect(
274 Rule,
275 Effect.gen(function* () {
276 const createRuleName = (id: string, props: { name?: string } = {}) => {
277 if (props.name) {
278 return Effect.succeed(props.name);
279 }
280 return createPhysicalName({
281 id,
282 maxLength: 64,
283 });
284 };
285
286 return {
287 stables: ["ruleName", "ruleArn", "eventBusName"],
288 diff: Effect.fn(function* ({ id, news, olds }) {
289 if (!isResolved(news)) return;
290 const oldName = yield* createRuleName(id, olds);
291 const newName = yield* createRuleName(id, news);
292 if (oldName !== newName) {
293 return { action: "replace" } as const;
294 }
295 const oldBus = (olds.eventBusName as string | undefined) ?? "default";
296 const newBus = (news.eventBusName as string | undefined) ?? "default";
297 if (oldBus !== newBus) {
298 return { action: "replace" } as const;
299 }
300 }),
301 read: Effect.fn(function* ({ id, olds, output }) {
302 const { accountId, region } = yield* AWSEnvironment.current;
303 // The engine guarantees `olds` is fully resolved before calling
304 // `read` (Plan.ts guards every read site with `isResolved`), so
305 // props can be used directly to derive identity.
306 const ruleName =
307 output?.ruleName ??
308 (yield* createRuleName(id, { name: olds?.name }));
309 const eventBusName =
310 output?.eventBusName ?? olds?.eventBusName ?? "default";
311 const described = yield* eventbridge
312 .describeRule({
313 Name: ruleName,
314 EventBusName:
315 eventBusName !== "default" ? eventBusName : undefined,
316 })
317 .pipe(
318 Effect.catchTag("ResourceNotFoundException", () =>
319 Effect.succeed(undefined),
320 ),
321 );
322
323 if (!described?.Name) {
324 return undefined;
325 }
326
327 const resolvedEventBusName = described.EventBusName ?? eventBusName;
328 const ruleArn = toRuleArn(
329 region,
330 accountId,
331 resolvedEventBusName,
332 described.Name,
333 );
334 const { Tags } = yield* eventbridge.listTagsForResource({
335 ResourceARN: described.Arn ?? ruleArn,
336 });
337 const attrs = {
338 ruleName: described.Name,
339 ruleArn,
340 eventBusName: resolvedEventBusName,
341 };
342 return (yield* hasAlchemyTags(id, Tags ?? []))
343 ? attrs
344 : Unowned(attrs);
345 }),
346 list: () =>
347 Effect.gen(function* () {
348 const { accountId, region } = yield* AWSEnvironment.current;
349 // A Rule belongs to an event bus and `listRules` is scoped to one
350 // bus (defaulting to "default"). To enumerate every rule in the
351 // account/region we first enumerate all event buses (manual
352 // NextToken pagination — neither op is a paginated distilled op),
353 // then list rules per bus with bounded concurrency.
354 const busNames: string[] = [];
355 let busToken: string | undefined;
356 do {
357 const page = yield* eventbridge.listEventBuses({
358 NextToken: busToken,
359 });
360 for (const bus of page.EventBuses ?? []) {
361 if (bus.Name) {
362 busNames.push(bus.Name);
363 }
364 }
365 busToken = page.NextToken;
366 } while (busToken);
367 // `listEventBuses` should include the default bus, but guarantee it.
368 if (!busNames.includes("default")) {
369 busNames.push("default");
370 }
371
372 const perBus = yield* Effect.forEach(
373 busNames,
374 (busName) =>
375 Effect.gen(function* () {
376 const eventBusParam =
377 busName !== "default" ? busName : undefined;
378 const attrs: {
379 ruleName: RuleName;
380 ruleArn: RuleArn;
381 eventBusName: string;
382 }[] = [];
383 let ruleToken: string | undefined;
384 do {
385 // A peer reconciler may delete an event bus between our
386 // `listEventBuses` snapshot and this `listRules` call —
387 // treat a vanished bus as contributing zero rules rather
388 // than failing the whole enumeration.
389 const page = yield* eventbridge
390 .listRules({
391 EventBusName: eventBusParam,
392 NextToken: ruleToken,
393 })
394 .pipe(
395 Effect.catchTag("ResourceNotFoundException", () =>
396 Effect.succeed(
397 undefined as
398 | eventbridge.ListRulesResponse
399 | undefined,
400 ),
401 ),
402 );
403 if (!page) {
404 break;
405 }
406 for (const rule of page.Rules ?? []) {
407 if (!rule.Name) {
408 continue;
409 }
410 // Rules created and owned by another AWS service (e.g.
411 // B2BI's DO-NOT-DELETE-* or DevOpsGuru's managed rules)
412 // reject DeleteRule without Force and are recreated by
413 // the owning service anyway — skip them in enumeration
414 // for account-wide teardown (nuke).
415 if (rule.ManagedBy) {
416 continue;
417 }
418 const resolvedBus = rule.EventBusName ?? busName;
419 attrs.push({
420 ruleName: rule.Name,
421 ruleArn:
422 (rule.Arn as RuleArn | undefined) ??
423 toRuleArn(region, accountId, resolvedBus, rule.Name),
424 eventBusName: resolvedBus,
425 });
426 }
427 ruleToken = page.NextToken;
428 } while (ruleToken);
429 return attrs;
430 }),
431 { concurrency: 5 },
432 );
433 return perBus.flat();
434 }),
435 reconcile: Effect.fn(function* ({ id, news = {}, output, session }) {
436 const { accountId, region } = yield* AWSEnvironment.current;
437 yield* validateRuleProps(news);
438 const ruleName =
439 output?.ruleName ?? (yield* createRuleName(id, news));
440 const eventBusName =
441 output?.eventBusName ??
442 (news.eventBusName as string | undefined) ??
443 "default";
444 const eventBusParam =
445 eventBusName !== "default" ? eventBusName : undefined;
446 const internalTags = yield* createInternalTags(id);
447 const desiredTags = {
448 ...internalTags,
449 ...(news.tags as Record<string, string> | undefined),
450 };
451
452 // Ensure + Sync rule definition — `putRule` is the single
453 // create-or-update API for the rule itself. It's idempotent on
454 // matching params and overwrites schedule/eventPattern/state/etc.
455 // on differences, so we call it unconditionally. `Tags` only
456 // applies when creating; tags are reconciled separately below
457 // against observed cloud tags.
458 const { RuleArn } = yield* eventbridge.putRule({
459 Name: ruleName,
460 Description: news.description,
461 EventBusName: eventBusParam,
462 EventPattern: news.eventPattern
463 ? JSON.stringify(news.eventPattern)
464 : undefined,
465 ScheduleExpression: news.scheduleExpression,
466 State: news.state ?? "ENABLED",
467 RoleArn: news.roleArn as string | undefined,
468 Tags: createTagsList(desiredTags),
469 });
470 const ruleArn =
471 (RuleArn as RuleArn | undefined) ??
472 toRuleArn(region, accountId, eventBusName, ruleName);
473
474 // Sync targets — observed cloud targets vs desired. `listTargetsByRule`
475 // gives us the live target ids; we remove anything no longer desired,
476 // and `putTargets` overwrites/upserts the rest.
477 const resolvedTargets =
478 (news.targets as Input.Resolve<RuleTarget>[] | undefined) ?? [];
479 const desiredTargetIds = new Set(resolvedTargets.map((t) => t.Id));
480 const observedTargets = yield* eventbridge
481 .listTargetsByRule({
482 Rule: ruleName,
483 EventBusName: eventBusParam,
484 })
485 .pipe(
486 Effect.map((r) => r.Targets ?? []),
487 Effect.catchTag("ResourceNotFoundException", () =>
488 Effect.succeed([] as eventbridge.Target[]),
489 ),
490 );
491 const removedTargetIds = observedTargets
492 .map((t) => t.Id)
493 .filter(
494 (tid): tid is string => !!tid && !desiredTargetIds.has(tid),
495 );
496
497 if (removedTargetIds.length > 0) {
498 const response = yield* eventbridge
499 .removeTargets({
500 Rule: ruleName,
501 EventBusName: eventBusParam,
502 Ids: removedTargetIds,
503 })
504 .pipe(
505 Effect.catchTag("ResourceNotFoundException", () =>
506 Effect.succeed(undefined),
507 ),
508 );
509
510 if (response) {
511 yield* assertRemoveTargetsSucceeded(response);
512 }
513 }
514
515 if (resolvedTargets.length > 0) {
516 const response = yield* eventbridge.putTargets({
517 Rule: ruleName,
518 EventBusName: eventBusParam,
519 Targets: resolvedTargets.map(toTarget),
520 });
521 yield* assertPutTargetsSucceeded(response);
522 }
523
524 // Sync tags — diff observed cloud tags against desired. Adoption
525 // and partial prior runs both converge here.
526 const observedTagsList = yield* eventbridge
527 .listTagsForResource({
528 ResourceARN: ruleArn,
529 })
530 .pipe(Effect.map((r) => r.Tags ?? []));
531 const observedTags: Record<string, string> = {};
532 for (const tag of observedTagsList) {
533 if (tag.Key && tag.Value !== undefined) {
534 observedTags[tag.Key] = tag.Value;
535 }
536 }
537 const { removed, upsert } = diffTags(observedTags, desiredTags);
538
539 if (removed.length > 0) {
540 yield* eventbridge.untagResource({
541 ResourceARN: ruleArn,
542 TagKeys: removed,
543 });
544 }
545
546 if (upsert.length > 0) {
547 yield* eventbridge.tagResource({
548 ResourceARN: ruleArn,
549 Tags: upsert,
550 });
551 }
552
553 yield* session.note(ruleArn);
554 return {
555 ruleName,
556 ruleArn,
557 eventBusName,
558 };
559 }),
560 delete: Effect.fn(function* (input) {
561 const ruleName = input.output.ruleName;
562 const eventBusName = input.output.eventBusName;
563 const eventBusParam =
564 eventBusName !== "default" ? eventBusName : undefined;
565
566 const { Targets } = yield* eventbridge
567 .listTargetsByRule({
568 Rule: ruleName,
569 EventBusName: eventBusParam,
570 })
571 .pipe(
572 Effect.catchTag("ResourceNotFoundException", () =>
573 Effect.succeed({ Targets: undefined }),
574 ),
575 );
576
577 if (Targets && Targets.length > 0) {
578 const response = yield* eventbridge
579 .removeTargets({
580 Rule: ruleName,
581 EventBusName: eventBusParam,
582 Ids: Targets.map((t) => t.Id),
583 })
584 .pipe(
585 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
586 );
587 if (response) {
588 yield* assertRemoveTargetsSucceeded(response);
589 }
590 }
591
592 yield* eventbridge
593 .deleteRule({
594 Name: ruleName,
595 EventBusName: eventBusParam,
596 })
597 .pipe(
598 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
599 );
600 }),
601 };
602 }),
603 );
604
605const toTarget = (target: Input.Resolve<RuleTarget>): eventbridge.Target => ({
606 Id: target.Id,
607 Arn: target.Arn,
608 RoleArn: target.RoleArn,
609 Input: target.Input,
610 InputPath: target.InputPath,
611 InputTransformer: target.InputTransformer,
612 KinesisParameters: target.KinesisParameters,
613 RunCommandParameters: target.RunCommandParameters,
614 EcsParameters: target.EcsParameters
615 ? {
616 ...target.EcsParameters,
617 TaskDefinitionArn: target.EcsParameters.TaskDefinitionArn,
618 }
619 : undefined,
620 BatchParameters: target.BatchParameters
621 ? {
622 ...target.BatchParameters,
623 JobDefinition: target.BatchParameters.JobDefinition,
624 }
625 : undefined,
626 SqsParameters: target.SqsParameters,
627 HttpParameters: target.HttpParameters,
628 RedshiftDataParameters: target.RedshiftDataParameters,
629 SageMakerPipelineParameters: target.SageMakerPipelineParameters,
630 DeadLetterConfig: target.DeadLetterConfig
631 ? { Arn: target.DeadLetterConfig.Arn }
632 : undefined,
633 RetryPolicy: target.RetryPolicy,
634 AppSyncParameters: target.AppSyncParameters,
635});
636
637const toRuleArn = (
638 region: RegionID,
639 accountId: AccountID,
640 eventBusName: string,
641 ruleName: string,
642): RuleArn =>
643 (eventBusName === "default"
644 ? `arn:aws:events:${region}:${accountId}:rule/${ruleName}`
645 : `arn:aws:events:${region}:${accountId}:rule/${eventBusName}/${ruleName}`) as RuleArn;
646
647const validateRuleProps = Effect.fn(function* (props: RuleProps) {
648 if (!props.eventPattern && !props.scheduleExpression) {
649 return yield* Effect.fail(
650 new Error(
651 "EventBridge Rule requires either `eventPattern` or `scheduleExpression`",
652 ),
653 );
654 }
655});
656
657const assertPutTargetsSucceeded = Effect.fn(function* (
658 response: eventbridge.PutTargetsResponse,
659) {
660 if ((response.FailedEntryCount ?? 0) > 0) {
661 return yield* Effect.fail(
662 new Error(
663 `Failed to attach EventBridge targets: ${JSON.stringify(response.FailedEntries ?? [])}`,
664 ),
665 );
666 }
667});
668
669const assertRemoveTargetsSucceeded = Effect.fn(function* (
670 response: eventbridge.RemoveTargetsResponse,
671) {
672 if ((response.FailedEntryCount ?? 0) > 0) {
673 return yield* Effect.fail(
674 new Error(
675 `Failed to remove EventBridge targets: ${JSON.stringify(response.FailedEntries ?? [])}`,
676 ),
677 );
678 }
679});