Open History to see the last commit
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, isNotNull, 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 LiveActivityIdleTargetListPersistenceError extends Schema.TaggedError<LiveActivityIdleTargetListPersistenceError>()(
47 "LiveActivityIdleTargetListPersistenceError",
48 {
49 deliveredBefore: Schema.String,
50 cause: Schema.Defect(),
51 },
52) {
53 override get message(): string {
54 return `Failed to list idle Live Activities delivered before ${this.deliveredBefore}.`;
55 }
56}
57
58export class LiveActivityDeliveryMarkPersistenceError extends Schema.TaggedError<LiveActivityDeliveryMarkPersistenceError>()(
59 "LiveActivityDeliveryMarkPersistenceError",
60 {
61 operation: Schema.Literals(["mark-delivery", "invalidate-delivery-token"]),
62 userId: Schema.String,
63 deviceId: Schema.String,
64 kind: Schema.NullOr(RelayDeliveryKindSchema),
65 cause: Schema.Defect(),
66 },
67) {
68 override get message(): string {
69 return `Failed to persist Live Activity state during ${this.operation} for user ${this.userId} and device ${this.deviceId}.`;
70 }
71}
72
73export interface DeviceRow {
74 readonly user_id: string;
75 readonly device_id: string;
76 readonly platform: "ios" | "android";
77 readonly ios_major_version: number | null;
78 readonly app_version: string | null;
79 readonly bundle_id: string | null;
80 readonly aps_environment: "sandbox" | "production" | null;
81 readonly push_token: string | null;
82 readonly push_to_start_token: string | null;
83 readonly preferences_json: string;
84}
85
86export interface LiveActivityRow {
87 readonly activity_push_token: string | null;
88 readonly remote_start_queued_at: string | null;
89 readonly remote_started_at: string | null;
90 readonly ended_at: string | null;
91 readonly last_aggregate_json: string | null;
92 readonly last_live_activity_delivery_at: string | null;
93}
94
95export type TargetRow = DeviceRow & LiveActivityRow;
96
97export class LiveActivities extends Context.Service<
98 LiveActivities,
99 {
100 readonly register: (input: {
101 readonly userId: string;
102 readonly registration: RelayLiveActivityRegistrationRequest;
103 }) => Effect.Effect<void, LiveActivityRegistrationPersistenceError>;
104 readonly listTargets: (input: {
105 readonly userId: string;
106 }) => Effect.Effect<ReadonlyArray<TargetRow>, LiveActivityTargetListPersistenceError>;
107 readonly listIdleArmedTargets: (input: {
108 readonly deliveredBefore: string;
109 }) => Effect.Effect<
110 ReadonlyArray<{ readonly user_id: string; readonly device_id: string }>,
111 LiveActivityIdleTargetListPersistenceError
112 >;
113 readonly markDelivery: (input: {
114 readonly userId: string;
115 readonly deviceId: string;
116 readonly kind: Exclude<RelayDeliveryKind, "live_activity_start">;
117 readonly aggregate: RelayAgentActivityAggregateState | null;
118 readonly deliveredAt: string;
119 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
120 readonly invalidateDeliveryToken: (input: {
121 readonly userId: string;
122 readonly deviceId: string;
123 readonly kind: Exclude<RelayDeliveryKind, "live_activity_start">;
124 readonly invalidatedAt: string;
125 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
126 }
127>()("t3code-relay/agentActivity/LiveActivities") {}
128
129const decodeJsonString = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown));
130const encodeJsonValue = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
131
132const encodeRelayAgentActivityAggregateStateJson = Schema.encodeEffect(
133 Schema.fromJsonString(RelayAgentActivityAggregateStateSchema),
134);
135
136const make = Effect.gen(function* () {
137 const db = yield* RelayDb.RelayDb;
138
139 return LiveActivities.of({
140 register: Effect.fn("relay.live_activities.register")(function* (input) {
141 yield* Effect.annotateCurrentSpan({
142 "relay.mobile.device_id": input.registration.deviceId,
143 });
144 yield* Effect.gen(function* () {
145 const updatedAt = DateTime.formatIso(yield* DateTime.now);
146 const registration = input.registration;
147 const [current] = yield* db
148 .select({
149 activityPushToken: relayLiveActivities.activityPushToken,
150 remoteStartedAt: relayLiveActivities.remoteStartedAt,
151 })
152 .from(relayLiveActivities)
153 .where(
154 and(
155 eq(relayLiveActivities.userId, input.userId),
156 eq(relayLiveActivities.deviceId, registration.deviceId),
157 ),
158 );
159 // Foreground token refresh must retain the original arming time or grace can swallow an owed end event.
160 const armedAt =
161 (current?.activityPushToken === registration.activityPushToken
162 ? current.remoteStartedAt
163 : null) ?? updatedAt;
164
165 yield* db
166 .update(relayLiveActivities)
167 .set({
168 activityPushToken: null,
169 remoteStartQueuedAt: null,
170 remoteStartedAt: null,
171 endedAt: updatedAt,
172 updatedAt,
173 })
174 .where(eq(relayLiveActivities.activityPushToken, registration.activityPushToken));
175
176 yield* db
177 .insert(relayLiveActivities)
178 .values({
179 userId: input.userId,
180 deviceId: registration.deviceId,
181 activityPushToken: registration.activityPushToken,
182 remoteStartQueuedAt: null,
183 remoteStartedAt: armedAt,
184 endedAt: null,
185 lastAggregateJson: null,
186 lastLiveActivityDeliveryAt: null,
187 createdAt: updatedAt,
188 updatedAt,
189 })
190 .onConflictDoUpdate({
191 target: [relayLiveActivities.userId, relayLiveActivities.deviceId],
192 set: {
193 activityPushToken: registration.activityPushToken,
194 remoteStartQueuedAt: null,
195 remoteStartedAt: armedAt,
196 endedAt: null,
197 lastAggregateJson: null,
198 lastLiveActivityDeliveryAt: null,
199 updatedAt,
200 },
201 });
202 }).pipe(
203 Effect.mapError(
204 (cause) =>
205 new LiveActivityRegistrationPersistenceError({
206 userId: input.userId,
207 deviceId: input.registration.deviceId,
208 cause,
209 }),
210 ),
211 );
212 }),
213
214 listTargets: Effect.fn("relay.live_activities.list_targets")(function* (input) {
215 return yield* db
216 .select({
217 device_id: relayMobileDevices.deviceId,
218 user_id: relayMobileDevices.userId,
219 platform: relayMobileDevices.platform,
220 ios_major_version: relayMobileDevices.iosMajorVersion,
221 app_version: relayMobileDevices.appVersion,
222 bundle_id: relayMobileDevices.bundleId,
223 aps_environment: relayMobileDevices.apsEnvironment,
224 push_token: relayMobileDevices.pushToken,
225 push_to_start_token: relayMobileDevices.pushToStartToken,
226 preferences_json: relayMobileDevices.preferencesJson,
227 activity_push_token: relayLiveActivities.activityPushToken,
228 remote_start_queued_at: relayLiveActivities.remoteStartQueuedAt,
229 remote_started_at: relayLiveActivities.remoteStartedAt,
230 ended_at: relayLiveActivities.endedAt,
231 last_aggregate_json: relayLiveActivities.lastAggregateJson,
232 last_live_activity_delivery_at: relayLiveActivities.lastLiveActivityDeliveryAt,
233 })
234 .from(relayMobileDevices)
235 .leftJoin(
236 relayLiveActivities,
237 and(
238 eq(relayLiveActivities.userId, relayMobileDevices.userId),
239 eq(relayLiveActivities.deviceId, relayMobileDevices.deviceId),
240 ),
241 )
242 .where(eq(relayMobileDevices.userId, input.userId))
243 .pipe(
244 Effect.flatMap((rows) =>
245 Effect.forEach(
246 rows,
247 (row) =>
248 Effect.all({
249 preferences_json: encodeJsonValue(row.preferences_json),
250 last_aggregate_json:
251 row.last_aggregate_json === null
252 ? Effect.succeed(null)
253 : encodeJsonValue(row.last_aggregate_json),
254 }).pipe(
255 Effect.map((json) => ({
256 ...row,
257 ...json,
258 })),
259 ),
260 { concurrency: "unbounded" },
261 ),
262 ),
263 Effect.map((rows): ReadonlyArray<TargetRow> => rows),
264 Effect.mapError(
265 (cause) =>
266 new LiveActivityTargetListPersistenceError({
267 userId: input.userId,
268 cause,
269 }),
270 ),
271 );
272 }),
273
274 listIdleArmedTargets: Effect.fn("relay.live_activities.list_idle_armed_targets")(
275 function* (input) {
276 return yield* db
277 .select({
278 user_id: relayLiveActivities.userId,
279 device_id: relayLiveActivities.deviceId,
280 })
281 .from(relayLiveActivities)
282 .where(
283 and(
284 isNotNull(relayLiveActivities.activityPushToken),
285 sql`coalesce(${relayLiveActivities.lastLiveActivityDeliveryAt}, ${relayLiveActivities.remoteStartedAt}, ${relayLiveActivities.updatedAt}) < ${input.deliveredBefore}`,
286 ),
287 )
288 .pipe(
289 Effect.mapError(
290 (cause) =>
291 new LiveActivityIdleTargetListPersistenceError({
292 deliveredBefore: input.deliveredBefore,
293 cause,
294 }),
295 ),
296 );
297 },
298 ),
299
300 markDelivery: Effect.fn("relay.live_activities.mark_delivery")(function* (input) {
301 yield* Effect.annotateCurrentSpan({
302 "relay.mobile.device_id": input.deviceId,
303 "relay.delivery.kind": input.kind,
304 });
305 yield* Effect.gen(function* () {
306 const aggregateJson =
307 input.aggregate === null
308 ? null
309 : yield* encodeRelayAgentActivityAggregateStateJson(input.aggregate).pipe(
310 Effect.flatMap(decodeJsonString),
311 Effect.map(Function.cast<unknown, RelayAgentActivityAggregateState>),
312 );
313
314 yield* db
315 .insert(relayLiveActivities)
316 .values({
317 userId: input.userId,
318 deviceId: input.deviceId,
319 remoteStartedAt: null,
320 remoteStartQueuedAt: null,
321 endedAt: input.kind === "live_activity_end" ? input.deliveredAt : null,
322 lastAggregateJson: aggregateJson,
323 lastLiveActivityDeliveryAt: input.deliveredAt,
324 createdAt: input.deliveredAt,
325 updatedAt: input.deliveredAt,
326 })
327 .onConflictDoUpdate({
328 target: [relayLiveActivities.userId, relayLiveActivities.deviceId],
329 set: {
330 activityPushToken:
331 input.kind === "live_activity_update"
332 ? sql`${relayLiveActivities.activityPushToken}`
333 : null,
334 remoteStartedAt: sql`coalesce(
335 ${relayLiveActivities.remoteStartedAt},
336 excluded.remote_started_at
337 )`,
338 remoteStartQueuedAt: null,
339 endedAt:
340 input.kind === "live_activity_end"
341 ? input.deliveredAt
342 : relayLiveActivities.endedAt,
343 lastAggregateJson: aggregateJson,
344 lastLiveActivityDeliveryAt: input.deliveredAt,
345 updatedAt: input.deliveredAt,
346 },
347 });
348 }).pipe(
349 Effect.mapError(
350 (cause) =>
351 new LiveActivityDeliveryMarkPersistenceError({
352 operation: "mark-delivery",
353 userId: input.userId,
354 deviceId: input.deviceId,
355 kind: input.kind,
356 cause,
357 }),
358 ),
359 );
360 }),
361
362 invalidateDeliveryToken: Effect.fn("relay.live_activities.invalidate_delivery_token")(
363 function* (input) {
364 yield* Effect.annotateCurrentSpan({
365 "relay.mobile.device_id": input.deviceId,
366 "relay.delivery.kind": input.kind,
367 });
368 yield* Effect.gen(function* () {
369 if (input.kind === "push_notification") {
370 yield* db
371 .update(relayMobileDevices)
372 .set({
373 pushToken: null,
374 updatedAt: input.invalidatedAt,
375 })
376 .where(
377 and(
378 eq(relayMobileDevices.userId, input.userId),
379 eq(relayMobileDevices.deviceId, input.deviceId),
380 ),
381 );
382 return;
383 }
384
385 yield* db
386 .update(relayLiveActivities)
387 .set({
388 activityPushToken: null,
389 remoteStartQueuedAt: null,
390 remoteStartedAt: null,
391 endedAt: input.invalidatedAt,
392 updatedAt: input.invalidatedAt,
393 })
394 .where(
395 and(
396 eq(relayLiveActivities.userId, input.userId),
397 eq(relayLiveActivities.deviceId, input.deviceId),
398 ),
399 );
400 }).pipe(
401 Effect.mapError(
402 (cause) =>
403 new LiveActivityDeliveryMarkPersistenceError({
404 operation: "invalidate-delivery-token",
405 userId: input.userId,
406 deviceId: input.deviceId,
407 kind: input.kind,
408 cause,
409 }),
410 ),
411 );
412 },
413 ),
414 });
415});
416
417export const layer = Layer.effect(LiveActivities, make);