1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import * as Binding from "../../Binding.ts";
5import * as Namespace from "../../Namespace.ts";
6import { ServerHost } from "../../Server/Process.ts";
7import { Acknowledge } from "../PubSub/Acknowledge.ts";
8import { AcknowledgeHttp } from "../PubSub/AcknowledgeHttp.ts";
9import { Pull } from "../PubSub/Pull.ts";
10import { PullHttp } from "../PubSub/PullHttp.ts";
11import { Subscription } from "../PubSub/Subscription.ts";
12import type { Topic } from "../PubSub/Topic.ts";
13import {
14 TopicEventSource as PubSubTopicEventSource,
15 type TopicEventSourceProps,
16 type TopicEventSourceService,
17 type TopicMessage,
18} from "../PubSub/TopicEventSource.ts";
19
20/**
21 * Subscription `deadLetterPolicy` and `retryPolicy` for the event source's
22 * props. Dead-lettering defaults to a backoff so its attempts span more
23 * than a transient failure.
24 */
25const deliveryPolicyOf = (props: TopicEventSourceProps) => ({
26 deadLetterPolicy:
27 props.deadLetter === undefined
28 ? undefined
29 : {
30 deadLetterTopic: props.deadLetter.topic.name,
31 maxDeliveryAttempts: props.deadLetter.maxDeliveryAttempts ?? 5,
32 },
33 retryPolicy:
34 props.retryPolicy ??
35 (props.deadLetter === undefined
36 ? undefined
37 : { minimumBackoff: "10s", maximumBackoff: "600s" }),
38});
39
40/**
41 * Pull implementation of `GCP.PubSub.TopicEventSource` for hosts without
42 * an inbound URL (`GCP.Run.Job`, `GCP.Run.WorkerPool`; any host with
43 * `ServerHost`).
44 *
45 * Deploy-time: creates a pull subscription owned by the host and grants
46 * the host's runtime service account `roles/pubsub.subscriber` on it
47 * (through the `Pull` / `Acknowledge` bindings). Runtime: a background
48 * loop pulls batches, runs the handler, and acks the batch once the
49 * handler succeeds. A failed handler leaves the batch unacked, so Pub/Sub
50 * redelivers after the ack deadline.
51 *
52 * @layer
53 * @provides GCP.PubSub.TopicEventSource
54 * @category Run
55 */
56export const TopicPullEventSource = Layer.effect(
57 PubSubTopicEventSource,
58 Effect.gen(function* () {
59 const { run } = yield* ServerHost;
60 const subscription = yield* Subscription;
61 const pull = yield* Pull;
62 const acknowledge = yield* Acknowledge;
63
64 return Effect.fn(function* <Req = never>(
65 topic: Topic,
66 props: TopicEventSourceProps,
67 process: (
68 messages: Stream.Stream<TopicMessage>,
69 ) => Effect.Effect<void, never, Req>,
70 ) {
71 const host = yield* Binding.Host;
72 const hostId = host?.LogicalId ?? "Host";
73 // Declared in both phases: at runtime the declaration resolves to the
74 // deployed subscription's outputs, which the bindings below read.
75 const sub = yield* Namespace.push(
76 hostId,
77 subscription(`${topic.LogicalId}-Pull`, {
78 topic: topic.name,
79 ackDeadlineSeconds: props.ackDeadlineSeconds ?? 60,
80 filter: props.filter,
81 ...deliveryPolicyOf(props),
82 }),
83 );
84 const pullBatch = yield* pull(sub);
85 const ackBatch = yield* acknowledge(sub);
86 const subscriptionName = yield* sub.name;
87
88 yield* run(
89 Effect.gen(function* () {
90 const name = yield* subscriptionName;
91 const { receivedMessages = [] } = yield* pullBatch({
92 body: { maxMessages: props.maxMessages ?? 10 },
93 });
94 const received = receivedMessages.filter(
95 (item) => item.ackId !== undefined && item.message !== undefined,
96 );
97 if (received.length === 0) return;
98 yield* process(
99 Stream.fromIterable(
100 received.map((item) => ({
101 message: item.message!,
102 subscription: name,
103 deliveryAttempt: item.deliveryAttempt,
104 })),
105 ),
106 );
107 yield* ackBatch({
108 body: { ackIds: received.map((item) => item.ackId!) },
109 });
110 }).pipe(
111 // A failed pull, ack, or handler must not end the consumer; the
112 // unacked batch is redelivered after its ack deadline.
113 Effect.catchCause((cause) =>
114 Effect.logWarning("Pub/Sub pull iteration failed", cause).pipe(
115 Effect.andThen(Effect.sleep("1 second")),
116 ),
117 ),
118 Effect.forever,
119 ),
120 );
121 }) as TopicEventSourceService;
122 }),
123).pipe(Layer.provide(Layer.mergeAll(PullHttp, AcknowledgeHttp)));