Change return type of executing bulk request

modules/library/src/it/scala/zio/elasticsearch/HttpExecutorSpec.scala

@@ -1436,11 +1436,19 @@ object HttpExecutorSpec extends IntegrationSpec {
                            firstDocumentId,
                            Script("ctx._source.intField = params['factor']").withParams("factor" -> 100)
                          )
-                  res  <- Executor.execute(ElasticRequest.bulk(req1, req2, req3, req4, req5, req6).refreshTrue)
+                  req7 =
+                    ElasticRequest
+                      .update[TestDocument](index, DocumentId("invalid-document-id"), document.copy(intField = 100))
+                  res  <- Executor.execute(ElasticRequest.bulk(req1, req2, req3, req4, req5, req6, req7).refreshTrue)
                   doc1 <- Executor.execute(ElasticRequest.getById(index, firstDocumentId)).documentAs[TestDocument]
                   doc2 <- Executor.execute(ElasticRequest.getById(index, secondDocumentId)).documentAs[TestDocument]
                   doc3 <- Executor.execute(ElasticRequest.getById(index, thirdDocumentId)).documentAs[TestDocument]
-                } yield assert(res)(isUnit) && assert(doc3)(isSome(equalTo(document.copy(intField = 100)))) &&
+                } yield assert(res.items.size)(equalTo(7)) &&
+                  assert(res.items.map(_.error.isDefined))(
+                    equalTo(List(false, false, false, false, false, false, true))
+                  ) &&
+                  assert(res.items(6).error.map(_.`type`))(equalTo(Some("document_missing_exception"))) &&
+                  assert(doc3)(isSome(equalTo(document.copy(intField = 100)))) &&
                   assert(doc2)(isNone) && assert(doc1)(
                     isSome(equalTo(document.copy(doubleField = 3000, intField = 100)))
                   )

modules/library/src/main/scala/zio/elasticsearch/ElasticRequest.scala

@@ -19,6 +19,7 @@ package zio.elasticsearch
 import zio.Chunk
 import zio.elasticsearch.ElasticPrimitive.ElasticPrimitiveOps
 import zio.elasticsearch.aggregation.ElasticAggregation
+import zio.elasticsearch.executor.response.BulkResponse
 import zio.elasticsearch.highlights.Highlights
 import zio.elasticsearch.query.ElasticQuery
 import zio.elasticsearch.query.sort.Sort
@@ -361,7 +362,10 @@ object ElasticRequest {
   private[elasticsearch] final case class Aggregate(index: IndexName, aggregation: ElasticAggregation)
       extends AggregateRequest
 
-  sealed trait BulkRequest extends ElasticRequest[Unit] with HasRefresh[BulkRequest] with HasRouting[BulkRequest]
+  sealed trait BulkRequest
+      extends ElasticRequest[BulkResponse]
+      with HasRefresh[BulkRequest]
+      with HasRouting[BulkRequest]
 
   private[elasticsearch] final case class Bulk(
     requests: List[BulkableRequest[_]],

modules/library/src/main/scala/zio/elasticsearch/executor/HttpExecutor.scala

@@ -33,6 +33,7 @@ import zio.ZIO.logDebug
 import zio.elasticsearch.ElasticRequest._
 import zio.elasticsearch._
 import zio.elasticsearch.executor.response.{
+  BulkResponse,
   CountResponse,
   CreateResponse,
   DocumentWithHighlightsAndSort,
@@ -44,7 +45,7 @@ import zio.elasticsearch.request.{CreationOutcome, DeletionOutcome, UpdateOutcom
 import zio.elasticsearch.result._
 import zio.json.ast.Json
 import zio.json.ast.Json.{Arr, Obj, Str}
-import zio.json.{DeriveJsonDecoder, JsonDecoder}
+import zio.json.{DeriveJsonDecoder, JsonDecoder, _}
 import zio.schema.Schema
 import zio.stream.{Stream, ZStream}
 import zio.{Chunk, Task, ZIO}
@@ -126,7 +127,7 @@ private[elasticsearch] final class HttpExecutor private (esConfig: ElasticConfig
       }
     }
 
-  private def executeBulk(r: Bulk): Task[Unit] = {
+  private def executeBulk[A](r: Bulk): Task[BulkResponse] = {
     val uri = (r.index match {
       case Some(index) => uri"${esConfig.uri}/$index/$Bulk"
       case None        => uri"${esConfig.uri}/$Bulk"
@@ -136,8 +137,16 @@ private[elasticsearch] final class HttpExecutor private (esConfig: ElasticConfig
       baseRequest.post(uri).contentType(ApplicationJson).body(r.body)
     ).flatMap { response =>
       response.code match {
-        case HttpOk => ZIO.unit
-        case _      => ZIO.fail(handleFailures(response))
+        case HttpOk =>
+          response.body match {
+            case Left(error) => ZIO.fail(new ElasticException(s"Bulk response body empty: ${error}"))
+            case Right(body) =>
+              body.replaceAll(" = ", ":").fromJson[BulkResponse] match {
+                case Left(error)     => ZIO.fail(new ElasticException(s"Bulk response body invalid: ${error}"))
+                case Right(response) => ZIO.succeed(response)
+              }
+          }
+        case _ => ZIO.fail(handleFailures(response))
       }
     }
   }

modules/library/src/main/scala/zio/elasticsearch/executor/response/BulkResponse.scala

@@ -0,0 +1,172 @@
+/*
+ * Copyright 2022 LambdaWorks
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zio.elasticsearch.executor.response
+
+import zio.json.{DeriveJsonDecoder, JsonDecoder, jsonField, jsonHint}
+
+// Bulk response format:
+// https://www.elastic.co/guide/en/elasticsearch/reference/8.7/docs-bulk.html
+
+private[elasticsearch] final case class Error(
+  `type`: String,
+  reason: String,
+  @jsonField("index_uuid")
+  indexUuid: String,
+  shard: String,
+  index: String
+)
+
+private[elasticsearch] object Error {
+  implicit val decoder: JsonDecoder[Error] = DeriveJsonDecoder.gen[Error]
+}
+
+private[elasticsearch] final case class Status(
+  status: Int,
+  error: Error
+)
+
+private[elasticsearch] object Status {
+  implicit val decoder: JsonDecoder[Status] = DeriveJsonDecoder.gen[Status]
+}
+
+private[elasticsearch] final case class ShardsResponse(
+  total: Int,
+  successful: Int,
+  failed: Int
+)
+
+private[elasticsearch] object ShardsResponse {
+  implicit val decoder: JsonDecoder[ShardsResponse] = DeriveJsonDecoder.gen[ShardsResponse]
+}
+
+private[elasticsearch] sealed trait Item {
+  def index: String
+  def id: String
+  def version: Option[Int]
+  def result: Option[String]
+  def shards: Option[ShardsResponse]
+  def seqNo: Option[Int]
+  def primaryTerm: Option[Int]
+  def status: Option[Int]
+  def error: Option[Error]
+}
+
+@jsonHint("create")
+private[elasticsearch] final case class Create(
+  @jsonField("_index")
+  index: String,
+  @jsonField("_id")
+  id: String,
+  @jsonField("_version")
+  version: Option[Int],
+  result: Option[String],
+  @jsonField("_shards")
+  shards: Option[ShardsResponse],
+  @jsonField("_seq_no")
+  seqNo: Option[Int],
+  @jsonField("_primary_term")
+  primaryTerm: Option[Int],
+  status: Option[Int],
+  error: Option[Error]
+) extends Item
+
+private[elasticsearch] object Create {
+  implicit val decoder: JsonDecoder[Create] = DeriveJsonDecoder.gen[Create]
+}
+
+@jsonHint("delete")
+private[elasticsearch] final case class Delete(
+  @jsonField("_index")
+  index: String,
+  @jsonField("_id")
+  id: String,
+  @jsonField("_version")
+  version: Option[Int],
+  result: Option[String],
+  @jsonField("_shards")
+  shards: Option[ShardsResponse],
+  @jsonField("_seq_no")
+  seqNo: Option[Int],
+  @jsonField("_primary_term")
+  primaryTerm: Option[Int],
+  status: Option[Int],
+  error: Option[Error]
+) extends Item
+
+private[elasticsearch] object Delete {
+  implicit val decoder: JsonDecoder[Delete] = DeriveJsonDecoder.gen[Delete]
+}
+
+@jsonHint("index")
+private[elasticsearch] final case class Index(
+  @jsonField("_index")
+  index: String,
+  @jsonField("_id")
+  id: String,
+  @jsonField("_version")
+  version: Option[Int],
+  result: Option[String],
+  @jsonField("_shards")
+  shards: Option[ShardsResponse],
+  @jsonField("_seq_no")
+  seqNo: Option[Int],
+  @jsonField("_primary_term")
+  primaryTerm: Option[Int],
+  status: Option[Int],
+  error: Option[Error]
+) extends Item
+
+private[elasticsearch] object Index {
+  implicit val decoder: JsonDecoder[Index] = DeriveJsonDecoder.gen[Index]
+}
+
+@jsonHint("update")
+private[elasticsearch] final case class Update(
+  @jsonField("_index")
+  index: String,
+  @jsonField("_id")
+  id: String,
+  @jsonField("_version")
+  version: Option[Int],
+  result: Option[String],
+  @jsonField("_shards")
+  shards: Option[ShardsResponse],
+  @jsonField("_seq_no")
+  seqNo: Option[Int],
+  @jsonField("_primary_term")
+  primaryTerm: Option[Int],
+  status: Option[Int],
+  error: Option[Error]
+) extends Item
+
+private[elasticsearch] object Update {
+  implicit val decoder: JsonDecoder[Update] = DeriveJsonDecoder.gen[Update]
+}
+
+private[elasticsearch] object Item {
+  implicit val decoder: JsonDecoder[Item] = DeriveJsonDecoder.gen[Item]
+}
+
+private[elasticsearch] final case class BulkResponse(
+  took: Int,
+  errors: Boolean,
+  items: List[Item]
+)
+
+private[elasticsearch] object BulkResponse {
+  implicit val decoder: JsonDecoder[BulkResponse] = DeriveJsonDecoder.gen[BulkResponse]
+}

modules/library/src/test/scala/zio/elasticsearch/HttpElasticExecutorSpec.scala

@@ -20,7 +20,13 @@ import zio.elasticsearch.ElasticAggregation.termsAggregation
 import zio.elasticsearch.ElasticQuery.{matchAll, term}
 import zio.elasticsearch.domain.TestDocument
 import zio.elasticsearch.executor.Executor
-import zio.elasticsearch.executor.response.{TermsAggregationBucket, TermsAggregationResponse}
+import zio.elasticsearch.executor.response.{
+  BulkResponse,
+  Create,
+  ShardsResponse,
+  TermsAggregationBucket,
+  TermsAggregationResponse
+}
 import zio.elasticsearch.request.CreationOutcome.Created
 import zio.elasticsearch.request.DeletionOutcome.Deleted
 import zio.elasticsearch.request.UpdateConflicts.Proceed
@@ -48,7 +54,31 @@ object HttpElasticExecutorSpec extends SttpBackendStubSpec {
         assertZIO(
           Executor.execute(ElasticRequest.bulk(ElasticRequest.create(index, doc)).refreshTrue)
         )(
-          isUnit
+          equalTo(
+            BulkResponse(
+              took = 3,
+              errors = false,
+              items = List(
+                Create(
+                  index = "repositories",
+                  id = "123",
+                  version = Some(1),
+                  result = Some("created"),
+                  shards = Some(
+                    ShardsResponse(
+                      total = 1,
+                      successful = 1,
+                      failed = 0
+                    )
+                  ),
+                  seqNo = Some(0),
+                  primaryTerm = Some(1),
+                  status = Some(201),
+                  error = None
+                )
+              )
+            )
+          )
         )
       },
       test("count request") {