finish cleaning up db access

backend/src/analyze-traces.ts

@@ -20,7 +20,12 @@ import { isGraphQlEndpoint } from "services/graphql"
 import { isQueryFailedError, retryTypeormTransaction } from "utils/db"
 import { MetloContext } from "types"
 import { DatabaseService } from "services/database"
-import { getEntityManager, getQB } from "services/database/utils"
+import {
+  getEntityManager,
+  getQB,
+  insertValueBuilder,
+  insertValuesBuilder,
+} from "services/database/utils"
 
 const getEndpointQuery = (ctx: MetloContext) => `
 SELECT
@@ -63,14 +68,13 @@ WHERE
   "apiEndpointUuid" = $1
 `
 
-const getQueuedApiTrace = async (
-  ctx: MetloContext,
-): Promise<QueuedApiTrace> => {
+const getQueuedApiTrace = async (): Promise<{
+  trace: QueuedApiTrace
+  ctx: MetloContext
+}> => {
   try {
-    const traceString = await RedisClient.popValueFromRedisList(
-      ctx,
-      TRACES_QUEUE,
-    )
+    const unsafeRedisClient = RedisClient.getInstance()
+    const traceString = await unsafeRedisClient.lpop(TRACES_QUEUE)
     return JSON.parse(traceString)
   } catch (err) {
     return null
@@ -115,18 +119,17 @@ const analyze = async (
   await queryRunner.startTransaction()
   await retryTypeormTransaction(
     () =>
-      getEntityManager(ctx, queryRunner).insert(ApiTrace, {
-        ...trace,
-        apiEndpointUuid: apiEndpoint.uuid,
-      }),
+      getEntityManager(ctx, queryRunner).insert(ApiTrace, [
+        {
+          ...trace,
+          apiEndpointUuid: apiEndpoint.uuid,
+        },
+      ]),
     5,
   )
   await retryTypeormTransaction(
     () =>
-      getQB(ctx, queryRunner)
-        .insert()
-        .into(DataField)
-        .values(dataFields)
+      insertValuesBuilder(ctx, queryRunner, DataField, dataFields)
         .orUpdate(
           [
             "dataClasses",
@@ -142,12 +145,7 @@ const analyze = async (
   )
   await retryTypeormTransaction(
     () =>
-      getQB(ctx, queryRunner)
-        .insert()
-        .into(Alert)
-        .values(alerts)
-        .orIgnore()
-        .execute(),
+      insertValuesBuilder(ctx, queryRunner, Alert, alerts).orIgnore().execute(),
     5,
   )
   await retryTypeormTransaction(
@@ -215,11 +213,12 @@ const generateEndpoint = async (
       await queryRunner.startTransaction()
       await retryTypeormTransaction(
         () =>
-          getQB(ctx, queryRunner)
-            .insert()
-            .into(ApiEndpoint)
-            .values(apiEndpoint)
-            .execute(),
+          insertValueBuilder(
+            ctx,
+            queryRunner,
+            ApiEndpoint,
+            apiEndpoint,
+          ).execute(),
         5,
       )
       await queryRunner.commitTransaction()
@@ -251,8 +250,6 @@ const generateEndpoint = async (
 }
 
 const analyzeTraces = async (): Promise<void> => {
-  const ctx: MetloContext = {}
-
   const datasource = await AppDataSource.initialize()
   if (!datasource.isInitialized) {
     console.error("Couldn't initialize datasource...")
@@ -264,8 +261,9 @@ const analyzeTraces = async (): Promise<void> => {
   await queryRunner.connect()
   while (true) {
     try {
-      const trace = await getQueuedApiTrace(ctx)
-      if (trace) {
+      const queued = await getQueuedApiTrace()
+      if (queued) {
+        const { trace, ctx } = queued
         trace.createdAt = new Date(trace.createdAt)
         const apiEndpoint: ApiEndpoint = (
           await queryRunner.query(getEndpointQuery(ctx), [

backend/src/api/tests/index.ts

@@ -3,8 +3,13 @@ import ApiResponseHandler from "api-response-handler"
 import { runTest } from "@metlo/testing"
 import { ApiEndpointTest } from "models"
 import { GetEndpointsService } from "services/get-endpoints"
-import { getRepoQB } from "services/database/utils"
+import {
+  getRepoQB,
+  insertValueBuilder,
+  insertValuesBuilder,
+} from "services/database/utils"
 import { MetloRequest } from "types"
+import { AppDataSource } from "data-source"
 
 export const runTestHandler = async (
   req: MetloRequest,
@@ -33,25 +38,36 @@ export const saveTest = async (
     test: { uuid, name, tags, requests },
     endpointUuid,
   } = req.body
-  let testInsert = await getRepoQB(req.ctx, ApiEndpointTest)
-    .insert()
-    .into(ApiEndpointTest)
-    .values({
-      uuid: uuid,
-      name,
-      tags,
-      requests,
-      apiEndpoint: {
-        uuid: endpointUuid,
+  let queryRunner = AppDataSource.createQueryRunner()
+  await queryRunner.connect()
+  try {
+    let testInsert = await insertValueBuilder(
+      req.ctx,
+      queryRunner,
+      ApiEndpointTest,
+      {
+        uuid: uuid,
+        name,
+        tags,
+        requests,
+        apiEndpoint: {
+          uuid: endpointUuid,
+        },
       },
-    })
-    .orUpdate(["name", "tags", "requests"], ["uuid"])
-    .execute()
-  let resp = await getRepoQB(req.ctx, ApiEndpointTest)
-    .select()
-    .where("uuid = :uuid", testInsert.identifiers[0])
-    .getOne()
-  await ApiResponseHandler.success(res, resp)
+    )
+      .orUpdate(["name", "tags", "requests"], ["uuid"])
+      .execute()
+    let resp = await getRepoQB(req.ctx, ApiEndpointTest)
+      .select()
+      .where("uuid = :uuid", testInsert.identifiers[0])
+      .getOne()
+    await ApiResponseHandler.success(res, resp)
+  } catch (err) {
+    console.error(`Error while saving test: ${err}`)
+    return await ApiResponseHandler.error(res, err)
+  } finally {
+    await queryRunner.release()
+  }
 }
 
 export const getTest = async (

backend/src/services/database/utils.ts

@@ -28,6 +28,24 @@ export const getQB = (ctx: MetloContext, queryRunner: QueryRunner) => {
   return qb
 }
 
+export function insertValueBuilder<Entity extends ObjectLiteral>(
+  ctx: MetloContext,
+  queryRunner: QueryRunner,
+  into: EntityTarget<Entity>,
+  value: QueryDeepPartialEntity<Entity>,
+) {
+  return getQB(ctx, queryRunner).insert().into(into).values(value)
+}
+
+export function insertValuesBuilder<Entity extends ObjectLiteral>(
+  ctx: MetloContext,
+  queryRunner: QueryRunner,
+  into: EntityTarget<Entity>,
+  values: QueryDeepPartialEntity<Entity>[],
+) {
+  return getQB(ctx, queryRunner).insert().into(into).values(values)
+}
+
 export function getRepoQB<Entity extends ObjectLiteral>(
   ctx: MetloContext,
   target: EntityTarget<Entity>,
@@ -125,16 +143,8 @@ export class WrappedEntityManager {
     return this.manager.findOneBy(entityClass, where)
   }
 
-  save<Entity>(
-    targetOrEntity: Entity,
-    maybeEntityOrOptions?: SaveOptions,
-  ): Promise<Entity>
-  save<Entity, T extends DeepPartial<Entity>>(
-    targetOrEntity: EntityTarget<Entity>,
-    maybeEntityOrOptions: T,
-    maybeOptions?: SaveOptions,
-  ) {
-    return this.manager.save(targetOrEntity, maybeEntityOrOptions, maybeOptions)
+  save<Entity>(target: Entity, options?: SaveOptions): Promise<Entity> {
+    return this.manager.save(target, options)
   }
 
   remove<Entity>(entity: Entity, options?: RemoveOptions): Promise<Entity> {
@@ -143,7 +153,7 @@ export class WrappedEntityManager {
 
   insert<Entity>(
     target: EntityTarget<Entity>,
-    entity: QueryDeepPartialEntity<Entity> | QueryDeepPartialEntity<Entity>[],
+    entity: QueryDeepPartialEntity<Entity>[],
   ): Promise<InsertResult> {
     return this.manager.insert(target, entity)
   }

backend/src/services/get-endpoints/index.ts

@@ -207,7 +207,6 @@ export class GetEndpointsService {
       const hosts: { [host: string]: string }[] = await getRepoQB(
         ctx,
         ApiEndpoint,
-        "apiEndpoint",
       )
         .select(["host"])
         .distinct(true)

backend/src/services/jobs/check-unauthenticated-endpoints.ts

@@ -6,10 +6,12 @@ import {
   getUnauthenticatedEndpointsSensitiveData,
 } from "./queries"
 import { AlertService } from "services/alert"
-import { getQB } from "services/database/utils"
+import { insertValuesBuilder } from "services/database/utils"
 import { MetloContext } from "types"
 
-const checkForUnauthenticatedEndpoints = async (ctx: MetloContext): Promise<void> => {
+const checkForUnauthenticatedEndpoints = async (
+  ctx: MetloContext,
+): Promise<void> => {
   const queryRunner = AppDataSource.createQueryRunner()
   try {
     await queryRunner.connect()
@@ -26,7 +28,7 @@ const checkForUnauthenticatedEndpoints = async (ctx: MetloContext): Promise<void
     const alerts = await AlertService.createUnauthEndpointSenDataAlerts(
       endpointsToAlert,
     )
-    await getQB(ctx, queryRunner).insert().into(Alert).values(alerts).execute()
+    await insertValuesBuilder(ctx, queryRunner, Alert, alerts).execute()
   } catch (err) {
     console.error(
       `Encountered error when checking for unauthenticated endpoints: ${err}`,

backend/src/services/jobs/generate-endpoints-traces.ts

@@ -9,7 +9,11 @@ import { skipAutoGeneratedMatch, isSuspectedParamater } from "utils"
 import { GenerateEndpoint } from "./types"
 import { retryTypeormTransaction } from "utils/db"
 import { MetloContext } from "types"
-import { getEntityManager, getQB } from "services/database/utils"
+import {
+  getEntityManager,
+  getQB,
+  insertValueBuilder,
+} from "services/database/utils"
 
 const generateEndpointsFromTraces = async (
   ctx: MetloContext,
@@ -43,19 +47,22 @@ const generateEndpointsFromTraces = async (
       const regexToTracesMap: Record<string, GenerateEndpoint> = {}
       for (let i = 0; i < traces.length; i++) {
         const trace = traces[i]
-        const apiEndpoint = await getEntityManager(ctx, queryRunner).findOne(ApiEndpoint, {
-          where: {
-            pathRegex: Raw(alias => `:path ~ ${alias}`, {
-              path: trace.path,
-            }),
-            method: trace.method,
-            host: trace.host,
+        const apiEndpoint = await getEntityManager(ctx, queryRunner).findOne(
+          ApiEndpoint,
+          {
+            where: {
+              pathRegex: Raw(alias => `:path ~ ${alias}`, {
+                path: trace.path,
+              }),
+              method: trace.method,
+              host: trace.host,
+            },
+            relations: { openapiSpec: true },
+            order: {
+              numberParams: "ASC",
+            },
           },
-          relations: { openapiSpec: true },
-          order: {
-            numberParams: "ASC",
-          },
-        })
+        )
         if (apiEndpoint && !skipAutoGeneratedMatch(apiEndpoint, trace.path)) {
           apiEndpoint.updateDates(trace.createdAt)
 
@@ -159,20 +166,16 @@ const generateEndpointsFromTraces = async (
         await queryRunner.startTransaction()
         await retryTypeormTransaction(
           () =>
-            getQB(ctx, queryRunner)
-              .insert()
-              .into(ApiEndpoint)
-              .values(apiEndpoint)
-              .execute(),
+            insertValueBuilder(
+              ctx,
+              queryRunner,
+              ApiEndpoint,
+              apiEndpoint,
+            ).execute(),
           5,
         )
         await retryTypeormTransaction(
-          () =>
-            getQB(ctx, queryRunner)
-              .insert()
-              .into(Alert)
-              .values(alert)
-              .execute(),
+          () => insertValueBuilder(ctx, queryRunner, Alert, alert).execute(),
           5,
         )
         await retryTypeormTransaction(
@@ -186,7 +189,10 @@ const generateEndpointsFromTraces = async (
         )
         await queryRunner.commitTransaction()
       }
-      traces = await getEntityManager(ctx, queryRunner).find(ApiTrace, tracesFindOptions)
+      traces = await getEntityManager(ctx, queryRunner).find(
+        ApiTrace,
+        tracesFindOptions,
+      )
     }
     console.log("Finished Generating Endpoints.")
   } catch (err) {

backend/src/services/log-request/index.ts

@@ -11,9 +11,13 @@ export class LogRequestService {
     ctx: MetloContext,
     traceParams: TraceParams,
   ): Promise<void> {
+    const unsafeRedisClient = RedisClient.getInstance()
     try {
       /** Log Request in ApiTrace table **/
-      const queueLength = await RedisClient.getListLength(ctx, TRACES_QUEUE)
+      let queueLength = 0
+      try {
+        await unsafeRedisClient.llen(TRACES_QUEUE)
+      } catch {}
       if (queueLength > 1000) {
         return
       }
@@ -45,11 +49,12 @@ export class LogRequestService {
       await BlockFieldsService.redactBlockedFields(ctx, apiTraceObj)
       await AuthenticationConfigService.setSessionMetadata(ctx, apiTraceObj)
 
-      RedisClient.pushValueToRedisList(
-        ctx,
+      await unsafeRedisClient.rpush(
         TRACES_QUEUE,
-        [JSON.stringify(apiTraceObj)],
-        true,
+        JSON.stringify({
+          ctx,
+          trace: apiTraceObj,
+        }),
       )
     } catch (err) {
       console.error(`Error in Log Request service: ${err}`)