Open History to see the last commit
1import { and, eq } from "drizzle-orm";
2import {
3 RelayAgentActivityState,
4 RelayAgentActivityAggregateState,
5 RelayAgentAwarenessPreferences,
6 type RelayDeliveryResult,
7} from "@t3tools/contracts/relay";
8import * as Digest from "effect/crypto/Digest";
9import type * as PlatformError from "effect/PlatformError";
10import * as Context from "effect/Context";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import * as Layer from "effect/Layer";
14import * as Option from "effect/Option";
15import * as Schema from "effect/Schema";
16
17import * as RelayConfiguration from "../Config.ts";
18import * as RelayDb from "../db.ts";
19import { relayMobileDevices } from "../persistence/schema.ts";
20import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
21import * as AgentActivityRows from "./AgentActivityRows.ts";
22import * as LiveActivities from "./LiveActivities.ts";
23import * as FcmDeliveryQueueSender from "./FcmDeliveryQueueSender.ts";
24import * as FcmClient from "./FcmClient.ts";
25import { androidActivityData, fitFcmData } from "./fcmPayloads.ts";
26import { makeAggregateState, statusForPhase } from "./agentActivityAggregate.ts";
27import { isExpiredAgentActivityState, notificationForActivity } from "./agentActivityPayloads.ts";
28import {
29 alertForActivityRows,
30 attentionTransitionRows,
31 terminalTransitionRows,
32 shouldAlertForActivity,
33} from "./agentActivityAlerts.ts";
34
35export const FcmDeliveryJob = Schema.Struct({
36 userId: Schema.String,
37 deviceId: Schema.String,
38 token: Schema.String,
39 state: Schema.NullOr(RelayAgentActivityState),
40 queuedAt: Schema.Number,
41 replay: Schema.optional(Schema.Boolean),
42});
43export type FcmDeliveryJob = typeof FcmDeliveryJob.Type;
44const decodeJob = Schema.decodeUnknownEffect(FcmDeliveryJob);
45const decodePreferences = Schema.decodeUnknownOption(
46 Schema.fromJsonString(RelayAgentAwarenessPreferences),
47);
48const decodePreviousActivity = Schema.decodeUnknownOption(
49 Schema.fromJsonString(RelayAgentActivityAggregateState),
50);
51
52export class FcmDeliveryError extends Schema.TaggedError<FcmDeliveryError>()("FcmDeliveryError", {
53 operation: Schema.Literals(["enqueue", "decode-job", "invalidate-token"]),
54 cause: Schema.Defect(),
55}) {
56 override get message() {
57 return `Failed to ${this.operation} Android notification delivery.`;
58 }
59}
60
61export function androidAlertForState(
62 state: RelayAgentActivityState,
63 preferences: RelayAgentAwarenessPreferences,
64 nowMs: number,
65) {
66 if (!shouldAlertForActivity({ ...state, preferences, nowMs })) return null;
67 const notification = notificationForActivity({ ...state, status: statusForPhase(state.phase) });
68 return {
69 alert_id: JSON.stringify([state.environmentId, state.threadId, state.phase, state.updatedAt]),
70 alert_group: JSON.stringify([state.environmentId, state.threadId]),
71 alert_title: notification.title,
72 alert_body: notification.body,
73 alert_path: notification.deepLink,
74 };
75}
76
77export function androidAlertForAggregate(input: {
78 readonly previousAggregate: RelayAgentActivityAggregateState;
79 readonly nextAggregate: RelayAgentActivityAggregateState;
80 readonly preferences: RelayAgentAwarenessPreferences;
81 readonly nowMs: number;
82}) {
83 if (!input.preferences.notificationsEnabled) return null;
84 const attention = attentionTransitionRows(input);
85 const activities =
86 attention.length > 0
87 ? attention
88 : terminalTransitionRows({ ...input, includeUnobserved: true });
89 const first = activities[0];
90 const alert = alertForActivityRows(activities);
91 if (!first || !alert) return null;
92 if (activities.length === 1) {
93 const notification = notificationForActivity(first);
94 return {
95 alert_id: JSON.stringify([first.environmentId, first.threadId, first.phase, first.updatedAt]),
96 alert_group: JSON.stringify([first.environmentId, first.threadId]),
97 alert_title: notification.title,
98 alert_body: notification.body,
99 alert_path: notification.deepLink,
100 };
101 }
102 return {
103 // Stable group IDs let the native handler deduplicate across jobs, retries, and row orders.
104 alert_id: JSON.stringify(
105 activities
106 .map((row) => [row.environmentId, row.threadId, row.phase, row.updatedAt])
107 .sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b))),
108 ),
109 alert_title: alert.title,
110 alert_body: alert.body,
111 // Target the overview so one visible thread cannot suppress other threads' alerts.
112 alert_path: "/",
113 };
114}
115
116export class FcmDeliveries extends Context.Service<
117 FcmDeliveries,
118 {
119 readonly enqueue: (input: {
120 readonly target: LiveActivities.TargetRow;
121 readonly state: RelayAgentActivityState | null;
122 readonly replay?: boolean;
123 }) => Effect.Effect<RelayDeliveryResult | null, FcmDeliveryError>;
124 readonly process: (
125 body: unknown,
126 ) => Effect.Effect<
127 void,
128 | FcmDeliveryError
129 | FcmClient.FcmClientError
130 | PlatformError.PlatformError
131 | LiveActivities.LiveActivityTargetListPersistenceError
132 | LiveActivities.LiveActivityDeliveryMarkPersistenceError
133 | AgentActivityRows.AgentActivityRowListPersistenceError
134 | EnvironmentLinks.EnvironmentLinkLookupPersistenceError
135 | EnvironmentLinks.EnvironmentLinkUserListPersistenceError
136 >;
137 }
138>()("t3code-relay/agentActivity/FcmDeliveries") {}
139
140const make = Effect.gen(function* () {
141 const config = yield* RelayConfiguration.RelayConfiguration;
142 const crypto = yield* Digest.Digest;
143 const sender = yield* FcmDeliveryQueueSender.FcmDeliveryQueueSender;
144 const client = yield* FcmClient.FcmClient;
145 const devices = yield* LiveActivities.LiveActivities;
146 const rows = yield* AgentActivityRows.AgentActivityRows;
147 const links = yield* EnvironmentLinks.EnvironmentLinks;
148 const db = yield* RelayDb.RelayDb;
149
150 return FcmDeliveries.of({
151 enqueue: Effect.fn("relay.fcm.enqueue")(function* (input) {
152 if (input.target.platform !== "android" || !input.target.push_token) return null;
153 if (!config.fcmServiceAccount) {
154 yield* Effect.logWarning("Android notifications are not configured for this relay");
155 return {
156 deviceId: input.target.device_id,
157 kind: "push_notification",
158 ok: false,
159 apnsStatus: null,
160 apnsReason: null,
161 apnsId: null,
162 };
163 }
164 const now = yield* DateTime.now;
165 yield* sender
166 .send({
167 userId: input.target.user_id,
168 deviceId: input.target.device_id,
169 token: input.target.push_token,
170 state: input.state,
171 ...(input.replay ? { replay: true } : {}),
172 queuedAt: now.epochMilliseconds,
173 })
174 .pipe(Effect.mapError((cause) => new FcmDeliveryError({ operation: "enqueue", cause })));
175 return {
176 deviceId: input.target.device_id,
177 kind: "push_notification",
178 ok: true,
179 queued: true,
180 apnsStatus: null,
181 apnsReason: null,
182 apnsId: null,
183 };
184 }),
185 process: Effect.fn("relay.fcm.process")(function* (body) {
186 const job = yield* decodeJob(body).pipe(
187 Effect.mapError((cause) => new FcmDeliveryError({ operation: "decode-job", cause })),
188 );
189 const now = yield* DateTime.now;
190 if (now.epochMilliseconds - job.queuedAt > 5 * 60_000) return;
191 const targets = yield* devices.listTargets({ userId: job.userId });
192 const target = targets.find(
193 (device) =>
194 device.device_id === job.deviceId &&
195 device.platform === "android" &&
196 device.push_token === job.token,
197 );
198 if (!target) return;
199 const preferences = decodePreferences(target.preferences_json);
200 if (Option.isNone(preferences)) return;
201
202 // Re-read at consumption to honor current links, tokens, preferences, and thread states.
203 const states = preferences.value.liveActivitiesEnabled
204 ? yield* rows.listForUser({ userId: job.userId })
205 : [];
206 const aggregate = makeAggregateState({
207 activeStates: states,
208 terminalState: null,
209 nowMs: now.epochMilliseconds,
210 });
211 const previousAggregate = target.last_aggregate_json
212 ? Option.getOrNull(decodePreviousActivity(target.last_aggregate_json))
213 : null;
214 let alert: ReturnType<typeof androidAlertForState | typeof androidAlertForAggregate> = null;
215 // Deletions update the card but must not consume another thread's pending completion alert.
216 let acknowledgeAggregate = job.state !== null || job.replay === true || aggregate === null;
217 if (job.state && preferences.value.notificationsEnabled) {
218 const state = yield* rows.getForUserThread({
219 userId: job.userId,
220 environmentId: job.state.environmentId,
221 threadId: job.state.threadId,
222 });
223 if (
224 !previousAggregate &&
225 (!state || state.phase !== job.state.phase || state.updatedAt !== job.state.updatedAt)
226 )
227 return;
228 const link = yield* links.getForUser({
229 userId: job.userId,
230 environmentId: job.state.environmentId,
231 });
232 const deliveryUsers = link
233 ? yield* links.listDeliveryUsersForEnvironment({
234 environmentId: job.state.environmentId,
235 environmentPublicKey: link.environmentPublicKey,
236 })
237 : [];
238 const deliveryUser = deliveryUsers.find((user) => user.userId === job.userId);
239 // Notification-only jobs must not acknowledge another environment's pending card transitions.
240 acknowledgeAggregate =
241 deliveryUser?.liveActivitiesEnabled === true || !preferences.value.liveActivitiesEnabled;
242 if (
243 deliveryUser?.liveActivitiesEnabled &&
244 preferences.value.liveActivitiesEnabled &&
245 previousAggregate &&
246 aggregate
247 ) {
248 const environmentIds = [...new Set(aggregate.activities.map((row) => row.environmentId))];
249 const allowedEnvironments = new Set<string>();
250 for (const environmentId of environmentIds) {
251 const environmentLink = yield* links.getForUser({
252 userId: job.userId,
253 environmentId,
254 });
255 if (!environmentLink) continue;
256 const users = yield* links.listDeliveryUsersForEnvironment({
257 environmentId,
258 environmentPublicKey: environmentLink.environmentPublicKey,
259 });
260 if (users.some((user) => user.userId === job.userId && user.notificationsEnabled)) {
261 allowedEnvironments.add(environmentId);
262 }
263 }
264 alert = androidAlertForAggregate({
265 previousAggregate,
266 nextAggregate: {
267 ...aggregate,
268 activities: aggregate.activities.filter((row) =>
269 allowedEnvironments.has(row.environmentId),
270 ),
271 },
272 preferences: preferences.value,
273 nowMs: now.epochMilliseconds,
274 });
275 } else if (
276 deliveryUser?.notificationsEnabled &&
277 state?.phase === job.state.phase &&
278 state.updatedAt === job.state.updatedAt &&
279 !isExpiredAgentActivityState(state, now.epochMilliseconds)
280 ) {
281 alert = androidAlertForState(state, preferences.value, now.epochMilliseconds);
282 }
283 }
284 const displayedAggregate = preferences.value.liveActivitiesEnabled ? aggregate : null;
285 const active = (displayedAggregate?.activeCount ?? 0) > 0;
286 if (!displayedAggregate && !alert && !previousAggregate && job.state !== null) return;
287 const data: Record<string, string> = {
288 t3_kind: "agent_activity",
289 device_id: job.deviceId,
290 user_id: job.userId,
291 updated_at: String(now.epochMilliseconds),
292 ...androidActivityData(displayedAggregate),
293 ...alert,
294 };
295 if (alert) {
296 // Hash the full stable group identity to stay within the payload budget.
297 const digest = yield* crypto.digest(
298 Digest.sha256,
299 new TextEncoder().encode(alert.alert_id),
300 );
301 data.alert_id = Array.from(digest, (byte) => byte.toString(16).padStart(2, "0")).join("");
302 }
303 const result = yield* client.send({
304 token: job.token,
305 packageName: target.bundle_id,
306 alert: alert !== null,
307 data: fitFcmData(data),
308 });
309 if (result.unregistered) {
310 yield* db
311 .update(relayMobileDevices)
312 .set({ pushToken: null })
313 .where(
314 and(
315 eq(relayMobileDevices.userId, job.userId),
316 eq(relayMobileDevices.deviceId, job.deviceId),
317 eq(relayMobileDevices.pushToken, job.token),
318 ),
319 )
320 .pipe(
321 Effect.mapError(
322 (cause) => new FcmDeliveryError({ operation: "invalidate-token", cause }),
323 ),
324 );
325 } else if (acknowledgeAggregate) {
326 yield* devices.markDelivery({
327 userId: job.userId,
328 deviceId: job.deviceId,
329 kind: active ? "live_activity_update" : "live_activity_end",
330 // Retain terminal rows as the baseline so finished-card replays cannot alert again.
331 aggregate: preferences.value.liveActivitiesEnabled ? aggregate : null,
332 deliveredAt: DateTime.formatIso(now),
333 });
334 }
335 }),
336 });
337});
338
339export const layer = Layer.effect(FcmDeliveries, make);