1import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import type { Providers } from "../Providers.ts";
11
12export type SchemaType = "AVRO" | "PROTOCOL_BUFFER";
13
14export type SchemaProps = {
15 /**
16 * Schema id (the `{schema}` segment of `projects/{project}/schemas/{schema}`).
17 * If omitted, a unique name is generated from the stack, stage, and
18 * logical id. Must start with a letter and be 3-255 characters.
19 * Immutable — changing it replaces the schema.
20 */
21 schemaId?: string;
22 /**
23 * Schema definition language. Immutable — changing it replaces the schema.
24 */
25 type: SchemaType;
26 /**
27 * Schema definition. Must be a valid Avro JSON schema or Protocol Buffer
28 * file matching `type`. Updates commit a new revision (Pub/Sub allows
29 * up to 20 revisions per schema).
30 */
31 definition: string;
32};
33
34export type Schema = Resource<
35 "GCP.PubSub.Schema",
36 SchemaProps,
37 {
38 /** Full resource name `projects/{project}/schemas/{schema}`. */
39 name: string;
40 /** Schema id (last path segment). */
41 schemaId: string;
42 /** Project id. */
43 project: string;
44 /** Schema definition language. */
45 type: string | undefined;
46 /** Schema definition (latest revision). */
47 definition: string | undefined;
48 /** Latest revision id. */
49 revisionId: string | undefined;
50 /** RFC3339 timestamp when the latest revision was created. */
51 revisionCreateTime: string | undefined;
52 },
53 never,
54 Providers
55>;
56
57/**
58 * A Google Cloud Pub/Sub schema used to validate published messages.
59 *
60 * Pub/Sub schemas have no labels field. Ownership is by resource name;
61 * `list` returns every schema in the project so `pnpm nuke:gcp` can
62 * clean leaked test schemas.
63 *
64 * ### Creating a Schema
65 * **Example:** Generated name (Avro)
66 * ```typescript
67 * const schema = yield* GCP.PubSub.Schema("Events", {
68 * type: "AVRO",
69 * definition: JSON.stringify({
70 * type: "record",
71 * name: "Event",
72 * fields: [{ name: "id", type: "string" }],
73 * }),
74 * });
75 * ```
76 *
77 * **Example:** Explicit id and Protocol Buffer definition
78 * ```typescript
79 * const schema = yield* GCP.PubSub.Schema("Events", {
80 * schemaId: "order-events",
81 * type: "PROTOCOL_BUFFER",
82 * definition: 'syntax = "proto3";\nmessage Event { string id = 1; }',
83 * });
84 * ```
85 *
86 * ### Updating a Schema
87 * Change `definition` on the same logical id; the schema keeps its name and
88 * a new revision is committed.
89 *
90 * **Example:** Commit a compatible Avro revision
91 * ```typescript
92 * const schema = yield* GCP.PubSub.Schema("Events", {
93 * type: "AVRO",
94 * definition: JSON.stringify({
95 * type: "record",
96 * name: "Event",
97 * fields: [
98 * { name: "id", type: "string" },
99 * { name: "count", type: "int", default: 0 },
100 * ],
101 * }),
102 * });
103 * ```
104 *
105 * @resource
106 * @category PubSub
107 */
108export const Schema = Resource<Schema>("GCP.PubSub.Schema");
109
110export class SchemaNotResolved extends Data.TaggedError(
111 "GCP.PubSub.SchemaNotResolved",
112)<{
113 name: string;
114}> {}
115
116const canonicalName = (name: string) => name.split("@")[0] ?? name;
117
118const schemaIdOf = (name: string) => {
119 const withoutRevision = canonicalName(name);
120 return withoutRevision.split("/").pop() ?? withoutRevision;
121};
122
123const resourceName = (project: string, schemaId: string) =>
124 `projects/${project}/schemas/${schemaId}`;
125
126const toId = (id: string, schemaId: string | undefined, existing?: string) =>
127 Effect.gen(function* () {
128 if (schemaId !== undefined) return schemaId;
129 if (existing !== undefined) return existing;
130 const generated = yield* createPhysicalName({
131 id,
132 maxLength: 255,
133 lowercase: true,
134 });
135 return /^[a-z]/.test(generated) ? generated : `s${generated}`.slice(0, 255);
136 });
137
138const toAttrs = (schema: pubsub.Pubsub_Schema, project: string) => {
139 const name = canonicalName(schema.name ?? "");
140 return {
141 name,
142 schemaId: schemaIdOf(name),
143 project,
144 type: schema.type,
145 definition: schema.definition,
146 revisionId: schema.revisionId,
147 revisionCreateTime: schema.revisionCreateTime,
148 };
149};
150
151const getByName = (name: string) =>
152 pubsub
153 .getProjectsSchemas({ name: canonicalName(name), view: "FULL" })
154 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
155
156const definitionsEqual = (
157 left: string | undefined,
158 right: string | undefined,
159) => {
160 const a = (left ?? "").trim();
161 const b = (right ?? "").trim();
162 if (a === b) return true;
163 try {
164 return JSON.stringify(JSON.parse(a)) === JSON.stringify(JSON.parse(b));
165 } catch {
166 return false;
167 }
168};
169
170export const SchemaProvider = () =>
171 Provider.succeed(Schema, {
172 stables: ["name", "schemaId", "project"],
173
174 diff: Effect.fn(function* ({ news, olds, output }) {
175 if (!isResolved(news)) return undefined;
176 const previousId = olds?.schemaId ?? output?.schemaId;
177 const nextId = news.schemaId ?? previousId;
178 const nameChanged =
179 previousId !== undefined &&
180 news.schemaId !== undefined &&
181 news.schemaId !== previousId;
182 const previousType = olds?.type ?? output?.type;
183 const typeChanged =
184 previousType !== undefined && previousType !== news.type;
185 if (nameChanged) {
186 return { action: "replace" as const, deleteFirst: false };
187 }
188 if (typeChanged) {
189 return {
190 action: "replace" as const,
191 deleteFirst: nextId !== undefined && nextId === previousId,
192 };
193 }
194 return undefined;
195 }),
196
197 read: Effect.fn(function* ({ id, olds, output }) {
198 const env = yield* GcpEnvironment.current;
199 const schemaId = yield* toId(id, olds?.schemaId, output?.schemaId);
200 const name = canonicalName(
201 output?.name ?? resourceName(env.project, schemaId),
202 );
203 const existing = yield* getByName(name);
204 if (existing === undefined) return undefined;
205 return toAttrs(existing, env.project);
206 }),
207
208 list: () =>
209 Effect.gen(function* () {
210 const env = yield* GcpEnvironment.current;
211 return yield* pubsub.listProjectsSchemas
212 .pages({
213 parent: `projects/${env.project}`,
214 pageSize: 1000,
215 view: "FULL",
216 })
217 .pipe(
218 Stream.flatMap((page) => Stream.fromIterable(page.schemas ?? [])),
219 Stream.map((schema) => toAttrs(schema, env.project)),
220 Stream.runCollect,
221 Effect.map((chunk) => Array.from(chunk)),
222 );
223 }),
224
225 reconcile: Effect.fn(function* ({ id, news, output }) {
226 const env = yield* GcpEnvironment.current;
227 const schemaId = yield* toId(id, news.schemaId, output?.schemaId);
228 const name = resourceName(env.project, schemaId);
229
230 let current = yield* getByName(name);
231
232 if (current === undefined) {
233 const created = yield* pubsub
234 .createProjectsSchemas({
235 parent: `projects/${env.project}`,
236 schemaId,
237 body: {
238 type: news.type,
239 definition: news.definition,
240 },
241 })
242 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
243 current = created ?? undefined;
244 }
245
246 if (current === undefined) {
247 return yield* new SchemaNotResolved({ name });
248 }
249
250 if (!definitionsEqual(current.definition, news.definition)) {
251 current = yield* pubsub.commitProjectsSchemas({
252 name,
253 body: {
254 schema: {
255 name,
256 type: news.type,
257 definition: news.definition,
258 },
259 },
260 });
261 }
262
263 return toAttrs(current, env.project);
264 }),
265
266 delete: Effect.fn(function* ({ output }) {
267 yield* pubsub
268 .deleteProjectsSchemas({ name: canonicalName(output.name) })
269 .pipe(Effect.catchTag("NotFound", () => Effect.void));
270 }),
271 });