# Nimbus
# Copyright (c) 2018-2021 Status Research & Development GmbH
# Licensed and distributed under either of
#   * MIT license (license terms in the root directory or at
#     https://opensource.org/licenses/MIT).
#   * Apache v2 license (license terms in the root directory or at
#     https://www.apache.org/licenses/LICENSE-2.0).
# at your option. This file may not be copied, modified, or distributed
# except according to those terms.

## Fetch and queue blocks
## ======================
##
## Worker items state diagram and sketch of sync algorithm:
## ::
##   unprocessed      |                   | ready for           | store into
##   block ranges     | peer workers      | persistent database | database
##   =======================================================================
##
##        +------------------------------------------+
##        |                                          |
##        |  +----------------------------+          |
##        |  |                            |          |
##        V  v                            |          |
##   <unprocessed> ---+---> <worker-0> ---+-----> <staged> -------> OUTPUT
##                    |                   |
##                    +---> <worker-1> ---+
##                    |                   |
##                    +---> <worker-2> ---+
##                    :                   :
##
## A work item is created from a range of block numbers extracted from the
## `<unprocessed>` set of block ranges.
##
## A work item consists of a
## * current state `<worker-#>` or `<staged>`
## * given range of consecutive block numbers `[from..to]`
## * sequence of block headers relating to `[from..to]` (to be completed)
## * sequence of block buddies relating to `[from..to]` (to be completed)
##
## Block ranges *may* be recycled back into the `<unprocessed>` set when a
## work item is destroyed. This is supposed to be an exceptional case.
## Typically, a `<staged>` work item is added to the persistent block chain
## database and destroyed without block range recycling.
##
## Beware of `<staged>` overflow
## -----------------------------
## When the `<staged>` queue gets too long in non-backtrack/re-org mode, this
## may be caused by a gap between the least `<unprocessed>` block number and
## the least `<staged>` block number. Then a mechanism is invoked where
## `<unprocessed>` block range is updated.
##
## For backtrack/re-org the system runs in single instance mode tracing
## backvards parent hash references. So updating `<unprocessed>` block numbers
## would have no effect. In that case, the record with the largest block
## numbers are deleted from the `<staged>` list.
##

import
  std/[algorithm, options, sequtils, strutils],
  chronicles,
  chronos,
  eth/[common/eth_types, p2p],
  stew/[byteutils, interval_set, sorted_set],
  "../.."/[db/db_chain, utils],
  ".."/[protocol, sync_desc, types]

{.push raises:[Defect].}

logScope:
  topics = "block-queue"

const
  maxStagedWorkItems = 70
    ## Maximal items in the `staged` list.

  stagedWorkItemsTrigger = 50
    ## Turn on the global `poolMode` if there are more than this many items
    ## staged.

type
  BlockQueueRC* = enum
    ## Return & error codes
    AllSmileOk
    EmptyQueue
    StagedQueueOverflow
    BlockNumberGap
    BacktrackDisabled
    FetchHeadersError
    FetchBodiesError
    NoMoreUnprocessed
    NoMorePeerBlocks

  BlockRangeSetRef = IntervalSetRef[BlockNumber,UInt256]
    ## Disjunct sets of block number intervals

  BlockRange = Interval[BlockNumber,UInt256]
    ## Block number interval

  BlockItemQueue = SortedSet[BlockNumber,BlockItemRef]
    ## Block intervals sorted by least block number

  BlockItemWalkRef = SortedSetWalkRef[BlockNumber,BlockItemRef]
    ## Fast traversal descriptor for `BlockItemQueue`

  BlockItemRef* = ref object
    ## Public block items, OUTPUT
    blocks*: BlockRange             ## Block numbers ranvge covered
    topHash*: Option[Hash256]       ## Fetched by top hash rather than block
    headers*: seq[BlockHeader]      ## Block headers received
    hashes*: seq[Hash256]           ## Hashed from `headers[]` for convenience
    bodies*: seq[BlockBody]         ## Block bodies received

  BlockQueueCtxRef* = ref object
    ## Globally shared data among `block` instances
    backtrack: Option[Hash256]      ## Find reverse block after re-org
    unprocessed: BlockRangeSetRef   ## Block ranges to fetch
    staged: BlockItemQueue          ## Blocks fetched but not stored yet
    topAccepted: BlockNumber        ## Up to this block number processed OK

  BlockQueueWorkerRef* = ref object
    ## Local descriptor data extension
    global: BlockQueueCtxRef        ## Common data
    bestNumber: Option[BlockNumber] ## Largest block number reported
    ctrl: BuddyCtrlRef              ## Control and state settings
    peer: Peer                      ## network peer

  BlockQueueStats* = object
    ## Statistics
    topAccepted*: BlockNumber
    nextUnprocessed*: Option[BlockNumber]
    nextStaged*: Option[BlockNumber]
    nStagedQueue*: int
    reOrg*: bool

let
  highBlockNumber = high(BlockNumber)
  highBlockRange = BlockRange.new(highBlockNumber,highBlockNumber)

static:
  doAssert stagedWorkItemsTrigger < maxStagedWorkItems

# ------------------------------------------------------------------------------
# Private helpers
# ------------------------------------------------------------------------------

proc `+`(n: BlockNumber; delta: static[int]): BlockNumber =
  ## Syntactic sugar for expressions like `xxx.toBlockNumber + 1`
  n + delta.toBlockNumber

proc `-`(n: BlockNumber; delta: static[int]): BlockNumber =
  ## Syntactic sugar for expressions like `xxx.toBlockNumber - 1`
  n - delta.toBlockNumber

proc merge(ivSet: BlockRangeSetRef; wi: BlockItemRef): Uint256 =
  ## Syntactic sugar
  ivSet.merge(wi.blocks)

proc reduce(ivSet: BlockRangeSetRef; wi: BlockItemRef): Uint256 =
  ## Syntactic sugar
  ivSet.reduce(wi.blocks)

# ---------------

proc pp(n: BlockNumber): string =
  ## Dedicated pretty printer (`$` is defined elsewhere using `UInt256`)
  if n == highBlockNumber: "high" else:"#" & $n

proc `$`(iv: BlockRange): string =
  ## Needed for macro generated DSL files like `snap.nim` because the
  ## `distinct` flavour of `NodeTag` is discarded there.
  result = "[" & iv.minPt.pp
  if iv.minPt != iv.maxPt:
    result &= "," & iv.maxPt.pp
  result &= "]"

proc `$`(n: Option[BlockRange]): string =
  if n.isNone: "n/a" else: $n.get

proc `$`(n: Option[BlockNumber]): string =
  if n.isNone: "n/a" else: n.get.pp

proc `$`(brs: BlockRangeSetRef): string =
  "{" & toSeq(brs.increasing).mapIt($it).join(",") & "}"

# ------------------------------------------------------------------------------
# Private helpers
# ------------------------------------------------------------------------------

proc nextUnprocessed(ctx: BlockQueueCtxRef): Option[BlockNumber] =
  ## Pseudo getter
  let rc = ctx.unprocessed.ge()
  if rc.isOK:
    result = some(rc.value.minPt)

proc nextStaged(ctx: BlockQueueCtxRef): Option[BlockRange] =
  ## Pseudo getter
  let rc = ctx.staged.ge(low(BlockNumber))
  if rc.isOK:
    result = some(rc.value.data.blocks)

template safeTransport(
    qd: BlockQueueWorkerRef;
    info: static[string];
    code: untyped) =
  try:
    code
  except TransportError as e:
    error info & ", stop", error=($e.name), msg=e.msg
    qd.ctrl.stopped = true

# ------------------------------------------------------------------------------
# Private functions
# ------------------------------------------------------------------------------

proc newWorkItem(qd: BlockQueueWorkerRef): Result[BlockItemRef,BlockQueueRC] =
  ## Fetch the next unprocessed block range and register it as work item.
  ##
  ## This function will grab a block range from the `unprocessed` range set,
  ## ove it and return it as a `BlockItemRef`. The returned range is registered
  ## in the `pending` list.
  let rc = qd.global.unprocessed.ge()
  if rc.isErr:
      return err(NoMoreUnprocessed) # no more data for this peer

  # Check whether there is somthing to do at all
  if qd.bestNumber.isNone or
     qd.bestNumber.unsafeGet < rc.value.minPt:
    return err(NoMorePeerBlocks) # no more data for this peer

  # Compute interval
  let iv = BlockRange.new(
    rc.value.minPt,
    min(rc.value.maxPt,
        min(rc.value.minPt + maxHeadersFetch - 1, qd.bestNumber.unsafeGet)))

  discard qd.global.unprocessed.reduce(iv)
  ok(BlockItemRef(blocks: iv))


proc stageItem(
    qd: BlockQueueWorkerRef;
    wi: BlockItemRef;
      ): Result[void,BlockQueueRC] =
  ## Add work item to the list of staged items
  ##
  ## Typically, the function returns `AllSmileOk` unless there is a queue
  ## oberflow (with return code`StagedQueueOverflow`) which needs to be handled
  ## in *pool mode* by running `blockQueueGrout()`.
  var
    error = AllSmileOk
  let
    peer = qd.peer
    rc = qd.global.staged.insert(wi.blocks.minPt)
  if rc.isOk:
    rc.value.data = wi

    # Return `true` if staged queue oberflows (unless backtracking.)
    if stagedWorkItemsTrigger < qd.global.staged.len and
       qd.global.backtrack.isNone and
       wi.topHash.isNone:
      debug "Staged queue too long", peer,
        staged=qd.global.staged.len, max=stagedWorkItemsTrigger
      error = StagedQueueOverflow

    # The list size is limited. So cut if necessary and recycle back the block
    # range of the discarded item (tough luck if the current work item is the
    # one removed from top.)
    while maxStagedWorkItems < qd.global.staged.len:
      let topValue = qd.global.staged.le(highBlockNumber).value
      discard qd.global.unprocessed.merge(topValue.data)
      discard qd.global.staged.delete(topValue.key)
  else:
    # Ooops, duplicates should not exist (but anyway ...)
    let wj = block:
      let rc = qd.global.staged.eq(wi.blocks.minPt)
      doAssert rc.isOk
      # Store `wi` and return offending entry
      let rcData = rc.value.data
      rc.value.data = wi
      rcData

    # Update `staged` list and `unprocessed` ranges
    block:
      debug "Replacing dup item in staged list", peer,
        range=($wi.blocks), discarded=($wj.blocks)
      let rc = wi.blocks - wj.blocks
      if rc.isOk:
        discard qd.global.unprocessed.merge(rc.value)

  if error != AllSmileOk:
    return err(error)
  ok()

# ------------------------------------------------------------------------------
# Private functions, asynchroneous data network activity
# ------------------------------------------------------------------------------

proc fetchHeaders(
    qd: BlockQueueWorkerRef;
    wi: BlockItemRef;
      ): Future[bool]
      {.async.} =
  ## Get the work item with the least interval and complete it. The function
  ## returns `true` if bodies were fetched and there were no inconsistencies.
  if 0 < wi.hashes.len:
    return true

  let peer = qd.peer
  var hdrReq: BlocksRequest
  if wi.topHash.isNone:
    hdrReq = BlocksRequest(
      startBlock: HashOrNum(
        isHash:   false,
        number:   wi.blocks.minPt),
      maxResults: wi.blocks.len.truncate(uint),
      skip:       0,
      reverse:    false)
    trace trEthSendSendingGetBlockHeaders, peer,
      blocks=($wi.blocks)

  else:
    hdrReq = BlocksRequest(
      startBlock: HashOrNum(
        isHash:   true,
        hash:     wi.topHash.get),
      maxResults: maxHeadersFetch,
      skip:       0,
      reverse:    true)
    trace trEthSendSendingGetBlockHeaders & " reverse", peer,
      topHash=hdrReq.startBlock.hash, reqLen=hdrReq.maxResults

  # Fetch headers from peer
  var hdrResp: Option[blockHeadersObj]
  block:
    let reqLen = hdrReq.maxResults
    qd.safeTransport("Error fetching block headers"):
      hdrResp = await peer.getBlockHeaders(hdrReq)
    # Beware of peer terminating the session
    if qd.ctrl.stopped:
      return false

    if hdrResp.isNone:
      trace trEthRecvReceivedBlockHeaders, peer, reqLen, respose="n/a"
      return false

    let hdrRespLen = hdrResp.get.headers.len
    trace trEthRecvReceivedBlockHeaders, peer, reqLen, hdrRespLen

    if hdrRespLen == 0:
      qd.ctrl.stopped = true
      return false

  # Update block range for reverse search
  if wi.topHash.isSome:
    # Headers are in reversed order
    wi.headers = hdrResp.get.headers.reversed
    wi.blocks = BlockRange.new(
      wi.headers[0].blockNumber, wi.headers[^1].blockNumber)
    discard qd.global.unprocessed.reduce(wi)
    trace "Updated reverse header range", peer, range=($wi.blocks)

  # Verify start block number
  elif hdrResp.get.headers[0].blockNumber != wi.blocks.minPt:
    trace "Header range starts with wrong block number", peer,
      startBlock=hdrResp.get.headers[0].blockNumber,
      requestedBlock=wi.blocks.minPt
    qd.ctrl.zombie = true
    return false

  # Import into `wi.headers`
  else:
    wi.headers.shallowCopy(hdrResp.get.headers)

  # Calculate block header hashes and verify it against parent links. If
  # necessary, cut off some offending block headers tail.
  wi.hashes.setLen(wi.headers.len)
  wi.hashes[0] = wi.headers[0].hash
  for n in 1 ..< wi.headers.len:
    if wi.headers[n-1].blockNumber + 1 != wi.headers[n].blockNumber:
      trace "Non-consecutive block numbers in header list response", peer
      qd.ctrl.zombie = true
      return false
    if wi.hashes[n-1] != wi.headers[n].parentHash:
      # Oops, cul-de-sac after block chain re-org?
      trace "Dangling parent link in header list response. Re-org?", peer
      wi.headers.setLen(n)
      wi.hashes.setLen(n)
      break
    wi.hashes[n] = wi.headers[n].hash

  # Adjust range length if necessary
  if wi.headers[^1].blockNumber < wi.blocks.maxPt:
    let redRng = BlockRange.new(
      wi.headers[0].blockNumber, wi.headers[^1].blockNumber)
    trace "Adjusting block range", peer, range=($wi.blocks), reduced=($redRng)
    discard qd.global.unprocessed.merge(redRng.maxPt + 1, wi.blocks.maxPt)
    wi.blocks = redRng

  return true


proc fetchBodies(
    qd: BlockQueueWorkerRef;
    wi: BlockItemRef
      ): Future[bool]
      {.async.} =
  ## Get the work item with the least interval and complete it. The function
  ## returns `true` if bodies were fetched and there were no inconsistencies.
  let peer = qd.peer

  # Complete group of bodies
  qd.safeTransport("Error fetching block bodies"):
    while wi.bodies.len < wi.hashes.len:
      let
        start = wi.bodies.len
        reqLen = min(wi.hashes.len - wi.bodies.len, maxBodiesFetch)
        top = start + reqLen
        hashes =  wi.hashes[start ..< top]

      trace trEthSendSendingGetBlockBodies, peer, reqLen

      # Append bodies from peer to `wi.bodies`
      block:
        let bdyResp = await peer.getBlockBodies(hashes)
        # Beware of peer terminating the session
        if qd.ctrl.stopped:
          return false

        if bdyResp.isNone:
          trace trEthRecvReceivedBlockBodies, peer, reqLen, respose="n/a"
          qd.ctrl.zombie = true
          return false

        let bdyRespLen = bdyResp.get.blocks.len
        trace trEthRecvReceivedBlockBodies, peer, reqLen, bdyRespLen

        if bdyRespLen == 0 or reqLen < bdyRespLen:
          qd.ctrl.zombie = true
          return false

        wi.bodies.add bdyResp.get.blocks

    return true


# ------------------------------------------------------------------------------
# Public functions, constructor
# ------------------------------------------------------------------------------

proc init*(
    T: type BlockQueueCtxRef;           ## Global data descriptor type
    firstBlockNumber = 0.toBlockNumber; ## Of first block to fetch from network
      ): T =
  ## Global constructor, shared data
  result = T(
    unprocessed: BlockRangeSetRef.init())
  result.staged.init()
  result.topAccepted = max(firstBlockNumber,1.toBlockNumber) - 1
  discard result.unprocessed.merge(result.topAccepted + 1, highBlockNumber)

proc init*(
    T: type BlockQueueWorkerRef;        ## Global data descriptor type
    ctx: BlockQueueCtxRef;              ## Global data descriptor
    ctrl: BuddyCtrlRef;                 ## Control and state settings
    peer: Peer;                         ## For fetching data from network
      ): T =
  ## Buddy/local constructor
  T(global: ctx,
    peer:   peer,
    ctrl:   ctrl)

# ------------------------------------------------------------------------------
# Public functions -- synchronous
# ------------------------------------------------------------------------------

proc blockQueueFetchStaged*(
    qd: BlockQueueWorkerRef;
      ): Result[BlockItemRef,BlockQueueRC]=
  ## Fetch the next item from the staged block queue. This item will be removed
  ## from the staged queue and must be recycled if it cannot be processed.
  ##
  ## On error, the function returns `EmptyQueue` if the queue was empty and
  ## `BlockNumberGap` if processing this item would result in a gap between the
  ## last accepted block number and the fitsr block number of the next queue
  ## item.
  ##
  ## This gap might appear if another function processes the in-beween block
  ## in paralell or if something went wrong, see `blockQueueGrout()`, below.
  let rc = qd.global.staged.ge(low(BlockNumber))
  if rc.isErr:
    # No more items in the database
    return err(EmptyQueue)

  let
    peer = qd.peer
    wi = rc.value.data
    topAccepted = qd.global.topAccepted
    startNumber = wi.headers[0].blockNumber

  # Check whether this record of blocks can be stored, at all
  if topAccepted + 1 < startNumber:
    trace "Staged work item postponed", peer, topAccepted,
      range=($wi.blocks), staged=qd.global.staged.len
    return err(BlockNumberGap)

  # Ok, store into the block chain database
  trace "Staged work item", peer,
    topAccepted, range=($wi.blocks)

  # Remove from staged DB
  discard qd.global.staged.delete(wi.blocks.minPt)

  ok(wi)


proc blockQueueAccept*(qd: BlockQueueWorkerRef; wi: BlockItemRef) =
  ## Mark this argument item `wi` to be the item with the topmost block number
  ## accepted. This statement comes tyipcally after the successful processing
  ## and storage of the work item fetched by `blockQueueFetchStaged()`.
  qd.global.topAccepted = wi.blocks.maxPt


proc blockQueueGrout*(qd: BlockQueueWorkerRef) =
  ## Fill the gap unprocessed and staged block numbers. If there is such a gap
  ## (which should not at all), the `blockQueueFetchStaged()` will always fail
  ## with a `true` error code because there is no next work item.
  ##
  ## To close the gap and avoid double processing, all other workers should
  ## have finished their tasks while this function is run. A way to achive that
  ## is to run this function in *pool mode* once.
  # Mind the gap, fill in if necessary
  let covered = min(
    qd.global.nextUnprocessed.get(otherwise = highBlockNumber),
    qd.global.nextStaged.get(otherwise = highBlockRange).minPt)
  if qd.global.topAccepted + 1 < covered:
    discard qd.global.unprocessed.merge(qd.global.topAccepted + 1, covered - 1)


proc blockQueueRecycle*(qd: BlockQueueWorkerRef; wi: BlockItemRef) =
  ## Put back and destroy the `wi` argument item. The covered block range needs
  ## to be re-fetched from the network. This statement is typically used instead
  ## of `blockQueueAccept()` after a failure tpo process and store the work item
  ## fetched by `blockQueueFetchStaged()`.
  discard qd.global.unprocessed.merge(wi.blocks)


proc blockQueueRecycleStaged*(qd: BlockQueueWorkerRef) =
  ## Similar to `blockQueueRecycle()`, recycle all items from the staged queue.
  # using fast traversal
  let
    walk = BlockItemWalkRef.init(qd.global.staged)
  var
    rc = walk.first()
  while rc.isOk:
    # Store back into `unprocessed` ranges set
    discard qd.global.unprocessed.merge(rc.value.data)
    rc = walk.next()
  # optional clean up, see comments on the destroy() directive
  walk.destroy()
  qd.global.staged.clear()


proc blockQueueBacktrackFrom*(qd: BlockQueueWorkerRef; wi: BlockItemRef) =
  ## Set backtrack mode starting with the blocks before the argument work
  ## item `wi`.
  qd.global.backtrack = some(wi.headers[0].parentHash)


proc blockQueueBacktrackOk*(qd: BlockQueueWorkerRef): bool =
  ## Returns `true` if the queue is in backtrack mode.
  qd.global.backtrack.isSome


proc blockQueueStats*(ctx: BlockQueueCtxRef; stats: var BlockQueueStats) =
  ## Collect global statistics
  stats.topAccepted = ctx.topAccepted
  stats.nextUnprocessed = ctx.nextUnprocessed
  stats.nStagedQueue = ctx.staged.len
  stats.reOrg = ctx.backtrack.isSome
  stats.nextStaged =
    if ctx.nextStaged.isSome: some(ctx.nextStaged.unsafeGet.minPt)
    else: none(BlockNumber)

# ------------------------------------------------------------------------------
# Public functions -- asynchronous
# ------------------------------------------------------------------------------

proc blockQueueBacktrackWorker*(
    qd: BlockQueueWorkerRef;
      ): Future[Result[void,BlockQueueRC]]
      {.async.} =
  ## This function does some backtrack processing on the queue. Backtracking
  ## is single threaded due to the fact that the next block is identified by
  ## the hash of the parent header. So this function needs to run in *single
  ## mode*.
  ##
  ## If backtracking is enabled, this function fetches the next parent work
  ## item from the network and makes it available on the staged queue to be
  ## retrieved with `blockQueueFetchStaged()`. In that case, the function
  ## succeeds and `blockQueueBacktrackOk()` will return `false`.
  ##
  ## In all other cases, the function returns an error code.
  var error = BacktrackDisabled
  if qd.global.backtrack.isSome:
    let
      peer = qd.peer
      wi = BlockItemRef(
        # This dummy interval can savely merged back without any effect
        blocks:  highBlockRange,
        # Enable backtrack
        topHash: some(qd.global.backtrack.unsafeGet))

    # Fetch headers and bodies for the current work item
    trace "Single mode worker, re-org backtracking", peer
    if not await qd.fetchHeaders(wi):
      error = FetchHeadersError
    elif not await qd.fetchBodies(wi):
      error = FetchBodiesError
    else:
      qd.global.backtrack = none(Hash256)
      discard qd.stageItem(wi)
      return ok()

    # This work item failed, nothing to do anymore.
    discard qd.global.unprocessed.merge(wi)

  return err(error)


proc blockQueueWorker*(
    qd: BlockQueueWorkerRef;
      ): Future[Result[void,BlockQueueRC]]
      {.async.} =
  ## Normal worker function used to stage another work item be retrieved by
  ## `blockQueueFetchStaged()`. This function may run in *multi mode*. Not
  ## until retrieving work items the queue will be synchronised in a way that
  ## after the next item can be retrieved the queue will be blocked by a
  ## *gap* until the item is commited by `blockQueueAccept()`.
  ##
  ## On error, with most error codes there is not much that can be done. The
  ## one remarcable error code is `StagedQueueOverflow` which pops up if there
  ## is a gap between unprocessed and staged block numbers. One of the actions
  ## to be considered here is to run `blockQueueGrout()` in *pool mode*.
  ## Otherwise, the `StagedQueueOverflow` can be treated as a success would be.
  ##
  # Fetch work item
  let wi = block:
    let rc = qd.newWorkItem()
    if rc.isErr:
      # No way, end of capacity for this peer => re-calibrate
      qd.bestNumber = none(BlockNumber)
      return err(rc.error)
    rc.value

  # Fetch headers and bodies for the current work item
  var error = AllSmileOk
  if not await qd.fetchHeaders(wi):
    error = FetchHeadersError
  elif not await qd.fetchBodies(wi):
    error = FetchBodiesError
  else:
    return qd.stageItem(wi)

  # This work item failed
  discard qd.global.unprocessed.merge(wi)
  return err(error)

# ------------------------------------------------------------------------------
# End
# ------------------------------------------------------------------------------