Search service backend implementation

agni/.env.local.sample

@@ -0,0 +1,2 @@
+export SEARCH_SERVER=elasticsearch://10.240.0.18:9300
+export PORT=9000

search-service/Makefile → agni/Makefile

@@ -1,14 +1,21 @@
 include ../makelib
-header = $(call baseheader, $(1), search-service)
+header = $(call baseheader, $(1), agni)
 
 DOCKER_REPO ?= $(DOCKER_STAGE_REPO)
-DOCKER_IMAGE ?= search-service
+DOCKER_IMAGE ?= agni
 DOCKER_TAG ?= master
 SBT_CMD = sbt -DDOCKER_REPO=$(DOCKER_REPO) -DDOCKER_TAG=${DOCKER_TAG}
 
+autoformat-check:
+	../utils/scalafmt/scalafmt.sh --test
+
+run:
+	$(call header, Run locally)
+	sbt '~api/re-start'
+
 clean:
 	$(call header, Cleaning)
-	${SBT_CMD} 'clean'
+	${SBT_CMD} '; clean'
 
 build: fmt-check
 	$(call header, Building)
@@ -22,17 +29,17 @@ fmt-check:
 
 test:
 	$(call header, Testing)
-	${SBT_CMD} 'test'
+	${SBT_CMD} '; test'
 
 docker:
 	$(call header, Dockerizing)
-	${SBT_CMD} 'api/docker'
+	${SBT_CMD} '; api/docker'
 
 docker-push:
 	$(call header, Registering)
-	${SBT_CMD} 'api/dockerPush'
+	${SBT_CMD} '; api/dockerPush'
 
-docker-build:
-	${SBT_CMD} 'api/dockerBuildAndPush'
+docker-all:
+	${SBT_CMD} '; api/dockerBuildAndPush'
 
-.PHONY: clean build fmt fmt-check test docker docker-push docker-build
+.PHONY: clean build fmt fmt-check test docker docker-push docker-all

agni/api/app/foxcomm/agni/api/Api.scala

@@ -0,0 +1,54 @@
+package foxcomm.agni.api
+
+import com.twitter.finagle.Http
+import com.twitter.finagle.http.Status
+import com.twitter.util.Await
+import foxcomm.agni._
+import foxcomm.agni.dsl.query._
+import foxcomm.agni.interpreter.es.queryInterpreter
+import foxcomm.utils.finch._
+import io.circe.generic.extras.auto._
+import io.finch._
+import io.finch.circe._
+import monix.execution.Scheduler
+import org.elasticsearch.common.ValidationException
+
+object Api extends App {
+  def endpoints(searchService: SearchService)(implicit s: Scheduler) =
+    post("api" :: "search" :: "translate" :: jsonBody[SearchPayload.fc]) { (searchPayload: SearchPayload.fc) ⇒
+      searchService
+        .translate(searchPayload = searchPayload)
+        .map(Ok)
+        .toTwitterFuture
+    } :+: post(
+      "api" :: "search" :: string :: string :: param("size")
+        .as[Int] :: paramOption("from").as[Int] :: jsonBody[SearchPayload]) {
+      (searchIndex: String, searchType: String, size: Int, from: Option[Int], searchQuery: SearchPayload) ⇒
+        searchService
+          .searchFor(searchIndex = searchIndex,
+                     searchType = searchType,
+                     searchPayload = searchQuery,
+                     searchSize = size,
+                     searchFrom = from)
+          .map(Ok)
+          .toTwitterFuture
+    } :+: get("ping") {
+      Ok("pong")
+    }
+
+  def errorHandler[A]: PartialFunction[Throwable, Output[A]] = {
+    case ex: ValidationException ⇒ Output.failure(ex, Status.BadRequest)
+    case ex: Exception           ⇒ Output.failure(ex, Status.InternalServerError)
+    case ex                      ⇒ Output.failure(new RuntimeException(ex), Status.InternalServerError)
+  }
+
+  implicit val s: Scheduler = Scheduler.global
+  val config                = AppConfig.load()
+  val svc                   = SearchService.fromConfig(config, queryInterpreter)
+
+  Await.result(
+    Http.server
+      .withStreaming(enabled = true)
+      .serve(s"${config.http.interface}:${config.http.port}",
+             endpoints(svc).handle(errorHandler).toServiceAs[Application.Json]))
+}

search-service/build.sbt → agni/build.sbt

@@ -1,20 +1,15 @@
-import sbtassembly.AssemblyKeys.assemblyExcludedJars
 import sbtassembly.{MergeStrategy, PathList}
 
-name := "search-service"
-
-version := "0.1-SNAPSHOT"
-
 lazy val core = (project in file("core"))
   .settings(Settings.common)
   .settings(
-    libraryDependencies ++= Dependencies.core ++ Dependencies.es ++ Dependencies.circe
+    libraryDependencies ++= Dependencies.core ++ Dependencies.es ++ Dependencies.circe ++ Dependencies.monix ++ Dependencies.test.core
   )
 
 lazy val finch = (project in file("finch"))
   .settings(Settings.common)
   .settings(
-    libraryDependencies ++= Dependencies.finch ++ Dependencies.circe :+ Dependencies.jwt
+    libraryDependencies ++= Dependencies.finch ++ Dependencies.circe ++ Dependencies.jwt ++ Dependencies.monix
   )
 
 lazy val api = (project in file("api"))
@@ -24,8 +19,9 @@ lazy val api = (project in file("api"))
     libraryDependencies ++= Dependencies.finch
   )
   .settings(
-    mainClass in assembly := Some("foxcomm.search.api.Api"),
-    assemblyJarName in assembly := "search-service.jar",
+    Settings.appName := "agni",
+    mainClass in assembly := Some("foxcomm.agni.api.Api"),
+    assemblyJarName in assembly := s"${Settings.appName.value}.jar",
     assemblyMergeStrategy in assembly := {
       case PathList("BUILD")                                    ⇒ MergeStrategy.discard
       case PathList("META-INF", "io.netty.versions.properties") ⇒ MergeStrategy.discard

agni/core/app/foxcomm/agni/AppConfig.scala

@@ -0,0 +1,36 @@
+package foxcomm.agni
+
+import cats.data.NonEmptyList
+import com.typesafe.config.ConfigFactory
+import java.net.InetSocketAddress
+import pureconfig._
+import scala.util.Try
+
+final case class AppConfig(http: AppConfig.Http, elasticsearch: AppConfig.ElasticSearch)
+
+@SuppressWarnings(Array("org.wartremover.warts.Equals"))
+object AppConfig {
+  implicit val readHostConfig: ConfigReader[NonEmptyList[InetSocketAddress]] =
+    ConfigReader.fromNonEmptyStringTry(s ⇒
+      Try {
+        val withoutPrefix = s.stripPrefix("elasticsearch://")
+        val hosts = withoutPrefix.split(',').map { host ⇒
+          val parts = host.split(':')
+          require(parts.length == 2,
+                  "ElasticSearch uri must be in format elasticsearch://host:port,host:port,...")
+          new InetSocketAddress(parts(0), parts(1).toInt)
+        }
+        require(hosts.length >= 1, "At least single ElasticSearch host should be specified")
+        NonEmptyList.fromListUnsafe(hosts.toList)
+    })
+
+  final case class Http(interface: String, port: Int)
+
+  final case class ElasticSearch(host: NonEmptyList[InetSocketAddress], cluster: String)
+
+  def load(): AppConfig = {
+    val config =
+      ConfigFactory.systemProperties.withFallback(ConfigFactory.load())
+    loadConfigOrThrow[AppConfig](config, "app")
+  }
+}

agni/core/app/foxcomm/agni/SearchService.scala

@@ -0,0 +1,108 @@
+package foxcomm.agni
+
+import cats.implicits._
+import foxcomm.agni.interpreter.es._
+import io.circe._
+import io.circe.jawn.parseByteBuffer
+import monix.eval.{Coeval, Task}
+import org.elasticsearch.action.search.{SearchAction, SearchRequestBuilder, SearchResponse}
+import org.elasticsearch.client.Client
+import org.elasticsearch.client.transport.TransportClient
+import org.elasticsearch.common.settings.Settings
+import org.elasticsearch.common.transport.InetSocketTransportAddress
+import org.elasticsearch.common.xcontent.{ToXContent, XContentFactory}
+import org.elasticsearch.index.query.QueryBuilder
+import org.elasticsearch.search.SearchHit
+
+@SuppressWarnings(Array("org.wartremover.warts.NonUnitStatements"))
+class SearchService private (client: Client, qi: ESQueryInterpreter) {
+  import SearchService.ExtractJsonObject
+
+  def translate(searchPayload: SearchPayload.fc): Task[Json] = {
+    def buildJson(qb: QueryBuilder): Coeval[Json] =
+      Coeval.eval {
+        val builder = XContentFactory.jsonBuilder()
+        builder.prettyPrint()
+        builder.startObject()
+        builder.field("query")
+        qb.toXContent(builder, ToXContent.EMPTY_PARAMS)
+        builder.endObject()
+        parseByteBuffer(builder.bytes().toChannelBuffer.toByteBuffer)
+          .fold(Coeval.raiseError(_), Coeval.eval(_))
+      }.flatten
+
+    for {
+      builder ← qi(searchPayload.query).task
+      json    ← buildJson(builder).task
+    } yield json
+  }
+
+  def searchFor(searchIndex: String,
+                searchType: String,
+                searchPayload: SearchPayload,
+                searchSize: Int,
+                searchFrom: Option[Int]): Task[SearchResult] = {
+    def prepareBuilder: Coeval[SearchRequestBuilder] = Coeval.eval {
+      val builder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
+      builder
+        .setIndices(searchIndex)
+        .setTypes(searchType)
+        .setSize(searchSize)
+      searchFrom.foreach(builder.setFrom)
+      searchPayload.fields.foreach(fs ⇒ builder.setFetchSource(fs.toList.toArray, Array.empty[String]))
+      builder
+    }
+
+    def evalQuery(builder: SearchRequestBuilder): Coeval[SearchRequestBuilder] = searchPayload match {
+      case SearchPayload.es(query, _) ⇒
+        Coeval.eval(builder.setQuery(Json.fromJsonObject(query).toBytes))
+      case SearchPayload.fc(query, _) ⇒
+        qi(query).map(builder.setQuery)
+    }
+
+    def setupBuilder: Task[SearchRequestBuilder] = (prepareBuilder flatMap evalQuery).task
+
+    for {
+      builder ← setupBuilder
+      request = builder.request()
+      response ← async[SearchResponse, SearchResult](client.search(request, _))
+    } yield {
+      val hits = response.getHits
+      SearchResult(
+        result = hits
+          .hits()
+          .view
+          .collect {
+            case ExtractJsonObject(obj) ⇒ obj
+          }
+          .toList,
+        pagination = SearchPagination(total = hits.totalHits()),
+        maxScore = hits.getMaxScore
+      )
+    }
+  }
+}
+
+object SearchService {
+  object ExtractJsonObject {
+    def unapply(hit: SearchHit): Option[JsonObject] =
+      parseByteBuffer(hit.sourceRef.toChannelBuffer.toByteBuffer).toOption
+        .flatMap(_.asObject)
+  }
+
+  def apply(client: Client, qi: ESQueryInterpreter): SearchService =
+    new SearchService(client, qi)
+
+  def fromConfig(config: AppConfig, qi: ESQueryInterpreter): SearchService = {
+    val esConfig = config.elasticsearch
+    val settings =
+      Settings.settingsBuilder().put("cluster.name", esConfig.cluster).build()
+    val client = TransportClient
+      .builder()
+      .settings(settings)
+      .build()
+      .addTransportAddresses(esConfig.host.toList.map(new InetSocketTransportAddress(_)): _*)
+
+    apply(client, qi)
+  }
+}

agni/core/app/foxcomm/agni/dsl/package.scala

@@ -0,0 +1,7 @@
+package foxcomm.agni
+
+import io.circe.generic.extras.Configuration
+
+package object dsl {
+  implicit def configuration: Configuration = foxcomm.agni.configuration
+}