perf: optimize streaming message projection path

Berkay2002 · Berkay2002 · commit 2c1f460d5dbe · 2026-04-16T15:48:30.000+02:00
Cherry-picked from upstream pingdotgg#1688
diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts
@@ -758,37 +758,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
     )(function* (event, attachmentSideEffects) {
       switch (event.type) {
         case "thread.message-sent": {
-          const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({
-            messageId: event.payload.messageId,
-          });
-          const previousMessage = Option.getOrUndefined(existingMessage);
-          const nextText = Option.match(existingMessage, {
-            onNone: () => event.payload.text,
-            onSome: (message) => {
-              if (event.payload.streaming) {
-                return `${message.text}${event.payload.text}`;
-              }
-              if (event.payload.text.length === 0) {
-                return message.text;
-              }
-              return event.payload.text;
-            },
-          });
           const nextAttachments =
             event.payload.attachments !== undefined
               ? yield* materializeAttachmentsForProjection({
                   attachments: event.payload.attachments,
                 })
-              : previousMessage?.attachments;
+              : undefined;
+          if (event.payload.streaming) {
+            yield* projectionThreadMessageRepository.appendTextDelta({
+              messageId: event.payload.messageId,
+              threadId: event.payload.threadId,
+              turnId: event.payload.turnId,
+              role: event.payload.role,
+              delta: event.payload.text,
+              ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),
+              isStreaming: true,
+              createdAt: event.payload.createdAt,
+              updatedAt: event.payload.updatedAt,
+            });
+            return;
+          }
+
+          const existingMessage = yield* projectionThreadMessageRepository
+            .getByMessageId({
+              messageId: event.payload.messageId,
+            })
+            .pipe(Effect.map(Option.getOrUndefined));
+          const nextText =
+            existingMessage !== undefined && event.payload.text.length === 0
+              ? existingMessage.text
+              : event.payload.text;
+          const persistedAttachments = nextAttachments ?? existingMessage?.attachments;
           yield* projectionThreadMessageRepository.upsert({
             messageId: event.payload.messageId,
             threadId: event.payload.threadId,
             turnId: event.payload.turnId,
             role: event.payload.role,
             text: nextText,
-            ...(nextAttachments !== undefined ? { attachments: [...nextAttachments] } : {}),
+            ...(persistedAttachments !== undefined
+              ? { attachments: [...persistedAttachments] }
+              : {}),
             isStreaming: event.payload.streaming,
-            createdAt: previousMessage?.createdAt ?? event.payload.createdAt,
+            createdAt: existingMessage?.createdAt ?? event.payload.createdAt,
             updatedAt: event.payload.updatedAt,
           });
           return;
diff --git a/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts b/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts
@@ -1,6 +1,6 @@
 import { MessageId, ThreadId } from "@t3tools/contracts";
 import { assert, it } from "@effect/vitest";
-import { Effect, Layer } from "effect";
+import { Effect, Layer, Option } from "effect";
 
 import { ProjectionThreadMessageRepository } from "../Services/ProjectionThreadMessages.ts";
 import { ProjectionThreadMessageRepositoryLive } from "./ProjectionThreadMessages.ts";
@@ -110,4 +110,84 @@ layer("ProjectionThreadMessageRepository", (it) => {
       assert.deepEqual(rows[0]?.attachments, []);
     }),
   );
+
+  it.effect("looks up a projected message by id", () =>
+    Effect.gen(function* () {
+      const repository = yield* ProjectionThreadMessageRepository;
+      const threadId = ThreadId.makeUnsafe("thread-get-by-message-id");
+      const messageId = MessageId.makeUnsafe("message-get-by-message-id");
+      const createdAt = "2026-02-28T19:20:00.000Z";
+      const updatedAt = "2026-02-28T19:20:01.000Z";
+
+      yield* repository.upsert({
+        messageId,
+        threadId,
+        turnId: null,
+        role: "assistant",
+        text: "lookup me",
+        isStreaming: false,
+        createdAt,
+        updatedAt,
+      });
+
+      const maybeRow = yield* repository.getByMessageId({ messageId });
+      assert.isTrue(Option.isSome(maybeRow));
+      const row = Option.getOrThrow(maybeRow);
+      assert.equal(row.messageId, messageId);
+      assert.equal(row.threadId, threadId);
+      assert.equal(row.text, "lookup me");
+      assert.equal(row.createdAt, createdAt);
+      assert.equal(row.updatedAt, updatedAt);
+    }),
+  );
+
+  it.effect("appends streaming deltas without losing createdAt or attachments", () =>
+    Effect.gen(function* () {
+      const repository = yield* ProjectionThreadMessageRepository;
+      const threadId = ThreadId.makeUnsafe("thread-append-delta");
+      const messageId = MessageId.makeUnsafe("message-append-delta");
+      const createdAt = "2026-02-28T19:30:00.000Z";
+      const persistedAttachments = [
+        {
+          type: "image" as const,
+          id: "thread-append-delta-att-1",
+          name: "example.png",
+          mimeType: "image/png",
+          sizeBytes: 5,
+        },
+      ];
+
+      yield* repository.appendTextDelta({
+        messageId,
+        threadId,
+        turnId: null,
+        role: "assistant",
+        delta: "Hello",
+        attachments: persistedAttachments,
+        isStreaming: true,
+        createdAt,
+        updatedAt: "2026-02-28T19:30:01.000Z",
+      });
+
+      yield* repository.appendTextDelta({
+        messageId,
+        threadId,
+        turnId: null,
+        role: "assistant",
+        delta: " world",
+        isStreaming: true,
+        createdAt: "2026-02-28T19:30:05.000Z",
+        updatedAt: "2026-02-28T19:30:06.000Z",
+      });
+
+      const maybeRow = yield* repository.getByMessageId({ messageId });
+      assert.isTrue(Option.isSome(maybeRow));
+      const row = Option.getOrThrow(maybeRow);
+      assert.equal(row.text, "Hello world");
+      assert.equal(row.createdAt, createdAt);
+      assert.equal(row.updatedAt, "2026-02-28T19:30:06.000Z");
+      assert.deepEqual(row.attachments, persistedAttachments);
+      assert.isTrue(row.isStreaming);
+    }),
+  );
 });
diff --git a/apps/server/src/persistence/Layers/ProjectionThreadMessages.ts b/apps/server/src/persistence/Layers/ProjectionThreadMessages.ts
@@ -5,6 +5,7 @@ import { ChatAttachment } from "@t3tools/contracts";
 
 import { toPersistenceSqlError } from "../Errors.ts";
 import {
+  AppendProjectionThreadMessageDeltaInput,
   GetProjectionThreadMessageInput,
   ProjectionThreadMessageRepository,
   type ProjectionThreadMessageRepositoryShape,
@@ -133,6 +134,50 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
       `,
   });
 
+  const appendProjectionThreadMessageDeltaRow = SqlSchema.void({
+    Request: AppendProjectionThreadMessageDeltaInput,
+    execute: (row) => {
+      const nextAttachmentsJson =
+        row.attachments !== undefined ? JSON.stringify(row.attachments) : null;
+      return sql`
+        INSERT INTO projection_thread_messages (
+          message_id,
+          thread_id,
+          turn_id,
+          role,
+          text,
+          attachments_json,
+          is_streaming,
+          created_at,
+          updated_at
+        )
+        VALUES (
+          ${row.messageId},
+          ${row.threadId},
+          ${row.turnId},
+          ${row.role},
+          ${row.delta},
+          ${nextAttachmentsJson},
+          ${row.isStreaming ? 1 : 0},
+          ${row.createdAt},
+          ${row.updatedAt}
+        )
+        ON CONFLICT (message_id)
+        DO UPDATE SET
+          thread_id = excluded.thread_id,
+          turn_id = COALESCE(excluded.turn_id, projection_thread_messages.turn_id),
+          role = excluded.role,
+          text = projection_thread_messages.text || excluded.text,
+          attachments_json = COALESCE(
+            excluded.attachments_json,
+            projection_thread_messages.attachments_json
+          ),
+          is_streaming = excluded.is_streaming,
+          updated_at = excluded.updated_at
+      `;
+    },
+  });
+
   const deleteProjectionThreadMessageRows = SqlSchema.void({
     Request: DeleteProjectionThreadMessagesInput,
     execute: ({ threadId }) =>
@@ -163,6 +208,13 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
       Effect.map((rows) => rows.map(toProjectionThreadMessage)),
     );
 
+  const appendTextDelta: ProjectionThreadMessageRepositoryShape["appendTextDelta"] = (input) =>
+    appendProjectionThreadMessageDeltaRow(input).pipe(
+      Effect.mapError(
+        toPersistenceSqlError("ProjectionThreadMessageRepository.appendTextDelta:query"),
+      ),
+    );
+
   const deleteByThreadId: ProjectionThreadMessageRepositoryShape["deleteByThreadId"] = (input) =>
     deleteProjectionThreadMessageRows(input).pipe(
       Effect.mapError(
@@ -174,6 +226,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
     upsert,
     getByMessageId,
     listByThreadId,
+    appendTextDelta,
     deleteByThreadId,
   } satisfies ProjectionThreadMessageRepositoryShape;
 });
diff --git a/apps/server/src/persistence/Services/ProjectionThreadMessages.ts b/apps/server/src/persistence/Services/ProjectionThreadMessages.ts
@@ -14,8 +14,7 @@ import {
   TurnId,
   IsoDateTime,
 } from "@t3tools/contracts";
-import { Schema, Context } from "effect";
-import type { Option } from "effect";
+import { Option, Schema, Context } from "effect";
 import type { Effect } from "effect";
 
 import type { ProjectionRepositoryError } from "../Errors.ts";
@@ -48,6 +47,20 @@ export const DeleteProjectionThreadMessagesInput = Schema.Struct({
 });
 export type DeleteProjectionThreadMessagesInput = typeof DeleteProjectionThreadMessagesInput.Type;
 
+export const AppendProjectionThreadMessageDeltaInput = Schema.Struct({
+  messageId: MessageId,
+  threadId: ThreadId,
+  turnId: Schema.NullOr(TurnId),
+  role: OrchestrationMessageRole,
+  delta: Schema.String,
+  attachments: Schema.optional(Schema.Array(ChatAttachment)),
+  isStreaming: Schema.Boolean,
+  createdAt: IsoDateTime,
+  updatedAt: IsoDateTime,
+});
+export type AppendProjectionThreadMessageDeltaInput =
+  typeof AppendProjectionThreadMessageDeltaInput.Type;
+
 /**
  * ProjectionThreadMessageRepositoryShape - Service API for projected thread messages.
  */
@@ -77,6 +90,13 @@ export interface ProjectionThreadMessageRepositoryShape {
     input: ListProjectionThreadMessagesInput,
   ) => Effect.Effect<ReadonlyArray<ProjectionThreadMessage>, ProjectionRepositoryError>;
 
+  /**
+   * Append a streaming text delta to an existing projected message row, or insert it.
+   */
+  readonly appendTextDelta: (
+    input: AppendProjectionThreadMessageDeltaInput,
+  ) => Effect.Effect<void, ProjectionRepositoryError>;
+
   /**
    * Delete projected thread messages by thread.
    */
