🚸 (whatsapp) Avoid multiple replies to be sent concurently

Closes #972, closes #1453

.vscode/settings.json

@@ -15,7 +15,8 @@
     "DATABASE_URL": "postgresql://postgres:typebot@127.0.0.1:5432/typebot",
     "NEXT_PUBLIC_VIEWER_URL": "http://localhost:3001",
     "NEXTAUTH_URL": "http://localhost:3000",
-    "ENCRYPTION_SECRET": "H+KbL/OFrqbEuDy/1zX8bsPG+spXri3S"
+    "ENCRYPTION_SECRET": "H+KbL/OFrqbEuDy/1zX8bsPG+spXri3S",
+    "S3_ENDPOINT": "http://localhost:9000"
   },
   "[prisma]": {
     "editor.defaultFormatter": "Prisma.prisma"

apps/docs/openapi/viewer.json

@@ -12473,10 +12473,18 @@
                     }
                   }
                 }
+              },
+              "runtime": {
+                "type": "string",
+                "enum": [
+                  "edge",
+                  "nodejs"
+                ]
               }
             },
             "required": [
-              "messages"
+              "messages",
+              "runtime"
             ]
           },
           "lastBubbleBlockId": {
@@ -12753,6 +12761,13 @@
               true
             ]
           },
+          "runtime": {
+            "type": "string",
+            "enum": [
+              "edge",
+              "nodejs"
+            ]
+          },
           "lastBubbleBlockId": {
             "type": "string"
           },
@@ -12762,7 +12777,8 @@
         },
         "required": [
           "type",
-          "stream"
+          "stream",
+          "runtime"
         ],
         "title": "Exec stream"
       },

packages/bot-engine/queries/createSession.ts

@@ -5,15 +5,18 @@ import { SessionState } from '@typebot.io/schemas'
 type Props = {
   id?: string
   state: SessionState
+  isReplying?: boolean
 }
 
 export const createSession = ({
   id,
   state,
+  isReplying,
 }: Props): Prisma.PrismaPromise<any> =>
   prisma.chatSession.create({
     data: {
       id,
       state,
+      isReplying,
     },
   })

packages/bot-engine/queries/getSession.ts

@@ -4,7 +4,7 @@ import { sessionStateSchema } from '@typebot.io/schemas'
 export const getSession = async (sessionId: string) => {
   const session = await prisma.chatSession.findUnique({
     where: { id: sessionId },
-    select: { id: true, state: true, updatedAt: true },
+    select: { id: true, state: true, updatedAt: true, isReplying: true },
   })
   if (!session) return null
   return { ...session, state: sessionStateSchema.parse(session.state) }

packages/bot-engine/queries/setChatSessionHasReplying.ts

@@ -0,0 +1,20 @@
+import prisma from '@typebot.io/lib/prisma'
+
+type Props = {
+  existingSessionId: string | undefined
+  newSessionId: string
+}
+export const setChatSessionHasReplying = async ({
+  existingSessionId,
+  newSessionId,
+}: Props) => {
+  if (existingSessionId) {
+    return prisma.chatSession.updateMany({
+      where: { id: existingSessionId },
+      data: { isReplying: true },
+    })
+  }
+  return prisma.chatSession.createMany({
+    data: { id: newSessionId, isReplying: true, state: {} },
+  })
+}

packages/bot-engine/queries/updateSession.ts

@@ -5,15 +5,18 @@ import { SessionState } from '@typebot.io/schemas'
 type Props = {
   id: string
   state: SessionState
+  isReplying: boolean
 }
 
 export const updateSession = ({
   id,
   state,
+  isReplying,
 }: Props): Prisma.PrismaPromise<any> =>
   prisma.chatSession.updateMany({
     where: { id },
     data: {
       state,
+      isReplying,
     },
   })

packages/bot-engine/saveStateToDatabase.ts

@@ -16,7 +16,6 @@ type Props = {
   logs: ContinueChatResponse['logs']
   clientSideActions: ContinueChatResponse['clientSideActions']
   visitedEdges: VisitedEdge[]
-  forceCreateSession?: boolean
   hasCustomEmbedBubble?: boolean
 }
 
@@ -25,7 +24,6 @@ export const saveStateToDatabase = async ({
   input,
   logs,
   clientSideActions,
-  forceCreateSession,
   visitedEdges,
   hasCustomEmbedBubble,
 }: Props) => {
@@ -43,13 +41,12 @@ export const saveStateToDatabase = async ({
 
   if (id) {
     if (isCompleted && resultId) queries.push(deleteSession(id))
-    else queries.push(updateSession({ id, state }))
+    else queries.push(updateSession({ id, state, isReplying: false }))
   }
 
-  const session =
+  const session = id
-    id && !forceCreateSession
+    ? { state, id }
-      ? { state, id }
+    : await createSession({ id, state, isReplying: false })
-      : await createSession({ id, state })
 
   if (!resultId) {
     if (queries.length > 0) await prisma.$transaction(queries)

packages/bot-engine/whatsapp/resumeWhatsAppFlow.ts

@@ -13,6 +13,7 @@ import { saveStateToDatabase } from '../saveStateToDatabase'
 import prisma from '@typebot.io/lib/prisma'
 import { isDefined } from '@typebot.io/lib/utils'
 import { Reply } from '../types'
+import { setChatSessionHasReplying } from '../queries/setChatSessionHasReplying'
 
 type Props = {
   receivedMessage: WhatsAppIncomingMessage
@@ -67,6 +68,18 @@ export const resumeWhatsAppFlow = async ({
 
   const session = await getSession(sessionId)
 
+  if (session?.isReplying) {
+    console.log('Is currently replying, skipping...')
+    return {
+      message: 'Message received',
+    }
+  }
+
+  await setChatSessionHasReplying({
+    existingSessionId: session?.id,
+    newSessionId: sessionId,
+  })
+
   const isSessionExpired =
     session &&
     isDefined(session.state.expiryTimeout) &&
@@ -116,7 +129,6 @@ export const resumeWhatsAppFlow = async ({
   })
 
   await saveStateToDatabase({
-    forceCreateSession: !session && isDefined(input),
     clientSideActions: [],
     input,
     logs,

packages/prisma/mysql/schema.prisma

@@ -349,10 +349,11 @@ model ClaimableCustomPlan {
 }
 
 model ChatSession {
-  id        String   @id @default(cuid())
+  id         String   @id @default(cuid())
-  createdAt DateTime @default(now())
+  createdAt  DateTime @default(now())
-  updatedAt DateTime @default(now()) @updatedAt
+  updatedAt  DateTime @default(now()) @updatedAt
-  state     Json
+  state      Json
+  isReplying Boolean?
 }
 
 model ThemeTemplate {

packages/prisma/postgresql/migrations/20240418062443_add_is_replying_in_chat_session/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "ChatSession" ADD COLUMN     "isReplying" BOOLEAN;

packages/prisma/postgresql/schema.prisma

@@ -328,10 +328,11 @@ model ClaimableCustomPlan {
 }
 
 model ChatSession {
-  id        String   @id @default(cuid())
+  id         String   @id @default(cuid())
-  createdAt DateTime @default(now())
+  createdAt  DateTime @default(now())
-  updatedAt DateTime @default(now()) @updatedAt
+  updatedAt  DateTime @default(now()) @updatedAt
-  state     Json
+  state      Json
+  isReplying Boolean?
 }
 
 model ThemeTemplate {

packages/schemas/features/chat/schema.ts

@@ -28,13 +28,20 @@ import { preprocessTypebot } from '../typebot/helpers/preprocessTypebot'
 import { typebotV5Schema, typebotV6Schema } from '../typebot/typebot'
 import { BubbleBlockType } from '../blocks/bubbles/constants'
 import { clientSideActionSchema } from './clientSideAction'
+import { ChatSession as ChatSessionFromPrisma } from '@typebot.io/prisma'
 
 const chatSessionSchema = z.object({
   id: z.string(),
   createdAt: z.date(),
   updatedAt: z.date(),
   state: sessionStateSchema,
-})
+  isReplying: z
+    .boolean()
+    .nullable()
+    .describe(
+      'Used in WhatsApp runtime to avoid concurrent replies from the bot'
+    ),
+}) satisfies z.ZodType<ChatSessionFromPrisma, z.ZodTypeDef, unknown>
 export type ChatSession = z.infer<typeof chatSessionSchema>
 
 const textMessageSchema = z