import { RadixUniverse , RadixAtom , RadixAtomStore , RadixAtomNodeStatus , RadixAtomNodeStatusUpdate , logger , RadixAtomObservation } from '../..' ; import { RadixAtomUpdate , RadixAddress , RadixSerializer , RadixAtomEvent } from '../atommodel' ; import { Subject , Observable , merge , BehaviorSubject , combineLatest , Subscription } from 'rxjs' ; import RadixNodeConnection , { AtomReceivedNotification } from '../universe/RadixNodeConnection' ; import { takeWhile , multicast , publish , tap } from 'rxjs/operators' ; import { TSMap } from 'typescript-map' ; import promiseRetry from 'promise-retry' ; export class RadixLedger { private networkAccountSubscriptions : { [ address ] : Observable < AtomReceivedNotification > ; } = { } ; private networkSyncedSubjects : { [ address ] : BehaviorSubject < boolean > ; } = { } ; private syncSubjects = new TSMap < Observable < RadixAtomObservation > , Observable < boolean > > ( ) ; private finalityTimeouts : { [ aid ] : NodeJS . Timeout ; } = { } ; private subs = new Subscription ( ) ; constructor ( readonly universe , readonly atomStore , readonly finalityTime ) { this . subs . add ( this . atomStore . getAtomObservations ( ) . subscribe ( this . monitorAtomFinality ) ) ; } param { RadixAddress } address returns public getAtomObservations ( address ) { this . setUpNetworkSubscription ( address ) . catch ( e => { logger . error ( e ) ; } ) ; const hasSyncedStore = new BehaviorSubject ( false ) ; const atomObservable = merge ( this . atomStore . getStoredAtomObservations ( address ) . pipe ( tap ( { complete : ( ) => { if ( ! hasSyncedStore . isStopped && ! hasSyncedStore . closed ) { hasSyncedStore . next ( true ) ; } else { logger . error ( `☢️ hasSyncedStore stopped/closed` ) ; } } , } ) ) , this . atomStore . getAtomObservations ( address ) ) ; this . syncSubjects . set ( atomObservable , combineLatest ( hasSyncedStore , this . getSyncedSubject ( address ) , ( val1 , val2 ) => { return val1 && val2 ; } ) ) ; return atomObservable ; } param { RadixAtom } atom param { RadixNodeConnection } node public submitAtom ( atom , node ) { this . atomStore . insert ( atom , { status : RadixAtomNodeStatus . PENDING , } ) . then ( ( ) => { this . subs . add ( node . submitAtom ( atom ) . subscribe ( { next : ( update ) => { this . atomStore . updateStatus ( atom . getAid ( ) , update ) ; } , error : ( e ) => { this . atomStore . updateStatus ( atom . getAid ( ) , { status : RadixAtomNodeStatus . SUBMISSION_ERROR , data : e , } ) ; } , } ) ) ; } ) ; return this . atomStore . getAtomStatusUpdates ( atom . getAid ( ) ) . pipe ( takeWhile ( ( update ) => { switch ( update . status ) { case RadixAtomNodeStatus . SUBMISSION_ERROR : case RadixAtomNodeStatus . MISSING_DEPENDENCY : case RadixAtomNodeStatus . EVICTED_FAILED_CM_VERIFICATION : case RadixAtomNodeStatus . EVICTED_CONFLICT_LOSER_FINAL : throw update ; case RadixAtomNodeStatus . STORED_FINAL : return false ; default : return true ; } } , true ) ) ; } param { Observable < RadixAtomObservation > } observable returns public onSynced ( observable ) { if ( ! this . syncSubjects . has ( observable ) ) { throw new Error ( `Subscription doesn't exist` ) ; } return this . syncSubjects . get ( observable ) ; } private getSyncedSubject ( address , defaultValue = false ) { const strAddress = address . toString ( ) ; if ( ! ( strAddress in this . networkSyncedSubjects ) ) { this . networkSyncedSubjects [ strAddress ] = new BehaviorSubject ( defaultValue ) ; } return this . networkSyncedSubjects [ strAddress ] ; } private onAtomUpdateReceived = ( notification , address ) => { const deserializedAtomEvents = RadixSerializer . fromJSON ( notification . atomEvents ) as RadixAtomEvent [ ] ; for ( const event of deserializedAtomEvents ) { if ( event . type . toUpperCase ( ) === 'STORE' ) { this . atomStore . insert ( event . atom , { status : RadixAtomNodeStatus . STORED , } ) ; } else if ( event . type . toUpperCase ( ) === 'DELETE' ) { this . atomStore . insert ( event . atom , { status : RadixAtomNodeStatus . CONFLICT_LOSER , } ) ; } else { logger . error ( `Unsupported atom event type received "${ event . type . toUpperCase ( ) }"` ) ; } } const syncer = this . getSyncedSubject ( address , notification . isHead ) ; if ( ! syncer . closed && ! syncer . isStopped ) { syncer . next ( notification . isHead ) ; } else { logger . error ( `☢️ syncer closed or stopped` ) ; } } ; private async setUpNetworkSubscription ( address ) { if ( address . toString ( ) in this . networkAccountSubscriptions ) { return this . networkAccountSubscriptions [ address . toString ( ) ] ; } const connection = await promiseRetry ( async ( retry , attempt ) => { try { return await this . universe . getNodeConnection ( ) ; } catch ( e ) { logger . error ( e ) ; logger . info ( 'retrying...' ) ; retry ( e ) ; } } , { retries : 1000 , maxtimeout : 60000 , } ) ; const subscription = connection . subscribe ( address . toString ( ) ) ; this . networkAccountSubscriptions [ address . toString ( ) ] = subscription ; subscription . subscribe ( { next : ( update ) => { this . onAtomUpdateReceived ( update , address ) ; } , error : ( ) => { setTimeout ( ( ) => { const syncedSubject = this . getSyncedSubject ( address , false ) ; if ( ! syncedSubject . closed && ! syncedSubject . isStopped ) { syncedSubject . next ( false ) ; } else { logger . error ( `☢️ syncedSubject closed or stopped` ) ; } this . setUpNetworkSubscription ( address ) ; } , 1000 ) ; } , } ) ; return subscription ; } private monitorAtomFinality = async ( observation ) => { const aid = observation . atom . getAid ( ) ; const aidString = aid . toString ( ) ; if ( aidString in this . finalityTimeouts ) { clearTimeout ( this . finalityTimeouts [ aidString ] ) ; } if ( observation . status . status === RadixAtomNodeStatus . STORED ) { this . finalityTimeouts [ aidString ] = setTimeout ( async ( ) => { await this . atomStore . updateStatus ( aid , { status : RadixAtomNodeStatus . STORED_FINAL } ) ; delete this . finalityTimeouts [ aidString ] ; } , this . finalityTime ) ; } else if ( observation . status . status === RadixAtomNodeStatus . CONFLICT_LOSER ) { this . finalityTimeouts [ aidString ] = setTimeout ( async ( ) => { await this . atomStore . updateStatus ( aid , { status : RadixAtomNodeStatus . EVICTED_CONFLICT_LOSER_FINAL } ) ; delete this . finalityTimeouts [ aidString ] ; } , this . finalityTime ) ; } } ; }