solution: support xpub addresses for bitcoin

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackBitcoinAddress.kt

@@ -47,25 +47,43 @@ class TrackBitcoinAddress(
                 && BlockchainType.fromBlockchain(chain) == BlockchainType.BITCOIN && multistreamHolder.isAvailable(chain)
     }
 
-    fun allAddresses(request: BlockchainOuterClass.BalanceRequest): List<String>? {
+    fun allAddresses(api: BitcoinMultistream, request: BlockchainOuterClass.BalanceRequest): Flux<String> {
         if (!request.hasAddress()) {
-            return null
+            return Flux.empty()
         }
         return when {
+            request.address.hasAddressXpub() -> {
+                val xpubAddresses = api.getXpubAddresses()
+                        ?: return Flux.error(IllegalStateException("Xpub verification is not available"))
+
+                val addressXpub = request.address.addressXpub
+                if (!addressXpub.xpub.isValidUtf8) {
+                    return Flux.error(IllegalArgumentException("Invalid xpub string"))
+                }
+                val xpub = addressXpub.xpub.toStringUtf8()
+                val start = Math.max(0, addressXpub.start).toInt()
+                val limit = Math.min(100, Math.max(1, addressXpub.limit)).toInt()
+                xpubAddresses.activeAddresses(xpub, start, limit)
+                        .map { it.toString() }
+                        .doOnError { t -> log.error("Failed to process xpub. ${t.javaClass}:${t.message}") }
+            }
             request.address.hasAddressSingle() -> {
-                listOf(request.address.addressSingle.address)
+                Flux.just(request.address.addressSingle.address)
             }
             request.address.hasAddressMulti() -> {
-                request.address.addressMulti.addressesList
-                        .map { addr -> addr.address }
-                        .sorted()
+                Flux.fromIterable(
+                        request.address.addressMulti.addressesList
+                                .map { addr -> addr.address }
+                                //TODO why sorted?
+                                .sorted()
+                )
             }
-            else -> null
+            else -> Flux.error(IllegalArgumentException("Unsupported address type"))
         }
     }
 
-    fun requestBalances(chain: Chain, api: BitcoinMultistream, addresses: List<String>, includeUtxo: Boolean): Flux<AddressBalance> {
-        return Flux.fromIterable(addresses)
+    fun requestBalances(chain: Chain, api: BitcoinMultistream, addresses: Flux<String>, includeUtxo: Boolean): Flux<AddressBalance> {
+        return addresses
                 .map { Address(chain, it) }
                 .flatMap { address ->
                     balanceForAddress(api, address, includeUtxo)
@@ -103,10 +121,7 @@ class TrackBitcoinAddress(
         val chain = Chain.byId(request.asset.chainValue)
         val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java)
                 ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
-        val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
-        if (addresses.isEmpty()) {
-            return Flux.empty()
-        }
+        val addresses = allAddresses(upstream, request) ?: return Flux.error(SilentException("Unsupported address"))
         return requestBalances(chain, upstream, addresses, request.includeUtxo)
                 .map(this@TrackBitcoinAddress::buildResponse)
     }
@@ -116,10 +131,7 @@ class TrackBitcoinAddress(
         val chain = Chain.byId(request.asset.chainValue)
         val upstream = multistreamHolder.getUpstream(chain)?.cast(BitcoinMultistream::class.java)
                 ?: return Flux.error(SilentException.UnsupportedBlockchain(request.asset.chainValue))
-        val addresses = allAddresses(request) ?: return Flux.error(SilentException("Unsupported address"))
-        if (addresses.isEmpty()) {
-            return Flux.empty()
-        }
+        val addresses = allAddresses(upstream, request) ?: return Flux.error(SilentException("Unsupported address"))
         val initial = requestBalances(chain, upstream, addresses, request.includeUtxo)
         val following = upstream.getHead().getFlux()
                 .flatMap { block ->
@@ -164,6 +176,7 @@ class TrackBitcoinAddress(
 
     open class BalanceUtxo(val txid: String, val vout: Int, val value: Long)
 
+    //TODO use bitcoin class for address
     class Address(val chain: Chain, val address: String) {
         val network = if (chain == Chain.BITCOIN) {
             MainNetParams()

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/AddressActiveCheck.kt

@@ -0,0 +1,35 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import org.bitcoinj.core.Address
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+open class AddressActiveCheck(
+        private val esploraClient: EsploraClient
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(AddressActiveCheck::class.java)
+    }
+
+    open fun isActive(address: Address): Mono<Boolean> {
+        //TODO cache with bloom filter
+        return esploraClient.getTransactions(address).map { it.isNotEmpty() }
+    }
+
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt

@@ -40,6 +40,8 @@ open class BitcoinMultistream(
     private var head: Head = EmptyHead()
     private var esplora = upstreams.find { it.esploraClient != null }?.esploraClient
     private var reader = BitcoinReader(this, head, esplora)
+    private var addressActiveCheck: AddressActiveCheck? = null
+    private var xpubAddresses: XpubAddresses? = null
 
     override fun init() {
         if (upstreams.size > 0) {
@@ -48,6 +50,10 @@ open class BitcoinMultistream(
         super.init()
     }
 
+    open fun getXpubAddresses(): XpubAddresses? {
+        return xpubAddresses
+    }
+
     override fun updateHead(): Head {
         head.let {
             if (it is Lifecycle) {
@@ -85,6 +91,8 @@ open class BitcoinMultistream(
         super.onUpstreamsUpdated()
         esplora = upstreams.find { it.esploraClient != null }?.esploraClient
         reader = BitcoinReader(this, this.head, esplora)
+        addressActiveCheck = esplora?.let { AddressActiveCheck(it) }
+        xpubAddresses = addressActiveCheck?.let { XpubAddresses(it) }
     }
 
     override fun setHead(head: Head) {

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt

@@ -80,17 +80,7 @@ class EsploraClient(
         val response = httpClient
                 .get()
                 .uri("$url/address/$address/utxo")
-
-        val json = response
-                .response { header, bytes ->
-                    if (header.status().code() != 200) {
-                        Mono.error(EsploraException("HTTP Code: ${header.status().code()}"))
-                    } else {
-                        bytes.aggregate().asByteArray()
-                    }
-                }
-                .single()
-                .map { bytes -> String(bytes) }
+        val json = parseResponse(response)
 
         return json.map {
             val parsed = Global.objectMapper.readerFor(EsploraUnspent::class.java)
@@ -99,6 +89,32 @@ class EsploraClient(
         }
     }
 
+    fun getTransactions(address: Address): Mono<List<Map<String, Any>>> {
+        val response = httpClient
+                .get()
+                .uri("$url/address/$address/txs")
+        val json = parseResponse(response)
+
+        return json.map {
+            val parsed = Global.objectMapper.readerFor(Map::class.java)
+                    .readValues<Map<String, Any>>(it);
+            parsed.readAll()
+        }
+    }
+
+    private fun parseResponse(response: HttpClient.ResponseReceiver<*>): Mono<String> {
+        return response
+                .response { header, bytes ->
+                    if (header.status().code() != 200) {
+                        Mono.error(EsploraException("HTTP Code: ${header.status().code()} for ${header.fullPath()}"))
+                    } else {
+                        bytes.aggregate().asByteArray()
+                    }
+                }
+                .single()
+                .map { bytes -> String(bytes) }
+    }
+
     class EsploraException(msg: String) : Exception(msg)
 
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/XpubAddresses.kt

@@ -0,0 +1,103 @@
+/**
+ * Copyright (c) 2020 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.upstream.bitcoin
+
+import org.bitcoinj.core.Address
+import org.bitcoinj.core.ECKey
+import org.bitcoinj.core.NetworkParameters
+import org.bitcoinj.crypto.ChildNumber
+import org.bitcoinj.crypto.DeterministicKey
+import org.bitcoinj.crypto.HDKeyDerivation
+import org.bitcoinj.params.MainNetParams
+import org.bitcoinj.params.TestNet3Params
+import org.bitcoinj.script.Script
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.util.function.Tuples
+import java.util.concurrent.atomic.AtomicInteger
+
+open class XpubAddresses(
+        private val addressActiveCheck: AddressActiveCheck
+) {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(XpubAddresses::class.java)
+        private val MAINNET = MainNetParams()
+        private val TESTNET = TestNet3Params()
+        private val INACTIVE_LIMIT = 20
+    }
+
+    open fun allAddresses(xpub: String, start: Int, limit: Int): Flux<Address> {
+        // versions:
+        // https://electrum.readthedocs.io/en/latest/xpub_version_bytes.html
+        // TODO doesn't support SH keys right now. should?
+        val prefix = xpub.substring(0, 4)
+        val type: Script.ScriptType
+        val network: NetworkParameters
+
+        when (prefix) {
+            "xpub" -> {
+                type = Script.ScriptType.P2PKH
+                network = MAINNET
+            }
+            "zpub" -> {
+                type = Script.ScriptType.P2WPKH
+                network = MAINNET
+            }
+            "tpub" -> {
+                type = Script.ScriptType.P2PKH
+                network = TESTNET
+            }
+            "vpub" -> {
+                type = Script.ScriptType.P2WPKH
+                network = TESTNET
+            }
+            else -> return Flux.error(IllegalArgumentException("Unsupported type: $prefix"))
+        }
+
+        val key: DeterministicKey
+        try {
+            key = DeterministicKey.deserializeB58(xpub, network)
+        } catch (t: Throwable) {
+            return Flux.error(t)
+        }
+
+        return Flux.range(start, limit)
+                .map { HDKeyDerivation.deriveChildKey(key, ChildNumber(it, false)) }
+                .map { Address.fromKey(network, ECKey.fromPublicOnly(it.pubKeyPoint), type) }
+    }
+
+    open fun activeAddresses(xpub: String, start: Int, limit: Int): Flux<Address> {
+        val lastActive = AtomicInteger(0)
+        return this.allAddresses(xpub, start, limit)
+                .zipWith(Flux.range(0, limit))
+                .takeUntil {
+                    it.t2 - lastActive.get() >= INACTIVE_LIMIT
+                }
+                .concatMap { toCheck ->
+                    addressActiveCheck.isActive(toCheck.t1)
+                            .doOnNext { active -> if (active) lastActive.set(toCheck.t2) }
+                            .map { Tuples.of(toCheck.t1, it) }
+                }
+                .filter {
+                    it.t2
+                }
+                .map {
+                    it.t1
+                }
+    }
+
+}