Replay.js

const fs = require('fs');
const readLastLines = require('read-last-lines');
const LineByLineReader = require('line-by-line');
const dhive = require('@hiveio/dhive');
const { IPC } = require('../libs/IPC');
const { Transaction } = require('../libs/Transaction');
const BC_PLUGIN_NAME = require('./Blockchain.constants').PLUGIN_NAME;
const BC_PLUGIN_ACTIONS = require('./Blockchain.constants').PLUGIN_ACTIONS;
const { PLUGIN_NAME, PLUGIN_ACTIONS } = require('./Replay.constants');

const PLUGIN_PATH = require.resolve(__filename);

const ipc = new IPC(PLUGIN_NAME);
let hiveClient = null;

let chainIdentifier;
let currentHiveBlock = 0;
let currentBlock = 0;
let filePath = '';
let hiveNode = '';

function getCurrentBlock() {
  return currentBlock;
}

function getcurrentHiveBlock() {
  return currentHiveBlock;
}

function sendBlock(block) {
  return ipc.send(
    { to: BC_PLUGIN_NAME, action: BC_PLUGIN_ACTIONS.PRODUCE_NEW_BLOCK_SYNC, payload: block },
  );
}

// parse the transactions found in a Hive block
const parseTransactions = (refBlockNumber, block) => {
  const newTransactions = [];
  const transactionsLength = block.transactions.length;

  for (let i = 0; i < transactionsLength; i += 1) {
    const nbOperations = block.transactions[i].operations.length;

    for (let indexOp = 0; indexOp < nbOperations; indexOp += 1) {
      const operation = block.transactions[i].operations[indexOp];

      if (operation[0] === 'custom_json'
        || operation[0] === 'transfer'
        || operation[0] === 'comment'
        || operation[0] === 'comment_options'
        || operation[0] === 'vote'
      ) {
        try {
          let id = null;
          let sender = null;
          let recipient = null;
          let amount = null;
          let permlink = null;
          let sscTransactions = [];
          let isSignedWithActiveKey = null;

          if (operation[0] === 'custom_json') {
            id = operation[1].id; // eslint-disable-line prefer-destructuring
            if (operation[1].required_auths.length > 0) {
              sender = operation[1].required_auths[0]; // eslint-disable-line
              isSignedWithActiveKey = true;
            } else {
              sender = operation[1].required_posting_auths[0]; // eslint-disable-line
              isSignedWithActiveKey = false;
            }
            let jsonObj = JSON.parse(operation[1].json); // eslint-disable-line
            sscTransactions = Array.isArray(jsonObj) ? jsonObj : [jsonObj];
          } else if (operation[0] === 'transfer') {
            isSignedWithActiveKey = true;
            sender = operation[1].from;
            recipient = operation[1].to;
            amount = operation[1].amount; // eslint-disable-line prefer-destructuring
            const transferParams = JSON.parse(operation[1].memo);
            id = transferParams.id; // eslint-disable-line prefer-destructuring
            // multi transactions is not supported for the Hive transfers
            if (Array.isArray(transferParams.json) && transferParams.json.length === 1) {
              sscTransactions = transferParams.json;
            } else if (!Array.isArray(transferParams.json)) {
              sscTransactions = [transferParams.json];
            }
          } else if (operation[0] === 'comment') {
            sender = operation[1].author;
            const commentMeta = operation[1].json_metadata !== '' ? JSON.parse(operation[1].json_metadata) : null;

            if (commentMeta && commentMeta.ssc) {
              id = commentMeta.ssc.id; // eslint-disable-line prefer-destructuring
              sscTransactions = commentMeta.ssc.transactions;
              permlink = operation[1].permlink; // eslint-disable-line prefer-destructuring
            } else {
              const commentBody = JSON.parse(operation[1].body);
              id = commentBody.id; // eslint-disable-line prefer-destructuring
              sscTransactions = Array.isArray(commentBody.json)
                ? commentBody.json : [commentBody.json];
            }
          } else if (operation[0] === 'comment_options') {
            id = `ssc-${chainIdentifier}`;
            sender = 'null';
            permlink = operation[1].permlink; // eslint-disable-line prefer-destructuring

            const extensions = operation[1].extensions; // eslint-disable-line prefer-destructuring
            let beneficiaries = [];
            if (extensions
              && extensions[0] && extensions[0].length > 1
              && extensions[0][1].beneficiaries) {
              beneficiaries = extensions[0][1].beneficiaries; // eslint-disable-line
            }

            sscTransactions = [
              {
                contractName: 'comments',
                contractAction: 'commentOptions',
                contractPayload: {
                  author: operation[1].author,
                  maxAcceptedPayout: operation[1].max_accepted_payout,
                  allowVotes: operation[1].allow_votes,
                  allowCurationRewards: operation[1].allow_curation_rewards,
                  beneficiaries,
                },
              },
            ];
          } else if (operation[0] === 'vote') {
            id = `ssc-${chainIdentifier}`;
            sender = 'null';
            permlink = operation[1].permlink; // eslint-disable-line prefer-destructuring

            sscTransactions = [
              {
                contractName: 'comments',
                contractAction: 'vote',
                contractPayload: {
                  voter: operation[1].voter,
                  author: operation[1].author,
                  weight: operation[1].weight,
                },
              },
            ];
          }

          if (id && id === `ssc-${chainIdentifier}` && sscTransactions.length > 0) {
            const nbTransactions = sscTransactions.length;
            for (let index = 0; index < nbTransactions; index += 1) {
              const sscTransaction = sscTransactions[index];

              const { contractName, contractAction, contractPayload } = sscTransaction;
              if (contractName && typeof contractName === 'string'
                && contractAction && typeof contractAction === 'string'
                && contractPayload && typeof contractPayload === 'object') {
                contractPayload.recipient = recipient;
                contractPayload.amountHIVEHBD = amount;
                contractPayload.isSignedWithActiveKey = isSignedWithActiveKey;
                contractPayload.permlink = permlink;

                if (recipient === null) {
                  delete contractPayload.recipient;
                }

                if (amount === null) {
                  delete contractPayload.amountHIVEHBD;
                }

                if (isSignedWithActiveKey === null) {
                  delete contractPayload.isSignedWithActiveKey;
                }

                if (permlink === null) {
                  delete contractPayload.permlink;
                }

                // callingContractInfo is a reserved property
                // it is used to provide information about a contract when calling
                // a contract action from another contract
                if (contractPayload.callingContractInfo) {
                  delete contractPayload.callingContractInfo;
                }

                // set the sender to null when calling the comment action
                // this way we allow people to create comments only via the comment operation
                if (operation[0] === 'comment' && contractName === 'comments' && contractAction === 'comment') {
                  contractPayload.author = sender;
                  sender = 'null';
                }

                // if multi transactions
                // append the index of the transaction to the Hive transaction id
                let SSCtransactionId = block.transaction_ids[i];

                if (nbOperations > 1) {
                  SSCtransactionId = `${SSCtransactionId}-${indexOp}`;
                }

                if (nbTransactions > 1) {
                  SSCtransactionId = `${SSCtransactionId}-${index}`;
                }

                /* console.log( // eslint-disable-line no-console
                  'sender:',
                  sender,
                  'recipient',
                  recipient,
                  'amount',
                  amount,
                  'contractName:',
                  contractName,
                  'contractAction:',
                  contractAction,
                  'contractPayload:',
                  contractPayload,
                ); */

                newTransactions.push(
                  new Transaction(
                    refBlockNumber,
                    SSCtransactionId,
                    sender,
                    contractName,
                    contractAction,
                    JSON.stringify(contractPayload),
                  ),
                );
              }
            }
          }
        } catch (e) {
          // console.error('Invalid transaction', e); // eslint-disable-line no-console
        }
      }
    }
  }

  return newTransactions;
};

async function getHiveBlock(hiveBlockNumber) {
  try {
    const block = await hiveClient.database.getBlock(hiveBlockNumber);
    return block;
  } catch (error) {
    return getHiveBlock(hiveBlockNumber);
  }
}


function replayFile(callback) {
  let lr;
  let lastRefHiveBlockNumber = 0;
  // make sure file exists
  fs.stat(filePath, async (err, stats) => {
    if (!err && stats.isFile()) {
      // read last line of the file to determine the number of blocks to replay

      const lastLine = await readLastLines.read(filePath, 1);
      const lastBlock = JSON.parse(lastLine);
      const lastBockNumber = lastBlock.blockNumber;

      // read the file from the start
      lr = new LineByLineReader(filePath);

      lr.on('line', async (line) => {
        lr.pause();
        if (line !== '') {
          const block = JSON.parse(line);
          const {
            blockNumber,
            timestamp,
            transactions,
            refHiveBlockNumber,
            refHiveBlockId,
            prevRefHiveBlockId,
            virtualTransactions,
          } = block;

          if (lastRefHiveBlockNumber === 0) {
            lastRefHiveBlockNumber = refHiveBlockNumber - 1;
          }

          let finalRefHiveBlockId = refHiveBlockId;
          let finalPrevRefHiveBlockId = prevRefHiveBlockId;

          if (blockNumber !== 0) {
            // if there are missing Hive blocks
            while (refHiveBlockNumber !== lastRefHiveBlockNumber + 1 
              && lastRefHiveBlockNumber !== -1) {
              console.log('processing missing Hive block', lastRefHiveBlockNumber + 1);
              // eslint-disable-next-line no-await-in-loop
              const hiveBlock = await getHiveBlock(lastRefHiveBlockNumber + 1);
              // eslint-disable-next-line no-await-in-loop
              await sendBlock(
                {
                  // we timestamp the block with the Hive block timestamp
                  timestamp: hiveBlock.timestamp,
                  refHiveBlockNumber: lastRefHiveBlockNumber + 1,
                  refHiveBlockId: hiveBlock.block_id,
                  prevRefHiveBlockId: hiveBlock.previous,
                  transactions: parseTransactions(
                    hiveBlock.blockNumber,
                    hiveBlock,
                  ),
                },
              );

              lastRefHiveBlockNumber += 1;
            }

            currentHiveBlock = refHiveBlockNumber;
            currentBlock = blockNumber;
            console.log(`replaying block ${currentBlock} / ${lastBockNumber}`); // eslint-disable-line no-console

            if (hiveClient !== null && finalRefHiveBlockId === undefined) {
              const hiveBlock = await hiveClient.database.getBlock(refHiveBlockNumber);
              finalRefHiveBlockId = hiveBlock.block_id;
              finalPrevRefHiveBlockId = hiveBlock.previous;
            }

            await sendBlock({
              blockNumber,
              timestamp,
              refHiveBlockNumber,
              refHiveBlockId: finalRefHiveBlockId,
              prevRefHiveBlockId: finalPrevRefHiveBlockId,
              transactions,
              virtualTransactions,
            });

            lastRefHiveBlockNumber = refHiveBlockNumber;
          }
        }
        lr.resume();
      });

      lr.on('error', (error) => {
        callback(error);
      });

      lr.on('end', () => {
        console.log('Replay done'); // eslint-disable-line no-console
        callback(null);
      });
    } else {
      // file does not exist, so callback with null
      callback(`file located at ${filePath} does not exist`);
    }
  });
}

function init(payload) {
  const { blocksLogFilePath, streamNodes, chainId } = payload;
  filePath = blocksLogFilePath;
  hiveNode = streamNodes[0]; // eslint-disable-line
  hiveClient = new dhive.Client(hiveNode);
  chainIdentifier = chainId;
}

ipc.onReceiveMessage((message) => {
  const {
    action,
    payload,
    // from,
  } = message;

  switch (action) {
    case 'init':
      init(payload);
      ipc.reply(message);
      console.log('successfully initialized'); // eslint-disable-line no-console
      break;
    case 'stop':
      ipc.reply(message, getcurrentHiveBlock() + 1);
      console.log('successfully stopped'); // eslint-disable-line no-console
      break;
    case PLUGIN_ACTIONS.REPLAY_FILE:
      replayFile((result) => {
        let finalResult = null;
        if (result === null) {
          finalResult = getCurrentBlock();
        }
        if (result) console.log('error encountered during the replay:', result); // eslint-disable-line no-console

        ipc.reply(message, finalResult);
      });
      break;
    case PLUGIN_ACTIONS.GET_CURRENT_BLOCK:
      ipc.reply(message, getCurrentBlock());
      break;
    default:
      ipc.reply(message);
      break;
  }
});

module.exports.PLUGIN_NAME = PLUGIN_NAME;
module.exports.PLUGIN_PATH = PLUGIN_PATH;
module.exports.PLUGIN_ACTIONS = PLUGIN_ACTIONS;