authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-01 14:48:39-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-01 14:48:39-07:00
log4963fccb05c9ae46fc5153bc76008d7c8e21ba0a
tree116cb8293186d7b32fe9c825b4f77470b0e0642f
parentdd89d5ce1a483f29e0d5429439a60da7a2296a3f
signature Signed by PGP key B5690EEEBB952194

perf(server): avoid thread-wide scans for message projection (#1647)

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

4 files changed, 87 insertions(+), 28 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.ts+17-13
...@@ -623,24 +623,28 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -623,24 +623,28 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
623 )(function* (event, attachmentSideEffects) {623 )(function* (event, attachmentSideEffects) {
624 switch (event.type) {624 switch (event.type) {
625 case "thread.message-sent": {625 case "thread.message-sent": {
626 const existingRows = yield* projectionThreadMessageRepository.listByThreadId({626 const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({
627 threadId: event.payload.threadId,627 messageId: event.payload.messageId,
628 });
629 const previousMessage = Option.getOrUndefined(existingMessage);
630 const nextText = Option.match(existingMessage, {
631 onNone: () => event.payload.text,
632 onSome: (message) => {
633 if (event.payload.streaming) {
634 return `${message.text}${event.payload.text}`;
635 }
636 if (event.payload.text.length === 0) {
637 return message.text;
638 }
639 return event.payload.text;
640 },
628 });641 });
629 const existingMessage = existingRows.find(
630 (row) => row.messageId === event.payload.messageId,
631 );
632 const nextText =
633 existingMessage && event.payload.streaming
634 ? `${existingMessage.text}${event.payload.text}`
635 : existingMessage && event.payload.text.length === 0
636 ? existingMessage.text
637 : event.payload.text;
638 const nextAttachments =642 const nextAttachments =
639 event.payload.attachments !== undefined643 event.payload.attachments !== undefined
640 ? yield* materializeAttachmentsForProjection({644 ? yield* materializeAttachmentsForProjection({
641 attachments: event.payload.attachments,645 attachments: event.payload.attachments,
642 })646 })
643 : existingMessage?.attachments;647 : previousMessage?.attachments;
644 yield* projectionThreadMessageRepository.upsert({648 yield* projectionThreadMessageRepository.upsert({
645 messageId: event.payload.messageId,649 messageId: event.payload.messageId,
646 threadId: event.payload.threadId,650 threadId: event.payload.threadId,
...@@ -649,7 +653,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -649,7 +653,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
649 text: nextText,653 text: nextText,
650 ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),654 ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),
651 isStreaming: event.payload.streaming,655 isStreaming: event.payload.streaming,
652 createdAt: existingMessage?.createdAt ?? event.payload.createdAt,656 createdAt: previousMessage?.createdAt ?? event.payload.createdAt,
653 updatedAt: event.payload.updatedAt,657 updatedAt: event.payload.updatedAt,
654 });658 });
655 return;659 return;
apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts+7
...@@ -55,6 +55,13 @@ layer("ProjectionThreadMessageRepository", (it) => {...@@ -55,6 +55,13 @@ layer("ProjectionThreadMessageRepository", (it) => {
55 assert.equal(rows.length, 1);55 assert.equal(rows.length, 1);
56 assert.equal(rows[0]?.text, "updated");56 assert.equal(rows[0]?.text, "updated");
57 assert.deepEqual(rows[0]?.attachments, persistedAttachments);57 assert.deepEqual(rows[0]?.attachments, persistedAttachments);
58
59 const rowById = yield* repository.getByMessageId({ messageId });
60 assert.equal(rowById._tag, "Some");
61 if (rowById._tag === "Some") {
62 assert.equal(rowById.value.text, "updated");
63 assert.deepEqual(rowById.value.attachments, persistedAttachments);
64 }
58 }),65 }),
59 );66 );
6067
apps/server/src/persistence/Layers/ProjectionThreadMessages.ts+49-14
...@@ -1,10 +1,11 @@...@@ -1,10 +1,11 @@
1import * as SqlClient from "effect/unstable/sql/SqlClient";1import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as SqlSchema from "effect/unstable/sql/SqlSchema";2import * as SqlSchema from "effect/unstable/sql/SqlSchema";
3import { Effect, Layer, Schema, Struct } from "effect";3import { Effect, Layer, Option, Schema, Struct } from "effect";
4import { ChatAttachment } from "@t3tools/contracts";4import { ChatAttachment } from "@t3tools/contracts";
55
6import { toPersistenceSqlError } from "../Errors.ts";6import { toPersistenceSqlError } from "../Errors.ts";
7import {7import {
8 GetProjectionThreadMessageInput,
8 ProjectionThreadMessageRepository,9 ProjectionThreadMessageRepository,
9 type ProjectionThreadMessageRepositoryShape,10 type ProjectionThreadMessageRepositoryShape,
10 DeleteProjectionThreadMessagesInput,11 DeleteProjectionThreadMessagesInput,
...@@ -19,6 +20,22 @@ const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields(...@@ -19,6 +20,22 @@ const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields(
19 }),20 }),
20);21);
2122
23function toProjectionThreadMessage(
24 row: Schema.Schema.Type<typeof ProjectionThreadMessageDbRowSchema>,
25): ProjectionThreadMessage {
26 return {
27 messageId: row.messageId,
28 threadId: row.threadId,
29 turnId: row.turnId,
30 role: row.role,
31 text: row.text,
32 isStreaming: row.isStreaming === 1,
33 createdAt: row.createdAt,
34 updatedAt: row.updatedAt,
35 ...(row.attachments !== null ? { attachments: row.attachments } : {}),
36 };
37}
38
22const makeProjectionThreadMessageRepository = Effect.gen(function* () {39const makeProjectionThreadMessageRepository = Effect.gen(function* () {
23 const sql = yield* SqlClient.SqlClient;40 const sql = yield* SqlClient.SqlClient;
2441
...@@ -74,6 +91,27 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -74,6 +91,27 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
74 },91 },
75 });92 });
7693
94 const getProjectionThreadMessageRow = SqlSchema.findOneOption({
95 Request: GetProjectionThreadMessageInput,
96 Result: ProjectionThreadMessageDbRowSchema,
97 execute: ({ messageId }) =>
98 sql`
99 SELECT
100 message_id AS "messageId",
101 thread_id AS "threadId",
102 turn_id AS "turnId",
103 role,
104 text,
105 attachments_json AS "attachments",
106 is_streaming AS "isStreaming",
107 created_at AS "createdAt",
108 updated_at AS "updatedAt"
109 FROM projection_thread_messages
110 WHERE message_id = ${messageId}
111 LIMIT 1
112 `,
113 });
114
77 const listProjectionThreadMessageRows = SqlSchema.findAll({115 const listProjectionThreadMessageRows = SqlSchema.findAll({
78 Request: ListProjectionThreadMessagesInput,116 Request: ListProjectionThreadMessagesInput,
79 Result: ProjectionThreadMessageDbRowSchema,117 Result: ProjectionThreadMessageDbRowSchema,
...@@ -109,24 +147,20 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -109,24 +147,20 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
109 Effect.mapError(toPersistenceSqlError("ProjectionThreadMessageRepository.upsert:query")),147 Effect.mapError(toPersistenceSqlError("ProjectionThreadMessageRepository.upsert:query")),
110 );148 );
111149
150 const getByMessageId: ProjectionThreadMessageRepositoryShape["getByMessageId"] = (input) =>
151 getProjectionThreadMessageRow(input).pipe(
152 Effect.mapError(
153 toPersistenceSqlError("ProjectionThreadMessageRepository.getByMessageId:query"),
154 ),
155 Effect.map(Option.map(toProjectionThreadMessage)),
156 );
157
112 const listByThreadId: ProjectionThreadMessageRepositoryShape["listByThreadId"] = (input) =>158 const listByThreadId: ProjectionThreadMessageRepositoryShape["listByThreadId"] = (input) =>
113 listProjectionThreadMessageRows(input).pipe(159 listProjectionThreadMessageRows(input).pipe(
114 Effect.mapError(160 Effect.mapError(
115 toPersistenceSqlError("ProjectionThreadMessageRepository.listByThreadId:query"),161 toPersistenceSqlError("ProjectionThreadMessageRepository.listByThreadId:query"),
116 ),162 ),
117 Effect.map((rows) =>163 Effect.map((rows) => rows.map(toProjectionThreadMessage)),
118 rows.map((row) => ({
119 messageId: row.messageId,
120 threadId: row.threadId,
121 turnId: row.turnId,
122 role: row.role,
123 text: row.text,
124 isStreaming: row.isStreaming === 1,
125 createdAt: row.createdAt,
126 updatedAt: row.updatedAt,
127 ...(row.attachments !== null ? { attachments: row.attachments } : {}),
128 })),
129 ),
130 );164 );
131165
132 const deleteByThreadId: ProjectionThreadMessageRepositoryShape["deleteByThreadId"] = (input) =>166 const deleteByThreadId: ProjectionThreadMessageRepositoryShape["deleteByThreadId"] = (input) =>
...@@ -138,6 +172,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -138,6 +172,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
138172
139 return {173 return {
140 upsert,174 upsert,
175 getByMessageId,
141 listByThreadId,176 listByThreadId,
142 deleteByThreadId,177 deleteByThreadId,
143 } satisfies ProjectionThreadMessageRepositoryShape;178 } satisfies ProjectionThreadMessageRepositoryShape;
apps/server/src/persistence/Services/ProjectionThreadMessages.ts+14-1
...@@ -8,13 +8,14 @@...@@ -8,13 +8,14 @@
8 */8 */
9import {9import {
10 ChatAttachment,10 ChatAttachment,
11 OrchestrationMessageRole,
12 MessageId,11 MessageId,
12 OrchestrationMessageRole,
13 ThreadId,13 ThreadId,
14 TurnId,14 TurnId,
15 IsoDateTime,15 IsoDateTime,
16} from "@t3tools/contracts";16} from "@t3tools/contracts";
17import { Schema, ServiceMap } from "effect";17import { Schema, ServiceMap } from "effect";
18import type { Option } from "effect";
18import type { Effect } from "effect";19import type { Effect } from "effect";
1920
20import type { ProjectionRepositoryError } from "../Errors.ts";21import type { ProjectionRepositoryError } from "../Errors.ts";
...@@ -37,6 +38,11 @@ export const ListProjectionThreadMessagesInput = Schema.Struct({...@@ -37,6 +38,11 @@ export const ListProjectionThreadMessagesInput = Schema.Struct({
37});38});
38export type ListProjectionThreadMessagesInput = typeof ListProjectionThreadMessagesInput.Type;39export type ListProjectionThreadMessagesInput = typeof ListProjectionThreadMessagesInput.Type;
3940
41export const GetProjectionThreadMessageInput = Schema.Struct({
42 messageId: MessageId,
43});
44export type GetProjectionThreadMessageInput = typeof GetProjectionThreadMessageInput.Type;
45
40export const DeleteProjectionThreadMessagesInput = Schema.Struct({46export const DeleteProjectionThreadMessagesInput = Schema.Struct({
41 threadId: ThreadId,47 threadId: ThreadId,
42});48});
...@@ -55,6 +61,13 @@ export interface ProjectionThreadMessageRepositoryShape {...@@ -55,6 +61,13 @@ export interface ProjectionThreadMessageRepositoryShape {
55 message: ProjectionThreadMessage,61 message: ProjectionThreadMessage,
56 ) => Effect.Effect<void, ProjectionRepositoryError>;62 ) => Effect.Effect<void, ProjectionRepositoryError>;
5763
64 /**
65 * Read a projected thread message by id.
66 */
67 readonly getByMessageId: (
68 input: GetProjectionThreadMessageInput,
69 ) => Effect.Effect<Option.Option<ProjectionThreadMessage>, ProjectionRepositoryError>;
70
58 /**71 /**
59 * List projected thread messages for a thread.72 * List projected thread messages for a thread.
60 *73 *