1import * as GCP from "@/GCP";
2import * as Test from "@/Test/Alchemy";
3import * as kafka from "@distilled.cloud/gcp/managedkafka_v1";
4import { expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import { GcpEnvironment } from "@/GCP/Environment";
9import { withKafkaClusterSlot } from "./quota.ts";
10
11const { test } = Test.make({ providers: GCP.providers() });
12
13// The Schema Registry API refuses every call until a Kafka cluster exists in
14// the region (`SchemaRegistryRequiresCluster`), so lifecycle tests deploy a
15// cluster first and make the registry depend on it. Clusters take ~30 minutes.
16const runLifecycle = !!process.env.GCP_TEST_SLOW && !process.env.FAST;
17
18const logLevel = Effect.provideService(
19 MinimumLogLevel,
20 process.env.DEBUG ? "Debug" : "Info",
21);
22
23const waitUntilGone = (name: string) =>
24 kafka.getProjectsLocationsSchemaRegistries({ name }).pipe(
25 Effect.as("found" as const),
26 Effect.catchTag(["NotFound", "SchemaRegistryRequiresCluster"], () =>
27 Effect.succeed("gone" as const),
28 ),
29 Effect.repeat({
30 schedule: Schedule.spaced("1 second"),
31 until: (status) => status === "gone",
32 times: 10,
33 }),
34 );
35
36test.provider(
37 "getProjectsLocationsSchemaRegistries on a missing registry fails with a typed tag",
38 (stack) =>
39 Effect.gen(function* () {
40 const { project } = yield* GcpEnvironment.current;
41
42 yield* stack.destroy();
43
44 const error = yield* Effect.flip(
45 kafka.getProjectsLocationsSchemaRegistries({
46 name: `projects/${project}/locations/us-central1/schemaRegistries/alchemy_missing_registry`,
47 }),
48 );
49 // Without a cluster in the region every call is refused; with one
50 // (other suites create them concurrently) the registry is just missing.
51 expect(["SchemaRegistryRequiresCluster", "NotFound"]).toContain(
52 error._tag,
53 );
54
55 yield* stack.destroy();
56 }).pipe(logLevel),
57 {
58 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
59 timeout: 90_000,
60 },
61);
62
63test.provider.skipIf(!runLifecycle)(
64 "create, refresh, and delete a schema registry",
65 (stack) =>
66 Effect.gen(function* () {
67 yield* stack.destroy();
68
69 const deployed = yield* stack.deploy(
70 Effect.gen(function* () {
71 const cluster = yield* GCP.ManagedKafka.Cluster("Brokers", {
72 location: "us-central1",
73 });
74 const registry = yield* GCP.ManagedKafka.SchemaRegistry("Schemas", {
75 location: cluster.location,
76 });
77 return { cluster, registry };
78 }),
79 );
80 const created = deployed.registry;
81
82 expect(created.name).toContain("/schemaRegistries/");
83 expect(created.schemaRegistryId).toEqual(expect.any(String));
84 expect(created.location).toEqual("us-central1");
85
86 const fetched = yield* kafka.getProjectsLocationsSchemaRegistries({
87 name: created.name,
88 });
89 expect(fetched.name).toEqual(created.name);
90
91 const updated = yield* stack.deploy(
92 Effect.gen(function* () {
93 const cluster = yield* GCP.ManagedKafka.Cluster("Brokers", {
94 clusterId: deployed.cluster.clusterId,
95 location: "us-central1",
96 });
97 return yield* GCP.ManagedKafka.SchemaRegistry("Schemas", {
98 schemaRegistryId: created.schemaRegistryId,
99 location: cluster.location,
100 });
101 }),
102 );
103 expect(updated.name).toEqual(created.name);
104
105 yield* stack.destroy();
106
107 const gone = yield* waitUntilGone(created.name);
108 expect(gone).toEqual("gone");
109 }).pipe(logLevel, withKafkaClusterSlot),
110 {
111 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
112 timeout: 10_800_000,
113 retry: 0,
114 },
115);