1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5import * as Namespace from "../../Namespace.ts";
6import * as Output from "../../Output.ts";
7import { Subscription } from "../PubSub/Subscription.ts";
8import type { Topic } from "../PubSub/Topic.ts";
9import {
10 TopicEventSource as PubSubTopicEventSource,
11 type TopicEventSourceProps,
12 type TopicEventSourceService,
13 type TopicMessage,
14} from "../PubSub/TopicEventSource.ts";
15import {
16 deliveryAudience,
17 grantSelfInvoker,
18 hostEndpoint,
19 listenForDeliveries,
20 pathSegment,
21 pushHost,
22} from "../PushDelivery.ts";
23
24/**
25 * Subscription `deadLetterPolicy` and `retryPolicy` for the event source's
26 * props. Dead-lettering defaults to a backoff so its attempts span more
27 * than a transient failure.
28 */
29const deliveryPolicyOf = (props: TopicEventSourceProps) => ({
30 deadLetterPolicy:
31 props.deadLetter === undefined
32 ? undefined
33 : {
34 deadLetterTopic: props.deadLetter.topic.name,
35 maxDeliveryAttempts: props.deadLetter.maxDeliveryAttempts ?? 5,
36 },
37 retryPolicy:
38 props.retryPolicy ??
39 (props.deadLetter === undefined
40 ? undefined
41 : { minimumBackoff: "10s", maximumBackoff: "600s" }),
42});
43
44/** Default push path for a topic's deliveries. */
45export const topicPushPath = (topic: Topic, props: TopicEventSourceProps) =>
46 props.path ?? `/__alchemy/pubsub/${pathSegment(topic.LogicalId)}`;
47
48interface PushEnvelope {
49 message?: TopicMessage["message"];
50 subscription?: string;
51 deliveryAttempt?: number;
52}
53
54/**
55 * Push implementation of `GCP.PubSub.TopicEventSource` for HTTP hosts
56 * (`GCP.Run.Service` / `GCP.Function`, `GCP.CloudFunctions.Function`).
57 *
58 * Deploy-time: grants the host's runtime service account
59 * `roles/run.invoker` on the host and creates a push subscription to the
60 * host's URL, signed with an OIDC token for that account. Runtime: claims
61 * `POST` deliveries on the push path, verifies the token, and runs the
62 * handler. 204 acks the message; a failed handler answers 500 and Pub/Sub
63 * redelivers per the subscription's retry policy.
64 *
65 * @layer
66 * @provides GCP.PubSub.TopicEventSource
67 * @category Run
68 */
69export const TopicEventSource = Layer.effect(
70 PubSubTopicEventSource,
71 Effect.gen(function* () {
72 const subscription = yield* Subscription;
73
74 return Effect.fn(function* <Req = never>(
75 topic: Topic,
76 props: TopicEventSourceProps,
77 process: (
78 messages: Stream.Stream<TopicMessage>,
79 ) => Effect.Effect<void, never, Req>,
80 ) {
81 const host = yield* pushHost("GCP.PubSub.TopicEventSource");
82 const path = topicPushPath(topic, props);
83
84 if (!globalThis.__ALCHEMY_RUNTIME__) {
85 const endpoint = hostEndpoint(host);
86 yield* Namespace.push(
87 host.LogicalId,
88 Effect.gen(function* () {
89 yield* grantSelfInvoker(host);
90 yield* subscription(`${topic.LogicalId}-Push`, {
91 topic: topic.name,
92 ackDeadlineSeconds: props.ackDeadlineSeconds ?? 60,
93 filter: props.filter,
94 ...deliveryPolicyOf(props),
95 pushConfig: {
96 pushEndpoint: Output.interpolate`${endpoint.url}${path}`,
97 oidcToken: {
98 serviceAccountEmail: endpoint.serviceAccount,
99 audience: deliveryAudience(endpoint.url, path),
100 },
101 },
102 });
103 }),
104 );
105 }
106
107 yield* listenForDeliveries(host, path, (request) =>
108 Effect.gen(function* () {
109 const envelope = (yield* request.json.pipe(
110 Effect.orElseSucceed(() => undefined),
111 )) as PushEnvelope | undefined;
112 if (envelope?.message === undefined) {
113 // Malformed deliveries can never succeed; ack so they don't loop.
114 return HttpServerResponse.empty({ status: 204 });
115 }
116 yield* process(
117 Stream.make({
118 message: envelope.message,
119 subscription: envelope.subscription ?? "",
120 deliveryAttempt: envelope.deliveryAttempt,
121 }),
122 ).pipe(Effect.orDie);
123 return HttpServerResponse.empty({ status: 204 });
124 }),
125 );
126 }) as TopicEventSourceService;
127 }),
128);