1import * as datalabeling from "@distilled.cloud/gcp/datalabeling_v1beta1";
2import { Retry as GcpRetry } from "@distilled.cloud/gcp/Retry";
3import * as Data from "effect/Data";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import {
10 alchemyLabelKeys,
11 createInternalLabels,
12 hasAlchemyLabels,
13} from "../Labels.ts";
14import { isTransientGcpError } from "../Errors.ts";
15
16/**
17 * Data Labeling is shut down. The frontend hangs ~20s then returns HTTP
18 * 502 (`BadGateway`), which distilled retries as transient. Disable that
19 * retry so list/get/create fail in one round trip instead of eight.
20 */
21export const noRetryLayer = Layer.succeed(GcpRetry, {
22 while: () => false,
23});
24
25export const MAX_DISPLAY_NAME_LENGTH = 64;
26export const MAX_DESCRIPTION_LENGTH = 10_000;
27export const MAX_EVALUATION_DESCRIPTION_LENGTH = 25_000;
28export const MAX_FEEDBACK_BODY_LENGTH = 10_000;
29
30export class DatalabelingPending extends Data.TaggedError(
31 "GCP.DataLabeling.Pending",
32)<{
33 name: string;
34}> {}
35
36export const lastSegment = (value: string) => {
37 const trimmed = value.replace(/\/+$/, "");
38 const parts = trimmed.split("/");
39 return parts[parts.length - 1] || trimmed;
40};
41
42export const parentOf = (name: string) => {
43 const parts = name.split("/").filter((part) => part.length > 0);
44 return parts.slice(0, -2).join("/");
45};
46
47export const projectParent = (project: string) => `projects/${project}`;
48
49export const parseResourceName = (name: string, collection: string) => {
50 const parts = name.split("/").filter((part) => part.length > 0);
51 const collectionAt = parts.lastIndexOf(collection);
52 const projectsAt = parts.lastIndexOf("projects");
53 const datasetsAt = parts.lastIndexOf("datasets");
54 return {
55 project:
56 projectsAt >= 0 && parts[projectsAt + 1] ? parts[projectsAt + 1]! : "",
57 datasetId:
58 datasetsAt >= 0 && parts[datasetsAt + 1] ? parts[datasetsAt + 1]! : "",
59 dataset:
60 datasetsAt >= 0
61 ? parts.slice(0, datasetsAt + 2).join("/")
62 : parentOf(name),
63 id:
64 collectionAt >= 0 && parts[collectionAt + 1]
65 ? parts[collectionAt + 1]!
66 : lastSegment(name),
67 parent:
68 collectionAt > 0
69 ? parts.slice(0, collectionAt).join("/")
70 : parts.slice(0, Math.max(0, parts.length - 1)).join("/"),
71 };
72};
73
74export const sameText = (left: string | undefined, right: string | undefined) =>
75 (left ?? "") === (right ?? "");
76
77export const canonical = (value: unknown): unknown => {
78 if (value === undefined || value === null) return undefined;
79 if (typeof value === "boolean" || typeof value === "number") return value;
80 if (typeof value === "string") {
81 return value.length === 0 ? undefined : value;
82 }
83 if (Array.isArray(value)) {
84 const items = value.map(canonical).filter((item) => item !== undefined);
85 return items.length === 0 ? undefined : items;
86 }
87 if (typeof value === "object") {
88 const entries = Object.entries(value as Record<string, unknown>)
89 .map(([key, item]) => [key, canonical(item)] as const)
90 .filter(([, item]) => item !== undefined)
91 .sort(([left], [right]) => left.localeCompare(right));
92 if (entries.length === 0) return undefined;
93 return Object.fromEntries(entries);
94 }
95 return undefined;
96};
97
98export const fingerprint = (value: unknown): string =>
99 JSON.stringify(canonical(value) ?? null);
100
101export const sameJson = (left: unknown, right: unknown) =>
102 fingerprint(left) === fingerprint(right);
103
104export const updateMaskOf = (...fields: Array<string | undefined>) =>
105 fields.filter((field): field is string => field !== undefined).join(",");
106
107export const toDisplayName = (
108 id: string,
109 explicit: string | undefined,
110 existing: string | undefined,
111 maxLength = MAX_DISPLAY_NAME_LENGTH,
112) =>
113 Effect.gen(function* () {
114 if (explicit !== undefined) return explicit;
115 if (existing !== undefined) return existing;
116 const generated = yield* createPhysicalName({
117 id,
118 maxLength,
119 lowercase: true,
120 });
121 const next = /^[a-z]/.test(generated)
122 ? generated
123 : `d${generated}`.slice(0, maxLength);
124 return next.length > 0 ? next : "d";
125 });
126
127const markerOf = (
128 labels: Record<string, string>,
129 stack: string,
130 stage: string,
131 id: string,
132) =>
133 `[alchemy ${alchemyLabelKeys.stack}=${stack} ${alchemyLabelKeys.stage}=${stage} ${alchemyLabelKeys.id}=${id}]`;
134
135const shrinkMarker = (labels: Record<string, string>, maxLength: number) => {
136 let stack = labels[alchemyLabelKeys.stack] ?? "x";
137 let stage = labels[alchemyLabelKeys.stage] ?? "x";
138 let id = labels[alchemyLabelKeys.id] ?? "x";
139 let marker = markerOf(labels, stack, stage, id);
140 while (
141 marker.length > maxLength &&
142 (stack.length > 1 || stage.length > 1 || id.length > 1)
143 ) {
144 if (stack.length >= stage.length && stack.length >= id.length) {
145 stack = stack.slice(0, -1);
146 } else if (stage.length >= id.length) {
147 stage = stage.slice(0, -1);
148 } else {
149 id = id.slice(0, -1);
150 }
151 marker = markerOf(labels, stack, stage, id);
152 }
153 return marker.slice(0, maxLength);
154};
155
156export const encodeOwnership = (
157 labels: Record<string, string>,
158 text: string | undefined,
159 maxLength = MAX_DESCRIPTION_LENGTH,
160): string => {
161 const marker = shrinkMarker(labels, maxLength);
162 const trimmed = text?.trim();
163 if (!trimmed) return marker;
164 return `${marker}\n${trimmed}`.slice(0, maxLength);
165};
166
167export const parseOwnership = (
168 text: string | undefined,
169): {
170 labels: Record<string, string>;
171 text: string | undefined;
172} => {
173 if (!text?.startsWith("[alchemy ")) {
174 return { labels: {}, text };
175 }
176 const end = text.indexOf("]");
177 if (end < 0) return { labels: {}, text };
178 const labels: Record<string, string> = {};
179 for (const part of text.slice("[alchemy ".length, end).split(/\s+/)) {
180 const eq = part.indexOf("=");
181 if (eq > 0) {
182 labels[part.slice(0, eq)] = part.slice(eq + 1);
183 }
184 }
185 const rest = text.slice(end + 1).replace(/^[\s\n]+/, "");
186 return { labels, text: rest.length > 0 ? rest : undefined };
187};
188
189export const hasOwnershipMarker = (text: string | undefined) =>
190 Object.keys(parseOwnership(text).labels).some((key) =>
191 key.startsWith("alchemy-"),
192 );
193
194const prefixMatch = (expected: string, observed: string) =>
195 expected === observed ||
196 expected.startsWith(observed) ||
197 observed.startsWith(expected);
198
199export const ownedByAlchemy = (id: string, text: string | undefined) =>
200 Effect.gen(function* () {
201 const expected = yield* createInternalLabels(id);
202 const { labels } = parseOwnership(text);
203 if (!hasOwnershipMarker(text)) return false;
204 const exact = yield* hasAlchemyLabels(id, labels);
205 if (exact) return true;
206 return (
207 prefixMatch(
208 expected[alchemyLabelKeys.stack] ?? "",
209 labels[alchemyLabelKeys.stack] ?? "",
210 ) &&
211 prefixMatch(
212 expected[alchemyLabelKeys.stage] ?? "",
213 labels[alchemyLabelKeys.stage] ?? "",
214 ) &&
215 prefixMatch(
216 expected[alchemyLabelKeys.id] ?? "",
217 labels[alchemyLabelKeys.id] ?? "",
218 )
219 );
220 });
221
222export const replaceOnIdentity = (input: {
223 previousId?: string;
224 nextId?: string;
225 previousParent?: string;
226 nextParent?: string;
227 extra?: boolean;
228}) => {
229 if (input.extra === true) {
230 return { action: "replace" as const, deleteFirst: false };
231 }
232 if (
233 input.previousId !== undefined &&
234 input.nextId !== undefined &&
235 input.previousId !== input.nextId
236 ) {
237 return { action: "replace" as const, deleteFirst: false };
238 }
239 if (
240 input.previousParent !== undefined &&
241 input.nextParent !== undefined &&
242 input.previousParent !== input.nextParent
243 ) {
244 return { action: "replace" as const, deleteFirst: false };
245 }
246 return undefined;
247};
248
249export const retryTransient = <A, E extends { readonly _tag: string }, R>(
250 effect: Effect.Effect<A, E, R>,
251) =>
252 effect.pipe(
253 Effect.provide(noRetryLayer),
254 Effect.retry({
255 while: isTransientGcpError,
256 times: 8,
257 schedule: Schedule.exponential("250 millis"),
258 }),
259 );
260
261export const retryDelete = <A, E extends { readonly _tag: string }, R>(
262 effect: Effect.Effect<A, E, R>,
263) =>
264 effect.pipe(
265 Effect.provide(noRetryLayer),
266 Effect.retry({
267 while: (error) => error._tag === "Conflict" || isTransientGcpError(error),
268 times: 8,
269 schedule: Schedule.exponential("250 millis"),
270 }),
271 );
272
273const isGone = <E extends { readonly _tag: string }>(
274 error: E,
275): error is E & { readonly _tag: "NotFound" | "BadGateway" } =>
276 error._tag === "NotFound" || error._tag === "BadGateway";
277
278/** A missing parent (NotFound) has no children. */
279const isMissingList = <E extends { readonly _tag: string }>(
280 error: E,
281): error is E & { readonly _tag: "NotFound" } => error._tag === "NotFound";
282
283export const ignoreGone = <A, E extends { readonly _tag: string }, R>(
284 effect: Effect.Effect<A, E, R>,
285) => effect.pipe(Effect.catchIf(isGone, () => Effect.void));
286
287export class DeleteNotConfirmed extends Data.TaggedError(
288 "GCP.DataLabeling.DeleteNotConfirmed",
289)<{}> {}
290
291/** Poll until the resource is gone; fails if it is still readable after ~60s. */
292export const waitUntilGone = <A, E, R>(
293 get: Effect.Effect<A | undefined, E, R>,
294) =>
295 get.pipe(
296 Effect.repeat({
297 schedule: Schedule.spaced("2 seconds"),
298 until: (value) => value === undefined,
299 times: 30,
300 }),
301 Effect.flatMap((value): Effect.Effect<void, DeleteNotConfirmed> =>
302 value === undefined ? Effect.void : Effect.fail(new DeleteNotConfirmed()),
303 ),
304 );
305
306export const waitForVisible = <A, E, R>(
307 get: Effect.Effect<A | undefined, E, R>,
308) =>
309 get.pipe(
310 Effect.filterOrFail(
311 (value): value is A => value !== undefined,
312 () => new DatalabelingPending({ name: "" }),
313 ),
314 Effect.retry({
315 while: (error) => error instanceof DatalabelingPending,
316 times: 8,
317 schedule: Schedule.exponential("250 millis"),
318 }),
319 Effect.catchIf(
320 (error): error is DatalabelingPending =>
321 error instanceof DatalabelingPending,
322 () => Effect.succeed(undefined),
323 ),
324 );
325
326const emptyList = <A>() => Effect.succeed<A[]>([]);
327
328export const collectPages = <
329 Page,
330 Item,
331 E extends { readonly _tag: string },
332 R,
333>(
334 pages: Stream.Stream<Page, E, R>,
335 items: (page: Page) => readonly Item[] | null | undefined,
336) =>
337 pages.pipe(
338 Stream.provide(noRetryLayer),
339 Stream.flatMap((page) => Stream.fromIterable(items(page) ?? [])),
340 Stream.runCollect,
341 Effect.map((chunk) => Array.from(chunk) as Item[]),
342 Effect.provide(noRetryLayer),
343 Effect.catchIf(isMissingList, () => emptyList<Item>()),
344 );
345
346export const listAnnotationSpecSets = (parent: string) =>
347 parent.length === 0
348 ? emptyList<datalabeling.GoogleCloudDatalabelingV1beta1AnnotationSpecSet>()
349 : collectPages(
350 datalabeling.listProjectsAnnotationSpecSets.pages({
351 parent,
352 pageSize: 100,
353 }),
354 (page) => page.annotationSpecSets,
355 );
356
357export const listDatasets = (parent: string) =>
358 parent.length === 0
359 ? emptyList<datalabeling.GoogleCloudDatalabelingV1beta1Dataset>()
360 : collectPages(
361 datalabeling.listProjectsDatasets.pages({ parent, pageSize: 100 }),
362 (page) => page.datasets,
363 );
364
365export const listInstructions = (parent: string) =>
366 parent.length === 0
367 ? emptyList<datalabeling.GoogleCloudDatalabelingV1beta1Instruction>()
368 : collectPages(
369 datalabeling.listProjectsInstructions.pages({ parent, pageSize: 100 }),
370 (page) => page.instructions,
371 );
372
373export const listEvaluationJobs = (parent: string) =>
374 parent.length === 0
375 ? emptyList<datalabeling.GoogleCloudDatalabelingV1beta1EvaluationJob>()
376 : collectPages(
377 datalabeling.listProjectsEvaluationJobs.pages({
378 parent,
379 pageSize: 100,
380 }),
381 (page) => page.evaluationJobs,
382 );
383
384export const listAnnotatedDatasets = (parent: string) =>
385 parent.length === 0
386 ? emptyList<datalabeling.GoogleCloudDatalabelingV1beta1AnnotatedDataset>()
387 : collectPages(
388 datalabeling.listProjectsDatasetsAnnotatedDatasets.pages({
389 parent,
390 pageSize: 100,
391 }),
392 (page) => page.annotatedDatasets,
393 );
394
395export const listFeedbackThreads = (parent: string) =>
396 parent.length === 0
397 ? emptyList<datalabeling.GoogleCloudDatalabelingV1beta1FeedbackThread>()
398 : collectPages(
399 datalabeling.listProjectsDatasetsAnnotatedDatasetsFeedbackThreads.pages(
400 {
401 parent,
402 pageSize: 100,
403 },
404 ),
405 (page) => page.feedbackThreads,
406 );
407
408export const listFeedbackMessages = (parent: string) =>
409 parent.length === 0
410 ? emptyList<datalabeling.GoogleCloudDatalabelingV1beta1FeedbackMessage>()
411 : collectPages(
412 datalabeling.listProjectsDatasetsAnnotatedDatasetsFeedbackThreadsFeedbackMessages.pages(
413 {
414 parent,
415 pageSize: 100,
416 },
417 ),
418 (page) => page.feedbackMessages,
419 );
420
421export const listAllFeedbackMessages = (project: string) =>
422 Effect.gen(function* () {
423 const datasets = yield* listDatasets(projectParent(project));
424 const namedDatasets = datasets.filter(
425 (
426 dataset,
427 ): dataset is datalabeling.GoogleCloudDatalabelingV1beta1Dataset & {
428 name: string;
429 } => typeof dataset.name === "string" && dataset.name.length > 0,
430 );
431 const groups = yield* Effect.forEach(
432 namedDatasets,
433 (dataset) =>
434 Effect.gen(function* () {
435 const annotated = yield* listAnnotatedDatasets(dataset.name);
436 const namedAnnotated = annotated.filter(
437 (
438 row,
439 ): row is datalabeling.GoogleCloudDatalabelingV1beta1AnnotatedDataset & {
440 name: string;
441 } => typeof row.name === "string" && row.name.length > 0,
442 );
443 const threadGroups = yield* Effect.forEach(
444 namedAnnotated,
445 (annotatedDataset) =>
446 Effect.gen(function* () {
447 const threads = yield* listFeedbackThreads(
448 annotatedDataset.name,
449 );
450 const namedThreads = threads.filter(
451 (
452 thread,
453 ): thread is datalabeling.GoogleCloudDatalabelingV1beta1FeedbackThread & {
454 name: string;
455 } =>
456 typeof thread.name === "string" && thread.name.length > 0,
457 );
458 const messages = yield* Effect.forEach(
459 namedThreads,
460 (thread) => listFeedbackMessages(thread.name),
461 { concurrency: 4 },
462 );
463 return messages.flat();
464 }),
465 { concurrency: 4 },
466 );
467 return threadGroups.flat();
468 }),
469 { concurrency: 4 },
470 );
471 return groups.flat();
472 });
473
474export const findOwned = <A>(
475 id: string,
476 items: readonly A[],
477 textOf: (item: A) => string | undefined,
478) =>
479 Effect.gen(function* () {
480 for (const item of items) {
481 if (yield* ownedByAlchemy(id, textOf(item))) {
482 return item;
483 }
484 }
485 return undefined;
486 });