1import type * as Logs from "@distilled.cloud/aws/cloudwatch-logs";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Binding from "../../Binding.ts";
5import * as Output from "../../Output.ts";
6import { makeBatchedSink } from "../internal/BatchedSink.ts";
7import { isBindingHost } from "../Lambda/Function.ts";
8import { LogEventSink, type LogEventSinkProps } from "./LogEventSink.ts";
9import type { LogGroup } from "./LogGroup.ts";
10import { PutLogEvents } from "./PutLogEvents.ts";
11
12const encoder = new TextEncoder();
13
14/**
15 * The `PutLogEvents` batch-size formula charges each event its UTF-8 message
16 * size plus 26 bytes of overhead.
17 */
18const PER_EVENT_OVERHEAD = 26;
19
20/**
21 * Select the events `rejectedLogEventsInfo` marks as permanently rejected,
22 * preserving input order. The API skips (never ingests) these while still
23 * processing the remaining valid events, so they must be dropped — not
24 * retried:
25 *
26 * - `tooOldLogEventEndIndex` / `expiredLogEventEndIndex` — everything up to
27 * and including that index is too old / past retention;
28 * - `tooNewLogEventStartIndex` — everything from that index on is more than
29 * 2 hours in the future.
30 */
31const selectRejected = (
32 out: Logs.PutLogEventsResponse,
33 batch: readonly Logs.InputLogEvent[],
34): readonly Logs.InputLogEvent[] => {
35 const info = out.rejectedLogEventsInfo;
36 if (info === undefined) {
37 return [];
38 }
39 const rejectedHeadEnd = Math.max(
40 info.tooOldLogEventEndIndex ?? -1,
41 info.expiredLogEventEndIndex ?? -1,
42 );
43 const tooNewStart = info.tooNewLogEventStartIndex ?? batch.length;
44 return batch.filter(
45 (_, index) => index <= rejectedHeadEnd || index >= tooNewStart,
46 );
47};
48
49export const LogEventSinkHttp = Layer.effect(
50 LogEventSink,
51 Effect.gen(function* () {
52 const putLogEvents = yield* PutLogEvents;
53
54 return Effect.fn(function* (logGroup: LogGroup, props: LogEventSinkProps) {
55 if (!globalThis.__ALCHEMY_RUNTIME__) {
56 const host = yield* Binding.Host;
57 if (isBindingHost(host)) {
58 yield* host.bind`Allow(${host}, AWS.Logs.LogEventSink(${logGroup}))`({
59 policyStatements: [
60 {
61 // PutLogEvents authorizes against the log-stream resource
62 // (`log-group:{name}:log-stream:{stream}`), covered by the
63 // `:*` wildcard on the group ARN.
64 Effect: "Allow",
65 Action: ["logs:PutLogEvents"],
66 Resource: [Output.interpolate`${logGroup.logGroupArn}:*`],
67 },
68 ],
69 });
70 }
71 }
72 const put = yield* putLogEvents(logGroup);
73 return makeBatchedSink<
74 Logs.InputLogEvent,
75 Logs.PutLogEventsResponse,
76 Logs.PutLogEventsError
77 >({
78 maxRecords: 10_000,
79 maxBytes: 1_048_576,
80 sizeOf: (event) =>
81 encoder.encode(event.message).length + PER_EVENT_OVERHEAD,
82 send: (batch) =>
83 put({
84 logStreamName: props.logStreamName,
85 logEvents: [...batch],
86 }),
87 // No `unprocessed` extractor: PutLogEvents has no transient per-event
88 // failure mode — timestamp rejections are permanent, so drop + surface.
89 rejected: selectRejected,
90 });
91 });
92 }),
93);