Add safe JSON parsing helper to improve error handling across services

Author: Roman Snapko

packages/engine/src/lib/handler/context/engine-constants.ts

@@ -14,6 +14,7 @@ import {
   TestRunLimitSettings,
   TriggerHookType,
 } from '@openops/shared';
+import { safeResponseJson } from '../../helper/response-helper';
 import {
   createPropsResolver,
   PropsResolver,
@@ -232,7 +233,7 @@ export class EngineConstants {
       },
     });
 
-    return (await response.json()) as Project;
+    return safeResponseJson<Project>(response);
   }
 }
 

packages/engine/src/lib/helper/response-helper.ts

@@ -0,0 +1,23 @@
+export const safeResponseJson = async <T>(response: Response): Promise<T> => {
+  const contentType = response.headers.get('content-type');
+  if (contentType && !contentType.includes('application/json')) {
+    const text = await response.text();
+    throw new Error(
+      `Expected JSON response, but received ${contentType}. Body: ${text.slice(
+        0,
+        100,
+      )}${text.length > 100 ? '...' : ''}`,
+    );
+  }
+
+  try {
+    return await response.json();
+  } catch (e) {
+    const text = await response.text();
+    throw new Error(
+      `Failed to parse JSON response. Body: ${text.slice(0, 100)}${
+        text.length > 100 ? '...' : ''
+      }`,
+    );
+  }
+};

packages/engine/src/lib/services/connections.service.ts

@@ -14,6 +14,7 @@ import {
   ExecutionError,
   FetchError,
 } from '../helper/execution-errors';
+import { safeResponseJson } from '../helper/response-helper';
 
 export const createConnectionService = ({
   projectId,
@@ -40,16 +41,17 @@ export const createConnectionService = ({
             httpStatus: response.status,
           });
         }
-        const connection: AppConnection = await response.json();
+        const connection: AppConnection = await safeResponseJson<AppConnection>(
+          response,
+        );
         if (connection.status === AppConnectionStatus.ERROR) {
           throw new ConnectionExpiredError(connectionName);
         }
         return getConnectionValue(connection);
       } catch (e) {
-        if (e instanceof ExecutionError) {
+        if (e instanceof ExecutionError || e instanceof Error) {
           throw e;
         }
-
         return handleFetchError({
           url,
           cause: e,

packages/engine/src/lib/services/files.service.ts

@@ -1,6 +1,7 @@
 import { FilesService, WorkflowFile } from '@openops/blocks-framework';
 import { isNil } from '@openops/shared';
 import fs from 'fs/promises';
+import { safeResponseJson } from '../helper/response-helper';
 
 const FILE_PREFIX_URL = 'file://';
 const MEMORY_PREFIX_URL = 'memory://';
@@ -129,7 +130,7 @@ async function writeDbFile({
     throw new Error('Failed to store entry ' + response.body);
   }
 
-  const result = await response.json();
+  const result = await safeResponseJson<{ url: string }>(response);
   return result.url;
 }
 

packages/engine/src/lib/services/storage.service.ts

@@ -19,6 +19,7 @@ import {
   StorageError,
   StorageLimitError,
 } from '../helper/execution-errors';
+import { safeResponseJson } from '../helper/response-helper';
 
 export const createStorageService = ({
   engineToken,
@@ -42,8 +43,11 @@ export const createStorageService = ({
           });
         }
 
-        return await response.json();
+        return await safeResponseJson<StoreEntry>(response);
       } catch (e) {
+        if (e instanceof ExecutionError || e instanceof Error) {
+          throw e;
+        }
         return handleFetchError({
           url,
           cause: e,
@@ -78,8 +82,11 @@ export const createStorageService = ({
           });
         }
 
-        return await response.json();
+        return await safeResponseJson<StoreEntry>(response);
       } catch (e) {
+        if (e instanceof ExecutionError || e instanceof Error) {
+          throw e;
+        }
         return handleFetchError({
           url,
           cause: e,
@@ -107,6 +114,9 @@ export const createStorageService = ({
 
         return null;
       } catch (e) {
+        if (e instanceof ExecutionError || e instanceof Error) {
+          throw e;
+        }
         return handleFetchError({
           url,
           cause: e,
@@ -136,9 +146,14 @@ export const createStorageService = ({
           throw new Error(`Failed to list keys: ${response.statusText}`);
         }
 
-        const result = await response.json();
+        const result = await safeResponseJson<{
+          entries: Array<{ key: string; value: unknown }>;
+        }>(response);
         return result.entries || [];
       } catch (e) {
+        if (e instanceof ExecutionError || e instanceof Error) {
+          throw e;
+        }
         return handleFetchError({
           url,
           cause: e,

packages/server/worker/src/lib/trigger/hooks/extract-trigger-payload-hooks.ts

@@ -38,19 +38,6 @@ export async function extractPayloads(
       handleFailureFlow(flowVersion, projectId, engineToken, true);
       return result.output as unknown[];
     } else {
-      if ('status' in result && result.status === 'INTERNAL_ERROR') {
-        logger.warn(
-          {
-            result,
-            blockName,
-            blockVersion,
-            flowId: flowVersion.flowId,
-          },
-          'Failed to execute trigger because of internal error in the engine',
-        );
-        return [];
-      }
-
       logger.error(
         {
           result,