package tf.bug.aoc.y2020

import cats.effect._
import cats.syntax.all._
import fs2._
import tf.bug.aoc.Day

object Day3 extends Day {

  case class Path(x: Int, treeCount: Long)

  override def part1[F[_]: Async]: Pipe[F, String, String] =
    _.through(checkSlope[F](3, 1)).map(_.show)

  override def part2[F[_]: Async]: Pipe[F, String, String] =
    (lines: Stream[F, String]) => {
      val pipes: Vector[Pipe[F, String, Long]] = Vector(
        checkSlope[F](1, 1),
        checkSlope[F](3, 1),
        checkSlope[F](5, 1),
        checkSlope[F](7, 1),
        checkSlope[F](1, 2),
      )
      val broadcast: Stream[F, String] => Stream[F, Stream[F, String]] = concurrent.Broadcast[F, String](pipes.size)
      val channels: Stream[F, Stream[F, String]] = lines.through(broadcast).take(pipes.size)
      val results: Stream[F, Long] = channels.zipWithIndex.map {
        case (channel, ind) =>
          val pipe = pipes(ind.toInt)
          channel.through(pipe)
      }.parJoin(pipes.size)
      results.fold(1L)(_ * _).map(_.show)
    }

  def checkSlope[F[_]](right: Int, down: Int): Pipe[F, String, Long] =
    (lines: Stream[F, String]) => {
      lines.chunkN(down, true).fold(Path(0, 0L)) {
        case (Path(x, numTrees), group) =>
          group.head match {
            case Some(line) =>
              val isTree = line.charAt(x % line.length) == '#'
              Path(x + right, if(isTree) numTrees + 1L else numTrees)
            case None =>
              Path(x, numTrees)
          }
      }.map(_.treeCount)
    }

}