1import type * as eventbridge from "@distilled.cloud/aws/eventbridge";
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 { AWSEnvironment } from "../Environment.ts";
7import { makeBatchedSink } from "../internal/BatchedSink.ts";
8import { isBindingHost } from "../Lambda/Function.ts";
9import { BusSink, type BusSinkEntry } from "./BusSink.ts";
10import type { EventBus } from "./EventBus.ts";
11import { PutEvents } from "./PutEvents.ts";
12
13const encoder = new TextEncoder();
14
15/**
16 * Approximate wire size of one entry, per AWS's documented calculation:
17 * 14 bytes if `Time` is specified, plus the UTF-8 byte lengths of `Source`,
18 * `DetailType`, `Detail`, and each entry in `Resources`.
19 */
20const entrySize = (entry: BusSinkEntry): number => {
21 let size = entry.Time !== undefined ? 14 : 0;
22 if (entry.Source !== undefined) {
23 size += encoder.encode(entry.Source).length;
24 }
25 if (entry.DetailType !== undefined) {
26 size += encoder.encode(entry.DetailType).length;
27 }
28 if (entry.Detail !== undefined) {
29 size += encoder.encode(entry.Detail).length;
30 }
31 for (const resource of entry.Resources ?? []) {
32 size += encoder.encode(resource).length;
33 }
34 return size;
35};
36
37/**
38 * Per-entry error codes EventBridge documents as retryable. Every other
39 * `ErrorCode` (`MalformedDetail`, `AccessDeniedException`, ...) is permanent.
40 */
41const TRANSIENT_ERROR_CODES: ReadonlySet<string> = new Set([
42 "ThrottlingException",
43 "InternalFailure",
44]);
45
46/**
47 * Select the original entries whose positional `PutEventsResultEntry` carries
48 * an `ErrorCode` of the given transience, preserving input order (the result
49 * `Entries` array is aligned index-for-index with the request entries).
50 */
51const selectFailed = (
52 out: eventbridge.PutEventsResponse,
53 batch: readonly BusSinkEntry[],
54 transient: boolean,
55): readonly BusSinkEntry[] => {
56 const results = out.Entries;
57 if (results === undefined || (out.FailedEntryCount ?? 0) === 0) {
58 return [];
59 }
60 return batch.filter((_, index) => {
61 const code = results[index]?.ErrorCode;
62 return code !== undefined && TRANSIENT_ERROR_CODES.has(code) === transient;
63 });
64};
65
66/**
67 * HTTP implementation of {@link BusSink}. At deploy time it grants
68 * `events:PutEvents` on the bound bus (or the default bus); at runtime it
69 * batches stream elements into `PutEvents` calls (10 entries / 256 KiB) with
70 * bounded retry of transient per-entry failures. Provide this layer on the
71 * Function using the sink.
72 */
73export const BusSinkHttp = Layer.effect(
74 BusSink,
75 Effect.gen(function* () {
76 const putEvents = yield* PutEvents;
77
78 return Effect.fn(function* (bus?: EventBus) {
79 if (!globalThis.__ALCHEMY_RUNTIME__) {
80 const host = yield* Binding.Host;
81 if (isBindingHost(host)) {
82 const { accountId, region } =
83 yield* AWSEnvironment.current as unknown as Effect.Effect<{
84 accountId: string;
85 region: string;
86 }>;
87 // Pass the ARN as an unresolved Output — binding data is resolved
88 // by the engine before the host reconciles (see PutEventsHttp).
89 const resource = bus
90 ? Output.interpolate`${bus.eventBusArn}`
91 : (`arn:aws:events:${region}:${accountId}:event-bus/default` as const);
92
93 yield* host.bind`Allow(${host}, AWS.EventBridge.BusSink(${bus ?? "default"}))`(
94 {
95 policyStatements: [
96 {
97 Effect: "Allow",
98 Action: ["events:PutEvents"],
99 Resource: [resource],
100 },
101 ],
102 },
103 );
104 }
105 }
106 const put = yield* putEvents(bus);
107 return makeBatchedSink<
108 BusSinkEntry,
109 eventbridge.PutEventsResponse,
110 eventbridge.PutEventsError
111 >({
112 maxRecords: 10,
113 maxBytes: 262_144,
114 sizeOf: entrySize,
115 send: (batch) => put({ Entries: [...batch] }),
116 // ThrottlingException / InternalFailure are transient — re-submit
117 // them on the bounded schedule.
118 unprocessed: (out, batch) => selectFailed(out, batch, true),
119 // Any other per-entry ErrorCode (e.g. MalformedDetail) is permanent —
120 // drop and surface.
121 rejected: (out, batch) => selectFailed(out, batch, false),
122 });
123 });
124 }),
125);