1import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
2import { describe, expect, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5
6import * as RelayDb from "../db.ts";
7import { relayDeliveryAttempts } from "../persistence/schema.ts";
8import * as DeliveryAttempts from "./DeliveryAttempts.ts";
9
10describe("DeliveryAttempts", () => {
11 it.effect("records the signed queue source job id for APNs delivery auditability", () => {
12 const insertedValues: Array<Record<string, unknown>> = [];
13 const fakeDb = {
14 insert: (table: unknown) => {
15 expect(table).toBe(relayDeliveryAttempts);
16 return {
17 values: (values: Record<string, unknown>) => {
18 insertedValues.push(values);
19 return Effect.void;
20 },
21 };
22 },
23 } as unknown as RelayDb.RelayDb["Service"];
24
25 return Effect.gen(function* () {
26 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
27 yield* attempts.record({
28 userId: "user-1",
29 environmentId: "env-1",
30 threadId: "thread-1",
31 deviceId: "device-1",
32 kind: "live_activity_update",
33 sourceJobId: "job-1",
34 token: "apns-token",
35 apnsStatus: 200,
36 apnsId: "apns-id",
37 });
38
39 expect(insertedValues).toHaveLength(1);
40 expect(insertedValues[0]).toMatchObject({
41 userId: "user-1",
42 environmentId: "env-1",
43 threadId: "thread-1",
44 deviceId: "device-1",
45 kind: "live_activity_update",
46 sourceJobId: "job-1",
47 tokenSuffix: "ns-token",
48 apnsStatus: 200,
49 apnsId: "apns-id",
50 });
51 }).pipe(
52 Effect.provide(
53 DeliveryAttempts.layer.pipe(
54 Layer.provide(NodeCryptoLayer.layer),
55 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
56 ),
57 ),
58 );
59 });
60
61 it.effect("claims signed queue source jobs before APNs delivery", () => {
62 const insertedValues: Array<Record<string, unknown>> = [];
63 const conflictTargets: Array<unknown> = [];
64 const fakeDb = {
65 insert: (table: unknown) => {
66 expect(table).toBe(relayDeliveryAttempts);
67 return {
68 values: (values: Record<string, unknown>) => {
69 insertedValues.push(values);
70 return {
71 onConflictDoNothing: (config: { readonly target: unknown }) => {
72 conflictTargets.push(config.target);
73 return {
74 returning: () => Effect.succeed([{ id: values.id }]),
75 };
76 },
77 };
78 },
79 };
80 },
81 } as unknown as RelayDb.RelayDb["Service"];
82
83 return Effect.gen(function* () {
84 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
85 const claimed = yield* attempts.claimSourceJob({
86 userId: "user-1",
87 environmentId: "env-1",
88 threadId: "thread-1",
89 deviceId: "device-1",
90 kind: "push_notification",
91 sourceJobId: "job-1",
92 token: "apns-token",
93 });
94
95 expect(claimed).toBe("claimed");
96 expect(conflictTargets).toEqual([relayDeliveryAttempts.sourceJobId]);
97 expect(insertedValues[0]).toMatchObject({
98 kind: "push_notification",
99 sourceJobId: "job-1",
100 tokenSuffix: "ns-token",
101 apnsStatus: null,
102 });
103 }).pipe(
104 Effect.provide(
105 DeliveryAttempts.layer.pipe(
106 Layer.provide(NodeCryptoLayer.layer),
107 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
108 ),
109 ),
110 );
111 });
112
113 it.effect("reports completed source jobs when the durable claim already exists", () => {
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: "2026-05-26T00:00:00.000Z",
129 apnsStatus: 200,
130 apnsReason: null,
131 apnsId: null,
132 transportError: null,
133 },
134 ]),
135 }),
136 }),
137 }),
138 } as unknown as RelayDb.RelayDb["Service"];
139
140 return Effect.gen(function* () {
141 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
142 const claimed = yield* attempts.claimSourceJob({
143 userId: "user-1",
144 environmentId: null,
145 threadId: null,
146 deviceId: "device-1",
147 kind: "live_activity_update",
148 sourceJobId: "job-1",
149 token: "apns-token",
150 });
151
152 expect(claimed).toBe("completed");
153 }).pipe(
154 Effect.provide(
155 DeliveryAttempts.layer.pipe(
156 Layer.provide(NodeCryptoLayer.layer),
157 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
158 ),
159 ),
160 );
161 });
162
163 it.effect("reports in-flight source jobs while an active claim lease exists", () => {
164 const fakeDb = {
165 insert: () => ({
166 values: () => ({
167 onConflictDoNothing: () => ({
168 returning: () => Effect.succeed([]),
169 }),
170 }),
171 }),
172 select: () => ({
173 from: () => ({
174 where: () => ({
175 limit: () =>
176 Effect.succeed([
177 {
178 createdAt: "2999-01-01T00:00:00.000Z",
179 apnsStatus: null,
180 apnsReason: null,
181 apnsId: null,
182 transportError: null,
183 },
184 ]),
185 }),
186 }),
187 }),
188 } as unknown as RelayDb.RelayDb["Service"];
189
190 return Effect.gen(function* () {
191 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
192 const claimed = yield* attempts.claimSourceJob({
193 userId: "user-1",
194 environmentId: null,
195 threadId: null,
196 deviceId: "device-1",
197 kind: "live_activity_update",
198 sourceJobId: "job-1",
199 token: "apns-token",
200 });
201
202 expect(claimed).toBe("in_flight");
203 }).pipe(
204 Effect.provide(
205 DeliveryAttempts.layer.pipe(
206 Layer.provide(NodeCryptoLayer.layer),
207 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
208 ),
209 ),
210 );
211 });
212
213 it.effect("reclaims source jobs after the claim lease expires", () => {
214 const updatedValues: Array<Record<string, unknown>> = [];
215 const fakeDb = {
216 insert: () => ({
217 values: () => ({
218 onConflictDoNothing: () => ({
219 returning: () => Effect.succeed([]),
220 }),
221 }),
222 }),
223 select: () => ({
224 from: () => ({
225 where: () => ({
226 limit: () =>
227 Effect.succeed([
228 {
229 createdAt: "1969-12-31T23:00:00.000Z",
230 apnsStatus: null,
231 apnsReason: null,
232 apnsId: null,
233 transportError: null,
234 },
235 ]),
236 }),
237 }),
238 }),
239 update: () => ({
240 set: (values: Record<string, unknown>) => {
241 updatedValues.push(values);
242 return {
243 where: () => ({
244 returning: () => Effect.succeed([{ id: "attempt-1" }]),
245 }),
246 };
247 },
248 }),
249 } as unknown as RelayDb.RelayDb["Service"];
250
251 return Effect.gen(function* () {
252 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
253 const claimed = yield* attempts.claimSourceJob({
254 userId: "user-1",
255 environmentId: null,
256 threadId: null,
257 deviceId: "device-1",
258 kind: "live_activity_update",
259 sourceJobId: "job-1",
260 token: "apns-token",
261 });
262
263 expect(claimed).toBe("claimed");
264 expect(updatedValues[0]?.createdAt).toEqual(expect.any(String));
265 }).pipe(
266 Effect.provide(
267 DeliveryAttempts.layer.pipe(
268 Layer.provide(NodeCryptoLayer.layer),
269 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
270 ),
271 ),
272 );
273 });
274
275 it.effect("completes a claimed source job with the APNs outcome", () => {
276 const updatedValues: Array<Record<string, unknown>> = [];
277 const whereClauses: Array<unknown> = [];
278 const fakeDb = {
279 update: (table: unknown) => {
280 expect(table).toBe(relayDeliveryAttempts);
281 return {
282 set: (values: Record<string, unknown>) => {
283 updatedValues.push(values);
284 return {
285 where: (clause: unknown) => {
286 whereClauses.push(clause);
287 return Effect.void;
288 },
289 };
290 },
291 };
292 },
293 } as unknown as RelayDb.RelayDb["Service"];
294
295 return Effect.gen(function* () {
296 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
297 yield* attempts.completeSourceJob({
298 sourceJobId: "job-1",
299 apnsStatus: 410,
300 apnsReason: "Unregistered",
301 apnsId: "apns-id",
302 });
303
304 expect(whereClauses).toHaveLength(1);
305 expect(updatedValues).toEqual([
306 {
307 createdAt: expect.any(String),
308 apnsStatus: 410,
309 apnsReason: "Unregistered",
310 apnsId: "apns-id",
311 transportError: null,
312 },
313 ]);
314 }).pipe(
315 Effect.provide(
316 DeliveryAttempts.layer.pipe(
317 Layer.provide(NodeCryptoLayer.layer),
318 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
319 ),
320 ),
321 );
322 });
323
324 it.effect("preserves operation context and causes for persistence failures", () => {
325 const cause = new Error("database unavailable");
326 const fakeDb = {
327 insert: () => ({
328 values: (values: Record<string, unknown>) =>
329 values.kind === "record"
330 ? Effect.fail(cause)
331 : {
332 onConflictDoNothing: () => ({
333 returning: () => Effect.fail(cause),
334 }),
335 },
336 }),
337 update: () => ({
338 set: () => ({
339 where: () => Effect.fail(cause),
340 }),
341 }),
342 } as unknown as RelayDb.RelayDb["Service"];
343
344 return Effect.gen(function* () {
345 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
346 const recordError = yield* Effect.flip(
347 attempts.record({
348 userId: "user-1",
349 environmentId: "env-1",
350 threadId: "thread-1",
351 deviceId: "device-1",
352 kind: "record",
353 sourceJobId: "job-1",
354 token: "apns-token",
355 }),
356 );
357 const claimError = yield* Effect.flip(
358 attempts.claimSourceJob({
359 userId: "user-2",
360 environmentId: "env-2",
361 threadId: "thread-2",
362 deviceId: "device-2",
363 kind: "claim",
364 sourceJobId: "job-2",
365 token: "apns-token",
366 }),
367 );
368 const completionError = yield* Effect.flip(
369 attempts.completeSourceJob({ sourceJobId: "job-3", apnsStatus: 500 }),
370 );
371
372 expect(recordError).toMatchObject({
373 operation: "record",
374 sourceJobId: "job-1",
375 userId: "user-1",
376 environmentId: "env-1",
377 threadId: "thread-1",
378 deviceId: "device-1",
379 kind: "record",
380 cause,
381 message: "Failed to persist APNs delivery attempt during record.",
382 });
383 expect(claimError).toMatchObject({
384 operation: "claim-source-job",
385 sourceJobId: "job-2",
386 userId: "user-2",
387 environmentId: "env-2",
388 threadId: "thread-2",
389 deviceId: "device-2",
390 kind: "claim",
391 cause,
392 message: "Failed to persist APNs delivery attempt during claim-source-job.",
393 });
394 expect(completionError).toMatchObject({
395 operation: "complete-source-job",
396 sourceJobId: "job-3",
397 userId: null,
398 environmentId: null,
399 threadId: null,
400 deviceId: null,
401 kind: null,
402 cause,
403 message: "Failed to persist APNs delivery attempt during complete-source-job.",
404 });
405 }).pipe(
406 Effect.provide(
407 DeliveryAttempts.layer.pipe(
408 Layer.provide(NodeCryptoLayer.layer),
409 Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)),
410 ),
411 ),
412 );
413 });
414});