HTTP4S customizable Request/Response Logger Middleware

Http4sExampleEndpoint.scala

import org.http4s.*
import org.http4s.dsl.*
import zio.{durationInt as _, *}
import zio.interop.catz.*
import com.comcast.ip4s.*
import fs2.*
import org.http4s.ember.server.EmberServerBuilder
import fs2.io.net.Network

val example =
  val dsl = new Http4sDsl[Task] {}
  import dsl.*
  HttpRoutes.of[Task] {
    case GET -> Root / "stream" =>
      Ok(
        Stream
          .iterate(1)(_ + 1)
          .take(2_000L)
          .map(_.toString())
          .intersperse(",")
          .covary[Task]
      )

    case req @ POST -> Root / "upload" =>
      val stream = req.body.through(fs2.text.utf8.decode)
      val text = stream.compile.string
      for
        t <- text
        res <- Ok("thanks for the upload")
      yield res
  }

val app = Http4sLogger.logMiddleware(example.orNotFound, "example")

object Http4sExampleEndpoint extends ZIOAppDefault {
  given Network[Task] = Network.forAsync
  def run =
    EmberServerBuilder
      .default[Task]
      .withHost(ipv4"0.0.0.0")
      .withPort(port"8080")
      .withHttpApp(app)
      .build
      .use(_ => ZIO.never)
}

Http4sLogger.scala

import org.http4s.*
import org.typelevel.ci.CIString
import zio.*
import zio.interop.catz.*

object Http4sLogger {
  def logMiddleware(
      httpApp: HttpApp[Task],
      serviceName: String
  ): HttpApp[Task] = HttpApp { (req: Request[Task]) =>
    (httpApp(logRequest(req)).map(logResponse))
      @@ ZIOAspect.annotated(
        "service_name" -> serviceName,
        "http_method" -> req.method.name,
        "http_path" -> req.uri.path.toString()
      )
  }

  private inline def logRequest(inline req: Request[Task]): Request[Task] =
    val body =
      req.body
        .broadcastThrough[Task, Byte](
          identity,
          _.chunks.foldMonoid
            .evalMap { rawReqBody =>
              ZIO.logInfo(s"Incoming request") @@
                ZIOAspect.annotated(
                  "http_request_body" -> chunk2String(rawReqBody),
                  "http_request_headers" -> req.headers.headers
                    .filterNot(_.name.equals(CIString("authorization")))
                    .toString()
                )
            }
            .drain
        )
    req.withBodyStream(body)

  private inline def logResponse(inline res: Response[Task]): Response[Task] =
    val body = res.body
      .broadcastThrough[Task, Byte](
        identity,
        _.chunks.foldMonoid
          .evalMap { rawResponseBody =>
            ZIO.logInfo(s"Outgoing response") @@
              ZIOAspect.annotated(
                "http_response_body" -> chunk2String(rawResponseBody),
                "http_response_status" -> res.status.toString()
              )
          }
          .drain
      )
    res.withBodyStream(body)

  private def chunk2String(chunk: fs2.Chunk[Byte]): String =
    new String(chunk.toArray, "UTF-8")
}