block23.scala

.tap(x => log.info(x.value.toString))
.foreachChunk { chunk =>
  val offsetBatch = OffsetBatch(chunk.map(_.offset))
  val offsets = offsetBatch.offsets
  val offsetsList = offsets
    .map { case (topicPartition, offset) => partitionToString(topicPartition) -> offset }
    .toList
  val commands = chunk.map(_.value)
  session.transaction.toManagedZIO.use { transaction =>
    for {
      _ <- ZIO.foreach_(commands) {
        case TransferBalance(from, to, amount) =>
          for {
            _ <- session
              .prepare(Sql.changeBalanceCommand)
              .toManagedZIO
              .use(_.execute(from ~ -amount)) // notice the minus sign
            
            _ <- session
              .prepare(Sql.changeBalanceCommand)
              .toManagedZIO
              .use(_.execute(to ~ amount))
          } yield ()
      }
      
      _ <- session
        .prepare(Sql.updatePartitionCommand(offsetsList))
        .toManagedZIO
        .use(_.execute(offsetsList))
      
      _ <- transaction.commit
    } yield ()
  }
}