feat(iapp): add bulk processing support

Author: abbesBenayache

dapp/src/executeTask.js

@@ -1,5 +1,6 @@
 import { IExecDataProtectorDeserializer } from '@iexec/dataprotector-deserializer';
 import { promises as fs } from 'fs';
+import path from 'node:path';
 import { decryptContent, downloadEncryptedContent } from './decryptContent.js';
 import sendTelegram from './telegramService.js';
 import {
@@ -19,45 +20,35 @@ async function writeTaskOutput(path, message) {
   }
 }
 
-async function start() {
-  const { IEXEC_OUT, IEXEC_APP_DEVELOPER_SECRET, IEXEC_REQUESTER_SECRET_1 } =
-    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');
+async function processProtectedData(
+  index,
+  {
+    IEXEC_IN,
+    IEXEC_OUT,
+    appDeveloperSecret,
+    requesterSecret,
+    datasetFilename = null,
   }
-  appDeveloperSecret = validateAppSecret(appDeveloperSecret);
-
-  // Parse the requester secret environment variable
-  let requesterSecret;
+) {
+  // Parse the protected data
+  let protectedData;
   try {
-    requesterSecret = IEXEC_REQUESTER_SECRET_1
-      ? JSON.parse(IEXEC_REQUESTER_SECRET_1)
+    const deserializerConfig = datasetFilename
+      ? { protectedDataPath: path.join(IEXEC_IN, datasetFilename) }
       : {};
-  } catch {
-    throw Error('Failed to parse requester secret');
-  }
-  requesterSecret = validateRequesterSecret(requesterSecret);
 
-  // Parse the protected data and get the requester secret (chatId)
-  let protectedData;
-  try {
-    const deserializer = new IExecDataProtectorDeserializer();
+    const deserializer = new IExecDataProtectorDeserializer(deserializerConfig);
     protectedData = {
       chatId: await deserializer.getValue('telegram_chatId', 'string'),
     };
   } catch (e) {
-    throw Error(`Failed to parse ProtectedData: ${e.message}`);
+    throw Error(`Failed to parse ProtectedData ${index}: ${e.message}`);
   }
-  // Validate the protected data (chatId)
+
+  // Validate the protected data
   validateProtectedData(protectedData);
 
+  // Download and decrypt content
   const encryptedTelegramContent = await downloadEncryptedContent(
     requesterSecret.telegramContentMultiAddr
   );
@@ -67,21 +58,101 @@ async function start() {
     requesterSecret.telegramContentEncryptionKey
   );
 
+  // Send telegram message
   const response = await sendTelegram({
     chatId: protectedData.chatId,
     message: telegramContent,
     botToken: appDeveloperSecret.TELEGRAM_BOT_TOKEN,
     senderName: requesterSecret.senderName,
   });
+
+  // Write individual result file
+  const resultFileName = index > 0 ? `${datasetFilename}.txt` : 'result.txt';
   await writeTaskOutput(
-    `${workerEnv.IEXEC_OUT}/result.txt`,
+    path.join(IEXEC_OUT, resultFileName),
     JSON.stringify(response, null, 2)
   );
+
+  return { index, response, resultFileName };
+}
+
+async function start() {
+  const {
+    IEXEC_OUT,
+    IEXEC_IN,
+    IEXEC_APP_DEVELOPER_SECRET,
+    IEXEC_REQUESTER_SECRET_1,
+    IEXEC_BULK_SLICE_SIZE,
+  } = process.env;
+
+  // Check worker env
+  const workerEnv = validateWorkerEnv({ IEXEC_OUT });
+
+  // Parse the app developer secret
+  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
+  let requesterSecret;
+  try {
+    requesterSecret = IEXEC_REQUESTER_SECRET_1
+      ? JSON.parse(IEXEC_REQUESTER_SECRET_1)
+      : {};
+  } catch {
+    throw Error('Failed to parse requester secret');
+  }
+  requesterSecret = validateRequesterSecret(requesterSecret);
+
+  const bulkSize = parseInt(IEXEC_BULK_SLICE_SIZE) || 0;
+
+  const results = [];
+
+  if (bulkSize > 0) {
+    // Process multiple protected data
+    for (let i = 1; i <= bulkSize; i++) {
+      const datasetFilename = process.env[`IEXEC_DATASET_${i}_FILENAME`];
+
+      const result = await processProtectedData(i, {
+        IEXEC_IN,
+        IEXEC_OUT: workerEnv.IEXEC_OUT,
+        appDeveloperSecret,
+        requesterSecret,
+        datasetFilename,
+      });
+
+      results.push(result);
+    }
+  } else {
+    // Process single protected data
+    const result = await processProtectedData(0, {
+      IEXEC_IN,
+      IEXEC_OUT: workerEnv.IEXEC_OUT,
+      appDeveloperSecret,
+      requesterSecret,
+    });
+
+    results.push(result);
+  }
+
+  // Write computed.json with all results
+  const computedOutput = {
+    'deterministic-output-path': workerEnv.IEXEC_OUT,
+    'bulk-results': results.map((r) => ({
+      index: r.index,
+      file: r.resultFileName,
+      status: r.response.status === 200 ? 'success' : 'error',
+    })),
+    'total-processed': results.length,
+  };
+
   await writeTaskOutput(
-    `${workerEnv.IEXEC_OUT}/computed.json`,
-    JSON.stringify({
-      'deterministic-output-path': `${workerEnv.IEXEC_OUT}/result.txt`,
-    })
+    path.join(workerEnv.IEXEC_OUT, 'computed.json'),
+    JSON.stringify(computedOutput, null, 2)
   );
 }