object PageRank {

  def main(args: Array[String]): Unit = {
    val graph = Map(
      "A" -> List("B", "C"),
      "B" -> List("C"),
      "C" -> List("A")
    )

    val init = Map(
      "A" -> 1.0,
      "B" -> 1.0,
      "C" -> 1.0
    )

    val d = 0.5

    println(init)

    val result: Map[String, Double] = pageRank(graph, init, d)

    println(result)
  }

  def pageRank(graph: Map[String, List[String]], initPageRanks: Map[String, Double], damping: Double): Map[String, Double] = {
    val pageRanks: Stream[Map[String, Double]] =
      Stream.continually(1)
        .scanLeft(initPageRanks) {
          case (currentPageRanks, _) =>
            val newPageRanks =
              for {currentNode <- graph.keys}
                yield currentNode -> ((1 - damping) + damping * computeNewRank(graph, currentNode, currentPageRanks))

            newPageRanks.toMap
        }

    val result =
      pageRanks
        .zip(pageRanks.tail)
        .takeWhile {
          case (pr1, pr2) =>
            pageRanksEqual(pr1, pr2)
        }
        .last._1

    result
  }

  def computeNewRank(graph: Map[String, List[String]],
                     currentNode: String,
                     currentPageRanks: Map[String, Double]): Double =
    referringNodes(graph, currentNode)
      .map(node => currentPageRanks(node) / graph(node).size)
      .sum

  def referringNodes(graph: Map[String, List[String]],
                     currentNode: String): Iterable[String] =
    for {
      (node, neighboors) <- graph
      if (currentNode != node) && (neighboors contains currentNode)
    }
      yield node

  def pageRanksEqual(pr1: Map[String, Double], pr2: Map[String, Double]): Boolean = {
    val r =
      for {key <- pr1.keys.toList}
        yield (pr1(key) - pr2(key)).abs < 1e-7

    !r.forall(identity)
  }
}