import CID from 'cids' ; import cloneDeep from 'lodash.clonedeep' ; import { DiagnosticsLogger , IpfsApi , ServiceLogger , StreamUtils , UnreachableCaseError , } from '@ceramicnetwork/common' ; import StreamID from '@ceramicnetwork/streamid' ; import { Repository } from './state-management/repository' ; import { MsgType , PubsubMessage , QueryMessage , ResponseMessage , UpdateMessage , } from './pubsub/pubsub-message' ; import { Pubsub } from './pubsub/pubsub' ; import { Subscription } from 'rxjs' ; import { MessageBus } from './pubsub/message-bus' ; import { LRUMap } from 'lru_map' ; import { PubsubKeepalive } from './pubsub/pubsub-keepalive' ; import { PubsubRateLimit } from './pubsub/pubsub-ratelimit' ; const IPFS_GET_RETRIES = 3 ; const IPFS_GET_TIMEOUT = 30000 ; const IPFS_MAX_COMMIT_SIZE = 256000 ; const IPFS_RESUBSCRIBE_INTERVAL_DELAY = 1000 * 15 ; const MAX_PUBSUB_PUBLISH_INTERVAL = 60 * 1000 ; const IPFS_CACHE_SIZE = 1024 ; function messageTypeToString ( type ) { switch ( type ) { case MsgType . UPDATE : return 'Update' ; case MsgType . QUERY : return 'Query' ; case MsgType . RESPONSE : return 'Response' ; case MsgType . KEEPALIVE : return 'Keepalive' ; default : throw new UnreachableCaseError ( type , `Unsupported message type` ) ; } } /**
 * Ceramic core Dispatcher used for handling messages from pub/sub topic.
 */ export class Dispatcher { readonly messageBus : MessageBus ; readonly dagNodeCache : LRUMap < string , any > ; constructor ( readonly _ipfs , private readonly topic , readonly repository , private readonly _logger , private readonly _pubsubLogger , maxQueriesPerSecond ) { const pubsub = new Pubsub ( _ipfs , topic , IPFS_RESUBSCRIBE_INTERVAL_DELAY , _pubsubLogger , _logger ) ; this . messageBus = new MessageBus ( new PubsubRateLimit ( new PubsubKeepalive ( pubsub , MAX_PUBSUB_PUBLISH_INTERVAL ) , _logger , maxQueriesPerSecond ) ) ; this . messageBus . subscribe ( this . handleMessage . bind ( this ) ) ; this . dagNodeCache = new LRUMap < string , any > ( IPFS_CACHE_SIZE ) ; } param data param streamId async storeCommit ( data , streamId ? ) { try { if ( StreamUtils . isSignedCommitContainer ( data ) ) { const { jws , linkedBlock } = data ; const cid = await this . _ipfs . dag . put ( jws , { format : 'dag-jose' , hashAlg : 'sha2-256' } ) ; await this . _ipfs . block . put ( linkedBlock , { cid : jws . link . toString ( ) } ) ; await this . _restrictCommitSize ( jws . link . toString ( ) ) ; await this . _restrictCommitSize ( cid ) ; return cid ; } const cid = await this . _ipfs . dag . put ( data ) ; await this . _restrictCommitSize ( cid ) ; return cid ; } catch ( e ) { if ( streamId ) { this . _logger . err ( `Error while storing commit to IPFS for stream ${ streamId . toString ( ) }: ${ e }` ) ; } else { this . _logger . err ( `Error while storing commit to IPFS: ${ e }` ) ; } throw e ; } } param cid param streamId async retrieveCommit ( cid , streamId ? ) { try { const result = await this . _getFromIpfs ( cid ) ; await this . _restrictCommitSize ( cid ) ; return result ; } catch ( e ) { if ( streamId ) { this . _logger . err ( `Error while loading commit CID ${ cid . toString ( ) } from IPFS for stream ${ streamId . toString ( ) }: ${ e }` ) ; } else { this . _logger . err ( `Error while loading commit CID ${ cid . toString ( ) } from IPFS: ${ e }` ) ; } throw e ; } } param cid param path async retrieveFromIPFS ( cid , path ? ) { try { return this . _getFromIpfs ( cid , path ) ; } catch ( e ) { this . _logger . err ( `Error while loading CID ${ cid . toString ( ) } from IPFS: ${ e }` ) ; throw e ; } } /**
     * Helper function for loading a CID from IPFS
     */ private async _getFromIpfs ( cid , path ? ) { const asCid = typeof cid === 'string' ? new CID ( cid ) : cid ; const cidAndPath = path ? asCid . toString ( ) + path : asCid . toString ( ) ; const cachedDagNode = await this . dagNodeCache . get ( cidAndPath ) ; if ( cachedDagNode ) return cloneDeep ( cachedDagNode ) ; let dagResult = null ; for ( let retries = IPFS_GET_RETRIES - 1 ; retries >= 0 && dagResult == null ; retries -- ) { try { dagResult = await this . _ipfs . dag . get ( asCid , { timeout : IPFS_GET_TIMEOUT , path } ) ; } catch ( err ) { if ( err . code == 'ERR_TIMEOUT' || err . name == 'TimeoutError' || err . message == 'Request timed out' ) { console . warn ( `Timeout error while loading CID ${ cid . toString ( ) } from IPFS. ${ retries } retries remain` ) ; if ( retries > 0 ) { continue ; } } throw err ; } } await this . dagNodeCache . set ( cidAndPath , dagResult . value ) ; return cloneDeep ( dagResult . value ) ; } param cid private async _restrictCommitSize ( cid ) { const stat = await this . _ipfs . block . stat ( cid , { timeout : IPFS_GET_TIMEOUT } ) ; if ( stat . size > IPFS_MAX_COMMIT_SIZE ) { throw new Error ( `${ cid . toString ( ) } commit size ${ stat . size } exceeds the maximum block size of ${ IPFS_MAX_COMMIT_SIZE }` ) ; } } param streamId param tip publishTip ( streamId , tip ) { return this . publish ( { typ : MsgType . UPDATE , stream : streamId , tip : tip } ) ; } /**
     * Handles one message from the pubsub topic.
     */ async handleMessage ( message ) { try { switch ( message . typ ) { case MsgType . UPDATE : await this . _handleUpdateMessage ( message ) ; break ; case MsgType . QUERY : await this . _handleQueryMessage ( message ) ; break ; case MsgType . RESPONSE : await this . _handleResponseMessage ( message ) ; break ; case MsgType . KEEPALIVE : break ; default : throw new UnreachableCaseError ( message , `Unsupported message type` ) ; } } catch ( e ) { this . _logger . err ( `Error while processing ${ messageTypeToString ( message . typ ) } message from pubsub: ${ e }` ) ; this . _logger . err ( e ) ; } } param message private async _handleUpdateMessage ( message ) { const { stream : streamId , tip } = message ; this . repository . stateManager . update ( streamId , tip ) ; } param message private async _handleQueryMessage ( message ) { const { stream : streamId , id } = message ; const streamState = await this . repository . streamState ( streamId ) ; if ( streamState ) { const tip = streamState . log [ streamState . log . length - 1 ] . cid ; const tipMap = new Map ( ) . set ( streamId . toString ( ) , tip ) ; this . publish ( { typ : MsgType . RESPONSE , id , tips : tipMap } ) ; } } param message private async _handleResponseMessage ( message ) { const { id : queryId , tips } = message ; const expectedStreamID = this . messageBus . outstandingQueries . get ( queryId ) ; if ( expectedStreamID ) { const newTip = tips . get ( expectedStreamID . toString ( ) ) ; if ( ! newTip ) { throw new Error ( "Response to query with ID '" + queryId + "' is missing expected new tip for StreamID '" + expectedStreamID + "'" ) ; } this . repository . stateManager . update ( expectedStreamID , newTip ) ; this . messageBus . outstandingQueries . delete ( queryId ) ; } } /**
     * Gracefully closes the Dispatcher.
     */ async close ( ) { this . messageBus . unsubscribe ( ) ; } /**
     * Publish a message to IPFS pubsub as a fire-and-forget operation.
     *
     * You could use returned Subscription to react when the operation is finished.
     * Feel free to disregard it though.
     */ private publish ( message ) { return this . messageBus . next ( message ) ; } }