Open History to see the last commit
1import * as AgentActivityAlerts from "./agentActivityAlerts.ts";
2import { EnvironmentId, ThreadId } from "@t3tools/contracts";
3import type {
4 RelayAgentActivityAggregateState,
5 RelayAgentActivityState,
6} from "@t3tools/contracts/relay";
7import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
8import { describe, expect, it } from "@effect/vitest";
9// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no generateKeyPairSync.
10import * as NodeCrypto from "node:crypto";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Redacted from "effect/Redacted";
14import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http";
15
16import {
17 makeApnsDeliveryJobPayload,
18 signApnsDeliveryJob,
19 type SignedApnsDeliveryJob,
20} from "./apnsDeliveryJobs.ts";
21import * as DeliveryAttempts from "./DeliveryAttempts.ts";
22import * as LiveActivities from "./LiveActivities.ts";
23import * as RelayConfiguration from "../Config.ts";
24import * as ApnsDeliveryQueue from "./ApnsDeliveryQueue.ts";
25import * as AgentActivityRows from "./AgentActivityRows.ts";
26import * as ApnsDeliveries from "./ApnsDeliveries.ts";
27import * as ApnsClient from "./ApnsClient.ts";
28import * as ApnsProviderTokens from "./ApnsProviderTokens.ts";
29
30const config = RelayConfiguration.RelayConfiguration.of({
31 relayIssuer: "https://relay.example.test",
32 apns: {
33 environment: "sandbox",
34 teamId: "team-id",
35 keyId: "key-id",
36 privateKey: Redacted.make("not-a-private-key"),
37 bundleId: "com.t3tools.t3code.dev",
38 },
39 apnsDeliveryJobSigningSecret: Redacted.make("job-signing-secret"),
40 clerkSecretKey: Redacted.make("clerk-secret"),
41 clerkPublishableKey: "pk_test_test",
42 clerkJwtAudience: "t3-code-relay",
43 cloudMintPrivateKey: Redacted.make("cloud-private-key"),
44 cloudMintPublicKey: "cloud-public-key",
45 managedEndpointBaseDomain: undefined,
46 managedEndpointNamespace: undefined,
47});
48
49const apnsSigningKeyPair = NodeCrypto.generateKeyPairSync("ec", {
50 namedCurve: "P-256",
51 privateKeyEncoding: { format: "pem", type: "pkcs8" },
52 publicKeyEncoding: { format: "pem", type: "spki" },
53});
54
55const signingConfig = RelayConfiguration.RelayConfiguration.of({
56 ...config,
57 apns: {
58 ...config.apns!,
59 privateKey: Redacted.make(apnsSigningKeyPair.privateKey),
60 },
61});
62
63const state: RelayAgentActivityState = {
64 environmentId: "env" as RelayAgentActivityState["environmentId"],
65 threadId: "thread" as RelayAgentActivityState["threadId"],
66 projectTitle: "Project",
67 threadTitle: "Thread",
68 modelTitle: "gpt-5.4",
69 phase: "running",
70 headline: "Running",
71 updatedAt: "1970-01-01T00:00:00.000Z",
72 deepLink: "/",
73};
74
75const aggregate: RelayAgentActivityAggregateState = {
76 title: "T3 Code",
77 subtitle: "Agent work in progress",
78 activeCount: 1,
79 updatedAt: state.updatedAt,
80 activities: [
81 {
82 environmentId: state.environmentId,
83 threadId: state.threadId,
84 projectTitle: state.projectTitle,
85 threadTitle: state.threadTitle,
86 modelTitle: state.modelTitle,
87 phase: state.phase,
88 status: "Working",
89 updatedAt: state.updatedAt,
90 deepLink: state.deepLink,
91 },
92 ],
93};
94
95const enabledPreferences = JSON.stringify({
96 liveActivitiesEnabled: true,
97 notificationsEnabled: true,
98 notifyOnApproval: true,
99 notifyOnInput: true,
100 notifyOnCompletion: true,
101 notifyOnFailure: true,
102});
103
104const disabledPreferences = JSON.stringify({
105 liveActivitiesEnabled: false,
106 notificationsEnabled: true,
107 notifyOnApproval: true,
108 notifyOnInput: true,
109 notifyOnCompletion: true,
110 notifyOnFailure: true,
111});
112
113const notificationsDisabledPreferences = JSON.stringify({
114 liveActivitiesEnabled: false,
115 notificationsEnabled: false,
116 notifyOnApproval: true,
117 notifyOnInput: true,
118 notifyOnCompletion: true,
119 notifyOnFailure: true,
120});
121
122const target: LiveActivities.TargetRow = {
123 user_id: "dev:julius",
124 device_id: "device-1",
125 platform: "ios",
126 ios_major_version: 18,
127 app_version: "1.0.0",
128 bundle_id: null,
129 aps_environment: null,
130 push_token: null,
131 push_to_start_token: "start-token",
132 preferences_json: enabledPreferences,
133 activity_push_token: "activity-token",
134 remote_start_queued_at: null,
135 remote_started_at: "1970-01-01T00:00:00.000Z",
136 ended_at: null,
137 last_aggregate_json: null,
138 last_live_activity_delivery_at: null,
139};
140
141function layerFor(input: {
142 readonly attempts: Array<DeliveryAttempts.DeliveryAttemptInput>;
143 readonly sourceJobClaims?: ReadonlyMap<string, DeliveryAttempts.DeliverySourceJobClaimResult>;
144 readonly queuedJobs?: Array<SignedApnsDeliveryJob>;
145 readonly markedDeliveries?: Array<
146 Parameters<LiveActivities.LiveActivities["Service"]["markDelivery"]>[0]
147 >;
148 readonly invalidatedTokens?: Array<
149 Parameters<LiveActivities.LiveActivities["Service"]["invalidateDeliveryToken"]>[0]
150 >;
151 readonly currentTargets?: ReadonlyArray<LiveActivities.TargetRow>;
152 readonly config?: RelayConfiguration.RelayConfiguration["Service"];
153 readonly activityStates?: ReadonlyArray<RelayAgentActivityState>;
154 readonly currentActivityStates?: ReadonlyArray<RelayAgentActivityState>;
155 readonly execute?: (
156 request: HttpClientRequest.HttpClientRequest,
157 ) => Effect.Effect<HttpClientResponse.HttpClientResponse>;
158}) {
159 return ApnsDeliveries.layer.pipe(
160 Layer.provideMerge(NodeCryptoLayer.layerMac),
161 Layer.provide(ApnsClient.layer),
162 Layer.provide(ApnsProviderTokens.layer),
163 Layer.provide(ApnsDeliveryQueue.layer.pipe(Layer.provide(NodeCryptoLayer.layer))),
164 Layer.provide(
165 Layer.mergeAll(
166 Layer.succeed(AgentActivityRows.AgentActivityRows, {
167 upsert: () => Effect.void,
168 remove: () => Effect.void,
169 pruneTerminal: () => Effect.void,
170 listForUser: () =>
171 input.activityStates !== undefined
172 ? Effect.succeed([...input.activityStates])
173 : Effect.succeed([state]),
174 getForUserThread: ({ environmentId, threadId }) =>
175 Effect.succeed(
176 (input.currentActivityStates ?? input.activityStates ?? [state]).find(
177 (current) =>
178 current.environmentId === environmentId && current.threadId === threadId,
179 ) ?? null,
180 ),
181 } satisfies AgentActivityRows.AgentActivityRows["Service"]),
182 Layer.succeed(ApnsDeliveryQueue.ApnsDeliveryQueueSender, {
183 send: (body) =>
184 Effect.sync(() => {
185 input.queuedJobs?.push(body);
186 }),
187 }),
188 Layer.succeed(DeliveryAttempts.DeliveryAttempts, {
189 pruneExpired: Effect.succeed(0),
190 claimSourceJob: (attempt) =>
191 Effect.sync(() => {
192 const claim = input.sourceJobClaims?.get(attempt.sourceJobId);
193 if (claim) {
194 return claim;
195 }
196 input.attempts.push(attempt);
197 return "claimed";
198 }),
199 completeSourceJob: (completion) =>
200 Effect.sync(() => {
201 const attempt = input.attempts.find(
202 (row) => row.sourceJobId === completion.sourceJobId,
203 );
204 if (attempt) {
205 Object.assign(attempt, completion);
206 }
207 }),
208 }),
209 Layer.succeed(LiveActivities.LiveActivities, {
210 register: () => Effect.void,
211 listTargets: () => Effect.succeed(input.currentTargets ?? [target]),
212 listIdleArmedTargets: () => Effect.succeed([]),
213 markDelivery: (delivery) =>
214 Effect.sync(() => {
215 input.markedDeliveries?.push(delivery);
216 }),
217 invalidateDeliveryToken: (invalidated) =>
218 Effect.sync(() => {
219 input.invalidatedTokens?.push(invalidated);
220 }),
221 }),
222 RelayConfiguration.layer(input.config ?? config),
223 input.execute
224 ? Layer.succeed(HttpClient.HttpClient, HttpClient.make(input.execute))
225 : FetchHttpClient.layer,
226 ),
227 ),
228 );
229}
230
231describe("ApnsDeliveries", () => {
232 it.effect("queues bounded Live Activity aggregate payloads", () => {
233 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
234 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
235 const longTitle = "x".repeat(300);
236 const inputAggregate: RelayAgentActivityAggregateState = {
237 ...aggregate,
238 title: longTitle,
239 subtitle: longTitle,
240 activities: [0, 1, 2, 3, 4, 5].map((index) =>
241 Object.assign({}, aggregate.activities[0]!, {
242 projectTitle: longTitle,
243 threadTitle: longTitle,
244 modelTitle: longTitle,
245 status: longTitle,
246 threadId: `thread-${index}` as RelayAgentActivityState["threadId"],
247 deepLink: "https://example.test/not-an-app-link",
248 }),
249 ),
250 };
251
252 return Effect.gen(function* () {
253 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
254 yield* deliveries.sendForTarget({
255 target,
256 aggregate: inputAggregate,
257 nowMs: 10_000,
258 });
259
260 const payloadAggregate = queuedJobs[0]?.payload.aggregate;
261 expect(payloadAggregate?.title.length).toBeLessThanOrEqual(120);
262 expect(payloadAggregate?.subtitle.length).toBeLessThanOrEqual(120);
263 expect(payloadAggregate?.activities).toHaveLength(5);
264 expect(payloadAggregate?.activities[0]?.projectTitle.length).toBeLessThanOrEqual(120);
265 expect(payloadAggregate?.activities[0]?.status.length).toBeLessThanOrEqual(40);
266 expect(payloadAggregate?.activities[0]?.deepLink).toBe("/");
267 expect(attempts).toEqual([]);
268 }).pipe(Effect.provide(layerFor({ attempts, queuedJobs })));
269 });
270
271 it.effect.each([
272 {
273 label: "connecting becomes working",
274 previousRows: [{ ...aggregate.activities[0]!, phase: "starting", status: "Connecting" }],
275 },
276 {
277 label: "input resumes working",
278 previousRows: [{ ...aggregate.activities[0]!, phase: "waiting_for_input", status: "Input" }],
279 },
280 {
281 label: "a settled terminal row leaves",
282 previousRows: [
283 aggregate.activities[0]!,
284 {
285 ...aggregate.activities[0]!,
286 threadId: ThreadId.make("finished"),
287 phase: "completed",
288 status: "Done",
289 },
290 ],
291 },
292 {
293 label: "the same thread id moves to another environment",
294 previousRows: [
295 { ...aggregate.activities[0]!, environmentId: EnvironmentId.make("other-env") },
296 ],
297 },
298 ] as const)("queues an update inside the throttle window when $label", ({ previousRows }) => {
299 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
300 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
301 const previousAggregate: RelayAgentActivityAggregateState = {
302 ...aggregate,
303 activities: previousRows,
304 };
305
306 return Effect.gen(function* () {
307 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
308 const result = yield* deliveries.sendForTarget({
309 target: {
310 ...target,
311 last_aggregate_json: JSON.stringify(previousAggregate),
312 last_live_activity_delivery_at: "1970-01-01T00:00:04.000Z",
313 },
314 aggregate,
315 nowMs: 5_000,
316 });
317
318 expect(result?.kind).toBe("live_activity_update");
319 expect(queuedJobs).toMatchObject([{ payload: { kind: "live_activity_update", aggregate } }]);
320 expect(queuedJobs[0]?.payload.alert ?? null).toBeNull();
321 expect(attempts).toEqual([]);
322 }).pipe(Effect.provide(layerFor({ attempts, queuedJobs })));
323 });
324
325 it.effect.each([
326 { label: "inside", nowMs: 5_000, queued: false },
327 { label: "after", nowMs: 19_000, queued: true },
328 ])("throttles timestamp-only updates $label the window", ({ nowMs, queued }) => {
329 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
330 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
331 const changedAggregate: RelayAgentActivityAggregateState = {
332 ...aggregate,
333 updatedAt: "1970-01-01T00:00:04.000Z",
334 activities: [{ ...aggregate.activities[0]!, updatedAt: "1970-01-01T00:00:04.000Z" }],
335 };
336 const previousAggregateJson = JSON.stringify(aggregate);
337
338 return Effect.gen(function* () {
339 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
340 const result = yield* deliveries.sendForTarget({
341 target: {
342 ...target,
343 last_aggregate_json: previousAggregateJson,
344 last_live_activity_delivery_at: "1970-01-01T00:00:04.000Z",
345 },
346 aggregate: changedAggregate,
347 nowMs,
348 });
349
350 expect(result?.kind ?? null).toBe(queued ? "live_activity_update" : null);
351 expect(queuedJobs).toHaveLength(queued ? 1 : 0);
352 }).pipe(Effect.provide(layerFor({ attempts, queuedJobs })));
353 });
354
355 it.effect(
356 "queues an alert while ending an active Live Activity when only Live Activities are disabled",
357 () => {
358 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
359 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
360 const inputAggregate: RelayAgentActivityAggregateState = {
361 ...aggregate,
362 activities: [
363 {
364 ...aggregate.activities[0]!,
365 phase: "waiting_for_input",
366 status: "Input",
367 },
368 ],
369 };
370
371 return Effect.gen(function* () {
372 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
373 const result = yield* deliveries.sendForTarget({
374 target: {
375 ...target,
376 push_token: "apns-device-token",
377 preferences_json: disabledPreferences,
378 },
379 aggregate: inputAggregate,
380 nowMs: 5_000,
381 });
382
383 expect(result?.kind).toBe("live_activity_end");
384 expect(queuedJobs).toMatchObject([
385 {
386 payload: {
387 kind: "live_activity_end",
388 target: {
389 token: "activity-token",
390 },
391 },
392 },
393 {
394 payload: {
395 kind: "push_notification",
396 target: {
397 token: "apns-device-token",
398 },
399 notification: {
400 title: "Thread",
401 body: "Input: Project",
402 environmentId: "env",
403 threadId: "thread",
404 deepLink: "/",
405 phase: "waiting_for_input",
406 updatedAt: state.updatedAt,
407 },
408 },
409 },
410 ]);
411 expect(attempts).toEqual([]);
412 }).pipe(Effect.provide(layerFor({ attempts, queuedJobs })));
413 },
414 );
415
416 it.effect("does not queue alert pushes when notification permission is disabled", () => {
417 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
418 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
419 const inputAggregate: RelayAgentActivityAggregateState = {
420 ...aggregate,
421 activities: [
422 {
423 ...aggregate.activities[0]!,
424 phase: "waiting_for_input",
425 status: "Input",
426 },
427 ],
428 };
429
430 return Effect.gen(function* () {
431 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
432 const result = yield* deliveries.sendForTarget({
433 target: {
434 ...target,
435 push_token: "apns-device-token",
436 preferences_json: notificationsDisabledPreferences,
437 },
438 aggregate: inputAggregate,
439 nowMs: 5_000,
440 });
441
442 expect(result?.kind).toBe("live_activity_end");
443 expect(queuedJobs).toMatchObject([
444 {
445 payload: {
446 kind: "live_activity_end",
447 target: {
448 token: "activity-token",
449 },
450 },
451 },
452 ]);
453 expect(attempts).toEqual([]);
454 }).pipe(Effect.provide(layerFor({ attempts, queuedJobs })));
455 });
456
457 it.effect(
458 "queues a push notification for approval and input states when no Live Activity delivery is available",
459 () => {
460 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
461 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
462 const inputAggregate: RelayAgentActivityAggregateState = {
463 ...aggregate,
464 activities: [
465 {
466 ...aggregate.activities[0]!,
467 phase: "waiting_for_input",
468 status: "Input",
469 },
470 ],
471 };
472
473 return Effect.gen(function* () {
474 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
475 const result = yield* deliveries.sendForTarget({
476 target: {
477 ...target,
478 push_token: "apns-device-token",
479 push_to_start_token: null,
480 activity_push_token: null,
481 remote_started_at: null,
482 },
483 aggregate: inputAggregate,
484 nowMs: 5_000,
485 });
486
487 expect(result?.kind).toBe("push_notification");
488 expect(result?.ok).toBe(true);
489 expect(queuedJobs).toMatchObject([
490 {
491 payload: {
492 kind: "push_notification",
493 target: {
494 token: "apns-device-token",
495 },
496 notification: {
497 title: "Thread",
498 body: "Input: Project",
499 environmentId: "env",
500 threadId: "thread",
501 deepLink: "/",
502 },
503 },
504 },
505 ]);
506 expect(attempts).toEqual([]);
507 }).pipe(Effect.provide(layerFor({ attempts, queuedJobs })));
508 },
509 );
510
511 it.effect("queues bounded alert notification payloads", () => {
512 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
513 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
514 const longTitle = "x".repeat(300);
515 const inputAggregate: RelayAgentActivityAggregateState = {
516 ...aggregate,
517 activities: [
518 {
519 ...aggregate.activities[0]!,
520 projectTitle: longTitle,
521 threadTitle: longTitle,
522 phase: "waiting_for_input",
523 status: "Input",
524 deepLink: "https://example.test/not-an-app-link",
525 },
526 ],
527 };
528
529 return Effect.gen(function* () {
530 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
531 yield* deliveries.sendForTarget({
532 target: {
533 ...target,
534 push_token: "apns-device-token",
535 push_to_start_token: null,
536 activity_push_token: null,
537 remote_started_at: null,
538 },
539 aggregate: inputAggregate,
540 nowMs: 5_000,
541 });
542
543 const notification = queuedJobs[0]?.payload.notification;
544 expect(notification?.title.length).toBeLessThanOrEqual(120);
545 expect(notification?.body.length).toBeLessThanOrEqual(120);
546 expect(notification?.deepLink).toBe("/");
547 expect(attempts).toEqual([]);
548 }).pipe(Effect.provide(layerFor({ attempts, queuedJobs })));
549 });
550
551 it.effect.each([
552 { phase: "running", updatedAt: state.updatedAt, disabled: false, sends: false },
553 { phase: "completed", updatedAt: state.updatedAt, disabled: false, sends: false },
554 { phase: "completed", updatedAt: "1969-12-31T23:44:00.000Z", disabled: false, sends: true },
555 { phase: "running", updatedAt: state.updatedAt, disabled: true, sends: true },
556 ] as const)(
557 "rechecks a queued contentless end: $phase at $updatedAt, disabled=$disabled",
558 ({ phase, updatedAt, disabled, sends }) => {
559 const requests: HttpClientRequest.HttpClientRequest[] = [];
560 const signedEnd = signApnsDeliveryJob({
561 secret: config.apnsDeliveryJobSigningSecret,
562 payload: makeApnsDeliveryJobPayload({
563 kind: "live_activity_end",
564 userId: target.user_id,
565 deviceId: target.device_id,
566 token: "activity-token",
567 aggregate: null,
568 createdAt: state.updatedAt,
569 expiresAt: "1970-01-01T00:10:00.000Z",
570 jobId: "job-end-1",
571 }),
572 });
573 return Effect.gen(function* () {
574 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
575 const result = yield* deliveries.processSignedJob(yield* signedEnd);
576 expect(result.ok).toBe(true);
577 expect(requests).toHaveLength(sends ? 1 : 0);
578 if (sends) {
579 const request = requests[0]!;
580 expect(request.url).toBe("https://api.sandbox.push.apple.com/3/device/activity-token");
581 expect(request.body._tag).toBe("Uint8Array");
582 if (request.body._tag === "Uint8Array") {
583 const { aps } = JSON.parse(new TextDecoder().decode(request.body.body));
584 expect(aps.event).toBe("end");
585 expect(aps.alert).toBeUndefined();
586 expect(aps.sound).toBeUndefined();
587 }
588 }
589 }).pipe(
590 Effect.provide(
591 layerFor({
592 attempts: [],
593 config: signingConfig,
594 currentTargets: [
595 { ...target, preferences_json: disabled ? disabledPreferences : enabledPreferences },
596 ],
597 activityStates: [{ ...state, phase, updatedAt }],
598 execute: (request) =>
599 Effect.sync(() => {
600 requests.push(request);
601 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
602 }),
603 }),
604 ),
605 );
606 },
607 );
608
609 it.effect("processes signed push notification jobs through APNs and records attempts", () => {
610 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
611 const payload = makeApnsDeliveryJobPayload({
612 kind: "push_notification",
613 userId: target.user_id,
614 deviceId: target.device_id,
615 token: "apns-device-token",
616 aggregate: null,
617 notification: {
618 title: "Thread",
619 body: "Input: Project",
620 environmentId: "env",
621 threadId: "thread",
622 deepLink: "/",
623 },
624 createdAt: "1970-01-01T00:00:00.000Z",
625 expiresAt: "1970-01-01T00:10:00.000Z",
626 jobId: "job-push-1",
627 });
628 const signed = signApnsDeliveryJob({
629 secret: config.apnsDeliveryJobSigningSecret,
630 payload,
631 });
632 const execute = (request: HttpClientRequest.HttpClientRequest) =>
633 Effect.succeed(HttpClientResponse.fromWeb(request, new Response("", { status: 200 })));
634
635 return Effect.gen(function* () {
636 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
637 const result = yield* deliveries.processSignedJob(yield* signed);
638
639 expect(result.kind).toBe("push_notification");
640 expect(result.ok).toBe(true);
641 expect(result.apnsStatus).toBe(200);
642 expect(attempts).toMatchObject([
643 {
644 kind: "push_notification",
645 sourceJobId: "job-push-1",
646 token: "apns-device-token",
647 environmentId: "env",
648 threadId: "thread",
649 deviceId: target.device_id,
650 apnsStatus: 200,
651 },
652 ]);
653 }).pipe(
654 Effect.provide(
655 layerFor({
656 attempts,
657 currentTargets: [
658 {
659 ...target,
660 push_token: "apns-device-token",
661 },
662 ],
663 config: signingConfig,
664 execute,
665 }),
666 ),
667 );
668 });
669
670 it.effect("skips duplicate signed queue jobs before calling APNs", () => {
671 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
672 let executeCount = 0;
673 const payload = makeApnsDeliveryJobPayload({
674 kind: "push_notification",
675 userId: target.user_id,
676 deviceId: target.device_id,
677 token: "apns-device-token",
678 aggregate: null,
679 notification: {
680 title: "Thread",
681 body: "Input: Project",
682 environmentId: "env",
683 threadId: "thread",
684 deepLink: "/",
685 },
686 createdAt: "1970-01-01T00:00:00.000Z",
687 expiresAt: "1970-01-01T00:10:00.000Z",
688 jobId: "job-push-duplicate",
689 });
690 const signed = signApnsDeliveryJob({
691 secret: config.apnsDeliveryJobSigningSecret,
692 payload,
693 });
694 const execute = (request: HttpClientRequest.HttpClientRequest) =>
695 Effect.sync(() => {
696 executeCount += 1;
697 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
698 });
699
700 return Effect.gen(function* () {
701 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
702 const result = yield* deliveries.processSignedJob(yield* signed);
703
704 expect(result).toMatchObject({
705 kind: "push_notification",
706 ok: true,
707 apnsStatus: null,
708 apnsReason: "Duplicate APNs delivery job skipped.",
709 });
710 expect(executeCount).toBe(0);
711 expect(attempts).toEqual([]);
712 }).pipe(
713 Effect.provide(
714 layerFor({
715 attempts,
716 sourceJobClaims: new Map([["job-push-duplicate", "completed"]]),
717 config: signingConfig,
718 execute,
719 }),
720 ),
721 );
722 });
723
724 it.effect("skips stale signed Live Activity jobs when the registered token changed", () => {
725 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
726 let executeCount = 0;
727 const payload = makeApnsDeliveryJobPayload({
728 kind: "live_activity_update",
729 userId: target.user_id,
730 deviceId: target.device_id,
731 token: "stale-activity-token",
732 aggregate,
733 createdAt: "1970-01-01T00:00:00.000Z",
734 expiresAt: "1970-01-01T00:10:00.000Z",
735 jobId: "job-update-stale-token",
736 });
737 const signed = signApnsDeliveryJob({
738 secret: config.apnsDeliveryJobSigningSecret,
739 payload,
740 });
741 const execute = (request: HttpClientRequest.HttpClientRequest) =>
742 Effect.sync(() => {
743 executeCount += 1;
744 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
745 });
746
747 return Effect.gen(function* () {
748 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
749 const result = yield* deliveries.processSignedJob(yield* signed);
750
751 expect(result).toMatchObject({
752 kind: "live_activity_update",
753 ok: true,
754 apnsStatus: null,
755 apnsReason: "Stale APNs delivery job skipped.",
756 });
757 expect(executeCount).toBe(0);
758 expect(attempts).toMatchObject([
759 {
760 kind: "live_activity_update",
761 sourceJobId: "job-update-stale-token",
762 token: "stale-activity-token",
763 apnsReason: "Stale APNs delivery job skipped.",
764 },
765 ]);
766 }).pipe(
767 Effect.provide(
768 layerFor({
769 attempts,
770 config: signingConfig,
771 execute,
772 }),
773 ),
774 );
775 });
776
777 it.effect("skips a queued Done Live Activity update after the thread resumes working", () => {
778 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
779 let executeCount = 0;
780 const completedAt = "1970-01-01T00:00:01.000Z";
781 const completedAggregate: RelayAgentActivityAggregateState = {
782 ...aggregate,
783 activeCount: 0,
784 updatedAt: completedAt,
785 activities: [
786 {
787 ...aggregate.activities[0]!,
788 phase: "completed",
789 status: "Done",
790 updatedAt: completedAt,
791 },
792 ],
793 };
794 const payload = makeApnsDeliveryJobPayload({
795 kind: "live_activity_update",
796 userId: target.user_id,
797 deviceId: target.device_id,
798 token: target.activity_push_token ?? "activity-token",
799 aggregate: completedAggregate,
800 alert: { title: "Thread", body: "Done: Project" },
801 createdAt: completedAt,
802 expiresAt: "1970-01-01T00:10:00.000Z",
803 jobId: "job-update-superseded-by-running",
804 });
805 const signed = signApnsDeliveryJob({
806 secret: config.apnsDeliveryJobSigningSecret,
807 payload,
808 });
809 const execute = (request: HttpClientRequest.HttpClientRequest) =>
810 Effect.sync(() => {
811 executeCount += 1;
812 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
813 });
814
815 return Effect.gen(function* () {
816 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
817 const result = yield* deliveries.processSignedJob(yield* signed);
818
819 expect(result).toMatchObject({
820 kind: "live_activity_update",
821 ok: true,
822 apnsStatus: null,
823 apnsReason: "Stale APNs delivery job skipped.",
824 });
825 expect(executeCount).toBe(0);
826 expect(attempts).toMatchObject([
827 {
828 sourceJobId: "job-update-superseded-by-running",
829 apnsReason: "Stale agent activity state skipped.",
830 },
831 ]);
832 }).pipe(
833 Effect.provide(
834 layerFor({
835 attempts,
836 activityStates: [
837 {
838 ...state,
839 phase: "running",
840 headline: "Running",
841 updatedAt: "1970-01-01T00:00:02.000Z",
842 },
843 ],
844 config: signingConfig,
845 execute,
846 }),
847 ),
848 );
849 });
850
851 it.effect("retries signed queue jobs that are already claimed but not completed", () => {
852 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
853 let executeCount = 0;
854 const payload = makeApnsDeliveryJobPayload({
855 kind: "push_notification",
856 userId: target.user_id,
857 deviceId: target.device_id,
858 token: "apns-device-token",
859 aggregate: null,
860 notification: {
861 title: "Thread",
862 body: "Input: Project",
863 environmentId: "env",
864 threadId: "thread",
865 deepLink: "/",
866 },
867 createdAt: "1970-01-01T00:00:00.000Z",
868 expiresAt: "1970-01-01T00:10:00.000Z",
869 jobId: "job-push-in-flight",
870 });
871 const signed = signApnsDeliveryJob({
872 secret: config.apnsDeliveryJobSigningSecret,
873 payload,
874 });
875 const execute = (request: HttpClientRequest.HttpClientRequest) =>
876 Effect.sync(() => {
877 executeCount += 1;
878 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
879 });
880
881 return Effect.gen(function* () {
882 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
883 const result = yield* Effect.exit(deliveries.processSignedJob(yield* signed));
884
885 expect(result._tag).toBe("Failure");
886 if (result._tag === "Failure") {
887 expect(result.cause.toString()).toContain("ApnsDeliveryJobClaimInFlight");
888 }
889 expect(executeCount).toBe(0);
890 expect(attempts).toEqual([]);
891 }).pipe(
892 Effect.provide(
893 layerFor({
894 attempts,
895 sourceJobClaims: new Map([["job-push-in-flight", "in_flight"]]),
896 config: signingConfig,
897 execute,
898 }),
899 ),
900 );
901 });
902
903 it.effect("invalidates dead device push tokens after permanent APNs alert failures", () => {
904 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
905 const invalidatedTokens: Array<
906 Parameters<LiveActivities.LiveActivities["Service"]["invalidateDeliveryToken"]>[0]
907 > = [];
908 const payload = makeApnsDeliveryJobPayload({
909 kind: "push_notification",
910 userId: target.user_id,
911 deviceId: target.device_id,
912 token: "apns-device-token",
913 aggregate: null,
914 notification: {
915 title: "Thread",
916 body: "Failed: Project",
917 environmentId: "env",
918 threadId: "thread",
919 deepLink: "/",
920 },
921 createdAt: "1970-01-01T00:00:00.000Z",
922 expiresAt: "1970-01-01T00:10:00.000Z",
923 jobId: "job-push-bad-token",
924 });
925 const signed = signApnsDeliveryJob({
926 secret: config.apnsDeliveryJobSigningSecret,
927 payload,
928 });
929 const execute = (request: HttpClientRequest.HttpClientRequest) =>
930 Effect.succeed(
931 HttpClientResponse.fromWeb(
932 request,
933 Response.json({ reason: "BadDeviceToken" }, { status: 400 }),
934 ),
935 );
936
937 return Effect.gen(function* () {
938 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
939 const result = yield* deliveries.processSignedJob(yield* signed);
940
941 expect(result.kind).toBe("push_notification");
942 expect(result.ok).toBe(false);
943 expect(result.apnsStatus).toBe(400);
944 expect(result.apnsReason).toBe("BadDeviceToken");
945 expect(invalidatedTokens).toMatchObject([
946 {
947 userId: target.user_id,
948 deviceId: target.device_id,
949 kind: "push_notification",
950 },
951 ]);
952 }).pipe(
953 Effect.provide(
954 layerFor({
955 attempts,
956 invalidatedTokens,
957 currentTargets: [
958 {
959 ...target,
960 push_token: "apns-device-token",
961 },
962 ],
963 config: signingConfig,
964 execute,
965 }),
966 ),
967 );
968 });
969
970 it.effect("invalidates dead Live Activity tokens after APNs unregisters them", () => {
971 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
972 const invalidatedTokens: Array<
973 Parameters<LiveActivities.LiveActivities["Service"]["invalidateDeliveryToken"]>[0]
974 > = [];
975 const payload = makeApnsDeliveryJobPayload({
976 kind: "live_activity_update",
977 userId: target.user_id,
978 deviceId: target.device_id,
979 token: target.activity_push_token ?? "activity-token",
980 aggregate,
981 createdAt: "1970-01-01T00:00:00.000Z",
982 expiresAt: "1970-01-01T00:10:00.000Z",
983 jobId: "job-update-unregistered",
984 });
985 const signed = signApnsDeliveryJob({
986 secret: config.apnsDeliveryJobSigningSecret,
987 payload,
988 });
989 const execute = (request: HttpClientRequest.HttpClientRequest) =>
990 Effect.succeed(
991 HttpClientResponse.fromWeb(
992 request,
993 Response.json({ reason: "Unregistered" }, { status: 410 }),
994 ),
995 );
996
997 return Effect.gen(function* () {
998 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
999 const result = yield* deliveries.processSignedJob(yield* signed);
1000
1001 expect(result.kind).toBe("live_activity_update");
1002 expect(result.ok).toBe(false);
1003 expect(result.apnsStatus).toBe(410);
1004 expect(result.apnsReason).toBe("Unregistered");
1005 expect(invalidatedTokens).toMatchObject([
1006 {
1007 userId: target.user_id,
1008 deviceId: target.device_id,
1009 kind: "live_activity_update",
1010 },
1011 ]);
1012 }).pipe(
1013 Effect.provide(layerFor({ attempts, invalidatedTokens, config: signingConfig, execute })),
1014 );
1015 });
1016});
1017
1018describe("live activity alert decisions", () => {
1019 const preferences = {
1020 liveActivitiesEnabled: true,
1021 notificationsEnabled: true,
1022 notifyOnApproval: true,
1023 notifyOnInput: true,
1024 notifyOnCompletion: true,
1025 notifyOnFailure: true,
1026 };
1027
1028 const attentionRow = {
1029 ...aggregate.activities[0]!,
1030 threadId: "thread-2" as RelayAgentActivityState["threadId"],
1031 threadTitle: "Blocked thread",
1032 phase: "waiting_for_approval" as const,
1033 status: "Approval",
1034 };
1035
1036 it("alerts when a thread newly enters an attention phase", () => {
1037 const alert = AgentActivityAlerts.alertForAttentionTransition({
1038 previousAggregate: aggregate,
1039 nextAggregate: {
1040 ...aggregate,
1041 activeCount: 2,
1042 activities: [...aggregate.activities, attentionRow],
1043 },
1044 preferences,
1045 });
1046 expect(alert).toEqual({ title: "Blocked thread", body: "Approval: Project" });
1047 });
1048
1049 it("stays silent without a delivered baseline so replays cannot buzz", () => {
1050 expect(
1051 AgentActivityAlerts.alertForAttentionTransition({
1052 previousAggregate: null,
1053 nextAggregate: { ...aggregate, activities: [attentionRow] },
1054 preferences,
1055 }),
1056 ).toBeNull();
1057 });
1058
1059 it("alerts when a previously active thread finishes mid-flight", () => {
1060 const doneRow = {
1061 ...aggregate.activities[0]!,
1062 phase: "completed" as const,
1063 status: "Done",
1064 };
1065 const next = {
1066 ...aggregate,
1067 activeCount: 0,
1068 activities: [attentionRow, doneRow],
1069 };
1070 expect(
1071 AgentActivityAlerts.alertForNewlyTerminal({
1072 previousAggregate: aggregate,
1073 nextAggregate: next,
1074 preferences,
1075 nowMs: 0,
1076 }),
1077 ).toEqual({ title: "Thread", body: "Done: Project" });
1078 expect(
1079 AgentActivityAlerts.alertForNewlyTerminal({
1080 previousAggregate: aggregate,
1081 nextAggregate: next,
1082 preferences: { ...preferences, notifyOnCompletion: false },
1083 nowMs: 0,
1084 }),
1085 ).toBeNull();
1086 expect(
1087 AgentActivityAlerts.alertForNewlyTerminal({
1088 previousAggregate: null,
1089 nextAggregate: next,
1090 preferences,
1091 nowMs: 0,
1092 }),
1093 ).toBeNull();
1094 expect(
1095 AgentActivityAlerts.alertForNewlyTerminal({
1096 previousAggregate: next,
1097 nextAggregate: next,
1098 preferences,
1099 nowMs: 0,
1100 }),
1101 ).toBeNull();
1102 expect(
1103 AgentActivityAlerts.alertForNewlyTerminal({
1104 previousAggregate: { ...aggregate, activities: [attentionRow] },
1105 nextAggregate: next,
1106 preferences,
1107 nowMs: 0,
1108 }),
1109 ).toBeNull();
1110 });
1111});
1112
1113describe("fast completion delivery", () => {
1114 it.effect("keeps a completion alert when work finishes before running delivery", () => {
1115 const queuedJobs: SignedApnsDeliveryJob[] = [];
1116 const old = {
1117 ...aggregate,
1118 activeCount: 0,
1119 activities: [
1120 {
1121 ...aggregate.activities[0]!,
1122 threadId: "old" as RelayAgentActivityState["threadId"],
1123 phase: "completed" as const,
1124 },
1125 ],
1126 };
1127 const device = { ...target, last_aggregate_json: JSON.stringify(old) };
1128 const done = {
1129 ...aggregate,
1130 activeCount: 0,
1131 activities: [{ ...aggregate.activities[0]!, phase: "completed" as const }],
1132 };
1133 return Effect.gen(function* () {
1134 const d = yield* ApnsDeliveries.ApnsDeliveries;
1135 yield* d.sendForTarget({ target: device, aggregate, nowMs: 0 });
1136 yield* d.sendForTarget({ target: device, aggregate: done, nowMs: 0 });
1137 expect(
1138 queuedJobs.some((x) => x.payload.alert !== null && x.payload.alert !== undefined),
1139 ).toBe(true);
1140 }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs, currentTargets: [device] })));
1141 });
1142 it.effect("replays a newly visible completion without alerting", () => {
1143 const queuedJobs: SignedApnsDeliveryJob[] = [];
1144 const done = {
1145 ...aggregate,
1146 activeCount: 0,
1147 activities: [{ ...aggregate.activities[0]!, phase: "completed" as const }],
1148 };
1149 const previous = {
1150 ...aggregate,
1151 activities: [
1152 { ...aggregate.activities[0]!, threadId: "other" as RelayAgentActivityState["threadId"] },
1153 ],
1154 };
1155 const device = { ...target, last_aggregate_json: JSON.stringify(previous) };
1156 return Effect.gen(function* () {
1157 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
1158 yield* deliveries.sendForTarget({
1159 target: device,
1160 aggregate: done,
1161 nowMs: 0,
1162 replay: true,
1163 });
1164 expect(queuedJobs).toHaveLength(1);
1165 expect(queuedJobs[0]?.payload.alert).toBeUndefined();
1166 }).pipe(Effect.provide(layerFor({ attempts: [], queuedJobs })));
1167 });
1168});
1169
1170describe("signed APNs registration metadata", () => {
1171 it.effect.each(
1172 (["live_activity_update", "push_notification"] as const).flatMap((kind) =>
1173 (["bundle", "environment", "legacy"] as const).map((changed) => ({ kind, changed })),
1174 ),
1175 )("routes $kind using current registration with $changed job metadata", ({ kind, changed }) => {
1176 const attempts: DeliveryAttempts.DeliveryAttemptInput[] = [];
1177 const requests: HttpClientRequest.HttpClientRequest[] = [];
1178 const payload = makeApnsDeliveryJobPayload({
1179 kind,
1180 userId: target.user_id,
1181 deviceId: target.device_id,
1182 token: "unchanged-token",
1183 ...(changed === "legacy"
1184 ? {}
1185 : { bundleId: "com.t3tools.t3code.dev", apsEnvironment: "sandbox" as const }),
1186 aggregate: kind === "live_activity_update" ? aggregate : null,
1187 ...(kind === "push_notification"
1188 ? {
1189 notification: {
1190 title: "Thread",
1191 body: "Input: Project",
1192 environmentId: "env",
1193 threadId: "thread",
1194 deepLink: "/",
1195 },
1196 }
1197 : {}),
1198 createdAt: "1970-01-01T00:00:00.000Z",
1199 expiresAt: "1970-01-01T00:10:00.000Z",
1200 jobId: `metadata-${kind}-${changed}`,
1201 });
1202 const signed = signApnsDeliveryJob({
1203 secret: config.apnsDeliveryJobSigningSecret,
1204 payload,
1205 });
1206 return Effect.gen(function* () {
1207 const deliveries = yield* ApnsDeliveries.ApnsDeliveries;
1208 const result = yield* deliveries.processSignedJob(yield* signed);
1209 expect(result.ok).toBe(true);
1210 expect(requests).toHaveLength(1);
1211 expect(requests[0]?.url).toBe(
1212 `${changed === "environment" ? "https://api.push.apple.com" : "https://api.sandbox.push.apple.com"}/3/device/unchanged-token`,
1213 );
1214 expect(requests[0]?.headers["apns-topic"]).toBe(
1215 `${changed === "bundle" ? "com.t3tools.t3code.preview" : "com.t3tools.t3code.dev"}${kind === "live_activity_update" ? ".push-type.liveactivity" : ""}`,
1216 );
1217 }).pipe(
1218 Effect.provide(
1219 layerFor({
1220 attempts,
1221 config: signingConfig,
1222 currentTargets: [
1223 {
1224 ...target,
1225 push_token: "unchanged-token",
1226 activity_push_token: "unchanged-token",
1227 bundle_id:
1228 changed === "bundle" ? "com.t3tools.t3code.preview" : "com.t3tools.t3code.dev",
1229 aps_environment: changed === "environment" ? "production" : "sandbox",
1230 },
1231 ],
1232 execute: (request) =>
1233 Effect.sync(() => {
1234 requests.push(request);
1235 return HttpClientResponse.fromWeb(request, new Response("", { status: 200 }));
1236 }),
1237 }),
1238 ),
1239 );
1240 });
1241});