import { Dispatcher } from './dispatcher' ; import StreamID , { CommitID , StreamRef } from '@ceramicnetwork/streamid' ; import { IpfsTopology } from '@ceramicnetwork/ipfs-topology' ; import { CreateOpts , Stream , StreamHandler , Context , DiagnosticsLogger , StreamUtils , LoadOpts , AnchorService , CeramicApi , CeramicCommit , IpfsApi , PinApi , MultiQuery , PinningBackendStatic , LoggerProvider , Networks , UpdateOpts , SyncOptions , AnchorValidator , AnchorStatus , } from '@ceramicnetwork/common' ; import { DID } from 'dids' ; import { PinStoreFactory } from './store/pin-store-factory' ; import { PathTrie , TrieNode , promiseTimeout } from './utils' ; import EthereumAnchorService from './anchor/ethereum/ethereum-anchor-service' ; import InMemoryAnchorService from './anchor/memory/in-memory-anchor-service' ; import { randomUint32 } from '@stablelib/random' ; import { LocalPinApi } from './local-pin-api' ; import { Repository } from './state-management/repository' ; import { HandlersMap } from './handlers-map' ; import { FauxStateValidation , RealStateValidation , StateValidation , } from './state-management/state-validation' ; import { streamFromState } from './state-management/stream-from-state' ; import { ConflictResolution } from './conflict-resolution' ; import EthereumAnchorValidator from './anchor/ethereum/ethereum-anchor-validator' ; const DEFAULT_CACHE_LIMIT = 500 ; const DEFAULT_QPS_LIMIT = 10 ; const TESTING = process . env . NODE_ENV == 'test' ; const TRAILING_SLASH = /\/$/ ; const DEFAULT_ANCHOR_SERVICE_URLS = { [ Networks . MAINNET ] : 'https://cas.3boxlabs.com' , [ Networks . ELP ] : 'https://cas.3boxlabs.com' , [ Networks . TESTNET_CLAY ] : 'https://cas-clay.3boxlabs.com' , [ Networks . DEV_UNSTABLE ] : 'https://cas-dev.3boxlabs.com' , [ Networks . LOCAL ] : 'http://localhost:8081' , } ; const DEFAULT_LOCAL_ETHEREUM_RPC = 'http://localhost:7545' ; const SUPPORTED_CHAINS_BY_NETWORK = { [ Networks . MAINNET ] : [ 'eip155:1' ] , [ Networks . ELP ] : [ 'eip155:1' ] , [ Networks . TESTNET_CLAY ] : [ 'eip155:3' , 'eip155:4' ] , [ Networks . DEV_UNSTABLE ] : [ 'eip155:3' , 'eip155:4' ] , [ Networks . LOCAL ] : [ 'eip155:1337' ] , [ Networks . INMEMORY ] : [ 'inmemory:12345' ] , } ; const DEFAULT_APPLY_COMMIT_OPTS = { anchor : true , publish : true , sync : SyncOptions . PREFER_CACHE } ; const DEFAULT_CREATE_FROM_GENESIS_OPTS = { anchor : true , publish : true , sync : SyncOptions . PREFER_CACHE , } ; const DEFAULT_LOAD_OPTS = { sync : SyncOptions . PREFER_CACHE } ; /**
 * Ceramic configuration
 */ export interface CeramicConfig { ethereumRpcUrl ? : string ; anchorServiceUrl ? : string ; stateStoreDirectory ? : string ; validateStreams ? : boolean ; ipfsPinningEndpoints ? : string [ ] ; pinningBackends ? : PinningBackendStatic [ ] ; loggerProvider ? : LoggerProvider ; gateway ? : boolean ; networkName ? : string ; pubsubTopic ? : string ; streamCacheLimit ? : number ; concurrentRequestsLimit ? : number ; useCentralizedPeerDiscovery ? : boolean ; syncOverride ? : SyncOptions ; [ index ] : any ; } /**
 * Modules that Ceramic uses internally.
 * Most users will not provide this directly but will let it be derived automatically from the
 * `CeramicConfig` via `Ceramic.create()`.
 */ export interface CeramicModules { anchorService : AnchorService | null ; anchorValidator : AnchorValidator ; dispatcher : Dispatcher ; ipfs : IpfsApi ; ipfsTopology : IpfsTopology ; loggerProvider : LoggerProvider ; pinStoreFactory : PinStoreFactory ; repository : Repository ; } /**
 * Parameters that control internal Ceramic behavior.
 * Most users will not provide this directly but will let it be derived automatically from the
 * `CeramicConfig` via `Ceramic.create()`.
 */ export interface CeramicParameters { gateway : boolean ; networkOptions : CeramicNetworkOptions ; validateStreams : boolean ; loadOptsOverride : LoadOpts ; } /**
 * Protocol options that are derived from the specified Ceramic network name (e.g. "mainnet", "testnet-clay", etc)
 */ interface CeramicNetworkOptions { name : string ; pubsubTopic : string ; } const DEFAULT_NETWORK = Networks . INMEMORY ; const normalizeStreamID = ( streamId ) : StreamID => { const streamRef = StreamRef . from ( streamId ) ; if ( StreamID . isInstance ( streamRef ) ) { return streamRef ; } else { throw new Error ( `Not StreamID: ${ streamRef }` ) ; } } ; const tryStreamId = ( id ) : StreamID | null => { try { return StreamID . fromString ( id ) ; } catch ( e ) { return null ; } } ; /**
 * ### Ceramic core implementation.<br/>
 *
 * To install this library:<br/>
 * `$ npm install --save @ceramicnetwork/core`
 */ export class Ceramic implements CeramicApi { public readonly context : Context ; public readonly dispatcher : Dispatcher ; public readonly loggerProvider : LoggerProvider ; public readonly pin : PinApi ; readonly repository : Repository ; readonly _streamHandlers : HandlersMap ; private readonly _anchorValidator : AnchorValidator ; private readonly _gateway : boolean ; private readonly _ipfsTopology : IpfsTopology ; private readonly _logger : DiagnosticsLogger ; private readonly _networkOptions : CeramicNetworkOptions ; private _supportedChains : Array < string > ; private readonly _validateStreams : boolean ; private readonly stateValidation : StateValidation ; private readonly _loadOptsOverride : LoadOpts ; constructor ( modules , params ) { this . _ipfsTopology = modules . ipfsTopology ; this . loggerProvider = modules . loggerProvider ; this . _logger = modules . loggerProvider . getDiagnosticsLogger ( ) ; this . repository = modules . repository ; this . dispatcher = modules . dispatcher ; this . pin = this . _buildPinApi ( ) ; this . _anchorValidator = modules . anchorValidator ; this . _gateway = params . gateway ; this . _networkOptions = params . networkOptions ; this . _validateStreams = params . validateStreams ; this . _loadOptsOverride = params . loadOptsOverride ; this . context = { api : this , anchorService : modules . anchorService , ipfs : modules . ipfs , loggerProvider : modules . loggerProvider , } ; if ( ! this . _gateway ) { this . context . anchorService . ceramic = this ; } this . _streamHandlers = new HandlersMap ( this . _logger ) ; this . stateValidation = this . _validateStreams ? new RealStateValidation ( this . loadStream . bind ( this ) ) : new FauxStateValidation ( ) ; const conflictResolution = new ConflictResolution ( modules . anchorValidator , this . stateValidation , this . dispatcher , this . context , this . _streamHandlers ) ; const pinStore = modules . pinStoreFactory . createPinStore ( ) ; this . repository . setDeps ( { dispatcher : this . dispatcher , pinStore : pinStore , context : this . context , handlers : this . _streamHandlers , anchorService : modules . anchorService , conflictResolution : conflictResolution , stateValidation : this . stateValidation , } ) ; } /**
     * Get IPFS instance
     */ get ipfs ( ) : IpfsApi { return this . context . ipfs ; } /**
     * Get DID
     */ get did ( ) : DID | undefined { return this . context . did ; } param did set did ( did ) { this . context . did = did ; } private _buildPinApi ( ) { return new LocalPinApi ( this . repository , this . _logger ) ; } private static _generateNetworkOptions ( config ) { const networkName = config . networkName || DEFAULT_NETWORK ; if ( config . pubsubTopic && networkName !== Networks . INMEMORY && networkName !== Networks . LOCAL ) { throw new Error ( "Specifying pub/sub topic is only supported for the 'inmemory' and 'local' networks" ) ; } let pubsubTopic ; switch ( networkName ) { case Networks . MAINNET : { pubsubTopic = '/ceramic/mainnet' ; break ; } case Networks . ELP : { pubsubTopic = '/ceramic/mainnet' ; break ; } case Networks . TESTNET_CLAY : { pubsubTopic = '/ceramic/testnet-clay' ; break ; } case Networks . DEV_UNSTABLE : { pubsubTopic = '/ceramic/dev-unstable' ; break ; } case Networks . LOCAL : { if ( config . pubsubTopic ) { pubsubTopic = config . pubsubTopic ; } else { const rand = randomUint32 ( ) ; pubsubTopic = '/ceramic/local-' + rand ; } break ; } case Networks . INMEMORY : { if ( config . pubsubTopic ) { pubsubTopic = config . pubsubTopic ; } else { const rand = randomUint32 ( ) ; pubsubTopic = '/ceramic/inmemory-' + rand ; } break ; } default : { throw new Error ( "Unrecognized Ceramic network name: '" + networkName + "'. Supported networks are: 'mainnet', 'testnet-clay', 'dev-unstable', 'local', 'inmemory'" ) ; } } if ( networkName == Networks . MAINNET ) { throw new Error ( 'Ceramic mainnet is not yet supported' ) ; } return { name : networkName , pubsubTopic } ; } private private async _loadSupportedChains ( ) { const networkName = this . _networkOptions . name ; const anchorService = this . context . anchorService ; const networkChains = SUPPORTED_CHAINS_BY_NETWORK [ networkName ] ; const anchorServiceChains = await anchorService . getSupportedChains ( ) ; const usableChains = networkChains . filter ( ( c ) => anchorServiceChains . includes ( c ) ) ; if ( usableChains . length === 0 ) { throw new Error ( "No usable chainId for anchoring was found.  The ceramic network '" + networkName + "' supports the chains: ['" + networkChains . join ( "', '" ) + "'], but the configured anchor service '" + anchorService . url + "' only supports the chains: ['" + anchorServiceChains . join ( "', '" ) + "']" ) ; } this . _supportedChains = usableChains ; } /**
     * Parses the given `CeramicConfig` and generates the appropriate `CeramicParameters` and
     * `CeramicModules` from it. This usually should not be called directly - most users will prefer
     * to call `Ceramic.create()` instead which calls this internally.
     */ static _processConfig ( ipfs , config ) { const loggerProvider = config . loggerProvider ?? new LoggerProvider ( ) ; const logger = loggerProvider . getDiagnosticsLogger ( ) ; const pubsubLogger = loggerProvider . makeServiceLogger ( 'pubsub' ) ; const networkOptions = Ceramic . _generateNetworkOptions ( config ) ; let anchorService = null ; if ( ! config . gateway ) { const anchorServiceUrl = config . anchorServiceUrl ?. replace ( TRAILING_SLASH , '' ) || DEFAULT_ANCHOR_SERVICE_URLS [ networkOptions . name ] ; if ( ( networkOptions . name == Networks . MAINNET || networkOptions . name == Networks . ELP ) && anchorServiceUrl !== 'https://cas-internal.3boxlabs.com' && anchorServiceUrl !== DEFAULT_ANCHOR_SERVICE_URLS [ networkOptions . name ] ) { throw new Error ( 'Cannot use custom anchor service on Ceramic mainnet' ) ; } anchorService = networkOptions . name != Networks . INMEMORY ? new EthereumAnchorService ( anchorServiceUrl , logger ) : new InMemoryAnchorService ( config as any ) ; } let ethereumRpcUrl = config . ethereumRpcUrl ; if ( ! ethereumRpcUrl && networkOptions . name == Networks . LOCAL ) { ethereumRpcUrl = DEFAULT_LOCAL_ETHEREUM_RPC ; } let anchorValidator ; if ( networkOptions . name == Networks . INMEMORY ) { anchorValidator = anchorService ; } else { anchorValidator = new EthereumAnchorValidator ( ethereumRpcUrl , logger ) ; } const pinStoreOptions = { networkName : networkOptions . name , stateStoreDirectory : config . stateStoreDirectory , pinningEndpoints : config . ipfsPinningEndpoints , pinningBackends : config . pinningBackends , } ; const loadOptsOverride = config . syncOverride ? { sync : config . syncOverride } : { } ; const streamCacheLimit = config . streamCacheLimit ?? DEFAULT_CACHE_LIMIT ; const concurrentRequestsLimit = config . concurrentRequestsLimit ?? streamCacheLimit ; const maxQueriesPerSecond = process . env . CERAMIC_PUBSUB_QPS_LIMIT ? parseInt ( process . env . CERAMIC_PUBSUB_QPS_LIMIT ) : DEFAULT_QPS_LIMIT ; const ipfsTopology = new IpfsTopology ( ipfs , networkOptions . name , logger ) ; const pinStoreFactory = new PinStoreFactory ( ipfs , pinStoreOptions ) ; const repository = new Repository ( streamCacheLimit , concurrentRequestsLimit , logger ) ; const dispatcher = new Dispatcher ( ipfs , networkOptions . pubsubTopic , repository , logger , pubsubLogger , maxQueriesPerSecond ) ; const params = { gateway : config . gateway , networkOptions , validateStreams : config . validateStreams ?? true , loadOptsOverride , } ; const modules = { anchorService , anchorValidator , dispatcher , ipfs , ipfsTopology , loggerProvider , pinStoreFactory , repository , } ; return [ modules , params ] ; } param ipfs param config static async create ( ipfs , config = { } ) { const [ modules , params ] = await Ceramic . _processConfig ( ipfs , config ) ; const ceramic = new Ceramic ( modules , params ) ; const doPeerDiscovery = config . useCentralizedPeerDiscovery ?? ! TESTING ; await ceramic . _init ( doPeerDiscovery ) ; return ceramic ; } param doPeerDiscovery async _init ( doPeerDiscovery ) { try { this . _logger . imp ( `Connecting to ceramic network '${ this . _networkOptions . name }' using pubsub topic '${ this . _networkOptions . pubsubTopic }'` ) ; if ( this . _gateway ) { this . _logger . warn ( `Starting in read-only gateway mode. All write operations will fail` ) ; } if ( doPeerDiscovery ) { await this . _ipfsTopology . start ( ) ; } if ( ! this . _gateway ) { await this . context . anchorService . init ( ) ; await this . _loadSupportedChains ( ) ; this . _logger . imp ( `Connected to anchor service '${ this . context . anchorService . url }' with supported anchor chains ['${ this . _supportedChains . join ( "','" ) }']` ) ; } await this . _anchorValidator . init ( this . _supportedChains ? this . _supportedChains [ 0 ] : null ) ; } catch ( err ) { await this . close ( ) ; throw err ; } } deprecated async setDID ( did ) { this . context . did = did ; } param streamHandler addStreamHandler < T extends Stream > ( streamHandler ) { this . _streamHandlers . add ( streamHandler ) ; } param streamId param commit param opts async applyCommit < T extends Stream > ( streamId , commit , opts = { } ) { if ( this . _gateway ) { throw new Error ( 'Writes to streams are not supported in gateway mode' ) ; } opts = { ... DEFAULT_APPLY_COMMIT_OPTS , ... opts , ... this . _loadOptsOverride } ; const state$ = await this . repository . applyCommit ( normalizeStreamID ( streamId ) , commit , opts as CreateOpts ) ; return streamFromState < T > ( this . context , this . _streamHandlers , state$ . value , this . repository . updates$ ) ; } param streamId param opts async requestAnchor ( streamId , opts = { } ) { opts = { ... DEFAULT_LOAD_OPTS , ... opts , ... this . _loadOptsOverride } ; const effectiveStreamId = normalizeStreamID ( streamId ) ; const state = await this . repository . load ( effectiveStreamId , opts ) ; await this . repository . stateManager . anchor ( state ) ; return state . state . anchorStatus ; } param type param genesis param opts async createStreamFromGenesis < T extends Stream > ( type , genesis , opts = { } ) { opts = { ... DEFAULT_CREATE_FROM_GENESIS_OPTS , ... opts , ... this . _loadOptsOverride } ; const genesisCid = await this . dispatcher . storeCommit ( genesis ) ; const streamId = new StreamID ( type , genesisCid ) ; const state$ = await this . repository . applyCreateOpts ( streamId , opts ) ; return streamFromState < T > ( this . context , this . _streamHandlers , state$ . value , this . repository . updates$ ) ; } param streamId param opts async loadStream < T extends Stream > ( streamId , opts = { } ) { opts = { ... DEFAULT_LOAD_OPTS , ... opts , ... this . _loadOptsOverride } ; const streamRef = StreamRef . from ( streamId ) ; if ( CommitID . isInstance ( streamRef ) ) { const snapshot$ = await this . repository . loadAtCommit ( streamRef , opts ) ; return streamFromState < T > ( this . context , this . _streamHandlers , snapshot$ . value ) ; } else if ( opts . atTime ) { const snapshot$ = await this . repository . loadAtTime ( streamRef , opts ) ; return streamFromState < T > ( this . context , this . _streamHandlers , snapshot$ . value ) ; } else { const base$ = await this . repository . load ( streamRef . baseID , opts ) ; return streamFromState < T > ( this . context , this . _streamHandlers , base$ . value , this . repository . updates$ ) ; } } param genesis param streamRef async _ensureGenesis ( genesis , streamRef ) { if ( StreamUtils . isSignedCommitContainer ( genesis ) || StreamUtils . isSignedCommit ( genesis ) ) { throw new Error ( 'Given genesis commit is not deterministic' ) ; } const genesisCID = await this . ipfs . dag . put ( genesis ) ; if ( ! streamRef . cid . equals ( genesisCID ) ) { throw new Error ( `Given StreamID CID ${ streamRef . cid . toString ( ) } does not match given genesis content` ) ; } } param query param timeout private async _loadLinkedStreams ( query , timeout ) { const id = StreamRef . from ( query . streamId ) ; const pathTrie = new PathTrie ( ) ; query . paths ?. forEach ( ( path ) => pathTrie . add ( path ) ) ; if ( query . genesis ) { await this . _ensureGenesis ( query . genesis , id ) ; } const index = { } ; const walkNext = async ( node , streamId ) => { let stream ; try { stream = await promiseTimeout ( this . loadStream ( streamId , { atTime : query . atTime } ) , timeout , `Timeout after ${ timeout }ms` ) ; } catch ( e ) { if ( CommitID . isInstance ( streamId ) ) { this . _logger . warn ( `Error loading stream ${ streamId . baseID . toString ( ) } at commit ${ streamId . commit . toString ( ) } at time ${ query . atTime } as part of a multiQuery request: ${ e . toString ( ) }` ) ; } else { this . _logger . warn ( `Error loading stream ${ streamId . toString ( ) } at time ${ query . atTime } as part of a multiQuery request: ${ e . toString ( ) }` ) ; } return Promise . resolve ( ) ; } const streamRef = query . atTime ? streamId . atCommit ( stream . tip ) : streamId ; index [ streamRef . toString ( ) ] = stream ; const promiseList = Object . keys ( node . children ) . map ( ( key ) => { const keyStreamId = stream . content [ key ] ? tryStreamId ( stream . content [ key ] ) : null ; if ( keyStreamId ) return walkNext ( node . children [ key ] , keyStreamId ) ; return Promise . resolve ( ) ; } ) ; await Promise . all ( promiseList ) ; } ; await walkNext ( pathTrie . root , id ) ; return index ; } param queries param timeout async multiQuery ( queries , timeout = 7000 ) { const queryResults = await Promise . all ( queries . map ( ( query ) => { try { return this . _loadLinkedStreams ( query , timeout ) ; } catch ( e ) { this . _logger . warn ( `Error during multiQuery: ${ e . toString ( ) }` ) ; return Promise . resolve ( { } ) ; } } ) ) ; const results = queryResults . reduce ( ( acc , res ) => ( { ... acc , ... res } ) , { } ) ; await Promise . all ( Object . values ( results ) . map ( ( stream ) => { if ( ! stream . isReadOnly ) { return stream . sync ( { sync : SyncOptions . NEVER_SYNC , syncTimeoutSeconds : 0 } ) ; } } ) ) ; return results ; } param streamId async loadStreamCommits ( streamId ) { const effectiveStreamId = normalizeStreamID ( streamId ) ; const stream = await this . loadStream ( effectiveStreamId ) ; const { state } = stream ; const results = await Promise . all ( state . log . map ( async ( { cid } ) => { const commit = await this . dispatcher . retrieveCommit ( cid ) ; return { cid : cid . toString ( ) , value : await StreamUtils . convertCommitToSignedCommitContainer ( commit , this . ipfs ) , } ; } ) ) ; this . _logger . verbose ( `Successfully loaded ${ results . length } commits for stream ${ streamId . toString ( ) }` ) ; return results ; } returns async getSupportedChains ( ) { return this . _supportedChains ; } /**
     * Close Ceramic instance gracefully
     */ async close ( ) { this . _logger . imp ( 'Closing Ceramic instance' ) ; await this . dispatcher . close ( ) ; await this . repository . close ( ) ; this . _ipfsTopology . stop ( ) ; this . _logger . imp ( 'Ceramic instance closed successfully' ) ; } } export default Ceramic ;