(example): Support initial loading of data in example app

modules/example/src/main/scala/example/GitHubRepo.scala

@@ -1,12 +1,13 @@
 package example
 
+import example.external.github.model.RepoResponse
 import zio.json.{DeriveJsonEncoder, JsonEncoder}
 import zio.schema.{DeriveSchema, Schema}
 
-import java.time.LocalDateTime
+import java.time.{Instant, LocalDateTime, ZoneId}
 
 final case class GitHubRepo(
- id: Option[String],
+ id: String,
  organization: String,
  name: String,
  url: String,
@@ -17,6 +18,18 @@ final case class GitHubRepo(
 )
 
 object GitHubRepo {
+ def fromResponse(response: RepoResponse): GitHubRepo =
+ GitHubRepo(
+ id = response.id.toString,
+ organization = response.owner.organization,
+ name = response.name,
+ url = response.url,
+ description = response.description,
+ lastCommitAt = LocalDateTime.ofInstant(Instant.parse(response.updatedAt), ZoneId.systemDefault()),
+ stars = response.stars,
+ forks = response.forks
+ )
+
  implicit val schema: Schema[GitHubRepo] = DeriveSchema.gen[GitHubRepo]
 
  implicit val encoder: JsonEncoder[GitHubRepo] = DeriveJsonEncoder.gen[GitHubRepo]

modules/example/src/main/scala/example/Main.scala

@@ -2,6 +2,8 @@ package example
 
 import example.api.{HealthCheck, Repositories}
 import example.config.{AppConfig, ElasticsearchConfig, HttpConfig}
+import example.external.github.RepoFetcher
+import sttp.client3.SttpBackend
 import sttp.client3.httpclient.zio.HttpClientZioBackend
 import zio._
 import zio.config.getConfig
@@ -23,7 +25,7 @@ object Main extends ZIOAppDefault {
  )
  }
 
- private[this] def prepare: RIO[ElasticExecutor, Unit] = {
+ private[this] def prepare: RIO[SttpBackend[Task, Any] with ElasticExecutor, Unit] = {
  val deleteIndex: RIO[ElasticExecutor, Unit] =
  for {
  _ <- ZIO.logInfo(s"Deleting index '$Index'...")
@@ -37,7 +39,14 @@ object Main extends ZIOAppDefault {
  _ <- ElasticRequest.createIndex(Index, Some(mapping)).execute
  } yield ()
 
- deleteIndex *> createIndex
+ val populate: RIO[SttpBackend[Task, Any] with ElasticExecutor, Unit] =
+ (for {
+ repositories <- RepoFetcher.fetchAllByOrganization("zio")
+ _ <- ZIO.logInfo("Adding GitHub repositories...")
+ _ <- RepositoriesElasticsearch.createAll(repositories)
+ } yield ()).provideSome(RepositoriesElasticsearch.live)
+
+ deleteIndex *> createIndex *> populate
  }
 
  private[this] def runServer: RIO[HttpConfig with ElasticExecutor, ExitCode] = {

modules/example/src/main/scala/example/RepositoriesElasticsearch.scala

@@ -2,6 +2,7 @@ package example
 
 import zio._
 import zio.elasticsearch.{DeletionOutcome, DocumentId, ElasticExecutor, ElasticRequest, Routing}
+import zio.prelude.Newtype.unsafeWrap
 
 final case class RepositoriesElasticsearch(executor: ElasticExecutor) {
 
@@ -19,6 +20,21 @@ final case class RepositoriesElasticsearch(executor: ElasticExecutor) {
  res <- executor.execute(req)
  } yield res
 
+ def createAll(repositories: List[GitHubRepo]): Task[Unit] =
+ for {
+ reqs <- ZIO.collectAllPar {
+ repositories.map { repository =>
+ routingOf(repository.organization).map(
+ ElasticRequest
+ .create[GitHubRepo](Index, unsafeWrap(DocumentId)(repository.id), repository)
+ .routing(_)
+ )
+ }
+ }
+ bulkReq = ElasticRequest.bulk(reqs: _*)
+ _ <- executor.execute(bulkReq)
+ } yield ()
+
  def upsert(id: String, repository: GitHubRepo): Task[Unit] =
  for {
  routing <- routingOf(repository.organization)
@@ -46,6 +62,9 @@ object RepositoriesElasticsearch {
  def create(repository: GitHubRepo): RIO[RepositoriesElasticsearch, DocumentId] =
  ZIO.serviceWithZIO[RepositoriesElasticsearch](_.create(repository))
 
+ def createAll(repositories: List[GitHubRepo]): RIO[RepositoriesElasticsearch, Unit] =
+ ZIO.serviceWithZIO[RepositoriesElasticsearch](_.createAll(repositories))
+
  def upsert(id: String, repository: GitHubRepo): RIO[RepositoriesElasticsearch, Unit] =
  ZIO.serviceWithZIO[RepositoriesElasticsearch](_.upsert(id, repository))
 

modules/example/src/main/scala/example/api/Repositories.scala

@@ -37,7 +37,7 @@ object Repositories {
  ZIO.succeed(Response.json(ErrorResponse.fromReasons(e.message).toJson).setStatus(BadRequest))
  case Right(repo) =>
  RepositoriesElasticsearch.create(repo).map { id =>
- Response.json(repo.copy(id = Some(DocumentId.unwrap(id))).toJson).setStatus(Created)
+ Response.json(repo.copy(id = DocumentId.unwrap(id)).toJson).setStatus(Created)
  }
  }
  .orDie
@@ -48,7 +48,7 @@ object Repositories {
  .flatMap {
  case Left(e) =>
  ZIO.succeed(Response.json(ErrorResponse.fromReasons(e.message).toJson).setStatus(BadRequest))
- case Right(repo) if repo.id.exists(_ != id) =>
+ case Right(repo) if repo.id == id =>
  ZIO.succeed(
  Response
  .json(
@@ -58,7 +58,7 @@ object Repositories {
  )
  case Right(repo) =>
  (RepositoriesElasticsearch
- .upsert(id, repo.copy(id = Some(id))) *> RepositoriesElasticsearch.findById(
+ .upsert(id, repo.copy(id = id)) *> RepositoriesElasticsearch.findById(
  repo.organization,
  id
  )).map {

modules/example/src/main/scala/example/external/github/RepoFetcher.scala

@@ -0,0 +1,22 @@
+package example.external.github
+
+import example.GitHubRepo
+import example.external.github.model.RepoResponse
+import sttp.client3.{SttpBackend, UriContext, basicRequest}
+import zio.json.DecoderOps
+import zio.{RIO, Task, ZIO}
+
+object RepoFetcher {
+
+ def fetchAllByOrganization(
+ organization: String,
+ limit: Int = 100
+ ): RIO[SttpBackend[Task, Any], List[GitHubRepo]] =
+ for {
+ client <- ZIO.service[SttpBackend[Task, Any]]
+ req = basicRequest.get(uri"https://api.github.com/orgs/$organization/repos?per_page=$limit")
+ res <- req.send(client)
+ } yield res.body.toOption
+ .map(_.fromJson[List[RepoResponse]].fold(_ => Nil, _.map(GitHubRepo.fromResponse).toList))
+ .getOrElse(Nil)
+}

modules/example/src/main/scala/example/external/github/model/RepoOwner.scala

@@ -0,0 +1,12 @@
+package example.external.github.model
+
+import zio.json.{DeriveJsonDecoder, JsonDecoder, jsonField}
+
+final case class RepoOwner(
+ @jsonField("login")
+ organization: String
+)
+
+object RepoOwner {
+ implicit val decoder: JsonDecoder[RepoOwner] = DeriveJsonDecoder.gen[RepoOwner]
+}

modules/example/src/main/scala/example/external/github/model/RepoResponse.scala

@@ -0,0 +1,20 @@
+package example.external.github.model
+
+import zio.json.{DeriveJsonDecoder, JsonDecoder, jsonField}
+
+final case class RepoResponse(
+ id: Int,
+ name: String,
+ url: String,
+ description: Option[String],
+ @jsonField("updated_at")
+ updatedAt: String,
+ @jsonField("stargazers_count")
+ stars: Int,
+ forks: Int,
+ owner: RepoOwner
+)
+
+object RepoResponse {
+ implicit val decoder: JsonDecoder[RepoResponse] = DeriveJsonDecoder.gen[RepoResponse]
+}

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

@@ -88,10 +88,15 @@ object ElasticRequest {
  private[elasticsearch] final case class BulkableRequest private (request: ElasticRequest[_, _])
 
  object BulkableRequest {
- implicit def toBulkable[ERT <: ElasticRequestType](req: ElasticRequest[_, ERT])(implicit
+ implicit def toBulkable[ERT <: ElasticRequestType](request: ElasticRequest[_, ERT])(implicit
  @unused ev: ERT <:< BulkableRequestType
  ): BulkableRequest =
- BulkableRequest(req)
+ BulkableRequest(request)
+
+ implicit def toBulkableList[ERT <: ElasticRequestType](requests: List[ElasticRequest[_, ERT]])(implicit
+ @unused ev: ERT <:< BulkableRequestType
+ ): List[BulkableRequest] =
+ requests.map(BulkableRequest(_))
  }
 
  private[elasticsearch] final case class BulkRequest(