Start refactor to wrappers round search client to make testing easier

chrissearle · chrissearle · commit d906553b6a0c · 2024-11-20T11:08:41.000+01:00
diff --git a/src/main/kotlin/no/java/conf/plugins/Search.kt b/src/main/kotlin/no/java/conf/plugins/Search.kt
@@ -14,6 +14,8 @@ import io.ktor.server.routing.route
 import io.ktor.server.routing.routing
 import no.java.conf.model.search.TextSearchRequest
 import no.java.conf.service.SearchService
+import no.java.conf.service.search.ElasticIndexer
+import no.java.conf.service.search.ElasticIngester
 
 private val logger = KotlinLogging.logger {}
 
@@ -49,21 +51,30 @@ fun Application.searchClient() =
 
 fun searchService(
     searchClient: SearchClient,
+    indexer: ElasticIndexer,
+    ingester: ElasticIngester,
     skipIndex: Boolean
 ) = SearchService(
     client = searchClient,
+    indexer = indexer,
+    ingester = ingester,
     skipIndex = skipIndex,
 )
 
-fun Application.searchService() =
-    searchService(
-        searchClient = searchClient(),
+fun Application.searchService(): SearchService {
+    val searchClient = searchClient()
+
+    return searchService(
+        searchClient = searchClient,
+        indexer = ElasticIndexer(searchClient),
+        ingester = ElasticIngester(searchClient),
         skipIndex =
             environment.config
                 .property("elastic.skipindex")
                 .getString()
                 .toBoolean()
     )
+}
 
 fun Application.configureSearchRouting(service: SearchService) {
     routing {
diff --git a/src/main/kotlin/no/java/conf/service/SearchService.kt b/src/main/kotlin/no/java/conf/service/SearchService.kt
@@ -5,14 +5,8 @@ import arrow.core.raise.ensure
 import arrow.core.raise.ensureNotNull
 import com.jillesvangurp.jsondsl.json
 import com.jillesvangurp.ktsearch.Aggregations
-import com.jillesvangurp.ktsearch.BulkItemCallBack
-import com.jillesvangurp.ktsearch.BulkResponse
-import com.jillesvangurp.ktsearch.OperationType
 import com.jillesvangurp.ktsearch.SearchClient
-import com.jillesvangurp.ktsearch.bulk
 import com.jillesvangurp.ktsearch.count
-import com.jillesvangurp.ktsearch.createIndex
-import com.jillesvangurp.ktsearch.deleteIndex
 import com.jillesvangurp.ktsearch.parseHits
 import com.jillesvangurp.ktsearch.parsedBuckets
 import com.jillesvangurp.ktsearch.search
@@ -25,7 +19,6 @@ import com.jillesvangurp.searchdsls.querydsl.exists
 import com.jillesvangurp.searchdsls.querydsl.nested
 import com.jillesvangurp.searchdsls.querydsl.simpleQueryString
 import com.jillesvangurp.searchdsls.querydsl.terms
-import com.jillesvangurp.serializationext.DEFAULT_JSON
 import io.github.oshai.kotlinlogging.KotlinLogging
 import no.java.conf.model.AggregationsNotFound
 import no.java.conf.model.ApiError
@@ -42,16 +35,12 @@ import no.java.conf.model.search.YearAggregate
 import no.java.conf.model.search.hasFilter
 import no.java.conf.model.search.hasQuery
 import no.java.conf.model.sessions.Session
-import no.java.conf.model.sessions.Speaker
-import kotlin.time.Duration.Companion.seconds
+import no.java.conf.service.search.ElasticIndexer
+import no.java.conf.service.search.ElasticIngester
 import kotlin.time.measureTimedValue
 
 private const val INDEX_NAME = "javazone"
 
-private const val REPLICAS = 0
-private const val SHARDS = 3
-private const val REFRESH = 10
-
 private val logger = KotlinLogging.logger {}
 
 enum class State {
@@ -62,6 +51,8 @@ enum class State {
 
 class SearchService(
     private val client: SearchClient,
+    private val indexer: ElasticIndexer,
+    private val ingester: ElasticIngester,
     private val skipIndex: Boolean,
 ) {
     private var readyState = State.NEW
@@ -78,30 +69,7 @@ class SearchService(
         if (!skipIndex) {
             logger.debug { "Creating index" }
 
-            client.deleteIndex(INDEX_NAME, ignoreUnavailable = true)
-
-            client.createIndex(INDEX_NAME) {
-                settings {
-                    replicas = REPLICAS
-                    shards = SHARDS
-                    refreshInterval = REFRESH.seconds
-                }
-                mappings(dynamicEnabled = false) {
-                    text(Session::title)
-                    text(Session::abstract)
-                    text(Session::intendedAudience)
-                    keyword(Session::year)
-                    keyword(Session::video)
-                    keyword(Session::sessionId)
-                    keyword(Session::format)
-                    keyword(Session::language)
-                    nestedField("speakers") {
-                        text(Speaker::name)
-                        keyword(Speaker::twitter)
-                        text(Speaker::bio)
-                    }
-                }
-            }
+            indexer.recreateIndex(INDEX_NAME)
         }
 
         logger.debug { "State -> Mapped" }
@@ -121,44 +89,7 @@ class SearchService(
         if (!skipIndex) {
             logger.debug { "Bulk" }
 
-            val itemCallBack =
-                object : BulkItemCallBack {
-                    override fun bulkRequestFailed(
-                        e: Exception,
-                        ops: List<Pair<String, String?>>,
-                    ) {
-                        logger.error(e) { "Bulk failed" }
-                    }
-
-                    override fun itemFailed(
-                        operationType: OperationType,
-                        item: BulkResponse.ItemDetails,
-                    ) {
-                        logger.warn { "${operationType.name} failed ${item.id} with ${item.status}" }
-                    }
-
-                    override fun itemOk(
-                        operationType: OperationType,
-                        item: BulkResponse.ItemDetails,
-                    ) {
-                        logger.trace {
-                            "${operationType.name} completed ${item.id} seq ${item.seqNo} p_term ${item.primaryTerm}"
-                        }
-                    }
-                }
-
-            val timeTaken =
-                measureTimedValue {
-                    client.bulk(callBack = itemCallBack) {
-                        sessions.forEach { session ->
-                            index(
-                                source = DEFAULT_JSON.encodeToString(Session.serializer(), session),
-                                index = INDEX_NAME,
-                                id = session.sessionId,
-                            )
-                        }
-                    }
-                }
+            val timeTaken = measureTimedValue { ingester.ingest(INDEX_NAME, sessions) }
 
             logger.info { "Time taken to index - $timeTaken" }
         } else {
diff --git a/src/main/kotlin/no/java/conf/service/search/ElasticIndexer.kt b/src/main/kotlin/no/java/conf/service/search/ElasticIndexer.kt
@@ -0,0 +1,43 @@
+package no.java.conf.service.search
+
+import com.jillesvangurp.ktsearch.SearchClient
+import com.jillesvangurp.ktsearch.createIndex
+import com.jillesvangurp.ktsearch.deleteIndex
+import no.java.conf.model.sessions.Session
+import no.java.conf.model.sessions.Speaker
+import kotlin.time.Duration.Companion.seconds
+
+private const val REPLICAS = 0
+private const val SHARDS = 3
+private const val REFRESH = 10
+
+class ElasticIndexer(
+    private val client: SearchClient,
+) {
+    suspend fun recreateIndex(indexName: String) {
+        client.deleteIndex(indexName, ignoreUnavailable = true)
+
+        client.createIndex(indexName) {
+            settings {
+                replicas = REPLICAS
+                shards = SHARDS
+                refreshInterval = REFRESH.seconds
+            }
+            mappings(dynamicEnabled = false) {
+                text(Session::title)
+                text(Session::abstract)
+                text(Session::intendedAudience)
+                keyword(Session::year)
+                keyword(Session::video)
+                keyword(Session::sessionId)
+                keyword(Session::format)
+                keyword(Session::language)
+                nestedField("speakers") {
+                    text(Speaker::name)
+                    keyword(Speaker::twitter)
+                    text(Speaker::bio)
+                }
+            }
+        }
+    }
+}
diff --git a/src/main/kotlin/no/java/conf/service/search/ElasticIngester.kt b/src/main/kotlin/no/java/conf/service/search/ElasticIngester.kt
@@ -0,0 +1,57 @@
+package no.java.conf.service.search
+
+import com.jillesvangurp.ktsearch.BulkItemCallBack
+import com.jillesvangurp.ktsearch.BulkResponse
+import com.jillesvangurp.ktsearch.OperationType
+import com.jillesvangurp.ktsearch.SearchClient
+import com.jillesvangurp.ktsearch.bulk
+import com.jillesvangurp.serializationext.DEFAULT_JSON
+import io.github.oshai.kotlinlogging.KotlinLogging
+import no.java.conf.model.sessions.Session
+
+private val logger = KotlinLogging.logger {}
+
+class ElasticIngester(
+    private val client: SearchClient,
+) {
+    val itemCallBack =
+        object : BulkItemCallBack {
+            override fun bulkRequestFailed(
+                e: Exception,
+                ops: List<Pair<String, String?>>,
+            ) {
+                logger.error(e) { "Bulk failed" }
+            }
+
+            override fun itemFailed(
+                operationType: OperationType,
+                item: BulkResponse.ItemDetails,
+            ) {
+                logger.warn { "${operationType.name} failed ${item.id} with ${item.status}" }
+            }
+
+            override fun itemOk(
+                operationType: OperationType,
+                item: BulkResponse.ItemDetails,
+            ) {
+                logger.trace {
+                    "${operationType.name} completed ${item.id} seq ${item.seqNo} p_term ${item.primaryTerm}"
+                }
+            }
+        }
+
+    suspend fun ingest(
+        indexName: String,
+        sessions: List<Session>
+    ) {
+        client.bulk(callBack = itemCallBack) {
+            sessions.forEach { session ->
+                index(
+                    source = DEFAULT_JSON.encodeToString(Session.serializer(), session),
+                    index = indexName,
+                    id = session.sessionId,
+                )
+            }
+        }
+    }
+}
diff --git a/src/test/kotlin/no/java/conf/StateTest.kt b/src/test/kotlin/no/java/conf/StateTest.kt
@@ -5,19 +5,26 @@ import io.ktor.client.request.get
 import io.ktor.client.statement.bodyAsText
 import io.ktor.http.HttpStatusCode
 import io.ktor.server.testing.testApplication
+import io.mockk.coEvery
+import io.mockk.just
 import io.mockk.mockk
+import io.mockk.runs
 import kotlinx.coroutines.runBlocking
 import no.java.conf.plugins.configureSearchRouting
 import no.java.conf.service.SearchService
+import no.java.conf.service.search.ElasticIndexer
+import no.java.conf.service.search.ElasticIngester
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
 class StateTest {
     @Test
-    fun testNewState() {
+    fun testNew() {
         val searchClient = mockk<SearchClient>()
+        val indexer = mockk<ElasticIndexer>()
+        val ingester = mockk<ElasticIngester>()
 
-        val service = SearchService(searchClient, false)
+        val service = SearchService(searchClient, indexer, ingester, false)
 
         testApplication {
             application {
@@ -32,10 +39,66 @@ class StateTest {
     }
 
     @Test
-    fun testSkipIndex() {
+    fun testMapped() {
         val searchClient = mockk<SearchClient>()
+        val indexer = mockk<ElasticIndexer>()
+        val ingester = mockk<ElasticIngester>()
 
-        val service = SearchService(searchClient, true)
+        val service = SearchService(searchClient, indexer, ingester, false)
+
+        coEvery { indexer.recreateIndex(any()) } just runs
+
+        runBlocking {
+            service.setup()
+        }
+
+        testApplication {
+            application {
+                configureSearchRouting(service)
+            }
+
+            client.get("/api/search/state").apply {
+                assertEquals(HttpStatusCode.OK, status)
+                assertEquals(bodyAsText(), "MAPPED")
+            }
+        }
+    }
+
+    @Test
+    fun testIndexed() {
+        val searchClient = mockk<SearchClient>()
+        val indexer = mockk<ElasticIndexer>()
+        val ingester = mockk<ElasticIngester>()
+
+        val service = SearchService(searchClient, indexer, ingester, false)
+
+        coEvery { indexer.recreateIndex(any()) } just runs
+        coEvery { ingester.ingest(any(), any()) } just runs
+
+        runBlocking {
+            service.setup()
+            service.ingest(emptyList())
+        }
+
+        testApplication {
+            application {
+                configureSearchRouting(service)
+            }
+
+            client.get("/api/search/state").apply {
+                assertEquals(HttpStatusCode.OK, status)
+                assertEquals(bodyAsText(), "INDEXED")
+            }
+        }
+    }
+
+    @Test
+    fun testSkipIndexNew() {
+        val searchClient = mockk<SearchClient>()
+        val indexer = mockk<ElasticIndexer>()
+        val ingester = mockk<ElasticIngester>()
+
+        val service = SearchService(searchClient, indexer, ingester, true)
 
         testApplication {
             application {
@@ -52,8 +115,10 @@ class StateTest {
     @Test
     fun testSkipIndexMapped() {
         val searchClient = mockk<SearchClient>()
+        val indexer = mockk<ElasticIndexer>()
+        val ingester = mockk<ElasticIngester>()
 
-        val service = SearchService(searchClient, true)
+        val service = SearchService(searchClient, indexer, ingester, true)
 
         runBlocking {
             service.setup()
@@ -74,8 +139,10 @@ class StateTest {
     @Test
     fun testSkipIndexIndexed() {
         val searchClient = mockk<SearchClient>()
+        val indexer = mockk<ElasticIndexer>()
+        val ingester = mockk<ElasticIngester>()
 
-        val service = SearchService(searchClient, true)
+        val service = SearchService(searchClient, indexer, ingester, true)
 
         runBlocking {
             service.setup()
