authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-01 04:09:06-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 04:09:06-07:00
loge86604d3372acccd9f6a33a2c4ae46f4e2685541
tree0bb7a188f387041840cec2dc6aa33970591bd8c2
parent3c73fa7ce02b7ee6b2904ec58e47c06262aebf5d
signature Signed by PGP key B5690EEEBB952194

perf(server): skip full-message reads while streaming (#9032)

5 files changed, 205 insertions(+), 17 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+52-7
...@@ -1462,6 +1462,8 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -1462,6 +1462,8 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
1462 const eventStore = yield* OrchestrationEventStore;1462 const eventStore = yield* OrchestrationEventStore;
1463 const sql = yield* SqlClient.SqlClient;1463 const sql = yield* SqlClient.SqlClient;
1464 const now = "2026-01-01T00:00:00.000Z";1464 const now = "2026-01-01T00:00:00.000Z";
1465 const streamingAt = "2026-01-01T00:00:01.000Z";
1466 const completedAt = "2026-01-01T00:00:02.000Z";
14651467
1466 yield* eventStore.append({1468 yield* eventStore.append({
1467 type: "project.created",1469 type: "project.created",
...@@ -1526,7 +1528,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -1526,7 +1528,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
1526 role: "assistant",1528 role: "assistant",
1527 text: "hello",1529 text: "hello",
1528 turnId: null,1530 turnId: null,
1529 streaming: false,1531 streaming: true,
1530 createdAt: now,1532 createdAt: now,
1531 updatedAt: now,1533 updatedAt: now,
1532 },1534 },
...@@ -1539,7 +1541,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -1539,7 +1541,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
1539 eventId: EventId.make("evt-a4"),1541 eventId: EventId.make("evt-a4"),
1540 aggregateKind: "thread",1542 aggregateKind: "thread",
1541 aggregateId: ThreadId.make("thread-a"),1543 aggregateId: ThreadId.make("thread-a"),
1542 occurredAt: now,1544 occurredAt: streamingAt,
1543 commandId: CommandId.make("cmd-a4"),1545 commandId: CommandId.make("cmd-a4"),
1544 causationEventId: null,1546 causationEventId: null,
1545 correlationId: CorrelationId.make("cmd-a4"),1547 correlationId: CorrelationId.make("cmd-a4"),
...@@ -1551,18 +1553,61 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -1551,18 +1553,61 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
1551 text: " world",1553 text: " world",
1552 turnId: null,1554 turnId: null,
1553 streaming: true,1555 streaming: true,
1554 createdAt: now,1556 createdAt: streamingAt,
1555 updatedAt: now,1557 updatedAt: streamingAt,
1556 },1558 },
1557 });1559 });
15581560
1559 yield* projectionPipeline.bootstrap;1561 yield* projectionPipeline.bootstrap;
1560 yield* projectionPipeline.bootstrap;1562 yield* projectionPipeline.bootstrap;
15611563
1562 const messageRows = yield* sql<{ readonly text: string }>`1564 yield* eventStore.append({
1563 SELECT text FROM projection_thread_messages WHERE message_id = 'message-a'1565 type: "thread.message-sent",
1566 eventId: EventId.make("evt-a5"),
1567 aggregateKind: "thread",
1568 aggregateId: ThreadId.make("thread-a"),
1569 occurredAt: completedAt,
1570 commandId: CommandId.make("cmd-a5"),
1571 causationEventId: null,
1572 correlationId: CorrelationId.make("cmd-a5"),
1573 metadata: {},
1574 payload: {
1575 threadId: ThreadId.make("thread-a"),
1576 messageId: MessageId.make("message-a"),
1577 role: "assistant",
1578 text: "",
1579 turnId: null,
1580 streaming: false,
1581 createdAt: completedAt,
1582 updatedAt: completedAt,
1583 },
1584 });
1585
1586 yield* projectionPipeline.bootstrap;
1587 yield* projectionPipeline.bootstrap;
1588
1589 const messageRows = yield* sql<{
1590 readonly text: string;
1591 readonly isStreaming: number;
1592 readonly createdAt: string;
1593 readonly updatedAt: string;
1594 }>`
1595 SELECT
1596 text,
1597 is_streaming AS "isStreaming",
1598 created_at AS "createdAt",
1599 updated_at AS "updatedAt"
1600 FROM projection_thread_messages
1601 WHERE message_id = 'message-a'
1564 `;1602 `;
1565 assert.deepEqual(messageRows, [{ text: "hello world" }]);1603 assert.deepEqual(messageRows, [
1604 {
1605 text: "hello world",
1606 isStreaming: 0,
1607 createdAt: now,
1608 updatedAt: completedAt,
1609 },
1610 ]);
15661611
1567 const stateRows = yield* sql<{1612 const stateRows = yield* sql<{
1568 readonly projector: string;1613 readonly projector: string;
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+23-10
...@@ -1008,21 +1008,34 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -1008,21 +1008,34 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
1008 return;1008 return;
10091009
1010 case "thread.message-sent": {1010 case "thread.message-sent": {
1011 if (event.payload.streaming) {
1012 const attachments =
1013 event.payload.attachments !== undefined
1014 ? yield* materializeAttachmentsForProjection({
1015 attachments: event.payload.attachments,
1016 })
1017 : undefined;
1018 yield* projectionThreadMessageRepository.appendStreaming({
1019 messageId: event.payload.messageId,
1020 threadId: event.payload.threadId,
1021 turnId: event.payload.turnId,
1022 role: event.payload.role,
1023 text: event.payload.text,
1024 ...(attachments !== undefined ? { attachments: [...attachments] } : {}),
1025 createdAt: event.payload.createdAt,
1026 updatedAt: event.payload.updatedAt,
1027 });
1028 return;
1029 }
1030
1011 const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({1031 const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({
1012 messageId: event.payload.messageId,1032 messageId: event.payload.messageId,
1013 });1033 });
1014 const previousMessage = Option.getOrUndefined(existingMessage);1034 const previousMessage = Option.getOrUndefined(existingMessage);
1015 const nextText = Option.match(existingMessage, {1035 const nextText = Option.match(existingMessage, {
1016 onNone: () => event.payload.text,1036 onNone: () => event.payload.text,
1017 onSome: (message) => {1037 onSome: (message) =>
1018 if (event.payload.streaming) {1038 event.payload.text.length === 0 ? message.text : event.payload.text,
1019 return `${message.text}${event.payload.text}`;
1020 }
1021 if (event.payload.text.length === 0) {
1022 return message.text;
1023 }
1024 return event.payload.text;
1025 },
1026 });1039 });
1027 const nextAttachments =1040 const nextAttachments =
1028 event.payload.attachments !== undefined1041 event.payload.attachments !== undefined
...@@ -1037,7 +1050,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -1037,7 +1050,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
1037 role: event.payload.role,1050 role: event.payload.role,
1038 text: nextText,1051 text: nextText,
1039 ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),1052 ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),
1040 isStreaming: event.payload.streaming,1053 isStreaming: false,
1041 createdAt: previousMessage?.createdAt ?? event.payload.createdAt,1054 createdAt: previousMessage?.createdAt ?? event.payload.createdAt,
1042 updatedAt: event.payload.updatedAt,1055 updatedAt: event.payload.updatedAt,
1043 });1056 });
apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts+65
...@@ -12,6 +12,71 @@ const layer = it.layer(...@@ -12,6 +12,71 @@ const layer = it.layer(
12);12);
1313
14layer("ProjectionThreadMessageRepository", (it) => {14layer("ProjectionThreadMessageRepository", (it) => {
15 it.effect("appends streaming text and applies attachment updates", () =>
16 Effect.gen(function* () {
17 const repository = yield* ProjectionThreadMessageRepository;
18 const threadId = ThreadId.make("thread-streaming-append");
19 const messageId = MessageId.make("message-streaming-append");
20 const createdAt = "2026-02-28T19:05:00.000Z";
21 const attachments = [
22 {
23 type: "image" as const,
24 id: "thread-streaming-append-att-1",
25 name: "example.png",
26 mimeType: "image/png",
27 sizeBytes: 5,
28 },
29 ];
30
31 yield* repository.appendStreaming({
32 messageId,
33 threadId,
34 turnId: null,
35 role: "assistant",
36 text: "hello",
37 attachments,
38 createdAt,
39 updatedAt: createdAt,
40 });
41 yield* repository.appendStreaming({
42 messageId,
43 threadId,
44 turnId: null,
45 role: "assistant",
46 text: " world",
47 createdAt: "2026-02-28T19:05:01.000Z",
48 updatedAt: "2026-02-28T19:05:01.000Z",
49 });
50
51 const rowWithPreservedAttachments = yield* repository.getByMessageId({ messageId });
52 assert.equal(rowWithPreservedAttachments._tag, "Some");
53 if (rowWithPreservedAttachments._tag === "Some") {
54 assert.deepEqual(rowWithPreservedAttachments.value.attachments, attachments);
55 }
56
57 yield* repository.appendStreaming({
58 messageId,
59 threadId,
60 turnId: null,
61 role: "assistant",
62 text: "",
63 attachments: [],
64 createdAt: "2026-02-28T19:05:02.000Z",
65 updatedAt: "2026-02-28T19:05:02.000Z",
66 });
67
68 const row = yield* repository.getByMessageId({ messageId });
69 assert.equal(row._tag, "Some");
70 if (row._tag === "Some") {
71 assert.equal(row.value.text, "hello world");
72 assert.deepEqual(row.value.attachments, []);
73 assert.equal(row.value.createdAt, createdAt);
74 assert.equal(row.value.updatedAt, "2026-02-28T19:05:02.000Z");
75 assert.isTrue(row.value.isStreaming);
76 }
77 }),
78 );
79
15 it.effect("preserves existing attachments when upsert omits attachments", () =>80 it.effect("preserves existing attachments when upsert omits attachments", () =>
16 Effect.gen(function* () {81 Effect.gen(function* () {
17 const repository = yield* ProjectionThreadMessageRepository;82 const repository = yield* ProjectionThreadMessageRepository;
apps/server/src/persistence/Layers/ProjectionThreadMessages.ts+53
...@@ -9,6 +9,7 @@ import { ChatAttachment } from "@t3tools/contracts";...@@ -9,6 +9,7 @@ import { ChatAttachment } from "@t3tools/contracts";
99
10import { toPersistenceSqlError } from "../Errors.ts";10import { toPersistenceSqlError } from "../Errors.ts";
11import {11import {
12 AppendStreamingProjectionThreadMessage,
12 GetProjectionThreadMessageInput,13 GetProjectionThreadMessageInput,
13 ProjectionThreadMessageRepository,14 ProjectionThreadMessageRepository,
14 type ProjectionThreadMessageRepositoryShape,15 type ProjectionThreadMessageRepositoryShape,
...@@ -95,6 +96,50 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -95,6 +96,50 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
95 },96 },
96 });97 });
9798
99 const appendStreamingProjectionThreadMessageRow = SqlSchema.void({
100 Request: AppendStreamingProjectionThreadMessage,
101 execute: (row) => {
102 const nextAttachmentsJson =
103 row.attachments !== undefined ? JSON.stringify(row.attachments) : null;
104 return sql`
105 INSERT INTO projection_thread_messages (
106 message_id,
107 thread_id,
108 turn_id,
109 role,
110 text,
111 attachments_json,
112 is_streaming,
113 created_at,
114 updated_at
115 )
116 VALUES (
117 ${row.messageId},
118 ${row.threadId},
119 ${row.turnId},
120 ${row.role},
121 ${row.text},
122 ${nextAttachmentsJson},
123 1,
124 ${row.createdAt},
125 ${row.updatedAt}
126 )
127 ON CONFLICT (message_id)
128 DO UPDATE SET
129 thread_id = excluded.thread_id,
130 turn_id = excluded.turn_id,
131 role = excluded.role,
132 text = projection_thread_messages.text || excluded.text,
133 attachments_json = COALESCE(
134 excluded.attachments_json,
135 projection_thread_messages.attachments_json
136 ),
137 is_streaming = 1,
138 updated_at = excluded.updated_at
139 `;
140 },
141 });
142
98 const getProjectionThreadMessageRow = SqlSchema.findOneOption({143 const getProjectionThreadMessageRow = SqlSchema.findOneOption({
99 Request: GetProjectionThreadMessageInput,144 Request: GetProjectionThreadMessageInput,
100 Result: ProjectionThreadMessageDbRowSchema,145 Result: ProjectionThreadMessageDbRowSchema,
...@@ -151,6 +196,13 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -151,6 +196,13 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
151 Effect.mapError(toPersistenceSqlError("ProjectionThreadMessageRepository.upsert:query")),196 Effect.mapError(toPersistenceSqlError("ProjectionThreadMessageRepository.upsert:query")),
152 );197 );
153198
199 const appendStreaming: ProjectionThreadMessageRepositoryShape["appendStreaming"] = (row) =>
200 appendStreamingProjectionThreadMessageRow(row).pipe(
201 Effect.mapError(
202 toPersistenceSqlError("ProjectionThreadMessageRepository.appendStreaming:query"),
203 ),
204 );
205
154 const getByMessageId: ProjectionThreadMessageRepositoryShape["getByMessageId"] = (input) =>206 const getByMessageId: ProjectionThreadMessageRepositoryShape["getByMessageId"] = (input) =>
155 getProjectionThreadMessageRow(input).pipe(207 getProjectionThreadMessageRow(input).pipe(
156 Effect.mapError(208 Effect.mapError(
...@@ -176,6 +228,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -176,6 +228,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
176228
177 return {229 return {
178 upsert,230 upsert,
231 appendStreaming,
179 getByMessageId,232 getByMessageId,
180 listByThreadId,233 listByThreadId,
181 deleteByThreadId,234 deleteByThreadId,
apps/server/src/persistence/Services/ProjectionThreadMessages.ts+12
...@@ -16,6 +16,7 @@ import {...@@ -16,6 +16,7 @@ import {
16} from "@t3tools/contracts";16} from "@t3tools/contracts";
17import * as Schema from "effect/Schema";17import * as Schema from "effect/Schema";
18import * as Context from "effect/Context";18import * as Context from "effect/Context";
19import * as Struct from "effect/Struct";
19import type * as Option from "effect/Option";20import type * as Option from "effect/Option";
20import type * as Effect from "effect/Effect";21import type * as Effect from "effect/Effect";
2122
...@@ -34,6 +35,12 @@ export const ProjectionThreadMessage = Schema.Struct({...@@ -34,6 +35,12 @@ export const ProjectionThreadMessage = Schema.Struct({
34});35});
35export type ProjectionThreadMessage = typeof ProjectionThreadMessage.Type;36export type ProjectionThreadMessage = typeof ProjectionThreadMessage.Type;
3637
38export const AppendStreamingProjectionThreadMessage = Schema.Struct(
39 Struct.omit(ProjectionThreadMessage.fields, ["isStreaming"]),
40);
41export type AppendStreamingProjectionThreadMessage =
42 typeof AppendStreamingProjectionThreadMessage.Type;
43
37export const ListProjectionThreadMessagesInput = Schema.Struct({44export const ListProjectionThreadMessagesInput = Schema.Struct({
38 threadId: ThreadId,45 threadId: ThreadId,
39});46});
...@@ -62,6 +69,11 @@ export interface ProjectionThreadMessageRepositoryShape {...@@ -62,6 +69,11 @@ export interface ProjectionThreadMessageRepositoryShape {
62 message: ProjectionThreadMessage,69 message: ProjectionThreadMessage,
63 ) => Effect.Effect<void, ProjectionRepositoryError>;70 ) => Effect.Effect<void, ProjectionRepositoryError>;
6471
72 /** Insert a streaming message or append text to its existing row. */
73 readonly appendStreaming: (
74 message: AppendStreamingProjectionThreadMessage,
75 ) => Effect.Effect<void, ProjectionRepositoryError>;
76
65 /**77 /**
66 * Read a projected thread message by id.78 * Read a projected thread message by id.
67 */79 */