feat(iapp)!: support bulk processing (#82)

BREAKING CHANGE: result file renamed to `result.json`; single protectedData result file now contains `{"success": <boolean>, "protectedData"?: <address>, "error"?: <string> }`; upon error, the iapp will now exit 0 and output `"success": false` and `"error": <string>` rather than falling the task.

Co-authored-by: Pierre Jeanjacquot <[email protected]>

Author: abbesBenayache

dapp/README.md

@@ -12,7 +12,11 @@ fill in the environment variables:
 
 - **IEXEC_IN**: The path to the input directory on your local machine where the unencrypted data .zip file will be stored. This file contains the telegram chat ID address to which the message will be sent.
 - **IEXEC_OUT**: The path on your local machine where the result of the Dapp execution will be written.
-- **IEXEC_DATASET_FILENAME**: The name of the data file that you place in the **IEXEC_IN** directory.
+- data to process
+  - either **IEXEC_DATASET_FILENAME**: The name of the data file that you place in the **IEXEC_IN** directory.
+  - or multiple datasets using:
+    - **IEXEC_BULK_SLICE_SIZE**
+    - **IEXEC_DATASET\_\<index\>\_FILENAME**: The name of the data file for dataset at index <index> (starting from 1) that you place in the **IEXEC_IN** directory.
 - **IEXEC_APP_DEVELOPER_SECRET**: A JSON string with the following keys:
   - **TELEGRAM_BOT_TOKEN**: The API key of the telegram bot used to send the message.
 - **IEXEC_REQUESTER_SECRET_1**: A JSON string with the following keys:

dapp/package-lock.json

@@ -33,7 +33,7 @@
     "eslint-config-prettier": "^9.0.0",
     "eslint-plugin-import": "^2.28.1",
     "eslint-plugin-sonarjs": "^0.21.0",
-    "iexec": "^8.2.1",
+    "iexec": "^8.22.0",
     "jest": "^29.7.0",
     "prettier": "^2.8.8"
   }

dapp/package.json

@@ -16,40 +16,43 @@ export const downloadEncryptedContent = async (
     }
     const arrayBuffer = await res.arrayBuffer();
     return new Uint8Array(arrayBuffer);
-  } catch (error) {
-    console.error('Error downloading encrypted content:', error);
-    throw error;
+  } catch {
+    throw new Error('Failed to download encrypted content');
   }
 };
 
 export const decryptContent = (encryptedContent, encryptionKey) => {
-  const ivBytes = encryptedContent.slice(0, 16);
-  let ciphertextBytes = encryptedContent.slice(16);
-  const key = forge.util.createBuffer(Buffer.from(encryptionKey, 'base64'));
-  const decipher = forge.cipher.createDecipher('AES-CBC', key);
-
-  decipher.start({ iv: forge.util.createBuffer(ivBytes) });
+  try {
+    const ivBytes = encryptedContent.slice(0, 16);
+    let ciphertextBytes = encryptedContent.slice(16);
+    const key = forge.util.createBuffer(Buffer.from(encryptionKey, 'base64'));
+    const decipher = forge.cipher.createDecipher('AES-CBC', key);
+
+    decipher.start({ iv: forge.util.createBuffer(ivBytes) });
+
+    const CHUNK_SIZE = 10 * 1000 * 1000;
+    let decryptedBuffer = Buffer.from([]);
+
+    while (ciphertextBytes.length > 0) {
+      // flush the decipher buffer
+      decryptedBuffer = Buffer.concat([
+        decryptedBuffer,
+        Buffer.from(decipher.output.getBytes(), 'binary'),
+      ]);
+      const chunk = ciphertextBytes.slice(0, CHUNK_SIZE);
+      ciphertextBytes = ciphertextBytes.slice(CHUNK_SIZE);
+      decipher.update(forge.util.createBuffer(chunk));
+    }
 
-  const CHUNK_SIZE = 10 * 1000 * 1000;
-  let decryptedBuffer = Buffer.from([]);
+    decipher.finish();
 
-  while (ciphertextBytes.length > 0) {
-    // flush the decipher buffer
     decryptedBuffer = Buffer.concat([
       decryptedBuffer,
       Buffer.from(decipher.output.getBytes(), 'binary'),
     ]);
-    const chunk = ciphertextBytes.slice(0, CHUNK_SIZE);
-    ciphertextBytes = ciphertextBytes.slice(CHUNK_SIZE);
-    decipher.update(forge.util.createBuffer(chunk));
-  }
 
-  decipher.finish();
-
-  decryptedBuffer = Buffer.concat([
-    decryptedBuffer,
-    Buffer.from(decipher.output.getBytes(), 'binary'),
-  ]);
-
-  return decryptedBuffer.toString();
+    return decryptedBuffer.toString();
+  } catch {
+    throw new Error('Failed to decrypt content');
+  }
 };

dapp/src/decryptContent.js

@@ -1,5 +1,5 @@
-import { IExecDataProtectorDeserializer } from '@iexec/dataprotector-deserializer';
 import { promises as fs } from 'fs';
+import { IExecDataProtectorDeserializer } from '@iexec/dataprotector-deserializer';
 import { decryptContent, downloadEncryptedContent } from './decryptContent.js';
 import sendTelegram from './telegramService.js';
 import {
@@ -9,79 +9,148 @@ import {
   validateWorkerEnv,
 } from './validation.js';
 
-async function writeTaskOutput(path, message) {
+async function processProtectedData({
+  index,
+  IEXEC_IN,
+  appDeveloperSecret,
+  requesterSecret,
+}) {
+  const datasetFilename =
+    index > 0
+      ? process.env[`IEXEC_DATASET_${index}_FILENAME`]
+      : process.env.IEXEC_DATASET_FILENAME;
+  const result = { index, protectedData: datasetFilename };
   try {
-    await fs.writeFile(path, message);
-    console.log(`File successfully written at path: ${path}`);
-  } catch {
-    console.error('Failed to write Task Output');
-    process.exit(1);
+    let protectedData;
+    try {
+      const deserializerConfig = datasetFilename
+        ? { protectedDataPath: `${IEXEC_IN}/${datasetFilename}` }
+        : {};
+
+      const deserializer = new IExecDataProtectorDeserializer(
+        deserializerConfig
+      );
+      protectedData = {
+        chatId: await deserializer.getValue('telegram_chatId', 'string'),
+      };
+    } catch (e) {
+      throw Error(`Failed to parse ProtectedData ${index}: ${e.message}`);
+    }
+
+    validateProtectedData(protectedData);
+
+    const encryptedTelegramContent = await downloadEncryptedContent(
+      requesterSecret.telegramContentMultiAddr
+    );
+
+    const telegramContent = decryptContent(
+      encryptedTelegramContent,
+      requesterSecret.telegramContentEncryptionKey
+    );
+
+    await sendTelegram({
+      chatId: protectedData.chatId,
+      message: telegramContent,
+      botToken: appDeveloperSecret.TELEGRAM_BOT_TOKEN,
+      senderName: requesterSecret.senderName,
+    });
+    result.success = true;
+  } catch (e) {
+    result.success = false;
+    result.error = e.message;
   }
+  console.log(`Protected data ${index} processed:`, result);
+  return result;
 }
 
 async function start() {
-  const { IEXEC_OUT, IEXEC_APP_DEVELOPER_SECRET, IEXEC_REQUESTER_SECRET_1 } =
-    process.env;
+  const {
+    IEXEC_OUT,
+    IEXEC_APP_DEVELOPER_SECRET,
+    IEXEC_REQUESTER_SECRET_1,
+    IEXEC_IN,
+    IEXEC_BULK_SLICE_SIZE,
+  } = process.env;
 
-  // Check worker env
   const workerEnv = validateWorkerEnv({ IEXEC_OUT });
-  // Parse the app developer secret environment variable
-  let appDeveloperSecret;
-  try {
-    appDeveloperSecret = JSON.parse(IEXEC_APP_DEVELOPER_SECRET);
-  } catch {
-    throw Error('Failed to parse the developer secret');
-  }
-  appDeveloperSecret = validateAppSecret(appDeveloperSecret);
 
-  // Parse the requester secret environment variable
-  let requesterSecret;
+  let result; // { success: boolean, error?: string ,  protectedData?: string,  results?: { index: number, protectedData: string, success: boolean, error?: string }[] }
   try {
-    requesterSecret = IEXEC_REQUESTER_SECRET_1
-      ? JSON.parse(IEXEC_REQUESTER_SECRET_1)
-      : {};
-  } catch {
-    throw Error('Failed to parse requester secret');
-  }
-  requesterSecret = validateRequesterSecret(requesterSecret);
+    let appDeveloperSecret;
+    try {
+      appDeveloperSecret = JSON.parse(IEXEC_APP_DEVELOPER_SECRET);
+    } catch {
+      throw Error('Failed to parse the developer secret');
+    }
+    appDeveloperSecret = validateAppSecret(appDeveloperSecret);
 
-  // Parse the protected data and get the requester secret (chatId)
-  let protectedData;
-  try {
-    const deserializer = new IExecDataProtectorDeserializer();
-    protectedData = {
-      chatId: await deserializer.getValue('telegram_chatId', 'string'),
-    };
-  } catch (e) {
-    throw Error(`Failed to parse ProtectedData: ${e.message}`);
-  }
-  // Validate the protected data (chatId)
-  validateProtectedData(protectedData);
+    let requesterSecret;
+    try {
+      requesterSecret = JSON.parse(IEXEC_REQUESTER_SECRET_1);
+    } catch {
+      throw Error('Failed to parse requester secret');
+    }
 
-  const encryptedTelegramContent = await downloadEncryptedContent(
-    requesterSecret.telegramContentMultiAddr
-  );
+    requesterSecret = validateRequesterSecret(requesterSecret);
 
-  const telegramContent = decryptContent(
-    encryptedTelegramContent,
-    requesterSecret.telegramContentEncryptionKey
-  );
+    const bulkSize = parseInt(IEXEC_BULK_SLICE_SIZE, 10) || 0;
+
+    // Process multiple protected data
+    if (bulkSize > 0) {
+      console.log(`Processing ${bulkSize} protected data...`);
+      const processPromises = new Array(bulkSize).fill(null).map((_, index) =>
+        processProtectedData({
+          index: index + 1,
+          IEXEC_IN,
+          appDeveloperSecret,
+          requesterSecret,
+        })
+      );
+      const results = await Promise.all(processPromises);
+      const successCount = results.filter((r) => r.success === true).length;
+      const errorCount = results.filter((r) => r.success !== true).length;
+      result = {
+        success: errorCount === 0,
+        error: errorCount > 0 ? 'Partial failure' : undefined,
+        totalCount: results.length,
+        successCount,
+        errorCount,
+        results: results.map((r) => ({
+          index: r.index,
+          protectedData: r.protectedData,
+          success: r.success,
+          error: r.error,
+        })),
+      };
+    } else {
+      console.log('Processing single protected data...');
+      const { protectedData, success, error } = await processProtectedData({
+        index: 0,
+        IEXEC_IN,
+        appDeveloperSecret,
+        requesterSecret,
+      });
+      result = { protectedData, success, error };
+    }
+  } catch (e) {
+    console.error('Something went wrong:', e.message);
+    result = { success: false, error: e.message };
+  }
 
-  const response = await sendTelegram({
-    chatId: protectedData.chatId,
-    message: telegramContent,
-    botToken: appDeveloperSecret.TELEGRAM_BOT_TOKEN,
-    senderName: requesterSecret.senderName,
-  });
-  await writeTaskOutput(
-    `${workerEnv.IEXEC_OUT}/result.txt`,
-    JSON.stringify(response, null, 2)
+  console.log('Writing results:', JSON.stringify(result));
+  await fs.writeFile(
+    `${workerEnv.IEXEC_OUT}/result.json`,
+    JSON.stringify(result, null, 2)
   );
-  await writeTaskOutput(
+  await fs.writeFile(
     `${workerEnv.IEXEC_OUT}/computed.json`,
-    JSON.stringify({
-      'deterministic-output-path': `${workerEnv.IEXEC_OUT}/result.txt`,
-    })
+    JSON.stringify(
+      {
+        'deterministic-output-path': `${workerEnv.IEXEC_OUT}/result.json`,
+      },
+      null,
+      2
+    )
   );
 }
 

dapp/src/executeTask.js

@@ -4,49 +4,27 @@ async function sendTelegram({
   botToken,
   senderName = 'Web3Telegram Dapp',
 }) {
-  if (!botToken || botToken.trim() === '')
-    throw new Error('Bot token is required');
-  if (!chatId || chatId.trim() === '') throw new Error('Chat ID is required');
-  if (!message || message.trim() === '')
-    throw new Error('Message content is required');
-
   const messageToSend = `Message from: ${senderName}\n${message}`;
-
-  try {
-    const response = await fetch(
-      `https://api.telegram.org/bot${botToken}/sendMessage`,
-      {
-        method: 'POST',
-        headers: {
-          'Content-Type': 'application/json',
-        },
-        body: JSON.stringify({
-          chat_id: chatId,
-          text: messageToSend,
-          parse_mode: 'HTML',
-        }),
-      }
-    );
-
-    if (!response.ok) {
-      console.error('Telegram API Error');
-      return {
-        message: 'Failed to send Telegram message.',
-        status: response.status,
-      };
+  const response = await fetch(
+    `https://api.telegram.org/bot${botToken}/sendMessage`,
+    {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+      },
+      body: JSON.stringify({
+        chat_id: chatId,
+        text: messageToSend,
+        parse_mode: 'HTML',
+      }),
     }
-
-    console.log('Message successfully sent by Telegram bot.');
-    return {
-      message: 'Your telegram message has been sent successfully.',
-      status: 200,
-    };
-  } catch (error) {
-    console.error('Failed to send Telegram message');
-    return {
-      message: 'Failed to send Telegram message.',
-      status: 500,
-    };
+  ).catch(() => {
+    throw new Error('Failed to reach Telegram bot API');
+  });
+  if (!response.ok) {
+    throw new Error(
+      `Failed to send Telegram message, bot API answered with status: ${response.status}`
+    );
   }
 }
 

dapp/src/telegramService.js

@@ -0,0 +1 @@
+invalid content

dapp/tests/_test_inputs_/invalid-data.zip

@@ -1 +1,3 @@
-{"deterministic-output-path":"./tests/_test_outputs_/iexec_out/result.txt"}
+{
+  "deterministic-output-path": "./tests/_test_outputs_/iexec_out/result.json"
+}