WIP Blockbook

src/hub.ts

@@ -10,6 +10,12 @@ import {
   SubscribeResult
 } from './types/changeProtocol'
 
+const pluginGauge = new Gauge({
+  name: 'change_plugin_count',
+  help: 'Active change-server plugins',
+  labelNames: ['pluginId'] as const
+})
+
 const connectionGauge = new Gauge({
   name: 'change_connection_count',
   help: 'Active websocket connections'
@@ -69,6 +75,16 @@ export function makeAddressHub(plugins: AddressPlugin[]): AddressHub {
         codec?.remoteMethods.update([pluginId, address, checkpoint])
       }
     })
+
+    plugin.on('connect', () => {
+      pluginGauge.inc({ pluginId })
+      // TODO: Tell clients
+    })
+    plugin.on('disconnect', () => {
+      pluginGauge.dec({ pluginId })
+      // TODO: Tell clients
+      // Clear active addresses list
+    })
   }
 
   function handleConnection(ws: WebSocket): void {

src/jsonRpc.ts

@@ -91,15 +91,26 @@ export function makeRpcProtocol<
    * Methods supported on the client side.
    */
   clientMethods: ClientCleaners
+
+  /**
+   * Optionally used if the protocol differs from strict JSON-RPC 2.0.
+   */
+  asCall?: Cleaner<JsonRpcCall>
+  asReturn?: Cleaner<JsonRpcReturn>
 }): RpcProtocol<Methods<ServerCleaners>, Methods<ClientCleaners>> {
-  const { serverMethods, clientMethods } = opts
+  const {
+    serverMethods,
+    clientMethods,
+    asCall = asJsonRpcCall,
+    asReturn = asJsonRpcReturn
+  } = opts
 
   return {
     makeServerCodec(opts) {
-      return makeCodec(serverMethods, clientMethods, opts)
+      return makeCodec(serverMethods, clientMethods, asCall, asReturn, opts)
     },
     makeClientCodec(opts) {
-      return makeCodec(clientMethods, serverMethods, opts)
+      return makeCodec(clientMethods, serverMethods, asCall, asReturn, opts)
     }
   }
 }
@@ -123,9 +134,13 @@ type Methods<T> = {
 function makeCodec(
   localCleaners: MethodCleaners,
   remoteCleaners: MethodCleaners,
+  asCall: Cleaner<JsonRpcCall>,
+  asReturn: Cleaner<JsonRpcReturn>,
   opts: RpcCodecOpts<any>
 ): RpcCodec<any> {
   const { handleError, handleSend, localMethods } = opts
+  const wasCall = uncleaner(asCall)
+  const wasReturn = uncleaner(asReturn)
 
   const sendError = async (
     code: number,
@@ -134,7 +149,7 @@ function makeCodec(
   ): Promise<void> =>
     await handleSend(
       JSON.stringify(
-        wasJsonRpcReturn({
+        wasReturn({
           jsonrpc: '2.0',
           result: undefined,
           error: { code, message, data: undefined },
@@ -160,7 +175,7 @@ function makeCodec(
       remoteMethods[name] = (params: unknown): void => {
         handleSend(
           JSON.stringify(
-            wasJsonRpcCall({
+            wasCall({
               jsonrpc: '2.0',
               method: name,
               params: wasParams(params),
@@ -183,7 +198,7 @@ function makeCodec(
 
         handleSend(
           JSON.stringify(
-            wasJsonRpcCall({
+            wasCall({
               jsonrpc: '2.0',
               method: name,
               params: wasParams(params),
@@ -208,8 +223,8 @@ function makeCodec(
     }
 
     // TODO: We need to add support for batch calls.
-    const call = asMaybe(asJsonRpcCall)(json)
-    const response = asMaybe(asJsonRpcReturn)(json)
+    const call = asMaybe(asCall)(json)
+    const response = asMaybe(asReturn)(json)
 
     if (call != null) {
       const { method, id, params } = call
@@ -251,7 +266,7 @@ function makeCodec(
           (result: unknown) => {
             handleSend(
               JSON.stringify(
-                wasJsonRpcReturn({
+                wasReturn({
                   jsonrpc: '2.0',
                   result: wasResult(result),
                   error: undefined,
@@ -350,7 +365,6 @@ const asJsonRpcCall = asObject<JsonRpcCall>({
   params: asUnknown,
   id: asOptional(asRpcId)
 })
-const wasJsonRpcCall = uncleaner(asJsonRpcCall)
 
 const asJsonRpcReturn = asObject<JsonRpcReturn>({
   jsonrpc: asValue('2.0'),
@@ -364,4 +378,3 @@ const asJsonRpcReturn = asObject<JsonRpcReturn>({
   ),
   id: asRpcId
 })
-const wasJsonRpcReturn = uncleaner(asJsonRpcReturn)

src/plugins/allPlugins.ts

@@ -1,6 +1,61 @@
+import { serverConfig } from '../server-config'
+import { AddressPlugin } from '../types/addressPlugin'
+import { BlockbookOptions, makeBlockbook } from './blockbook'
 import { makeFakePlugin } from './fakePlugin'
 
+function makeNowNode(opts: BlockbookOptions): AddressPlugin {
+  return makeBlockbook({
+    ...opts,
+    safeUrl: opts.url,
+    url: opts.url + '/' + serverConfig.nowNodesApiKey
+  })
+}
+
 export const allPlugins = [
+  // Bitcoin family:
+  makeNowNode({
+    pluginId: 'bitcoin',
+    url: 'wss://btcbook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'bitcoincash',
+    url: 'wss://bchbook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'dogecoin',
+    url: 'wss://dogebook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'litecoin',
+    url: 'wss://ltcbook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'qtum',
+    url: 'wss://qtum-blockbook.nownodes.io/wss'
+  }),
+
+  // Ethereum family:
+  makeNowNode({
+    pluginId: 'arbitrum',
+    url: 'wss://arb-blockbook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'avalanche',
+    url: 'wss://avax-blockbook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'base',
+    url: 'wss://base-blockbook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'ethereum',
+    url: 'wss://eth-blockbook.nownodes.io/wss'
+  }),
+  makeNowNode({
+    pluginId: 'polygon',
+    url: 'wss://maticbook.nownodes.io/wss'
+  }),
+
   // Testing:
   makeFakePlugin()
 ]

src/plugins/blockbook.ts

@@ -0,0 +1,87 @@
+import WebSocket from 'ws'
+import { makeEvents } from 'yavent'
+
+import { messageToString } from '../messageToString'
+import { AddressPlugin, PluginEvents } from '../types/addressPlugin'
+import { blockbookProtocol } from '../types/blockbookProtocol'
+
+export interface BlockbookOptions {
+  pluginId: string
+
+  /** A clean URL for logging */
+  safeUrl?: string
+
+  /** The actual connection URL */
+  url: string
+}
+
+export function makeBlockbook(opts: BlockbookOptions): AddressPlugin {
+  const { pluginId, url } = opts
+
+  const ws = new WebSocket(url)
+  const [on, emit] = makeEvents<PluginEvents>()
+  const codec = blockbookProtocol.makeClientCodec({
+    handleError(error) {
+      console.log(error)
+    },
+    async handleSend(text) {
+      console.log('send', text)
+      ws.send(text)
+    },
+    localMethods: {
+      subscribeAddresses({ address }) {
+        emit('update', { address })
+      }
+    }
+  })
+
+  ws.on('message', message => {
+    const text = messageToString(message)
+    console.log(text)
+    codec.handleMessage(text)
+  })
+  ws.on('open', () => emit('connect', undefined))
+  ws.on('close', () => {
+    codec.handleClose()
+    emit('disconnect', undefined)
+  })
+
+  return {
+    pluginId,
+    on,
+
+    subscribe(address) {
+      codec.remoteMethods
+        .subscribeAddresses({ addresses: [address] })
+        .catch(error => console.log(error))
+    },
+
+    unsubscribe(address) {
+      codec.remoteMethods
+        .unsubscribeAddresses({ addresses: [address] })
+        .catch(error => console.log(error))
+    },
+
+    async scanAddress(address, checkpoint): Promise<boolean> {
+      const out = await codec.remoteMethods.getAccountInfo({
+        descriptor: address, // Address or xpub
+        details: 'txids',
+        tokens: undefined, // 'derived',
+        from: 860728, // checkpoint == null ? checkpoint : parseInt(checkpoint),
+        to: undefined, // checkpoint == null ? checkpoint : parseInt(checkpoint),
+        page: undefined,
+        pageSize: undefined,
+        contractFilter: undefined,
+        secondaryCurrency: undefined,
+        gap: undefined
+      })
+
+      console.log(out)
+      // return out.unconfirmedTxs > 0 || out.txids?.length > 0
+
+      if (out.unconfirmedTxs > 0) return true
+      if (out.txids != null && out.txids.length > 0) return true
+      return false
+    }
+  }
+}

src/plugins/blocktest.ts

@@ -0,0 +1,16 @@
+import { allPlugins } from './allPlugins'
+
+const bitcoin = allPlugins.find(plugin => plugin.pluginId === 'bitcoin')
+if (bitcoin != null) {
+  bitcoin.on('connect', () => {
+    bitcoin.subscribe('bc1qmgwnfjlda4ns3g6g3yz74w6scnn9yu2ts82yyc')
+    bitcoin
+      .scanAddress?.(
+        'bc1qmgwnfjlda4ns3g6g3yz74w6scnn9yu2ts82yyc' /* '860728' */
+      )
+      .then(
+        x => console.log(x),
+        e => console.log(e)
+      )
+  })
+}

src/plugins/ethereum.ts

@@ -0,0 +1,15 @@
+import { AddressPlugin } from '../types/addressPlugin'
+
+export function makeEthereum(opts: {}): AddressPlugin {
+  return {
+    pluginId: 'ethereum',
+
+    on(event, callback) {
+      return () => {}
+    },
+
+    subscribe() {},
+
+    unsubscribe(address) {}
+  }
+}

src/serverConfig.ts

@@ -15,7 +15,10 @@ const asServerConfig = asObject({
   listenPort: asOptional(asNumber, 8008),
   metricsHost: asOptional(asString, '127.0.0.1'),
   metricsPort: asOptional(asNumber, 8009),
-  publicUri: asOptional(asString, 'https://address1.edge.app')
+  publicUri: asOptional(asString, 'https://address1.edge.app'),
+
+  // Resources:
+  nowNodesApiKey: asOptional(asString, '')
 })
 
 export const serverConfig = makeConfig(