import type CID from 'cids' ; import { AnchorProof , AnchorStatus , AnchorValidator , CommitData , CommitType , Context , InternalOpts , LogEntry , Stream , StreamHandler , StreamState , StreamStateHolder , StreamUtils , } from '@ceramicnetwork/common' ; import { Dispatcher } from './dispatcher' ; import cloneDeep from 'lodash.clonedeep' ; import { CommitID } from '@ceramicnetwork/streamid' ; import { StateValidation } from './state-management/state-validation' ; import { HandlersMap } from './handlers-map' ; import Utils from './utils' ; param dispatcher param anchorValidator param commitData async function verifyAnchorCommit ( dispatcher , anchorValidator , commitData ) { const proof = commitData . proof ; const commitPath = commitData . commit . path ; let prevCIDViaMerkleTree ; try { if ( commitPath . length === 0 ) { prevCIDViaMerkleTree = proof . root ; } else { const merkleTreeParentCommitPath = '/root/' + commitPath . substr ( 0 , commitPath . lastIndexOf ( '/' ) ) ; const last = commitPath . substr ( commitPath . lastIndexOf ( '/' ) + 1 ) ; const merkleTreeParentCommit = await dispatcher . retrieveFromIPFS ( commitData . commit . proof , merkleTreeParentCommitPath ) ; prevCIDViaMerkleTree = merkleTreeParentCommit [ last ] ; } } catch ( e ) { throw new Error ( `The anchor commit couldn't be verified. Reason ${ e . message }` ) ; } if ( commitData . commit . prev . toString ( ) !== prevCIDViaMerkleTree . toString ( ) ) { throw new Error ( `The anchor commit proof ${ proof . toString ( ) } with path ${ commitPath } points to invalid 'prev' commit` ) ; } await anchorValidator . validateChainInclusion ( proof ) ; return proof ; } param state1 param state2 returns export async function pickLogToAccept ( state1 , state2 ) { const isState1Anchored = state1 . anchorStatus === AnchorStatus . ANCHORED ; const isState2Anchored = state2 . anchorStatus === AnchorStatus . ANCHORED ; if ( isState1Anchored != isState2Anchored ) { return isState1Anchored ? state1 : state2 ; } if ( isState1Anchored && isState2Anchored ) { const { anchorProof : proof1 } = state1 ; const { anchorProof : proof2 } = state2 ; if ( proof1 . chainId != proof2 . chainId ) { throw new Error ( 'Conflicting logs on the same stream are anchored on different chains. Chain1: ' + proof1 . chainId + ', chain2: ' + proof2 . chainId ) ; } if ( proof1 . blockNumber < proof2 . blockNumber ) { return state1 ; } else if ( proof2 . blockNumber < proof1 . blockNumber ) { return state2 ; } } if ( state1 . log . length > state2 . log . length ) { return state1 ; } else if ( state1 . log . length < state2 . log . length ) { return state2 ; } return state1 . log [ state1 . log . length - 1 ] . cid . bytes < state2 . log [ state2 . log . length - 1 ] . cid . bytes ? state1 : state2 ; } export class HistoryLog { static fromState ( dispatcher , state ) { return new HistoryLog ( dispatcher , state . log ) ; } constructor ( private readonly dispatcher , readonly items ) { } get length ( ) : number { return this . items . length ; } /**
     * Determines if the HistoryLog includes a CID, returning true or false as appropriate.
     */ includes ( cid ) { const index = this . findIndex ( cid ) ; return index !== - 1 ; } get last ( ) : CID { return this . items [ this . items . length - 1 ] . cid ; } param cid findIndex ( cid ) { return this . items . findIndex ( ( entry ) => entry . cid . equals ( cid ) ) ; } slice ( start ? , end ? ) { const next = this . items . slice ( start , end ) ; return new HistoryLog ( this . dispatcher , next ) ; } async toCommitData ( ) { return await Promise . all ( this . items . map ( async ( logEntry ) => await Utils . getCommitData ( this . dispatcher , logEntry . cid , logEntry . timestamp ) ) ) ; } } param dispatcher param cid param stateLog param unappliedCommits param timestamp private export async function fetchLog ( dispatcher , cid , stateLog , unappliedCommits = [ ] , timestamp ? ) { if ( stateLog . includes ( cid ) ) { return [ ] ; } const nextCommitData = await Utils . getCommitData ( dispatcher , cid , timestamp ) ; timestamp = nextCommitData . timestamp ; const prevCid = nextCommitData . commit . prev ; if ( ! prevCid ) { return [ ] ; } unappliedCommits . push ( nextCommitData ) ; if ( stateLog . includes ( prevCid ) ) { return unappliedCommits . reverse ( ) ; } return fetchLog ( dispatcher , prevCid , stateLog , unappliedCommits , timestamp ) ; } export function commitAtTime ( stateHolder , timestamp ) { let commitCid = stateHolder . state . log [ 0 ] . cid ; for ( const entry of stateHolder . state . log ) { if ( entry . type === CommitType . ANCHOR ) { if ( entry . timestamp <= timestamp ) { commitCid = entry . cid ; } else { break ; } } } return stateHolder . id . atCommit ( commitCid ) ; } export class ConflictResolution { constructor ( public anchorValidator , private readonly stateValidation , private readonly dispatcher , private readonly context , private readonly handlers ) { } param commitData param state param handler private private async applyCommitDataToState < T extends Stream > ( commitData , state , handler ) { if ( StreamUtils . isAnchorCommitData ( commitData ) ) { await verifyAnchorCommit ( this . dispatcher , this . anchorValidator , commitData ) ; return await handler . applyCommit ( commitData , this . context , state ) ; } else { const tmpState = await handler . applyCommit ( commitData , this . context , state ) ; const isGenesis = ! commitData . commit . prev ; const effectiveState = isGenesis ? tmpState : tmpState . next ; await this . stateValidation . validate ( effectiveState , effectiveState . content ) ; return tmpState ; } } /**
     * Applies the log to the stream and updates the state.
     * If an error is encountered while applying a commit, commit application stops and the state
     * that was built thus far is returned, unless 'opts.throwOnInvalidCommit' is true.
     */ private async applyLogToState < T extends Stream > ( handler , unappliedCommits , state , breakOnAnchor , opts ) { if ( state && state . log . length === 1 ) { const timestamp = unappliedCommits [ 0 ] . timestamp ; const genesis = await Utils . getCommitData ( this . dispatcher , state . log [ 0 ] . cid , timestamp ) ; await handler . applyCommit ( genesis , this . context ) ; } for ( const entry of unappliedCommits ) { try { state = await this . applyCommitDataToState ( entry , state , handler ) ; } catch ( err ) { const streamId = state ? StreamUtils . streamIdFromState ( state ) . toString ( ) : null ; this . context . loggerProvider . getDiagnosticsLogger ( ) . warn ( `Error while applying commit ${ entry . cid . toString ( ) } to stream ${ streamId }: ${ err }` ) ; if ( opts . throwOnInvalidCommit ) { throw err ; } else { return state ; } } if ( breakOnAnchor && AnchorStatus . ANCHORED === state . anchorStatus ) { return state ; } } return state ; } param initialState param initialStateLog param unappliedCommits param opts private async applyLog ( initialState , initialStateLog , unappliedCommits , opts ) { const handler = this . handlers . get ( initialState . type ) ; const tip = initialStateLog . last ; if ( unappliedCommits [ unappliedCommits . length - 1 ] . cid . equals ( tip ) ) { return null ; } const commitData = unappliedCommits [ 0 ] ; if ( commitData . commit . prev . equals ( tip ) ) { return this . applyLogToState ( handler , unappliedCommits , cloneDeep ( initialState ) , false , opts ) ; } const conflictIdx = initialStateLog . findIndex ( commitData . commit . prev ) + 1 ; const canonicalLog = await initialStateLog . toCommitData ( ) ; const localLog = canonicalLog . splice ( conflictIdx ) ; const state = await this . applyLogToState ( handler , canonicalLog , null , false , opts ) ; const localState = await this . applyLogToState ( handler , localLog , cloneDeep ( state ) , true , opts ) ; const remoteState = await this . applyLogToState ( handler , unappliedCommits , cloneDeep ( state ) , true , opts ) ; const selectedState = await pickLogToAccept ( localState , remoteState ) ; if ( selectedState === localState ) { return null ; } return this . applyLogToState ( handler , unappliedCommits , cloneDeep ( state ) , false , opts ) ; } param initialState param tip param opts async applyTip ( initialState , tip , opts ) { if ( initialState . log . find ( ( logEntry ) => logEntry . cid . equals ( tip ) ) ) { return null ; } const stateLog = HistoryLog . fromState ( this . dispatcher , initialState ) ; const log = await fetchLog ( this . dispatcher , tip , stateLog ) ; if ( log . length ) { return this . applyLog ( initialState , stateLog , log , opts ) ; } } /**
     * Verify signature of a lone genesis commit, using current time to check for revoked key.
     */ async verifyLoneGenesis ( state ) { const handler = this . handlers . get ( state . type ) ; const genesis = await Utils . getCommitData ( this . dispatcher , state . log [ 0 ] . cid , state . log [ 0 ] . timestamp ) ; await handler . applyCommit ( genesis , this . context ) ; } /**
     * Return state at `commitId` version.
     */ async snapshotAtCommit ( initialState , commitId ) { const opts = { throwOnInvalidCommit : true } ; const baseState = ( await this . applyTip ( initialState , commitId . commit , opts ) ) || initialState ; const baseStateLog = HistoryLog . fromState ( this . dispatcher , baseState ) ; const commitIndex = baseStateLog . findIndex ( commitId . commit ) ; if ( commitIndex < 0 ) { throw new Error ( `Requested commit CID ${ commitId . commit . toString ( ) } not found in the log for stream ${ commitId . baseID . toString ( ) }` ) ; } const resetLog = await baseStateLog . slice ( 0 , commitIndex + 1 ) . toCommitData ( ) ; const handler = this . handlers . get ( initialState . type ) ; return this . applyLogToState ( handler , resetLog , null , false , opts ) ; } }