fix(worker): reduce and refactor number of logs

apps/api/src/app/events/e2e/send-message-push.e2e.ts

@@ -76,12 +76,10 @@ describe('Trigger event - Send Push Notification - /v1/events/trigger (POST) #no
         _environmentId: session.environment._id,
       });
 
-      expect(executionDetails.length).to.equal(8);
+      expect(executionDetails.length).to.equal(7);
       const noActiveChannel = executionDetails.find((ex) => ex.detail === DetailEnum.SUBSCRIBER_NO_ACTIVE_CHANNEL);
       expect(noActiveChannel).to.be.ok;
       expect(noActiveChannel?.providerId).to.equal('fcm');
-      const genericError = executionDetails.find((ex) => ex.detail === DetailEnum.NOTIFICATION_ERROR);
-      expect(genericError).to.be.ok;
     });
 
     it('should not create any message if subscriber has configured two providers without device tokens', async () => {

apps/worker/src/app/workflow/services/active-jobs-metric.service.ts

@@ -115,8 +115,6 @@ export class ActiveJobsMetricService {
           // eslint-disable-next-line no-promise-executor-return
           return resolve();
         } catch (error) {
-          Logger.error({ error }, 'Error occurred while processing metrics', LOG_CONTEXT);
-
           // eslint-disable-next-line no-promise-executor-return
           return reject(error);
         }

apps/worker/src/app/workflow/usecases/run-job/run-job.usecase.ts

@@ -110,7 +110,6 @@ export class RunJob {
         await this.jobRepository.updateStatus(job._environmentId, job._id, JobStatusEnum.COMPLETED);
       }
     } catch (error: any) {
-      Logger.error({ error }, `Running job ${job._id} has thrown an error`, LOG_CONTEXT);
       if (job.step.shouldStopOnFail || this.shouldBackoff(error)) {
         shouldQueueNextJob = false;
       }

apps/worker/src/app/workflow/usecases/send-message/digest/digest.usecase.ts

@@ -122,7 +122,7 @@ export class Digest extends SendMessageType {
 
     if (!currentJob) {
       const message = `Digest job ${command.jobId} is not found`;
-      Logger.error(message, LOG_CONTEXT);
+      Logger.log(message, LOG_CONTEXT);
       throw new PlatformException(message);
     }
 

apps/worker/src/app/workflow/usecases/send-message/digest/get-digest-events.usecase.ts

@@ -56,7 +56,7 @@ export abstract class GetDigestEvents {
       );
 
       const message = `Trigger job for jobId ${currentJob._id} is not found`;
-      Logger.error(message, LOG_CONTEXT);
+      Logger.log(message, LOG_CONTEXT);
       throw new PlatformException(message);
     }
 

apps/worker/src/app/workflow/usecases/send-message/send-message-push.usecase.ts

@@ -134,10 +134,22 @@ export class SendMessagePush extends SendMessageBase {
     for (const channel of pushChannels) {
       const { deviceTokens } = channel.credentials || {};
 
-      const [isChannelMissingDeviceTokens, integration] = await Promise.all([
-        this.isChannelMissingDeviceTokens(channel, command),
-        this.getSubscriberIntegration(channel, command),
-      ]);
+      let isChannelMissingDeviceTokens;
+      let integration;
+      try {
+        [isChannelMissingDeviceTokens, integration] = await Promise.all([
+          this.isChannelMissingDeviceTokens(channel, command),
+          this.getSubscriberIntegration(channel, command),
+        ]);
+      } catch (error) {
+        integrationsWithErrors += 1;
+        Logger.error(
+          { jobId: command.jobId },
+          [`Error processing channel for jobId ${command.jobId}`, error.message || error.toString()].join(' '),
+          LOG_CONTEXT
+        );
+        continue;
+      }
 
       // We avoid to send a message if subscriber has not an integration or if the subscriber has no device tokens for said integration
       if (!deviceTokens || !integration || isChannelMissingDeviceTokens) {
@@ -263,24 +275,19 @@ export class SendMessagePush extends SendMessageBase {
       raw?: string;
     }
   ): Promise<void> {
-    // We avoid to throw the errors to be able to execute all actions in the loop
-    try {
-      await this.executionLogRoute.execute(
-        ExecutionLogRouteCommand.create({
-          ...ExecutionLogRouteCommand.getDetailsFromJob(job),
-          detail,
-          source: ExecutionDetailsSourceEnum.INTERNAL,
-          status: ExecutionDetailsStatusEnum.FAILED,
-          isTest: false,
-          isRetry: false,
-          ...(contextData?.providerId && { providerId: contextData.providerId }),
-          ...(contextData?.messageId && { messageId: contextData.messageId }),
-          ...(contextData?.raw && { raw: contextData.raw }),
-        })
-      );
-    } catch (error) {
-      Logger.error(error, 'Error creating execution details error');
-    }
+    await this.executionLogRoute.execute(
+      ExecutionLogRouteCommand.create({
+        ...ExecutionLogRouteCommand.getDetailsFromJob(job),
+        detail,
+        source: ExecutionDetailsSourceEnum.INTERNAL,
+        status: ExecutionDetailsStatusEnum.FAILED,
+        isTest: false,
+        isRetry: false,
+        ...(contextData?.providerId && { providerId: contextData.providerId }),
+        ...(contextData?.messageId && { messageId: contextData.messageId }),
+        ...(contextData?.raw && { raw: contextData.raw }),
+      })
+    );
   }
 
   private async sendMessage(
@@ -336,7 +343,15 @@ export class SendMessagePush extends SendMessageBase {
 
       const raw = JSON.stringify(e) !== JSON.stringify({}) ? JSON.stringify(e) : JSON.stringify(e.message);
 
-      await this.sendProviderError(command.job, message._id, raw);
+      try {
+        await this.sendProviderError(command.job, message._id, raw);
+      } catch (err) {
+        Logger.error(
+          { jobId: command.jobId },
+          [`Error sending provider error for jobId ${command.jobId}`, err.message || err.toString()].join(' '),
+          LOG_CONTEXT
+        );
+      }
 
       return { success: false, error: e };
     }