1import * as Clock from "effect/Clock";
2import * as Effect from "effect/Effect";
3import * as Schema from "effect/Schema";
4
5export const makeSubscriptionCleanup = () => {
6 let deadline: number | undefined;
7 return <A, E, R>(effect: Effect.Effect<A, E, R>) =>
8 Effect.gen(function* () {
9 const now = yield* Clock.currentTimeMillis;
10 deadline ??= now + 20_000;
11 const remaining = deadline - now;
12 if (remaining <= 0) {
13 return yield* Effect.die(
14 new Error("Subscription cleanup deadline exceeded"),
15 );
16 }
17 return yield* effect.pipe(Effect.timeout(Math.min(5_000, remaining)));
18 }).pipe(Effect.orDie, Effect.interruptible);
19};
20
21export const SubscriptionEvent = Schema.fromJsonString(
22 Schema.Struct({
23 type: Schema.String,
24 source: Schema.Struct({ type: Schema.String }),
25 metadata: Schema.Struct({
26 accountId: Schema.String,
27 eventSubscriptionId: Schema.String,
28 eventTimestamp: Schema.String,
29 }),
30 payload: Schema.Struct({
31 id: Schema.optional(Schema.String),
32 name: Schema.optional(Schema.String),
33 }),
34 }),
35);
36
37export type SubscriptionEvent = typeof SubscriptionEvent.Type;
38
39interface SubscriptionEventTarget {
40 source: "images" | "kv" | "r2" | "vectorize";
41 type: string;
42 accountId: string;
43 subscriptionId: string;
44 identity: string;
45}
46
47export interface SubscriptionProbe {
48 identity: string;
49 createdAt: number;
50}
51
52export const hasReadySubscriptionEvent = (
53 events: readonly SubscriptionEvent[],
54 probes: readonly SubscriptionProbe[],
55 expected: Omit<SubscriptionEventTarget, "identity"> & { readyAfter: number },
56) =>
57 probes.some(
58 ({ identity, createdAt }) =>
59 createdAt >= expected.readyAfter &&
60 events.some((event) =>
61 matchesSubscriptionEvent(event, { ...expected, identity }),
62 ),
63 );
64
65export const matchesSubscriptionEvent = (
66 event: SubscriptionEvent,
67 expected: SubscriptionEventTarget,
68) =>
69 event.type === `cf.${expected.source}.${expected.type}` &&
70 event.source.type === expected.source &&
71 event.metadata.accountId === expected.accountId &&
72 event.metadata.eventSubscriptionId === expected.subscriptionId &&
73 event.payload[
74 expected.source === "kv" || expected.source === "images" ? "id" : "name"
75 ] === expected.identity;