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 AVRO = JSON.stringify({
24 type: "record",
25 name: "Order",
26 namespace: "alchemy",
27 fields: [{ name: "id", type: "string" }],
28});
29
30const waitUntilGone = (name: string) =>
31 kafka.getProjectsLocationsSchemaRegistriesSubjectsVersions({ name }).pipe(
32 Effect.as("found" as const),
33 // Once the parent registry is deleted too, the path itself is invalid.
34 Effect.catchTag(
35 [
36 "NotFound",
37 "SchemaRegistryRequiresCluster",
38 "SchemaRegistryPathNotFound",
39 ],
40 () => Effect.succeed("gone" as const),
41 ),
42 Effect.repeat({
43 schedule: Schedule.spaced("1 second"),
44 until: (status) => status === "gone",
45 times: 10,
46 }),
47 );
48
49test.provider(
50 "getProjectsLocationsSchemaRegistriesSubjectsVersions on a missing version fails with a typed tag",
51 (stack) =>
52 Effect.gen(function* () {
53 const { project } = yield* GcpEnvironment.current;
54
55 yield* stack.destroy();
56
57 const error = yield* Effect.flip(
58 kafka.getProjectsLocationsSchemaRegistriesSubjectsVersions({
59 name: `projects/${project}/locations/us-central1/schemaRegistries/alchemy_missing/subjects/missing/versions/1`,
60 }),
61 );
62 // Without a cluster in the region every call is refused; with one
63 // (other suites create them concurrently) the registry is just missing.
64 expect(["SchemaRegistryRequiresCluster", "NotFound"]).toContain(
65 error._tag,
66 );
67
68 yield* stack.destroy();
69 }).pipe(logLevel),
70 {
71 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
72 timeout: 90_000,
73 },
74);
75
76test.provider.skipIf(!runLifecycle)(
77 "create and delete a schema subject version",
78 (stack) =>
79 Effect.gen(function* () {
80 yield* stack.destroy();
81
82 const created = yield* stack.deploy(
83 Effect.gen(function* () {
84 const cluster = yield* GCP.ManagedKafka.Cluster("Brokers", {
85 location: "us-central1",
86 });
87 const registry = yield* GCP.ManagedKafka.SchemaRegistry("Schemas", {
88 location: cluster.location,
89 });
90 const version =
91 yield* GCP.ManagedKafka.SchemaRegistriesSubjectsVersion(
92 "OrderSchema",
93 {
94 schemaRegistry: registry.name,
95 subject: "orders",
96 schemaType: "AVRO",
97 schema: AVRO,
98 },
99 );
100 return { registry, version };
101 }),
102 );
103
104 expect(created.version.name).toContain("/subjects/");
105 expect(created.version.subject).toEqual("orders");
106 expect(created.version.version).toBeGreaterThan(0);
107
108 const fetched =
109 yield* kafka.getProjectsLocationsSchemaRegistriesSubjectsVersions({
110 name: created.version.name,
111 });
112 expect(fetched.subject).toEqual("orders");
113 expect(fetched.schema).toContain("Order");
114
115 yield* stack.destroy();
116
117 const gone = yield* waitUntilGone(created.version.name);
118 expect(gone).toEqual("gone");
119 }).pipe(logLevel, withKafkaClusterSlot),
120 {
121 tags: ["provider:gcp", "provider:gcp:managedkafka", "live"],
122 timeout: 10_800_000,
123 retry: 0,
124 },
125);