1import * as Context from "effect/Context";
2import type * as Effect from "effect/Effect";
3import type * as Stream from "effect/Stream";
4import type { LogGroup } from "./LogGroup.ts";
5
6/**
7 * Raw Lambda invocation payload produced by a CloudWatch Logs subscription
8 * filter: the log events are gzipped, base64-encoded JSON under
9 * `awslogs.data`.
10 */
11export interface CloudWatchLogsEvent {
12 /** Envelope holding the gzipped, base64-encoded subscription payload. */
13 awslogs: {
14 /** Base64-encoded gzip of a {@link LogsSubscriptionPayload} JSON document. */
15 data: string;
16 };
17}
18
19/**
20 * Decoded CloudWatch Logs subscription payload (after base64 + gunzip).
21 */
22export interface LogsSubscriptionPayload {
23 /** `DATA_MESSAGE` carries log events; `CONTROL_MESSAGE` is a delivery check. */
24 messageType: "DATA_MESSAGE" | "CONTROL_MESSAGE";
25 /** AWS account id that owns the source log group. */
26 owner: string;
27 /** Name of the log group the events came from. */
28 logGroup: string;
29 /** Name of the log stream the events came from. */
30 logStream: string;
31 /** Names of the subscription filters that matched the events. */
32 subscriptionFilters: string[];
33 /** The batch of matched log events. */
34 logEvents: {
35 /** Unique identifier of the log event. */
36 id: string;
37 /** Event timestamp (epoch milliseconds). */
38 timestamp: number;
39 /** The raw log line. */
40 message: string;
41 }[];
42}
43
44/**
45 * A single decoded log event delivered through a subscription filter,
46 * flattened with its source metadata.
47 */
48export interface LogEventRecord {
49 /** Unique identifier of the log event. */
50 id: string;
51 /** Event timestamp (epoch milliseconds). */
52 timestamp: number;
53 /** The raw log line. */
54 message: string;
55 /** Name of the log group the event came from. */
56 logGroup: string;
57 /** Name of the log stream the event came from. */
58 logStream: string;
59 /** AWS account id that owns the source log group. */
60 owner: string;
61 /** Names of the subscription filters that matched the event. */
62 subscriptionFilters: string[];
63}
64
65export interface LogGroupEventSourceProps {
66 /**
67 * Name for the backing subscription filter. If omitted, a deterministic
68 * name is generated from the logical ids.
69 */
70 filterName?: string;
71 /**
72 * Filter pattern selecting which log events are delivered.
73 * An empty string matches every log event.
74 * @default ""
75 */
76 filterPattern?: string;
77}
78
79type LogEventsHandler<Req> = (
80 events: Stream.Stream<LogEventRecord>,
81) => Effect.Effect<void, never, Req>;
82
83/**
84 * Subscribe an Effect handler to log events produced by a CloudWatch Logs
85 * {@link LogGroup}.
86 *
87 * At deploy time this creates the backing subscription filter targeting the
88 * host Lambda function plus the `lambda:InvokeFunction` permission for
89 * `logs.amazonaws.com`. At runtime the gzipped subscription payload is decoded
90 * and each log event is delivered to the handler as a {@link LogEventRecord}.
91 *
92 * :::warning
93 * Never subscribe a function to its own log group — every line the handler
94 * logs would re-invoke the handler in an infinite loop.
95 * :::
96 *
97 * @param logGroup The log group to consume log events from.
98 * @param props Optional subscription filter configuration.
99 * @param process The handler invoked with a stream of decoded log events
100 * (last argument).
101 * ### Consuming Log Events
102 * **Example:** Forward Error Logs
103 * ```typescript
104 * yield* AWS.Logs.consumeLogEvents(
105 * logGroup,
106 * { filterPattern: "?ERROR ?Error" },
107 * (events) =>
108 * Stream.runForEach(events, (event) =>
109 * Effect.log(`${event.logStream}: ${event.message}`),
110 * ),
111 * );
112 * ```
113 *
114 * **Example:** Wire the Event Source into a Lambda Function
115 * ```typescript
116 * // The Lambda implementation layer (AWS.Lambda.LogGroupEventSource)
117 * // satisfies the LogGroupEventSource requirement of consumeLogEvents.
118 * export default AlertsFunction.make(
119 * { main: import.meta.url },
120 * Effect.gen(function* () {
121 * const source = yield* AWS.Logs.LogGroup("AppLogs", {
122 * retention: "7 days",
123 * });
124 * yield* AWS.Logs.consumeLogEvents(source, (events) =>
125 * Stream.runForEach(events, (event) => Effect.log(event.message)),
126 * );
127 * return {
128 * fetch: Effect.succeed(HttpServerResponse.text("ok")),
129 * };
130 * }).pipe(Effect.provide(AWS.Lambda.LogGroupEventSource)),
131 * );
132 * ```
133 *
134 * @binding
135 */
136export function consumeLogEvents<G extends LogGroup, Req = never>(
137 logGroup: G,
138 process: LogEventsHandler<Req>,
139): Effect.Effect<void, never, LogGroupEventSource>;
140export function consumeLogEvents<G extends LogGroup, Req = never>(
141 logGroup: G,
142 props: LogGroupEventSourceProps,
143 process: LogEventsHandler<Req>,
144): Effect.Effect<void, never, LogGroupEventSource>;
145export function consumeLogEvents<G extends LogGroup, Req = never>(
146 logGroup: G,
147 propsOrProcess: LogGroupEventSourceProps | LogEventsHandler<Req>,
148 maybeProcess?: LogEventsHandler<Req>,
149): Effect.Effect<void, never, LogGroupEventSource> {
150 const [props, process] =
151 typeof propsOrProcess === "function"
152 ? [{} as LogGroupEventSourceProps, propsOrProcess]
153 : [propsOrProcess, maybeProcess!];
154 return LogGroupEventSource.use((source) => source(logGroup, props, process));
155}
156
157export class LogGroupEventSource extends Context.Service<
158 LogGroupEventSource,
159 LogGroupEventSourceService
160>()("AWS.Logs.LogGroupEventSource") {}
161
162export type LogGroupEventSourceService = <Req = never>(
163 logGroup: LogGroup,
164 props: LogGroupEventSourceProps,
165 process: (
166 events: Stream.Stream<LogEventRecord>,
167 ) => Effect.Effect<void, never, Req>,
168) => Effect.Effect<void, never, never>;