1import { and, eq } from "drizzle-orm";
2import {
3 RelayAgentActivityState,
4 RelayAgentActivityAggregateState,
5 RelayAgentAwarenessPreferences,
6 type RelayDeliveryResult,
7} from "@t3tools/contracts/relay";
8import * as Crypto from "effect/Crypto";
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 // Every contributing queue job identifies the same group, including after
104 // retries or a different database row order. The native handler deduplicates it.
105 alert_id: JSON.stringify(
106 activities
107 .map((row) => [row.environmentId, row.threadId, row.phase, row.updatedAt])
108 .sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b))),
109 ),
110 alert_title: alert.title,
111 alert_body: alert.body,
112 // A multi-thread alert targets the overview so one visible thread cannot
113 // suppress notifications for the other threads in the group.
114 alert_path: "/",
115 };
116}
117
118export class FcmDeliveries extends Context.Service<
119 FcmDeliveries,
120 {
121 readonly enqueue: (input: {
122 readonly target: LiveActivities.TargetRow;
123 readonly state: RelayAgentActivityState | null;
124 readonly replay?: boolean;
125 }) => Effect.Effect<RelayDeliveryResult | null, FcmDeliveryError>;
126 readonly process: (
127 body: unknown,
128 ) => Effect.Effect<
129 void,
130 | FcmDeliveryError
131 | FcmClient.FcmClientError
132 | PlatformError.PlatformError
133 | LiveActivities.LiveActivityTargetListPersistenceError
134 | LiveActivities.LiveActivityDeliveryMarkPersistenceError
135 | AgentActivityRows.AgentActivityRowListPersistenceError
136 | EnvironmentLinks.EnvironmentLinkLookupPersistenceError
137 | EnvironmentLinks.EnvironmentLinkUserListPersistenceError
138 >;
139 }
140>()("t3code-relay/agentActivity/FcmDeliveries") {}
141
142export const make = Effect.gen(function* () {
143 const config = yield* RelayConfiguration.RelayConfiguration;
144 const crypto = yield* Crypto.Crypto;
145 const sender = yield* FcmDeliveryQueueSender.FcmDeliveryQueueSender;
146 const client = yield* FcmClient.FcmClient;
147 const devices = yield* LiveActivities.LiveActivities;
148 const rows = yield* AgentActivityRows.AgentActivityRows;
149 const links = yield* EnvironmentLinks.EnvironmentLinks;
150 const db = yield* RelayDb.RelayDb;
151
152 return FcmDeliveries.of({
153 enqueue: Effect.fn("relay.fcm.enqueue")(function* (input) {
154 if (input.target.platform !== "android" || !input.target.push_token) return null;
155 if (!config.fcmServiceAccount) {
156 yield* Effect.logWarning("Android notifications are not configured for this relay");
157 return {
158 deviceId: input.target.device_id,
159 kind: "push_notification",
160 ok: false,
161 apnsStatus: null,
162 apnsReason: null,
163 apnsId: null,
164 };
165 }
166 const now = yield* DateTime.now;
167 yield* sender
168 .send({
169 userId: input.target.user_id,
170 deviceId: input.target.device_id,
171 token: input.target.push_token,
172 state: input.state,
173 ...(input.replay ? { replay: true } : {}),
174 queuedAt: now.epochMilliseconds,
175 })
176 .pipe(Effect.mapError((cause) => new FcmDeliveryError({ operation: "enqueue", cause })));
177 return {
178 deviceId: input.target.device_id,
179 kind: "push_notification",
180 ok: true,
181 queued: true,
182 apnsStatus: null,
183 apnsReason: null,
184 apnsId: null,
185 };
186 }),
187 process: Effect.fn("relay.fcm.process")(function* (body) {
188 const job = yield* decodeJob(body).pipe(
189 Effect.mapError((cause) => new FcmDeliveryError({ operation: "decode-job", cause })),
190 );
191 const now = yield* DateTime.now;
192 if (now.epochMilliseconds - job.queuedAt > 5 * 60_000) return;
193 const targets = yield* devices.listTargets({ userId: job.userId });
194 const target = targets.find(
195 (device) =>
196 device.device_id === job.deviceId &&
197 device.platform === "android" &&
198 device.push_token === job.token,
199 );
200 if (!target) return;
201 const preferences = decodePreferences(target.preferences_json);
202 if (Option.isNone(preferences)) return;
203
204 // Re-read links and state when consuming: queued messages must honor
205 // sign-out, token rotation, disabled publishing, and newer thread states.
206 const states = preferences.value.liveActivitiesEnabled
207 ? yield* rows.listForUser({ userId: job.userId })
208 : [];
209 const aggregate = makeAggregateState({
210 activeStates: states,
211 terminalState: null,
212 nowMs: now.epochMilliseconds,
213 });
214 const previousAggregate = target.last_aggregate_json
215 ? Option.getOrNull(decodePreviousActivity(target.last_aggregate_json))
216 : null;
217 let alert: ReturnType<typeof androidAlertForState | typeof androidAlertForAggregate> = null;
218 // Deletion jobs can observe another thread's newly completed state. They
219 // update the card, but must leave that transition for its own alert job.
220 // Registration replay deliberately establishes a silent baseline.
221 let acknowledgeAggregate = job.state !== null || job.replay === true || aggregate === null;
222 if (job.state && preferences.value.notificationsEnabled) {
223 const state = yield* rows.getForUserThread({
224 userId: job.userId,
225 environmentId: job.state.environmentId,
226 threadId: job.state.threadId,
227 });
228 if (
229 !previousAggregate &&
230 (!state || state.phase !== job.state.phase || state.updatedAt !== job.state.updatedAt)
231 )
232 return;
233 const link = yield* links.getForUser({
234 userId: job.userId,
235 environmentId: job.state.environmentId,
236 });
237 const deliveryUsers = link
238 ? yield* links.listDeliveryUsersForEnvironment({
239 environmentId: job.state.environmentId,
240 environmentPublicKey: link.environmentPublicKey,
241 })
242 : [];
243 const deliveryUser = deliveryUsers.find((user) => user.userId === job.userId);
244 // A notification-only job must not acknowledge transitions on another
245 // environment's live card before that environment's own job can alert.
246 acknowledgeAggregate =
247 deliveryUser?.liveActivitiesEnabled === true || !preferences.value.liveActivitiesEnabled;
248 if (
249 deliveryUser?.liveActivitiesEnabled &&
250 preferences.value.liveActivitiesEnabled &&
251 previousAggregate &&
252 aggregate
253 ) {
254 const environmentIds = [...new Set(aggregate.activities.map((row) => row.environmentId))];
255 const allowedEnvironments = new Set<string>();
256 for (const environmentId of environmentIds) {
257 const environmentLink = yield* links.getForUser({
258 userId: job.userId,
259 environmentId,
260 });
261 if (!environmentLink) continue;
262 const users = yield* links.listDeliveryUsersForEnvironment({
263 environmentId,
264 environmentPublicKey: environmentLink.environmentPublicKey,
265 });
266 if (users.some((user) => user.userId === job.userId && user.notificationsEnabled)) {
267 allowedEnvironments.add(environmentId);
268 }
269 }
270 alert = androidAlertForAggregate({
271 previousAggregate,
272 nextAggregate: {
273 ...aggregate,
274 activities: aggregate.activities.filter((row) =>
275 allowedEnvironments.has(row.environmentId),
276 ),
277 },
278 preferences: preferences.value,
279 nowMs: now.epochMilliseconds,
280 });
281 } else if (
282 deliveryUser?.notificationsEnabled &&
283 state?.phase === job.state.phase &&
284 state.updatedAt === job.state.updatedAt &&
285 !isExpiredAgentActivityState(state, now.epochMilliseconds)
286 ) {
287 alert = androidAlertForState(state, preferences.value, now.epochMilliseconds);
288 }
289 }
290 const displayedAggregate =
291 preferences.value.notificationsEnabled && preferences.value.liveActivitiesEnabled
292 ? aggregate
293 : null;
294 const active = (displayedAggregate?.activeCount ?? 0) > 0;
295 // A registration replay must clear an orphan even when the relay has
296 // already forgotten its baseline. Finished cards are visible, but idle.
297 if (!displayedAggregate && !alert && !previousAggregate && job.state !== null) return;
298 const data: Record<string, string> = {
299 t3_kind: "agent_activity",
300 device_id: job.deviceId,
301 user_id: job.userId,
302 updated_at: String(now.epochMilliseconds),
303 ...androidActivityData(displayedAggregate),
304 ...alert,
305 };
306 if (alert) {
307 // Group identities can contain five sets of IDs. Hash the full,
308 // stable identity rather than spending the payload budget on it.
309 const digest = yield* crypto.digest("SHA-256", new TextEncoder().encode(alert.alert_id));
310 data.alert_id = Array.from(digest, (byte) => byte.toString(16).padStart(2, "0")).join("");
311 }
312 const result = yield* client.send({
313 token: job.token,
314 packageName: target.bundle_id,
315 alert: alert !== null,
316 data: fitFcmData(data),
317 });
318 if (result.unregistered) {
319 yield* db
320 .update(relayMobileDevices)
321 .set({ pushToken: null })
322 .where(
323 and(
324 eq(relayMobileDevices.userId, job.userId),
325 eq(relayMobileDevices.deviceId, job.deviceId),
326 eq(relayMobileDevices.pushToken, job.token),
327 ),
328 )
329 .pipe(
330 Effect.mapError(
331 (cause) => new FcmDeliveryError({ operation: "invalidate-token", cause }),
332 ),
333 );
334 } else if (acknowledgeAggregate) {
335 yield* devices.markDelivery({
336 userId: job.userId,
337 deviceId: job.deviceId,
338 kind: active ? "live_activity_update" : "live_activity_end",
339 // Keep the delivered terminal rows as the next transition baseline,
340 // so replaying the finished card cannot alert again.
341 aggregate: preferences.value.liveActivitiesEnabled ? aggregate : null,
342 deliveredAt: DateTime.formatIso(now),
343 });
344 }
345 }),
346 });
347});
348
349export const layer = Layer.effect(FcmDeliveries, make);