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