(executor): Support Create and CreateOrUpdate requests

modules/example/src/main/scala/example/ExampleApp.scala

@@ -7,13 +7,30 @@ import zio.elasticsearch._
 object ExampleApp extends ZIOAppDefault {
 
   override def run: Task[Unit] = {
-    val index   = IndexName("kibana_sample_data_ecommerce")
-    val docId   = DocumentId("wzvbqIQB22WMP-4s4h8K")
+    val index   = IndexName("documentindex")
+    val docId   = DocumentId("SrBtwIQBGQ08MH8or_0t")
     val routing = Some(Routing("10"))
 
     (for {
-      _   <- Console.printLine("Welcome to an example app...")
-      _   <- Console.printLine(s"Looking for the document '$docId' in '$index' index...'")
+      _ <- Console.printLine("Welcome to an example app...")
+      _ <- Console.printLine(s"Creating document with id '$docId' in '$index' index...'")
+      newDocId <-
+        ElasticRequest
+          .create(index = index, id = docId, doc = ExampleDocument("docId", "docName", 11), routing = routing)
+          .execute
+      _ <- Console.printLine(newDocId)
+      _ <- Console.printLine("Creating document with same id as previous (unsuccessfully)...")
+      newDocId2 <-
+        ElasticRequest
+          .create(index = index, id = docId, doc = ExampleDocument("docId2", "docName2", 22), routing = routing)
+          .execute
+      _ <- Console.printLine(newDocId2)
+      _ <- Console.printLine("Updating existing document...")
+      _ <-
+        ElasticRequest
+          .upsert(index = index, id = docId, doc = ExampleDocument("docId3", "docName3", 33), routing = routing)
+          .execute
+      _   <- Console.printLine("Getting updated document...")
       res <- ElasticRequest.getById[ExampleDocument](index, docId, routing).execute
       _   <- Console.printLine(res)
     } yield ()).provide(ElasticExecutor.local, HttpClientZioBackend.layer())

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

@@ -0,0 +1,24 @@
+package zio.elasticsearch
+
+import zio.json.ast.Json
+import zio.json.{DeriveJsonDecoder, JsonDecoder, jsonField}
+
+final case class ElasticCreateResponse(
+  @jsonField("_index")
+  index: String,
+  @jsonField("_type")
+  `type`: String,
+  @jsonField("_id")
+  id: String,
+  @jsonField("_version")
+  version: Int,
+  @jsonField("result")
+  result: String,
+  @jsonField("_shards")
+  shards: Json
+)
+
+object ElasticCreateResponse {
+  implicit val decoder: JsonDecoder[ElasticCreateResponse] =
+    DeriveJsonDecoder.gen[ElasticCreateResponse]
+}

modules/library/src/main/scala/zio/elasticsearch/ElasticResponse.scala → modules/library/src/main/scala/zio/elasticsearch/ElasticGetResponse.scala

@@ -3,12 +3,12 @@ package zio.elasticsearch
 import zio.json.ast.Json
 import zio.json.{DeriveJsonDecoder, JsonDecoder, jsonField}
 
-final case class ElasticResponse(
+final case class ElasticGetResponse(
   found: Boolean,
   @jsonField("_source")
   source: Json
 )
 
-object ElasticResponse {
-  implicit val decoder: JsonDecoder[ElasticResponse] = DeriveJsonDecoder.gen[ElasticResponse]
+object ElasticGetResponse {
+  implicit val decoder: JsonDecoder[ElasticGetResponse] = DeriveJsonDecoder.gen[ElasticGetResponse]
 }

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

@@ -20,13 +20,13 @@ object ElasticRequest {
     doc: A,
     routing: Option[Routing] = None
   ): ElasticRequest[Unit] =
-    Create(index, Some(id), Document.from(doc), routing)
+    Create(index, Some(id), Document.from(doc), routing).map(_ => ())
 
   def create[A: Schema](
     index: IndexName,
     doc: A,
     routing: Option[Routing]
-  ): ElasticRequest[Unit] =
+  ): ElasticRequest[Option[DocumentId]] =
     Create(index, None, Document.from(doc), routing)
 
   def getById[A: Schema](
@@ -52,7 +52,7 @@ object ElasticRequest {
     id: Option[DocumentId],
     document: Document,
     routing: Option[Routing] = None
-  ) extends ElasticRequest[Unit]
+  ) extends ElasticRequest[Option[DocumentId]]
 
   private[elasticsearch] final case class CreateOrUpdate(
     index: IndexName,

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

@@ -3,8 +3,8 @@ package zio.elasticsearch
 import sttp.client3._
 import sttp.client3.ziojson._
 import sttp.model.Uri
+import zio.Task
 import zio.elasticsearch.ElasticRequest._
-import zio.{Task, ZIO}
 
 private[elasticsearch] final class HttpElasticExecutor private (config: ElasticConfig, client: SttpBackend[Task, Any])
     extends ElasticExecutor {
@@ -15,27 +15,60 @@ private[elasticsearch] final class HttpElasticExecutor private (config: ElasticC
 
   override def execute[A](request: ElasticRequest[A]): Task[A] =
     request match {
-      case _: Create         => ZIO.unit
-      case _: CreateOrUpdate => ZIO.unit
-      case r: GetById        => executeGetById(r)
-      case map @ Map(_, _)   => execute(map.request).map(map.mapper)
+      case c: Create           => executeCreate(c)
+      case cou: CreateOrUpdate => executeCreateOrUpdate(cou)
+      case r: GetById          => executeGetById(r)
+      case map @ Map(_, _)     => execute(map.request).map(map.mapper)
     }
 
   private def executeGetById(getById: GetById): Task[Option[Document]] = {
     val uri = uri"$basePath/${getById.index}/$Doc/${getById.id}".withParam("routing", getById.routing.map(_.value))
     basicRequest
       .get(uri)
-      .response(asJson[ElasticResponse])
+      .response(asJson[ElasticGetResponse])
       .send(client)
       .map(_.body.toOption)
       .map(_.flatMap(d => if (d.found) Option(Document.from(d.source)) else None))
   }
 
+  private def executeCreate(create: Create): Task[Option[DocumentId]] = {
+    def createUri(maybeDocumentId: Option[DocumentId]) =
+      maybeDocumentId match {
+        case Some(documentId) =>
+          uri"$basePath/${create.index}/$Create/$documentId".withParam("routing", create.routing.map(_.value))
+        case None =>
+          uri"$basePath/${create.index}/$Doc/".withParam("routing", create.routing.map(_.value))
+      }
+
+    basicRequest
+      .post(createUri(create.id))
+      .header(ContentType, ContentTypeVal)
+      .response(asJson[ElasticCreateResponse])
+      .body(create.document.json)
+      .send(client)
+      .map(_.body.toOption)
+      .map(_.flatMap(body => Option(DocumentId(body.id))))
+  }
+
+  private def executeCreateOrUpdate(createOrUpdate: CreateOrUpdate): Task[Unit] = {
+    val u = uri"$basePath/${createOrUpdate.index}/$Doc/${createOrUpdate.id}"
+      .withParam("routing", createOrUpdate.routing.map(_.value))
+
+    basicRequest
+      .put(u)
+      .header(ContentType, ContentTypeVal)
+      .body(createOrUpdate.document.json)
+      .send(client)
+      .map(_ => ())
+  }
 }
 
 private[elasticsearch] object HttpElasticExecutor {
 
-  private final val Doc = "_doc"
+  private final val Doc            = "_doc"
+  private final val Create         = "_create"
+  private final val ContentType    = "Content-Type"
+  private final val ContentTypeVal = "application/json"
 
   def apply(config: ElasticConfig, client: SttpBackend[Task, Any]) =
     new HttpElasticExecutor(config, client)