// Copyright 2017-2020 @polkadot/rpc-core authors & contributors // SPDX-License-Identifier: Apache-2.0 import { ProviderInterface, ProviderInterfaceCallback } from '@polkadot/rpc-provider/types'; import { Hash } from '@polkadot/types/interfaces'; import { AnyJson, Codec, DefinitionRpc, DefinitionRpcExt, DefinitionRpcSub, Registry } from '@polkadot/types/types'; import { RpcInterface, RpcInterfaceMethod } from './types'; import memoizee from 'memoizee'; import { Observable, Observer } from 'rxjs'; import { publishReplay, refCount } from 'rxjs/operators'; import jsonrpc from '@polkadot/types/interfaces/jsonrpc'; import { Option, StorageKey, Vec } from '@polkadot/types'; import { createClass, createTypeUnsafe } from '@polkadot/types/create'; import { assert, hexToU8a, isFunction, isNull, isUndefined, logger, u8aToU8a } from '@polkadot/util'; import { drr, normalizer, refCountDelay } from './util'; interface StorageChangeSetJSON { block: string; changes: [string, string | null][]; } type OutputType = 'json' | 'raw' | 'scale'; const l = logger('rpc-core'); const EMPTY_META = { fallback: undefined, modifier: { isOptional: true }, type: { asMap: { linked: { isTrue: false } }, isMap: false } }; // utility method to create a nicely-formatted error /** @internal */ function logErrorMessage (method: string, { params, type }: DefinitionRpc, error: Error): void { const inputs = params.map(({ isOptional, name, type }): string => `${name}${isOptional ? '?' : ''}: ${type}` ).join(', '); l.error(`${method}(${inputs}): ${type}:: ${error.message}`); } /** * @name Rpc * @summary The API may use a HTTP or WebSockets provider. * @description It allows for querying a Polkadot Client Node. * WebSockets provider is recommended since HTTP provider only supports basic querying. * * ```mermaid * graph LR; * A[Api] --> |WebSockets| B[WsProvider]; * B --> |endpoint| C[ws://127.0.0.1:9944] * ``` * * @example *
* * ```javascript * import Rpc from '@polkadot/rpc-core'; * import { WsProvider } from '@polkadot/rpc-provider/ws'; * * const provider = new WsProvider('ws://127.0.0.1:9944'); * const rpc = new Rpc(provider); * ``` */ export class RpcCore implements RpcInterface { #instanceId: string; #registryDefault: Registry; #getBlockRegistry?: (blockHash: string | Uint8Array) => Promise<{ registry: Registry }>; readonly #storageCache = new Map(); public readonly mapping = new Map(); public readonly provider: ProviderInterface; public readonly sections: string[] = []; // Ok, this is quite horrible - we really should not be using the ! here, but we are actually assigning // these via the createInterfaces inside the constructor. However... this is not quite visible. The reason // why we don't do for individual assignments is to allow user-defined RPCs to also be defined public readonly author!: RpcInterface['author']; public readonly babe!: RpcInterface['babe']; public readonly chain!: RpcInterface['chain']; public readonly childstate!: RpcInterface['childstate']; public readonly contracts!: RpcInterface['contracts']; public readonly engine!: RpcInterface['engine']; public readonly grandpa!: RpcInterface['grandpa']; public readonly offchain!: RpcInterface['offchain']; public readonly payment!: RpcInterface['payment']; public readonly rpc!: RpcInterface['rpc']; public readonly state!: RpcInterface['state']; public readonly syncstate!: RpcInterface['syncstate']; public readonly system!: RpcInterface['system']; /** * @constructor * Default constructor for the Api Object * @param {ProviderInterface} provider An API provider using HTTP or WebSocket */ constructor (instanceId: string, registry: Registry, provider: ProviderInterface, userRpc: Record> = {}) { // eslint-disable-next-line @typescript-eslint/unbound-method assert(provider && isFunction(provider.send), 'Expected Provider to API create'); this.#instanceId = instanceId; this.#registryDefault = registry; this.provider = provider; const sectionNames = Object.keys(jsonrpc); // these are the base keys (i.e. part of jsonrpc) this.sections.push(...sectionNames); this.addUserInterfaces(userRpc); } /** * @description Returns the connected status of a provider */ public get isConnected (): boolean { return this.provider.isConnected; } /** * @description Manually connect from the attached provider */ public connect (): Promise { return this.provider.connect(); } /** * @description Manually disconnect from the attached provider */ public disconnect (): Promise { return this.provider.disconnect(); } /** * @description Sets a registry swap (typically from Api) */ public setRegistrySwap (registrySwap: (blockHash: string | Uint8Array) => Promise<{ registry: Registry }>): void { this.#getBlockRegistry = registrySwap; } public addUserInterfaces
(userRpc: Record>): void { // add any extra user-defined sections this.sections.push(...Object.keys(userRpc).filter((key) => !this.sections.includes(key))); // decorate the sections with base and user methods this.sections.forEach((sectionName): void => { (this as Record)[sectionName as Section] ||= {}; const section = this[sectionName as Section] as Record; Object .entries({ ...this._createInterface(sectionName, jsonrpc[sectionName as 'babe'] || {}), ...this._createInterface(sectionName, userRpc[sectionName] || {}) }) .forEach(([key, value]): void => { section[key] ||= value; }); }); } private _createInterface
(section: string, methods: Record): RpcInterface[Section] { return Object .entries(methods) .filter(([method, { endpoint }]) => !this.mapping.has(endpoint || `${section}_${method}`)) .reduce((exposed, [method, { endpoint }]): RpcInterface[Section] => { const def = methods[method]; const isSubscription = !!(def as DefinitionRpcSub).pubsub; const jsonrpc = endpoint || `${section}_${method}`; this.mapping.set(jsonrpc, { ...def, isSubscription, jsonrpc, method, section }); // FIXME Remove any here // To do so, remove `RpcInterfaceMethod` from './types.ts', and refactor // every method inside this class to take: // `` // Not doing so, because it makes this class a little bit less readable, // and leaving it as-is doesn't harm much (exposed as Record)[method] = isSubscription ? this._createMethodSubscribe(section, method, def as DefinitionRpcSub) : this._createMethodSend(section, method, def); return exposed; }, {} as RpcInterface[Section]); } private _memomize (creator: (outputAs: OutputType) => (...values: any[]) => Observable): RpcInterfaceMethod & memoizee.Memoized { const memoized = memoizee(creator('scale') as RpcInterfaceMethod, { // Dynamic length for argument length: false, // Normalize args so that different args that should be cached // together are cached together. // E.g.: `query.my.method('abc') === query.my.method(createType('AccountId', 'abc'));` normalizer: normalizer(this.#instanceId) }); memoized.json = creator('json'); memoized.raw = creator('raw'); return memoized; } private _createMethodSend (section: string, method: string, def: DefinitionRpc): RpcInterfaceMethod { const rpcName = def.endpoint || `${section}_${method}`; const hashIndex = def.params.findIndex(({ isHistoric }) => isHistoric); const cacheIndex = def.params.findIndex(({ isCached }) => isCached); let memoized: null | RpcInterfaceMethod & memoizee.Memoized = null; // execute the RPC call, doing a registry swap for historic as applicable const callWithRegistry = async (outputAs: OutputType, values: any[]): Promise => { const hash = hashIndex === -1 ? undefined : values[hashIndex] as Uint8Array; const { registry } = hash && this.#getBlockRegistry ? await this.#getBlockRegistry(hash) : { registry: this.#registryDefault }; const params = this._formatInputs(registry, def, values); const data = await this.provider.send(rpcName, params.map((param): AnyJson => param.toJSON())) as AnyJson; return outputAs === 'scale' ? this._formatOutput(registry, method, def, params, data) : registry.createType(outputAs === 'raw' ? 'Raw' : 'Json', data); }; const creator = (outputAs: OutputType) => (...values: any[]): Observable => { const isDelayed = (hashIndex !== -1 && !!values[hashIndex]) || (cacheIndex !== -1 && !!values[cacheIndex]); return new Observable((observer: Observer): VoidCallback => { callWithRegistry(outputAs, values) .then((value): void => { observer.next(value); observer.complete(); }) .catch((error): void => { logErrorMessage(method, def, error); observer.error(error); observer.complete(); }); return (): void => { // delete old results from cache memoized?.delete(...values); }; }).pipe( publishReplay(1), // create a Replay(1) isDelayed ? refCountDelay() // Unsubscribe after delay : refCount() ); }; memoized = this._memomize(creator); return memoized; } // create a subscriptor, it subscribes once and resolves with the id as subscribe private _createSubscriber ({ paramsJson, subName, subType, update }: { subType: string; subName: string; paramsJson: AnyJson[]; update: ProviderInterfaceCallback }, errorHandler: (error: Error) => void): Promise { return new Promise((resolve, reject): void => { this.provider .subscribe(subType, subName, paramsJson, update) .then(resolve) .catch((error): void => { errorHandler(error); reject(error); }); }); } private _createMethodSubscribe (section: string, method: string, def: DefinitionRpcSub): RpcInterfaceMethod { const [updateType, subMethod, unsubMethod] = def.pubsub; const subName = `${section}_${subMethod}`; const unsubName = `${section}_${unsubMethod}`; const subType = `${section}_${updateType}`; let memoized: null | RpcInterfaceMethod & memoizee.Memoized = null; const creator = (outputAs: OutputType) => (...values: unknown[]): Observable => { return new Observable((observer: Observer): VoidCallback => { // Have at least an empty promise, as used in the unsubscribe let subscriptionPromise: Promise = Promise.resolve(null); const registry = this.#registryDefault; const errorHandler = (error: Error): void => { logErrorMessage(method, def, error); observer.error(error); }; try { const params = this._formatInputs(registry, def, values); const paramsJson = params.map((param): AnyJson => param.toJSON()); const update = (error?: Error | null, result?: any): void => { if (error) { logErrorMessage(method, def, error); return; } try { observer.next( outputAs === 'scale' ? this._formatOutput(registry, method, def, params, result) : registry.createType(outputAs === 'raw' ? 'Raw' : 'Json', result) ); } catch (error) { observer.error(error); } }; subscriptionPromise = this._createSubscriber({ paramsJson, subName, subType, update }, errorHandler); } catch (error) { errorHandler(error); } // Teardown logic return (): void => { // Delete from cache, so old results don't hang around memoized?.delete(...values); // Unsubscribe from provider subscriptionPromise .then((subscriptionId): Promise => isNull(subscriptionId) ? Promise.resolve(false) : this.provider.unsubscribe(subType, unsubName, subscriptionId) ) .catch((error: Error) => logErrorMessage(method, def, error)); }; }).pipe(drr()); }; memoized = this._memomize(creator); return memoized; } private _formatInputs (registry: Registry, def: DefinitionRpc, inputs: any[]): Codec[] { const reqArgCount = def.params.filter(({ isOptional }): boolean => !isOptional).length; const optText = reqArgCount === def.params.length ? '' : ` (${def.params.length - reqArgCount} optional)`; assert(inputs.length >= reqArgCount && inputs.length <= def.params.length, `Expected ${def.params.length} parameters${optText}, ${inputs.length} found instead`); return inputs.map((input, index): Codec => createTypeUnsafe(registry, def.params[index].type, [input]) ); } private _treatAsHex (key: StorageKey): boolean { // :code is problematic - it does not have the length attached, which is // unlike all other storage entries where it is indeed properly encoded return ['0x3a636f6465'].includes(key.toHex()); } private _formatOutput (registry: Registry, method: string, rpc: DefinitionRpc, params: Codec[], result?: any): Codec | Codec[] { if (rpc.type === 'StorageData') { const key = params[0] as StorageKey; return this._formatStorageData(registry, key, result); } else if (rpc.type === 'StorageChangeSet') { const keys = params[0] as Vec; return keys ? this._formatStorageSet(registry, keys, (result as StorageChangeSetJSON).changes) : registry.createType('StorageChangeSet', result); } else if (rpc.type === 'Vec') { const mapped = (result as StorageChangeSetJSON[]).map(({ block, changes }): [Hash, Codec[]] => [ registry.createType('Hash', block), this._formatStorageSet(registry, params[0] as Vec, changes) ]); // we only query at a specific block, not a range - flatten return method === 'queryStorageAt' ? mapped[0][1] : mapped as unknown as Codec[]; } return createTypeUnsafe(registry, rpc.type, [result]); } private _formatStorageData (registry: Registry, key: StorageKey, value: string | null): Codec { // single return value (via state.getStorage), decode the value based on the // outputType that we have specified. Fallback to Raw on nothing const type = key.outputType || 'Raw'; const meta = key.meta || EMPTY_META; const isEmpty = isNull(value); // we convert to Uint8Array since it maps to the raw encoding, all // data will be correctly encoded (incl. numbers, excl. :code) const input = isEmpty ? null : this._treatAsHex(key) ? value : u8aToU8a(value); if (meta.modifier.isOptional) { let inner = null; if (!isEmpty) { try { inner = createTypeUnsafe(registry, type, [input], true); } catch (error) { l.error(`Unable to decode storage ${key.section || 'unknown'}.${key.method || 'unknown'}:`, (error as Error).message); } } return new Option(registry, createClass(registry, type), inner); } try { return createTypeUnsafe(registry, type, [ isEmpty ? meta.fallback ? hexToU8a(meta.fallback.toHex()) : undefined : input ], true); } catch (error) { l.error(`Unable to decode storage ${key.section || 'unknown'}.${key.method || 'unknown'}:`, (error as Error).message); return registry.createType('Raw', input); } } private _formatStorageSet (registry: Registry, keys: Vec, changes: [string, string | null][]): Codec[] { // For StorageChangeSet, the changes has the [key, value] mappings const withCache = keys.length !== 1; // multiple return values (via state.storage subscription), decode the values // one at a time, all based on the query types. Three values can be returned - // - Codec - There is a valid value, non-empty // - null - The storage key is empty return keys.reduce((results: Codec[], key: StorageKey, index: number): Codec[] => { results.push(this._formatStorageSetEntry(registry, key, changes, withCache, index)); return results; }, []); } private _formatStorageSetEntry (registry: Registry, key: StorageKey, changes: [string, string | null][], witCache: boolean, entryIndex: number): Codec { // Fallback to Raw (i.e. just the encoding) if we don't have a specific type const type = key.outputType || 'Raw'; const hexKey = key.toHex(); const meta = key.meta || EMPTY_META; const found = changes.find(([key]) => key === hexKey); // if we don't find the value, this is our fallback // - in the case of an array of values, fill the hole from the cache // - if a single result value, don't fill - it is not an update hole // - fallback to an empty option in all cases const value = isUndefined(found) ? (witCache && this.#storageCache.get(hexKey)) || null : found[1]; const isEmpty = isNull(value); const input = isEmpty || this._treatAsHex(key) ? value : u8aToU8a(value); // store the retrieved result - the only issue with this cache is that there is no // clearing of it, so very long running processes (not just a couple of hours, longer) // will increase memory beyond what is allowed. this.#storageCache.set(hexKey, value); if (meta.modifier.isOptional) { let inner = null; if (!isEmpty) { try { inner = createTypeUnsafe(registry, type, [input], true); } catch (error) { l.error(`Unable to decode storage ${key.section || 'unknown'}.${key.method || 'unknown'}: entry ${entryIndex}:`, (error as Error).message); } } return new Option(registry, createClass(registry, type), inner); } try { return createTypeUnsafe(registry, type, [ isEmpty ? meta.fallback ? hexToU8a(meta.fallback.toHex()) : undefined : input ], true); } catch (error) { l.error(`Unable to decode storage ${key.section || 'unknown'}.${key.method || 'unknown'}: entry ${entryIndex}:`, (error as Error).message); return registry.createType('Raw', input); } } }