Fixed problems in ingestion in ai-assistant

Author: preeesha

ai-assistant/src/core/db/neo4j.ts

@@ -119,10 +119,7 @@ export class Neo4j implements IDB {
         this.transactionUrl = undefined;
     }
 
-    async run(
-        query: string,
-        params?: any
-    ): Promise<Record<string, any>[] | null> {
+    async run(query: string, params?: any): Promise<Record<string, any>[]> {
         const data = {
             statements: [
                 {
@@ -152,7 +149,9 @@ export class Neo4j implements IDB {
         }
 
         if (response.errors.length) {
-            return null;
+            throw new Error(
+                response.errors.map((x) => JSON.stringify(x)).join("\n\n")
+            );
         }
 
         const nodes: Record<string, any>[] = [];

ai-assistant/src/endpoints/ingest.ts

@@ -22,12 +22,7 @@ import {
 namespace Helpers {
     async function insertNode(db: IDB, node: DBNode) {
         const query = new DBNode(node).getDBInsertQuery();
-        try {
-            await db.run(query);
-        } catch (e) {
-            console.log(e);
-            console.error("Failed to insert node");
-        }
+        await db.run(query, node);
     }
 
     export async function insertNodes(db: IDB, nodes: DBNode[]) {
@@ -41,35 +36,15 @@ export class IngestEndpoint extends ApiEndpoint {
     makeBodies(
         content: any
     ): [IngestEndpointRequestBody, IngestEndpointResponseBody] {
-        const requestBody = JSON.parse(content) as IngestEndpointRequestBody;
+        const requestBody = content as IngestEndpointRequestBody;
         const responseBody: IngestEndpointResponseBody = {
-            batchID: requestBody.batchID,
+            batchID: "hey",
             status: 200,
         };
 
         return [requestBody, responseBody];
     }
 
-    async commitProgress(
-        db: IDB
-    ): Promise<IngestEndpointResponseBody["status"]> {
-        try {
-            try {
-                await db.commitTransaction();
-            } catch (e) {
-                console.error(e);
-                await db.rollbackTransaction();
-
-                return 500;
-            }
-        } catch (e) {
-            console.error(e);
-            return 500;
-        }
-
-        return 200;
-    }
-
     public async post(
         request: IApiRequest,
         endpoint: IApiEndpointInfo,
@@ -85,15 +60,13 @@ export class IngestEndpoint extends ApiEndpoint {
         nodes = nodes.map((node) => new DBNode(node));
         await Promise.all(nodes.map((x) => x.fillEmbeddings(embeddingModel)));
         // -----------------------------------------------------------------------------------
-
         const db = new Neo4j(http);
         await db.verifyConnectivity();
         // -----------------------------------------------------------------------------------
-        await db.beginTransaction();
-        // -----------------------------------------------------------------------------------
-        await Helpers.insertNodes(db, nodes);
-        // -----------------------------------------------------------------------------------
-        responseBody.status = await this.commitProgress(db);
+        const jobs = nodes.map((node) =>
+            db.run(new DBNode(node).getDBInsertQuery(), node)
+        );
+        await Promise.all(jobs);
         // -----------------------------------------------------------------------------------
 
         return this.success(JSON.stringify(responseBody));