[TTAHUB-3097] Implement transactions for worker processes

email_templates/changes_requested_by_manager/html.pug

@@ -2,7 +2,7 @@ style
   include ../email.css
 p Hello,
 p 
-p #{managerName} requested changed to report #{displayId}. 
+p #{managerName} requested changes to report #{displayId}. 
 if comments
   p #{managerName} provided the following comments:
   blockquote !{comments} 

src/lib/apiErrorHandler.js

@@ -13,37 +13,28 @@ import { sequelize } from '../models';
  * @returns {Promise<number|null>} - The ID of the stored request error, or null if storing failed.
  */
 async function logRequestError(req, operation, error, logContext) {
-  // Check if error logging should be suppressed
   if (
     operation !== 'SequelizeError'
     && process.env.SUPPRESS_ERROR_LOGGING
     && process.env.SUPPRESS_ERROR_LOGGING.toLowerCase() === 'true'
   ) {
     return 0;
   }
+  if (!error) {
+    return 0;
+  }
 
   try {
-    // Prepare the response body for storage
-    const responseBody = typeof error === 'object'
-      && error !== null ? { ...error, errorStack: error?.stack } : error;
+    const responseBody = typeof error === 'object' && error !== null
+      ? { ...error, errorStack: error?.stack }
+      : error;
 
-    // Prepare the request body for storage
     const requestBody = {
-      ...(req.body
-        && typeof req.body === 'object'
-        && Object.keys(req.body).length > 0
-        && { body: req.body }),
-      ...(req.params
-        && typeof req.params === 'object'
-        && Object.keys(req.params).length > 0
-        && { params: req.params }),
-      ...(req.query
-        && typeof req.query === 'object'
-        && Object.keys(req.query).length > 0
-        && { query: req.query }),
+      ...(req.body && typeof req.body === 'object' && Object.keys(req.body).length > 0 && { body: req.body }),
+      ...(req.params && typeof req.params === 'object' && Object.keys(req.params).length > 0 && { params: req.params }),
+      ...(req.query && typeof req.query === 'object' && Object.keys(req.query).length > 0 && { query: req.query }),
     };
 
-    // Create a request error in the database and get its ID
     const requestErrorId = await createRequestError({
       operation,
       uri: req.originalUrl,
@@ -69,15 +60,13 @@ async function logRequestError(req, operation, error, logContext) {
  * @param {Object} logContext - The context for logging.
  */
 export const handleError = async (req, res, error, logContext) => {
-  // Check if the environment is development
   if (process.env.NODE_ENV === 'development') {
     logger.error(error);
   }
 
   let operation;
   let label;
 
-  // Check if the error is an instance of Sequelize.Error
   if (error instanceof Sequelize.Error) {
     operation = 'SequelizeError';
     label = 'Sequelize error';
@@ -86,24 +75,14 @@ export const handleError = async (req, res, error, logContext) => {
     label = 'UNEXPECTED ERROR';
   }
 
-  // eslint-disable-next-line max-len
-  if (error instanceof Sequelize.ConnectionError || error instanceof Sequelize.ConnectionAcquireTimeoutError) {
-    logger.error(`${logContext.namespace} Connection Pool: ${JSON.stringify(sequelize.connectionManager.pool)}`);
+  if (error instanceof Sequelize.ConnectionError
+    || error instanceof Sequelize.ConnectionAcquireTimeoutError) {
+    logger.error(`${logContext.namespace} Connection Pool: ${JSON.stringify(sequelize?.connectionManager?.pool)}`);
   }
-
-  // Log the request error and get the error ID
   const requestErrorId = await logRequestError(req, operation, error, logContext);
 
-  let errorMessage;
-
-  // Check if the error has a stack property
-  if (error?.stack) {
-    errorMessage = error.stack;
-  } else {
-    errorMessage = error;
-  }
+  const errorMessage = error?.stack || error;
 
-  // Log the error message with the error ID if available
   if (requestErrorId) {
     logger.error(`${logContext.namespace} - id: ${requestErrorId} ${label} - ${errorMessage}`);
   } else {
@@ -114,12 +93,11 @@ export const handleError = async (req, res, error, logContext) => {
 };
 
 /**
- * Handles any unexpected errors in an error handler catch block
- *
- * @param {*} req - request
- * @param {*} res - response
- * @param {*} error - error
- * @param {*} logContext - useful data for logging
+ * Handles any unexpected errors in an error handler catch block.
+ * @param {Object} req - The request object.
+ * @param {Object} res - The response object.
+ * @param {Error} error - The error object.
+ * @param {Object} logContext - The context for logging.
  */
 export function handleUnexpectedErrorInCatchBlock(req, res, error, logContext) {
   logger.error(`${logContext.namespace} - Unexpected error in catch block - ${error}`);
@@ -128,11 +106,10 @@ export function handleUnexpectedErrorInCatchBlock(req, res, error, logContext) {
 
 /**
  * Handles API errors. Saves data in the RequestErrors table and sends 500 error.
- *
- * @param {*} req - request
- * @param {*} res - response
- * @param {*} error - error
- * @param {*} logContext - useful data for logging
+ * @param {Object} req - The request object.
+ * @param {Object} res - The response object.
+ * @param {Error} error - The error object.
+ * @param {Object} logContext - The context for logging.
  */
 export default async function handleErrors(req, res, error, logContext) {
   try {
@@ -141,3 +118,110 @@ export default async function handleErrors(req, res, error, logContext) {
     handleUnexpectedErrorInCatchBlock(req, res, e, logContext);
   }
 }
+
+/**
+ * Logs a worker error and stores it in the database.
+ * @param {Object} job - The job object.
+ * @param {string} operation - The operation name.
+ * @param {Error} error - The error object.
+ * @param {Object} logContext - The logging context.
+ * @returns {Promise<number|null>} - The ID of the stored request error, or null if storing failed.
+ */
+const logWorkerError = async (job, operation, error, logContext) => {
+  if (
+    operation !== 'SequelizeError'
+    && process.env.SUPPRESS_ERROR_LOGGING
+    && process.env.SUPPRESS_ERROR_LOGGING.toLowerCase() === 'true'
+  ) {
+    return 0;
+  }
+
+  try {
+    const responseBody = typeof error === 'object' && error !== null
+      ? { ...error, errorStack: error?.stack }
+      : error;
+
+    const requestBody = {
+      ...(job.data && typeof job.data === 'object' && Object.keys(job.data).length > 0 && { data: job.data }),
+    };
+
+    const requestErrorId = await createRequestError({
+      operation,
+      uri: job.queue.name,
+      method: 'PROCESS_JOB',
+      requestBody,
+      responseBody,
+      responseCode: INTERNAL_SERVER_ERROR,
+    });
+
+    return requestErrorId;
+  } catch (e) {
+    logger.error(`${logContext.namespace} - Sequelize error - unable to store RequestError - ${e}`);
+  }
+
+  return null;
+};
+
+/**
+ * Handles errors in a worker job.
+ * @param {Object} job - The job object.
+ * @param {Error} error - The error object.
+ * @param {Object} logContext - The context for logging.
+ */
+export const handleWorkerError = async (job, error, logContext) => {
+  if (process.env.NODE_ENV === 'development') {
+    logger.error(error);
+  }
+
+  let operation;
+  let label;
+
+  if (error instanceof Sequelize.Error) {
+    operation = 'SequelizeError';
+    label = 'Sequelize error';
+  } else {
+    operation = 'UNEXPECTED_ERROR';
+    label = 'UNEXPECTED ERROR';
+  }
+
+  if (error instanceof Sequelize.ConnectionError
+    || error instanceof Sequelize.ConnectionAcquireTimeoutError) {
+    logger.error(`${logContext.namespace} Connection Pool: ${JSON.stringify(sequelize.connectionManager.pool)}`);
+  }
+
+  const requestErrorId = await logWorkerError(job, operation, error, logContext);
+
+  const errorMessage = error?.stack || error;
+
+  if (requestErrorId) {
+    logger.error(`${logContext.namespace} - id: ${requestErrorId} ${label} - ${errorMessage}`);
+  } else {
+    logger.error(`${logContext.namespace} - ${label} - ${errorMessage}`);
+  }
+
+  // Handle job failure as needed
+};
+
+/**
+ * Handles any unexpected errors in a worker error handler catch block.
+ * @param {Object} job - The job object.
+ * @param {Error} error - The error object.
+ * @param {Object} logContext - The context for logging.
+ */
+export const handleUnexpectedWorkerError = (job, error, logContext) => {
+  logger.error(`${logContext.namespace} - Unexpected error in catch block - ${error}`);
+};
+
+/**
+ * Handles worker job errors. Logs the error and stores it in the database.
+ * @param {Object} job - The job object.
+ * @param {Error} error - The error object.
+ * @param {Object} logContext - The context for logging.
+ */
+export const handleWorkerErrors = async (job, error, logContext) => {
+  try {
+    await handleWorkerError(job, error, logContext);
+  } catch (e) {
+    handleUnexpectedWorkerError(job, e, logContext);
+  }
+};

src/lib/apiErrorHandler.test.js

@@ -1,7 +1,7 @@
 import Sequelize from 'sequelize';
 import { INTERNAL_SERVER_ERROR } from 'http-codes';
 import db, { RequestErrors } from '../models';
-import handleErrors, { handleUnexpectedErrorInCatchBlock } from './apiErrorHandler';
+import handleErrors, { handleUnexpectedErrorInCatchBlock, handleWorkerErrors, handleUnexpectedWorkerError } from './apiErrorHandler';
 
 const mockUser = {
   id: 47,
@@ -31,6 +31,11 @@ const mockResponse = {
   })),
 };
 
+const mockJob = {
+  data: { jobDetail: 'example job detail' },
+  queue: { name: 'exampleQueue' },
+};
+
 const mockSequelizeError = new Sequelize.Error('Not all ok here');
 
 const mockLogContext = {
@@ -54,6 +59,7 @@ describe('apiErrorHandler', () => {
     const requestErrors = await RequestErrors.findAll();
 
     expect(requestErrors.length).not.toBe(0);
+    expect(requestErrors[0].operation).toBe('SequelizeError');
   });
 
   it('handles a generic error', async () => {
@@ -65,9 +71,10 @@ describe('apiErrorHandler', () => {
     const requestErrors = await RequestErrors.findAll();
 
     expect(requestErrors.length).not.toBe(0);
+    expect(requestErrors[0].operation).toBe('UNEXPECTED_ERROR');
   });
 
-  it('can handle unexpected error in catch block', async () => {
+  it('handles unexpected error in catch block', async () => {
     const mockUnexpectedErr = new Error('Unexpected error');
     handleUnexpectedErrorInCatchBlock(mockRequest, mockResponse, mockUnexpectedErr, mockLogContext);
 
@@ -77,4 +84,97 @@ describe('apiErrorHandler', () => {
 
     expect(requestErrors.length).toBe(0);
   });
+
+  it('handles error suppression when SUPPRESS_ERROR_LOGGING is true', async () => {
+    process.env.SUPPRESS_ERROR_LOGGING = 'true';
+    const mockGenericError = new Error('Unknown error');
+    await handleErrors(mockRequest, mockResponse, mockGenericError, mockLogContext);
+
+    expect(mockResponse.status).toHaveBeenCalledWith(INTERNAL_SERVER_ERROR);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).toBe(0);
+
+    delete process.env.SUPPRESS_ERROR_LOGGING;
+  });
+
+  it('logs connection pool information on connection errors', async () => {
+    const mockConnectionError = new Sequelize.ConnectionError(new Error('Connection error'));
+    await handleErrors(mockRequest, mockResponse, mockConnectionError, mockLogContext);
+
+    expect(mockResponse.status).toHaveBeenCalledWith(INTERNAL_SERVER_ERROR);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).not.toBe(0);
+    expect(requestErrors[0].operation).toBe('SequelizeError');
+  });
+
+  it('handles worker errors', async () => {
+    const mockWorkerError = new Error('Worker error');
+    await handleWorkerErrors(mockJob, mockWorkerError, mockLogContext);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).not.toBe(0);
+    expect(requestErrors[0].operation).toBe('UNEXPECTED_ERROR');
+  });
+
+  it('handles worker Sequelize errors', async () => {
+    const mockSequelizeWorkerError = new Sequelize.Error('Sequelize worker error');
+    await handleWorkerErrors(mockJob, mockSequelizeWorkerError, mockLogContext);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).not.toBe(0);
+    expect(requestErrors[0].operation).toBe('SequelizeError');
+  });
+
+  it('handles unexpected worker error in catch block', async () => {
+    const mockUnexpectedWorkerError = new Error('Unexpected worker error');
+    handleUnexpectedWorkerError(mockJob, mockUnexpectedWorkerError, mockLogContext);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).toBe(0);
+  });
+
+  it('handles null error', async () => {
+    await handleErrors(mockRequest, mockResponse, null, mockLogContext);
+
+    expect(mockResponse.status).toHaveBeenCalledWith(INTERNAL_SERVER_ERROR);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).toBe(0);
+  });
+
+  it('handles undefined error', async () => {
+    await handleErrors(mockRequest, mockResponse, undefined, mockLogContext);
+
+    expect(mockResponse.status).toHaveBeenCalledWith(INTERNAL_SERVER_ERROR);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).toBe(0);
+  });
+
+  it('handles specific Sequelize connection acquire timeout error', async () => {
+    const mockConnectionAcquireTimeoutError = new Sequelize
+      .ConnectionAcquireTimeoutError(new Error('Connection acquire timeout error'));
+    await handleErrors(
+      mockRequest,
+      mockResponse,
+      mockConnectionAcquireTimeoutError,
+      mockLogContext,
+    );
+
+    expect(mockResponse.status).toHaveBeenCalledWith(INTERNAL_SERVER_ERROR);
+
+    const requestErrors = await RequestErrors.findAll();
+
+    expect(requestErrors.length).not.toBe(0);
+    expect(requestErrors[0].operation).toBe('SequelizeError');
+  });
 });