1import type * as Logs from "@distilled.cloud/aws/cloudwatch-logs";
2import type * as Effect from "effect/Effect";
3import type * as Sink from "effect/Sink";
4import * as Binding from "../../Binding.ts";
5import type { BatchRetryExhaustedError } from "../internal/BatchedSink.ts";
6import type { LogGroup } from "./LogGroup.ts";
7
8export interface LogEventSinkProps {
9 /**
10 * Name of the target log stream. The stream must already exist — declare an
11 * `AWS.Logs.LogStream` alongside the log group.
12 */
13 readonly logStreamName: string;
14}
15
16export type LogEventSinkError =
17 | Logs.PutLogEventsError
18 | BatchRetryExhaustedError<Logs.InputLogEvent>;
19
20/**
21 * A batching sink over CloudWatch Logs `PutLogEvents` (10,000 events /
22 * 1,048,576 bytes per call, where each event costs its UTF-8 message size
23 * plus 26 bytes of overhead).
24 *
25 * Each input element is a raw `InputLogEvent`, so ordering stays with the
26 * caller: `PutLogEvents` requires events in chronological order by
27 * `timestamp`, and the time span within one batch must not exceed 24 hours.
28 *
29 * Events the API reports via `rejectedLogEventsInfo` (older than 14 days or
30 * the group's retention period, expired, or more than 2 hours in the future)
31 * are **permanently** rejected — they are dropped and surfaced with a
32 * warning, never retried. The remaining valid events in the batch are still
33 * ingested by the API.
34 *
35 * ### Streaming Log Events
36 * **Example:** Drain a Stream of Events into a Log Stream
37 * ```typescript
38 * const sink = yield* AWS.Logs.LogEventSink(logGroup, {
39 * logStreamName: "audit-stream",
40 * });
41 *
42 * // Inside a handler: drain fully before returning.
43 * yield* Stream.fromIterable(entries).pipe(
44 * Stream.map((entry) => ({ timestamp: entry.at, message: entry.text })),
45 * Stream.run(sink),
46 * );
47 * ```
48 *
49 * **Example:** Wire into a Lambda Function
50 * ```typescript
51 * // LogEventSinkHttp batches over the PutLogEvents binding, so provide
52 * // PutLogEventsHttp into it with Layer.provideMerge.
53 * export default IngestFunction.make(
54 * { main: import.meta.url, functionUrl: true },
55 * Effect.gen(function* () {
56 * const logGroup = yield* AWS.Logs.LogGroup("IngestLogs", {});
57 * yield* AWS.Logs.LogStream("IngestStream", {
58 * logGroupName: logGroup.logGroupName,
59 * logStreamName: "ingest-stream",
60 * });
61 * const sink = yield* AWS.Logs.LogEventSink(logGroup, {
62 * logStreamName: "ingest-stream",
63 * });
64 * // ... run streams of InputLogEvents into `sink` in the fetch handler
65 * return { fetch: handler };
66 * }).pipe(
67 * Effect.provide(
68 * Layer.provideMerge(AWS.Logs.LogEventSinkHttp, AWS.Logs.PutLogEventsHttp),
69 * ),
70 * ),
71 * );
72 * ```
73 *
74 * @binding
75 */
76export interface LogEventSink extends Binding.Service<
77 LogEventSink,
78 "AWS.Logs.LogEventSink",
79 (
80 logGroup: LogGroup,
81 props: LogEventSinkProps,
82 ) => Effect.Effect<
83 Sink.Sink<
84 void,
85 Logs.InputLogEvent,
86 readonly Logs.InputLogEvent[],
87 LogEventSinkError
88 >
89 >
90> {}
91
92export const LogEventSink = Binding.Service<LogEventSink>(
93 "AWS.Logs.LogEventSink",
94);