🚸 (openai) Parse stream on client to correctly handle errors

packages/embeds/js/package.json

@@ -14,6 +14,7 @@
   "dependencies": {
     "@stripe/stripe-js": "1.53.0",
     "@udecode/plate-common": "^21.1.5",
+    "eventsource-parser": "^1.0.0",
     "solid-element": "1.7.0",
     "solid-js": "1.7.5"
   },

packages/embeds/js/src/components/ConversationContainer/ConversationContainer.tsx

@@ -1,4 +1,4 @@
-import { ChatReply, Theme } from '@typebot.io/schemas'
+import { ChatReply, SendMessageInput, Theme } from '@typebot.io/schemas'
 import { InputBlockType } from '@typebot.io/schemas/features/blocks/inputs/enums'
 import { createEffect, createSignal, For, onMount, Show } from 'solid-js'
 import { sendMessageQuery } from '@/queries/sendMessageQuery'
@@ -79,7 +79,7 @@ export const ConversationContainer = (props: Props) => {
             sessionId: props.initialChatReply.sessionId,
           })
           if (response && 'replyToSend' in response) {
-            sendMessage(response.replyToSend)
+            sendMessage(response.replyToSend, response.logs)
             return
           }
           if (response && 'blockedPopupUrl' in response)
@@ -95,7 +95,11 @@ export const ConversationContainer = (props: Props) => {
     )
   })
 
-  const sendMessage = async (message: string | undefined) => {
+  const sendMessage = async (
+    message: string | undefined,
+    clientLogs?: SendMessageInput['clientLogs']
+  ) => {
+    if (clientLogs) props.onNewLogs?.(clientLogs)
     setHasError(false)
     const currentInputBlock = [...chatChunks()].pop()?.input
     if (currentInputBlock?.id && props.onAnswer && message)
@@ -114,6 +118,7 @@ export const ConversationContainer = (props: Props) => {
       apiHost: props.context.apiHost,
       sessionId: props.initialChatReply.sessionId,
       message,
+      clientLogs,
     })
     clearTimeout(longRequest)
     setIsSending(false)
@@ -151,7 +156,7 @@ export const ConversationContainer = (props: Props) => {
           sessionId: props.initialChatReply.sessionId,
         })
         if (response && 'replyToSend' in response) {
-          sendMessage(response.replyToSend)
+          sendMessage(response.replyToSend, response.logs)
           return
         }
         if (response && 'blockedPopupUrl' in response)
@@ -200,7 +205,7 @@ export const ConversationContainer = (props: Props) => {
           sessionId: props.initialChatReply.sessionId,
         })
         if (response && 'replyToSend' in response) {
-          sendMessage(response.replyToSend)
+          sendMessage(response.replyToSend, response.logs)
           return
         }
         if (response && 'blockedPopupUrl' in response)

packages/embeds/js/src/features/blocks/integrations/openai/streamChat.ts

@@ -1,5 +1,10 @@
 import { getOpenAiStreamerQuery } from '@/queries/getOpenAiStreamerQuery'
 import { ClientSideActionContext } from '@/types'
+import {
+  createParser,
+  ParsedEvent,
+  ReconnectInterval,
+} from 'eventsource-parser'
 
 export const streamChat =
   (context: ClientSideActionContext) =>
@@ -8,25 +13,59 @@ export const streamChat =
       content?: string | undefined
       role?: 'system' | 'user' | 'assistant' | undefined
     }[],
-    { onStreamedMessage }: { onStreamedMessage?: (message: string) => void }
-  ) => {
+    {
+      onStreamedMessage,
+      isRetrying,
+    }: { onStreamedMessage?: (message: string) => void; isRetrying?: boolean }
+  ): Promise<{ message?: string; error?: object }> => {
     const data = await getOpenAiStreamerQuery(context)(messages)
 
-    if (!data) {
-      return
-    }
+    if (!data) return { error: { message: "Couldn't get streamer data" } }
+
+    let message = ''
 
     const reader = data.getReader()
     const decoder = new TextDecoder()
-    let done = false
 
-    let message = ''
-    while (!done) {
-      const { value, done: doneReading } = await reader.read()
-      done = doneReading
-      const chunkValue = decoder.decode(value)
-      message += chunkValue
-      onStreamedMessage?.(message)
+    const onParse = (event: ParsedEvent | ReconnectInterval) => {
+      if (event.type === 'event') {
+        const data = event.data
+        try {
+          const json = JSON.parse(data) as {
+            choices: { delta: { content: string } }[]
+          }
+          const text = json.choices.at(0)?.delta.content
+          if (!text) return
+          message += text
+          onStreamedMessage?.(message)
+        } catch (e) {
+          console.error(e)
+        }
+      }
     }
-    return message
+
+    const parser = createParser(onParse)
+
+    // eslint-disable-next-line no-constant-condition
+    while (true) {
+      const { value, done } = await reader.read()
+      if (done || !value) break
+      const dataString = decoder.decode(value)
+      if (dataString.includes('503 Service Temporarily Unavailable')) {
+        if (isRetrying)
+          return { error: { message: "Couldn't get streamer data" } }
+        await new Promise((resolve) => setTimeout(resolve, 1000))
+        return streamChat(context)(messages, {
+          onStreamedMessage,
+          isRetrying: true,
+        })
+      }
+      if (dataString.includes('[DONE]')) break
+      if (dataString.includes('"error":')) {
+        return { error: JSON.parse(dataString).error }
+      }
+      parser.feed(dataString)
+    }
+
+    return { message }
   }

packages/embeds/js/src/utils/executeClientSideActions.ts

@@ -7,14 +7,16 @@ import { executeSetVariable } from '@/features/blocks/logic/setVariable/executeS
 import { executeWait } from '@/features/blocks/logic/wait/utils/executeWait'
 import { executeWebhook } from '@/features/blocks/integrations/webhook/executeWebhook'
 import { ClientSideActionContext } from '@/types'
-import type { ChatReply } from '@typebot.io/schemas'
+import type { ChatReply, ReplyLog } from '@typebot.io/schemas'
 
 export const executeClientSideAction = async (
   clientSideAction: NonNullable<ChatReply['clientSideActions']>[0],
   context: ClientSideActionContext,
   onStreamedMessage?: (message: string) => void
 ): Promise<
-  { blockedPopupUrl: string } | { replyToSend: string | undefined } | void
+  | { blockedPopupUrl: string }
+  | { replyToSend: string | undefined; logs?: ReplyLog[] }
+  | void
 > => {
   if ('chatwoot' in clientSideAction) {
     return executeChatwoot(clientSideAction.chatwoot)
@@ -35,11 +37,22 @@ export const executeClientSideAction = async (
     return executeSetVariable(clientSideAction.setVariable.scriptToExecute)
   }
   if ('streamOpenAiChatCompletion' in clientSideAction) {
-    const text = await streamChat(context)(
+    const { error, message } = await streamChat(context)(
       clientSideAction.streamOpenAiChatCompletion.messages,
       { onStreamedMessage }
     )
-    return { replyToSend: text }
+    if (error)
+      return {
+        replyToSend: undefined,
+        logs: [
+          {
+            status: 'error',
+            description: 'Failed to stream OpenAI completion',
+            details: JSON.stringify(error, null, 2),
+          },
+        ],
+      }
+    return { replyToSend: message }
   }
   if ('webhookToExecute' in clientSideAction) {
     const response = await executeWebhook(clientSideAction.webhookToExecute)