1import * as GCP from "@/GCP";
2import { makeObjectMedia } from "@/GCP/Storage/ObjectMedia.ts";
3import * as Test from "@/Test/Alchemy";
4import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
5import { expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7import { MinimumLogLevel } from "effect/References";
8import * as Schedule from "effect/Schedule";
9import { spawnSync } from "node:child_process";
10import {
11 Markers,
12 markerFor,
13 PullConsumer,
14 PullOrders,
15 PushConsumer,
16 PushOrders,
17} from "./fixtures/consumers.ts";
18
19const { test } = Test.make({ providers: GCP.providers() });
20
21const logLevel = Effect.provideService(
22 MinimumLogLevel,
23 process.env.DEBUG ? "Debug" : "Info",
24);
25
26const dockerAvailable = (() => {
27 try {
28 return (
29 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 })
30 .status === 0
31 );
32 } catch {
33 return false;
34 }
35})();
36
37// Pub/Sub reads are eventually consistent; ride out a lagging replica.
38const readSubscription = (subscription: string) =>
39 pubsub.getProjectsSubscriptions({ subscription }).pipe(
40 Effect.retry({
41 while: (error) => error._tag === "NotFound",
42 schedule: Schedule.spaced("2 seconds"),
43 times: 15,
44 }),
45 );
46
47const subscriptionStatus = (subscription: string) =>
48 pubsub.getProjectsSubscriptions({ subscription }).pipe(
49 Effect.as("found" as const),
50 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
51 );
52
53const publishAndAwaitMarker = (topic: string, bucket: string, data: string) =>
54 Effect.gen(function* () {
55 const { messageIds = [] } = yield* pubsub.publishProjectsTopics({
56 topic,
57 body: { messages: [{ data: btoa(data), attributes: { kind: "test" } }] },
58 });
59 expect(messageIds.length).toEqual(1);
60 const media = yield* makeObjectMedia;
61 const marker = yield* media
62 .download({ bucket, object: markerFor(messageIds[0]!) })
63 .pipe(
64 Effect.retry({
65 while: (error) => error._tag === "GCP.Storage.ObjectNotFound",
66 schedule: Schedule.spaced("5 seconds"),
67 times: 36,
68 }),
69 );
70 return JSON.parse(new TextDecoder().decode(marker.body)) as {
71 data: string;
72 attributes: Record<string, string>;
73 subscription: string;
74 };
75 });
76
77test.provider.skipIf(!dockerAvailable)(
78 "topic messages are pushed to an effect-native Function",
79 (stack) =>
80 Effect.gen(function* () {
81 yield* stack.destroy();
82
83 const out = yield* stack.deploy(
84 Effect.gen(function* () {
85 const service = yield* PushConsumer;
86 const topic = yield* PushOrders;
87 const bucket = yield* Markers;
88 return {
89 uri: service.uri,
90 topic: topic.name,
91 bucket: bucket.bucketName,
92 };
93 }),
94 );
95
96 const { subscriptions = [] } =
97 yield* pubsub.listProjectsTopicsSubscriptions({ topic: out.topic });
98 expect(subscriptions.length).toEqual(1);
99 const subscription = yield* readSubscription(subscriptions[0]!);
100 const pushEndpoint = `${out.uri}/__alchemy/pubsub/pushorders`;
101 expect(subscription.pushConfig?.pushEndpoint).toEqual(pushEndpoint);
102 expect(subscription.pushConfig?.oidcToken?.audience).toEqual(
103 pushEndpoint,
104 );
105
106 const event = yield* publishAndAwaitMarker(out.topic, out.bucket, "push");
107 expect(event).toEqual({
108 data: "push",
109 attributes: { kind: "test" },
110 subscription: subscriptions[0]!,
111 });
112
113 yield* stack.destroy();
114
115 expect(yield* subscriptionStatus(subscriptions[0]!)).toEqual("gone");
116 }).pipe(logLevel),
117 { tags: ["provider:gcp", "provider:gcp:pubsub", "live"], timeout: 600_000 },
118);
119
120test.provider.skipIf(!dockerAvailable)(
121 "topic messages are pulled by an effect-native WorkerPool",
122 (stack) =>
123 Effect.gen(function* () {
124 yield* stack.destroy();
125
126 const out = yield* stack.deploy(
127 Effect.gen(function* () {
128 yield* PullConsumer;
129 const topic = yield* PullOrders;
130 const bucket = yield* Markers;
131 return { topic: topic.name, bucket: bucket.bucketName };
132 }),
133 );
134
135 const { subscriptions = [] } =
136 yield* pubsub.listProjectsTopicsSubscriptions({ topic: out.topic });
137 expect(subscriptions.length).toEqual(1);
138 const subscription = yield* readSubscription(subscriptions[0]!);
139 expect(subscription.pushConfig?.pushEndpoint).toBeUndefined();
140
141 const event = yield* publishAndAwaitMarker(out.topic, out.bucket, "pull");
142 expect(event).toEqual({
143 data: "pull",
144 attributes: { kind: "test" },
145 subscription: subscriptions[0]!,
146 });
147
148 yield* stack.destroy();
149
150 expect(yield* subscriptionStatus(subscriptions[0]!)).toEqual("gone");
151 }).pipe(logLevel),
152 { tags: ["provider:gcp", "provider:gcp:pubsub", "live"], timeout: 600_000 },
153);