fix(chat): keep the chat log's send path free of database reads

Author: waleedlatif1

apps/sim/lib/mothership/chat/chat-log.ts

@@ -10,19 +10,31 @@ const logger = createLogger('ChatLog')
 
 const CHAT_LOG_TIMEOUT_MS = 5_000
 
-/** The turn a Chat send admitted — rebuilt identically when a relay pod recovers it. */
+/** The turn a Chat send admitted — rebuilt when a relay pod recovers it. */
 export interface ChatTurnLogContext {
   chatId: string
   messageId: string
   requestId: string
   userId: string
+  userEmail?: string
   userMessage: string
   mode: 'assistant' | 'agent' | 'plan'
   startedAt: number
 }
 
 export type ChatTurnStatus = 'success' | 'error' | 'aborted'
 
+/** The user's email for a recovered turn, whose send-time session is gone; skipped when the log is off. */
+export async function readChatLogEmail(userId: string): Promise<string | undefined> {
+  if (!env.SIM_LOGGING_WORKFLOW_URL) return undefined
+  const [row] = await db
+    .select({ email: user.email })
+    .from(user)
+    .where(eq(user.id, userId))
+    .limit(1)
+  return row?.email
+}
+
 /**
  * Operator funnel: posts each finished Chat turn to the Sim workflow at
  * `SIM_LOGGING_WORKFLOW_URL` (off when unset). The body keeps the shape the Go
@@ -53,12 +65,6 @@ async function sendChatTurn(
   status: ChatTurnStatus,
   durationMs: number
 ): Promise<void> {
-  const [row] = await db
-    .select({ email: user.email })
-    .from(user)
-    .where(eq(user.id, context.userId))
-    .limit(1)
-
   const headers: Record<string, string> = { 'Content-Type': 'application/json' }
   if (env.SIM_LOGGING_WORKFLOW_API_KEY) headers['X-API-Key'] = env.SIM_LOGGING_WORKFLOW_API_KEY
 
@@ -69,7 +75,7 @@ async function sendChatTurn(
     chatId: context.chatId,
     messageId: context.messageId,
     userId: context.userId,
-    userEmail: row?.email,
+    userEmail: context.userEmail,
     userMessage: context.userMessage,
     assistantResponse: result.content,
     status,

apps/sim/lib/mothership/chat/post.ts

@@ -1495,6 +1495,7 @@ export async function handleUnifiedChatPost(req: NextRequest) {
               messageId: userMessageId,
               requestId,
               userId: authenticatedUserId,
+              ...(authenticatedUserEmail ? { userEmail: authenticatedUserEmail } : {}),
               userMessage: body.message,
               mode: requestMode,
               startedAt: Date.now(),

apps/sim/lib/mothership/request/application/recover-stream.ts

@@ -13,7 +13,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { getLatestRunForStream } from '@/lib/mothership/async-runs/repository'
 import { defineAuthorizedChatUseCase } from '@/lib/mothership/chat/application/authorized-chat-use-case'
 import { resolveOwnedChatContext } from '@/lib/mothership/chat/application/context'
-import type { ChatTurnLogContext } from '@/lib/mothership/chat/chat-log'
+import { type ChatTurnLogContext, readChatLogEmail } from '@/lib/mothership/chat/chat-log'
 import { buildOnComplete, buildOnError } from '@/lib/mothership/chat/completion'
 import { restoreBillingAdmission } from '@/lib/mothership/request/lifecycle/admission'
 import { claimRunController } from '@/lib/mothership/request/lifecycle/controller-ownership'
@@ -108,7 +108,8 @@ export const readChatStream = defineAuthorizedChatUseCase({
             organizationId,
           })
         : undefined
-      const [events, billingAttribution, userPermission] = await Promise.all([
+      const logsChatTurn = config.data.goRoute === '/api/mothership'
+      const [events, billingAttribution, userPermission, userEmail] = await Promise.all([
         readEvents(run.streamId, '0'),
         restoredAdmission
           ? Promise.resolve(restoredAdmission.attribution)
@@ -118,6 +119,7 @@ export const readChatStream = defineAuthorizedChatUseCase({
         workspaceId
           ? getUserEntityPermissions(userId, 'workspace', workspaceId)
           : Promise.resolve(undefined),
+        logsChatTurn ? readChatLogEmail(userId) : Promise.resolve(undefined),
       ])
       if (workspaceId && !userPermission)
         throw new OrchestrationError('forbidden', 'Workspace access revoked')
@@ -134,13 +136,14 @@ export const readChatStream = defineAuthorizedChatUseCase({
         requestMode,
         notifyWorkspaceStatus: true,
         runController: { id: run.id, token: lease.value },
-        ...(config.data.goRoute === '/api/mothership'
+        ...(logsChatTurn
           ? {
               chatLog: {
                 chatId,
                 messageId: run.streamId,
                 requestId,
                 userId,
+                ...(userEmail ? { userEmail } : {}),
                 userMessage: intent.message,
                 mode: requestMode,
                 startedAt: run.startedAt.getTime(),