Improvements to ZQuery

core/src/main/scala/caliban/execution/Executor.scala

@@ -118,9 +118,10 @@ object Executor {
           reduceObject(items)
         case QueryStep(inner) =>
           ReducedStep.QueryStep(
-            inner
-              .map(reduceStep(_, currentField, arguments))
-              .mapError(GenericSchema.effectfulExecutionError(currentField.name, _))
+            inner.bimap(
+              GenericSchema.effectfulExecutionError(currentField.name, _),
+              reduceStep(_, currentField, arguments)
+            )
           )
         case StreamStep(stream) =>
           ReducedStep.StreamStep(
@@ -142,10 +143,10 @@ object Executor {
             val queries = steps.map { case (name, field) => loop(field).map(name -> _) }
             (if (allowParallelism) ZQuery.collectAllPar(queries) else ZQuery.collectAll(queries)).map(ObjectValue)
           case ReducedStep.QueryStep(step) =>
-            step.fold(Left(_), Right(_)).flatMap {
-              case Left(error)  => ZQuery.fromEffect(errors.update(error :: _)).map(_ => NullValue)
-              case Right(query) => loop(query)
-            }
+            step.foldM(
+              error => ZQuery.fromEffect(errors.update(error :: _)).map(_ => NullValue),
+              query => loop(query)
+            )
           case ReducedStep.StreamStep(stream) =>
             ZQuery
               .fromEffect(ZIO.environment[R])

core/src/main/scala/zquery/BlockedRequestMap.scala

@@ -7,13 +7,13 @@ import zio.ZIO
  * requests from those data sources.
  */
 private[zquery] final class BlockedRequestMap[-R](
-  private val map: Map[DataSource.Service[Any, Any], Vector[BlockedRequest[Any]]]
+  private val map: Map[DataSource[Any, Any], Vector[BlockedRequest[Any]]]
 ) { self =>
 
   def ++[R1 <: R](that: BlockedRequestMap[R1]): BlockedRequestMap[R1] =
     new BlockedRequestMap(
       (self.map.toVector ++ that.map.toVector)
-        .foldLeft[Map[DataSource.Service[Any, Any], Vector[BlockedRequest[Any]]]](Map()) {
+        .foldLeft[Map[DataSource[Any, Any], Vector[BlockedRequest[Any]]]](Map()) {
           case (acc, (key, value)) =>
             acc + (key -> acc.get(key).fold(value)(_ ++ value))
         }
@@ -24,8 +24,8 @@ private[zquery] final class BlockedRequestMap[-R](
    * can change the environment and error types of data sources but must
    * preserve the request type of each data source.
    */
-  final def mapDataSources[R1](f: DataSourceFunction[R, R1]): BlockedRequestMap[R1] =
-    new BlockedRequestMap(self.map.map { case (k, v) => (f(k).asInstanceOf[DataSource.Service[Any, Any]], v) })
+  def mapDataSources[R1](f: DataSourceFunction[R, R1]): BlockedRequestMap[R1] =
+    new BlockedRequestMap(self.map.map { case (k, v) => (f(k).asInstanceOf[DataSource[Any, Any]], v) })
 
   /**
    * Executes all requests, submitting batched requests to each data source in
@@ -50,18 +50,18 @@ object BlockedRequestMap {
    * specified data source to the specified request.
    */
   def apply[R, E, K](
-    dataSource: DataSource.Service[R, K],
+    dataSource: DataSource[R, K],
     blockedRequest: BlockedRequest[K]
   ): BlockedRequestMap[R] =
     new BlockedRequestMap(
-      Map(dataSource.asInstanceOf[DataSource.Service[Any, Any]] -> Vector(blockedRequest))
+      Map(dataSource.asInstanceOf[DataSource[Any, Any]] -> Vector(blockedRequest))
     )
 
   /**
    * An empty blocked requests map.
    */
   val empty: BlockedRequestMap[Any] =
     new BlockedRequestMap(
-      Map.empty[DataSource.Service[Any, Any], Vector[BlockedRequest[Any]]]
+      Map.empty[DataSource[Any, Any], Vector[BlockedRequest[Any]]]
     )
 }

core/src/main/scala/zquery/Cache.scala

@@ -1,20 +1,20 @@
 package zquery
 
-import zio.{ Ref, UIO }
+import zio.{ IO, Ref, UIO }
 
 /**
  * A `Cache` maintains an internal state with a mapping from requests to `Ref`s
  * that will contain the result of those requests when they are executed. This
  * is used internally by the library to provide deduplication and caching of
  * requests.
  */
-class Cache private (private val state: Ref[Map[Any, Any]]) {
+final class Cache private (private val state: Ref[Map[Any, Any]]) {
 
   /**
    * Inserts a request and a `Ref` that will contain the result of the request
    * when it is executed into the cache.
    */
-  final def insert[E, A](request: Request[E, A], result: Ref[Option[Either[E, A]]]): UIO[Unit] =
+  def insert[E, A](request: Request[E, A], result: Ref[Option[Either[E, A]]]): UIO[Unit] =
     state.update(_ + (request -> result)).unit
 
   /**
@@ -23,8 +23,8 @@ class Cache private (private val state: Ref[Map[Any, Any]]) {
    * been executed yet, or `Some(Ref(Some(value)))` if the request has been
    * executed.
    */
-  final def lookup[E, A](request: Request[E, A]): UIO[Option[Ref[Option[Either[E, A]]]]] =
-    state.get.map(_.get(request).asInstanceOf[Option[Ref[Option[Either[E, A]]]]])
+  def lookup[E, A](request: Request[E, A]): IO[Unit, Ref[Option[Either[E, A]]]] =
+    state.get.map(_.get(request).asInstanceOf[Option[Ref[Option[Either[E, A]]]]]).get
 }
 
 object Cache {

core/src/main/scala/zquery/CompletedRequestMap.scala

@@ -11,19 +11,19 @@ package zquery
  */
 final class CompletedRequestMap private (private val map: Map[Any, Either[Any, Any]]) { self =>
 
-  final def ++(that: CompletedRequestMap): CompletedRequestMap =
+  def ++(that: CompletedRequestMap): CompletedRequestMap =
     new CompletedRequestMap(self.map ++ that.map)
 
   /**
    * Appends the specified result to the completed requests map.
    */
-  final def insert[E, A](request: Request[E, A])(result: Either[E, A]): CompletedRequestMap =
+  def insert[E, A](request: Request[E, A])(result: Either[E, A]): CompletedRequestMap =
     new CompletedRequestMap(self.map + (request -> result))
 
   /**
    * Retrieves the result of the specified request if it exists.
    */
-  final def lookup[E, A](request: Request[E, A]): Option[Either[E, A]] =
+  def lookup[E, A](request: Request[E, A]): Option[Either[E, A]] =
     map.get(request).asInstanceOf[Option[Either[E, A]]]
 }