1import type * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
2import * as Effect from "effect/Effect";
3import type * as Stream from "effect/Stream";
4import * as Binding from "../../Binding.ts";
5import type { Topic } from "./Topic.ts";
6
7/** One delivered Pub/Sub message, as push and pull deliver it. */
8export interface TopicMessage {
9 /** The message: base64 `data`, `attributes`, `messageId`, `publishTime`. */
10 message: pubsub.PubsubMessage;
11 /** Subscription the message arrived on (`projects/p/subscriptions/s`). */
12 subscription: string;
13 /** Delivery attempt, when a dead-letter policy is configured. */
14 deliveryAttempt: number | undefined;
15}
16
17export interface TopicEventSourceProps {
18 /**
19 * Seconds Pub/Sub waits for an ack (a 2xx push response, or an explicit
20 * pull ack) before redelivering.
21 * @default 60
22 */
23 ackDeadlineSeconds?: number;
24 /**
25 * Pub/Sub filter expression on message attributes; only matching
26 * messages are delivered.
27 */
28 filter?: string;
29 /**
30 * Push delivery path on the host. Defaults to a deterministic
31 * per-topic path under `/__alchemy/pubsub/`. Push hosts only.
32 */
33 path?: string;
34 /**
35 * Maximum messages per pull. Pull hosts only.
36 * @default 10
37 */
38 maxMessages?: number;
39 /**
40 * Forward messages the handler keeps failing on to another topic instead
41 * of redelivering them forever. The Pub/Sub service agent is granted
42 * `roles/pubsub.publisher` on `topic` and `roles/pubsub.subscriber` on
43 * the event source's subscription. Give `topic` its own subscription, or
44 * dead-lettered messages are dropped.
45 */
46 deadLetter?: TopicDeadLetter;
47 /**
48 * Backoff between redeliveries of a failed message. Without one, Pub/Sub
49 * redelivers almost immediately, so a `deadLetter` policy would spend
50 * every attempt within seconds of a transient failure (e.g. a fresh
51 * deploy's IAM grants still propagating).
52 * @default { minimumBackoff: "10s", maximumBackoff: "600s" } with `deadLetter`, otherwise Pub/Sub's immediate redelivery
53 */
54 retryPolicy?: TopicRetryPolicy;
55}
56
57export interface TopicRetryPolicy {
58 /** Minimum delay before redelivering a failed message, e.g. `"10s"` (0–600s). */
59 minimumBackoff?: string;
60 /** Maximum delay before redelivering a failed message, e.g. `"600s"` (0–600s). */
61 maximumBackoff?: string;
62}
63
64export interface TopicDeadLetter {
65 /** Topic that receives dead-lettered messages. */
66 topic: Topic;
67 /**
68 * Delivery attempts before a message is dead-lettered. Between 5 and
69 * 100; approximate, since Pub/Sub counts attempts best-effort.
70 * @default 5
71 */
72 maxDeliveryAttempts?: number;
73}
74
75export type TopicMessagesHandler<Req> = (
76 messages: Stream.Stream<TopicMessage>,
77) => Effect.Effect<void, never, Req>;
78
79export type TopicEventSourceService = <Req = never>(
80 topic: Topic,
81 props: TopicEventSourceProps,
82 process: TopicMessagesHandler<Req>,
83) => Effect.Effect<void, never, never>;
84
85/**
86 * Event source connecting a Pub/Sub {@link Topic} to the hosting compute.
87 *
88 * The host-specific implementation layers are:
89 *
90 * - `GCP.Run.TopicEventSource` — push delivery to an HTTP host
91 * (`GCP.Run.Service` / `GCP.Function`, `GCP.CloudFunctions.Function`).
92 * Provisions a push subscription at the host's URL, authenticated with an
93 * OIDC token for the host's runtime service account, grants that account
94 * `roles/run.invoker` on the host, and verifies the token on every
95 * delivery. A 2xx response acks; a failed handler returns 500 and Pub/Sub
96 * redelivers.
97 * - `GCP.Run.TopicPullEventSource` — a pull loop for hosts without an
98 * inbound URL (`GCP.Run.Job`, `GCP.Run.WorkerPool`). Provisions a pull
99 * subscription, grants `roles/pubsub.subscriber` on it, and acks each
100 * batch after the handler succeeds.
101 *
102 * Consume it through {@link consumeTopicMessages}.
103 *
104 * ### Consuming a Topic
105 * **Example:** Push to a Cloud Run service
106 * ```typescript
107 * export class Worker extends GCP.Function<Worker>()(
108 * "Worker",
109 * { main: import.meta.url },
110 * Effect.gen(function* () {
111 * const orders = yield* GCP.PubSub.Topic("Orders", {});
112 * yield* GCP.PubSub.consumeTopicMessages(orders, (messages) =>
113 * messages.pipe(
114 * Stream.runForEach(({ message }) =>
115 * Effect.log(atob(message.data ?? "")),
116 * ),
117 * ),
118 * );
119 * }).pipe(Effect.provide(GCP.Run.TopicEventSource)),
120 * ) {}
121 * ```
122 *
123 * **Example:** Dead-letter messages the handler keeps failing on
124 * ```typescript
125 * const failed = yield* GCP.PubSub.Topic("FailedOrders", {});
126 * yield* GCP.PubSub.consumeTopicMessages(
127 * orders,
128 * { deadLetter: { topic: failed, maxDeliveryAttempts: 5 } },
129 * (messages) => messages.pipe(Stream.runForEach(handle)),
130 * );
131 * ```
132 *
133 * **Example:** Pull from a worker pool
134 * ```typescript
135 * yield* GCP.PubSub.consumeTopicMessages(
136 * orders,
137 * { maxMessages: 50 },
138 * (messages) => messages.pipe(Stream.runForEach(handle)),
139 * );
140 * // …provided with Effect.provide(GCP.Run.TopicPullEventSource)
141 * ```
142 *
143 * @binding
144 * @category PubSub
145 */
146export interface TopicEventSource extends Binding.Service<
147 TopicEventSource,
148 "GCP.PubSub.TopicEventSource",
149 TopicEventSourceService
150> {}
151
152export const TopicEventSource = Binding.Service<TopicEventSource>(
153 "GCP.PubSub.TopicEventSource",
154);
155
156/**
157 * Subscribe an Effect handler to messages published to a Pub/Sub
158 * {@link Topic}. See {@link TopicEventSource} for the host implementations.
159 */
160export function consumeTopicMessages<Req = never, ResourceReq = never>(
161 topic: Topic | Effect.Effect<Topic, never, ResourceReq>,
162 process: TopicMessagesHandler<Req>,
163): Effect.Effect<void, never, TopicEventSource | ResourceReq>;
164export function consumeTopicMessages<Req = never, ResourceReq = never>(
165 topic: Topic | Effect.Effect<Topic, never, ResourceReq>,
166 props: TopicEventSourceProps,
167 process: TopicMessagesHandler<Req>,
168): Effect.Effect<void, never, TopicEventSource | ResourceReq>;
169export function consumeTopicMessages<Req = never, ResourceReq = never>(
170 topic: Topic | Effect.Effect<Topic, never, ResourceReq>,
171 propsOrProcess: TopicEventSourceProps | TopicMessagesHandler<Req>,
172 maybeProcess?: TopicMessagesHandler<Req>,
173): Effect.Effect<void, never, TopicEventSource | ResourceReq> {
174 const [props, process] =
175 typeof propsOrProcess === "function"
176 ? [{} as TopicEventSourceProps, propsOrProcess]
177 : [propsOrProcess, maybeProcess!];
178 // Accept the resource or the Effect that declares it, like bindings do.
179 const resolved = Effect.isEffect(topic) ? topic : Effect.succeed(topic);
180 return resolved.pipe(
181 Effect.flatMap((value) =>
182 TopicEventSource.use((source) => source(value, props, process)),
183 ),
184 );
185}