1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import { gunzipSync } from "node:zlib";
5
6import * as Namespace from "../../Namespace.ts";
7import * as Output from "../../Output.ts";
8import type { LogGroup } from "../Logs/LogGroup.ts";
9import {
10 LogGroupEventSource as LogsLogGroupEventSource,
11 type CloudWatchLogsEvent,
12 type LogEventRecord,
13 type LogGroupEventSourceProps,
14 type LogGroupEventSourceService,
15 type LogsSubscriptionPayload,
16} from "../Logs/LogGroupEventSource.ts";
17import { SubscriptionFilter } from "../Logs/SubscriptionFilter.ts";
18import * as Lambda from "./Function.ts";
19import { Permission as LambdaPermission } from "./Permission.ts";
20
21/**
22 * Narrow an arbitrary Lambda invocation payload to a CloudWatch Logs
23 * subscription event.
24 */
25export const isCloudWatchLogsEvent = (
26 event: any,
27): event is CloudWatchLogsEvent => typeof event?.awslogs?.data === "string";
28
29/**
30 * Decode the gzipped, base64-encoded CloudWatch Logs subscription payload.
31 */
32export const decodeCloudWatchLogsEvent = (
33 event: CloudWatchLogsEvent,
34): Effect.Effect<LogsSubscriptionPayload, Error> =>
35 Effect.try({
36 try: () =>
37 JSON.parse(
38 gunzipSync(Buffer.from(event.awslogs.data, "base64")).toString(),
39 ) as LogsSubscriptionPayload,
40 catch: (cause) =>
41 new Error("failed to decode CloudWatch Logs subscription payload", {
42 cause,
43 }),
44 });
45
46/**
47 * Lambda runtime implementation for `AWS.Logs.consumeLogEvents(...)`.
48 *
49 * This layer does two things:
50 *
51 * 1. At deploy time it creates the backing `AWS.Logs.SubscriptionFilter`
52 * targeting the current Lambda function plus the `lambda:InvokeFunction`
53 * permission for `logs.amazonaws.com`.
54 * 2. At runtime it decodes the gzipped/base64 `awslogs.data` payload of
55 * incoming invocations and forwards each log event into the supplied
56 * handler as a typed `LogEventRecord` stream.
57 * ### Consuming Log Events
58 * **Example:** Forward Another Function's Error Logs
59 * ```typescript
60 * yield* AWS.Logs.consumeLogEvents(
61 * logGroup,
62 * { filterPattern: "?ERROR ?Error" },
63 * (events) =>
64 * Stream.runForEach(events, (event) =>
65 * Effect.log(`${event.logStream}: ${event.message}`),
66 * ),
67 * );
68 * ```
69 *
70 * @binding
71 */
72export const LogGroupEventSource = Layer.effect(
73 LogsLogGroupEventSource,
74 // The impl resolves plan-time services (SubscriptionFilter, Permission)
75 // whereas LogGroupEventSourceService erases the requirement channel to
76 // `never`.
77 // @effect-diagnostics-next-line missingEffectContext:off
78 Effect.gen(function* () {
79 const host = yield* Lambda.Function;
80 const Permission = yield* LambdaPermission;
81 const Filter = yield* SubscriptionFilter;
82
83 return Effect.fn(function* <Req = never>(
84 logGroup: LogGroup,
85 props: LogGroupEventSourceProps,
86 process: (
87 events: Stream.Stream<LogEventRecord>,
88 ) => Effect.Effect<void, never, Req>,
89 ) {
90 // this adds it to the Lambda Function's environment variables
91 const LogGroupName = yield* logGroup.logGroupName;
92
93 // Deploy-time: grant CloudWatch Logs permission to invoke this function
94 // and create the subscription filter targeting it. Skipped once running
95 // inside the deployed Function (the global guard), where the only work
96 // is registering the runtime handler below. Namespaced under the host so
97 // the sub-resources' logical identity is stable per host function.
98 if (!globalThis.__ALCHEMY_RUNTIME__) {
99 yield* Namespace.push(
100 host.LogicalId,
101 Effect.gen(function* () {
102 yield* Permission(
103 `AWS.Logs.InvokePermission(${logGroup.LogicalId})`,
104 {
105 action: "lambda:InvokeFunction",
106 functionName: host.functionName,
107 principal: "logs.amazonaws.com",
108 sourceArn: Output.interpolate`${logGroup.logGroupArn}:*`,
109 },
110 );
111 yield* Filter(
112 `AWS.Logs.SubscriptionFilter(${logGroup.LogicalId})`,
113 {
114 logGroupName: logGroup.logGroupName,
115 filterName: props.filterName,
116 filterPattern: props.filterPattern ?? "",
117 destinationArn: host.functionArn,
118 },
119 );
120 }),
121 );
122 }
123
124 yield* host.listen(
125 Effect.gen(function* () {
126 const logGroupName = yield* LogGroupName;
127 return (event: any) => {
128 if (isCloudWatchLogsEvent(event)) {
129 return Effect.gen(function* () {
130 const payload = yield* decodeCloudWatchLogsEvent(event);
131 if (
132 payload.messageType !== "DATA_MESSAGE" ||
133 payload.logGroup !== logGroupName
134 ) {
135 return;
136 }
137 yield* process(
138 Stream.fromArray(
139 payload.logEvents.map((logEvent): LogEventRecord => ({
140 id: logEvent.id,
141 timestamp: logEvent.timestamp,
142 message: logEvent.message,
143 logGroup: payload.logGroup,
144 logStream: payload.logStream,
145 owner: payload.owner,
146 subscriptionFilters: payload.subscriptionFilters,
147 })),
148 ),
149 );
150 }).pipe(Effect.orDie);
151 }
152 };
153 }),
154 );
155 }) as LogGroupEventSourceService;
156 }),
157);