add initial height query to block fetcher (by querying genesis)

- bespoke, hand-coded updates to openapi file, as we don't know how this was generated still and it doesn't appear to match the reality of the genesis response produced from testnet

es-api-model/src/main/resources/tendermint-v0.34.12-rpc-openapi-FIXED.yaml

@@ -1905,7 +1905,7 @@ components:
                 - "chain_id"
                 - "initial_height"
                 - "consensus_params"
-                - "validators"
+#                - "validators" # for some reason this schema has this as required, but it comes back as empty from the chain... maybe this is just out of date?
                 - "app_hash"
               properties:
                 genesis_time:
@@ -2938,11 +2938,19 @@ components:
         evidence:
           type: object
           required:
-            - "max_age"
+            - "max_age_num_blocks"
+            - "max_age_duration"
+            - "max_bytes"
           properties:
-            max_age:
+            max_age_num_blocks: # Manually updated these properties to reflect the current properties here in cosmos/tendermint
               type: string
               example: "100000"
+            max_age_duration:
+              type: string
+              example: "172800000000000"
+            max_bytes:
+              type: string
+              example: "1048576"
         validator:
           type: object
           required:

es-core/src/main/kotlin/tech/figure/eventstream/stream/clients/BlockFetcher.kt

@@ -68,6 +68,7 @@ open class BlockFetchException(m: String) : Exception(m)
 interface BlockFetcher {
     suspend fun getBlocksMeta(min: Long, max: Long): List<BlockMeta>?
     suspend fun getCurrentHeight(): Long?
+    suspend fun getInitialHeight(): Long
     suspend fun getBlock(height: Long): BlockData
     suspend fun getBlockResults(height: Long): BlockResultsResponse?
     suspend fun getBlocks(heights: List<Long>, concurrency: Int = DEFAULT_CONCURRENCY, context: CoroutineContext = EmptyCoroutineContext): Flow<BlockData>

es-core/src/main/kotlin/tech/figure/eventstream/stream/clients/TendermintBlockFetcher.kt

@@ -29,6 +29,10 @@ class TendermintBlockFetcher(
             ?: throw BlockFetchException("failed to fetch current block height")
     }
 
+    override suspend fun getInitialHeight(): Long {
+        return tendermintServiceClient.genesis().result.genesis.initialHeight
+    }
+
     override suspend fun getBlock(height: Long): BlockData {
         log.trace("getBlock($height)")
         val block = tendermintServiceClient.block(height).result?.block

es-core/src/main/kotlin/tech/figure/eventstream/stream/clients/TendermintServiceClient.kt

@@ -4,12 +4,14 @@ import tech.figure.eventstream.stream.models.ABCIInfoResponse
 import tech.figure.eventstream.stream.models.BlockResponse
 import tech.figure.eventstream.stream.models.BlockResultsResponse
 import tech.figure.eventstream.stream.models.BlockchainResponse
+import tech.figure.eventstream.stream.models.GenesisResponse
 
 /**
  * A client designed to interact with the Tendermint RPC API.
  */
 interface TendermintServiceClient {
     suspend fun abciInfo(): ABCIInfoResponse
+    suspend fun genesis(): GenesisResponse
     suspend fun block(height: Long?): BlockResponse
     suspend fun blockResults(height: Long?): BlockResultsResponse
     suspend fun blockchain(minHeight: Long?, maxHeight: Long?): BlockchainResponse

es-core/src/main/kotlin/tech/figure/eventstream/stream/clients/TendermintServiceOpenApiClient.kt

@@ -8,6 +8,7 @@ import tech.figure.eventstream.stream.models.ABCIInfoResponse
 import tech.figure.eventstream.stream.models.BlockResponse
 import tech.figure.eventstream.stream.models.BlockResultsResponse
 import tech.figure.eventstream.stream.models.BlockchainResponse
+import tech.figure.eventstream.stream.models.GenesisResponse
 
 /**
  * An OpenAPI generated client designed to interact with the Tendermint RPC API.
@@ -30,6 +31,8 @@ class TendermintServiceOpenApiClient(
 
     override suspend fun abciInfo(): ABCIInfoResponse = abciApi.abciInfo()
 
+    override suspend fun genesis(): GenesisResponse = infoApi.genesis()
+
     override suspend fun block(height: Long?): BlockResponse = infoApi.block(height)
 
     override suspend fun blockResults(height: Long?): BlockResultsResponse = infoApi.blockResults(height)

es-core/src/test/kotlin/tech/figure/eventstream/StreamTests.kt

@@ -32,6 +32,7 @@ import org.junit.jupiter.api.TestInstance
 import org.junit.jupiter.api.assertThrows
 import tech.figure.eventstream.mocks.MockEventStreamService
 import tech.figure.eventstream.stream.clients.TendermintServiceClient
+import tech.figure.eventstream.stream.models.GenesisResponse
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class StreamTests : TestBase() {
@@ -153,6 +154,27 @@ class StreamTests : TestBase() {
             }
         }
 
+        @OptIn(ExperimentalCoroutinesApi::class)
+        @Test
+        fun testGenesisResponse() {
+            val expectedInitialHeight = 10000L
+
+            val tm = ServiceMocker.Builder()
+                .doFor("genesis") {
+                    templates.readAs(
+                        GenesisResponse::class.java,
+                        "genesis/success.json",
+                        mapOf("initial_height" to expectedInitialHeight),
+                    )
+                }
+                .build(MockTendermintServiceClient::class.java)
+
+            dispatcherProvider.runBlockingTest {
+                val initialHeight = tm.genesis().result.genesis.initialHeight
+                assert(initialHeight == expectedInitialHeight) { "expected: $expectedInitialHeight received: $initialHeight" }
+            }
+        }
+
         // duped
         @OptIn(ExperimentalCoroutinesApi::class)
         @Test

es-core/src/testFixtures/kotlin/tech/figure/eventstream/mocks/MockNetAdapter.kt

@@ -23,6 +23,7 @@ import tech.figure.eventstream.utils.Template
 import io.reactivex.Flowable
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
+import tech.figure.eventstream.stream.models.GenesisResponse
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.coroutines.CoroutineContext
 
@@ -90,6 +91,11 @@ fun mockBlockFetcher(template: Template, currentHeight: Long? = null): BlockFetc
             return response!!.result!!.response.lastBlockHeight!!
         }
 
+        override suspend fun getInitialHeight(): Long {
+            val response = template.readAs(GenesisResponse::class.java, "genesis/success.json")
+            return response!!.result.genesis.initialHeight
+        }
+
         override suspend fun getBlock(height: Long): BlockData {
             val block = template.readAs(BlockResponse::class.java, "block/$height.json")
             val results = template.readAs(BlockResultsResponse::class.java, "block_results/$height.json")

es-core/src/testFixtures/kotlin/tech/figure/eventstream/mocks/MockTendermintServiceClient.kt

@@ -5,12 +5,15 @@ import tech.figure.eventstream.stream.models.BlockResultsResponse
 import tech.figure.eventstream.stream.models.BlockchainResponse
 import tech.figure.eventstream.stream.clients.TendermintServiceClient
 import tech.figure.eventstream.stream.models.ABCIInfoResponse
+import tech.figure.eventstream.stream.models.GenesisResponse
 
 class MockTendermintServiceClient(mocker: ServiceMock) : TendermintServiceClient, ServiceMock by mocker {
 
     override suspend fun abciInfo() =
         respondWith<ABCIInfoResponse>("abciInfo")
 
+    override suspend fun genesis() = respondWith<GenesisResponse>("genesis")
+
     override suspend fun block(height: Long?) =
         respondWith<BlockResponse>("block", height)