1import type * as lambda from "aws-lambda";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import * as Binding from "../../Binding.ts";
5import type { Topic } from "./Topic.ts";
6
7export type TopicNotification = lambda.SNSMessage;
8
9export interface TopicEventSourceProps {
10 /**
11 * Raw SNS subscription attributes for the Lambda subscription, such as
12 * `FilterPolicy` or `RedrivePolicy`.
13 */
14 attributes?: Record<string, string>;
15}
16
17/**
18 * Event source connecting an SNS {@link Topic} to the hosting Lambda
19 * function: it creates the `lambda`-protocol subscription (plus the invoke
20 * permission) at deploy time and dispatches delivered notifications to the
21 * registered handler at runtime.
22 *
23 * The contract is a `Binding.Service`; the Lambda implementation layer is
24 * `Lambda.TopicEventSource`. Consume it through the
25 * {@link consumeTopicNotifications} helper.
26 * ### Consuming a Topic
27 * **Example:** Consume Notifications in a Lambda Function
28 * ```typescript
29 * export default WorkerFunction.make(
30 * { main: import.meta.url },
31 * Effect.gen(function* () {
32 * const topic = yield* SNS.Topic("Events");
33 *
34 * // registers the subscription and the runtime dispatcher
35 * yield* SNS.consumeTopicNotifications(topic, (stream) =>
36 * stream.pipe(
37 * Stream.runForEach((message) => Effect.log(message.Message)),
38 * ),
39 * );
40 * }).pipe(Effect.provide(Lambda.TopicEventSource)),
41 * );
42 * ```
43 *
44 * @binding
45 */
46export interface TopicEventSource extends Binding.Service<
47 TopicEventSource,
48 "AWS.SNS.TopicEventSource",
49 TopicEventSourceService
50> {}
51export const TopicEventSource = Binding.Service<TopicEventSource>(
52 "AWS.SNS.TopicEventSource",
53);
54
55export type TopicEventSourceService = <StreamReq = never, Req = never>(
56 topic: Topic,
57 props: TopicEventSourceProps,
58 process: (
59 stream: Stream.Stream<TopicNotification, never, StreamReq>,
60 ) => Effect.Effect<void, never, Req>,
61) => Effect.Effect<void, never, never>;
62
63type TopicEventSourceHandler<Req, StreamReq> = (
64 stream: Stream.Stream<TopicNotification, never, StreamReq>,
65) => Effect.Effect<void, never, Req>;
66
67/**
68 * Subscribe a Lambda Function to an SNS {@link Topic}, processing published
69 * notifications as a stream.
70 *
71 * @example
72 * ```typescript
73 * yield* consumeTopicNotifications(topic, (stream) =>
74 * stream.pipe(Stream.runForEach((message) => Effect.log(message.Message))),
75 * );
76 * ```
77 *
78 * @example With subscription attributes
79 * ```typescript
80 * yield* consumeTopicNotifications(
81 * topic,
82 * { attributes: { FilterPolicy: JSON.stringify({ type: ["order"] }) } },
83 * (stream) =>
84 * stream.pipe(Stream.runForEach((message) => Effect.log(message.Message))),
85 * );
86 * ```
87 */
88export function consumeTopicNotifications<
89 T extends Topic,
90 Req = never,
91 StreamReq = never,
92>(
93 topic: T,
94 process: TopicEventSourceHandler<Req, StreamReq>,
95): Effect.Effect<void, never, TopicEventSource>;
96export function consumeTopicNotifications<
97 T extends Topic,
98 Req = never,
99 StreamReq = never,
100>(
101 topic: T,
102 props: TopicEventSourceProps,
103 process: TopicEventSourceHandler<Req, StreamReq>,
104): Effect.Effect<void, never, TopicEventSource>;
105export function consumeTopicNotifications<
106 T extends Topic,
107 Req = never,
108 StreamReq = never,
109>(
110 topic: T,
111 propsOrProcess:
112 | TopicEventSourceProps
113 | TopicEventSourceHandler<Req, StreamReq>,
114 maybeProcess?: TopicEventSourceHandler<Req, StreamReq>,
115) {
116 const [props, process] =
117 typeof propsOrProcess === "function"
118 ? [{} as TopicEventSourceProps, propsOrProcess]
119 : [propsOrProcess, maybeProcess!];
120 return TopicEventSource.use((source) => source(topic, props, process));
121}