1import * as GCP from "@/GCP";
2import { GcpEnvironment } from "@/GCP/Environment";
3import * as Test from "@/Test/Alchemy";
4import * as Core from "@/Test/Core";
5import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
6import * as resourcemanager from "@distilled.cloud/gcp/cloudresourcemanager_v3";
7import { describe, expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import { callProbe, dockerAvailable, expectProbe } from "../bindingHost.ts";
11import PubSubBindingsHost, {
12 EventSchema,
13 PUBLISH_PAYLOAD,
14 PublishTap,
15 PublishTopic,
16 PullInbox,
17 PullTopic,
18 ReadInbox,
19 ReadTopic,
20 WRITE_BATCH_BYTES,
21 WRITE_BATCH_TEXT,
22 WRITE_PAYLOAD,
23 WriteTap,
24 WriteTopicTarget,
25} from "./fixtures/bindings-host.ts";
26
27const testOptions = { providers: GCP.providers() };
28const { test, beforeAll, afterAll } = Test.make(testOptions);
29const sharedStack = Core.scratchStack(testOptions, "PubSubBindings");
30
31let baseUrl: string;
32let member: string;
33let names: {
34 publishTopic: string;
35 publishTap: string;
36 writeTopic: string;
37 writeTap: string;
38 pullTopic: string;
39 pullInbox: string;
40 readTopic: string;
41 readInbox: string;
42 schema: string;
43};
44
45const decode = (data: string | undefined) =>
46 Buffer.from(data ?? "", "base64").toString("utf8");
47
48/** Roles `member` holds on a resource's IAM policy. */
49const rolesOf = (
50 policy: { bindings?: { role?: string; members?: string[] }[] },
51 member: string,
52) =>
53 (policy.bindings ?? [])
54 .filter((binding) => (binding.members ?? []).includes(member))
55 .map((binding) => binding.role)
56 .sort();
57
58/** Pull (out of band) until `count` messages arrived, acknowledging them. */
59const drain = (subscription: string, count: number) =>
60 Effect.gen(function* () {
61 const seen: pubsub.ReceivedMessage[] = [];
62 yield* pubsub
63 .pullProjectsSubscriptions({
64 subscription,
65 body: { maxMessages: 10, returnImmediately: false },
66 })
67 .pipe(
68 Effect.tap((page) =>
69 Effect.gen(function* () {
70 const received = page.receivedMessages ?? [];
71 seen.push(...received);
72 if (received.length > 0) {
73 yield* pubsub.acknowledgeProjectsSubscriptions({
74 subscription,
75 body: { ackIds: received.map((m) => m.ackId ?? "") },
76 });
77 }
78 }),
79 ),
80 Effect.repeat({
81 schedule: Schedule.spaced("1 second"),
82 until: () => seen.length >= count,
83 times: 30,
84 }),
85 );
86 return seen;
87 });
88
89/** Publish (out of band, as the deployer) one message. */
90const publishOutOfBand = (topic: string, text: string) =>
91 pubsub.publishProjectsTopics({
92 topic,
93 body: { messages: [{ data: Buffer.from(text).toString("base64") }] },
94 });
95
96/** Nothing is redelivered once the ack deadline (10s) has passed. */
97const expectNoRedelivery = (subscription: string) =>
98 Effect.gen(function* () {
99 yield* Effect.sleep("15 seconds");
100 const page = yield* pubsub.pullProjectsSubscriptions({
101 subscription,
102 body: { maxMessages: 10, returnImmediately: true },
103 });
104 expect(page.receivedMessages ?? []).toEqual([]);
105 });
106
107describe.skipIf(!dockerAvailable)(
108 "PubSub Bindings",
109 { tags: ["provider:gcp", "provider:gcp:pubsub", "provider:gcp:run", "live"] },
110 () => {
111 beforeAll(
112 Effect.gen(function* () {
113 yield* sharedStack.destroy();
114 const out = yield* sharedStack.deploy(
115 Effect.gen(function* () {
116 const host = yield* PubSubBindingsHost;
117 return {
118 uri: host.uri,
119 serviceAccount: host.serviceAccount,
120 publishTopic: (yield* PublishTopic).name,
121 publishTap: (yield* PublishTap).name,
122 writeTopic: (yield* WriteTopicTarget).name,
123 writeTap: (yield* WriteTap).name,
124 pullTopic: (yield* PullTopic).name,
125 pullInbox: (yield* PullInbox).name,
126 readTopic: (yield* ReadTopic).name,
127 readInbox: (yield* ReadInbox).name,
128 schema: (yield* EventSchema).name,
129 };
130 }),
131 );
132 baseUrl = out.uri!;
133 member = `serviceAccount:${out.serviceAccount!}`;
134 names = out;
135 }),
136 { timeout: 900_000 },
137 );
138
139 afterAll(sharedStack.destroy(), { timeout: 600_000 });
140
141 describe("Publish", () => {
142 test.provider(
143 "publishes to the topic, granted publisher on the topic only",
144 (_stack) =>
145 Effect.gen(function* () {
146 const out = yield* expectProbe<{ messageIds?: string[] }>(
147 baseUrl,
148 "publish",
149 );
150 expect(out.messageIds).toHaveLength(1);
151
152 const [message] = yield* drain(names.publishTap, 1);
153 expect(message?.message?.messageId).toEqual(out.messageIds?.[0]);
154 expect(decode(message?.message?.data)).toEqual(PUBLISH_PAYLOAD);
155 expect(message?.message?.attributes).toEqual({ via: "Publish" });
156
157 const policy = yield* pubsub.getIamPolicyProjectsTopics({
158 resource: names.publishTopic,
159 });
160 expect(rolesOf(policy, member)).toEqual(["roles/pubsub.publisher"]);
161 }),
162 {
163 tags: ["provider:gcp", "provider:gcp:pubsub", "live"],
164 timeout: 600_000,
165 },
166 );
167 });
168
169 describe("Pull / Acknowledge", () => {
170 test.provider(
171 "pulls and acknowledges, granted subscriber on the subscription only",
172 (_stack) =>
173 Effect.gen(function* () {
174 yield* publishOutOfBand(names.pullTopic, "pull-me");
175 const texts = yield* expectProbe<string[]>(
176 baseUrl,
177 "pullAndAcknowledge",
178 );
179 expect(texts).toEqual(["pull-me"]);
180 yield* expectNoRedelivery(names.pullInbox);
181
182 const policy = yield* pubsub.getIamPolicyProjectsSubscriptions({
183 resource: names.pullInbox,
184 });
185 expect(rolesOf(policy, member)).toEqual([
186 "roles/pubsub.subscriber",
187 ]);
188 }),
189 {
190 tags: ["provider:gcp", "provider:gcp:pubsub", "live"],
191 timeout: 600_000,
192 },
193 );
194 });
195
196 describe("WriteTopic", () => {
197 test.provider(
198 "publish and publishBatch, granted publisher on the topic only",
199 (_stack) =>
200 Effect.gen(function* () {
201 const out = yield* expectProbe<{
202 id: string;
203 batch: string[];
204 empty: string[];
205 }>(baseUrl, "writeTopic");
206 expect(out.batch).toHaveLength(2);
207 // An empty batch is a no-op, not a request.
208 expect(out.empty).toEqual([]);
209
210 // Out of band: every message landed with its payload intact.
211 const received = yield* drain(names.writeTap, 3);
212 const byId = new Map(
213 received.map((m) => [m.message?.messageId, m.message]),
214 );
215 expect(decode(byId.get(out.id)?.data)).toEqual(WRITE_PAYLOAD);
216 expect(byId.get(out.id)?.attributes).toEqual({ via: "WriteTopic" });
217 expect(decode(byId.get(out.batch[0])?.data)).toEqual(
218 WRITE_BATCH_TEXT,
219 );
220 expect([
221 ...Buffer.from(byId.get(out.batch[1])?.data ?? "", "base64"),
222 ]).toEqual(WRITE_BATCH_BYTES);
223
224 const policy = yield* pubsub.getIamPolicyProjectsTopics({
225 resource: names.writeTopic,
226 });
227 expect(rolesOf(policy, member)).toEqual(["roles/pubsub.publisher"]);
228 }),
229 {
230 tags: ["provider:gcp", "provider:gcp:pubsub", "live"],
231 timeout: 600_000,
232 },
233 );
234 });
235
236 describe("ReadSubscription", () => {
237 test.provider(
238 "pull, modifyAckDeadline and acknowledge, granted subscriber on the subscription only",
239 (_stack) =>
240 Effect.gen(function* () {
241 yield* publishOutOfBand(names.readTopic, "read-me");
242 const messages = yield* expectProbe<
243 { text: string; attributes: Record<string, string> }[]
244 >(baseUrl, "readSubscription");
245 expect(messages).toEqual([{ text: "read-me", attributes: {} }]);
246 yield* expectNoRedelivery(names.readInbox);
247
248 const policy = yield* pubsub.getIamPolicyProjectsSubscriptions({
249 resource: names.readInbox,
250 });
251 expect(rolesOf(policy, member)).toEqual([
252 "roles/pubsub.subscriber",
253 ]);
254 }),
255 {
256 tags: ["provider:gcp", "provider:gcp:pubsub", "live"],
257 timeout: 600_000,
258 },
259 );
260 });
261
262 describe("GetSchema", () => {
263 test.provider(
264 "reads the schema, granted viewer on the schema",
265 (_stack) =>
266 Effect.gen(function* () {
267 const live = yield* expectProbe<pubsub.Pubsub_Schema>(
268 baseUrl,
269 "getSchema",
270 );
271 expect(live.name).toEqual(names.schema);
272 expect(live.type).toEqual("AVRO");
273 expect(JSON.parse(live.definition ?? "{}")).toMatchObject({
274 name: "Event",
275 });
276
277 const policy = yield* pubsub.getIamPolicyProjectsSchemas({
278 resource: names.schema,
279 });
280 expect(rolesOf(policy, member)).toEqual(["roles/pubsub.viewer"]);
281 }),
282 {
283 tags: ["provider:gcp", "provider:gcp:pubsub", "live"],
284 timeout: 600_000,
285 },
286 );
287 });
288
289 describe("ValidateMessage", () => {
290 test.provider(
291 "accepts a conforming message and rejects another, granted viewer on the project",
292 (_stack) =>
293 Effect.gen(function* () {
294 const valid = yield* expectProbe<object>(baseUrl, "validateValid");
295 expect(valid).toEqual({});
296
297 const invalid = yield* callProbe(baseUrl, "validateInvalid");
298 expect(invalid.ok ? "ok" : invalid.error._tag).toEqual(
299 "BadRequest",
300 );
301
302 const { project } = yield* GcpEnvironment.current;
303 const policy = yield* resourcemanager.getIamPolicyProjects({
304 resource: `projects/${project}`,
305 body: { options: { requestedPolicyVersion: 3 } },
306 });
307 // schemas.validateMessage is authorized on the parent project
308 // (a viewer grant on the schema alone answers 403).
309 expect(
310 (policy.bindings ?? [])
311 .filter((binding) => (binding.members ?? []).includes(member))
312 .map((binding) => [binding.role, binding.condition]),
313 ).toEqual([["roles/pubsub.viewer", undefined]]);
314 }),
315 {
316 tags: ["provider:gcp", "provider:gcp:pubsub", "live"],
317 timeout: 600_000,
318 },
319 );
320 });
321 },
322);