<commit_before>import logger from '../../logger';
import config from '../../config';
import { isChainSupported } from '../../types/SupportedChain';
import { BlockModel } from '../../models/block';
import { TransactionModel } from '../../models/transaction';
import { ChainNetwork, Chain } from '../../types/ChainNetwork';
import { Block } from '../../models/block';
import { sleep } from '../../utils/async';
import { Transaction } from '../../models/transaction';
import { BtcP2pService } from './bitcoin';
import { Bitcoin } from '../../types/namespaces/Bitcoin';
import { CSP } from '../../types/namespaces/ChainStateProvider';
import { EventEmitter } from 'events';
import { LoggifyClass } from '../../decorators/Loggify';

const P2PClasses: {
  [key: string]: Class<StandardP2p>;
} = {
  BCH: BtcP2pService,
  BTC: BtcP2pService
};

export interface P2pService<Block, _Transaction> {
  // a stream of blocks and transactions
  stream: EventEmitter;

  // connect to the peers and begin emitting data
  start(): Promise<void>;

  // get the max height of every peer in the pool
  height(): number;

  // get information about the chain this forked from (if applicable)
  parent(): ChainNetwork & { height: number } | undefined;

  // disconnects from peer and stops all pending tasks,
  // afterwards `start()` can be called again.
  stop(): Promise<void>;

  // when `true` only emit blocks that result from the syncing process
  syncing: boolean;

  // Get block hashes from a list of headers
  getMissingBlockHashes(hashes: string[]): Promise<string[]>;

  // Get a block from its hash
  getBlock(hash: string): Promise<Block>;
}

export type StandardP2p = P2pService<Bitcoin.Block, Bitcoin.Transaction>;

export enum P2pEvents {
  SYNC_COMPLETE = 'SYNC_COMPLETE'
}

@LoggifyClass
export class P2pRunner {
  private service: StandardP2p;
  private chain: string;
  private network: string;
  private blocks: Block;
  private transactions: Transaction;

  public events: EventEmitter;

  constructor(chain: string, network: string, blocks: Block, transactions: Transaction, service: StandardP2p) {
    this.service = service;
    this.chain = chain;
    this.network = network;
    this.blocks = blocks;
    this.transactions = transactions;
    this.events = new EventEmitter();
  }

  private wireupBlockStream(parent?: { height: number; chain: string }) {
    this.service.stream.on('block', async (block: Bitcoin.Block) => {
      await this.blocks.addBlock({
        chain: this.chain,
        network: this.network,
        forkHeight: parent ? parent.height : 0,
        parentChain: parent ? parent.chain : this.chain,
        block: block
      });
      logger.info(`Added block ${block.hash}`, {
        chain: this.chain,
        network: this.network
      });
    });
  }

  private wireupTxStream() {
    this.service.stream.on('tx', async (tx: Bitcoin.Transaction) => {
      await this.transactions.batchImport({
        txs: [tx],
        height: -1,
        network: this.network,
        chain: this.chain,
        blockTime: new Date(),
        blockTimeNormalized: new Date()
      });
      logger.debug(`Added transaction ${tx.hash}`, {
        chain: this.chain,
        network: this.network
      });
    });
  }

  async start() {
    logger.debug(`Started worker for chain ${this.chain}`);
    const parent = this.service.parent();
    this.service.syncing = true;
    this.wireupBlockStream(parent);
    this.wireupTxStream();
    // wait for it to get connected
    await this.service.start();
    await this.sync();
  }

  async sync() {
    this.service.syncing = true;
    const parent = this.service.parent();
    const tip = () =>
      this.blocks.getLocalTip({
        chain: this.chain,
        network: this.network
      });

    // remove the previous block to ensure consistency through process termination
    await this.blocks.handleReorg({ chain: this.chain, network: this.network });

    // get best block we currently have to see if we're synced
    let bestBlock = await tip();

    // wait for the parent fork to sync first
    while (parent && bestBlock.height < parent.height) {
      logger.info(`Waiting until ${parent.chain} syncs before ${this.chain}`);
      await sleep(5000);
      bestBlock = await tip();
    }

    const getHeaders = async () => {
      const locators = await this.blocks.getLocatorHashes({
        chain: this.chain,
        network: this.network
      });
      logger.debug(`Received ${locators.length} headers`);

      return this.service.getMissingBlockHashes(locators);
    };


    let hashes;
    while (!hashes || hashes.length > 0) {
      hashes = await getHeaders();
      bestBlock = await tip();
      let lastLog = 0;
      let counter = bestBlock.height;
      logger.info(
        `Syncing from ${
        bestBlock.height
        } to ${this.service.height()} for chain ${this.chain}`
      );

      for (const hash of hashes) {
        const block = await this.service.getBlock(hash);
        logger.debug('Block received', block.hash);
        await this.blocks.addBlock({
          chain: this.chain,
          network: this.network,
          forkHeight: parent ? parent.height : 0,
          parentChain: parent ? parent.chain : this.chain,
          block
        });
        logger.debug(`Syncing block ${block.hash}`, {
          chain: this.chain,
          network: this.network
        });
        counter += 1;
        if (Date.now() - lastLog > 100) {

          logger.info(`Sync progress ${((counter * 100) / this.service.height()).toFixed(3)}%`, {
            chain: this.chain,
            network: this.network,
            height: counter
          });
          lastLog = Date.now();
        }
      }
    }
    logger.info(`${this.chain}:${this.network} up to date.`);
    this.service.syncing = false;
    this.events.emit(P2pEvents.SYNC_COMPLETE, true);
  }<commit_msg>unneeded newlines<commit_after>import logger from '../../logger';
import config from '../../config';
import { isChainSupported } from '../../types/SupportedChain';
import { BlockModel } from '../../models/block';
import { TransactionModel } from '../../models/transaction';
import { ChainNetwork, Chain } from '../../types/ChainNetwork';
import { Block } from '../../models/block';
import { sleep } from '../../utils/async';
import { Transaction } from '../../models/transaction';
import { BtcP2pService } from './bitcoin';
import { Bitcoin } from '../../types/namespaces/Bitcoin';
import { CSP } from '../../types/namespaces/ChainStateProvider';
import { EventEmitter } from 'events';
import { LoggifyClass } from '../../decorators/Loggify';

const P2PClasses: {
  [key: string]: Class<StandardP2p>;
} = {
  BCH: BtcP2pService,
  BTC: BtcP2pService
};

export interface P2pService<Block, _Transaction> {
  // a stream of blocks and transactions
  stream: EventEmitter;

  // connect to the peers and begin emitting data
  start(): Promise<void>;

  // get the max height of every peer in the pool
  height(): number;

  // get information about the chain this forked from (if applicable)
  parent(): ChainNetwork & { height: number } | undefined;

  // disconnects from peer and stops all pending tasks,
  // afterwards `start()` can be called again.
  stop(): Promise<void>;

  // when `true` only emit blocks that result from the syncing process
  syncing: boolean;

  // Get block hashes from a list of headers
  getMissingBlockHashes(hashes: string[]): Promise<string[]>;

  // Get a block from its hash
  getBlock(hash: string): Promise<Block>;
}

export type StandardP2p = P2pService<Bitcoin.Block, Bitcoin.Transaction>;

export enum P2pEvents {
  SYNC_COMPLETE = 'SYNC_COMPLETE'
}

@LoggifyClass
export class P2pRunner {
  private service: StandardP2p;
  private chain: string;
  private network: string;
  private blocks: Block;
  private transactions: Transaction;

  public events: EventEmitter;

  constructor(chain: string, network: string, blocks: Block, transactions: Transaction, service: StandardP2p) {
    this.service = service;
    this.chain = chain;
    this.network = network;
    this.blocks = blocks;
    this.transactions = transactions;
    this.events = new EventEmitter();
  }

  private wireupBlockStream(parent?: { height: number; chain: string }) {
    this.service.stream.on('block', async (block: Bitcoin.Block) => {
      await this.blocks.addBlock({
        chain: this.chain,
        network: this.network,
        forkHeight: parent ? parent.height : 0,
        parentChain: parent ? parent.chain : this.chain,
        block: block
      });
      logger.info(`Added block ${block.hash}`, {
        chain: this.chain,
        network: this.network
      });
    });
  }

  private wireupTxStream() {
    this.service.stream.on('tx', async (tx: Bitcoin.Transaction) => {
      await this.transactions.batchImport({
        txs: [tx],
        height: -1,
        network: this.network,
        chain: this.chain,
        blockTime: new Date(),
        blockTimeNormalized: new Date()
      });
      logger.debug(`Added transaction ${tx.hash}`, {
        chain: this.chain,
        network: this.network
      });
    });
  }

  async start() {
    logger.debug(`Started worker for chain ${this.chain}`);
    const parent = this.service.parent();
    this.service.syncing = true;
    this.wireupBlockStream(parent);
    this.wireupTxStream();
    // wait for it to get connected
    await this.service.start();
    await this.sync();
  }

  async sync() {
    this.service.syncing = true;
    const parent = this.service.parent();
    const tip = () =>
      this.blocks.getLocalTip({
        chain: this.chain,
        network: this.network
      });

    // remove the previous block to ensure consistency through process termination
    await this.blocks.handleReorg({ chain: this.chain, network: this.network });

    // get best block we currently have to see if we're synced
    let bestBlock = await tip();

    // wait for the parent fork to sync first
    while (parent && bestBlock.height < parent.height) {
      logger.info(`Waiting until ${parent.chain} syncs before ${this.chain}`);
      await sleep(5000);
      bestBlock = await tip();
    }

    const getHeaders = async () => {
      const locators = await this.blocks.getLocatorHashes({
        chain: this.chain,
        network: this.network
      });
      logger.debug(`Received ${locators.length} headers`);

      return this.service.getMissingBlockHashes(locators);
    };

    let hashes;
    while (!hashes || hashes.length > 0) {
      hashes = await getHeaders();
      bestBlock = await tip();
      let lastLog = 0;
      let counter = bestBlock.height;
      logger.info(
        `Syncing from ${
        bestBlock.height
        } to ${this.service.height()} for chain ${this.chain}`
      );

      for (const hash of hashes) {
        const block = await this.service.getBlock(hash);
        logger.debug('Block received', block.hash);
        await this.blocks.addBlock({
          chain: this.chain,
          network: this.network,
          forkHeight: parent ? parent.height : 0,
          parentChain: parent ? parent.chain : this.chain,
          block
        });
        logger.debug(`Syncing block ${block.hash}`, {
          chain: this.chain,
          network: this.network
        });
        counter += 1;
        if (Date.now() - lastLog > 100) {
          logger.info(`Sync progress ${((counter * 100) / this.service.height()).toFixed(3)}%`, {
            chain: this.chain,
            network: this.network,
            height: counter
          });
          lastLog = Date.now();
        }
      }
    }
    logger.info(`${this.chain}:${this.network} up to date.`);
    this.service.syncing = false;
    this.events.emit(P2pEvents.SYNC_COMPLETE, true);
  }
{'fraction_non_alphanumeric': 0.1344295643329378, 'fraction_numerical': 0.0050856077301237495, 'mean_word_length': 2.6371763255240444, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4145326', 'n_tokens_mistral': 3389, 'n_tokens_neox': 3374, 'n_words': 955}