authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-20 12:05:13-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-20 12:05:13-07:00
loge06c33c38dba362e6a9a0733542cb37d56c024a3
tree59656ef1574a0c1016d92ef08c4066824ff6eca1
parent5f860439e9c981c29fc4385ad505ffebfbb63343
signature Signed by PGP key B5690EEEBB952194

[codex] Structure relay Live Activity errors (#3314)

Co-authored-by: codex <codex@users.noreply.github.com>

2 files changed, 241 insertions(+), 74 deletions(-)

infra/relay/src/agentActivity/LiveActivities.test.ts+88
...@@ -197,4 +197,92 @@ describe("LiveActivities", () => {...@@ -197,4 +197,92 @@ describe("LiveActivities", () => {
197 ),197 ),
198 );198 );
199 });199 });
200
201 it.effect("preserves correlation context and causes for persistence failures", () => {
202 const cause = new Error("database unavailable");
203 const registration: RelayLiveActivityRegistrationRequest = {
204 deviceId: "device-1" as RelayLiveActivityRegistrationRequest["deviceId"],
205 activityPushToken:
206 "activity-push-token" as RelayLiveActivityRegistrationRequest["activityPushToken"],
207 };
208 const fakeDb = {
209 update: () => ({
210 set: () => ({ where: () => Effect.fail(cause) }),
211 }),
212 insert: () => ({
213 values: () => ({ onConflictDoUpdate: () => Effect.fail(cause) }),
214 }),
215 select: () => ({
216 from: () => ({
217 leftJoin: () => ({ where: () => Effect.fail(cause) }),
218 }),
219 }),
220 } as unknown as RelayDb.RelayDb["Service"];
221
222 return Effect.gen(function* () {
223 const liveActivities = yield* LiveActivities.LiveActivities;
224 const registrationError = yield* Effect.flip(
225 liveActivities.register({ userId: "user-1", registration }),
226 );
227 const targetListError = yield* Effect.flip(liveActivities.listTargets({ userId: "user-1" }));
228 const deliveryErrors = yield* Effect.all(
229 [
230 liveActivities.markDelivery({
231 userId: "user-1",
232 deviceId: "device-1",
233 kind: "live_activity_update",
234 aggregate: null,
235 deliveredAt: "2026-05-25T00:00:10.000Z",
236 }),
237 liveActivities.markStartQueued({
238 userId: "user-1",
239 deviceId: "device-1",
240 queuedAt: "2026-05-25T00:00:10.000Z",
241 }),
242 liveActivities.clearStartQueued({ userId: "user-1", deviceId: "device-1" }),
243 liveActivities.invalidateDeliveryToken({
244 userId: "user-1",
245 deviceId: "device-1",
246 kind: "push_notification",
247 invalidatedAt: "2026-05-25T00:00:10.000Z",
248 }),
249 ].map(Effect.flip),
250 { concurrency: 1 },
251 );
252
253 expect(registrationError).toMatchObject({
254 userId: "user-1",
255 deviceId: "device-1",
256 cause,
257 message:
258 "Failed to persist Live Activity registration for user user-1 and device device-1.",
259 });
260 expect(targetListError).toMatchObject({
261 userId: "user-1",
262 cause,
263 message: "Failed to list Live Activity delivery targets for user user-1.",
264 });
265
266 const expectedDeliveryContext = [
267 ["mark-delivery", "live_activity_update"],
268 ["mark-start-queued", null],
269 ["clear-start-queued", null],
270 ["invalidate-delivery-token", "push_notification"],
271 ] as const;
272 for (const [index, [operation, kind]] of expectedDeliveryContext.entries()) {
273 expect(deliveryErrors[index]).toMatchObject({
274 operation,
275 userId: "user-1",
276 deviceId: "device-1",
277 kind,
278 cause,
279 message: `Failed to persist Live Activity state during ${operation} for user user-1 and device device-1.`,
280 });
281 }
282 }).pipe(
283 Effect.provide(
284 LiveActivities.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb))),
285 ),
286 );
287 });
200});288});
infra/relay/src/agentActivity/LiveActivities.ts+153-74
...@@ -3,7 +3,10 @@ import type {...@@ -3,7 +3,10 @@ import type {
3 RelayDeliveryKind,3 RelayDeliveryKind,
4 RelayLiveActivityRegistrationRequest,4 RelayLiveActivityRegistrationRequest,
5} from "@t3tools/contracts/relay";5} from "@t3tools/contracts/relay";
6import { RelayAgentActivityAggregateState as RelayAgentActivityAggregateStateSchema } from "@t3tools/contracts/relay";6import {
7 RelayAgentActivityAggregateState as RelayAgentActivityAggregateStateSchema,
8 RelayDeliveryKind as RelayDeliveryKindSchema,
9} from "@t3tools/contracts/relay";
7import * as Context from "effect/Context";10import * as Context from "effect/Context";
8import * as DateTime from "effect/DateTime";11import * as DateTime from "effect/DateTime";
9import * as Effect from "effect/Effect";12import * as Effect from "effect/Effect";
...@@ -17,28 +20,46 @@ import { relayLiveActivities, relayMobileDevices } from "../persistence/schema.t...@@ -17,28 +20,46 @@ import { relayLiveActivities, relayMobileDevices } from "../persistence/schema.t
1720
18export class LiveActivityRegistrationPersistenceError extends Schema.TaggedErrorClass<LiveActivityRegistrationPersistenceError>()(21export class LiveActivityRegistrationPersistenceError extends Schema.TaggedErrorClass<LiveActivityRegistrationPersistenceError>()(
19 "LiveActivityRegistrationPersistenceError",22 "LiveActivityRegistrationPersistenceError",
20 { cause: Schema.Defect() },23 {
24 userId: Schema.String,
25 deviceId: Schema.String,
26 cause: Schema.Defect(),
27 },
21) {28) {
22 override get message(): string {29 override get message(): string {
23 return "Failed to persist Live Activity registration";30 return `Failed to persist Live Activity registration for user ${this.userId} and device ${this.deviceId}.`;
24 }31 }
25}32}
2633
27export class LiveActivityTargetListPersistenceError extends Schema.TaggedErrorClass<LiveActivityTargetListPersistenceError>()(34export class LiveActivityTargetListPersistenceError extends Schema.TaggedErrorClass<LiveActivityTargetListPersistenceError>()(
28 "LiveActivityTargetListPersistenceError",35 "LiveActivityTargetListPersistenceError",
29 { cause: Schema.Defect() },36 {
37 userId: Schema.String,
38 cause: Schema.Defect(),
39 },
30) {40) {
31 override get message(): string {41 override get message(): string {
32 return "Failed to list Live Activity delivery targets";42 return `Failed to list Live Activity delivery targets for user ${this.userId}.`;
33 }43 }
34}44}
3545
36export class LiveActivityDeliveryMarkPersistenceError extends Schema.TaggedErrorClass<LiveActivityDeliveryMarkPersistenceError>()(46export class LiveActivityDeliveryMarkPersistenceError extends Schema.TaggedErrorClass<LiveActivityDeliveryMarkPersistenceError>()(
37 "LiveActivityDeliveryMarkPersistenceError",47 "LiveActivityDeliveryMarkPersistenceError",
38 { cause: Schema.Defect() },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 },
39) {60) {
40 override get message(): string {61 override get message(): string {
41 return "Failed to persist Live Activity delivery state";62 return `Failed to persist Live Activity state during ${this.operation} for user ${this.userId} and device ${this.deviceId}.`;
42 }63 }
43}64}
4465
...@@ -110,11 +131,11 @@ export const make = Effect.gen(function* () {...@@ -110,11 +131,11 @@ export const make = Effect.gen(function* () {
110 const db = yield* RelayDb.RelayDb;131 const db = yield* RelayDb.RelayDb;
111132
112 return LiveActivities.of({133 return LiveActivities.of({
113 register: Effect.fn("relay.live_activities.register")(134 register: Effect.fn("relay.live_activities.register")(function* (input) {
114 function* (input) {135 yield* Effect.annotateCurrentSpan({
115 yield* Effect.annotateCurrentSpan({136 "relay.mobile.device_id": input.registration.deviceId,
116 "relay.mobile.device_id": input.registration.deviceId,137 });
117 });138 yield* Effect.gen(function* () {
118 const updatedAt = DateTime.formatIso(yield* DateTime.now);139 const updatedAt = DateTime.formatIso(yield* DateTime.now);
119 const registration = input.registration;140 const registration = input.registration;
120141
...@@ -155,9 +176,17 @@ export const make = Effect.gen(function* () {...@@ -155,9 +176,17 @@ export const make = Effect.gen(function* () {
155 updatedAt,176 updatedAt,
156 },177 },
157 });178 });
158 },179 }).pipe(
159 Effect.mapError((cause) => new LiveActivityRegistrationPersistenceError({ cause })),180 Effect.mapError(
160 ),181 (cause) =>
182 new LiveActivityRegistrationPersistenceError({
183 userId: input.userId,
184 deviceId: input.registration.deviceId,
185 cause,
186 }),
187 ),
188 );
189 }),
161190
162 listTargets: Effect.fn("relay.live_activities.list_targets")(function* (input) {191 listTargets: Effect.fn("relay.live_activities.list_targets")(function* (input) {
163 return yield* db192 return yield* db
...@@ -207,16 +236,22 @@ export const make = Effect.gen(function* () {...@@ -207,16 +236,22 @@ export const make = Effect.gen(function* () {
207 ),236 ),
208 ),237 ),
209 Effect.map((rows): ReadonlyArray<TargetRow> => rows),238 Effect.map((rows): ReadonlyArray<TargetRow> => rows),
210 Effect.mapError((cause) => new LiveActivityTargetListPersistenceError({ cause })),239 Effect.mapError(
240 (cause) =>
241 new LiveActivityTargetListPersistenceError({
242 userId: input.userId,
243 cause,
244 }),
245 ),
211 );246 );
212 }),247 }),
213248
214 markDelivery: Effect.fn("relay.live_activities.mark_delivery")(249 markDelivery: Effect.fn("relay.live_activities.mark_delivery")(function* (input) {
215 function* (input) {250 yield* Effect.annotateCurrentSpan({
216 yield* Effect.annotateCurrentSpan({251 "relay.mobile.device_id": input.deviceId,
217 "relay.mobile.device_id": input.deviceId,252 "relay.delivery.kind": input.kind,
218 "relay.delivery.kind": input.kind,253 });
219 });254 yield* Effect.gen(function* () {
220 const aggregateJson =255 const aggregateJson =
221 input.aggregate === null256 input.aggregate === null
222 ? null257 ? null
...@@ -257,9 +292,19 @@ export const make = Effect.gen(function* () {...@@ -257,9 +292,19 @@ export const make = Effect.gen(function* () {
257 updatedAt: input.deliveredAt,292 updatedAt: input.deliveredAt,
258 },293 },
259 });294 });
260 },295 }).pipe(
261 Effect.mapError((cause) => new LiveActivityDeliveryMarkPersistenceError({ cause })),296 Effect.mapError(
262 ),297 (cause) =>
298 new LiveActivityDeliveryMarkPersistenceError({
299 operation: "mark-delivery",
300 userId: input.userId,
301 deviceId: input.deviceId,
302 kind: input.kind,
303 cause,
304 }),
305 ),
306 );
307 }),
263308
264 markStartQueued: Effect.fn("relay.live_activities.mark_start_queued")(function* (input) {309 markStartQueued: Effect.fn("relay.live_activities.mark_start_queued")(function* (input) {
265 yield* Effect.annotateCurrentSpan({310 yield* Effect.annotateCurrentSpan({
...@@ -287,7 +332,18 @@ export const make = Effect.gen(function* () {...@@ -287,7 +332,18 @@ export const make = Effect.gen(function* () {
287 updatedAt: input.queuedAt,332 updatedAt: input.queuedAt,
288 },333 },
289 })334 })
290 .pipe(Effect.mapError((cause) => new LiveActivityDeliveryMarkPersistenceError({ cause })));335 .pipe(
336 Effect.mapError(
337 (cause) =>
338 new LiveActivityDeliveryMarkPersistenceError({
339 operation: "mark-start-queued",
340 userId: input.userId,
341 deviceId: input.deviceId,
342 kind: null,
343 cause,
344 }),
345 ),
346 );
291 }),347 }),
292348
293 clearStartQueued: Effect.fn("relay.live_activities.clear_start_queued")(function* (input) {349 clearStartQueued: Effect.fn("relay.live_activities.clear_start_queued")(function* (input) {
...@@ -303,7 +359,18 @@ export const make = Effect.gen(function* () {...@@ -303,7 +359,18 @@ export const make = Effect.gen(function* () {
303 eq(relayLiveActivities.deviceId, input.deviceId),359 eq(relayLiveActivities.deviceId, input.deviceId),
304 ),360 ),
305 )361 )
306 .pipe(Effect.mapError((cause) => new LiveActivityDeliveryMarkPersistenceError({ cause })));362 .pipe(
363 Effect.mapError(
364 (cause) =>
365 new LiveActivityDeliveryMarkPersistenceError({
366 operation: "clear-start-queued",
367 userId: input.userId,
368 deviceId: input.deviceId,
369 kind: null,
370 cause,
371 }),
372 ),
373 );
307 }),374 }),
308375
309 invalidateDeliveryToken: Effect.fn("relay.live_activities.invalidate_delivery_token")(376 invalidateDeliveryToken: Effect.fn("relay.live_activities.invalidate_delivery_token")(
...@@ -312,39 +379,58 @@ export const make = Effect.gen(function* () {...@@ -312,39 +379,58 @@ export const make = Effect.gen(function* () {
312 "relay.mobile.device_id": input.deviceId,379 "relay.mobile.device_id": input.deviceId,
313 "relay.delivery.kind": input.kind,380 "relay.delivery.kind": input.kind,
314 });381 });
315 if (input.kind === "push_notification") {382 yield* Effect.gen(function* () {
316 yield* db383 if (input.kind === "push_notification") {
317 .update(relayMobileDevices)384 yield* db
318 .set({385 .update(relayMobileDevices)
319 pushToken: null,386 .set({
320 updatedAt: input.invalidatedAt,387 pushToken: null,
321 })388 updatedAt: input.invalidatedAt,
322 .where(389 })
323 and(390 .where(
324 eq(relayMobileDevices.userId, input.userId),391 and(
325 eq(relayMobileDevices.deviceId, input.deviceId),392 eq(relayMobileDevices.userId, input.userId),
326 ),393 eq(relayMobileDevices.deviceId, input.deviceId),
327 );394 ),
328 return;395 );
329 }396 return;
397 }
398
399 if (input.kind === "live_activity_start") {
400 yield* db
401 .update(relayMobileDevices)
402 .set({
403 pushToStartToken: null,
404 updatedAt: input.invalidatedAt,
405 })
406 .where(
407 and(
408 eq(relayMobileDevices.userId, input.userId),
409 eq(relayMobileDevices.deviceId, input.deviceId),
410 ),
411 );
412 yield* db
413 .update(relayLiveActivities)
414 .set({
415 remoteStartQueuedAt: null,
416 updatedAt: input.invalidatedAt,
417 })
418 .where(
419 and(
420 eq(relayLiveActivities.userId, input.userId),
421 eq(relayLiveActivities.deviceId, input.deviceId),
422 ),
423 );
424 return;
425 }
330426
331 if (input.kind === "live_activity_start") {
332 yield* db
333 .update(relayMobileDevices)
334 .set({
335 pushToStartToken: null,
336 updatedAt: input.invalidatedAt,
337 })
338 .where(
339 and(
340 eq(relayMobileDevices.userId, input.userId),
341 eq(relayMobileDevices.deviceId, input.deviceId),
342 ),
343 );
344 yield* db427 yield* db
345 .update(relayLiveActivities)428 .update(relayLiveActivities)
346 .set({429 .set({
430 activityPushToken: null,
347 remoteStartQueuedAt: null,431 remoteStartQueuedAt: null,
432 remoteStartedAt: null,
433 endedAt: input.invalidatedAt,
348 updatedAt: input.invalidatedAt,434 updatedAt: input.invalidatedAt,
349 })435 })
350 .where(436 .where(
...@@ -353,26 +439,19 @@ export const make = Effect.gen(function* () {...@@ -353,26 +439,19 @@ export const make = Effect.gen(function* () {
353 eq(relayLiveActivities.deviceId, input.deviceId),439 eq(relayLiveActivities.deviceId, input.deviceId),
354 ),440 ),
355 );441 );
356 return;442 }).pipe(
357 }443 Effect.mapError(
358444 (cause) =>
359 yield* db445 new LiveActivityDeliveryMarkPersistenceError({
360 .update(relayLiveActivities)446 operation: "invalidate-delivery-token",
361 .set({447 userId: input.userId,
362 activityPushToken: null,448 deviceId: input.deviceId,
363 remoteStartQueuedAt: null,449 kind: input.kind,
364 remoteStartedAt: null,450 cause,
365 endedAt: input.invalidatedAt,451 }),
366 updatedAt: input.invalidatedAt,452 ),
367 })453 );
368 .where(
369 and(
370 eq(relayLiveActivities.userId, input.userId),
371 eq(relayLiveActivities.deviceId, input.deviceId),
372 ),
373 );
374 },454 },
375 Effect.mapError((cause) => new LiveActivityDeliveryMarkPersistenceError({ cause })),
376 ),455 ),
377 });456 });
378});457});