<commit_before>package scorex.network

import scorex.app.Application
import scorex.block.Block
import scorex.block.Block.BlockId
import scorex.crypto.encode.Base58
import scorex.network.NetworkController.{DataFromPeer, SendToNetwork}
import scorex.network.NetworkObject.ConsideredValue
import scorex.network.message.Message
import scorex.transaction.History
import scorex.utils.ScorexLogging
import shapeless.Typeable._

import scala.collection.mutable
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.util.{Failure, Success, Try}


//todo: write tests
class HistorySynchronizer(application: Application) extends ViewSynchronizer with ScorexLogging {

  import HistorySynchronizer._
  import application.basicMessagesSpecsRepo._

  private implicit val consensusModule = application.consensusModule
  private implicit val transactionalModule = application.transactionModule

  override val messageSpecs = Seq(ScoreMessageSpec, SignaturesSpec, BlockMessageSpec)

  private lazy val scoreSyncer = new ScoreNetworkObject(self)

  private lazy val history = application.history

  protected override lazy val networkControllerRef = application.networkController

  private lazy val blockGenerator = application.blockGenerator

  //todo: make configurable
  private val GettingExtensionTimeout = 40.seconds
  private val GettingBlockTimeout = 10.seconds

  override def preStart: Unit = {
    super.preStart()
    //todo: make configurable
    context.system.scheduler.schedule(1.second, 2.seconds) {
      val msg = Message(ScoreMessageSpec, Right(history.score()), None)
      networkControllerRef ! NetworkController.SendToNetwork(msg, SendToRandom)
    }
  }

  override def receive: Receive =
    if (application.settings.offlineGeneration) gotoSynced() else gotoSyncing()

  def state(status: Status, logic: Receive): Receive =
    logic orElse ({
      case HistorySynchronizer.GetStatus =>
        sender() ! status.name
    }: Receive) orElse commonLogic


  def syncing: Receive = state(HistorySynchronizer.Syncing, {
    case ConsideredValue(Some(networkScore: History.BlockchainScore), witnesses) =>
      val localScore = history.score()
      if (networkScore > localScore) {
        log.info(s"networkScore=$networkScore > localScore=$localScore")
        val lastIds = history.lastBlocks(100).map(_.uniqueId)
        val msg = Message(GetSignaturesSpec, Right(lastIds), None)
        networkControllerRef ! NetworkController.SendToNetwork(msg, SendToChosen(witnesses))
        gotoGettingExtension(witnesses)
      } else gotoSynced()
  }: Receive)

  private val blocksToReceive = mutable.Queue[BlockId]()

  def gettingExtension(witnesses: Seq[ConnectedPeer]): Receive = state(HistorySynchronizer.GettingExtension, {
    case GettingExtensionTimeout => gotoSyncing()

    //todo: aggregating function for block ids (like score has)
    case DataFromPeer(msgId, blockIds: Seq[Block.BlockId]@unchecked, remote)
      if msgId == SignaturesSpec.messageCode &&
        blockIds.cast[Seq[Block.BlockId]].isDefined &&
        witnesses.contains(remote) => //todo: ban if non-expected sender

      val newBLockIds = blockIds.filter(!history.contains(_))
      log.info(s"Got SignaturesMessage with ${blockIds.length} sigs, ${newBLockIds.size} are new")

      val common = blockIds.head
      assert(application.history.contains(common)) //todo: what if not?
      Try(application.blockStorage.removeAfter(common)) //todo we don't need this call for blockTree

      blocksToReceive.clear()

      val newContext = if (newBLockIds.nonEmpty) {
        newBLockIds.foreach { blockId =>
          blocksToReceive += blockId
        }

        networkControllerRef ! NetworkController.SendToNetwork(Message(GetBlockSpec, Right(blocksToReceive.front), None),
          SendToChosen(Seq(remote)))

        gotoGettingBlock(witnesses)
      } else syncing

      context become newContext
  }: Receive)

  def gettingBlock(witnesses: Seq[ConnectedPeer]): Receive = state(HistorySynchronizer.GettingBlock, {
    case GettingBlockTimeout => //15.seconds
      blocksToReceive.clear()
      gotoSyncing()

    case CheckBlock(blockId) =>
      if (blocksToReceive.nonEmpty && blocksToReceive.front.sameElements(blockId)) {
        val sendTo = SendToRandomFromChosen(witnesses)
        val stn = NetworkController.SendToNetwork(Message(GetBlockSpec, Right(blockId), None), sendTo)
        networkControllerRef ! stn
      }

    case DataFromPeer(msgId, block: Block@unchecked, remote)
      if msgId == BlockMessageSpec.messageCode && block.cast[Block].isDefined
        && blocksToReceive.front.sameElements(block.uniqueId) =>

      val blockId = block.uniqueId
      log.info("Got block: " + Base58.encode(blockId))

      if (processNewBlock(block, local = false)) {
        if (blocksToReceive.nonEmpty && blocksToReceive.front.sameElements(blockId)) blocksToReceive.dequeue()

        if (blocksToReceive.nonEmpty) {
          val blockId = blocksToReceive.front
          val ss = SendToRandomFromChosen(Seq(remote))
          val stn = NetworkController.SendToNetwork(Message(GetBlockSpec, Right(blockId), None), ss)
          networkControllerRef ! stn
          context.system.scheduler.scheduleOnce(5.seconds)(self ! CheckBlock(blockId))
        }
      } else if (!history.contains(block.referenceField.value)) {
        log.warn("No parent block in history")
        //blocksToReceive.clear()
        //blocksToReceive.enqueue(block.referenceField.value)
      }

      if (blocksToReceive.nonEmpty) {
        self ! CheckBlock(blocksToReceive.front)
      } else {
        context become syncing
      }
  }: Receive)

  //accept only new block from local or remote
  def synced: Receive = state(HistorySynchronizer.Synced, {
    case block: Block =>
      processNewBlock(block, local = true)

    case ConsideredValue(Some(networkScore: History.BlockchainScore), witnesses) =>
      val localScore = history.score()
      if (networkScore > localScore) {
        blockGenerator ! BlockGenerator.StopGeneration
        gotoGettingExtension(witnesses)
      }

    case DataFromPeer(msgId, block: Block@unchecked, remote)
      if msgId == BlockMessageSpec.messageCode && block.cast[Block].isDefined =>
      processNewBlock(block, local = false)
  }: Receive)

  //common logic for all the states
  def commonLogic: Receive = {
    //todo: check sender
    case DataFromPeer(msgId, content: History.BlockchainScore, remote)
      if msgId == ScoreMessageSpec.messageCode =>

      scoreSyncer.networkUpdate(remote, content)

    case ConsideredValue(Some(networkScore: History.BlockchainScore), witnesses) =>
      log.info("Got unhandled ConsideredValue(score)")

    //the signals to initialize
    case Unit =>

    //timeout signals
    case f: FiniteDuration =>

    case nonsense: Any =>
      log.warn(s"Got something strange: $nonsense")
  }

  private def gotoSyncing() = {
    log.debug("Transition to syncing")
    scoreSyncer.consideredValue.foreach(cv => self ! cv)
    context become syncing
    syncing
  }

  private def gotoGettingExtension(witnesses: Seq[ConnectedPeer]): Unit = {
    log.debug("Transition to gettingExtension")
    context.system.scheduler.scheduleOnce(GettingExtensionTimeout)(self ! GettingExtensionTimeout)
    context become gettingExtension(witnesses)
  }

  private def gotoGettingBlock(witnesses: Seq[ConnectedPeer]): Receive = {
    log.debug("Transition to gettingBlock")
    context.system.scheduler.scheduleOnce(GettingBlockTimeout)(self ! GettingBlockTimeout)
    gettingBlock(witnesses)
  }

  private def gotoSynced() = {
    log.debug("Transition to synced")<commit_msg>commonLogic elimination<commit_after>package scorex.network

import scorex.app.Application
import scorex.block.Block
import scorex.block.Block.BlockId
import scorex.crypto.encode.Base58
import scorex.network.NetworkController.{DataFromPeer, SendToNetwork}
import scorex.network.NetworkObject.ConsideredValue
import scorex.network.message.Message
import scorex.transaction.History
import scorex.utils.ScorexLogging
import shapeless.Typeable._

import scala.collection.mutable
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.util.{Failure, Success, Try}


//todo: write tests
class HistorySynchronizer(application: Application) extends ViewSynchronizer with ScorexLogging {

  import HistorySynchronizer._
  import application.basicMessagesSpecsRepo._

  private implicit val consensusModule = application.consensusModule
  private implicit val transactionalModule = application.transactionModule

  override val messageSpecs = Seq(ScoreMessageSpec, SignaturesSpec, BlockMessageSpec)

  private lazy val scoreSyncer = new ScoreNetworkObject(self)

  private lazy val history = application.history

  protected override lazy val networkControllerRef = application.networkController

  private lazy val blockGenerator = application.blockGenerator

  //todo: make configurable
  private val GettingExtensionTimeout = 40.seconds
  private val GettingBlockTimeout = 10.seconds

  override def preStart: Unit = {
    super.preStart()
    //todo: make configurable
    context.system.scheduler.schedule(1.second, 2.seconds) {
      val msg = Message(ScoreMessageSpec, Right(history.score()), None)
      networkControllerRef ! NetworkController.SendToNetwork(msg, SendToRandom)
    }
  }

  override def receive: Receive =
    if (application.settings.offlineGeneration) gotoSynced() else gotoSyncing()

  def state(status: Status, logic: Receive): Receive =
    //combine specific logic with common for all the states
    logic orElse ({
      case HistorySynchronizer.GetStatus =>
        sender() ! status.name

      //todo: check sender
      case DataFromPeer(msgId, content: History.BlockchainScore, remote)
        if msgId == ScoreMessageSpec.messageCode =>

        scoreSyncer.networkUpdate(remote, content)

      case ConsideredValue(Some(networkScore: History.BlockchainScore), witnesses) =>
        log.info("Got unhandled ConsideredValue(score)")

      //the signals to initialize
      case Unit =>

      //timeout signals
      case f: FiniteDuration =>

      case nonsense: Any =>
        log.warn(s"Got something strange: $nonsense")
    }: Receive)

  def syncing: Receive = state(HistorySynchronizer.Syncing, {
    case ConsideredValue(Some(networkScore: History.BlockchainScore), witnesses) =>
      val localScore = history.score()
      if (networkScore > localScore) {
        log.info(s"networkScore=$networkScore > localScore=$localScore")
        val lastIds = history.lastBlocks(100).map(_.uniqueId)
        val msg = Message(GetSignaturesSpec, Right(lastIds), None)
        networkControllerRef ! NetworkController.SendToNetwork(msg, SendToChosen(witnesses))
        gotoGettingExtension(witnesses)
      } else gotoSynced()
  }: Receive)

  private val blocksToReceive = mutable.Queue[BlockId]()

  def gettingExtension(witnesses: Seq[ConnectedPeer]): Receive = state(HistorySynchronizer.GettingExtension, {
    case GettingExtensionTimeout => gotoSyncing()

    //todo: aggregating function for block ids (like score has)
    case DataFromPeer(msgId, blockIds: Seq[Block.BlockId]@unchecked, remote)
      if msgId == SignaturesSpec.messageCode &&
        blockIds.cast[Seq[Block.BlockId]].isDefined &&
        witnesses.contains(remote) => //todo: ban if non-expected sender

      val newBLockIds = blockIds.filter(!history.contains(_))
      log.info(s"Got SignaturesMessage with ${blockIds.length} sigs, ${newBLockIds.size} are new")

      val common = blockIds.head
      assert(application.history.contains(common)) //todo: what if not?
      Try(application.blockStorage.removeAfter(common)) //todo we don't need this call for blockTree

      blocksToReceive.clear()

      val newContext = if (newBLockIds.nonEmpty) {
        newBLockIds.foreach { blockId =>
          blocksToReceive += blockId
        }

        networkControllerRef ! NetworkController.SendToNetwork(Message(GetBlockSpec, Right(blocksToReceive.front), None),
          SendToChosen(Seq(remote)))

        gotoGettingBlock(witnesses)
      } else syncing

      context become newContext
  }: Receive)

  def gettingBlock(witnesses: Seq[ConnectedPeer]): Receive = state(HistorySynchronizer.GettingBlock, {
    case GettingBlockTimeout => //15.seconds
      blocksToReceive.clear()
      gotoSyncing()

    case CheckBlock(blockId) =>
      if (blocksToReceive.nonEmpty && blocksToReceive.front.sameElements(blockId)) {
        val sendTo = SendToRandomFromChosen(witnesses)
        val stn = NetworkController.SendToNetwork(Message(GetBlockSpec, Right(blockId), None), sendTo)
        networkControllerRef ! stn
      }

    case DataFromPeer(msgId, block: Block@unchecked, remote)
      if msgId == BlockMessageSpec.messageCode && block.cast[Block].isDefined
        && blocksToReceive.front.sameElements(block.uniqueId) =>

      val blockId = block.uniqueId
      log.info("Got block: " + Base58.encode(blockId))

      if (processNewBlock(block, local = false)) {
        if (blocksToReceive.nonEmpty && blocksToReceive.front.sameElements(blockId)) blocksToReceive.dequeue()

        if (blocksToReceive.nonEmpty) {
          val blockId = blocksToReceive.front
          val ss = SendToRandomFromChosen(Seq(remote))
          val stn = NetworkController.SendToNetwork(Message(GetBlockSpec, Right(blockId), None), ss)
          networkControllerRef ! stn
          context.system.scheduler.scheduleOnce(5.seconds)(self ! CheckBlock(blockId))
        }
      } else if (!history.contains(block.referenceField.value)) {
        log.warn("No parent block in history")
        //blocksToReceive.clear()
        //blocksToReceive.enqueue(block.referenceField.value)
      }

      if (blocksToReceive.nonEmpty) {
        self ! CheckBlock(blocksToReceive.front)
      } else {
        context become syncing
      }
  }: Receive)

  //accept only new block from local or remote
  def synced: Receive = state(HistorySynchronizer.Synced, {
    case block: Block =>
      processNewBlock(block, local = true)

    case ConsideredValue(Some(networkScore: History.BlockchainScore), witnesses) =>
      val localScore = history.score()
      if (networkScore > localScore) {
        blockGenerator ! BlockGenerator.StopGeneration
        gotoGettingExtension(witnesses)
      }

    case DataFromPeer(msgId, block: Block@unchecked, remote)
      if msgId == BlockMessageSpec.messageCode && block.cast[Block].isDefined =>
      processNewBlock(block, local = false)
  }: Receive)

  private def gotoSyncing() = {
    log.debug("Transition to syncing")
    scoreSyncer.consideredValue.foreach(cv => self ! cv)
    context become syncing
    syncing
  }

  private def gotoGettingExtension(witnesses: Seq[ConnectedPeer]): Unit = {
    log.debug("Transition to gettingExtension")
    context.system.scheduler.scheduleOnce(GettingExtensionTimeout)(self ! GettingExtensionTimeout)
    context become gettingExtension(witnesses)
  }

  private def gotoGettingBlock(witnesses: Seq[ConnectedPeer]): Receive = {
    log.debug("Transition to gettingBlock")
    context.system.scheduler.scheduleOnce(GettingBlockTimeout)(self ! GettingBlockTimeout)
    gettingBlock(witnesses)
  }

  private def gotoSynced() = {
    log.debug("Transition to synced")
{'fraction_non_alphanumeric': 0.09385765355866103, 'fraction_numerical': 0.0020799480012999674, 'mean_word_length': 3.9, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 43, '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': '4671018', 'n_tokens_mistral': 4396, 'n_tokens_neox': 4174, 'n_words': 948}