1import type * as Effect from "effect/Effect";
2import type * as Stream from "effect/Stream";
3import {
4 consumeBusEvents,
5 type EventRecord,
6 type EventRouteProps,
7} from "../EventBridge/EventSource.ts";
8
9/**
10 * The `detail` payload SES delivers to EventBridge for an email sending
11 * event. Every event carries `eventType` and `mail`; the event-specific
12 * member (`bounce`, `complaint`, `delivery`, …) is present only on that
13 * event kind.
14 */
15export interface EmailEventDetail {
16 /** The event kind, e.g. `Send`, `Delivery`, `Bounce`, `Complaint`. */
17 eventType?: string;
18 /**
19 * The original message: `messageId`, `source`, `destination`, and the
20 * message `tags` (including `ses:configuration-set`).
21 */
22 mail?: {
23 messageId?: string;
24 source?: string;
25 destination?: string[];
26 tags?: Record<string, string[]>;
27 [key: string]: unknown;
28 };
29 /** Bounce events: bounce type/subtype and the bounced recipients. */
30 bounce?: Record<string, unknown>;
31 /** Complaint events: the complained recipients and feedback type. */
32 complaint?: Record<string, unknown>;
33 /** Delivery events: recipients, SMTP response, processing time. */
34 delivery?: Record<string, unknown>;
35 /** Additional event fields (the schema grows over time). */
36 [key: string]: unknown;
37}
38
39/** An SES email EventBridge event delivered to the handler. */
40export type EmailEvent = EventRecord<EmailEventDetail>;
41
42/** Which SES email sending events to subscribe to. */
43export type EmailEventKind =
44 | "send"
45 | "reject"
46 | "delivery"
47 | "delivery-delay"
48 | "bounce"
49 | "complaint"
50 | "open"
51 | "click"
52 | "rendering-failure"
53 | "subscription";
54
55const DETAIL_TYPES: Record<EmailEventKind, string> = {
56 send: "Email Sent",
57 reject: "Email Rejected",
58 delivery: "Email Delivered",
59 "delivery-delay": "Email Delivery Delayed",
60 bounce: "Email Bounced",
61 complaint: "Email Complaint Received",
62 open: "Email Opened",
63 click: "Email Clicked",
64 "rendering-failure": "Email Rendering Failed",
65 subscription: "Email Subscription",
66};
67
68export interface EmailEventSourceProps extends EventRouteProps {
69 /**
70 * Logical id for the backing EventBridge rule.
71 * @default "SESEmailEvents"
72 */
73 id?: string;
74 /**
75 * Which email sending events to subscribe to.
76 * @default ["bounce", "complaint"]
77 */
78 kinds?: readonly EmailEventKind[];
79 /**
80 * Restrict to events sent through specific configuration sets (matched
81 * against the message's `ses:configuration-set` tag).
82 */
83 configurationSets?: readonly string[];
84}
85
86/**
87 * Event source connecting SES email sending events to the hosting compute.
88 * SES publishes send/delivery/bounce/complaint (and open/click) events to
89 * the account's default EventBridge bus (source `aws.ses`) for every
90 * configuration set that has an EventBridge event destination; this
91 * subscribes the host Function to those events so it can suppress bouncing
92 * addresses, update subscriber state, or alert on complaints.
93 *
94 * Events only flow for messages sent through a `ConfigurationSet` that has
95 * a `ConfigurationSetEventDestination` with an `eventBridgeDestination`
96 * (the default bus). Provide the host-specific implementation layer
97 * (e.g. `AWS.Lambda.EventSource`) on the Function effect.
98 *
99 * ### Consuming Email Events
100 * **Example:** Suppress Hard-Bouncing Addresses
101 * ```typescript
102 * import * as AWS from "alchemy/AWS";
103 *
104 * export default FeedbackFunction.make(
105 * { main: import.meta.url },
106 * Effect.gen(function* () {
107 * yield* AWS.SES.consumeEmailEvents(
108 * { kinds: ["bounce", "complaint"] },
109 * (events) =>
110 * Stream.runForEach(events, (event) =>
111 * Effect.log(`${event["detail-type"]}: ${event.detail.mail?.messageId}`),
112 * ),
113 * );
114 * return {};
115 * }).pipe(Effect.provide(AWS.Lambda.EventSource)),
116 * );
117 * ```
118 */
119export const consumeEmailEvents = <StreamReq = never, Req = never>(
120 props: EmailEventSourceProps,
121 process: (
122 events: Stream.Stream<EmailEvent, never, StreamReq>,
123 ) => Effect.Effect<void, never, Req>,
124) =>
125 consumeBusEvents(
126 props.id ?? "SESEmailEvents",
127 {
128 source: ["aws.ses"],
129 "detail-type": (props.kinds ?? (["bounce", "complaint"] as const)).map(
130 (kind) => DETAIL_TYPES[kind],
131 ),
132 ...(props.configurationSets !== undefined
133 ? {
134 detail: {
135 mail: {
136 tags: {
137 "ses:configuration-set": [...props.configurationSets],
138 },
139 },
140 },
141 }
142 : {}),
143 },
144 { description: props.description, state: props.state },
145 process,
146 );