⚡ Introduce a new high-performing standalone chat API (#1200)

Closes #1154

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
	- Added authentication functionality for user sessions in chat API.
- Introduced chat-related API endpoints for starting, previewing, and
continuing chat sessions, and streaming messages.
- Implemented WhatsApp API webhook handling for receiving and processing
messages.
- Added environment variable `NEXT_PUBLIC_CHAT_API_URL` for chat API URL
configuration.

- **Bug Fixes**
	- Adjusted file upload logic to correctly determine the API host.
	- Fixed message streaming URL in chat integration with OpenAI.

- **Documentation**
- Updated guides for creating blocks, local installation, self-hosting,
and deployment to use `bun` instead of `pnpm`.

- **Refactor**
	- Refactored chat API functionalities to use modular architecture.
- Simplified client log saving and session update functionalities by
using external functions.
	- Transitioned package management and workflow commands to use `bun`.

- **Chores**
- Switched to `bun` for package management in Dockerfiles and GitHub
workflows.
	- Added new Dockerfile for chat API service setup with Bun framework.
	- Updated `.prettierignore` and documentation with new commands.

- **Style**
	- No visible changes to end-users.

- **Tests**
	- No visible changes to end-users.

- **Revert**
	- No reverts in this release.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/bot-engine/apiHandlers/continueChat.ts

@@ -0,0 +1,102 @@
+import { TRPCError } from '@trpc/server'
+import { isDefined, isNotDefined } from '@typebot.io/lib/utils'
+import { getSession } from '../queries/getSession'
+import { continueBotFlow } from '../continueBotFlow'
+import { filterPotentiallySensitiveLogs } from '../logs/filterPotentiallySensitiveLogs'
+import { parseDynamicTheme } from '../parseDynamicTheme'
+import { saveStateToDatabase } from '../saveStateToDatabase'
+import { computeCurrentProgress } from '../computeCurrentProgress'
+
+type Props = {
+  origin: string | undefined
+  message?: string
+  sessionId: string
+}
+export const continueChat = async ({ origin, sessionId, message }: Props) => {
+  const session = await getSession(sessionId)
+
+  if (!session) {
+    throw new TRPCError({
+      code: 'NOT_FOUND',
+      message: 'Session not found.',
+    })
+  }
+
+  const isSessionExpired =
+    session &&
+    isDefined(session.state.expiryTimeout) &&
+    session.updatedAt.getTime() + session.state.expiryTimeout < Date.now()
+
+  if (isSessionExpired)
+    throw new TRPCError({
+      code: 'NOT_FOUND',
+      message: 'Session expired. You need to start a new session.',
+    })
+
+  let corsOrigin
+
+  if (
+    session?.state.allowedOrigins &&
+    session.state.allowedOrigins.length > 0
+  ) {
+    if (origin && session.state.allowedOrigins.includes(origin))
+      corsOrigin = origin
+    else corsOrigin = session.state.allowedOrigins[0]
+  }
+
+  const {
+    messages,
+    input,
+    clientSideActions,
+    newSessionState,
+    logs,
+    lastMessageNewFormat,
+    visitedEdges,
+  } = await continueBotFlow(message, {
+    version: 2,
+    state: session.state,
+    startTime: Date.now(),
+  })
+
+  if (newSessionState)
+    await saveStateToDatabase({
+      session: {
+        id: session.id,
+        state: newSessionState,
+      },
+      input,
+      logs,
+      clientSideActions,
+      visitedEdges,
+      hasCustomEmbedBubble: messages.some(
+        (message) => message.type === 'custom-embed'
+      ),
+    })
+
+  const isPreview = isNotDefined(session.state.typebotsQueue[0].resultId)
+
+  const isEnded =
+    newSessionState.progressMetadata &&
+    !input?.id &&
+    (clientSideActions?.filter((c) => c.expectsDedicatedReply).length ?? 0) ===
+      0
+
+  return {
+    messages,
+    input,
+    clientSideActions,
+    dynamicTheme: parseDynamicTheme(newSessionState),
+    logs: isPreview ? logs : logs?.filter(filterPotentiallySensitiveLogs),
+    lastMessageNewFormat,
+    corsOrigin,
+    progress: newSessionState.progressMetadata
+      ? isEnded
+        ? 100
+        : computeCurrentProgress({
+            typebotsQueue: newSessionState.typebotsQueue,
+            progressMetadata: newSessionState.progressMetadata,
+            currentInputBlockId: input?.id,
+          })
+      : undefined,
+  }
+}

packages/bot-engine/apiHandlers/getMessageStream.ts

@@ -0,0 +1,130 @@
+import { IntegrationBlockType } from '@typebot.io/schemas/features/blocks/integrations/constants'
+import { ChatCompletionOpenAIOptions } from '@typebot.io/schemas/features/blocks/integrations/openai'
+import { OpenAI } from 'openai'
+import { decryptV2 } from '@typebot.io/lib/api/encryption/decryptV2'
+import { forgedBlocks } from '@typebot.io/forge-repository/definitions'
+import { ReadOnlyVariableStore } from '@typebot.io/forge'
+import {
+  ParseVariablesOptions,
+  parseVariables,
+} from '@typebot.io/variables/parseVariables'
+import { getOpenAIChatCompletionStream } from './legacy/getOpenAIChatCompletionStream'
+import { getCredentials } from '../queries/getCredentials'
+import { getSession } from '../queries/getSession'
+import { getBlockById } from '@typebot.io/schemas/helpers'
+import { isForgedBlockType } from '@typebot.io/schemas/features/blocks/forged/helpers'
+
+type Props = {
+  sessionId: string
+  messages: OpenAI.Chat.ChatCompletionMessage[] | undefined
+}
+
+export const getMessageStream = async ({ sessionId, messages }: Props) => {
+  const session = await getSession(sessionId)
+
+  if (!session?.state || !session.state.currentBlockId)
+    return { status: 404, message: 'Could not find session' }
+
+  const { group, block } = getBlockById(
+    session.state.currentBlockId,
+    session.state.typebotsQueue[0].typebot.groups
+  )
+  if (!block || !group)
+    return {
+      status: 404,
+      message: 'Could not find block or group',
+    }
+
+  if (!('options' in block))
+    return {
+      status: 400,
+      message: 'This block does not have options',
+    }
+
+  if (block.type === IntegrationBlockType.OPEN_AI && messages) {
+    try {
+      const stream = await getOpenAIChatCompletionStream(
+        session.state,
+        block.options as ChatCompletionOpenAIOptions,
+        messages
+      )
+      if (!stream)
+        return {
+          status: 500,
+          message: 'Could not create stream',
+        }
+
+      return { stream }
+    } catch (error) {
+      if (error instanceof OpenAI.APIError) {
+        const { message } = error
+        return {
+          status: 500,
+          message,
+        }
+      } else {
+        throw error
+      }
+    }
+  }
+  if (!isForgedBlockType(block.type))
+    return {
+      status: 400,
+      message: 'This block does not have a stream function',
+    }
+
+  const blockDef = forgedBlocks[block.type]
+  const action = blockDef?.actions.find((a) => a.name === block.options?.action)
+
+  if (!action || !action.run?.stream)
+    return {
+      status: 400,
+      message: 'This block does not have a stream function',
+    }
+
+  try {
+    if (!block.options.credentialsId)
+      return { status: 404, message: 'Could not find credentials' }
+    const credentials = await getCredentials(block.options.credentialsId)
+    if (!credentials)
+      return { status: 404, message: 'Could not find credentials' }
+    const decryptedCredentials = await decryptV2(
+      credentials.data,
+      credentials.iv
+    )
+    const variables: ReadOnlyVariableStore = {
+      list: () => session.state.typebotsQueue[0].typebot.variables,
+      get: (id: string) => {
+        const variable = session.state.typebotsQueue[0].typebot.variables.find(
+          (variable) => variable.id === id
+        )
+        return variable?.value
+      },
+      parse: (text: string, params?: ParseVariablesOptions) =>
+        parseVariables(
+          session.state.typebotsQueue[0].typebot.variables,
+          params
+        )(text),
+    }
+    const stream = await action.run.stream.run({
+      credentials: decryptedCredentials,
+      options: block.options,
+      variables,
+    })
+    if (!stream) return { status: 500, message: 'Could not create stream' }
+
+    return { stream }
+  } catch (error) {
+    if (error instanceof OpenAI.APIError) {
+      const { message } = error
+      return {
+        status: 500,
+        message,
+      }
+    }
+    return {
+      status: 500,
+      message: 'Could not create stream',
+    }
+  }
+}

packages/bot-engine/apiHandlers/legacy/getOpenAIChatCompletionStream.ts

@@ -0,0 +1,58 @@
+import { decryptV2 } from '@typebot.io/lib/api/encryption/decryptV2'
+import { isNotEmpty } from '@typebot.io/lib/utils'
+import {
+  ChatCompletionOpenAIOptions,
+  OpenAICredentials,
+} from '@typebot.io/schemas/features/blocks/integrations/openai'
+import { SessionState } from '@typebot.io/schemas/features/chat/sessionState'
+import { OpenAIStream } from 'ai'
+import { parseVariableNumber } from '@typebot.io/variables/parseVariableNumber'
+import { ClientOptions, OpenAI } from 'openai'
+import { defaultOpenAIOptions } from '@typebot.io/schemas/features/blocks/integrations/openai/constants'
+import { getCredentials } from '../../queries/getCredentials'
+
+export const getOpenAIChatCompletionStream = async (
+  state: SessionState,
+  options: ChatCompletionOpenAIOptions,
+  messages: OpenAI.Chat.ChatCompletionMessageParam[]
+) => {
+  if (!options.credentialsId) return
+  const credentials = await getCredentials(options.credentialsId)
+  if (!credentials) {
+    console.error('Could not find credentials in database')
+    return
+  }
+  const { apiKey } = (await decryptV2(
+    credentials.data,
+    credentials.iv
+  )) as OpenAICredentials['data']
+
+  const { typebot } = state.typebotsQueue[0]
+  const temperature = parseVariableNumber(typebot.variables)(
+    options.advancedSettings?.temperature
+  )
+
+  const config = {
+    apiKey,
+    baseURL: options.baseUrl,
+    defaultHeaders: {
+      'api-key': apiKey,
+    },
+    defaultQuery: isNotEmpty(options.apiVersion)
+      ? {
+          'api-version': options.apiVersion,
+        }
+      : undefined,
+  } satisfies ClientOptions
+
+  const openai = new OpenAI(config)
+
+  const response = await openai.chat.completions.create({
+    model: options.model ?? defaultOpenAIOptions.model,
+    temperature,
+    stream: true,
+    messages,
+  })
+
+  return OpenAIStream(response)
+}

packages/bot-engine/apiHandlers/receiveMessage.ts

@@ -0,0 +1,35 @@
+import { WhatsAppWebhookRequestBody } from '@typebot.io/schemas/features/whatsapp'
+import { isNotDefined } from '@typebot.io/lib'
+import { resumeWhatsAppFlow } from '../whatsapp/resumeWhatsAppFlow'
+
+type Props = {
+  entry: WhatsAppWebhookRequestBody['entry']
+  credentialsId: string
+  workspaceId: string
+}
+
+export const receiveMessage = async ({
+  entry,
+  credentialsId,
+  workspaceId,
+}: Props) => {
+  const receivedMessage = entry.at(0)?.changes.at(0)?.value.messages?.at(0)
+  if (isNotDefined(receivedMessage)) return { message: 'No message found' }
+  const contactName =
+    entry.at(0)?.changes.at(0)?.value?.contacts?.at(0)?.profile?.name ?? ''
+  const contactPhoneNumber =
+    entry.at(0)?.changes.at(0)?.value?.messages?.at(0)?.from ?? ''
+  const phoneNumberId = entry.at(0)?.changes.at(0)?.value
+    .metadata.phone_number_id
+  if (!phoneNumberId) return { message: 'No phone number id found' }
+  return resumeWhatsAppFlow({
+    receivedMessage,
+    sessionId: `wa-${phoneNumberId}-${receivedMessage.from}`,
+    credentialsId,
+    workspaceId,
+    contact: {
+      name: contactName,
+      phoneNumber: contactPhoneNumber,
+    },
+  })
+}

packages/bot-engine/apiHandlers/receiveMessagePreview.ts

@@ -0,0 +1,31 @@
+import { WhatsAppWebhookRequestBody } from '@typebot.io/schemas/features/whatsapp'
+import { isNotDefined } from '@typebot.io/lib'
+import { TRPCError } from '@trpc/server'
+import { env } from '@typebot.io/env'
+import { resumeWhatsAppFlow } from '../whatsapp/resumeWhatsAppFlow'
+
+type Props = {
+  entry: WhatsAppWebhookRequestBody['entry']
+}
+export const receiveMessagePreview = ({ entry }: Props) => {
+  if (!env.WHATSAPP_PREVIEW_FROM_PHONE_NUMBER_ID)
+    throw new TRPCError({
+      code: 'INTERNAL_SERVER_ERROR',
+      message: 'WHATSAPP_PREVIEW_FROM_PHONE_NUMBER_ID is not defined',
+    })
+  const receivedMessage = entry.at(0)?.changes.at(0)?.value.messages?.at(0)
+  if (isNotDefined(receivedMessage)) return { message: 'No message found' }
+  const contactName =
+    entry.at(0)?.changes.at(0)?.value?.contacts?.at(0)?.profile?.name ?? ''
+  const contactPhoneNumber =
+    entry.at(0)?.changes.at(0)?.value?.messages?.at(0)?.from ?? ''
+
+  return resumeWhatsAppFlow({
+    receivedMessage,
+    sessionId: `wa-preview-${receivedMessage.from}`,
+    contact: {
+      name: contactName,
+      phoneNumber: contactPhoneNumber,
+    },
+  })
+}

packages/bot-engine/apiHandlers/saveClientLogs.ts

@@ -0,0 +1,49 @@
+import { TRPCError } from '@trpc/server'
+import { ChatLog } from '@typebot.io/schemas'
+import { formatLogDetails } from '../logs/helpers/formatLogDetails'
+import { getSession } from '../queries/getSession'
+import { saveLogs } from '../queries/saveLogs'
+
+type Props = {
+  sessionId: string
+  clientLogs: ChatLog[]
+}
+
+export const saveClientLogs = async ({ sessionId, clientLogs }: Props) => {
+  const session = await getSession(sessionId)
+
+  if (!session) {
+    throw new TRPCError({
+      code: 'NOT_FOUND',
+      message: 'Session not found.',
+    })
+  }
+
+  const resultId = session.state.typebotsQueue[0].resultId
+
+  if (!resultId) {
+    throw new TRPCError({
+      code: 'NOT_FOUND',
+      message: 'Result not found.',
+    })
+  }
+
+  try {
+    await saveLogs(
+      clientLogs.map((log) => ({
+        ...log,
+        resultId,
+        details: formatLogDetails(log.details),
+      }))
+    )
+    return {
+      message: 'Logs successfully saved.',
+    }
+  } catch (e) {
+    console.error('Failed to save logs', e)
+    throw new TRPCError({
+      code: 'INTERNAL_SERVER_ERROR',
+      message: 'Failed to save logs.',
+    })
+  }
+}

packages/bot-engine/apiHandlers/startChat.ts

@@ -0,0 +1,107 @@
+import { computeCurrentProgress } from '../computeCurrentProgress'
+import { filterPotentiallySensitiveLogs } from '../logs/filterPotentiallySensitiveLogs'
+import { restartSession } from '../queries/restartSession'
+import { saveStateToDatabase } from '../saveStateToDatabase'
+import { startSession } from '../startSession'
+
+type Props = {
+  origin: string | undefined
+  message?: string
+  isOnlyRegistering: boolean
+  publicId: string
+  isStreamEnabled: boolean
+  prefilledVariables?: Record<string, unknown>
+  resultId?: string
+}
+
+export const startChat = async ({
+  origin,
+  message,
+  isOnlyRegistering,
+  publicId,
+  isStreamEnabled,
+  prefilledVariables,
+  resultId: startResultId,
+}: Props) => {
+  const {
+    typebot,
+    messages,
+    input,
+    resultId,
+    dynamicTheme,
+    logs,
+    clientSideActions,
+    newSessionState,
+    visitedEdges,
+  } = await startSession({
+    version: 2,
+    startParams: {
+      type: 'live',
+      isOnlyRegistering,
+      isStreamEnabled,
+      publicId,
+      prefilledVariables,
+      resultId: startResultId,
+    },
+    message,
+  })
+
+  let corsOrigin
+
+  if (
+    newSessionState.allowedOrigins &&
+    newSessionState.allowedOrigins.length > 0
+  ) {
+    if (origin && newSessionState.allowedOrigins.includes(origin))
+      corsOrigin = origin
+    else corsOrigin = newSessionState.allowedOrigins[0]
+  }
+
+  const session = isOnlyRegistering
+    ? await restartSession({
+        state: newSessionState,
+      })
+    : await saveStateToDatabase({
+        session: {
+          state: newSessionState,
+        },
+        input,
+        logs,
+        clientSideActions,
+        visitedEdges,
+        hasCustomEmbedBubble: messages.some(
+          (message) => message.type === 'custom-embed'
+        ),
+      })
+
+  const isEnded =
+    newSessionState.progressMetadata &&
+    !input?.id &&
+    (clientSideActions?.filter((c) => c.expectsDedicatedReply).length ?? 0) ===
+      0
+
+  return {
+    sessionId: session.id,
+    typebot: {
+      id: typebot.id,
+      theme: typebot.theme,
+      settings: typebot.settings,
+    },
+    messages,
+    input,
+    resultId,
+    dynamicTheme,
+    logs: logs?.filter(filterPotentiallySensitiveLogs),
+    clientSideActions,
+    corsOrigin,
+    progress: newSessionState.progressMetadata
+      ? isEnded
+        ? 100
+        : computeCurrentProgress({
+            typebotsQueue: newSessionState.typebotsQueue,
+            progressMetadata: newSessionState.progressMetadata,
+            currentInputBlockId: input?.id,
+          })
+      : undefined,
+  }
+}

packages/bot-engine/apiHandlers/startChatPreview.ts

@@ -0,0 +1,97 @@
+import { StartFrom, StartTypebot } from '@typebot.io/schemas'
+import { restartSession } from '../queries/restartSession'
+import { saveStateToDatabase } from '../saveStateToDatabase'
+import { startSession } from '../startSession'
+import { computeCurrentProgress } from '../computeCurrentProgress'
+
+type Props = {
+  message?: string
+  isOnlyRegistering: boolean
+  isStreamEnabled: boolean
+  startFrom?: StartFrom
+  typebotId: string
+  typebot?: StartTypebot
+  userId?: string
+  prefilledVariables?: Record<string, unknown>
+}
+
+export const startChatPreview = async ({
+  message,
+  isOnlyRegistering,
+  isStreamEnabled,
+  startFrom,
+  typebotId,
+  typebot: startTypebot,
+  userId,
+  prefilledVariables,
+}: Props) => {
+  const {
+    typebot,
+    messages,
+    input,
+    dynamicTheme,
+    logs,
+    clientSideActions,
+    newSessionState,
+    visitedEdges,
+  } = await startSession({
+    version: 2,
+    startParams: {
+      type: 'preview',
+      isOnlyRegistering,
+      isStreamEnabled,
+      startFrom,
+      typebotId,
+      typebot: startTypebot,
+      userId,
+      prefilledVariables,
+    },
+    message,
+  })
+
+  const session = isOnlyRegistering
+    ? await restartSession({
+        state: newSessionState,
+      })
+    : await saveStateToDatabase({
+        session: {
+          state: newSessionState,
+        },
+        input,
+        logs,
+        clientSideActions,
+        visitedEdges,
+        hasCustomEmbedBubble: messages.some(
+          (message) => message.type === 'custom-embed'
+        ),
+      })
+
+  const isEnded =
+    newSessionState.progressMetadata &&
+    !input?.id &&
+    (clientSideActions?.filter((c) => c.expectsDedicatedReply).length ?? 0) ===
+      0
+
+  return {
+    sessionId: session.id,
+    typebot: {
+      id: typebot.id,
+      theme: typebot.theme,
+      settings: typebot.settings,
+    },
+    messages,
+    input,
+    dynamicTheme,
+    logs,
+    clientSideActions,
+    progress: newSessionState.progressMetadata
+      ? isEnded
+        ? 100
+        : computeCurrentProgress({
+            typebotsQueue: newSessionState.typebotsQueue,
+            progressMetadata: newSessionState.progressMetadata,
+            currentInputBlockId: input?.id,
+          })
+      : undefined,
+  }
+}

packages/bot-engine/apiHandlers/updateTypebotInSession.ts

@@ -0,0 +1,97 @@
+import { TRPCError } from '@trpc/server'
+import prisma from '@typebot.io/lib/prisma'
+import {
+  SessionState,
+  Variable,
+  PublicTypebot,
+  Typebot,
+} from '@typebot.io/schemas'
+import { getSession } from '../queries/getSession'
+
+type Props = {
+  user?: { id: string }
+  sessionId: string
+}
+
+export const updateTypebotInSession = async ({ user, sessionId }: Props) => {
+  if (!user)
+    throw new TRPCError({ code: 'UNAUTHORIZED', message: 'Unauthorized' })
+  const session = await getSession(sessionId)
+  if (!session)
+    throw new TRPCError({ code: 'NOT_FOUND', message: 'Session not found' })
+
+  const publicTypebot = (await prisma.publicTypebot.findFirst({
+    where: {
+      typebot: {
+        id: session.state.typebotsQueue[0].typebot.id,
+        OR: [
+          {
+            workspace: {
+              members: {
+                some: { userId: user.id, role: { in: ['ADMIN', 'MEMBER'] } },
+              },
+            },
+          },
+          {
+            collaborators: {
+              some: { userId: user.id, type: { in: ['WRITE'] } },
+            },
+          },
+        ],
+      },
+    },
+    select: {
+      edges: true,
+      groups: true,
+      variables: true,
+    },
+  })) as Pick<PublicTypebot, 'edges' | 'variables' | 'groups'> | null
+
+  if (!publicTypebot)
+    throw new TRPCError({ code: 'UNAUTHORIZED', message: 'Unauthorized' })
+
+  const newSessionState = updateSessionState(session.state, publicTypebot)
+
+  await prisma.chatSession.updateMany({
+    where: { id: session.id },
+    data: { state: newSessionState },
+  })
+
+  return { message: 'success' } as const
+}
+
+const updateSessionState = (
+  currentState: SessionState,
+  newTypebot: Pick<PublicTypebot, 'edges' | 'variables' | 'groups'>
+): SessionState => ({
+  ...currentState,
+  typebotsQueue: currentState.typebotsQueue.map((typebotInQueue, index) =>
+    index === 0
+      ? {
+          ...typebotInQueue,
+          typebot: {
+            ...typebotInQueue.typebot,
+            edges: newTypebot.edges,
+            groups: newTypebot.groups,
+            variables: updateVariablesInSession(
+              typebotInQueue.typebot.variables,
+              newTypebot.variables
+            ),
+          },
+        }
+      : typebotInQueue
+  ) as SessionState['typebotsQueue'],
+})
+
+const updateVariablesInSession = (
+  currentVariables: Variable[],
+  newVariables: Typebot['variables']
+): Variable[] => [
+  ...currentVariables,
+  ...newVariables.filter(
+    (newVariable) =>
+      !currentVariables.find(
+        (currentVariable) => currentVariable.id === newVariable.id
+      )
+  ),
+]