1import { poll, PredicateFailed } from "@/Util/poll.ts";
2import * as vectorize from "@distilled.cloud/cloudflare/vectorize";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5
6// Cloudflare reports p99 vector-write visibility below two minutes. Allow
7// 150 seconds including requests, with ten retries, and reserve separate
8// lifecycle time for deployment and cleanup.
9export const waitForVectorize = <A, E, R>(input: {
10 description: string;
11 effect: Effect.Effect<A, E, R>;
12 predicate: (value: A) => boolean;
13}) =>
14 Effect.gen(function* () {
15 let latest: A | undefined;
16 return yield* poll({
17 ...input,
18 effect: input.effect.pipe(
19 Effect.tap((value) =>
20 Effect.sync(() => {
21 latest = value;
22 }),
23 ),
24 ),
25 schedule: Schedule.max([
26 Schedule.spaced("12 seconds"),
27 Schedule.recurs(10),
28 ]),
29 }).pipe(
30 Effect.timeoutOrElse({
31 duration: "150 seconds",
32 orElse: () =>
33 Effect.fail(
34 new PredicateFailed({ message: input.description, actual: latest }),
35 ),
36 }),
37 Effect.catchTag("PredicateFailed", () =>
38 Effect.fail(
39 new PredicateFailed({
40 message: `${input.description}; last observed: ${JSON.stringify(latest)}`,
41 actual: latest,
42 }),
43 ),
44 ),
45 );
46 });
47
48/** Metadata must be observable before the fixture inserts its vectors. */
49export const waitForMetadata = (
50 accountId: string,
51 indexName: string,
52 expected: ReadonlyArray<{ propertyName: string; indexType: string }>,
53) =>
54 waitForVectorize({
55 description: `Metadata ready on ${indexName}: ${JSON.stringify(expected)}`,
56 effect: vectorize
57 .listIndexMetadataIndexes({ accountId, indexName })
58 .pipe(Effect.timeout("10 seconds")),
59 predicate: (metadata) =>
60 expected.every(
61 (field) =>
62 metadata.metadataIndexes?.some(
63 (entry) =>
64 entry.propertyName === field.propertyName &&
65 entry.indexType?.toLowerCase() === field.indexType,
66 ) === true,
67 ),
68 }).pipe(Effect.map((metadata) => metadata.metadataIndexes!));