fix(bpmn-service): add new worker key to handle OptimisticLockingException error (#2245)

* fix(bpmn-service): add new worker key to handle OptimisticLockingException error

* refactor(bpmn-service): rename inProgress to isInProgress

* test(bpmn-service): add test cases for worker implementation provider

* refactor(bpmn-service): remove multiple expect from test case

---------

Co-authored-by: Namandeep  Singh <namandeep.singh>

Author: namandeep-s1ngh

services/bpmn-service/src/__tests__/unit/register-worker.unit.ts

@@ -2,10 +2,10 @@
 //
 // This software is released under the MIT License.
 // https://opensource.org/licenses/MIT
-import {WorkerRegisterFnProvider} from '../../providers/register-worker.service';
+import {AnyObject} from '@loopback/repository';
 import {expect} from '@loopback/testlab';
+import {WorkerRegisterFnProvider} from '../../providers/register-worker.service';
 import {BPMTask, WorkerMap} from '../../types';
-import {AnyObject} from '@loopback/repository';
 
 describe('RegisterWorker unit', () => {
   let map: WorkerMap = {};
@@ -34,23 +34,27 @@ describe('RegisterWorker unit', () => {
           command: task1,
           running: false,
           topic: 'topic1',
+          isInProgress: false,
         },
         {
           command: task2,
           running: false,
           topic: 'topic2',
+          isInProgress: false,
         },
       ],
       workflow2: [
         {
           command: task3,
           running: false,
           topic: 'topic3',
+          isInProgress: false,
         },
         {
           command: task4,
           running: false,
           topic: 'topic4',
+          isInProgress: false,
         },
       ],
     };

services/bpmn-service/src/__tests__/unit/worker-implementation.provider.unit.ts

@@ -0,0 +1,157 @@
+// Copyright (c) 2023 Sourcefuse Technologies
+//
+// This software is released under the MIT License.
+// https://opensource.org/licenses/MIT
+import {AnyObject} from '@loopback/repository';
+import {expect} from '@loopback/testlab';
+import {ICommand, ILogger} from '@sourceloop/core';
+import {Client} from 'camunda-external-task-client-js';
+import sinon from 'sinon';
+import {WorkerImplementationProvider} from '../../providers/worker-implementation.provider';
+import {BPMTask, IWorkflowServiceConfig, WorkerNameCmdPair} from '../../types';
+
+describe('WorkerImplementationProvider Unit Tests', () => {
+  let worker: WorkerNameCmdPair;
+  let provider: WorkerImplementationProvider;
+  let mockClient: sinon.SinonStubbedInstance<Client>;
+  let mockLogger: sinon.SinonStubbedInstance<ILogger>;
+  let clock: sinon.SinonFakeTimers;
+
+  beforeEach(() => {
+    // Mock Logger
+    mockLogger = {
+      info: sinon.stub(),
+      error: sinon.stub(),
+      warn: sinon.stub(),
+      debug: sinon.stub(),
+    } as unknown as sinon.SinonStubbedInstance<ILogger>;
+
+    // Mock Workflow Service Config
+    const config: IWorkflowServiceConfig = {
+      useCustomSequence: true,
+      workflowEngineBaseUrl: 'http://localhost:8080',
+    };
+
+    // Mock Camunda Client
+    mockClient = sinon.createStubInstance(Client);
+    sinon.stub(Client.prototype, 'subscribe').callsFake((_topic, callback) => {
+      callback({
+        task: {
+          variables: {
+            get: sinon.stub(),
+            getTyped: sinon.stub(),
+            getAll: sinon.stub(),
+            getAllTyped: sinon.stub(),
+            set: sinon.stub(),
+            setTyped: sinon.stub(),
+            setAll: sinon.stub(),
+            setAllTyped: sinon.stub(),
+          },
+        },
+        taskService: {
+          complete: sinon.stub(),
+          handleFailure: sinon.stub(),
+          handleBpmnError: sinon.stub(),
+          extendLock: sinon.stub(),
+          unlock: sinon.stub(),
+        },
+      });
+      return {unsubscribe: sinon.stub()};
+    });
+
+    // Use fake timers to control setTimeout/setInterval
+    clock = sinon.useFakeTimers();
+
+    provider = new WorkerImplementationProvider(config, mockLogger);
+
+    const mockCommand: ICommand = {
+      parameters: {},
+      execute: sinon.stub().resolves({}), // Mock execute() to return a resolved promise
+    };
+
+    // Initialize Worker
+    worker = {
+      topic: 'test-topic',
+      command: new BPMTask<AnyObject, AnyObject>(mockCommand),
+      running: false,
+      isInProgress: false,
+    };
+  });
+
+  afterEach(() => {
+    clock.restore();
+    sinon.restore();
+  });
+
+  it('should initialize worker with isInProgress as false', async () => {
+    const implementationFn = provider.value();
+    await implementationFn(worker);
+    expect(worker.isInProgress).to.be.false();
+  });
+
+  it('should set isInProgress to true when processing starts', async () => {
+    const implementationFn = provider.value();
+
+    // Modify existing stub instead of re-stubbing
+    (worker.command.command.execute as sinon.SinonStub).callsFake(async () => {
+      // verify isInProgress is true during worker execution flow
+      expect(worker.isInProgress).to.be.true();
+      return {};
+    });
+
+    await implementationFn(worker);
+  });
+
+  it('should return early if worker is already in progress', async () => {
+    worker.isInProgress = true;
+    provider.client = mockClient as unknown as Client;
+
+    const implementationFn = provider.value();
+    await implementationFn(worker);
+
+    const operationStub = sinon.stub(worker.command, 'operation');
+    await implementationFn(worker);
+
+    expect(operationStub.called).to.be.false();
+  });
+
+  it('should set isInProgress to false when task completes', async () => {
+    const implementationFn = provider.value();
+    await implementationFn(worker);
+
+    // Properly stub the 'operation' method
+    const operationStub = sinon.stub(worker.command, 'operation');
+
+    // Simulate task completion by invoking the callback with an empty object
+    operationStub.callsFake((_data, done) => {
+      if (done) done({});
+    });
+
+    expect(worker.isInProgress).to.be.false();
+
+    // Restore the stub after the test
+    operationStub.restore();
+  });
+
+  it('should set isInProgress to false if an error occurs during polling', async () => {
+    sinon.restore();
+    const onStub = sinon
+      .stub(mockClient, 'on')
+      .callsArgWith(1, new Error('Test Poll Error'));
+
+    const implementationFn = provider.value();
+    await implementationFn(worker);
+
+    // Simulate poll error
+    /* eslint-disable @typescript-eslint/no-explicit-any */
+    (provider.client as any).emit(
+      'poll:error',
+      new Error('Simulated Polling Error'),
+    );
+
+    expect(worker.isInProgress).to.be.false();
+    expect(worker.running).to.be.false();
+
+    onStub.restore();
+  });
+});

services/bpmn-service/src/providers/register-worker.service.ts

@@ -21,9 +21,16 @@ export class WorkerRegisterFnProvider implements Provider<WorkerRegisterFn> {
         workerMap = {};
       }
       if (workerMap[workflowName]) {
-        workerMap[workflowName].push({topic, command, running: false});
+        workerMap[workflowName].push({
+          topic,
+          command,
+          running: false,
+          isInProgress: false,
+        });
       } else {
-        workerMap[workflowName] = [{topic, command, running: false}];
+        workerMap[workflowName] = [
+          {topic, command, running: false, isInProgress: false},
+        ];
         this.workerMapSetter(workerMap);
       }
     };

services/bpmn-service/src/providers/worker-implementation.provider.ts

@@ -3,11 +3,11 @@
 // This software is released under the MIT License.
 // https://opensource.org/licenses/MIT
 import {inject, Provider} from '@loopback/context';
-import {IWorkflowServiceConfig, WorkerImplementationFn} from '../types';
-import {Client, logger} from 'camunda-external-task-client-js';
-import {WorkflowServiceBindings} from '../keys';
 import {AnyObject} from '@loopback/repository';
 import {ILogger, LOGGER} from '@sourceloop/core';
+import {Client, logger} from 'camunda-external-task-client-js';
+import {WorkflowServiceBindings} from '../keys';
+import {IWorkflowServiceConfig, WorkerImplementationFn} from '../types';
 
 export class WorkerImplementationProvider
   implements Provider<WorkerImplementationFn>
@@ -31,22 +31,27 @@ export class WorkerImplementationProvider
   value(): WorkerImplementationFn {
     return async worker => {
       if (this.client) {
+        worker.isInProgress = false;
         worker.running = true;
         const subscription = this.client.subscribe(
           worker.topic,
           ({task, taskService}) => {
+            if (worker.isInProgress) return;
+            worker.isInProgress = true;
             worker.command.operation(
               {task, taskService},
               (result: AnyObject) => {
                 if (result) {
                   this.ilogger.info(`Worker task completed - ${worker.topic}`);
                 }
+                worker.isInProgress = false;
               },
             );
           },
         );
         this.client.on('poll:error', () => {
           worker.running = false;
+          worker.isInProgress = false;
           subscription.unsubscribe();
         });
       } else {

services/bpmn-service/src/types/types.ts

@@ -4,8 +4,8 @@
 // https://opensource.org/licenses/MIT
 import {AnyObject} from '@loopback/repository';
 import {IServiceConfig} from '@sourceloop/core';
+import {Workflow, WorkflowDto, WorkflowVersion} from '../models';
 import {BPMTask} from './bpm-task';
-import {Workflow, WorkflowVersion, WorkflowDto} from '../models';
 
 /* eslint-disable @typescript-eslint/no-explicit-any */
 export interface ICommand {
@@ -57,6 +57,7 @@ export type WorkerNameCmdPair<T = AnyObject, R = AnyObject> = {
   topic: string;
   command: BPMTask<T, R>;
   running: boolean;
+  isInProgress: boolean;
 };
 
 export type WorkerImplementationFn<T = AnyObject, R = AnyObject> = (

services/task-service/src/lifecycle-observers/command.observer.ts

@@ -34,6 +34,7 @@ export class CommandObserver implements LifeCycleObserver {
         topic: command.topic,
         command: new BPMTask(command),
         running: false,
+        isInProgress: false,
       });
     }
     this.logger.debug('Commands registered');