1import type {
2 RelayAgentActivityAggregateState,
3 RelayDeliveryKind,
4 RelayLiveActivityRegistrationRequest,
5} from "@t3tools/contracts/relay";
6import {
7 RelayAgentActivityAggregateState as RelayAgentActivityAggregateStateSchema,
8 RelayDeliveryKind as RelayDeliveryKindSchema,
9} from "@t3tools/contracts/relay";
10import * as Context from "effect/Context";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import * as Function from "effect/Function";
14import * as Layer from "effect/Layer";
15import * as Schema from "effect/Schema";
16import { and, eq, sql } from "drizzle-orm";
17
18import * as RelayDb from "../db.ts";
19import { relayLiveActivities, relayMobileDevices } from "../persistence/schema.ts";
20
21export class LiveActivityRegistrationPersistenceError extends Schema.TaggedError<LiveActivityRegistrationPersistenceError>()(
22 "LiveActivityRegistrationPersistenceError",
23 {
24 userId: Schema.String,
25 deviceId: Schema.String,
26 cause: Schema.Defect(),
27 },
28) {
29 override get message(): string {
30 return `Failed to persist Live Activity registration for user ${this.userId} and device ${this.deviceId}.`;
31 }
32}
33
34export class LiveActivityTargetListPersistenceError extends Schema.TaggedError<LiveActivityTargetListPersistenceError>()(
35 "LiveActivityTargetListPersistenceError",
36 {
37 userId: Schema.String,
38 cause: Schema.Defect(),
39 },
40) {
41 override get message(): string {
42 return `Failed to list Live Activity delivery targets for user ${this.userId}.`;
43 }
44}
45
46export class LiveActivityDeliveryMarkPersistenceError extends Schema.TaggedError<LiveActivityDeliveryMarkPersistenceError>()(
47 "LiveActivityDeliveryMarkPersistenceError",
48 {
49 operation: Schema.Literals([
50 "mark-delivery",
51 "mark-start-queued",
52 "clear-start-queued",
53 "invalidate-delivery-token",
54 ]),
55 userId: Schema.String,
56 deviceId: Schema.String,
57 kind: Schema.NullOr(RelayDeliveryKindSchema),
58 cause: Schema.Defect(),
59 },
60) {
61 override get message(): string {
62 return `Failed to persist Live Activity state during ${this.operation} for user ${this.userId} and device ${this.deviceId}.`;
63 }
64}
65
66export interface DeviceRow {
67 readonly user_id: string;
68 readonly device_id: string;
69 readonly platform: "ios" | "android";
70 readonly ios_major_version: number | null;
71 readonly app_version: string | null;
72 readonly bundle_id: string | null;
73 readonly aps_environment: "sandbox" | "production" | null;
74 readonly push_token: string | null;
75 readonly push_to_start_token: string | null;
76 readonly preferences_json: string;
77}
78
79export interface LiveActivityRow {
80 readonly activity_push_token: string | null;
81 readonly remote_start_queued_at: string | null;
82 readonly remote_started_at: string | null;
83 readonly ended_at: string | null;
84 readonly last_aggregate_json: string | null;
85 readonly last_live_activity_delivery_at: string | null;
86}
87
88export type TargetRow = DeviceRow & LiveActivityRow;
89
90export class LiveActivities extends Context.Service<
91 LiveActivities,
92 {
93 readonly register: (input: {
94 readonly userId: string;
95 readonly registration: RelayLiveActivityRegistrationRequest;
96 }) => Effect.Effect<void, LiveActivityRegistrationPersistenceError>;
97 readonly listTargets: (input: {
98 readonly userId: string;
99 }) => Effect.Effect<ReadonlyArray<TargetRow>, LiveActivityTargetListPersistenceError>;
100 readonly markDelivery: (input: {
101 readonly userId: string;
102 readonly deviceId: string;
103 readonly kind: RelayDeliveryKind;
104 readonly aggregate: RelayAgentActivityAggregateState | null;
105 readonly deliveredAt: string;
106 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
107 readonly markStartQueued: (input: {
108 readonly userId: string;
109 readonly deviceId: string;
110 readonly queuedAt: string;
111 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
112 readonly clearStartQueued: (input: {
113 readonly userId: string;
114 readonly deviceId: string;
115 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
116 readonly invalidateDeliveryToken: (input: {
117 readonly userId: string;
118 readonly deviceId: string;
119 readonly kind: RelayDeliveryKind;
120 readonly invalidatedAt: string;
121 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
122 }
123>()("t3code-relay/agentActivity/LiveActivities") {}
124
125const decodeJsonString = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown));
126const encodeJsonValue = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
127
128const encodeRelayAgentActivityAggregateStateJson = Schema.encodeEffect(
129 Schema.fromJsonString(RelayAgentActivityAggregateStateSchema),
130);
131
132export const make = Effect.gen(function* () {
133 const db = yield* RelayDb.RelayDb;
134
135 return LiveActivities.of({
136 register: Effect.fn("relay.live_activities.register")(function* (input) {
137 yield* Effect.annotateCurrentSpan({
138 "relay.mobile.device_id": input.registration.deviceId,
139 });
140 yield* Effect.gen(function* () {
141 const updatedAt = DateTime.formatIso(yield* DateTime.now);
142 const registration = input.registration;
143
144 yield* db
145 .update(relayLiveActivities)
146 .set({
147 activityPushToken: null,
148 remoteStartQueuedAt: null,
149 remoteStartedAt: null,
150 endedAt: updatedAt,
151 updatedAt,
152 })
153 .where(eq(relayLiveActivities.activityPushToken, registration.activityPushToken));
154
155 yield* db
156 .insert(relayLiveActivities)
157 .values({
158 userId: input.userId,
159 deviceId: registration.deviceId,
160 activityPushToken: registration.activityPushToken,
161 remoteStartQueuedAt: null,
162 remoteStartedAt: updatedAt,
163 endedAt: null,
164 lastAggregateJson: null,
165 lastLiveActivityDeliveryAt: null,
166 createdAt: updatedAt,
167 updatedAt,
168 })
169 .onConflictDoUpdate({
170 target: [relayLiveActivities.userId, relayLiveActivities.deviceId],
171 set: {
172 activityPushToken: registration.activityPushToken,
173 remoteStartQueuedAt: null,
174 remoteStartedAt: updatedAt,
175 endedAt: null,
176 lastAggregateJson: null,
177 lastLiveActivityDeliveryAt: null,
178 updatedAt,
179 },
180 });
181 }).pipe(
182 Effect.mapError(
183 (cause) =>
184 new LiveActivityRegistrationPersistenceError({
185 userId: input.userId,
186 deviceId: input.registration.deviceId,
187 cause,
188 }),
189 ),
190 );
191 }),
192
193 listTargets: Effect.fn("relay.live_activities.list_targets")(function* (input) {
194 return yield* db
195 .select({
196 device_id: relayMobileDevices.deviceId,
197 user_id: relayMobileDevices.userId,
198 platform: relayMobileDevices.platform,
199 ios_major_version: relayMobileDevices.iosMajorVersion,
200 app_version: relayMobileDevices.appVersion,
201 bundle_id: relayMobileDevices.bundleId,
202 aps_environment: relayMobileDevices.apsEnvironment,
203 push_token: relayMobileDevices.pushToken,
204 push_to_start_token: relayMobileDevices.pushToStartToken,
205 preferences_json: relayMobileDevices.preferencesJson,
206 activity_push_token: relayLiveActivities.activityPushToken,
207 remote_start_queued_at: relayLiveActivities.remoteStartQueuedAt,
208 remote_started_at: relayLiveActivities.remoteStartedAt,
209 ended_at: relayLiveActivities.endedAt,
210 last_aggregate_json: relayLiveActivities.lastAggregateJson,
211 last_live_activity_delivery_at: relayLiveActivities.lastLiveActivityDeliveryAt,
212 })
213 .from(relayMobileDevices)
214 .leftJoin(
215 relayLiveActivities,
216 and(
217 eq(relayLiveActivities.userId, relayMobileDevices.userId),
218 eq(relayLiveActivities.deviceId, relayMobileDevices.deviceId),
219 ),
220 )
221 .where(eq(relayMobileDevices.userId, input.userId))
222 .pipe(
223 Effect.flatMap((rows) =>
224 Effect.forEach(
225 rows,
226 (row) =>
227 Effect.all({
228 preferences_json: encodeJsonValue(row.preferences_json),
229 last_aggregate_json:
230 row.last_aggregate_json === null
231 ? Effect.succeed(null)
232 : encodeJsonValue(row.last_aggregate_json),
233 }).pipe(
234 Effect.map((json) => ({
235 ...row,
236 ...json,
237 })),
238 ),
239 { concurrency: "unbounded" },
240 ),
241 ),
242 Effect.map((rows): ReadonlyArray<TargetRow> => rows),
243 Effect.mapError(
244 (cause) =>
245 new LiveActivityTargetListPersistenceError({
246 userId: input.userId,
247 cause,
248 }),
249 ),
250 );
251 }),
252
253 markDelivery: Effect.fn("relay.live_activities.mark_delivery")(function* (input) {
254 yield* Effect.annotateCurrentSpan({
255 "relay.mobile.device_id": input.deviceId,
256 "relay.delivery.kind": input.kind,
257 });
258 yield* Effect.gen(function* () {
259 const aggregateJson =
260 input.aggregate === null
261 ? null
262 : yield* encodeRelayAgentActivityAggregateStateJson(input.aggregate).pipe(
263 Effect.flatMap(decodeJsonString),
264 Effect.map(Function.cast<unknown, RelayAgentActivityAggregateState>),
265 );
266
267 yield* db
268 .insert(relayLiveActivities)
269 .values({
270 userId: input.userId,
271 deviceId: input.deviceId,
272 remoteStartedAt: input.kind === "live_activity_start" ? input.deliveredAt : null,
273 remoteStartQueuedAt: null,
274 endedAt: input.kind === "live_activity_end" ? input.deliveredAt : null,
275 lastAggregateJson: aggregateJson,
276 lastLiveActivityDeliveryAt: input.deliveredAt,
277 createdAt: input.deliveredAt,
278 updatedAt: input.deliveredAt,
279 })
280 .onConflictDoUpdate({
281 target: [relayLiveActivities.userId, relayLiveActivities.deviceId],
282 set: {
283 // A delivered start begins a NEW activity generation: the stored
284 // update token belongs to the previous activity (dead once a new
285 // one starts, and certainly dead after an end), so keep it only
286 // for plain updates. Deliveries pause until the app registers
287 // the fresh activity's token; registerLiveActivity + replay then
288 // reconcile content (or end the activity if work already
289 // finished). Without this, updates and ends route to the dead
290 // token and the new lock-screen card is stranded at its start
291 // content forever.
292 activityPushToken:
293 input.kind === "live_activity_update"
294 ? sql`${relayLiveActivities.activityPushToken}`
295 : null,
296 remoteStartedAt:
297 input.kind === "live_activity_start"
298 ? input.deliveredAt
299 : sql`coalesce(
300 ${relayLiveActivities.remoteStartedAt},
301 excluded.remote_started_at
302 )`,
303 remoteStartQueuedAt: null,
304 endedAt:
305 input.kind === "live_activity_start"
306 ? null
307 : input.kind === "live_activity_end"
308 ? input.deliveredAt
309 : relayLiveActivities.endedAt,
310 lastAggregateJson: aggregateJson,
311 lastLiveActivityDeliveryAt: input.deliveredAt,
312 updatedAt: input.deliveredAt,
313 },
314 });
315 }).pipe(
316 Effect.mapError(
317 (cause) =>
318 new LiveActivityDeliveryMarkPersistenceError({
319 operation: "mark-delivery",
320 userId: input.userId,
321 deviceId: input.deviceId,
322 kind: input.kind,
323 cause,
324 }),
325 ),
326 );
327 }),
328
329 markStartQueued: Effect.fn("relay.live_activities.mark_start_queued")(function* (input) {
330 yield* Effect.annotateCurrentSpan({
331 "relay.mobile.device_id": input.deviceId,
332 });
333 yield* db
334 .insert(relayLiveActivities)
335 .values({
336 userId: input.userId,
337 deviceId: input.deviceId,
338 remoteStartQueuedAt: input.queuedAt,
339 remoteStartedAt: null,
340 endedAt: null,
341 createdAt: input.queuedAt,
342 updatedAt: input.queuedAt,
343 })
344 .onConflictDoUpdate({
345 target: [relayLiveActivities.userId, relayLiveActivities.deviceId],
346 set: {
347 remoteStartQueuedAt: sql`coalesce(
348 ${relayLiveActivities.remoteStartQueuedAt},
349 excluded.remote_start_queued_at
350 )`,
351 endedAt: null,
352 updatedAt: input.queuedAt,
353 },
354 })
355 .pipe(
356 Effect.mapError(
357 (cause) =>
358 new LiveActivityDeliveryMarkPersistenceError({
359 operation: "mark-start-queued",
360 userId: input.userId,
361 deviceId: input.deviceId,
362 kind: null,
363 cause,
364 }),
365 ),
366 );
367 }),
368
369 clearStartQueued: Effect.fn("relay.live_activities.clear_start_queued")(function* (input) {
370 yield* Effect.annotateCurrentSpan({
371 "relay.mobile.device_id": input.deviceId,
372 });
373 yield* db
374 .update(relayLiveActivities)
375 .set({ remoteStartQueuedAt: null })
376 .where(
377 and(
378 eq(relayLiveActivities.userId, input.userId),
379 eq(relayLiveActivities.deviceId, input.deviceId),
380 ),
381 )
382 .pipe(
383 Effect.mapError(
384 (cause) =>
385 new LiveActivityDeliveryMarkPersistenceError({
386 operation: "clear-start-queued",
387 userId: input.userId,
388 deviceId: input.deviceId,
389 kind: null,
390 cause,
391 }),
392 ),
393 );
394 }),
395
396 invalidateDeliveryToken: Effect.fn("relay.live_activities.invalidate_delivery_token")(
397 function* (input) {
398 yield* Effect.annotateCurrentSpan({
399 "relay.mobile.device_id": input.deviceId,
400 "relay.delivery.kind": input.kind,
401 });
402 yield* Effect.gen(function* () {
403 if (input.kind === "push_notification") {
404 yield* db
405 .update(relayMobileDevices)
406 .set({
407 pushToken: null,
408 updatedAt: input.invalidatedAt,
409 })
410 .where(
411 and(
412 eq(relayMobileDevices.userId, input.userId),
413 eq(relayMobileDevices.deviceId, input.deviceId),
414 ),
415 );
416 return;
417 }
418
419 if (input.kind === "live_activity_start") {
420 yield* db
421 .update(relayMobileDevices)
422 .set({
423 pushToStartToken: null,
424 updatedAt: input.invalidatedAt,
425 })
426 .where(
427 and(
428 eq(relayMobileDevices.userId, input.userId),
429 eq(relayMobileDevices.deviceId, input.deviceId),
430 ),
431 );
432 yield* db
433 .update(relayLiveActivities)
434 .set({
435 remoteStartQueuedAt: null,
436 updatedAt: input.invalidatedAt,
437 })
438 .where(
439 and(
440 eq(relayLiveActivities.userId, input.userId),
441 eq(relayLiveActivities.deviceId, input.deviceId),
442 ),
443 );
444 return;
445 }
446
447 yield* db
448 .update(relayLiveActivities)
449 .set({
450 activityPushToken: null,
451 remoteStartQueuedAt: null,
452 remoteStartedAt: null,
453 endedAt: input.invalidatedAt,
454 updatedAt: input.invalidatedAt,
455 })
456 .where(
457 and(
458 eq(relayLiveActivities.userId, input.userId),
459 eq(relayLiveActivities.deviceId, input.deviceId),
460 ),
461 );
462 }).pipe(
463 Effect.mapError(
464 (cause) =>
465 new LiveActivityDeliveryMarkPersistenceError({
466 operation: "invalidate-delivery-token",
467 userId: input.userId,
468 deviceId: input.deviceId,
469 kind: input.kind,
470 cause,
471 }),
472 ),
473 );
474 },
475 ),
476 });
477});
478
479export const layer = Layer.effect(LiveActivities, make);