Open History to see the last commit
1import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
2import { describe, expect, it } from "@effect/vitest";
3import * as Drizzle from "@alchemy.run/drizzle/Postgres";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Redacted from "effect/Redacted";
7
8import * as RelayDb from "../db.ts";
9import * as DeliveryAttempts from "./DeliveryAttempts.ts";
10
11describe("DeliveryAttempts", () => {
12 it.effect("reports completed source jobs when the durable claim already exists", () => {
13 const fakeDb = {
14 insert: () => ({
15 values: () => ({
16 onConflictDoNothing: () => ({
17 returning: () => Effect.succeed([]),
18 }),
19 }),
20 }),
21 select: () => ({
22 from: () => ({
23 where: () => ({
24 limit: () =>
25 Effect.succeed([
26 {
27 createdAt: "2026-05-26T00:00:00.000Z",
28 apnsStatus: 200,
29 apnsReason: null,
30 apnsId: null,
31 transportError: null,
32 },
33 ]),
34 }),
35 }),
36 }),
37 } as unknown as RelayDb.RelayDb["Service"];
38
39 return Effect.gen(function* () {
40 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
41 const claimed = yield* attempts.claimSourceJob({
42 userId: "user-1",
43 environmentId: null,
44 threadId: null,
45 deviceId: "device-1",
46 kind: "live_activity_update",
47 sourceJobId: "job-1",
48 token: "apns-token",
49 });
50
51 expect(claimed).toBe("completed");
52 }).pipe(
53 Effect.provide(
54 DeliveryAttempts.layer.pipe(
55 Layer.provide(NodeCryptoLayer.layer),
56 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
57 ),
58 ),
59 );
60 });
61
62 it.effect("reports in-flight source jobs while an active claim lease exists", () => {
63 const fakeDb = {
64 insert: () => ({
65 values: () => ({
66 onConflictDoNothing: () => ({
67 returning: () => Effect.succeed([]),
68 }),
69 }),
70 }),
71 select: () => ({
72 from: () => ({
73 where: () => ({
74 limit: () =>
75 Effect.succeed([
76 {
77 createdAt: "2999-01-01T00:00:00.000Z",
78 apnsStatus: null,
79 apnsReason: null,
80 apnsId: null,
81 transportError: null,
82 },
83 ]),
84 }),
85 }),
86 }),
87 } as unknown as RelayDb.RelayDb["Service"];
88
89 return Effect.gen(function* () {
90 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
91 const claimed = yield* attempts.claimSourceJob({
92 userId: "user-1",
93 environmentId: null,
94 threadId: null,
95 deviceId: "device-1",
96 kind: "live_activity_update",
97 sourceJobId: "job-1",
98 token: "apns-token",
99 });
100
101 expect(claimed).toBe("in_flight");
102 }).pipe(
103 Effect.provide(
104 DeliveryAttempts.layer.pipe(
105 Layer.provide(NodeCryptoLayer.layer),
106 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
107 ),
108 ),
109 );
110 });
111
112 it.effect("reclaims source jobs after the claim lease expires", () => {
113 const updatedValues: Array<Record<string, unknown>> = [];
114 const fakeDb = {
115 insert: () => ({
116 values: () => ({
117 onConflictDoNothing: () => ({
118 returning: () => Effect.succeed([]),
119 }),
120 }),
121 }),
122 select: () => ({
123 from: () => ({
124 where: () => ({
125 limit: () =>
126 Effect.succeed([
127 {
128 createdAt: "1969-12-31T23:00:00.000Z",
129 apnsStatus: null,
130 apnsReason: null,
131 apnsId: null,
132 transportError: null,
133 },
134 ]),
135 }),
136 }),
137 }),
138 update: () => ({
139 set: (values: Record<string, unknown>) => {
140 updatedValues.push(values);
141 return {
142 where: () => ({
143 returning: () => Effect.succeed([{ id: "attempt-1" }]),
144 }),
145 };
146 },
147 }),
148 } as unknown as RelayDb.RelayDb["Service"];
149
150 return Effect.gen(function* () {
151 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
152 const claimed = yield* attempts.claimSourceJob({
153 userId: "user-1",
154 environmentId: null,
155 threadId: null,
156 deviceId: "device-1",
157 kind: "live_activity_update",
158 sourceJobId: "job-1",
159 token: "apns-token",
160 });
161
162 expect(claimed).toBe("claimed");
163 expect(updatedValues[0]?.createdAt).toEqual(expect.any(String));
164 }).pipe(
165 Effect.provide(
166 DeliveryAttempts.layer.pipe(
167 Layer.provide(NodeCryptoLayer.layer),
168 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
169 ),
170 ),
171 );
172 });
173
174 const pruneWith = (batches: ReadonlyArray<number>) => {
175 const deleteCalls: number[] = [];
176 const fakeDb = {
177 delete: () => ({
178 where: () => ({
179 returning: () => {
180 const size = batches[deleteCalls.length] ?? 0;
181 deleteCalls.push(size);
182 return Effect.succeed(Array.from({ length: size }, (_, index) => ({ id: `${index}` })));
183 },
184 }),
185 }),
186 } as unknown as RelayDb.RelayDb["Service"];
187 const run = DeliveryAttempts.DeliveryAttempts.pipe(
188 Effect.flatMap((attempts) => attempts.pruneExpired),
189 Effect.provide(
190 DeliveryAttempts.layer.pipe(
191 Layer.provide(NodeCryptoLayer.layer),
192 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
193 ),
194 ),
195 );
196 return { run, deleteCalls };
197 };
198
199 it.effect("stops pruning once a batch comes back short", () => {
200 const full = DeliveryAttempts.DELIVERY_ATTEMPT_PRUNE_BATCH_SIZE;
201 const { run, deleteCalls } = pruneWith([full, 12]);
202 return Effect.gen(function* () {
203 expect(yield* run).toBe(full + 12);
204 expect(deleteCalls).toEqual([full, 12]);
205 });
206 });
207
208 it.effect("caps each run at the batch limit while a backlog remains", () => {
209 const full = DeliveryAttempts.DELIVERY_ATTEMPT_PRUNE_BATCH_SIZE;
210 const { run, deleteCalls } = pruneWith(Array.from({ length: 10 }, () => full));
211 return Effect.gen(function* () {
212 expect(yield* run).toBe(full * DeliveryAttempts.DELIVERY_ATTEMPT_PRUNE_MAX_BATCHES);
213 expect(deleteCalls).toHaveLength(DeliveryAttempts.DELIVERY_ATTEMPT_PRUNE_MAX_BATCHES);
214 });
215 });
216
217 // The Worker's db is Alchemy's proxyChain, whose builders render as nothing inside another
218 // query. Nothing listens on the port; the failed query still reports the SQL it sent.
219 it.live("prunes with a subquery the Worker's proxied database can render", () =>
220 Effect.gen(function* () {
221 const db = yield* Drizzle.Postgres(
222 Effect.succeed(Redacted.make("postgres://relay:relay@127.0.0.1:1/relay")),
223 { client: { prepare: false } },
224 );
225 const error = yield* DeliveryAttempts.DeliveryAttempts.pipe(
226 Effect.flatMap((attempts) => attempts.pruneExpired),
227 Effect.provide(
228 DeliveryAttempts.layer.pipe(
229 Layer.provide(NodeCryptoLayer.layer),
230 Layer.provide(Layer.succeed(RelayDb.RelayDb, db)),
231 ),
232 ),
233 Effect.flip,
234 );
235 expect((error.cause as { readonly query: string }).query).toMatch(
236 /"relay_delivery_attempts"\."id" in \(select "id" from "relay_delivery_attempts" where "relay_delivery_attempts"\."created_at" < \$1 limit \$2\)/,
237 );
238 }).pipe(Effect.scoped),
239 );
240});