1import type {
2 RelayAgentActivityAggregateState,
3 RelayLiveActivityRegistrationRequest,
4} from "@t3tools/contracts/relay";
5import { describe, expect, it } from "@effect/vitest";
6import type { SQL } from "drizzle-orm";
7import { PgDialect } from "drizzle-orm/pg-core";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10
11import * as RelayDb from "../db.ts";
12import { relayLiveActivities } from "../persistence/schema.ts";
13import * as LiveActivities from "./LiveActivities.ts";
14
15const aggregate: RelayAgentActivityAggregateState = {
16 title: "T3 Code",
17 subtitle: "Agent work in progress",
18 activeCount: 1,
19 updatedAt: "2026-05-25T00:00:00.000Z",
20 activities: [
21 {
22 environmentId:
23 "env" as RelayAgentActivityAggregateState["activities"][number]["environmentId"],
24 threadId: "thread" as RelayAgentActivityAggregateState["activities"][number]["threadId"],
25 projectTitle: "Project",
26 threadTitle: "Thread",
27 modelTitle: "gpt-5.4",
28 phase: "running",
29 status: "Working",
30 updatedAt: "2026-05-25T00:00:00.000Z",
31 deepLink: "/threads/env/thread",
32 },
33 ],
34};
35
36describe("LiveActivities", () => {
37 it.effect(
38 "claims Live Activity push tokens globally before upserting the current user device",
39 () => {
40 const registration: RelayLiveActivityRegistrationRequest = {
41 deviceId: "device-1" as RelayLiveActivityRegistrationRequest["deviceId"],
42 activityPushToken:
43 "activity-push-token" as RelayLiveActivityRegistrationRequest["activityPushToken"],
44 };
45 const calls: Array<string> = [];
46 const updateSets: Array<Record<string, unknown>> = [];
47 const updateConditions: Array<SQL> = [];
48 const insertedValues: Array<Record<string, unknown>> = [];
49 const conflictConfigs: Array<{
50 readonly set?: Record<string, unknown>;
51 }> = [];
52 const dialect = new PgDialect();
53
54 const fakeDb = {
55 update: (table: unknown) => {
56 expect(table).toBe(relayLiveActivities);
57 calls.push("update");
58 return {
59 set: (values: Record<string, unknown>) => {
60 updateSets.push(values);
61 calls.push("update.set");
62 return {
63 where: (condition: SQL) => {
64 expect(condition).toBeDefined();
65 updateConditions.push(condition);
66 calls.push("update.where");
67 return Effect.void;
68 },
69 };
70 },
71 };
72 },
73 insert: (table: unknown) => {
74 expect(table).toBe(relayLiveActivities);
75 calls.push("insert");
76 return {
77 values: (values: Record<string, unknown>) => {
78 insertedValues.push(values);
79 calls.push("insert.values");
80 return {
81 onConflictDoUpdate: (config: { readonly set?: Record<string, unknown> }) => {
82 expect(config).toBeDefined();
83 conflictConfigs.push(config);
84 calls.push("insert.onConflictDoUpdate");
85 return Effect.void;
86 },
87 };
88 },
89 };
90 },
91 } as unknown as RelayDb.RelayDb["Service"];
92
93 return Effect.gen(function* () {
94 const liveActivities = yield* LiveActivities.LiveActivities;
95 yield* liveActivities.register({ userId: "user-2", registration });
96
97 expect(calls).toEqual([
98 "update",
99 "update.set",
100 "update.where",
101 "insert",
102 "insert.values",
103 "insert.onConflictDoUpdate",
104 ]);
105 expect(updateSets).toEqual([
106 expect.objectContaining({
107 activityPushToken: null,
108 remoteStartQueuedAt: null,
109 remoteStartedAt: null,
110 }),
111 ]);
112 expect(updateConditions.map((condition) => dialect.sqlToQuery(condition))).toEqual([
113 {
114 sql: '"relay_live_activities"."activity_push_token" = $1',
115 params: ["activity-push-token"],
116 },
117 ]);
118 expect(insertedValues).toEqual([
119 expect.objectContaining({
120 userId: "user-2",
121 deviceId: "device-1",
122 activityPushToken: "activity-push-token",
123 remoteStartQueuedAt: null,
124 remoteStartedAt: expect.any(String),
125 endedAt: null,
126 lastAggregateJson: null,
127 lastLiveActivityDeliveryAt: null,
128 }),
129 ]);
130 expect(conflictConfigs[0]?.set).toEqual(
131 expect.objectContaining({
132 activityPushToken: "activity-push-token",
133 remoteStartQueuedAt: null,
134 remoteStartedAt: expect.any(String),
135 endedAt: null,
136 lastAggregateJson: null,
137 lastLiveActivityDeliveryAt: null,
138 }),
139 );
140 }).pipe(
141 Effect.provide(
142 LiveActivities.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb))),
143 ),
144 );
145 },
146 );
147
148 it.effect("preserves ended state when a delayed update delivery is marked", () => {
149 const insertedValues: Array<Record<string, unknown>> = [];
150 const conflictConfigs: Array<{
151 readonly set?: Record<string, unknown>;
152 }> = [];
153
154 const fakeDb = {
155 insert: (table: unknown) => {
156 expect(table).toBe(relayLiveActivities);
157 return {
158 values: (values: Record<string, unknown>) => {
159 insertedValues.push(values);
160 return {
161 onConflictDoUpdate: (config: { readonly set?: Record<string, unknown> }) => {
162 conflictConfigs.push(config);
163 return Effect.void;
164 },
165 };
166 },
167 };
168 },
169 } as unknown as RelayDb.RelayDb["Service"];
170
171 return Effect.gen(function* () {
172 const liveActivities = yield* LiveActivities.LiveActivities;
173 yield* liveActivities.markDelivery({
174 userId: "user-2",
175 deviceId: "device-1",
176 kind: "live_activity_update",
177 aggregate,
178 deliveredAt: "2026-05-25T00:00:10.000Z",
179 });
180
181 expect(insertedValues).toEqual([
182 expect.objectContaining({
183 userId: "user-2",
184 deviceId: "device-1",
185 endedAt: null,
186 }),
187 ]);
188 expect(conflictConfigs[0]?.set).toEqual(
189 expect.objectContaining({
190 endedAt: relayLiveActivities.endedAt,
191 lastLiveActivityDeliveryAt: "2026-05-25T00:00:10.000Z",
192 }),
193 );
194 }).pipe(
195 Effect.provide(
196 LiveActivities.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb))),
197 ),
198 );
199 });
200
201 it.effect("retires the previous activity token when a start or end is delivered", () => {
202 const conflictConfigs: Array<{ readonly set?: Record<string, unknown> }> = [];
203 const fakeDb = {
204 insert: () => ({
205 values: () => ({
206 onConflictDoUpdate: (config: { readonly set?: Record<string, unknown> }) => {
207 conflictConfigs.push(config);
208 return Effect.void;
209 },
210 }),
211 }),
212 } as unknown as RelayDb.RelayDb["Service"];
213
214 return Effect.gen(function* () {
215 const liveActivities = yield* LiveActivities.LiveActivities;
216 const mark = (kind: "live_activity_start" | "live_activity_update" | "live_activity_end") =>
217 liveActivities.markDelivery({
218 userId: "user-2",
219 deviceId: "device-1",
220 kind,
221 aggregate,
222 deliveredAt: "2026-05-25T00:00:10.000Z",
223 });
224 yield* mark("live_activity_start");
225 yield* mark("live_activity_update");
226 yield* mark("live_activity_end");
227
228 // A start begins a new activity generation and an end retires the
229 // current one; both must drop the stored update token so later
230 // deliveries can't route to the dead activity. Plain updates keep it.
231 expect(conflictConfigs[0]?.set).toEqual(
232 expect.objectContaining({
233 activityPushToken: null,
234 remoteStartedAt: "2026-05-25T00:00:10.000Z",
235 endedAt: null,
236 }),
237 );
238 expect(conflictConfigs[1]?.set?.activityPushToken).not.toBeNull();
239 expect(conflictConfigs[2]?.set).toEqual(
240 expect.objectContaining({
241 activityPushToken: null,
242 endedAt: "2026-05-25T00:00:10.000Z",
243 }),
244 );
245 }).pipe(
246 Effect.provide(
247 LiveActivities.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb))),
248 ),
249 );
250 });
251
252 it.effect("preserves correlation context and causes for persistence failures", () => {
253 const cause = new Error("database unavailable");
254 const registration: RelayLiveActivityRegistrationRequest = {
255 deviceId: "device-1" as RelayLiveActivityRegistrationRequest["deviceId"],
256 activityPushToken:
257 "activity-push-token" as RelayLiveActivityRegistrationRequest["activityPushToken"],
258 };
259 const fakeDb = {
260 update: () => ({
261 set: () => ({ where: () => Effect.fail(cause) }),
262 }),
263 insert: () => ({
264 values: () => ({ onConflictDoUpdate: () => Effect.fail(cause) }),
265 }),
266 select: () => ({
267 from: () => ({
268 leftJoin: () => ({ where: () => Effect.fail(cause) }),
269 }),
270 }),
271 } as unknown as RelayDb.RelayDb["Service"];
272
273 return Effect.gen(function* () {
274 const liveActivities = yield* LiveActivities.LiveActivities;
275 const registrationError = yield* Effect.flip(
276 liveActivities.register({ userId: "user-1", registration }),
277 );
278 const targetListError = yield* Effect.flip(liveActivities.listTargets({ userId: "user-1" }));
279 const deliveryErrors = yield* Effect.forEach(
280 [
281 liveActivities.markDelivery({
282 userId: "user-1",
283 deviceId: "device-1",
284 kind: "live_activity_update",
285 aggregate: null,
286 deliveredAt: "2026-05-25T00:00:10.000Z",
287 }),
288 liveActivities.markStartQueued({
289 userId: "user-1",
290 deviceId: "device-1",
291 queuedAt: "2026-05-25T00:00:10.000Z",
292 }),
293 liveActivities.clearStartQueued({ userId: "user-1", deviceId: "device-1" }),
294 liveActivities.invalidateDeliveryToken({
295 userId: "user-1",
296 deviceId: "device-1",
297 kind: "push_notification",
298 invalidatedAt: "2026-05-25T00:00:10.000Z",
299 }),
300 ],
301 (effect) => Effect.flip(effect),
302 { concurrency: 1 },
303 );
304
305 expect(registrationError).toMatchObject({
306 userId: "user-1",
307 deviceId: "device-1",
308 cause,
309 message:
310 "Failed to persist Live Activity registration for user user-1 and device device-1.",
311 });
312 expect(targetListError).toMatchObject({
313 userId: "user-1",
314 cause,
315 message: "Failed to list Live Activity delivery targets for user user-1.",
316 });
317
318 const expectedDeliveryContext = [
319 ["mark-delivery", "live_activity_update"],
320 ["mark-start-queued", null],
321 ["clear-start-queued", null],
322 ["invalidate-delivery-token", "push_notification"],
323 ] as const;
324 for (const [index, [operation, kind]] of expectedDeliveryContext.entries()) {
325 expect(deliveryErrors[index]).toMatchObject({
326 operation,
327 userId: "user-1",
328 deviceId: "device-1",
329 kind,
330 cause,
331 message: `Failed to persist Live Activity state during ${operation} for user user-1 and device device-1.`,
332 });
333 }
334 }).pipe(
335 Effect.provide(
336 LiveActivities.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb))),
337 ),
338 );
339 });
340});