Open History to see the last commit
1import { EnvironmentId, ThreadId } from "@t3tools/contracts";
2import type { RelayAgentActivityState } from "@t3tools/contracts/relay";
3import { describe, expect, it } from "@effect/vitest";
4import * as NodeCryptoLayer from "@effect/platform-node/NodeCrypto";
5import * as Effect from "effect/Effect";
6import * as DateTime from "effect/DateTime";
7import * as Layer from "effect/Layer";
8import * as Redacted from "effect/Redacted";
9import * as Schema from "effect/Schema";
10import * as Stream from "effect/Stream";
11import type * as Cloudflare from "@alchemy.run/cloudflare";
12import * as FcmDeliveryQueueConsumer from "./FcmDeliveryQueueConsumer.ts";
13
14import * as RelayConfiguration from "../Config.ts";
15import * as RelayDb from "../db.ts";
16import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
17import * as AgentActivityRows from "./AgentActivityRows.ts";
18import * as LiveActivities from "./LiveActivities.ts";
19import * as FcmDeliveryQueueSender from "./FcmDeliveryQueueSender.ts";
20import * as FcmClient from "./FcmClient.ts";
21import * as FcmDeliveries from "./FcmDeliveries.ts";
22import { androidActivityData, fitFcmData } from "./fcmPayloads.ts";
23import { makeAggregateState } from "./agentActivityAggregate.ts";
24
25const aggregateFor = (states: ReadonlyArray<RelayAgentActivityState>) =>
26 makeAggregateState({ activeStates: states, terminalState: null, nowMs: 0 })!;
27
28const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
29const state: RelayAgentActivityState = {
30 environmentId: EnvironmentId.make("env"),
31 threadId: ThreadId.make("thread"),
32 projectTitle: "Project",
33 threadTitle: "Fix notifications",
34 phase: "running",
35 headline: "Working",
36 modelTitle: "Codex",
37 updatedAt: "1970-01-01T00:00:00.000Z",
38 deepLink: "/threads/env/thread",
39};
40const preferences = {
41 notificationsEnabled: true,
42 liveActivitiesEnabled: true,
43 notifyOnApproval: true,
44 notifyOnInput: true,
45 notifyOnCompletion: true,
46 notifyOnFailure: true,
47};
48const target: LiveActivities.TargetRow = {
49 user_id: "user",
50 device_id: "phone",
51 platform: "android",
52 ios_major_version: null,
53 app_version: null,
54 bundle_id: "com.t3tools.t3code.dev",
55 aps_environment: null,
56 push_token: "fcm-token",
57 push_to_start_token: null,
58 preferences_json: encodeJson(preferences),
59 activity_push_token: null,
60 remote_start_queued_at: null,
61 remote_started_at: null,
62 ended_at: null,
63 last_aggregate_json: null,
64 last_live_activity_delivery_at: null,
65};
66const config = {
67 relayIssuer: "https://relay.test",
68 fcmServiceAccount: Redacted.make("configured"),
69 apns: {
70 environment: "sandbox",
71 teamId: "unused",
72 keyId: "unused",
73 bundleId: "unused",
74 privateKey: Redacted.make("unused"),
75 },
76 clerkSecretKey: Redacted.make("unused"),
77 clerkPublishableKey: "unused",
78 clerkJwtAudience: "unused",
79 apnsDeliveryJobSigningSecret: Redacted.make("unused"),
80 cloudMintPrivateKey: Redacted.make("unused"),
81 cloudMintPublicKey: "unused",
82 managedEndpointBaseDomain: undefined,
83 managedEndpointNamespace: undefined,
84} satisfies RelayConfiguration.RelayConfiguration["Service"];
85
86function harness() {
87 const sent: Array<Parameters<FcmClient.FcmClient["Service"]["send"]>[0]> = [];
88 const queued: FcmDeliveries.FcmDeliveryJob[] = [];
89 const marked: Array<Parameters<LiveActivities.LiveActivities["Service"]["markDelivery"]>[0]> = [];
90 const current = {
91 target: { ...target },
92 state: { ...state } as RelayAgentActivityState | null,
93 otherStates: [] as RelayAgentActivityState[],
94 mutedEnvironments: [] as string[],
95 notificationOnlyEnvironments: [] as string[],
96 revokedEnvironments: [] as string[],
97 linked: true,
98 deliveryFailure: null as FcmClient.FcmClientError | null,
99 };
100 const layerServices = Layer.mergeAll(
101 NodeCryptoLayer.layer,
102 Layer.succeed(RelayConfiguration.RelayConfiguration, config),
103 Layer.succeed(FcmDeliveryQueueSender.FcmDeliveryQueueSender, {
104 send: (job) =>
105 Effect.sync(() => {
106 queued.push(job);
107 }),
108 }),
109 Layer.succeed(FcmClient.FcmClient, {
110 send: (input) =>
111 Effect.suspend(() =>
112 current.deliveryFailure
113 ? Effect.fail(current.deliveryFailure)
114 : Effect.sync(() => {
115 sent.push(input);
116 return { unregistered: false };
117 }),
118 ),
119 }),
120 Layer.succeed(LiveActivities.LiveActivities, {
121 register: () => Effect.void,
122 listTargets: () => Effect.sync(() => [current.target]),
123 listIdleArmedTargets: () => Effect.succeed([]),
124 markDelivery: (input) =>
125 Effect.sync(() => {
126 marked.push(input);
127 current.target.last_aggregate_json = input.aggregate ? encodeJson(input.aggregate) : null;
128 }),
129
130 invalidateDeliveryToken: () => Effect.void,
131 }),
132 Layer.succeed(AgentActivityRows.AgentActivityRows, {
133 upsert: () => Effect.void,
134 remove: () => Effect.void,
135 pruneTerminal: () => Effect.void,
136 listForUser: () =>
137 Effect.sync(() =>
138 current.linked
139 ? [...(current.state ? [current.state] : []), ...current.otherStates].filter(
140 (row) =>
141 !current.revokedEnvironments.includes(row.environmentId) &&
142 !current.notificationOnlyEnvironments.includes(row.environmentId),
143 )
144 : [],
145 ),
146 getForUserThread: (input) =>
147 Effect.sync(() =>
148 current.linked
149 ? ([current.state, ...current.otherStates].find(
150 (row) =>
151 row?.environmentId === input.environmentId && row.threadId === input.threadId,
152 ) ?? null)
153 : null,
154 ),
155 }),
156 Layer.succeed(EnvironmentLinks.EnvironmentLinks, {
157 upsert: () => Effect.void,
158 listDeliveryUsersForEnvironment: (input) =>
159 Effect.sync(() =>
160 current.linked && !current.revokedEnvironments.includes(input.environmentId)
161 ? [
162 {
163 userId: "user",
164 notificationsEnabled: !current.mutedEnvironments.includes(input.environmentId),
165 liveActivitiesEnabled: !current.notificationOnlyEnvironments.includes(
166 input.environmentId,
167 ),
168 },
169 ]
170 : [],
171 ),
172 listForUser: () => Effect.succeed([]),
173 findActiveManagedForEnvironment: () => Effect.succeed([]),
174 setHoldWebhooksWhileOffline: () => Effect.void,
175 revokeForUser: () => Effect.succeed(false),
176 getForUser: (input) =>
177 Effect.sync(() =>
178 current.linked && !current.revokedEnvironments.includes(input.environmentId)
179 ? {
180 environmentId: EnvironmentId.make(input.environmentId),
181 label: "Desktop",
182 environmentPublicKey: "key",
183 linkedAt: state.updatedAt,
184 endpoint: {
185 httpBaseUrl: "https://env.test",
186 wsBaseUrl: "wss://env.test",
187 providerKind: "manual",
188 },
189 }
190 : null,
191 ),
192 }),
193 Layer.succeed(RelayDb.RelayDb, {} as RelayDb.RelayDb["Service"]),
194 );
195 return {
196 sent,
197 queued,
198 marked,
199 current,
200 layer: FcmDeliveries.layer.pipe(Layer.provide(layerServices)),
201 job: {
202 userId: "user",
203 deviceId: "phone",
204 token: "fcm-token",
205 state,
206 queuedAt: 0,
207 } satisfies FcmDeliveries.FcmDeliveryJob,
208 };
209}
210
211describe("Android delivery routing", () => {
212 const secondState: RelayAgentActivityState = {
213 ...state,
214 threadId: ThreadId.make("second-thread"),
215 threadTitle: "Second thread",
216 deepLink: "/threads/env/second-thread",
217 };
218
219 it.effect.each([
220 {
221 firstPhase: "waiting_for_approval",
222 secondPhase: "waiting_for_input",
223 title: "2 agents need attention",
224 active: "true",
225 },
226 { firstPhase: "completed", secondPhase: "failed", title: "2 agents finished", active: "false" },
227 ] as const)(
228 "routes grouped $firstPhase and $secondPhase to the overview once across their queued jobs",
229 ({ firstPhase, secondPhase, title, active }) => {
230 const h = harness();
231 h.current.otherStates = [secondState];
232 return Effect.gen(function* () {
233 const delivery = yield* FcmDeliveries.FcmDeliveries;
234 yield* delivery.process(h.job);
235 h.current.state = { ...state, phase: firstPhase };
236 h.current.otherStates = [{ ...secondState, phase: secondPhase }];
237 yield* delivery.process({ ...h.job, state: h.current.state });
238 yield* delivery.process({ ...h.job, state: h.current.otherStates[0] });
239 yield* delivery.process({ ...h.job, state: h.current.state });
240 const alerts = h.sent.filter((message) => message.alert);
241 expect(alerts).toHaveLength(1);
242 expect(alerts[0]?.data).toMatchObject({
243 alert_title: title,
244 alert_body: "Fix notifications, Second thread",
245 alert_path: "/",
246 active,
247 });
248 expect(h.marked.at(-1)?.aggregate?.activities).toHaveLength(2);
249 }).pipe(Effect.provide(h.layer));
250 },
251 );
252
253 it.effect("filters disabled event types before counting a group", () => {
254 const h = harness();
255 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, secondState]));
256 h.current.target.preferences_json = encodeJson({ ...preferences, notifyOnApproval: false });
257 h.current.state = { ...state, phase: "waiting_for_approval" };
258 h.current.otherStates = [{ ...secondState, phase: "waiting_for_input" }];
259 return Effect.gen(function* () {
260 const delivery = yield* FcmDeliveries.FcmDeliveries;
261 yield* delivery.process({ ...h.job, state: h.current.state });
262 expect(h.sent[0]?.data).toMatchObject({
263 alert_title: "Second thread",
264 alert_body: "Input: Project",
265 alert_path: "/threads/env/second-thread",
266 });
267 }).pipe(Effect.provide(h.layer));
268 });
269
270 it.effect("excludes already delivered attention and stale completions from groups", () => {
271 const h = harness();
272 h.current.state = { ...state, phase: "waiting_for_approval" };
273 h.current.target.last_aggregate_json = encodeJson(aggregateFor([h.current.state, secondState]));
274 h.current.otherStates = [
275 { ...secondState, phase: "completed", updatedAt: "1969-12-31T23:57:00.000Z" },
276 ];
277 return Effect.gen(function* () {
278 const delivery = yield* FcmDeliveries.FcmDeliveries;
279 yield* delivery.process({ ...h.job, state: h.current.state });
280 expect(h.sent.every((message) => !message.alert)).toBe(true);
281 }).pipe(Effect.provide(h.layer));
282 });
283
284 it.effect("prioritizes attention over simultaneous completions like iOS", () => {
285 const h = harness();
286 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, secondState]));
287 h.current.state = { ...state, phase: "completed" };
288 h.current.otherStates = [{ ...secondState, phase: "waiting_for_input" }];
289 return Effect.gen(function* () {
290 const delivery = yield* FcmDeliveries.FcmDeliveries;
291 yield* delivery.process({ ...h.job, state: h.current.state });
292 expect(h.sent[0]?.data).toMatchObject({
293 alert_title: "Second thread",
294 alert_body: "Input: Project",
295 });
296 yield* delivery.process({ ...h.job, state: h.current.otherStates[0] });
297 expect(h.sent.filter((message) => message.alert)).toHaveLength(1);
298 }).pipe(Effect.provide(h.layer));
299 });
300
301 it.effect.each(["completed", "waiting_for_approval", "waiting_for_input", "failed"] as const)(
302 "deleting one thread preserves another thread's %s alert",
303 (phase) => {
304 const h = harness();
305 h.current.otherStates = [secondState];
306 return Effect.gen(function* () {
307 const delivery = yield* FcmDeliveries.FcmDeliveries;
308 yield* delivery.process(h.job);
309 h.current.otherStates = [];
310 h.current.state = { ...state, phase };
311 yield* delivery.enqueue({ target, state: null });
312 yield* delivery.process(h.queued[0]);
313 expect(h.sent.at(-1)?.alert).toBe(false);
314 yield* delivery.process({ ...h.job, state: h.current.state });
315 expect(h.sent.filter((message) => message.alert)).toHaveLength(1);
316 yield* delivery.process({ ...h.job, state: h.current.state });
317 expect(h.sent.filter((message) => message.alert)).toHaveLength(1);
318 }).pipe(Effect.provide(h.layer));
319 },
320 );
321
322 it.effect("registration replay establishes a baseline without alerting", () => {
323 const h = harness();
324 h.current.state = { ...state, phase: "waiting_for_approval" };
325 h.current.otherStates = [{ ...secondState, phase: "waiting_for_input" }];
326 return Effect.gen(function* () {
327 const delivery = yield* FcmDeliveries.FcmDeliveries;
328 yield* delivery.enqueue({ target, state: null, replay: true });
329 yield* delivery.process(h.queued[0]);
330 yield* delivery.process({ ...h.job, state: h.current.state });
331 expect(h.sent.every((message) => !message.alert)).toBe(true);
332 expect(h.marked[0]?.aggregate?.activities).toHaveLength(2);
333 }).pipe(Effect.provide(h.layer));
334 });
335
336 it.effect.each(["mutedEnvironments", "revokedEnvironments"] as const)(
337 "excludes %s when forming cross-environment groups",
338 (restriction) => {
339 const h = harness();
340 const other = { ...secondState, environmentId: EnvironmentId.make("other-env") };
341 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state, other]));
342 h.current.state = { ...state, phase: "waiting_for_approval" };
343 h.current.otherStates = [{ ...other, phase: "waiting_for_input" }];
344 h.current[restriction] = [other.environmentId];
345 return Effect.gen(function* () {
346 const delivery = yield* FcmDeliveries.FcmDeliveries;
347 yield* delivery.process({ ...h.job, state: h.current.state });
348 expect(h.sent[0]?.data).toMatchObject({
349 alert_title: "Fix notifications",
350 alert_body: "Approval: Project",
351 });
352 }).pipe(Effect.provide(h.layer));
353 },
354 );
355
356 it("gives a group a stable retry identity independent of row order", () => {
357 const other = {
358 ...secondState,
359 environmentId: EnvironmentId.make("other-env"),
360 threadId: state.threadId,
361 };
362 const input = {
363 previousAggregate: aggregateFor([state, other]),
364 nextAggregate: aggregateFor([
365 { ...state, phase: "completed" },
366 { ...other, phase: "failed" },
367 ]),
368 preferences,
369 nowMs: 0,
370 };
371 const alert = FcmDeliveries.androidAlertForAggregate(input);
372 expect(alert?.alert_title).toBe("2 agents finished");
373 expect(
374 FcmDeliveries.androidAlertForAggregate({
375 ...input,
376 nextAggregate: {
377 ...input.nextAggregate,
378 activities: input.nextAggregate.activities.toReversed(),
379 },
380 })?.alert_id,
381 ).toBe(alert?.alert_id);
382 expect(
383 FcmDeliveries.androidAlertForAggregate({
384 ...input,
385 nextAggregate: {
386 ...input.nextAggregate,
387 activities: input.nextAggregate.activities.map((row) => ({
388 ...row,
389 updatedAt: "1970-01-01T00:00:01.000Z",
390 })),
391 },
392 })?.alert_id,
393 ).not.toBe(alert?.alert_id);
394 });
395
396 it("distinguishes matching thread IDs in different environments", () => {
397 const other = {
398 ...secondState,
399 environmentId: EnvironmentId.make("other-env"),
400 threadId: state.threadId,
401 };
402 const alreadyWaiting = { ...state, phase: "waiting_for_approval" as const };
403 expect(
404 FcmDeliveries.androidAlertForAggregate({
405 previousAggregate: aggregateFor([alreadyWaiting, other]),
406 nextAggregate: aggregateFor([alreadyWaiting, { ...other, phase: "waiting_for_input" }]),
407 preferences,
408 nowMs: 0,
409 }),
410 ).toMatchObject({ alert_title: "Second thread", alert_body: "Input: Project" });
411 });
412
413 it("keeps notification groups distinct when identifiers contain slashes", () => {
414 const left = FcmDeliveries.androidAlertForState(
415 {
416 ...state,
417 environmentId: EnvironmentId.make("a/b"),
418 threadId: ThreadId.make("c"),
419 phase: "completed",
420 },
421 preferences,
422 0,
423 );
424 const right = FcmDeliveries.androidAlertForState(
425 {
426 ...state,
427 environmentId: EnvironmentId.make("a"),
428 threadId: ThreadId.make("b/c"),
429 phase: "completed",
430 },
431 preferences,
432 0,
433 );
434 expect(left?.alert_group).not.toBe(right?.alert_group);
435 });
436
437 it.effect(
438 "queues Android devices and sends the latest aggregate instead of a stale running state",
439 () => {
440 const h = harness();
441 return Effect.gen(function* () {
442 const delivery = yield* FcmDeliveries.FcmDeliveries;
443 yield* delivery.enqueue({ target, state });
444 h.current.state = { ...state, phase: "completed" };
445 yield* delivery.process(h.queued[0]);
446 expect(h.sent).toHaveLength(0);
447 yield* delivery.process({ ...h.job, state: h.current.state });
448 expect(h.sent[0]?.data).toMatchObject({
449 active: "false",
450 alert_title: "Fix notifications",
451 alert_body: "Done: Project",
452 alert_path: "/threads/env/thread",
453 });
454 }).pipe(Effect.provide(h.layer));
455 },
456 );
457
458 it.effect("keeps completion alerts working with ongoing activity disabled", () => {
459 const h = harness();
460 h.current.state = { ...state, phase: "completed" };
461 h.current.target.preferences_json = encodeJson({
462 ...preferences,
463 liveActivitiesEnabled: false,
464 });
465 return Effect.gen(function* () {
466 const delivery = yield* FcmDeliveries.FcmDeliveries;
467 yield* delivery.process({ ...h.job, state: h.current.state });
468 expect(h.sent[0]?.data).toMatchObject({
469 active: "false",
470 alert_title: "Fix notifications",
471 alert_body: "Done: Project",
472 });
473 expect(h.sent[0]?.data.user_id).toBe("user");
474 }).pipe(Effect.provide(h.layer));
475 });
476
477 it.effect("drops jobs for rotated tokens, expired jobs, and revoked links", () => {
478 const h = harness();
479 return Effect.gen(function* () {
480 const delivery = yield* FcmDeliveries.FcmDeliveries;
481 yield* delivery.process({ ...h.job, token: "old-token" });
482 yield* delivery.process({ ...h.job, queuedAt: -400_000 });
483 h.current.linked = false;
484 yield* delivery.process(h.job);
485 expect(h.sent).toHaveLength(0);
486 }).pipe(Effect.provide(h.layer));
487 });
488
489 it.effect("ends an existing ongoing notification when the environment stops publishing", () => {
490 const h = harness();
491 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state]));
492 h.current.state = null;
493 return Effect.gen(function* () {
494 const delivery = yield* FcmDeliveries.FcmDeliveries;
495 yield* delivery.process({ ...h.job, state: null });
496 expect(h.sent[0]?.data).toMatchObject({ active: "false" });
497 expect(h.sent[0]?.alert).toBe(false);
498 expect(h.marked[0]?.kind).toBe("live_activity_end");
499 }).pipe(Effect.provide(h.layer));
500 });
501
502 it.effect("leaves iOS devices on their existing delivery path", () => {
503 const h = harness();
504 return Effect.gen(function* () {
505 const delivery = yield* FcmDeliveries.FcmDeliveries;
506 expect(yield* delivery.enqueue({ target: { ...target, platform: "ios" }, state })).toBeNull();
507 expect(h.queued).toHaveLength(0);
508 }).pipe(Effect.provide(h.layer));
509 });
510
511 it.effect(
512 "continues ongoing activity when alerts are disabled, with or without a previous card",
513 () => {
514 const h = harness();
515 h.current.state = { ...state, phase: "waiting_for_approval" };
516 h.current.target.preferences_json = encodeJson({
517 ...preferences,
518 notificationsEnabled: false,
519 });
520 return Effect.gen(function* () {
521 const delivery = yield* FcmDeliveries.FcmDeliveries;
522 yield* delivery.process({ ...h.job, state: h.current.state });
523 expect(h.sent).toHaveLength(1);
524 expect(h.sent[0]?.data.active).toBe("true");
525 expect(h.sent[0]?.alert).toBe(false);
526 expect(h.sent[0]?.data.alert_id).toBeUndefined();
527 h.current.target.preferences_json = encodeJson({ ...preferences, notifyOnApproval: false });
528 h.current.target.last_aggregate_json = encodeJson(aggregateFor([state]));
529 yield* delivery.process({ ...h.job, state: h.current.state });
530 expect(h.sent[1]?.data.active).toBe("true");
531 expect(h.sent[1]?.data.alert_id).toBeUndefined();
532 expect(h.marked.at(-1)?.kind).toBe("live_activity_update");
533 }).pipe(Effect.provide(h.layer));
534 },
535 );
536
537 it.effect("replays an empty card to repair an orphan without a delivery baseline", () => {
538 const h = harness();
539 h.current.state = null;
540 return Effect.gen(function* () {
541 const delivery = yield* FcmDeliveries.FcmDeliveries;
542 yield* delivery.process({ ...h.job, state: null });
543 expect(h.sent[0]?.data.activity_expires_at).toBe("0");
544 expect(h.sent[0]?.alert).toBe(false);
545 }).pipe(Effect.provide(h.layer));
546 });
547
548 it("fits five Unicode rows and a grouped alert in the FCM budget without corrupting text or routes", () => {
549 const longTitle = '🤖漢字"\\'.repeat(30);
550 const aggregate = aggregateFor(
551 Array.from({ length: 5 }, (_, i) => ({
552 ...state,
553 threadId: ThreadId.make(`thread-${i}`),
554 threadTitle: longTitle,
555 projectTitle: longTitle,
556 })),
557 );
558 const data = fitFcmData({
559 ...androidActivityData(aggregate),
560 t3_kind: "agent_activity",
561 device_id: "d".repeat(128),
562 user_id: "u".repeat(128),
563 updated_at: "1788780000000",
564 alert_id: "a".repeat(64),
565 alert_title: "5 agents finished",
566 alert_body: Array(5).fill(longTitle).join(", "),
567 alert_path: "/threads/env/thread",
568 });
569 expect(new TextEncoder().encode(JSON.stringify(data)).length).toBeLessThanOrEqual(3800);
570 expect(data.alert_path).toBe("/threads/env/thread");
571 expect(data.activity_path).toBe("/threads/env/thread");
572 expect(data.alert_id).toBe("a".repeat(64));
573 expect(data.activity_line_4).toContain("Working\t");
574 for (const value of Object.values(data))
575 expect(new TextDecoder().decode(new TextEncoder().encode(value))).toBe(value);
576 });
577});
578
579describe("delivery policy regressions", () => {
580 it.effect("alerts a quick completion even with a previous unrelated card", () => {
581 const h = harness();
582 const old = { ...state, threadId: ThreadId.make("old"), phase: "completed" as const };
583 h.current.target.last_aggregate_json = encodeJson(aggregateFor([old]));
584 h.current.otherStates = [old];
585 h.current.state = { ...state, phase: "completed" };
586 return Effect.gen(function* () {
587 const d = yield* FcmDeliveries.FcmDeliveries;
588 yield* d.process(h.job);
589 yield* d.process({ ...h.job, state: h.current.state });
590 expect(h.sent.filter((x) => x.alert)).toHaveLength(1);
591 }).pipe(Effect.provide(h.layer));
592 });
593});
594
595describe("notification-only environments", () => {
596 it.effect("alerts independently while another environment has a live card", () => {
597 const h = harness();
598 const other = {
599 ...state,
600 environmentId: EnvironmentId.make("other"),
601 threadId: ThreadId.make("other"),
602 };
603 h.current.target.last_aggregate_json = encodeJson(aggregateFor([other]));
604 h.current.otherStates = [other];
605 h.current.state = { ...state, phase: "waiting_for_input" };
606 h.current.notificationOnlyEnvironments = [state.environmentId];
607 return Effect.gen(function* () {
608 const deliveries = yield* FcmDeliveries.FcmDeliveries;
609 yield* deliveries.process({ ...h.job, state: h.current.state });
610 expect(h.sent).toHaveLength(1);
611 expect(h.sent[0]?.alert).toBe(true);
612 expect(h.sent[0]?.data.alert_path).toBe(state.deepLink);
613 }).pipe(Effect.provide(h.layer));
614 });
615});
616
617it("stops reducing five-character row fields and fits the remaining alert", () => {
618 const data = fitFcmData({
619 device_id: "x".repeat(3710),
620 activity_line_0: "Approval\taaaaa\tbbbbb",
621 alert_body: "y".repeat(200),
622 });
623 expect(data.activity_line_0).toBe("Approval\taaaaa\tbbbbb");
624 expect(new TextEncoder().encode(encodeJson(data)).length).toBeLessThanOrEqual(3800);
625});
626
627describe("FCM queue message isolation", () => {
628 it.effect.each(["invalid-job", "fcm-rejection"] as const)(
629 "retries only the %s message and delivers the rest of its batch",
630 (failure) => {
631 const h = harness();
632 const outcomes = new Map<string, "ack" | "retry">();
633 const message = (id: string, body: unknown): Cloudflare.Queues.Message<unknown> => ({
634 id,
635 body,
636 timestamp: DateTime.toDateUtc(DateTime.makeUnsafe(0)),
637 attempts: 1,
638 ack: () => {
639 if (!outcomes.has(id)) outcomes.set(id, "ack");
640 },
641 retry: () => {
642 if (!outcomes.has(id)) outcomes.set(id, "retry");
643 },
644 });
645 const batch = [
646 message("failed", failure === "invalid-job" ? {} : h.job),
647 message("healthy", h.job),
648 ];
649 return Effect.gen(function* () {
650 yield* Stream.fromIterable(batch).pipe(
651 Stream.tap((item) =>
652 Effect.sync(() => {
653 h.current.deliveryFailure =
654 item.id === "failed" && failure === "fcm-rejection"
655 ? new FcmClient.FcmClientError({ operation: "send", status: 400 })
656 : null;
657 }),
658 ),
659 Stream.runForEach(FcmDeliveryQueueConsumer.processMessage),
660 );
661 // Cloudflare ignores Alchemy's final batch acknowledgements for already-retried messages.
662 for (const item of batch) item.ack();
663 expect([...outcomes]).toEqual([
664 ["failed", "retry"],
665 ["healthy", "ack"],
666 ]);
667 expect(h.sent).toHaveLength(1);
668 expect(h.marked).toHaveLength(1);
669 }).pipe(Effect.provide(h.layer));
670 },
671 );
672});