import CID from 'cids' ; import { Observable , Subject , concat , of } from 'rxjs' ; import { filter } from 'rxjs/operators' ; import { AnchorProof , AnchorService , AnchorStatus , StreamUtils , AnchorServiceResponse , AnchorValidator , AnchorCommit , } from '@ceramicnetwork/common' ; import type { Dispatcher } from '../../dispatcher' ; import Ceramic from '../../ceramic' ; import StreamID from '@ceramicnetwork/streamid' ; import { DiagnosticsLogger } from '@ceramicnetwork/common' ; import type { DagJWS } from 'dids' ; import Utils from '../../utils' ; import dagCBOR from 'ipld-dag-cbor' ; const DID_MATCHER = '^(did:([a-zA-Z0-9_]+):([a-zA-Z0-9_.-]+(:[a-zA-Z0-9_.-]+)*)((;[a-zA-Z0-9_.:%-]+=[a-zA-Z0-9_.:%-]*)*)(/[^#?]*)?)([?][^#]*)?(#.*)?' ; const CHAIN_ID = 'inmemory:12345' ; class Candidate { constructor ( readonly cid , readonly streamId ? , readonly log ? ) { } get key ( ) : string { return this . streamId . toString ( ) ; } } interface InMemoryAnchorConfig { anchorDelay ? : number ; anchorOnRequest ? : boolean ; verifySignatures ? : boolean ; } const SAMPLE_ETH_TX_HASH = 'bagjqcgzaday6dzalvmy5ady2m5a5legq5zrbsnlxfc2bfxej532ds7htpova' ; /**
 * In-memory anchor service - used locally, not meant to be used in production code
 */ class InMemoryAnchorService implements AnchorService , AnchorValidator { #ceramic : Ceramic ; #dispatcher : Dispatcher ; #logger : DiagnosticsLogger ; readonly #anchorDelay : number ; readonly #anchorOnRequest : boolean ; readonly #verifySignatures : boolean ; readonly #feed : Subject < AnchorServiceResponse > = new Subject ( ) ; readonly #anchors : Map < string , AnchorServiceResponse > = new Map ( ) ; #queue : Candidate [ ] = [ ] ; constructor ( _config ) { this . #anchorDelay = _config ?. anchorDelay ?? 0 ; this . #anchorOnRequest = _config ?. anchorOnRequest ?? true ; this . #verifySignatures = _config ?. verifySignatures ?? true ; this . #feed . subscribe ( ( asr ) => this . #anchors . set ( asr . cid . toString ( ) , asr ) ) ; } async init ( ) { return ; } returns async getSupportedChains ( ) { return [ CHAIN_ID ] ; } /**
     * Anchor requests
     */ async anchor ( ) { const candidates = await this . _findCandidates ( ) ; for ( const candidate of candidates ) { await this . _process ( candidate ) ; } this . #queue = [ ] ; } /**
     * Fails all pending anchors. Useful for testing.
     */ async failPendingAnchors ( ) { const candidates = await this . _findCandidates ( ) ; for ( const candidate of candidates ) { this . _failCandidate ( candidate , 'anchor failed' ) ; } this . #queue = [ ] ; } private async _findCandidates ( ) { const groupedCandidates = await this . _groupCandidatesByStreamId ( this . #queue ) ; return this . _selectValidCandidates ( groupedCandidates ) ; } async _groupCandidatesByStreamId ( candidates ) { const result = { } ; await Promise . all ( candidates . map ( async ( req ) => { try { const commitData = await Utils . getCommitData ( this . #dispatcher , req . cid , null , req . streamId ) ; if ( this . #verifySignatures && StreamUtils . isSignedCommitData ( commitData ) ) { await this . verifySignedCommit ( commitData . envelope ) ; } const log = await this . _loadCommitHistory ( req . cid ) ; const candidate = new Candidate ( req . cid , req . streamId , log ) ; if ( ! result [ candidate . key ] ) { result [ candidate . key ] = [ ] ; } result [ candidate . key ] . push ( candidate ) ; } catch ( e ) { this . #logger . err ( e ) ; this . _failCandidate ( req , e . message ) ; } } ) ) ; return result ; } _selectValidCandidates ( groupedCandidates ) { const result = [ ] ; for ( const compositeKey of Object . keys ( groupedCandidates ) ) { const candidates = groupedCandidates [ compositeKey ] ; let selected = null ; for ( const c of candidates ) { if ( selected == null ) { selected = c ; continue ; } if ( c . log . length < selected . log . length ) { this . _failCandidate ( c ) ; } else if ( c . log . length > selected . log . length ) { this . _failCandidate ( selected ) ; selected = c ; } else { if ( c . cid . bytes < selected . cid . bytes ) { this . _failCandidate ( selected ) ; selected = c ; } else { this . _failCandidate ( c ) ; } } } result . push ( selected ) ; } return result ; } _failCandidate ( candidate , message ? ) { if ( ! message ) { message = `Rejecting request to anchor CID ${ candidate . cid . toString ( ) } for stream ${ candidate . streamId . toString ( ) } because there is a better CID to anchor for the same stream` ; } this . #feed . next ( { status : AnchorStatus . FAILED , streamId : candidate . streamId , cid : candidate . cid , message , } ) ; } param commitId private async _loadCommitHistory ( commitId ) { const history = [ ] ; let currentCommitId = commitId ; for ( ; ; ) { const commitData = await Utils . getCommitData ( this . #dispatcher , currentCommitId ) ; if ( StreamUtils . isAnchorCommitData ( commitData ) ) { return history ; } const prevCommitId = commitData . commit . prev ; if ( prevCommitId == null ) { return history ; } history . push ( prevCommitId ) ; currentCommitId = prevCommitId ; } } param ceramic set ceramic ( ceramic ) { this . #ceramic = ceramic ; this . #dispatcher = this . #ceramic . dispatcher ; this . #logger = this . #ceramic ?. context ?. loggerProvider . getDiagnosticsLogger ( ) ; } get url ( ) { return '<inmemory>' ; } param streamId param tip requestAnchor ( streamId , tip ) { const candidate = new Candidate ( tip , streamId ) ; if ( this . #anchorOnRequest ) { this . _process ( candidate ) . catch ( ( error ) => { this . #feed . next ( { status : AnchorStatus . FAILED , streamId : candidate . streamId , cid : candidate . cid , message : error . message , } ) ; } ) ; } else { this . #queue . push ( candidate ) ; } this . #feed . next ( { status : AnchorStatus . PENDING , streamId : streamId , cid : tip , message : 'Sending anchoring request' , anchorScheduledFor : null , } ) ; return this . pollForAnchorResponse ( streamId , tip ) ; } param streamId param tip pollForAnchorResponse ( streamId , tip ) { const anchorResponse = this . #anchors . get ( tip . toString ( ) ) ; const feed$ = this . #feed . pipe ( filter ( ( asr ) => asr . streamId . equals ( streamId ) && asr . cid . equals ( tip ) ) ) ; if ( anchorResponse ) { return concat ( of < AnchorServiceResponse > ( anchorResponse ) , feed$ ) ; } else { return of < AnchorServiceResponse > ( { status : AnchorStatus . FAILED , streamId , cid : tip , message : 'Request not found' , } ) ; } } param streamId param commit async _publishAnchorCommit ( streamId , commit ) { const expectedCID = await dagCBOR . util . cid ( new Uint8Array ( dagCBOR . util . serialize ( commit ) ) ) ; const stream = await this . #ceramic . applyCommit ( streamId , commit , { publish : true , anchor : false , } ) ; const commitFound = null != stream . state . log . find ( ( logEntry ) => logEntry . cid . equals ( expectedCID ) ) ; if ( ! commitFound ) { throw new Error ( `Anchor commit not found in stream log after being applied to Ceramic node. This most likely means the commit was rejected by Ceramic's conflict resolution. StreamID: ${ streamId . toString ( ) }, found tip: ${ stream . tip . toString ( ) }` ) ; } return expectedCID ; } private async _process ( leaf ) { const timestamp = Math . floor ( Date . now ( ) / 1000 ) ; const proofData = { chainId : CHAIN_ID , blockNumber : timestamp , blockTimestamp : timestamp , txHash : new CID ( SAMPLE_ETH_TX_HASH ) , root : leaf . cid , } ; const proof = await this . #dispatcher . storeCommit ( proofData ) ; const commit = { proof , path : '' , prev : leaf . cid , id : leaf . streamId . cid } ; const cid = await this . _publishAnchorCommit ( leaf . streamId , commit ) ; const handle = setTimeout ( ( ) => { this . #feed . next ( { status : AnchorStatus . ANCHORED , streamId : leaf . streamId , cid : leaf . cid , message : 'CID successfully anchored' , anchorCommit : cid , } ) ; clearTimeout ( handle ) ; } , this . #anchorDelay ) ; } param envelope return private async verifySignedCommit ( envelope ) { try { const { kid } = await this . #ceramic . did . verifyJWS ( envelope ) ; return kid . match ( RegExp ( DID_MATCHER ) ) [ 1 ] ; } catch ( e ) { throw new Error ( 'Invalid signature for signed commit. ' + e ) ; } } async validateChainInclusion ( proof ) { } } export default InMemoryAnchorService ;