1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
4import { expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8
9const { test } = Test.make({ providers: GCP.providers() });
10
11const logLevel = Effect.provideService(
12 MinimumLogLevel,
13 process.env.DEBUG ? "Debug" : "Info",
14);
15
16const waitUntilGone = (name: string) =>
17 pubsub.getProjectsSubscriptions({ subscription: name }).pipe(
18 Effect.as("found" as const),
19 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
20 Effect.repeat({
21 schedule: Schedule.spaced("1 second"),
22 until: (status) => status === "gone",
23 times: 10,
24 }),
25 );
26
27const getSubscription = (name: string) =>
28 pubsub.getProjectsSubscriptions({ subscription: name }).pipe(
29 Effect.retry({
30 while: (error) => error._tag === "NotFound",
31 schedule: Schedule.spaced("1 second"),
32 times: 8,
33 }),
34 );
35
36const pullUntilMessage = (name: string) =>
37 pubsub
38 .pullProjectsSubscriptions({
39 subscription: name,
40 body: { maxMessages: 1, returnImmediately: true },
41 })
42 .pipe(
43 Effect.repeat({
44 schedule: Schedule.spaced("1 second"),
45 until: (response) => (response.receivedMessages?.length ?? 0) > 0,
46 times: 10,
47 }),
48 );
49
50test.provider(
51 "create, update, and delete a subscription",
52 (stack) =>
53 Effect.gen(function* () {
54 yield* stack.destroy();
55
56 const created = yield* stack.deploy(
57 Effect.gen(function* () {
58 const topic = yield* GCP.PubSub.Topic("Events", {
59 labels: { env: "test" },
60 });
61 const subscription = yield* GCP.PubSub.Subscription("Orders", {
62 topic: topic.name,
63 labels: { env: "test" },
64 });
65 return { topic, subscription };
66 }),
67 );
68
69 expect(created.subscription.name).toContain("/subscriptions/");
70 expect(created.subscription.subscriptionId).toEqual(expect.any(String));
71 expect(created.subscription.topic).toEqual(created.topic.name);
72 expect(created.subscription.labels).toMatchObject({ env: "test" });
73 expect(created.subscription.ackDeadlineSeconds).toEqual(10);
74
75 const fetched = yield* getSubscription(created.subscription.name);
76 expect(fetched.name).toEqual(created.subscription.name);
77 expect(fetched.topic).toEqual(created.topic.name);
78 expect(fetched.labels?.env).toEqual("test");
79
80 yield* pubsub.publishProjectsTopics({
81 topic: created.topic.name,
82 body: { messages: [{ data: btoa("hello-pubsub") }] },
83 });
84
85 const pulled = yield* pullUntilMessage(created.subscription.name);
86 const received = pulled.receivedMessages ?? [];
87 expect(received.length).toBeGreaterThan(0);
88 const ackId = received[0]?.ackId;
89 expect(ackId).toEqual(expect.any(String));
90 if (ackId !== undefined) {
91 yield* pubsub.acknowledgeProjectsSubscriptions({
92 subscription: created.subscription.name,
93 body: { ackIds: [ackId] },
94 });
95 }
96
97 const updated = yield* stack.deploy(
98 Effect.gen(function* () {
99 const topic = yield* GCP.PubSub.Topic("Events", {
100 topicId: created.topic.topicId,
101 labels: { env: "test" },
102 });
103 const subscription = yield* GCP.PubSub.Subscription("Orders", {
104 subscriptionId: created.subscription.subscriptionId,
105 topic: topic.name,
106 labels: { env: "prod", role: "orders" },
107 ackDeadlineSeconds: 30,
108 messageRetentionDuration: "86400s",
109 });
110 return { topic, subscription };
111 }),
112 );
113
114 expect(updated.subscription.name).toEqual(created.subscription.name);
115 expect(updated.subscription.topic).toEqual(created.topic.name);
116 expect(updated.subscription.labels).toMatchObject({
117 env: "prod",
118 role: "orders",
119 });
120 expect(updated.subscription.ackDeadlineSeconds).toEqual(30);
121 expect(updated.subscription.messageRetentionDuration).toEqual("86400s");
122
123 const fetchedUpdated = yield* getSubscription(updated.subscription.name);
124 expect(fetchedUpdated.ackDeadlineSeconds).toEqual(30);
125 expect(fetchedUpdated.labels?.env).toEqual("prod");
126 expect(fetchedUpdated.labels?.role).toEqual("orders");
127 expect(fetchedUpdated.messageRetentionDuration).toEqual("86400s");
128
129 yield* stack.destroy();
130
131 const gone = yield* waitUntilGone(created.subscription.name);
132 expect(gone).toEqual("gone");
133 }).pipe(logLevel),
134 { tags: ["provider:gcp", "provider:gcp:pubsub", "live"], timeout: 90_000 },
135);
136
137test.provider(
138 "dead-letters a message after maxDeliveryAttempts",
139 (stack) =>
140 Effect.gen(function* () {
141 yield* stack.destroy();
142
143 const out = yield* stack.deploy(
144 Effect.gen(function* () {
145 const topic = yield* GCP.PubSub.Topic("DlqSource", {});
146 const deadLetters = yield* GCP.PubSub.Topic("DlqTarget", {});
147 const inbox = yield* GCP.PubSub.Subscription("DlqInbox", {
148 topic: deadLetters.name,
149 });
150 const subscription = yield* GCP.PubSub.Subscription("DlqWork", {
151 topic: topic.name,
152 deadLetterPolicy: {
153 deadLetterTopic: deadLetters.name,
154 maxDeliveryAttempts: 5,
155 },
156 });
157 return {
158 topic: topic.name,
159 deadLetters: deadLetters.name,
160 inbox: inbox.name,
161 subscription: subscription.name,
162 };
163 }),
164 );
165
166 // The Pub/Sub service agent forwards dead letters, so it must be
167 // able to publish to the target and ack on the source subscription.
168 const topicPolicy = yield* pubsub.getIamPolicyProjectsTopics({
169 resource: out.deadLetters,
170 });
171 const agent = (topicPolicy.bindings ?? [])
172 .find((binding) => binding.role === "roles/pubsub.publisher")
173 ?.members?.find((member) => member.includes("@gcp-sa-pubsub."));
174 expect(agent).toBeDefined();
175 const subscriptionPolicy =
176 yield* pubsub.getIamPolicyProjectsSubscriptions({
177 resource: out.subscription,
178 });
179 expect(
180 (subscriptionPolicy.bindings ?? []).some(
181 (binding) =>
182 binding.role === "roles/pubsub.subscriber" &&
183 binding.members?.includes(agent!),
184 ),
185 ).toBe(true);
186
187 yield* pubsub.publishProjectsTopics({
188 topic: out.topic,
189 body: { messages: [{ data: btoa("poison") }] },
190 });
191
192 // Pull and nack until Pub/Sub gives up on the message.
193 const nackOnce = pubsub
194 .pullProjectsSubscriptions({
195 subscription: out.subscription,
196 body: { maxMessages: 1, returnImmediately: true },
197 })
198 .pipe(
199 Effect.tap(({ receivedMessages = [] }) => {
200 const ackIds = receivedMessages.flatMap((m) =>
201 m.ackId ? [m.ackId] : [],
202 );
203 return ackIds.length === 0
204 ? Effect.void
205 : pubsub.modifyAckDeadlineProjectsSubscriptions({
206 subscription: out.subscription,
207 body: { ackIds, ackDeadlineSeconds: 0 },
208 });
209 }),
210 );
211 const deadLetter = pubsub
212 .pullProjectsSubscriptions({
213 subscription: out.inbox,
214 body: { maxMessages: 1, returnImmediately: true },
215 })
216 .pipe(Effect.map(({ receivedMessages = [] }) => receivedMessages[0]));
217
218 const letter = yield* nackOnce.pipe(
219 Effect.andThen(deadLetter),
220 Effect.repeat({
221 schedule: Schedule.spaced("2 seconds"),
222 until: (message) => message !== undefined,
223 times: 60,
224 }),
225 );
226 expect(atob(letter!.message?.data ?? "")).toEqual("poison");
227 expect(
228 letter!.message?.attributes?.CloudPubSubDeadLetterSourceSubscription,
229 ).toEqual(out.subscription.split("/").pop());
230
231 yield* stack.destroy();
232 expect(yield* waitUntilGone(out.subscription)).toEqual("gone");
233 expect(yield* waitUntilGone(out.inbox)).toEqual("gone");
234 }).pipe(logLevel),
235 { tags: ["provider:gcp", "provider:gcp:pubsub", "live"], timeout: 240_000 },
236);