1import type {
2 RelayAgentActivityAggregateState,
3 RelayAgentAwarenessPreferences,
4 RelayDeliveryKind,
5 RelayDeliveryResult,
6} from "@t3tools/contracts/relay";
7import {
8 RelayAgentActivityAggregateState as RelayAgentActivityAggregateStateSchema,
9 RelayAgentAwarenessPreferences as RelayAgentAwarenessPreferencesSchema,
10 RelayDeliveryKind as RelayDeliveryKindSchema,
11} from "@t3tools/contracts/relay";
12import * as Context from "effect/Context";
13import * as DateTime from "effect/DateTime";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Redacted from "effect/Redacted";
18import * as Schema from "effect/Schema";
19
20import {
21 isExpiredAgentActivityState,
22 isTerminalPhase,
23 notificationForActivity,
24 sanitizeAgentActivityAggregateState,
25 sanitizeApnsNotificationPayload,
26} from "./agentActivityPayloads.ts";
27import * as Apns from "./ApnsClient.ts";
28import {
29 ApnsDeliveryJobLiveActivityAggregateMissing,
30 ApnsDeliveryJobPushNotificationMissing,
31 ApnsDeliveryJobQueuePayloadInvalid,
32 type ApnsLiveActivityAlert,
33 type ApnsNotificationPayload,
34 SignedApnsDeliveryJob,
35 isApnsDeliveryJobVerificationError,
36 verifySignedApnsDeliveryJob,
37 type ApnsDeliveryJobVerificationError,
38} from "./apnsDeliveryJobs.ts";
39import * as AgentActivityRows from "./AgentActivityRows.ts";
40import * as DeliveryAttempts from "./DeliveryAttempts.ts";
41import * as LiveActivities from "./LiveActivities.ts";
42import * as RelayConfiguration from "../Config.ts";
43import * as ApnsDeliveryQueue from "./ApnsDeliveryQueue.ts";
44import { withSpanAttributes } from "../observability.ts";
45
46import {
47 alertForAttentionTransition,
48 alertForNewlyTerminal,
49 alertForTerminalAggregate,
50 newlyTerminalRows,
51 shouldAlertForActivity,
52} from "./agentActivityAlerts.ts";
53export {
54 alertForAttentionTransition,
55 alertForNewlyTerminal,
56 alertForTerminalAggregate,
57} from "./agentActivityAlerts.ts";
58
59const MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS = 15_000;
60// How long a just-armed card may sit with an empty aggregate before an end is
61// warranted; covers the gap between arming on send and the environment's
62// first publish reaching the relay.
63const FRESHLY_ARMED_GRACE_MS = 2 * 60 * 1_000;
64const PERMANENT_APNS_TOKEN_REASONS = new Set([
65 "BadDeviceToken",
66 "DeviceTokenNotForTopic",
67 "Unregistered",
68]);
69
70type LiveActivityDeliveryKind = Extract<
71 RelayDeliveryKind,
72 "live_activity_start" | "live_activity_update" | "live_activity_end"
73>;
74
75type ChosenLiveActivityDelivery =
76 | {
77 readonly kind: "live_activity_start" | "live_activity_update";
78 readonly token: string;
79 readonly aggregate: RelayAgentActivityAggregateState;
80 readonly alert: ApnsLiveActivityAlert | null;
81 }
82 | {
83 readonly kind: "live_activity_end";
84 readonly token: string;
85 readonly aggregate: RelayAgentActivityAggregateState | null;
86 readonly alert: ApnsLiveActivityAlert | null;
87 };
88
89type ChosenPushNotificationDelivery = {
90 readonly kind: "push_notification";
91 readonly token: string;
92 readonly notification: ApnsNotificationPayload;
93};
94
95type ChosenDelivery = ChosenLiveActivityDelivery | ChosenPushNotificationDelivery;
96
97export type ApnsDeliveryError =
98 | ApnsDeliveryQueue.ApnsDeliveryQueueError
99 | ApnsDeliveryJobVerificationError
100 | ApnsDeliveryJobClaimInFlight
101 | DeliveryAttempts.DeliveryAttemptRecordPersistenceError
102 | LiveActivities.LiveActivityTargetListPersistenceError
103 | LiveActivities.LiveActivityDeliveryMarkPersistenceError;
104
105export class ApnsDeliveryJobClaimInFlight extends Schema.TaggedError<ApnsDeliveryJobClaimInFlight>()(
106 "ApnsDeliveryJobClaimInFlight",
107 {
108 sourceJobId: Schema.String,
109 },
110) {
111 override get message(): string {
112 return `APNs delivery job '${this.sourceJobId}' is already in flight`;
113 }
114}
115
116export class ApnsDeliveryTransportError extends Schema.TaggedError<ApnsDeliveryTransportError>()(
117 "ApnsDeliveryTransportError",
118 {
119 deviceId: Schema.String,
120 kind: RelayDeliveryKindSchema,
121 sourceJobId: Schema.NullOr(Schema.String),
122 apnsErrorTag: Schema.Literals([
123 "ApnsJwtEncodingError",
124 "ApnsJwtSigningError",
125 "ApnsHttpRequestError",
126 ]),
127 requestStage: Schema.NullOr(Schema.Literals(["send", "read-response"])),
128 cause: Schema.Defect(),
129 },
130) {
131 override get message(): string {
132 return `APNs ${this.kind} delivery failed for device ${this.deviceId}.`;
133 }
134}
135
136export const isApnsDeliveryTransportError = Schema.is(ApnsDeliveryTransportError);
137
138const decodeRelayAgentActivityAggregateStateJson = Schema.decodeUnknownOption(
139 Schema.fromJsonString(RelayAgentActivityAggregateStateSchema),
140);
141const decodeRelayAgentAwarenessPreferencesJson = Schema.decodeUnknownOption(
142 Schema.fromJsonString(RelayAgentAwarenessPreferencesSchema),
143);
144const decodeSignedApnsDeliveryJob = Schema.decodeUnknownEffect(SignedApnsDeliveryJob);
145
146function parseAggregate(value: string | null): RelayAgentActivityAggregateState | null {
147 if (!value) {
148 return null;
149 }
150 return Option.getOrNull(decodeRelayAgentActivityAggregateStateJson(value));
151}
152
153function parsePreferences(value: string): RelayAgentAwarenessPreferences | null {
154 return Option.getOrNull(decodeRelayAgentAwarenessPreferencesJson(value));
155}
156
157function aggregateNeedsAttention(aggregate: RelayAgentActivityAggregateState): boolean {
158 return aggregate.activities.some(
159 (row) => row.phase === "waiting_for_approval" || row.phase === "waiting_for_input",
160 );
161}
162
163function shouldUpdateLiveActivity(input: {
164 readonly previousAggregate: RelayAgentActivityAggregateState | null;
165 readonly nextAggregate: RelayAgentActivityAggregateState;
166 readonly lastDeliveryAt: string | null;
167 readonly nowMs: number;
168}): boolean {
169 if (!input.previousAggregate) {
170 return true;
171 }
172 if (JSON.stringify(input.previousAggregate) === JSON.stringify(input.nextAggregate)) {
173 return false;
174 }
175 if (input.previousAggregate.activeCount !== input.nextAggregate.activeCount) {
176 return true;
177 }
178 if (aggregateNeedsAttention(input.nextAggregate)) {
179 return true;
180 }
181 // A thread finishing must never be throttled away: when a completion and a
182 // new start land in the same window, activeCount is unchanged and the Done
183 // transition (and its alert) would otherwise be suppressed.
184 if (newlyTerminalRows(input.previousAggregate, input.nextAggregate, true).length > 0) {
185 return true;
186 }
187 const lastDeliveryAtMs =
188 input.lastDeliveryAt === null
189 ? null
190 : Option.match(DateTime.make(input.lastDeliveryAt), {
191 onNone: () => Number.NaN,
192 onSome: (dt) => dt.epochMilliseconds,
193 });
194 return (
195 lastDeliveryAtMs === null ||
196 Number.isNaN(lastDeliveryAtMs) ||
197 input.nowMs - lastDeliveryAtMs >= MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS
198 );
199}
200
201// Completions replayed long after the fact (server restarts republish every
202// recently-finished thread) must not ring the device again.
203
204function notificationForAggregate(input: {
205 readonly target: LiveActivities.TargetRow;
206 readonly aggregate: RelayAgentActivityAggregateState | null;
207 readonly nowMs: number;
208}): ApnsNotificationPayload | null {
209 if (!input.target.push_token || input.aggregate === null) {
210 return null;
211 }
212 const preferences = parsePreferences(input.target.preferences_json);
213 if (!preferences?.notificationsEnabled) {
214 return null;
215 }
216 const activity = input.aggregate.activities[0];
217 if (!activity) {
218 return null;
219 }
220 if (!shouldAlertForActivity({ ...activity, preferences, nowMs: input.nowMs })) return null;
221 return notificationForActivity(activity);
222}
223
224// "suppressed" means a Live Activity owns this state but no update is due
225// (unchanged or throttled); callers must not fall back to an alert push, or
226// every republish of a waiting aggregate would ring the device.
227function chooseLiveActivityDelivery(input: {
228 readonly target: LiveActivities.TargetRow;
229 readonly aggregate: RelayAgentActivityAggregateState | null;
230 readonly nowMs: number;
231 readonly replay?: boolean;
232}): ChosenLiveActivityDelivery | "suppressed" | null {
233 const preferences = parsePreferences(input.target.preferences_json);
234 if (preferences?.liveActivitiesEnabled === false) {
235 return input.target.activity_push_token
236 ? {
237 kind: "live_activity_end",
238 token: input.target.activity_push_token,
239 aggregate: null,
240 alert: null,
241 }
242 : null;
243 }
244 // Activities are started by the app in the foreground, never remotely.
245 // Without a registered token there is nothing addressable; attention
246 // transitions fall back to the push notification channel until the user
247 // next arms the card from the app.
248 if (!input.target.activity_push_token) {
249 return null;
250 }
251 // An armed card always shows content: live agents, or recently finished
252 // ones (the publisher keeps Done/Failed rows in the aggregate for a
253 // while). A null aggregate means there is truly nothing left to show, so
254 // the card ends — arming is cheap now that the app re-arms on any open
255 // with content.
256 if (input.aggregate === null) {
257 // Except right after arming: the app arms the card the moment the user
258 // starts work, and the token registration's replay can land before the
259 // environment's first publish for the brand-new thread. Ending here
260 // would retire the token and orphan the card at its seed content, so a
261 // freshly armed card keeps its seed until real state arrives.
262 const armedAtMs = Option.match(
263 input.target.remote_started_at === null
264 ? Option.none()
265 : DateTime.make(input.target.remote_started_at),
266 { onNone: () => null, onSome: (dt) => dt.epochMilliseconds },
267 );
268 if (armedAtMs !== null && input.nowMs - armedAtMs < FRESHLY_ARMED_GRACE_MS) {
269 return null;
270 }
271 return {
272 kind: "live_activity_end",
273 token: input.target.activity_push_token,
274 aggregate: null,
275 alert: null,
276 };
277 }
278 const nextAggregate = input.aggregate;
279 const previousAggregate = parseAggregate(input.target.last_aggregate_json);
280 return shouldUpdateLiveActivity({
281 previousAggregate,
282 nextAggregate,
283 lastDeliveryAt: input.target.last_live_activity_delivery_at,
284 nowMs: input.nowMs,
285 })
286 ? {
287 kind: "live_activity_update",
288 token: input.target.activity_push_token,
289 aggregate: nextAggregate,
290 alert: input.replay
291 ? null
292 : (alertForAttentionTransition({
293 previousAggregate,
294 nextAggregate,
295 preferences,
296 }) ??
297 alertForNewlyTerminal({
298 previousAggregate,
299 nextAggregate,
300 preferences,
301 nowMs: input.nowMs,
302 includeUnobserved: true,
303 })),
304 }
305 : "suppressed";
306}
307
308function chooseDelivery(input: {
309 readonly target: LiveActivities.TargetRow;
310 readonly aggregate: RelayAgentActivityAggregateState | null;
311 readonly nowMs: number;
312 readonly replay?: boolean;
313}): ChosenDelivery | null {
314 const liveActivityDelivery = chooseLiveActivityDelivery(input);
315 if (liveActivityDelivery === "suppressed") {
316 return null;
317 }
318 if (liveActivityDelivery) {
319 return liveActivityDelivery;
320 }
321 const notification = input.replay ? null : notificationForAggregate(input);
322 return notification && input.target.push_token
323 ? {
324 kind: "push_notification",
325 token: input.target.push_token,
326 notification,
327 }
328 : null;
329}
330
331function deliveryEvent(kind: LiveActivityDeliveryKind): Apns.ApnsLiveActivityEvent {
332 switch (kind) {
333 case "live_activity_start":
334 return "start";
335 case "live_activity_update":
336 return "update";
337 case "live_activity_end":
338 return "end";
339 }
340}
341
342function isPermanentApnsTokenFailure(result: Apns.ApnsDeliveryResult): boolean {
343 return (
344 !result.ok &&
345 (result.status === 410 ||
346 (result.status === 400 &&
347 result.reason !== undefined &&
348 PERMANENT_APNS_TOKEN_REASONS.has(result.reason)))
349 );
350}
351
352function duplicateJobResult(input: {
353 readonly deviceId: string;
354 readonly kind: RelayDeliveryKind;
355}): RelayDeliveryResult {
356 return {
357 deviceId: input.deviceId,
358 kind: input.kind,
359 ok: true,
360 apnsStatus: null,
361 apnsReason: "Duplicate APNs delivery job skipped.",
362 apnsId: null,
363 };
364}
365
366function staleJobResult(input: {
367 readonly deviceId: string;
368 readonly kind: RelayDeliveryKind;
369}): RelayDeliveryResult {
370 return {
371 deviceId: input.deviceId,
372 kind: input.kind,
373 ok: true,
374 apnsStatus: null,
375 apnsReason: "Stale APNs delivery job skipped.",
376 apnsId: null,
377 };
378}
379
380function deliveryAttemptOutcome(result: Apns.ApnsDeliveryResult) {
381 return {
382 ...(result.status === 0 ? {} : { apnsStatus: result.status }),
383 ...(result.reason === undefined ? {} : { apnsReason: result.reason }),
384 apnsId: result.apnsId,
385 ...(result.status === 0 ? { transportError: result.reason ?? "APNs request failed." } : {}),
386 };
387}
388
389const recoverApnsDeliveryTransportError = (
390 input: {
391 readonly deviceId: string;
392 readonly kind: RelayDeliveryKind;
393 readonly sourceJobId: string | null;
394 },
395 cause: Apns.ApnsError,
396): Effect.Effect<Apns.ApnsDeliveryResult> => {
397 const error = new ApnsDeliveryTransportError({
398 deviceId: input.deviceId,
399 kind: input.kind,
400 sourceJobId: input.sourceJobId,
401 apnsErrorTag: cause._tag,
402 requestStage: cause._tag === "ApnsHttpRequestError" ? cause.stage : null,
403 cause,
404 });
405 return Effect.logError(error.message).pipe(
406 Effect.annotateLogs({
407 error: Redacted.make(error, { label: error._tag }),
408 "error.type": error._tag,
409 "error.apns_error_tag": error.apnsErrorTag,
410 ...(error.requestStage === null ? {} : { "error.request_stage": error.requestStage }),
411 ...(error.stack === undefined ? {} : { "error.stack": error.stack }),
412 "relay.mobile.device_id": error.deviceId,
413 "relay.delivery.kind": error.kind,
414 ...(error.sourceJobId === null ? {} : { "relay.delivery.job_id": error.sourceJobId }),
415 }),
416 Effect.as({
417 ok: false,
418 status: 0,
419 reason: cause.message,
420 apnsId: null,
421 }),
422 );
423};
424
425interface LiveActivityDeliveryTarget {
426 readonly user_id: string;
427 readonly device_id: string;
428 readonly bundle_id?: string | null;
429 readonly aps_environment?: "sandbox" | "production" | null;
430}
431
432// Devices register the bundle id and APS environment of the build they run
433// (dev/preview/prod variants have distinct bundle ids; development-signed
434// builds get sandbox tokens). Sending with mismatched routing yields
435// DeviceTokenNotForTopic/BadDeviceToken, so per-device values override the
436// relay-wide defaults when present.
437function credentialsForTarget(
438 credentials: RelayConfiguration.ApnsCredentials,
439 target: LiveActivityDeliveryTarget,
440): RelayConfiguration.ApnsCredentials {
441 return {
442 ...credentials,
443 ...(target.bundle_id ? { bundleId: target.bundle_id } : {}),
444 ...(target.aps_environment ? { environment: target.aps_environment } : {}),
445 };
446}
447
448function expectedCurrentToken(input: {
449 readonly target: LiveActivities.TargetRow;
450 readonly kind: RelayDeliveryKind;
451}): string | null {
452 switch (input.kind) {
453 case "live_activity_start":
454 return input.target.push_to_start_token;
455 case "live_activity_update":
456 case "live_activity_end":
457 return input.target.activity_push_token;
458 case "push_notification":
459 return input.target.push_token;
460 }
461}
462
463interface SendLiveActivityDeliveryInputBase {
464 readonly target: LiveActivityDeliveryTarget;
465 readonly token: string;
466 readonly sourceJobId?: string | null;
467}
468
469export type SendLiveActivityDeliveryInput =
470 | (SendLiveActivityDeliveryInputBase & {
471 readonly kind: "live_activity_start" | "live_activity_update";
472 readonly aggregate: RelayAgentActivityAggregateState;
473 readonly alert?: ApnsLiveActivityAlert | null;
474 })
475 | (SendLiveActivityDeliveryInputBase & {
476 readonly kind: "live_activity_end";
477 readonly aggregate: RelayAgentActivityAggregateState | null;
478 readonly alert?: ApnsLiveActivityAlert | null;
479 });
480
481function makeLiveActivityDeliveryRequest(
482 apns: Apns.ApnsClient["Service"],
483 input: SendLiveActivityDeliveryInput,
484 now: DateTime.DateTime,
485) {
486 const epochSeconds = Math.floor(now.epochMilliseconds / 1_000);
487 const base = {
488 token: input.token,
489 nowEpochSeconds: epochSeconds,
490 nowIso: DateTime.formatIso(now),
491 };
492 switch (input.kind) {
493 case "live_activity_start":
494 case "live_activity_update":
495 return {
496 epochSeconds,
497 iso: base.nowIso,
498 request: apns.makeLiveActivityRequest({
499 ...base,
500 event: deliveryEvent(input.kind),
501 state: input.aggregate,
502 alert: input.alert ?? null,
503 }),
504 };
505 case "live_activity_end":
506 return {
507 epochSeconds,
508 iso: base.nowIso,
509 request: apns.makeLiveActivityRequest({
510 ...base,
511 event: "end",
512 state: input.aggregate,
513 alert: input.alert ?? null,
514 }),
515 };
516 }
517}
518
519export class ApnsDeliveries extends Context.Service<
520 ApnsDeliveries,
521 {
522 readonly sendForTarget: (input: {
523 readonly target: LiveActivities.TargetRow;
524 readonly aggregate: RelayAgentActivityAggregateState | null;
525 readonly nowMs: number;
526 readonly replay?: boolean;
527 }) => Effect.Effect<RelayDeliveryResult | null, ApnsDeliveryError>;
528 readonly sendPushNotificationForTarget: (input: {
529 readonly target: LiveActivities.TargetRow;
530 readonly aggregate: RelayAgentActivityAggregateState | null;
531 }) => Effect.Effect<RelayDeliveryResult | null, ApnsDeliveryError>;
532 readonly sendLiveActivity: (
533 input: SendLiveActivityDeliveryInput,
534 ) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;
535 readonly processSignedJob: (
536 body: unknown,
537 ) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;
538 readonly sendPushNotification: (input: {
539 readonly target: LiveActivityDeliveryTarget;
540 readonly token: string;
541 readonly sourceJobId?: string | null;
542 readonly notification: ApnsNotificationPayload;
543 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;
544 }
545>()("t3code-relay/agentActivity/ApnsDeliveries") {}
546
547export const make = Effect.gen(function* () {
548 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
549 const liveActivities = yield* LiveActivities.LiveActivities;
550 const deliveryQueue = yield* ApnsDeliveryQueue.ApnsDeliveryQueue;
551 const config = yield* RelayConfiguration.RelayConfiguration;
552 const apns = yield* Apns.ApnsClient;
553 const activityRows = yield* AgentActivityRows.AgentActivityRows;
554
555 // Start jobs are decided at publish time, but consecutive publishes land in
556 // the same queue batch: a start chosen from a running aggregate can be
557 // delivered moments after a newer terminal publish already ended the user's
558 // work, birthing an orphan activity that shows stale content forever (no
559 // token is ever registered for it, so nothing can update or end it).
560 // Re-validate at delivery time that the user still has live work; fail open
561 // on persistence errors so a database hiccup never drops a legitimate start.
562 const userStillHasLiveWork = Effect.fnUntraced(function* (userId: string) {
563 const now = yield* DateTime.now;
564 return yield* activityRows.listForUser({ userId }).pipe(
565 Effect.map((states) =>
566 states.some(
567 (state) =>
568 !isTerminalPhase(state) && !isExpiredAgentActivityState(state, now.epochMilliseconds),
569 ),
570 ),
571 Effect.catchCause((cause) =>
572 Effect.logWarning("live-work recheck failed; allowing queued start", { cause }).pipe(
573 Effect.as(true),
574 ),
575 ),
576 );
577 });
578
579 const stateIdentityIsCurrent = Effect.fnUntraced(function* (input: {
580 readonly userId: string;
581 readonly environmentId: string;
582 readonly threadId: string;
583 readonly phase: RelayAgentActivityAggregateState["activities"][number]["phase"];
584 readonly updatedAt: string;
585 }) {
586 return yield* activityRows
587 .getForUserThread({
588 userId: input.userId,
589 environmentId: input.environmentId,
590 threadId: input.threadId,
591 })
592 .pipe(
593 Effect.map(
594 (current) =>
595 current !== null &&
596 current.phase === input.phase &&
597 current.updatedAt === input.updatedAt,
598 ),
599 // A transient persistence failure must not permanently discard a
600 // legitimate alert. Fail open and let the signed job's retry/dedupe
601 // protections handle transport failures as usual.
602 Effect.catchCause((cause) =>
603 Effect.logWarning("agent-activity state recheck failed; allowing queued delivery", {
604 cause,
605 environmentId: input.environmentId,
606 threadId: input.threadId,
607 }).pipe(Effect.as(true)),
608 ),
609 );
610 });
611
612 const aggregateRowsAreCurrent = Effect.fnUntraced(function* (input: {
613 readonly userId: string;
614 readonly aggregate: RelayAgentActivityAggregateState;
615 }) {
616 return yield* activityRows.listForUser({ userId: input.userId }).pipe(
617 Effect.map((currentStates) => {
618 const currentByThread = new Map(
619 currentStates.map((current) => [
620 `${current.environmentId}\u0000${current.threadId}`,
621 current,
622 ]),
623 );
624 return input.aggregate.activities.every((row) => {
625 const current = currentByThread.get(`${row.environmentId}\u0000${row.threadId}`);
626 return (
627 current !== undefined &&
628 current.phase === row.phase &&
629 current.updatedAt === row.updatedAt
630 );
631 });
632 }),
633 Effect.catchCause((cause) =>
634 Effect.logWarning("agent-activity aggregate recheck failed; allowing queued delivery", {
635 cause,
636 userId: input.userId,
637 }).pipe(Effect.as(true)),
638 ),
639 );
640 });
641
642 const notificationStateIsCurrent = Effect.fnUntraced(function* (input: {
643 readonly userId: string;
644 readonly notification: ApnsNotificationPayload;
645 }) {
646 // Jobs from older relay versions do not carry a state identity. Preserve
647 // backwards compatibility and only revalidate newly queued jobs.
648 if (input.notification.phase === undefined || input.notification.updatedAt === undefined) {
649 return true;
650 }
651 return yield* stateIdentityIsCurrent({
652 userId: input.userId,
653 environmentId: input.notification.environmentId,
654 threadId: input.notification.threadId,
655 phase: input.notification.phase,
656 updatedAt: input.notification.updatedAt,
657 });
658 });
659
660 const currentSignedJobTarget = Effect.fnUntraced(function* (input: {
661 readonly target: LiveActivityDeliveryTarget;
662 readonly kind: RelayDeliveryKind;
663 readonly token: string;
664 }) {
665 return yield* liveActivities.listTargets({ userId: input.target.user_id }).pipe(
666 Effect.map((targets) => {
667 const currentTarget = targets.find((row) => row.device_id === input.target.device_id);
668 return currentTarget &&
669 expectedCurrentToken({ target: currentTarget, kind: input.kind }) === input.token
670 ? currentTarget
671 : null;
672 }),
673 );
674 });
675
676 const sendLiveActivity: ApnsDeliveries["Service"]["sendLiveActivity"] = Effect.fn(
677 "relay.apns_deliveries.send_live_activity",
678 )(function* (input) {
679 if (!config.apns) {
680 return {
681 deviceId: input.target.device_id,
682 kind: input.kind,
683 ok: false,
684 apnsStatus: null,
685 apnsReason: "APNs is disabled for this relay.",
686 apnsId: null,
687 };
688 }
689 yield* Effect.annotateCurrentSpan({
690 "relay.mobile.device_id": input.target.device_id,
691 "relay.delivery.kind": input.kind,
692 ...(input.sourceJobId ? { "relay.delivery.job_id": input.sourceJobId } : {}),
693 });
694 let deliveryTarget = input.target;
695 const now = yield* DateTime.now;
696 const aggregate =
697 input.aggregate === null ? null : sanitizeAgentActivityAggregateState(input.aggregate);
698 let alert = input.alert ?? null;
699 const recoverTransportError = (cause: Apns.ApnsError) =>
700 recoverApnsDeliveryTransportError(
701 {
702 deviceId: input.target.device_id,
703 kind: input.kind,
704 sourceJobId: input.sourceJobId ?? null,
705 },
706 cause,
707 );
708 if (input.sourceJobId) {
709 const claim = yield* attempts.claimSourceJob({
710 userId: input.target.user_id,
711 environmentId: null,
712 threadId: null,
713 deviceId: input.target.device_id,
714 kind: input.kind,
715 sourceJobId: input.sourceJobId,
716 token: input.token,
717 });
718 if (claim === "completed") {
719 return duplicateJobResult({ deviceId: input.target.device_id, kind: input.kind });
720 }
721 if (claim === "in_flight") {
722 return yield* new ApnsDeliveryJobClaimInFlight({ sourceJobId: input.sourceJobId });
723 }
724 const currentTarget = yield* currentSignedJobTarget({
725 target: input.target,
726 kind: input.kind,
727 token: input.token,
728 });
729 if (!currentTarget) {
730 yield* attempts.completeSourceJob({
731 sourceJobId: input.sourceJobId,
732 apnsReason: "Stale APNs delivery job skipped.",
733 });
734 return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
735 }
736 deliveryTarget = currentTarget;
737 if (alert) {
738 const preferences = parsePreferences(currentTarget.preferences_json);
739 const previousAggregate = parseAggregate(currentTarget.last_aggregate_json);
740 alert =
741 !preferences?.notificationsEnabled || !aggregate
742 ? null
743 : (alertForAttentionTransition({
744 previousAggregate,
745 nextAggregate: aggregate,
746 preferences,
747 }) ??
748 alertForNewlyTerminal({
749 previousAggregate,
750 nextAggregate: aggregate,
751 preferences,
752 nowMs: now.epochMilliseconds,
753 includeUnobserved: true,
754 }));
755 }
756 if (
757 input.kind !== "live_activity_start" &&
758 aggregate !== null &&
759 !(yield* aggregateRowsAreCurrent({
760 userId: input.target.user_id,
761 aggregate,
762 }))
763 ) {
764 yield* attempts.completeSourceJob({
765 sourceJobId: input.sourceJobId,
766 apnsReason: "Stale agent activity state skipped.",
767 });
768 return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
769 }
770 }
771 if (
772 input.kind === "live_activity_start" &&
773 !(yield* userStillHasLiveWork(input.target.user_id))
774 ) {
775 yield* liveActivities.clearStartQueued({
776 userId: input.target.user_id,
777 deviceId: input.target.device_id,
778 });
779 if (input.sourceJobId) {
780 yield* attempts.completeSourceJob({
781 sourceJobId: input.sourceJobId,
782 apnsReason: "Stale APNs start job skipped.",
783 });
784 }
785 return staleJobResult({ deviceId: input.target.device_id, kind: input.kind });
786 }
787 const { epochSeconds, iso, request } = makeLiveActivityDeliveryRequest(
788 apns,
789 { ...input, aggregate, alert } as SendLiveActivityDeliveryInput,
790 now,
791 );
792 const result = yield* apns
793 .sendLiveActivityRequest({
794 credentials: credentialsForTarget(config.apns, deliveryTarget),
795 request,
796 issuedAtUnixSeconds: epochSeconds,
797 })
798 .pipe(
799 Effect.catchTags({
800 ApnsJwtEncodingError: recoverTransportError,
801 ApnsJwtSigningError: recoverTransportError,
802 ApnsHttpRequestError: recoverTransportError,
803 }),
804 );
805 if (result.ok) {
806 yield* liveActivities.markDelivery({
807 userId: input.target.user_id,
808 deviceId: input.target.device_id,
809 kind: input.kind,
810 aggregate,
811 deliveredAt: iso,
812 });
813 } else if (isPermanentApnsTokenFailure(result)) {
814 yield* liveActivities.invalidateDeliveryToken({
815 userId: input.target.user_id,
816 deviceId: input.target.device_id,
817 kind: input.kind,
818 invalidatedAt: iso,
819 });
820 } else if (input.kind === "live_activity_start") {
821 yield* liveActivities.clearStartQueued({
822 userId: input.target.user_id,
823 deviceId: input.target.device_id,
824 });
825 }
826 if (input.sourceJobId) {
827 yield* attempts.completeSourceJob({
828 sourceJobId: input.sourceJobId,
829 ...deliveryAttemptOutcome(result),
830 });
831 } else {
832 yield* attempts.record({
833 userId: input.target.user_id,
834 environmentId: null,
835 threadId: null,
836 deviceId: input.target.device_id,
837 kind: input.kind,
838 token: input.token,
839 ...deliveryAttemptOutcome(result),
840 });
841 }
842 return {
843 deviceId: input.target.device_id,
844 kind: input.kind,
845 ok: result.ok,
846 apnsStatus: result.status === 0 ? null : result.status,
847 apnsReason: result.reason ?? null,
848 apnsId: result.apnsId,
849 };
850 });
851
852 const sendPushNotification: ApnsDeliveries["Service"]["sendPushNotification"] = Effect.fn(
853 "relay.apns_deliveries.send_push_notification",
854 )(function* (input) {
855 if (!config.apns) {
856 return {
857 deviceId: input.target.device_id,
858 kind: "push_notification",
859 ok: false,
860 apnsStatus: null,
861 apnsReason: "APNs is disabled for this relay.",
862 apnsId: null,
863 };
864 }
865 yield* Effect.annotateCurrentSpan({
866 "relay.mobile.device_id": input.target.device_id,
867 "relay.delivery.kind": "push_notification",
868 ...(input.sourceJobId ? { "relay.delivery.job_id": input.sourceJobId } : {}),
869 });
870 let deliveryTarget = input.target;
871 const now = yield* DateTime.now;
872 const epochSeconds = Math.floor(now.epochMilliseconds / 1_000);
873 const notification = sanitizeApnsNotificationPayload(input.notification);
874 yield* Effect.annotateCurrentSpan({
875 "relay.environment_id": notification.environmentId,
876 "relay.thread_id": notification.threadId,
877 });
878 const request = apns.makePushNotificationRequest({
879 token: input.token,
880 notification,
881 });
882 const recoverTransportError = (cause: Apns.ApnsError) =>
883 recoverApnsDeliveryTransportError(
884 {
885 deviceId: input.target.device_id,
886 kind: "push_notification",
887 sourceJobId: input.sourceJobId ?? null,
888 },
889 cause,
890 );
891 if (input.sourceJobId) {
892 const claim = yield* attempts.claimSourceJob({
893 userId: input.target.user_id,
894 environmentId: notification.environmentId,
895 threadId: notification.threadId,
896 deviceId: input.target.device_id,
897 kind: "push_notification",
898 sourceJobId: input.sourceJobId,
899 token: input.token,
900 });
901 if (claim === "completed") {
902 return duplicateJobResult({
903 deviceId: input.target.device_id,
904 kind: "push_notification",
905 });
906 }
907 if (claim === "in_flight") {
908 return yield* new ApnsDeliveryJobClaimInFlight({ sourceJobId: input.sourceJobId });
909 }
910 const currentTarget = yield* currentSignedJobTarget({
911 target: input.target,
912 kind: "push_notification",
913 token: input.token,
914 });
915 if (!currentTarget) {
916 yield* attempts.completeSourceJob({
917 sourceJobId: input.sourceJobId,
918 apnsReason: "Stale APNs delivery job skipped.",
919 });
920 return staleJobResult({
921 deviceId: input.target.device_id,
922 kind: "push_notification",
923 });
924 }
925 if (
926 !(yield* notificationStateIsCurrent({
927 userId: input.target.user_id,
928 notification,
929 }))
930 ) {
931 yield* attempts.completeSourceJob({
932 sourceJobId: input.sourceJobId,
933 apnsReason: "Stale agent activity state skipped.",
934 });
935 return staleJobResult({
936 deviceId: input.target.device_id,
937 kind: "push_notification",
938 });
939 }
940 deliveryTarget = currentTarget;
941 const preferences = parsePreferences(currentTarget.preferences_json);
942 const alertAllowed =
943 notification.phase !== undefined && notification.updatedAt !== undefined
944 ? shouldAlertForActivity({
945 ...notification,
946 phase: notification.phase,
947 updatedAt: notification.updatedAt,
948 preferences,
949 nowMs: now.epochMilliseconds,
950 })
951 : preferences?.notificationsEnabled === true;
952 if (!alertAllowed) {
953 yield* attempts.completeSourceJob({
954 sourceJobId: input.sourceJobId,
955 apnsReason: "Notification is disabled or no longer fresh.",
956 });
957 return staleJobResult({ deviceId: input.target.device_id, kind: "push_notification" });
958 }
959 }
960 const result = yield* apns
961 .sendPushNotificationRequest({
962 credentials: credentialsForTarget(config.apns, deliveryTarget),
963 request,
964 issuedAtUnixSeconds: epochSeconds,
965 })
966 .pipe(
967 Effect.catchTags({
968 ApnsJwtEncodingError: recoverTransportError,
969 ApnsJwtSigningError: recoverTransportError,
970 ApnsHttpRequestError: recoverTransportError,
971 }),
972 );
973 if (isPermanentApnsTokenFailure(result)) {
974 yield* liveActivities.invalidateDeliveryToken({
975 userId: input.target.user_id,
976 deviceId: input.target.device_id,
977 kind: "push_notification",
978 invalidatedAt: DateTime.formatIso(now),
979 });
980 }
981 if (input.sourceJobId) {
982 yield* attempts.completeSourceJob({
983 sourceJobId: input.sourceJobId,
984 ...deliveryAttemptOutcome(result),
985 });
986 } else {
987 yield* attempts.record({
988 userId: input.target.user_id,
989 environmentId: notification.environmentId,
990 threadId: notification.threadId,
991 deviceId: input.target.device_id,
992 kind: "push_notification",
993 token: input.token,
994 ...deliveryAttemptOutcome(result),
995 });
996 }
997 return {
998 deviceId: input.target.device_id,
999 kind: "push_notification" as const,
1000 ok: result.ok,
1001 apnsStatus: result.status === 0 ? null : result.status,
1002 apnsReason: result.reason ?? null,
1003 apnsId: result.apnsId,
1004 };
1005 });
1006
1007 const processSignedJob: ApnsDeliveries["Service"]["processSignedJob"] = Effect.fn(
1008 "relay.apns_deliveries.process_signed_job",
1009 )(function* (body) {
1010 const signedJob = yield* decodeSignedApnsDeliveryJob(body).pipe(
1011 Effect.mapError(
1012 (cause) =>
1013 new ApnsDeliveryJobQueuePayloadInvalid({
1014 receivedType: Array.isArray(body) ? "array" : body === null ? "null" : typeof body,
1015 cause,
1016 }),
1017 ),
1018 );
1019 const now = yield* DateTime.now;
1020 const payload = verifySignedApnsDeliveryJob({
1021 secret: config.apnsDeliveryJobSigningSecret,
1022 job: signedJob,
1023 nowMs: now.epochMilliseconds,
1024 });
1025 if (isApnsDeliveryJobVerificationError(payload)) {
1026 return yield* payload;
1027 }
1028 yield* Effect.annotateCurrentSpan({
1029 "relay.mobile.device_id": payload.target.deviceId,
1030 "relay.delivery.kind": payload.kind,
1031 "relay.delivery.job_id": payload.jobId,
1032 });
1033 return yield* Effect.suspend(() => {
1034 switch (payload.kind) {
1035 case "live_activity_start":
1036 case "live_activity_update":
1037 if (payload.aggregate === null) {
1038 return Effect.fail(
1039 new ApnsDeliveryJobLiveActivityAggregateMissing({
1040 jobId: payload.jobId,
1041 kind: payload.kind,
1042 userId: payload.target.userId,
1043 deviceId: payload.target.deviceId,
1044 }),
1045 );
1046 }
1047 return sendLiveActivity({
1048 target: {
1049 user_id: payload.target.userId,
1050 device_id: payload.target.deviceId,
1051 bundle_id: payload.target.bundleId ?? null,
1052 aps_environment: payload.target.apsEnvironment ?? null,
1053 },
1054 token: payload.target.token,
1055 sourceJobId: payload.jobId,
1056 kind: payload.kind,
1057 aggregate: payload.aggregate,
1058 alert: payload.alert ?? null,
1059 });
1060 case "live_activity_end":
1061 return sendLiveActivity({
1062 target: {
1063 user_id: payload.target.userId,
1064 device_id: payload.target.deviceId,
1065 bundle_id: payload.target.bundleId ?? null,
1066 aps_environment: payload.target.apsEnvironment ?? null,
1067 },
1068 token: payload.target.token,
1069 sourceJobId: payload.jobId,
1070 kind: payload.kind,
1071 aggregate: payload.aggregate,
1072 alert: payload.alert ?? null,
1073 });
1074 case "push_notification":
1075 if (payload.notification === null) {
1076 return Effect.fail(
1077 new ApnsDeliveryJobPushNotificationMissing({
1078 jobId: payload.jobId,
1079 userId: payload.target.userId,
1080 deviceId: payload.target.deviceId,
1081 }),
1082 );
1083 }
1084 return sendPushNotification({
1085 target: {
1086 user_id: payload.target.userId,
1087 device_id: payload.target.deviceId,
1088 bundle_id: payload.target.bundleId ?? null,
1089 aps_environment: payload.target.apsEnvironment ?? null,
1090 },
1091 token: payload.target.token,
1092 sourceJobId: payload.jobId,
1093 notification: payload.notification,
1094 });
1095 }
1096 }).pipe(withSpanAttributes({ "user.id": payload.target.userId }));
1097 });
1098
1099 return ApnsDeliveries.of({
1100 sendLiveActivity,
1101 sendPushNotification,
1102 processSignedJob,
1103 sendPushNotificationForTarget: Effect.fnUntraced(function* (input) {
1104 if (!config.apns) return null;
1105 const now = yield* DateTime.now;
1106 const notification = notificationForAggregate({
1107 target: input.target,
1108 aggregate: input.aggregate,
1109 nowMs: now.epochMilliseconds,
1110 });
1111 const token = input.target.push_token;
1112 return yield* notification && token
1113 ? deliveryQueue.enqueuePushNotification({
1114 userId: input.target.user_id,
1115 deviceId: input.target.device_id,
1116 token,
1117 bundleId: input.target.bundle_id,
1118 apsEnvironment: input.target.aps_environment,
1119 notification,
1120 })
1121 : Effect.succeed(null);
1122 }),
1123 sendForTarget: Effect.fnUntraced(function* (input) {
1124 if (!config.apns) return null;
1125 const delivery = chooseDelivery({
1126 target: input.target,
1127 aggregate: input.aggregate,
1128 nowMs: input.nowMs,
1129 replay: input.replay ?? false,
1130 });
1131 if (!delivery) {
1132 return null;
1133 }
1134 if (delivery.kind === "push_notification") {
1135 const result = yield* deliveryQueue.enqueuePushNotification({
1136 userId: input.target.user_id,
1137 deviceId: input.target.device_id,
1138 token: delivery.token,
1139 bundleId: input.target.bundle_id,
1140 apsEnvironment: input.target.aps_environment,
1141 notification: delivery.notification,
1142 });
1143 return result;
1144 }
1145 const notification = input.replay
1146 ? null
1147 : notificationForAggregate({
1148 target: input.target,
1149 aggregate: input.aggregate,
1150 nowMs: input.nowMs,
1151 });
1152 // The end event doubles as the "task finished" moment. When a companion
1153 // push notification is about to ring the device (below), the activity end
1154 // stays silent; otherwise the end itself carries the alert so LA-only
1155 // users still get the buzz.
1156 const alert = input.replay
1157 ? null
1158 : delivery.kind === "live_activity_end"
1159 ? notification && input.target.push_token
1160 ? null
1161 : alertForTerminalAggregate({
1162 aggregate: delivery.aggregate,
1163 preferences: parsePreferences(input.target.preferences_json),
1164 })
1165 : delivery.alert;
1166 const result = yield* deliveryQueue.enqueueLiveActivity({
1167 userId: input.target.user_id,
1168 deviceId: input.target.device_id,
1169 kind: delivery.kind,
1170 token: delivery.token,
1171 bundleId: input.target.bundle_id,
1172 apsEnvironment: input.target.aps_environment,
1173 aggregate: delivery.aggregate,
1174 alert,
1175 });
1176 if (delivery.kind === "live_activity_end" && notification && input.target.push_token) {
1177 yield* deliveryQueue.enqueuePushNotification({
1178 userId: input.target.user_id,
1179 deviceId: input.target.device_id,
1180 token: input.target.push_token,
1181 bundleId: input.target.bundle_id,
1182 apsEnvironment: input.target.aps_environment,
1183 notification,
1184 });
1185 }
1186 if (delivery.kind === "live_activity_start") {
1187 const now = yield* DateTime.now;
1188 yield* liveActivities.markStartQueued({
1189 userId: input.target.user_id,
1190 deviceId: input.target.device_id,
1191 queuedAt: DateTime.formatIso(now),
1192 });
1193 }
1194 return result;
1195 }),
1196 });
1197});
1198
1199export const layer = Layer.effect(ApnsDeliveries, make);