| author |  Theo Browne <me@t3.gg> 2026-09-04 09:01:57-07:00 |
| committer |  GitHub <noreply@github.com> 2026-09-04 09:01:57-07:00 |
| log | c75299ee2085a121bceb6df76796e971fe92b5b6 |
| tree | 4db8c44b3bf06e3d6cd1b698ecc8ace8a3fc987f |
| parent | dffb4cd3b16dc6f41aced99922950ee3083082c6 |
| signature | Signed by PGP key B5690EEEBB952194 |
perf(relay): avoid repeated activity decoding (#9708)
4 files changed, 44 insertions(+), 25 deletions(-)
infra
infra/relay/src/agentActivity/AgentActivityPublisher.test.ts+2-8
| ... | ... | @@ -517,7 +517,7 @@ describe("AgentActivityPublisher", () => { |
| 517 | 517 | }); |
| 518 | 518 | |
| 519 | 519 | it.effect( |
| 520 | "does not build Live Activity aggregates for links with Live Activities disabled", | |
| 520 | "delivers notifications without querying activity rows when Live Activities are disabled", | |
| 521 | 521 | () => { |
| 522 | 522 | const notificationState: RelayAgentActivityState = { |
| 523 | 523 | ...state, |
| ... | ... | @@ -549,13 +549,7 @@ describe("AgentActivityPublisher", () => { |
| 549 | 549 | AgentActivityRows.AgentActivityRows, |
| 550 | 550 | makeAgentActivityRows({ |
| 551 | 551 | listForUser: () => |
| 552 | Effect.succeed([ | |
| 553 | { | |
| 554 | ...state, | |
| 555 | environmentId: "other-env" as RelayAgentActivityState["environmentId"], | |
| 556 | threadId: "other-thread" as RelayAgentActivityState["threadId"], | |
| 557 | }, | |
| 558 | ]), | |
| 552 | Effect.die("Notification-only delivery must not read rows"), | |
| 559 | 553 | }), |
| 560 | 554 | ), |
| 561 | 555 | Layer.succeed( |
infra/relay/src/agentActivity/AgentActivityPublisher.ts+3-1
| ... | ... | @@ -58,7 +58,9 @@ export const make = Effect.gen(function* () { |
| 58 | 58 | readonly state: RelayAgentActivityState | null; |
| 59 | 59 | readonly nowMs: number; |
| 60 | 60 | }) { |
| 61 | const activeStates = yield* rows.listForUser({ userId: input.deliveryUser.userId }); | |
| 61 | const activeStates = input.deliveryUser.liveActivitiesEnabled | |
| 62 | ? yield* rows.listForUser({ userId: input.deliveryUser.userId }) | |
| 63 | : []; | |
| 62 | 64 | const liveActivityAggregate = input.deliveryUser.liveActivitiesEnabled |
| 63 | 65 | ? makeAggregateState({ |
| 64 | 66 | activeStates, |
infra/relay/src/agentActivity/AgentActivityRows.test.ts+36
| ... | ... | @@ -19,6 +19,42 @@ const state: RelayAgentActivityState = { |
| 19 | 19 | }; |
| 20 | 20 | |
| 21 | 21 | describe("AgentActivityRows", () => { |
| 22 | it.effect("validates database JSON objects and ignores invalid activity rows", () => { | |
| 23 | const queryRows = [ | |
| 24 | { stateJson: null }, | |
| 25 | { stateJson: JSON.stringify(state) }, | |
| 26 | { stateJson: { ...state, phase: "invalid" } }, | |
| 27 | { stateJson: { ...state, unrelated: "ignored" } }, | |
| 28 | ]; | |
| 29 | const db = { | |
| 30 | select: () => ({ | |
| 31 | from: () => ({ | |
| 32 | innerJoin: () => ({ | |
| 33 | where: () => ({ | |
| 34 | orderBy: () => Effect.succeed(queryRows), | |
| 35 | }), | |
| 36 | }), | |
| 37 | }), | |
| 38 | }), | |
| 39 | } as unknown as RelayDb.RelayDb["Service"]; | |
| 40 | ||
| 41 | return Effect.gen(function* () { | |
| 42 | const rows = yield* AgentActivityRows.AgentActivityRows; | |
| 43 | expect(yield* rows.listForUser({ userId: "user-2" })).toEqual([state]); | |
| 44 | expect( | |
| 45 | yield* rows.getForUserThread({ | |
| 46 | userId: "user-2", | |
| 47 | environmentId: "env-1", | |
| 48 | threadId: "thread-1", | |
| 49 | }), | |
| 50 | ).toEqual(state); | |
| 51 | }).pipe( | |
| 52 | Effect.provide( | |
| 53 | AgentActivityRows.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, db))), | |
| 54 | ), | |
| 55 | ); | |
| 56 | }); | |
| 57 | ||
| 22 | 58 | it.effect("preserves activity context on persistence failures", () => { |
| 23 | 59 | const cause = new Error("database unavailable"); |
| 24 | 60 | const failingDb = { |
infra/relay/src/agentActivity/AgentActivityRows.ts+3-16
| ... | ... | @@ -92,15 +92,12 @@ export class AgentActivityRows extends Context.Service< |
| 92 | 92 | >()("t3code-relay/agentActivity/AgentActivityRows") {} |
| 93 | 93 | |
| 94 | 94 | const decodeJsonString = Schema.decodeEffect(Schema.fromJsonString(Schema.Unknown)); |
| 95 | const encodeJsonValue = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); | |
| 96 | 95 | |
| 97 | 96 | const encodeRelayAgentActivityStateJson = Schema.encodeEffect( |
| 98 | 97 | Schema.fromJsonString(RelayAgentActivityStateSchema), |
| 99 | 98 | ); |
| 100 | 99 | |
| 101 | const decodeRelayAgentActivityStateJson = Schema.decodeUnknownOption( | |
| 102 | Schema.fromJsonString(RelayAgentActivityStateSchema), | |
| 103 | ); | |
| 100 | const decodeRelayAgentActivityState = Schema.decodeUnknownOption(RelayAgentActivityStateSchema); | |
| 104 | 101 | |
| 105 | 102 | export const make = Effect.gen(function* () { |
| 106 | 103 | const db = yield* RelayDb.RelayDb; |
| ... | ... | @@ -229,13 +226,8 @@ export const make = Effect.gen(function* () { |
| 229 | 226 | ) |
| 230 | 227 | .orderBy(desc(relayAgentActivityRows.updatedAt)) |
| 231 | 228 | .pipe( |
| 232 | Effect.flatMap((rows) => | |
| 233 | Effect.forEach(rows, (row) => encodeJsonValue(row.stateJson), { | |
| 234 | concurrency: "unbounded", | |
| 235 | }), | |
| 236 | ), | |
| 237 | 229 | Effect.map((rows) => |
| 238 | rows.flatMap((row) => Option.toArray(decodeRelayAgentActivityStateJson(row))), | |
| 230 | rows.flatMap((row) => Option.toArray(decodeRelayAgentActivityState(row.stateJson))), | |
| 239 | 231 | ), |
| 240 | 232 | Effect.mapError( |
| 241 | 233 | (cause) => |
| ... | ... | @@ -271,14 +263,9 @@ export const make = Effect.gen(function* () { |
| 271 | 263 | ) |
| 272 | 264 | .orderBy(desc(relayAgentActivityRows.updatedAt)) |
| 273 | 265 | .pipe( |
| 274 | Effect.flatMap((rows) => | |
| 275 | Effect.forEach(rows, (row) => encodeJsonValue(row.stateJson), { | |
| 276 | concurrency: "unbounded", | |
| 277 | }), | |
| 278 | ), | |
| 279 | 266 | Effect.map((rows) => { |
| 280 | 267 | for (const row of rows) { |
| 281 | const decoded = decodeRelayAgentActivityStateJson(row); | |
| 268 | const decoded = decodeRelayAgentActivityState(row.stateJson); | |
| 282 | 269 | if (Option.isSome(decoded)) { |
| 283 | 270 | return decoded.value; |
| 284 | 271 | } |