feat(zeebe): support Job Corrections in complete command fixes #347

docker/.env

@@ -2,8 +2,8 @@
 # CAMUNDA_CONNECTORS_VERSION=0.23.2
 CAMUNDA_CONNECTORS_VERSION=8.5.0
 CAMUNDA_OPTIMIZE_VERSION=8.5.0
-CAMUNDA_PLATFORM_VERSION=8.6.0
-CAMUNDA_ZEEBE_VERSION=8.6.3
+CAMUNDA_PLATFORM_VERSION=8.7.0-alpha2
+CAMUNDA_ZEEBE_VERSION=8.7.0-alpha2
 CAMUNDA_WEB_MODELER_VERSION=8.5.0
 ELASTIC_VERSION=8.9.0
 KEYCLOAK_SERVER_VERSION=22.0.3

src/__tests__/testdata/Worker-JobResult.bpmn

@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:zeebe="http://camunda.org/schema/zeebe/1.0" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" xmlns:modeler="http://camunda.org/schema/modeler/1.0" id="Definitions_1ob4kc6" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="5.30.0" modeler:executionPlatform="Camunda Cloud" modeler:executionPlatformVersion="8.6.0">
+  <bpmn:process id="job-correction" name="Job Correction Test" isExecutable="true">
+    <bpmn:startEvent id="StartEvent_1" name="Start Job Correction Test">
+      <bpmn:outgoing>Flow_107kyon</bpmn:outgoing>
+    </bpmn:startEvent>
+    <bpmn:sequenceFlow id="Flow_107kyon" sourceRef="StartEvent_1" targetRef="Activity_01g5b4t" />
+    <bpmn:serviceTask id="Activity_01g5b4t" name="Job Correction Job">
+      <bpmn:extensionElements>
+        <zeebe:taskDefinition type="job-correction" />
+      </bpmn:extensionElements>
+      <bpmn:incoming>Flow_107kyon</bpmn:incoming>
+      <bpmn:outgoing>Flow_0y0j0w9</bpmn:outgoing>
+    </bpmn:serviceTask>
+    <bpmn:endEvent id="Event_1hmqn2y" name="Job Correction Test Completed Successfully">
+      <bpmn:incoming>Flow_0y0j0w9</bpmn:incoming>
+    </bpmn:endEvent>
+    <bpmn:sequenceFlow id="Flow_0y0j0w9" sourceRef="Activity_01g5b4t" targetRef="Event_1hmqn2y" />
+  </bpmn:process>
+  <bpmndi:BPMNDiagram id="BPMNDiagram_1">
+    <bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="job-correction">
+      <bpmndi:BPMNShape id="StartEvent_1_di" bpmnElement="StartEvent_1">
+        <dc:Bounds x="182" y="102" width="36" height="36" />
+        <bpmndi:BPMNLabel>
+          <dc:Bounds x="163" y="145" width="74" height="27" />
+        </bpmndi:BPMNLabel>
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="Activity_0s25xnu_di" bpmnElement="Activity_01g5b4t">
+        <dc:Bounds x="270" y="80" width="100" height="80" />
+        <bpmndi:BPMNLabel />
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNShape id="Event_1hmqn2y_di" bpmnElement="Event_1hmqn2y">
+        <dc:Bounds x="422" y="102" width="36" height="36" />
+        <bpmndi:BPMNLabel>
+          <dc:Bounds x="402" y="145" width="77" height="40" />
+        </bpmndi:BPMNLabel>
+      </bpmndi:BPMNShape>
+      <bpmndi:BPMNEdge id="Flow_107kyon_di" bpmnElement="Flow_107kyon">
+        <di:waypoint x="218" y="120" />
+        <di:waypoint x="270" y="120" />
+      </bpmndi:BPMNEdge>
+      <bpmndi:BPMNEdge id="Flow_0y0j0w9_di" bpmnElement="Flow_0y0j0w9">
+        <di:waypoint x="370" y="120" />
+        <di:waypoint x="422" y="120" />
+      </bpmndi:BPMNEdge>
+    </bpmndi:BPMNPlane>
+  </bpmndi:BPMNDiagram>
+</bpmn:definitions>

src/__tests__/zeebe/integration/Worker-CompleteWithResult.spec.ts

@@ -0,0 +1,62 @@
+import { restoreZeebeLogging, suppressZeebeLogging } from '../../../lib'
+import { ZeebeGrpcClient } from '../../../zeebe'
+import { cancelProcesses } from '../../../zeebe/lib/cancelProcesses'
+import { CreateProcessInstanceResponse } from '../../../zeebe/lib/interfaces-grpc-1.0'
+
+jest.setTimeout(30000)
+suppressZeebeLogging()
+
+const zbc = new ZeebeGrpcClient()
+let wf: CreateProcessInstanceResponse | undefined
+let processDefinitionKey1: string
+let bpmnProcessId1: string
+
+beforeAll(async () => {
+	const res1 = await zbc.deployResource({
+		processFilename: './src/__tests__/testdata/Worker-JobResult.bpmn',
+	})
+	;({
+		processDefinitionKey: processDefinitionKey1,
+		bpmnProcessId: bpmnProcessId1,
+	} = res1.deployments[0].process)
+	await cancelProcesses(processDefinitionKey1)
+})
+
+afterEach(async () => {
+	if (wf?.processInstanceKey) {
+		await zbc.cancelProcessInstance(wf.processInstanceKey).catch((e) => e)
+	}
+})
+
+afterAll(async () => {
+	await zbc.close()
+	await cancelProcesses(processDefinitionKey1)
+	restoreZeebeLogging()
+})
+
+test('Can complete a task with job corrections', (done) => {
+	zbc
+		.createProcessInstance({
+			bpmnProcessId: bpmnProcessId1,
+			variables: {},
+		})
+		.then((res) => {
+			wf = res
+			zbc.createWorker({
+				taskType: 'job-correction',
+				taskHandler: async (job) => {
+					expect(job.processInstanceKey).toBe(wf?.processInstanceKey)
+					const res1 = await job.completeWithJobResult({
+						variables: {},
+						result: {
+							denied: true,
+						},
+					})
+					// @TODO: correction interface
+					done(null)
+					return res1
+				},
+				loglevel: 'NONE',
+			})
+		})
+})

src/proto/zeebe.proto

@@ -6,7 +6,7 @@ option java_package = "io.camunda.zeebe.gateway.protocol";
 option go_package = "./;pb";
 
 // For a more complete documentation, refer to Zeebe documentation at:
-// https://docs.camunda.io/docs/reference/grpc
+// https://docs.camunda.io/docs/apis-tools/zeebe-api/gateway-service/
 
 message StreamActivatedJobsRequest {
   // the job type, as defined in the BPMN process (e.g. <zeebe:taskDefinition
@@ -101,6 +101,46 @@ message CompleteJobRequest {
   int64 jobKey = 1;
   // a JSON document representing the variables in the current task scope
   string variables = 2;
+  // The result of the completed job as determined by the worker.
+  optional JobResult result = 3;
+}
+
+message JobResult{
+  // Indicates whether the worker denies the work, i.e. explicitly doesn't approve it.
+  // For example, a Task Listener can deny the completion of a task by setting this flag to true.
+  // In this example, the completion of a task is represented by a job that the worker can complete as denied.
+  // As a result, the completion request is rejected and the task remains active.
+  // Defaults to false.
+  optional bool denied = 1;
+  // Attributes that were corrected by the worker.
+  // The following attributes can be corrected, additional attributes will be ignored:
+  //   * `assignee` - reset by providing an empty String
+  //   * `dueDate` - reset by providing an empty String
+  //   * `followUpDate` - reset by providing an empty String
+  //   * `candidateGroups` - reset by providing an empty list
+  //   * `candidateUsers` - reset by providing an empty list
+  //   * `priority` - minimum 0, maximum 100, default 50
+  //  Omitting any of the attributes will preserve the persisted attribute's value.
+  optional JobResultCorrections corrections = 2;
+}
+
+message JobResultCorrections {
+  // The assignee of the task.
+  optional string assignee = 1;
+  // The due date of the task.
+  optional string dueDate = 2;
+  // The follow-up date of the task.
+  optional string followUpDate = 3;
+  // The list of candidate users of the task.
+  optional StringList candidateUsers = 4;
+  // The list of candidate groups of the task.
+  optional StringList candidateGroups = 5;
+  // The priority of the task.
+  optional int32 priority = 6;
+}
+
+message StringList {
+  repeated string values = 1;
 }
 
 message CompleteJobResponse {

src/zeebe/lib/ZBStreamWorker.ts

@@ -15,7 +15,11 @@ import {
 	ZBGrpc,
 	ZBWorkerTaskHandler,
 } from './interfaces-1.0'
-import { ActivatedJob, StreamActivatedJobsRequest } from './interfaces-grpc-1.0'
+import {
+	ActivatedJob,
+	CompleteJobRequest,
+	StreamActivatedJobsRequest,
+} from './interfaces-grpc-1.0'
 
 import { parseVariablesAndCustomHeadersToJSON } from '.'
 
@@ -195,13 +199,30 @@ You should call only one job action method in the worker handler. This is a bug
 				})
 			}
 
+		const correctJob =
+			(job: Job) => (req: Pick<CompleteJobRequest, 'result' | 'variables'>) =>
+				this.completeJob(
+					job.key,
+					{
+						result: req.result,
+						variables: req.variables,
+					},
+					taskType
+				)
+
 		const fail = failJob(thisJob)
 		const succeed = succeedJob(thisJob)
+		const completeWithResult = correctJob(thisJob)
+
 		return {
 			cancelWorkflow: cancelWorkflow(thisJob),
 			complete: errorMsgOnPriorMessageCall('job.complete', succeed),
 			error: errorMsgOnPriorMessageCall('error', errorJob(thisJob)),
 			fail: errorMsgOnPriorMessageCall('job.fail', fail),
+			completeWithJobResult: errorMsgOnPriorMessageCall(
+				'job.completeWithJobResult',
+				completeWithResult
+			),
 			forward: errorMsgOnPriorMessageCall('job.forward', () => {
 				return JOB_ACTION_ACKNOWLEDGEMENT
 			}),

src/zeebe/lib/ZBWorkerBase.ts

@@ -20,6 +20,8 @@ import * as ZB from './interfaces-1.0'
 import {
 	ActivateJobsRequest,
 	ActivateJobsResponse,
+	CompleteJobRequest,
+	JobResult,
 } from './interfaces-grpc-1.0'
 import { ZBClientOptions } from './interfaces-published-contract'
 
@@ -340,6 +342,14 @@ You should call only one job action method in the worker handler. This is a bug
 			(completedVariables?: T) =>
 				this.completeJob(job.key, completedVariables ?? {})
 
+		const correctJob =
+			(job: ZB.Job<WorkerInputVariables, CustomHeaderShape>) =>
+			(req: Pick<CompleteJobRequest, 'result' | 'variables'>) =>
+				this.completeJob(job.key, {
+					result: req.result,
+					variables: req.variables,
+				})
+
 		const errorJob =
 			(job: ZB.Job<WorkerInputVariables, CustomHeaderShape>) =>
 			(e: string | ZB.ErrorJobWithVariables, errorMessage: string = '') => {
@@ -362,9 +372,14 @@ You should call only one job action method in the worker handler. This is a bug
 
 		const fail = failJob(thisJob)
 		const succeed = succeedJob(thisJob)
+		const completeWithResult = correctJob(thisJob)
 		return {
 			cancelWorkflow: cancelWorkflow(thisJob),
 			complete: errorMsgOnPriorMessageCall('job.complete', succeed),
+			completeWithJobResult: errorMsgOnPriorMessageCall(
+				'job.completeWithJobResult',
+				completeWithResult
+			),
 			error: errorMsgOnPriorMessageCall('error', errorJob(thisJob)),
 			fail: errorMsgOnPriorMessageCall('job.fail', fail),
 			forward: errorMsgOnPriorMessageCall('job.forward', () => {
@@ -402,11 +417,16 @@ You should call only one job action method in the worker handler. This is a bug
 			})
 	}
 
-	private completeJob(jobKey: string, completedVariables = {}) {
+	private completeJob(
+		jobKey: string,
+		completedVariables = {},
+		result: JobResult = {}
+	) {
 		return this.zbClient
 			.completeJob({
 				jobKey,
 				variables: completedVariables,
+				result,
 			})
 			.then((res) => {
 				this.logger.logDebug(`Completed job ${jobKey} for ${this.taskType}`)

src/zeebe/lib/interfaces-1.0.ts

@@ -178,6 +178,12 @@ export interface JobCompletionInterface<WorkerOutputVariables> {
 	complete: (
 		updatedVariables?: WorkerOutputVariables
 	) => Promise<JOB_ACTION_ACKNOWLEDGEMENT>
+	/**
+	 * Complete the job with success, and a job result that may contain corrections. Since Camunda 8.7.
+	 */
+	completeWithJobResult: (
+		req: Pick<CompleteJobRequest, 'result' | 'variables'>
+	) => Promise<JOB_ACTION_ACKNOWLEDGEMENT>
 	/**
 	 * Fail the job with an informative message as to the cause. Optionally, pass in a
 	 * value remaining retries. If no value is passed for retries then the current retry

src/zeebe/lib/interfaces-grpc-1.0.ts

@@ -506,6 +506,45 @@ export interface ThrowErrorRequest {
 export interface CompleteJobRequest<Variables = IProcessVariables> {
 	readonly jobKey: string
 	variables: Variables
+	/** Since Camunda 8.7 */
+	result?: JobResult
+}
+
+/**
+ * Since Camunda 8.7
+ */
+export interface JobResult {
+	// Indicates whether the worker denies the work, i.e. explicitly doesn't approve it.
+	// For example, a Task Listener can deny the completion of a task by setting this flag to true.
+	// In this example, the completion of a task is represented by a job that the worker can complete as denied.
+	// As a result, the completion request is rejected and the task remains active.
+	// Defaults to false.
+	denied?: boolean
+	// Attributes that were corrected by the worker.
+	// The following attributes can be corrected, additional attributes will be ignored:
+	//   * `assignee` - reset by providing an empty String
+	//   * `dueDate` - reset by providing an empty String
+	//   * `followUpDate` - reset by providing an empty String
+	//   * `candidateGroups` - reset by providing an empty list
+	//   * `candidateUsers` - reset by providing an empty list
+	//   * `priority` - minimum 0, maximum 100, default 50
+	//  Omitting any of the attributes will preserve the persisted attribute's value.
+	corrections?: JobResultCorrections
+}
+
+export interface JobResultCorrections {
+	// The assignee of the task.
+	assignee?: string
+	// The due date of the task.
+	dueDate?: string
+	// The follow-up date of the task.
+	followUpDate?: string
+	// The list of candidate users of the task.
+	candidateUsers?: { values: string[] }
+	// The list of candidate groups of the task.
+	candidateGroups?: { values: string[] }
+	// The priority of the task.
+	priority?: number
 }
 
 interface SetVariablesRequestBase {

src/zeebe/zb/ZBWorker.ts

@@ -49,11 +49,7 @@ export class ZBWorker<
 			)(
 				{
 					...job,
-					cancelWorkflow: workerCallback.cancelWorkflow,
-					complete: workerCallback.complete,
-					fail: workerCallback.fail,
-					error: workerCallback.error,
-					forward: workerCallback.forward,
+					...workerCallback,
 				},
 				this
 			)