1import * as cloudfront from "@distilled.cloud/aws/cloudfront";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { isResolved } from "../../Diff.ts";
5import { createPhysicalName } from "../../PhysicalName.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import type { Providers } from "../Providers.ts";
9
10export interface RealtimeLogEndpoint {
11 /**
12 * ARN of the Kinesis data stream that receives the real-time log records.
13 */
14 streamArn: string;
15 /**
16 * ARN of an IAM role that CloudFront can assume to write to the stream.
17 * The role's trust policy must allow the `cloudfront.amazonaws.com`
18 * service principal, and it needs `kinesis:DescribeStreamSummary`,
19 * `kinesis:DescribeStream`, `kinesis:PutRecord` and `kinesis:PutRecords`
20 * on the stream.
21 */
22 roleArn: string;
23}
24
25export interface RealtimeLogConfigProps {
26 /**
27 * Name of the real-time log configuration. If omitted, a deterministic
28 * name is generated. Changing this forces replacement.
29 */
30 name?: string;
31 /**
32 * Percentage of viewer requests sampled into the log stream, between 1 and
33 * 100. Updated in place.
34 */
35 samplingRate: number;
36 /**
37 * Real-time log record fields to include (e.g. `timestamp`, `c-ip`,
38 * `cs-uri-stem`, `sc-status`). Updated in place.
39 */
40 fields: string[];
41 /**
42 * Kinesis endpoints that receive the log records. Updated in place.
43 */
44 endpoints: RealtimeLogEndpoint[];
45}
46
47export interface RealtimeLogConfig extends Resource<
48 "AWS.CloudFront.RealtimeLogConfig",
49 RealtimeLogConfigProps,
50 {
51 /**
52 * ARN of the real-time log configuration.
53 */
54 arn: string;
55 /**
56 * Name of the configuration.
57 */
58 name: string;
59 /**
60 * Current sampling rate.
61 */
62 samplingRate: number;
63 /**
64 * Current log record fields.
65 */
66 fields: string[];
67 /**
68 * Current Kinesis endpoints.
69 */
70 endpoints: RealtimeLogEndpoint[];
71 },
72 never,
73 Providers
74> {}
75
76/**
77 * A CloudFront real-time log configuration.
78 *
79 * Real-time logs deliver per-request records to a Kinesis data stream within
80 * seconds. Attach the configuration to a distribution's cache behavior via
81 * `Distribution` (`realtimeLogConfigArn`).
82 * ### Creating Real-Time Log Configs
83 * **Example:** Stream Viewer Requests to Kinesis
84 * ```typescript
85 * const stream = yield* Kinesis.Stream("EdgeLogs", {});
86 *
87 * const role = yield* IAM.Role("EdgeLogsRole", {
88 * assumeRolePolicyDocument: {
89 * Version: "2012-10-17",
90 * Statement: [
91 * {
92 * Effect: "Allow",
93 * Principal: { Service: "cloudfront.amazonaws.com" },
94 * Action: "sts:AssumeRole",
95 * },
96 * ],
97 * },
98 * inlinePolicies: {
99 * kinesis: {
100 * Version: "2012-10-17",
101 * Statement: [
102 * {
103 * Effect: "Allow",
104 * Action: [
105 * "kinesis:DescribeStreamSummary",
106 * "kinesis:DescribeStream",
107 * "kinesis:PutRecord",
108 * "kinesis:PutRecords",
109 * ],
110 * Resource: stream.streamArn,
111 * },
112 * ],
113 * },
114 * },
115 * });
116 *
117 * const logConfig = yield* RealtimeLogConfig("EdgeLogConfig", {
118 * samplingRate: 100,
119 * fields: ["timestamp", "c-ip", "cs-uri-stem", "sc-status"],
120 * endpoints: [{ streamArn: stream.streamArn, roleArn: role.roleArn }],
121 * });
122 * ```
123 *
124 * @resource
125 */
126export const RealtimeLogConfig = Resource<RealtimeLogConfig>(
127 "AWS.CloudFront.RealtimeLogConfig",
128);
129
130const createName = (id: string, props: RealtimeLogConfigProps) =>
131 props.name
132 ? Effect.succeed(props.name)
133 : createPhysicalName({ id, maxLength: 64 });
134
135const toEndPoints = (endpoints: RealtimeLogEndpoint[]): cloudfront.EndPoint[] =>
136 endpoints.map((endpoint) => ({
137 StreamType: "Kinesis",
138 KinesisStreamConfig: {
139 RoleARN: endpoint.roleArn,
140 StreamARN: endpoint.streamArn,
141 },
142 }));
143
144const fromEndPoints = (
145 endpoints: cloudfront.EndPoint[] | undefined,
146): RealtimeLogEndpoint[] =>
147 (endpoints ?? []).flatMap((endpoint) =>
148 endpoint.KinesisStreamConfig
149 ? [
150 {
151 streamArn: endpoint.KinesisStreamConfig.StreamARN,
152 roleArn: endpoint.KinesisStreamConfig.RoleARN,
153 },
154 ]
155 : [],
156 );
157
158const toAttrs = (
159 config: cloudfront.RealtimeLogConfig,
160): RealtimeLogConfig["Attributes"] => ({
161 arn: config.ARN,
162 name: config.Name,
163 samplingRate: config.SamplingRate,
164 fields: [...config.Fields],
165 endpoints: fromEndPoints([...config.EndPoints]),
166});
167
168const sameEndpoints = (a: RealtimeLogEndpoint[], b: RealtimeLogEndpoint[]) =>
169 a.length === b.length &&
170 a.every(
171 (endpoint, index) =>
172 endpoint.streamArn === b[index]?.streamArn &&
173 endpoint.roleArn === b[index]?.roleArn,
174 );
175
176// CloudFront canonicalizes (reorders) the field list, so compare as sets.
177const sameFields = (a: readonly string[], b: readonly string[]) => {
178 if (a.length !== b.length) return false;
179 const sortedA = [...a].sort();
180 const sortedB = [...b].sort();
181 return sortedA.every((field, index) => field === sortedB[index]);
182};
183
184/**
185 * CloudFront validates at create/update time that it can assume the
186 * endpoint's IAM role. A role created in the same deploy can take a few
187 * seconds to propagate through IAM, surfacing as `InvalidArgument` — retry
188 * it on a bounded schedule (~40s). Genuine validation errors fail after the
189 * cap.
190 *
191 * Explicitly typed: inlining `Effect.retry` with options in provider
192 * lifecycle code can widen the provider layer to `unknown` in declaration
193 * emit.
194 *
195 * @internal
196 */
197const retryIamPropagation = <A, E extends { _tag: string }, R>(
198 self: Effect.Effect<A, E, R>,
199): Effect.Effect<A, E, R> =>
200 Effect.retry(self, {
201 while: (e) => e._tag === "InvalidArgument",
202 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(20)]),
203 });
204
205/**
206 * Deleting a config shortly after a distribution stopped referencing it can
207 * surface `RealtimeLogConfigInUse` until the distribution update propagates
208 * — retry on a bounded schedule (~40s).
209 *
210 * @internal
211 */
212const retryConfigInUse = <A, E extends { _tag: string }, R>(
213 self: Effect.Effect<A, E, R>,
214): Effect.Effect<A, E, R> =>
215 Effect.retry(self, {
216 while: (e) => e._tag === "RealtimeLogConfigInUse",
217 schedule: Schedule.max([Schedule.fixed("2 seconds"), Schedule.recurs(20)]),
218 });
219
220export const RealtimeLogConfigProvider = () =>
221 Provider.effect(
222 RealtimeLogConfig,
223 Effect.gen(function* () {
224 const observe = Effect.fn(function* (name: string) {
225 const response = yield* cloudfront
226 .getRealtimeLogConfig({ Name: name })
227 .pipe(
228 Effect.catchTag("NoSuchRealtimeLogConfig", () =>
229 Effect.succeed(undefined),
230 ),
231 );
232 return response?.RealtimeLogConfig;
233 });
234
235 return {
236 stables: ["arn", "name"],
237 // Enumerate every config in the account. `listRealtimeLogConfigs`
238 // pages via Marker/NextMarker inside a wrapper struct (no distilled
239 // paginator) — hand-roll a bounded loop; an empty NextMarker is
240 // terminal.
241 list: () =>
242 Effect.gen(function* () {
243 const items: RealtimeLogConfig["Attributes"][] = [];
244 let marker: string | undefined;
245 for (let page = 0; page < 20; page++) {
246 const response = yield* cloudfront.listRealtimeLogConfigs({
247 Marker: marker,
248 MaxItems: 100,
249 });
250 for (const config of response.RealtimeLogConfigs?.Items ?? []) {
251 items.push(toAttrs(config));
252 }
253 if (!response.RealtimeLogConfigs?.NextMarker) {
254 break;
255 }
256 marker = response.RealtimeLogConfigs.NextMarker;
257 }
258 return items;
259 }),
260 diff: Effect.fn(function* ({ id, olds, news }) {
261 if (!isResolved(news)) return undefined;
262 if ((yield* createName(id, olds)) !== (yield* createName(id, news))) {
263 return { action: "replace" } as const;
264 }
265 }),
266 read: Effect.fn(function* ({ id, olds, output }) {
267 const name =
268 output?.name ??
269 (yield* createName(id, olds ?? ({} as RealtimeLogConfigProps)));
270 const observed = yield* observe(name);
271 if (!observed) {
272 return undefined;
273 }
274 return toAttrs(observed);
275 }),
276 reconcile: Effect.fn(function* ({ id, news, output, session }) {
277 const name = output?.name ?? (yield* createName(id, news));
278 const desiredEndpoints = news.endpoints;
279 const desiredFields = news.fields;
280
281 // Observe — live config is authoritative; `output` only caches the
282 // name.
283 let observed = yield* observe(name);
284
285 // Ensure — create when missing; `RealtimeLogConfigAlreadyExists`
286 // is a race with a peer reconciler, re-observe.
287 if (!observed) {
288 observed = yield* retryIamPropagation(
289 cloudfront.createRealtimeLogConfig({
290 Name: name,
291 SamplingRate: news.samplingRate,
292 Fields: desiredFields,
293 EndPoints: toEndPoints(desiredEndpoints),
294 }),
295 ).pipe(
296 Effect.map((response) => response.RealtimeLogConfig),
297 Effect.catchTag("RealtimeLogConfigAlreadyExists", () =>
298 observe(name),
299 ),
300 );
301 if (!observed) {
302 return yield* Effect.die(
303 `CloudFront RealtimeLogConfig '${name}' could not be observed after create`,
304 );
305 }
306 yield* session.note(observed.ARN);
307 return toAttrs(observed);
308 }
309
310 // Sync — `updateRealtimeLogConfig` replaces every mutable aspect
311 // at once; skip the call entirely when observed matches desired.
312 const current = toAttrs(observed);
313 if (
314 current.samplingRate !== news.samplingRate ||
315 !sameFields(current.fields, desiredFields) ||
316 !sameEndpoints(current.endpoints, desiredEndpoints)
317 ) {
318 const updated = yield* retryIamPropagation(
319 cloudfront.updateRealtimeLogConfig({
320 Name: name,
321 SamplingRate: news.samplingRate,
322 Fields: desiredFields,
323 EndPoints: toEndPoints(desiredEndpoints),
324 }),
325 );
326 if (updated.RealtimeLogConfig) {
327 observed = updated.RealtimeLogConfig;
328 }
329 }
330
331 yield* session.note(observed.ARN);
332 return toAttrs(observed);
333 }),
334 delete: Effect.fn(function* ({ output }) {
335 yield* retryConfigInUse(
336 cloudfront
337 .deleteRealtimeLogConfig({ Name: output.name })
338 .pipe(Effect.asVoid),
339 ).pipe(Effect.catchTag("NoSuchRealtimeLogConfig", () => Effect.void));
340 }),
341 };
342 }),
343 );