1import * as logs from "@distilled.cloud/aws/cloudwatch-logs";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import type { Providers } from "../Providers.ts";
10
11export interface LogStreamProps {
12 /**
13 * Name of the log group the stream belongs to.
14 * Changing this value replaces the log stream.
15 */
16 logGroupName: string;
17 /**
18 * Name of the log stream. If omitted, a unique name is generated.
19 * Changing this value replaces the log stream.
20 */
21 logStreamName?: string;
22}
23
24export interface LogStream extends Resource<
25 "AWS.Logs.LogStream",
26 LogStreamProps,
27 {
28 logStreamName: string;
29 logGroupName: string;
30 logStreamArn?: string;
31 },
32 never,
33 Providers
34> {}
35
36/**
37 * A CloudWatch Logs log stream — a sequence of log events within a log group.
38 *
39 * Most log streams are created automatically by the emitting service (Lambda
40 * creates its own streams under `/aws/lambda/...`); declare one explicitly
41 * only when writing custom log events via `putLogEvents`.
42 * ### Creating Log Streams
43 * **Example:** Custom Audit Stream
44 * ```typescript
45 * const stream = yield* LogStream("AuditStream", {
46 * logGroupName: logGroup.logGroupName,
47 * });
48 * ```
49 *
50 * @resource
51 */
52export const LogStream = Resource<LogStream>("AWS.Logs.LogStream");
53
54export const LogStreamProvider = () =>
55 Provider.effect(
56 LogStream,
57 Effect.gen(function* () {
58 const toLogStreamName = (
59 id: string,
60 props: { logStreamName?: string } = {},
61 ) =>
62 props.logStreamName
63 ? Effect.succeed(props.logStreamName)
64 : createPhysicalName({ id, maxLength: 512 });
65
66 const observe = Effect.fn(function* (
67 logGroupName: string,
68 logStreamName: string,
69 ) {
70 const described = yield* logs
71 .describeLogStreams({
72 logGroupName,
73 logStreamNamePrefix: logStreamName,
74 })
75 .pipe(
76 Effect.catchTag("ResourceNotFoundException", () =>
77 Effect.succeed({ logStreams: [] }),
78 ),
79 );
80 return (described.logStreams ?? []).find(
81 (stream) => stream.logStreamName === logStreamName,
82 );
83 });
84
85 return {
86 stables: ["logStreamName", "logGroupName", "logStreamArn"],
87 // No account-wide stream list — fan out over every log group.
88 list: () =>
89 Effect.gen(function* () {
90 const groups = yield* logs.describeLogGroups.pages({}).pipe(
91 Stream.runCollect,
92 Effect.map((chunk) =>
93 Array.from(chunk)
94 .flatMap((page) => page.logGroups ?? [])
95 .map((group) => group.logGroupName)
96 .filter((name): name is string => name != null),
97 ),
98 );
99 const perGroup = yield* Effect.forEach(
100 groups,
101 (logGroupName) =>
102 logs.describeLogStreams.pages({ logGroupName }).pipe(
103 Stream.runCollect,
104 Effect.map((chunk) =>
105 Array.from(chunk)
106 .flatMap((page) => page.logStreams ?? [])
107 .filter(
108 (
109 stream,
110 ): stream is logs.LogStream & {
111 logStreamName: string;
112 } => stream.logStreamName != null,
113 )
114 .map((stream) => ({
115 logStreamName: stream.logStreamName,
116 logGroupName,
117 logStreamArn: stream.arn,
118 })),
119 ),
120 // group deleted between list and describe — skip
121 Effect.catchTag("ResourceNotFoundException", () =>
122 Effect.succeed([]),
123 ),
124 ),
125 { concurrency: 10 },
126 );
127 return perGroup.flat();
128 }),
129 diff: Effect.fn(function* ({ id, olds, news }) {
130 if (!isResolved(news)) return;
131 if (olds.logGroupName !== news.logGroupName) {
132 return { action: "replace" } as const;
133 }
134 if (
135 (yield* toLogStreamName(id, olds)) !==
136 (yield* toLogStreamName(id, news))
137 ) {
138 return { action: "replace" } as const;
139 }
140 }),
141 read: Effect.fn(function* ({ id, olds, output }) {
142 const logGroupName = output?.logGroupName ?? olds?.logGroupName;
143 if (logGroupName === undefined) return undefined;
144 const logStreamName =
145 output?.logStreamName ?? (yield* toLogStreamName(id, olds ?? {}));
146 const observed = yield* observe(logGroupName, logStreamName);
147 if (!observed) return undefined;
148 return {
149 logStreamName,
150 logGroupName,
151 logStreamArn: observed.arn,
152 };
153 }),
154 // Existence-only resource: observe → if missing, create. Nothing on a
155 // log stream is mutable.
156 reconcile: Effect.fn(function* ({ id, news, output, session }) {
157 const logGroupName = news.logGroupName;
158 const logStreamName =
159 output?.logStreamName ?? (yield* toLogStreamName(id, news));
160
161 let observed = yield* observe(logGroupName, logStreamName);
162 if (!observed) {
163 yield* logs
164 .createLogStream({ logGroupName, logStreamName })
165 .pipe(
166 Effect.catchTag(
167 "ResourceAlreadyExistsException",
168 () => Effect.void,
169 ),
170 );
171 observed = yield* observe(logGroupName, logStreamName);
172 }
173
174 yield* session.note(`${logGroupName}:${logStreamName}`);
175
176 return {
177 logStreamName,
178 logGroupName,
179 logStreamArn: observed?.arn,
180 };
181 }),
182 delete: Effect.fn(function* ({ output }) {
183 yield* logs
184 .deleteLogStream({
185 logGroupName: output.logGroupName,
186 logStreamName: output.logStreamName,
187 })
188 .pipe(
189 Effect.retry({
190 while: (error) =>
191 error._tag === "OperationAbortedException" ||
192 error._tag === "ServiceUnavailableException",
193 schedule: Schedule.exponential(100),
194 times: 8,
195 }),
196 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
197 );
198
199 const remaining = yield* Effect.repeat(
200 observe(output.logGroupName, output.logStreamName),
201 {
202 schedule: Schedule.fixed("250 millis"),
203 until: (stream) => stream === undefined,
204 times: 20,
205 },
206 );
207 if (remaining !== undefined) {
208 yield* Effect.die(
209 new Error(
210 `CloudWatch log stream ${output.logGroupName}:${output.logStreamName} remained observable after delete`,
211 ),
212 );
213 }
214 }),
215 };
216 }),
217 );