Read response pathfinder2

.env.example

@@ -4,7 +4,7 @@ RPC_URL=http://localhost:8545
 # Service Endpoints
 API_SERVICE_ENDPOINT=http://api.circles.local
 GRAPH_NODE_ENDPOINT=http://graph.circles.local
-PATHFINDER_SERVICE_ENDPOINT=http://localhost:8081
+PATHFINDER_SERVICE_ENDPOINT=http://localhost:54389
 RELAY_SERVICE_ENDPOINT=http://relay.circles.local
 
 # Database Endpoint

.github/workflows/tests.yml

@@ -36,7 +36,7 @@ jobs:
         uses: actions/checkout@v3
         with:
           repository: CirclesUBI/circles-docker.git
-          ref: main
+          ref: test-iterative
           path: circles-docker
 
       - name: Setup docker repo
@@ -45,15 +45,15 @@ jobs:
 
       - name: Container setup via docker-compose without pathfinder
         working-directory: circles-docker
-        run: docker compose -f docker-compose.yml -f docker-compose.relayer-pull.yml -f docker-compose.api-pull.yml -p circles  up --detach --remove-orphans --build
+        run: docker compose -f docker-compose.yml -f docker-compose.relayer-pull.yml -f docker-compose.api-pull.yml -p circles up --detach --remove-orphans --build
 
       - name: Download and migrate contracts
         working-directory: circles-docker
         run: ./scripts/migrate-contracts.sh
 
       - name: Try starting failed services
         working-directory: circles-docker
-        run: docker compose -f docker-compose.yml -f docker-compose.relayer-pull.yml -f docker-compose.api-pull.yml -p circles  up --detach --remove-orphans --build
+        run: docker compose -f docker-compose.yml -f docker-compose.relayer-pull.yml -f docker-compose.api-pull.yml -p circles up --detach --remove-orphans --build
 
       - name: Container setup via docker-compose for pathfinder
         working-directory: circles-docker

src/token.js

@@ -18,6 +18,7 @@ import { getVersion } from '~/safe';
  * @param {Object} userOptions - search arguments
  * @param {string} pathfinderType - pathfinder execution type
  * @param {number} pathfinderMaxTransferSteps - default pathfinder server max transfer steps
+ * @param {Boolean} returnIterativeFirstMatch - if true, the pathfinder service iteratively optimizes the result and returns the first match with 'value'. Only available when pathfinderType is 'cli'
  *
  * @return {Object[]} - transaction steps
  */
@@ -27,6 +28,7 @@ export async function findTransitiveTransfer(
   userOptions,
   pathfinderType,
   pathfinderMaxTransferSteps,
+  returnIterativeFirstMatch = false,
 ) {
   let result;
   if (pathfinderType == 'cli') {
@@ -39,6 +41,7 @@ export async function findTransitiveTransfer(
       utils,
       userOptions,
       pathfinderMaxTransferSteps,
+      returnIterativeFirstMatch,
     );
   }
   return result;
@@ -109,6 +112,7 @@ async function findTransitiveTransferCli(web3, utils, userOptions) {
  * @param {string} userOptions.to - receiver Safe address
  * @param {BN} userOptions.value - value of Circles tokens
  * @param {number} userOptions.maxTransfers - limit of steps returned by the pathfinder service
+ * @param {Boolean} returnIterativeFirstMatch - if true, the pathfinder service iteratively optimizes the result and returns the first match with 'value'. Only available when pathfinderType is 'cli'
  *
  * @return {Object[]} - transaction steps
  */
@@ -117,6 +121,7 @@ async function findTransitiveTransferServer(
   utils,
   userOptions,
   pathfinderMaxTransferSteps,
+  returnIterativeFirstMatch,
 ) {
   const options = checkOptions(userOptions, {
     from: {
@@ -136,7 +141,6 @@ async function findTransitiveTransferServer(
 
   try {
     const response = await utils.requestPathfinderAPI({
-      method: 'POST',
       data: {
         id: crypto.randomUUID(),
         method: 'compute_transfer',
@@ -145,9 +149,9 @@ async function findTransitiveTransferServer(
           to: options.to,
           value: options.value.toString(),
           max_transfers: options.maxTransfers,
+          iterative: returnIterativeFirstMatch,
         },
       },
-      isTrailingSlash: false,
     });
     return response.result;
   } catch (error) {
@@ -586,6 +590,7 @@ export default function createTokenModule(
               options,
               pathfinderType,
               pathfinderMaxTransferSteps,
+              true,
             );
             if (web3.utils.toBN(response.maxFlowValue).lt(options.value)) {
               throw new TransferError(

src/utils.js

@@ -17,7 +17,87 @@ import { getTokenContract, getSafeContract } from '~/common/getContracts';
 /** @access private */
 const transactionQueue = new TransactionQueue();
 
-async function request(endpoint, userOptions) {
+async function processResponseJson(response) {
+  return new Promise((resolve, reject) => {
+    const getJson = (response) => {
+      return response.json().then((json) => {
+        if (response.status >= 400) {
+          throw new RequestError(response.url, json, response.status);
+        }
+        return json;
+      });
+    };
+    const contentType = response.headers.get('Content-Type');
+    if (contentType && contentType.includes('application/json')) {
+      getJson(response).then(resolve).catch(reject);
+    } else {
+      if (response.status >= 400) {
+        reject(new RequestError(response.url, response.body, response.status));
+      }
+      resolve(response.body);
+    }
+  });
+}
+
+async function processResponseNdjson(response, data) {
+  let buffer = '';
+  let jsons = [];
+  let final;
+  return new Promise((resolve) => {
+    resolve(response.body);
+  }).then((res) => {
+    return new Promise((resolve, reject) => {
+      res.on('readable', () => {
+        // console.log('readable...*');
+        let result;
+        const decoder = new TextDecoder();
+        while (null !== (result = res.read())) {
+          buffer += decoder.decode(result);
+          let idx = buffer.indexOf('\n');
+          while (idx !== -1) {
+            const text = buffer.substring(0, idx);
+            try {
+              const jsonText = JSON.parse(text);
+              // console.log(jsonText);
+              jsons.push(jsonText);
+              if (jsonText.result.maxFlowValue === data.params.value) {
+                final = jsonText;
+                res.destroy();
+              }
+            } catch (error) {
+              // console.warn(text);
+              reject(error);
+            }
+            buffer = buffer.substring(idx + 1);
+            idx = buffer.indexOf('\n');
+          }
+        }
+      });
+      res.on('end', () => {
+        // If haven't received a matching result yet, then return the last result
+        // console.log('END!');
+        // console.log({ final });
+        // console.log({ jsons });
+        resolve(jsons.pop());
+      });
+      res.on('close', function (err) {
+        // console.log('Stream has been destroyed and file has been closed');
+        // console.log({ final });
+        // console.log({ jsons });
+        if (err) {
+          reject(err);
+        }
+        resolve(final);
+      });
+    });
+  });
+}
+
+async function request(
+  endpoint,
+  userOptions,
+  processResponse = processResponseJson,
+) {
   const options = checkOptions(userOptions, {
     path: {
       type: 'array',
@@ -60,25 +140,9 @@ async function request(endpoint, userOptions) {
   const url = `${endpoint}/${path.join('/')}${slash}${paramsStr}`;
 
   try {
-    return fetch(url, request).then((response) => {
-      const contentType = response.headers.get('Content-Type');
-
-      if (contentType && contentType.includes('application/json')) {
-        return response.json().then((json) => {
-          if (response.status >= 400) {
-            throw new RequestError(url, json, response.status);
-          }
-
-          return json;
-        });
-      } else {
-        if (response.status >= 400) {
-          throw new RequestError(url, response.body, response.status);
-        }
-
-        return response.body;
-      }
-    });
+    return fetch(url, request).then((response) =>
+      processResponse(response, data),
+    );
   } catch (err) {
     throw new RequestError(url, err.message);
   }
@@ -1120,28 +1184,27 @@ export default function createUtilsModule(web3, contracts, globalOptions) {
      * @namespace core.utils.requestPathfinderAPI
      *
      * @param {Object} userOptions - Pathfinder API query options
-     * @param {string} userOptions.method - HTTP method
      * @param {Object} userOptions.data - Request body (JSON)
      *
      * @return {Object} - API response
      */
     requestPathfinderAPI: async (userOptions) => {
       const options = checkOptions(userOptions, {
-        method: {
-          type: 'string',
-          default: 'GET',
-        },
         data: {
           type: 'object',
           default: {},
         },
       });
-      return request(pathfinderServiceEndpoint, {
-        data: options.data,
-        method: options.method,
-        path: [],
-        isTrailingSlash: false,
-      });
+      return request(
+        pathfinderServiceEndpoint,
+        {
+          data: options.data,
+          method: 'POST',
+          path: [],
+          isTrailingSlash: false,
+        },
+        processResponseNdjson,
+      );
     },
 
     /**