(EAI-611) chatbot server 500 errs (#592)

Author: nlarew

package-lock.json

@@ -42,6 +42,7 @@
     "cors": "^2.8.5",
     "dotenv": "^16.0.3",
     "express": "^4.18.2",
+    "express-promise-router": "^4.1.1",
     "express-rate-limit": "^6.10.0",
     "express-slow-down": "^1.6.0",
     "gpt3-tokenizer": "^1.1.5",

packages/mongodb-chatbot-server/package.json

@@ -1,4 +1,5 @@
-import { Request, Router, RequestHandler, Response } from "express";
+import { Request, RequestHandler, Response } from "express";
+import Router from "express-promise-router";
 import { rateLimit, Options as RateLimitOptions } from "express-rate-limit";
 import slowDown, { Options as SlowDownOptions } from "express-slow-down";
 import validateRequestSchema from "../../middleware/validateRequestSchema";

packages/mongodb-chatbot-server/src/routes/conversations/conversationsRouter.ts

@@ -53,7 +53,9 @@ export const sendErrorResponse = ({
     message: stripIndent`Responding with ${httpStatus} status and error message: ${errorMessage}.
     ${errorDetails ? `Error details: ${errorDetails}` : ""}`,
   });
-  return res.status(httpStatus).json({ error: errorMessage });
+  if (!res.writableEnded) {
+    return res.status(httpStatus).json({ error: errorMessage });
+  }
 };
 
 export function retryAsyncOperation<T>(

packages/mongodb-chatbot-server/src/utils.ts

@@ -80,28 +80,6 @@ describe("Data Streaming", () => {
       `data: {"type":"delta","data":"Once upon"}\n\ndata: {"type":"delta","data":" a time there was a"}\n\ndata: {"type":"delta","data":" very long string."}\n\n`
     );
   });
-
-  it("Bails out when a client closes a connection", async () => {
-    res.emit("close");
-    expect(dataStreamer.connected).toBe(false);
-    expect(() =>
-      dataStreamer.streamData({ type: "delta", data: "test" })
-    ).toThrow(Error);
-    await expect(async () => {
-      const stream = {
-        [Symbol.asyncIterator]() {
-          return {
-            async next() {
-              return {
-                done: true,
-              };
-            },
-          };
-        },
-      } as OpenAiStreamingResponse;
-      await dataStreamer.stream({ stream });
-    }).rejects.toThrow(Error);
-  });
 });
 
 function createChatCompletionWithDelta(

packages/mongodb-rag-core/src/DataStreamer.test.ts

@@ -1,6 +1,7 @@
 import { Response } from "express";
 import { OpenAiStreamingResponse } from "./llm";
 import { References } from "./References";
+import { logger } from "./logger";
 
 export function escapeNewlines(str: string): string {
   return str.replaceAll(`\n`, `\\n`);
@@ -135,11 +136,8 @@ export function makeDataStreamer(): DataStreamer {
         throw new Error("Tried to connect SSE, but it was already connected.");
       }
       sse = makeServerSentEventDispatcher<SomeStreamEvent>(res);
-      // If the client closes the connection, stop sending events
       res.on("close", () => {
-        if (this.connected) {
-          this.disconnect();
-        }
+        logger.info("SSE connection was closed.");
       });
       sse.connect();
       connected = true;