Http4s Module upload support (#945)

* Initial Upload support

* Make this PR work

* Make uploadService independent

* Generate UUID via ZIO Random

* Change rootUploadPath type

* Code cleanup

* Use Random.Service

* Move uploads into the core project

* Add Http4sAdapterSpec

* use a test cope variable instead of a string

* Change http4s tests blaze bindings

* Enable http4s tests in CI

* Get rid of asInstanceOf in tests

.circleci/config.yml

@@ -21,7 +21,7 @@ jobs:
       - checkout
       - restore_cache:
           key: sbtcache
-      - run: sbt ++2.12.13! core/test http4s/compile akkaHttp/test finch/compile play/test zioHttp/compile examples/compile catsInterop/compile benchmarks/compile tools/test codegenSbt/test clientJVM/test monixInterop/compile tapirInterop/test federation/test codegenSbt/scripted
+      - run: sbt ++2.12.13! core/test http4s/test akkaHttp/test finch/compile play/test zioHttp/compile examples/compile catsInterop/compile benchmarks/compile tools/test codegenSbt/test clientJVM/test monixInterop/compile tapirInterop/test federation/test codegenSbt/scripted
       - save_cache:
           key: sbtcache
           paths:
@@ -35,7 +35,7 @@ jobs:
       - checkout
       - restore_cache:
           key: sbtcache
-      - run: sbt ++2.13.6! core/test http4s/compile akkaHttp/test finch/compile play/test zioHttp/compile examples/compile catsInterop/compile monixInterop/compile tapirInterop/test clientJVM/test federation/test
+      - run: sbt ++2.13.6! core/test http4s/test akkaHttp/test finch/compile play/test zioHttp/compile examples/compile catsInterop/compile monixInterop/compile tapirInterop/test clientJVM/test federation/test
       - save_cache:
           key: sbtcache
           paths:

adapters/http4s/src/main/scala/caliban/Http4sAdapter.scala

@@ -3,33 +3,47 @@ package caliban
 import caliban.ResponseValue.{ ObjectValue, StreamValue }
 import caliban.Value.NullValue
 import caliban.execution.QueryExecution
+import caliban.uploads._
 import cats.arrow.FunctionK
 import cats.data.{ Kleisli, OptionT }
-import cats.effect.Effect
+import cats.syntax.either._
+import cats.syntax.traverse._
+import cats.effect.{ Blocker, Effect }
 import cats.effect.syntax.all._
 import cats.~>
 import fs2.{ Pipe, Stream }
+import fs2.text.utf8Decode
 import io.circe.Decoder.Result
-import io.circe.Json
+import io.circe.{ DecodingFailure, Json }
 import io.circe.parser._
 import io.circe.syntax._
 import org.http4s._
 import org.http4s.circe.CirceEntityCodec._
 import org.http4s.dsl.Http4sDsl
+import org.http4s.headers.`Content-Disposition`
 import org.http4s.implicits._
 import org.http4s.server.websocket.WebSocketBuilder
 import org.http4s.websocket.WebSocketFrame
 import org.http4s.websocket.WebSocketFrame.Text
+import org.http4s.multipart.{ Multipart, Part }
 import zio.Exit.Failure
 import zio._
 import zio.clock.Clock
 import zio.duration.Duration
 import zio.interop.catz._
+import zio.random.Random
+
+import java.io.File
+import java.nio.file.Path
+import scala.util.Try
 
 object Http4sAdapter {
 
   val `application/graphql`: MediaType = mediaType"application/graphql"
 
+  private def parsePath(path: String): List[Either[String, Int]] =
+    path.split('.').map(c => Try(c.toInt).toEither.left.map(_ => c)).toList
+
   private def executeToJson[R, E](
     interpreter: GraphQLInterpreter[R, E],
     request: GraphQLRequest,
@@ -46,6 +60,24 @@ object Http4sAdapter {
       )
       .foldCause(cause => GraphQLResponse(NullValue, cause.defects).asJson, _.asJson)
 
+  private def executeToJsonWithUpload[R <: Has[_], E](
+    interpreter: GraphQLInterpreter[R, E],
+    request: GraphQLRequest,
+    skipValidation: Boolean,
+    enableIntrospection: Boolean,
+    queryExecution: QueryExecution,
+    fileHandle: ZLayer[Any, Nothing, Uploads]
+  ): URIO[R, Json] =
+    interpreter
+      .executeRequest(
+        request,
+        skipValidation = skipValidation,
+        enableIntrospection = enableIntrospection,
+        queryExecution
+      )
+      .foldCause(cause => GraphQLResponse(NullValue, cause.defects).asJson, _.asJson)
+      .provideSomeLayer[R](fileHandle)
+
   @deprecated("Use makeHttpService instead", "0.4.0")
   def makeRestService[R, E](interpreter: GraphQLInterpreter[R, E]): HttpRoutes[RIO[R, *]] =
     makeHttpService(interpreter)
@@ -75,6 +107,149 @@ object Http4sAdapter {
       params.get("extensions")
     )
 
+  private def parseGraphQLRequest(body: String): Result[GraphQLRequest] =
+    parse(body).flatMap(_.as[GraphQLRequest]).leftMap(e => DecodingFailure(e.getMessage, Nil))
+
+  private def parsePaths(map: Map[String, Seq[String]]): List[(String, List[Either[String, Int]])] =
+    map.map { case (k, v) => k -> v.map(parsePath).toList }.toList.flatMap(kv => kv._2.map(kv._1 -> _))
+
+  def makeHttpUploadService[R <: Has[_] with Random, E](
+    interpreter: GraphQLInterpreter[R, E],
+    rootUploadPath: Path,
+    blocker: Blocker,
+    skipValidation: Boolean = false,
+    enableIntrospection: Boolean = true,
+    queryExecution: QueryExecution = QueryExecution.Parallel
+  ): HttpRoutes[RIO[R, *]] = {
+    object dsl extends Http4sDsl[RIO[R, *]]
+    import dsl._
+
+    HttpRoutes.of[RIO[R, *]] {
+      case req @ POST -> Root if req.contentType.exists(_.mediaType.isMultipart) =>
+        def getFileRefs(
+          parts: Vector[Part[RIO[R, *]]]
+        )(random: Random.Service): RIO[R, Map[String, (File, Part[RIO[R, *]])]] =
+          parts
+            .filter(_.headers.exists(_.value.contains("filename")))
+            .traverse { p =>
+              p.name.traverse { n =>
+                random.nextUUID.flatMap { uuid =>
+                  val path = rootUploadPath.resolve(uuid.toString)
+                  p.body
+                    .through(
+                      fs2.io.file.writeAll(
+                        path,
+                        blocker
+                      )
+                    )
+                    .compile
+                    .foldMonoid
+                    .as((n, path.toFile -> p))
+                }
+              }
+            }
+            .map(_.flatten.toMap)
+
+        def getUploadQuery(
+          operations: GraphQLRequest,
+          map: Map[String, Seq[String]],
+          parts: Vector[Part[RIO[R, *]]]
+        )(random: Random.Service): RIO[R, GraphQLUploadRequest] = {
+          val fileRefs  = getFileRefs(parts)(random)
+          val filePaths = parsePaths(map)
+
+          fileRefs.map { fileRef =>
+            def handler(handle: String): UIO[Option[FileMeta]] =
+              fileRef
+                .get(handle)
+                .traverse { case (file, fp) =>
+                  random.nextUUID.asSomeError
+                    .map(uuid =>
+                      FileMeta(
+                        uuid.toString,
+                        file.getAbsoluteFile.toPath,
+                        fp.headers.get(`Content-Disposition`).map(_.dispositionType),
+                        fp.contentType.map { ct =>
+                          val mt = ct.mediaType
+                          s"${mt.mainType}/${mt.subType}"
+                        },
+                        fp.filename.getOrElse(file.getName),
+                        file.length
+                      )
+                    )
+                    .optional
+                }
+                .map(_.flatten)
+
+            GraphQLUploadRequest(
+              operations,
+              filePaths,
+              Uploads.handler(handler)
+            )
+          }
+        }
+
+        req.decode[Multipart[RIO[R, *]]] { m =>
+          // First bit is always a standard graphql payload, it comes from the `operations` field
+          val optOperations =
+            m.parts.find(_.name.contains("operations")).traverse {
+              _.body
+                .through(utf8Decode)
+                .compile
+                .foldMonoid
+                .flatMap(body => Task.fromEither(parseGraphQLRequest(body)))
+            }
+
+          // Second bit is the mapping field
+          val optMap =
+            m.parts.find(_.name.contains("map")).traverse {
+              _.body
+                .through(utf8Decode)
+                .compile
+                .foldMonoid
+                .flatMap { body =>
+                  Task.fromEither(
+                    parse(body)
+                      .flatMap(_.as[Map[String, Seq[String]]])
+                      .leftMap(msg => msg.fillInStackTrace())
+                  )
+                }
+            }
+
+          for {
+            ooperations <- optOperations
+            omap        <- optMap
+            random      <- ZIO.service[Random.Service]
+            result      <- (ooperations, omap) match {
+                             case (Some(operations), Some(map)) =>
+                               for {
+                                 query           <- getUploadQuery(operations, map, m.parts)(random)
+                                 queryWithTracing =
+                                   req.headers
+                                     .find(r =>
+                                       r.name == GraphQLRequest.`apollo-federation-include-trace` && r.value == GraphQLRequest.ftv1
+                                     )
+                                     .foldLeft(query.remap)((q, _) => q.withFederatedTracing)
+
+                                 result   <- executeToJsonWithUpload(
+                                               interpreter,
+                                               queryWithTracing,
+                                               skipValidation = skipValidation,
+                                               enableIntrospection = enableIntrospection,
+                                               queryExecution,
+                                               query.fileHandle.toLayerMany
+                                             )
+                                 response <- Ok(result)
+                               } yield response
+
+                             case (None, _) => BadRequest("Missing multipart field 'operations'")
+                             case (_, None) => BadRequest("Missing multipart field 'map'")
+                           }
+          } yield result
+        }
+    }
+  }
+
   def makeHttpService[R, E](
     interpreter: GraphQLInterpreter[R, E],
     skipValidation: Boolean = false,

adapters/http4s/src/test/resources/d6359a52607b6953b2cb96be00b228c6030f11806e380c29c3f5d8db608c399b.txt

@@ -0,0 +1 @@
+Lorem ipsum dolor sit amet, consectetur adipiscing elit. Praesent et massa quam. Etiam maximus, nibh eu facilisis facilisis, tortor ex vestibulum tellus, ac semper elit justo eget purus. Duis iaculis metus elit, a semper sem sodales sed. Nam bibendum gravida gravida. Suspendisse ut ipsum iaculis, posuere enim ac, maximus nisi. Sed eleifend purus nunc. Quisque vel purus ligula. Pellentesque id ligula imperdiet, pulvinar magna sed, ultricies dolor. Donec ac neque mauris. In non est magna. Vivamus porttitor consequat est, quis pharetra odio viverra sit amet. Mauris pretium nunc lobortis nulla ultricies, at tempus quam sodales. Nam a odio dictum, ultricies elit tempor, fermentum urna. Cras lacus sem, luctus sed elementum non, malesuada et massa.