1import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import type { PolicyDocument } from "../IAM/Policy.ts";
11import type { Providers } from "../Providers.ts";
12
13export interface DestinationProps {
14 /**
15 * Name of the destination. The destination name is its identity within the
16 * account and region (put semantics upsert by name). If omitted, a unique
17 * name is generated. Changing this value replaces the destination.
18 */
19 destinationName?: string;
20 /**
21 * ARN of the physical target that receives the log events — a Kinesis
22 * stream in this account.
23 */
24 targetArn: string;
25 /**
26 * ARN of an IAM role that CloudWatch Logs assumes to write to the target.
27 * The role must trust `logs.amazonaws.com`.
28 */
29 roleArn: string;
30 /**
31 * Access policy governing which accounts may create subscription filters
32 * against this destination, either as a JSON string or a structured
33 * document. Required before another account can subscribe.
34 */
35 accessPolicy?: PolicyDocument | string;
36}
37
38export interface Destination extends Resource<
39 "AWS.Logs.Destination",
40 DestinationProps,
41 {
42 destinationName: string;
43 destinationArn: string;
44 targetArn: string;
45 roleArn: string;
46 accessPolicy?: string;
47 },
48 never,
49 Providers
50> {}
51
52/**
53 * A CloudWatch Logs destination — a cross-account subscription target that
54 * forwards log events to a Kinesis stream. Producers in other accounts create
55 * subscription filters whose `destinationArn` points at this destination;
56 * the `accessPolicy` controls which accounts may subscribe.
57 * ### Cross-Account Log Fan-Out
58 * **Example:** Kinesis-Backed Destination
59 * ```typescript
60 * const destination = yield* Destination("CentralLogs", {
61 * targetArn: stream.streamArn,
62 * roleArn: role.roleArn,
63 * accessPolicy: {
64 * Version: "2012-10-17",
65 * Statement: [
66 * {
67 * Effect: "Allow",
68 * Principal: { AWS: "123456789012" },
69 * Action: ["logs:PutSubscriptionFilter"],
70 * Resource: "*",
71 * },
72 * ],
73 * },
74 * });
75 * ```
76 *
77 * @resource
78 */
79export const Destination = Resource<Destination>("AWS.Logs.Destination");
80
81/**
82 * `putDestination` validates that CloudWatch Logs can assume the delivery role
83 * at call time; when the role was created in the same deploy this fails with
84 * `InvalidParameterException` until IAM propagates. Bounded retry.
85 *
86 * NOTE: explicit return annotation is load-bearing — an inlined `Effect.retry`
87 * in provider lifecycle code widens the provider layer during declaration emit
88 * (see PATTERNS.md §7).
89 */
90const retryThroughRolePropagation = <A, E extends { _tag: string }, R>(
91 self: Effect.Effect<A, E, R>,
92): Effect.Effect<A, E, R> =>
93 Effect.retry(self, {
94 while: (e) =>
95 e._tag === "InvalidParameterException" ||
96 e._tag === "OperationAbortedException",
97 schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(9)]),
98 });
99
100export const DestinationProvider = () =>
101 Provider.effect(
102 Destination,
103 Effect.gen(function* () {
104 const toDestinationName = (
105 id: string,
106 props: { destinationName?: string } = {},
107 ) =>
108 props.destinationName
109 ? Effect.succeed(props.destinationName)
110 : createPhysicalName({ id, maxLength: 512 });
111
112 const toPolicyString = (policy: PolicyDocument | string | undefined) =>
113 policy === undefined
114 ? undefined
115 : typeof policy === "string"
116 ? policy
117 : JSON.stringify(policy);
118
119 const observe = Effect.fn(function* (destinationName: string) {
120 return yield* logs.describeDestinations
121 .items({ DestinationNamePrefix: destinationName })
122 .pipe(
123 Stream.filter(
124 (destination) => destination.destinationName === destinationName,
125 ),
126 Stream.runHead,
127 Effect.map(Option.getOrUndefined),
128 );
129 });
130
131 return {
132 stables: ["destinationName", "destinationArn"],
133 list: () =>
134 logs.describeDestinations.pages({}).pipe(
135 Stream.runCollect,
136 Effect.map((chunk) =>
137 Array.from(chunk)
138 .flatMap((page) => page.destinations ?? [])
139 .filter(
140 (
141 destination,
142 ): destination is logs.Destination & {
143 destinationName: string;
144 arn: string;
145 } =>
146 destination.destinationName != null &&
147 destination.arn != null,
148 )
149 .map((destination) => ({
150 destinationName: destination.destinationName,
151 destinationArn: destination.arn,
152 targetArn: destination.targetArn ?? "",
153 roleArn: destination.roleArn ?? "",
154 accessPolicy: destination.accessPolicy,
155 })),
156 ),
157 ),
158 diff: Effect.fn(function* ({ id, olds, news }) {
159 if (!isResolved(news)) return;
160 if (
161 (yield* toDestinationName(id, olds)) !==
162 (yield* toDestinationName(id, news))
163 ) {
164 return { action: "replace" } as const;
165 }
166 }),
167 read: Effect.fn(function* ({ id, olds, output }) {
168 const destinationName =
169 output?.destinationName ??
170 (yield* toDestinationName(id, olds ?? {}));
171 const observed = yield* observe(destinationName);
172 if (!observed?.arn) return undefined;
173 return {
174 destinationName,
175 destinationArn: observed.arn,
176 targetArn: observed.targetArn ?? "",
177 roleArn: observed.roleArn ?? "",
178 accessPolicy: observed.accessPolicy,
179 };
180 }),
181 reconcile: Effect.fn(function* ({ id, news, output, session }) {
182 const destinationName =
183 output?.destinationName ?? (yield* toDestinationName(id, news));
184 const desiredPolicy = toPolicyString(news.accessPolicy);
185
186 // Observe — putDestination upserts by name; skip when target/role
187 // already match.
188 let observed = yield* observe(destinationName);
189 if (
190 observed?.targetArn !== news.targetArn ||
191 observed?.roleArn !== news.roleArn
192 ) {
193 const put = yield* retryThroughRolePropagation(
194 logs.putDestination({
195 destinationName,
196 targetArn: news.targetArn,
197 roleArn: news.roleArn,
198 }),
199 );
200 observed = put.destination ?? (yield* observe(destinationName));
201 }
202
203 // Sync access policy against the observed policy.
204 if (
205 desiredPolicy !== undefined &&
206 observed?.accessPolicy !== desiredPolicy
207 ) {
208 yield* retryThroughRolePropagation(
209 logs.putDestinationPolicy({
210 destinationName,
211 accessPolicy: desiredPolicy,
212 }),
213 );
214 }
215
216 yield* session.note(destinationName);
217
218 return {
219 destinationName,
220 destinationArn: observed?.arn ?? "",
221 targetArn: news.targetArn,
222 roleArn: news.roleArn,
223 accessPolicy: desiredPolicy,
224 };
225 }),
226 delete: Effect.fn(function* ({ output }) {
227 yield* logs
228 .deleteDestination({ destinationName: output.destinationName })
229 .pipe(
230 Effect.retry({
231 while: (error) =>
232 error._tag === "OperationAbortedException" ||
233 error._tag === "ServiceUnavailableException",
234 schedule: Schedule.exponential(100),
235 times: 8,
236 }),
237 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
238 );
239 }),
240 };
241 }),
242 );