feat: add waitForTaskCompletion

Author: PierreJeanjacquot

packages/sdk/src/lib/dataProtectorCore/IExecDataProtectorCore.ts

@@ -17,6 +17,8 @@ import {
   RevokedAccess,
   TransferParams,
   TransferResponse,
+  WaitForTaskCompletionResponse,
+  WaitForTaskCompletionParams,
 } from '../types/index.js';
 import { getGrantedAccess } from './getGrantedAccess.js';
 import { getProtectedData } from './getProtectedData.js';
@@ -27,6 +29,7 @@ import { protectData } from './protectData.js';
 import { revokeAllAccess } from './revokeAllAccess.js';
 import { revokeOneAccess } from './revokeOneAccess.js';
 import { transferOwnership } from './transferOwnership.js';
+import { waitForTaskCompletion } from './waitForTaskCompletion.js';
 
 class IExecDataProtectorCore extends IExecDataProtectorModule {
   async protectData(
@@ -99,6 +102,17 @@ class IExecDataProtectorCore extends IExecDataProtectorModule {
     return getGrantedAccess({ ...args, iexec: this.iexec });
   }
 
+  async waitForTaskCompletion(
+    args: WaitForTaskCompletionParams
+  ): Promise<WaitForTaskCompletionResponse> {
+    await this.init();
+    await isValidProvider(this.iexec);
+    return waitForTaskCompletion({
+      ...args,
+      iexec: this.iexec,
+    });
+  }
+
   async getResultFromCompletedTask(
     args: GetResultFromCompletedTaskParams
   ): Promise<GetResultFromCompletedTaskResponse> {

packages/sdk/src/lib/dataProtectorCore/processProtectedData.ts

@@ -40,6 +40,7 @@ import { IExecConsumer, VoucherInfo } from '../types/internalTypes.js';
 import { getResultFromCompletedTask } from './getResultFromCompletedTask.js';
 import { getWhitelistContract } from './smartContract/getWhitelistContract.js';
 import { isAddressInWhitelist } from './smartContract/whitelistContract.read.js';
+import { waitForTaskCompletion } from './waitForTaskCompletion.js';
 
 export type ProcessProtectedData = typeof processProtectedData;
 
@@ -399,15 +400,11 @@ export const processProtectedData = async <
         taskId: taskId,
       },
     });
-    const taskObservable = await iexec.task.obsTask(taskId, { dealid: dealid });
-    await new Promise((resolve, reject) => {
-      taskObservable.subscribe({
-        next: () => {},
-        error: (e) => {
-          reject(e);
-        },
-        complete: () => resolve(undefined),
-      });
+
+    await waitForTaskCompletion({
+      iexec,
+      dealid,
+      taskId,
     });
 
     vOnStatusUpdate({

packages/sdk/src/lib/dataProtectorCore/waitForTaskCompletion.ts

@@ -0,0 +1,78 @@
+import { WorkflowError } from '../../utils/errors.js';
+import {
+  taskIdSchema,
+  throwIfMissing,
+  validateOnStatusUpdateCallback,
+} from '../../utils/validators.js';
+import {
+  OnStatusUpdateFn,
+  WaitForTaskCompletionParams,
+  WaitForTaskCompletionResponse,
+  WaitForTaskCompletionStatuses,
+} from '../types/index.js';
+import { IExecConsumer } from '../types/internalTypes.js';
+
+export const waitForTaskCompletion = async ({
+  iexec = throwIfMissing(),
+  dealid,
+  taskId,
+  onStatusUpdate = () => {},
+}: IExecConsumer &
+  WaitForTaskCompletionParams): Promise<WaitForTaskCompletionResponse> => {
+  const vTaskId = taskIdSchema()
+    .required()
+    .label('taskId')
+    .validateSync(taskId);
+  const vDealId = taskIdSchema()
+    .required()
+    .label('dealId')
+    .validateSync(dealid);
+  const vOnStatusUpdate =
+    validateOnStatusUpdateCallback<
+      OnStatusUpdateFn<WaitForTaskCompletionStatuses>
+    >(onStatusUpdate);
+
+  try {
+    const taskObservable = await iexec.task.obsTask(vTaskId, {
+      dealid: vDealId,
+    });
+    let status: 'COMPLETED' | 'FAILED' | 'TIMEOUT';
+    let success: boolean;
+    await new Promise((resolve, reject) => {
+      taskObservable.subscribe({
+        next: (data) => {
+          const isDone =
+            data?.task?.statusName === 'COMPLETED' ||
+            data?.task?.statusName === 'FAILED' ||
+            data?.task?.statusName === 'TIMEOUT';
+          if (isDone) {
+            status = data?.task?.statusName as
+              | 'COMPLETED'
+              | 'FAILED'
+              | 'TIMEOUT';
+            success = data?.task?.statusName === 'COMPLETED';
+          }
+          vOnStatusUpdate({
+            title: 'TASK_UPDATED',
+            isDone,
+            payload: {
+              taskId: vTaskId,
+              status: data?.task?.statusName,
+            },
+          });
+        },
+        error: (e) => {
+          reject(e);
+        },
+        complete: () => resolve(undefined),
+      });
+    });
+    return { status, success };
+  } catch (error) {
+    console.error('Error in waitForTaskCompletion:', error);
+    throw new WorkflowError({
+      message: 'Failed to wait for task completion',
+      errorCause: error,
+    });
+  }
+};

packages/sdk/src/lib/types/coreTypes.ts

@@ -208,6 +208,21 @@ export type GrantedAccessResponse = {
   grantedAccess: GrantedAccess[];
 };
 
+// ---------------------waitForTaskCompletion Types------------------------------------
+
+export type WaitForTaskCompletionStatuses = 'TASK_UPDATED';
+
+export type WaitForTaskCompletionParams = {
+  taskId: string;
+  dealid: string;
+  onStatusUpdate?: OnStatusUpdateFn<WaitForTaskCompletionStatuses>;
+};
+
+export type WaitForTaskCompletionResponse = {
+  status: 'COMPLETED' | 'FAILED' | 'TIMEOUT';
+  success: boolean;
+};
+
 // ---------------------GetResultFromCompletedTask Types------------------------------------
 
 export type GetResultFromCompletedTaskStatuses =

packages/sdk/tests/e2e/dataProtectorCore/waitForTaskCompletion.test.ts

@@ -0,0 +1,83 @@
+import { beforeAll, describe, expect, it, jest } from '@jest/globals';
+import { Wallet } from 'ethers';
+import { IExecDataProtectorCore } from '../../../src/index.js';
+import { getTestConfig } from '../../test-utils.js';
+
+describe('dataProtectorCore.waitForTaskCompletion()', () => {
+  let dataProtectorCore: IExecDataProtectorCore;
+
+  beforeAll(async () => {
+    dataProtectorCore = new IExecDataProtectorCore(
+      ...getTestConfig(Wallet.createRandom().privateKey)
+    );
+  });
+
+  it('should return when the task is completed', async () => {
+    // https://explorer.iex.ec/bellecour/task/0xb4655f62bdb841a3b54363b113c4204bf4fee76ab9029f33dc1218ab495970d7
+    const onStatusUpdate = jest.fn();
+    const COMPLETED_TASKID =
+      '0xb4655f62bdb841a3b54363b113c4204bf4fee76ab9029f33dc1218ab495970d7';
+    const COMPLETED_DEALID =
+      '0xb5091be0385c80545cdd12e7c678b96dbb6338cf699324f8f2aa94d3f33f6eda';
+    const res = await dataProtectorCore.waitForTaskCompletion({
+      dealid: COMPLETED_DEALID,
+      taskId: COMPLETED_TASKID,
+      onStatusUpdate,
+    });
+    expect(res).toEqual({ status: 'COMPLETED', success: true });
+    expect(onStatusUpdate).toHaveBeenLastCalledWith({
+      title: 'TASK_UPDATED',
+      isDone: true,
+      payload: {
+        taskId: COMPLETED_TASKID,
+        status: 'COMPLETED',
+      },
+    });
+  });
+
+  it('should return when the task is failed', async () => {
+    // https://explorer.iex.ec/bellecour/task/0x000b16d5517e44ca70744ec156e8374ae525c9ab902169fe01d909370e5778e0
+    const FAILED_TASKID =
+      '0x000b16d5517e44ca70744ec156e8374ae525c9ab902169fe01d909370e5778e0';
+    const FAILED_DEALID =
+      '0xd613b7c6c4a022efe129fd93ce547eba71fc1055e0b42d20b11ad1f3505ad0a5';
+    const onStatusUpdate = jest.fn();
+    const res = await dataProtectorCore.waitForTaskCompletion({
+      dealid: FAILED_DEALID,
+      taskId: FAILED_TASKID,
+      onStatusUpdate,
+    });
+    expect(res).toEqual({ status: 'FAILED', success: false });
+    expect(onStatusUpdate).toHaveBeenLastCalledWith({
+      title: 'TASK_UPDATED',
+      isDone: true,
+      payload: {
+        taskId: FAILED_TASKID,
+        status: 'FAILED',
+      },
+    });
+  });
+
+  it('should return when the task is in timeout', async () => {
+    // https://explorer.iex.ec/bellecour/task/0x012b3d2f21ea3c8c0cc2a40ce06df028df84d1b53b7dae98d5352e79427b93a6
+    const TIMEOUT_TASKID =
+      '0x012b3d2f21ea3c8c0cc2a40ce06df028df84d1b53b7dae98d5352e79427b93a6';
+    const TIMEOUT_DEALID =
+      '0xab15a51de7a3829fca1d3666b81b53e9e9ced0aa71bf20e7ebee1be1bdb3ee33';
+    const onStatusUpdate = jest.fn();
+    const res = await dataProtectorCore.waitForTaskCompletion({
+      dealid: TIMEOUT_DEALID,
+      taskId: TIMEOUT_TASKID,
+      onStatusUpdate,
+    });
+    expect(res).toEqual({ status: 'TIMEOUT', success: false });
+    expect(onStatusUpdate).toHaveBeenLastCalledWith({
+      title: 'TASK_UPDATED',
+      isDone: true,
+      payload: {
+        taskId: TIMEOUT_TASKID,
+        status: 'TIMEOUT',
+      },
+    });
+  });
+});