Shared resources

Shared.scala

import cats.effect._
import cats.effect.kernel.Resource
import cats.syntax.all._

trait Shared[F[_], A] {
  def resource: Resource[F, A]
}

object Shared {
  def allocate[F[_], A](resource: Resource[F, A])(implicit F: Concurrent[F]): Resource[F, (Shared[F, A], A)] = {
    final case class State(value: A, finalizer: F[Unit], permits: Int) {
      def addPermit: State = copy(permits = permits + 1)
      def releasePermit: State = copy(permits = permits - 1)
    }
    
    for {
      shared <- Resource.eval(resource.allocated.flatMap { case (a, fin) =>
        F.ref[Option[State]](Some(State(a, fin, 0))).map { state =>
          def acquire: F[A] =
            state.modify {
              case Some(st) => (Some(st.addPermit), F.pure(st.value))
              case None => (None, F.raiseError[A](new Throwable("finalization has already occurred")))
            }.flatten

          def release: F[Unit] =
            state.modify {
              case Some(st) if st.permits > 1 => (Some(st.releasePermit), F.unit)
              case Some(st) => (None, st.finalizer)
              case None => (None, F.raiseError[Unit](new Throwable("can't finalize")))
            }.flatten

          new Shared[F, A] {
            override def resource: Resource[F, A] =
              Resource.make(acquire)(_ => release)
          }
        }
      })
      // i think we can cancel here, that would be wrong
      a <- shared.resource
    } yield (shared, a)
  }
}