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 * as storage from "@distilled.cloud/gcp/storage_v1";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10import { spawnSync } from "node:child_process";
11import BucketEventsService, {
12 INCOMING_PREFIX,
13 markerFor,
14 Uploads,
15} from "./fixtures/bucket-events.ts";
16
17const { test } = Test.make({ providers: GCP.providers() });
18
19const logLevel = Effect.provideService(
20 MinimumLogLevel,
21 process.env.DEBUG ? "Debug" : "Info",
22);
23
24const dockerAvailable = (() => {
25 try {
26 return (
27 spawnSync("docker", ["info"], { stdio: "ignore", timeout: 15_000 })
28 .status === 0
29 );
30 } catch {
31 return false;
32 }
33})();
34
35test.provider.skipIf(!dockerAvailable)(
36 "bucket finalize events are pushed to an effect-native Function",
37 (stack) =>
38 Effect.gen(function* () {
39 yield* stack.destroy();
40
41 const out = yield* stack.deploy(
42 Effect.gen(function* () {
43 const service = yield* BucketEventsService;
44 const bucket = yield* Uploads;
45 return { uri: service.uri, bucket: bucket.bucketName };
46 }),
47 );
48 const bucket = out.bucket;
49
50 // The notification publishes only `incoming/` finalizes to a topic
51 // with one push subscription to the service.
52 const notifications = (yield* storage.listNotifications({ bucket }))
53 .items;
54 expect(notifications?.length).toEqual(1);
55 const notification = notifications![0]!;
56 expect(notification.event_types).toEqual(["OBJECT_FINALIZE"]);
57 expect(notification.object_name_prefix).toEqual(INCOMING_PREFIX);
58 expect(notification.payload_format).toEqual("JSON_API_V1");
59 const topic = notification.topic!.replace("//pubsub.googleapis.com/", "");
60 const { subscriptions = [] } =
61 yield* pubsub.listProjectsTopicsSubscriptions({ topic });
62 expect(subscriptions.length).toEqual(1);
63 const subscription = yield* pubsub.getProjectsSubscriptions({
64 subscription: subscriptions[0]!,
65 });
66 const pushEndpoint = `${out.uri}/__alchemy/pubsub/uploads-bucketevents`;
67 expect(subscription.pushConfig?.pushEndpoint).toEqual(pushEndpoint);
68 expect(subscription.pushConfig?.oidcToken?.audience).toEqual(
69 pushEndpoint,
70 );
71
72 const media = yield* makeObjectMedia;
73 // Outside the prefix: must not be delivered.
74 yield* media.upload(bucket, { name: "other/skip.txt", body: "skip" });
75 const uploaded = yield* media.upload(bucket, {
76 name: `${INCOMING_PREFIX}hello.txt`,
77 body: "Hello, events!",
78 });
79
80 const marker = yield* media
81 .download({ bucket, object: markerFor(`${INCOMING_PREFIX}hello.txt`) })
82 .pipe(
83 Effect.retry({
84 while: (error) => error._tag === "GCP.Storage.ObjectNotFound",
85 schedule: Schedule.spaced("5 seconds"),
86 times: 36,
87 }),
88 );
89 const event = JSON.parse(new TextDecoder().decode(marker.body));
90 expect(event).toMatchObject({
91 eventType: "OBJECT_FINALIZE",
92 bucket,
93 object: `${INCOMING_PREFIX}hello.txt`,
94 generation: uploaded.generation,
95 size: "14",
96 });
97 expect(event.contentType).toContain("text/plain");
98 expect(event.eventTime).toEqual(expect.any(String));
99
100 const skipped = yield* media
101 .download({ bucket, object: markerFor("other/skip.txt") })
102 .pipe(
103 Effect.as("delivered" as const),
104 Effect.catchTag("GCP.Storage.ObjectNotFound", () =>
105 Effect.succeed("filtered" as const),
106 ),
107 );
108 expect(skipped).toEqual("filtered");
109
110 yield* stack.destroy();
111
112 const topicGone = yield* pubsub.getProjectsTopics({ topic }).pipe(
113 Effect.as("found" as const),
114 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
115 );
116 expect(topicGone).toEqual("gone");
117 const subscriptionGone = yield* pubsub
118 .getProjectsSubscriptions({ subscription: subscriptions[0]! })
119 .pipe(
120 Effect.as("found" as const),
121 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
122 );
123 expect(subscriptionGone).toEqual("gone");
124 // The bucket (and with it the notification) is gone.
125 const notificationsGone = yield* storage
126 .getNotifications({ bucket, notification: notification.id! })
127 .pipe(
128 Effect.as("found" as const),
129 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
130 );
131 expect(notificationsGone).toEqual("gone");
132 }).pipe(logLevel),
133 { tags: ["provider:gcp", "provider:gcp:storage", "live"], timeout: 600_000 },
134);