1import * as Lambda from "@distilled.cloud/aws/lambda";
2import type * as Duration from "effect/Duration";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { deepEqual } from "../../Diff.ts";
6import { toWireSeconds } from "../../Util/Duration.ts";
7
8const DEFAULT_MAXIMUM_RETRY_ATTEMPTS = 2;
9const DEFAULT_MAXIMUM_EVENT_AGE_SECONDS = 21_600;
10
11/**
12 * Asynchronous invocation settings for a Lambda function or alias.
13 *
14 * Configured via {@link FunctionProps.eventInvokeConfig} for the unqualified
15 * function, or {@link AliasProps.eventInvokeConfig} for a specific alias.
16 */
17export interface EventInvokeConfig {
18 /**
19 * Maximum number of times Lambda retries an asynchronous invocation.
20 * @default 2
21 */
22 maximumRetryAttempts?: number;
23 /**
24 * Maximum age that Lambda retains an asynchronous event (e.g. `"6 hours"`).
25 * Rounded to whole seconds on the wire.
26 * @default "6 hours"
27 */
28 maximumEventAge?: Duration.Input;
29 /**
30 * Destinations for successful or failed asynchronous invocation records.
31 */
32 destinationConfig?: Lambda.DestinationConfig;
33}
34
35const retryOnConflict = <A, E extends { _tag: string }, R>(
36 effect: Effect.Effect<A, E, R>,
37) =>
38 effect.pipe(
39 Effect.retry({
40 while: (e) => e._tag === "ResourceConflictException",
41 schedule: Schedule.max([Schedule.exponential(500), Schedule.recurs(10)]),
42 }),
43 );
44
45const normalizeDestinationConfig = (
46 config: Lambda.DestinationConfig | undefined,
47): Lambda.DestinationConfig | undefined => {
48 const onSuccess = config?.OnSuccess?.Destination
49 ? { Destination: config.OnSuccess.Destination }
50 : undefined;
51 const onFailure = config?.OnFailure?.Destination
52 ? { Destination: config.OnFailure.Destination }
53 : undefined;
54 return onSuccess || onFailure
55 ? {
56 OnSuccess: onSuccess,
57 OnFailure: onFailure,
58 }
59 : undefined;
60};
61
62const observeConfig = (functionName: string, qualifier: string | undefined) =>
63 Lambda.getFunctionEventInvokeConfig({
64 FunctionName: functionName,
65 Qualifier: qualifier,
66 }).pipe(
67 Effect.catchTag("ResourceNotFoundException", () =>
68 Effect.succeed(undefined),
69 ),
70 );
71
72/**
73 * Converge the async invocation config of a function or alias to the desired
74 * state: put it when set, delete it when omitted. Diffs against the observed
75 * cloud config so a no-op deploy skips the write entirely.
76 */
77export const syncEventInvokeConfig = Effect.fn(function* ({
78 functionName,
79 qualifier,
80 config,
81}: {
82 functionName: string;
83 /** Alias name (or version) to scope the config to. Omit for `$LATEST`. */
84 qualifier?: string;
85 config: EventInvokeConfig | undefined;
86}) {
87 const observed = yield* observeConfig(functionName, qualifier);
88
89 if (config === undefined) {
90 if (observed) {
91 yield* retryOnConflict(
92 Lambda.deleteFunctionEventInvokeConfig({
93 FunctionName: functionName,
94 Qualifier: qualifier,
95 }),
96 ).pipe(Effect.catchTag("ResourceNotFoundException", () => Effect.void));
97 }
98 return;
99 }
100
101 const desired = {
102 maximumRetryAttempts:
103 config.maximumRetryAttempts ?? DEFAULT_MAXIMUM_RETRY_ATTEMPTS,
104 maximumEventAgeSeconds:
105 toWireSeconds(config.maximumEventAge) ??
106 DEFAULT_MAXIMUM_EVENT_AGE_SECONDS,
107 destinationConfig: normalizeDestinationConfig(config.destinationConfig),
108 };
109
110 if (
111 observed &&
112 (observed.MaximumRetryAttempts ?? DEFAULT_MAXIMUM_RETRY_ATTEMPTS) ===
113 desired.maximumRetryAttempts &&
114 (observed.MaximumEventAgeInSeconds ?? DEFAULT_MAXIMUM_EVENT_AGE_SECONDS) ===
115 desired.maximumEventAgeSeconds &&
116 deepEqual(
117 normalizeDestinationConfig(observed.DestinationConfig),
118 desired.destinationConfig,
119 )
120 ) {
121 return;
122 }
123
124 yield* Lambda.putFunctionEventInvokeConfig({
125 FunctionName: functionName,
126 Qualifier: qualifier,
127 MaximumRetryAttempts: desired.maximumRetryAttempts,
128 MaximumEventAgeInSeconds: desired.maximumEventAgeSeconds,
129 DestinationConfig: desired.destinationConfig,
130 }).pipe(
131 Effect.retry({
132 while: (
133 e,
134 ): e is
135 | Lambda.ResourceConflictException
136 | Lambda.InvalidParameterValueException =>
137 e._tag === "ResourceConflictException" ||
138 // Destination validation races IAM policy propagation on the
139 // execution role — Lambda rejects the put until the role can reach
140 // the destination.
141 (e._tag === "InvalidParameterValueException" &&
142 (e.message?.includes(
143 "The function execution role does not have permissions to call",
144 ) ??
145 false)),
146 schedule: Schedule.max([Schedule.exponential(500), Schedule.recurs(10)]),
147 }),
148 );
149});