import { BehaviorSubject , Observable , Observer , Subject , Subscription } from 'rxjs' ; import { TSMap } from 'typescript-map' ; import RadixAccountSystem from './RadixAccountSystem' ; import RadixTransaction from './RadixTransaction' ; import RadixTransactionUpdate from './RadixTransactionUpdate' ; import { RadixAddress , RadixConsumable , RadixSpin , RadixTransferrableTokensParticle , RadixUniqueParticle } from '../atommodel' ; import { RadixDecryptionState } from './RadixDecryptionAccountSystem' ; import BN from 'bn.js' ; import { radixTokenManager } from '../token/RadixTokenManager' ; import Decimal from 'decimal.js' ; import { RadixTokenDefinition } from '../token/RadixTokenDefinition' ; import { logger , RadixAtomObservation , RadixAtomStatusIsInsert , radixUniverse } from '../..' ; import { share } from 'rxjs/operators' ; export default class RadixTransferAccountSystem implements RadixAccountSystem { public name = 'TRANSFER' ; public transactions : TSMap < string , RadixTransaction > = new TSMap ( ) ; public balance : { [ tokenId ] : BN ; } = { } ; public tokenUnitsBalance : { [ tokenId ] : Decimal ; } = { } ; public transactionSubject : Subject < RadixTransactionUpdate > = new Subject ( ) ; public balanceSubject : BehaviorSubject < { [ tokenId ] : BN ; } > ; private tokenUnitsBalanceSubject : BehaviorSubject < { [ tokenId ] : Decimal ; } > ; private unspentConsumables : TSMap < string , RadixConsumable > = new TSMap ( ) ; private spentConsumables : TSMap < string , RadixConsumable > = new TSMap ( ) ; private subs = new Subscription ( ) ; constructor ( readonly address ) { this . balance [ radixTokenManager . nativeToken . toString ( ) ] = new BN ( 0 ) ; this . balanceSubject = new BehaviorSubject ( this . balance ) ; this . tokenUnitsBalance [ radixTokenManager . nativeToken . toString ( ) ] = new Decimal ( 0 ) ; this . tokenUnitsBalanceSubject = new BehaviorSubject ( this . tokenUnitsBalance ) ; } public unsubscribeSubscribers ( ) { this . subs . unsubscribe ( ) ; } public async processAtomUpdate ( atomUpdate ) { const atom = atomUpdate . atom ; if ( ! atom . containsParticle ( RadixTransferrableTokensParticle ) ) { return ; } if ( RadixAtomStatusIsInsert [ atomUpdate . status . status ] ) { this . processStoreAtom ( atomUpdate ) ; } else { this . processDeleteAtom ( atomUpdate ) ; } } private async processStoreAtom ( atomUpdate ) { const atom = atomUpdate . atom ; if ( this . transactions . has ( atom . getAidString ( ) ) ) { return ; } const transactionUpdate = { action : 'STORE' , aid : atom . getAidString ( ) , transaction : { aid : atom . getAidString ( ) , balance : { } , tokenUnitsBalance : { } , fee : undefined , from : undefined , to : undefined , message : '' , unique : atom . getParticlesOfType ( RadixUniqueParticle ) . map ( p => p . getRRI ( ) . toString ( ) ) , } , } ; const transaction = transactionUpdate . transaction ; if ( atomUpdate . processedData . decryptedData && atomUpdate . processedData . decryptedData . decryptionState !== RadixDecryptionState . CANNOT_DECRYPT ) { transaction . message = atomUpdate . processedData . decryptedData . data ; } const consumables = atom . getSpunParticlesOfType ( RadixTransferrableTokensParticle ) ; for ( const consumable of consumables ) { const spin = consumable . spin ; const particle = consumable . particle as RadixConsumable ; if ( spin === RadixSpin . DOWN ) { if ( ! transactionUpdate . transaction . from ) { transactionUpdate . transaction . from = particle . getOwner ( ) ; } } else if ( spin === RadixSpin . UP ) { if ( ! transactionUpdate . transaction . to ) { transactionUpdate . transaction . to = particle . getOwner ( ) ; } } const tokenClassReference = particle . getTokenDefinitionReference ( ) ; const ownedByMe = particle . getOwner ( ) . equals ( this . address ) ; if ( ownedByMe ) { const quantity = new BN ( 0 ) ; const hid = particle . getHidString ( ) ; if ( spin === RadixSpin . DOWN ) { quantity . isub ( particle . getAmount ( ) ) ; this . unspentConsumables . delete ( hid ) ; this . spentConsumables . set ( hid , particle ) ; } else if ( spin === RadixSpin . UP ) { quantity . iadd ( particle . getAmount ( ) ) ; this . unspentConsumables . set ( hid , particle ) ; } if ( ! ( tokenClassReference . toString ( ) in transaction . balance ) ) { transaction . balance [ tokenClassReference . toString ( ) ] = new BN ( 0 ) ; } transaction . balance [ tokenClassReference . toString ( ) ] . iadd ( quantity ) ; } } if ( ! transactionUpdate . transaction . from ) { transactionUpdate . transaction . from = transactionUpdate . transaction . to ; } if ( Object . keys ( transaction . balance ) . length === 0 ) { return ; } for ( const tokenId in transaction . balance ) { if ( ! ( tokenId in this . balance ) || ! this . balance [ tokenId ] ) { this . balance [ tokenId ] = new BN ( 0 ) ; } this . balance [ tokenId ] . iadd ( transaction . balance [ tokenId ] ) ; transaction . tokenUnitsBalance [ tokenId ] = RadixTokenDefinition . fromSubunitsToDecimal ( transaction . balance [ tokenId ] ) ; if ( ! ( tokenId in this . tokenUnitsBalance ) || ! this . balance [ tokenId ] ) { this . tokenUnitsBalance [ tokenId ] = new Decimal ( 0 ) ; } this . tokenUnitsBalance [ tokenId ] = this . tokenUnitsBalance [ tokenId ] . add ( transaction . tokenUnitsBalance [ tokenId ] ) ; } const feeParticleGroup = atom . particleGroups [ atom . particleGroups . length - 1 ] ; const sumReduce = ( previousValue , currentValue ) : BN => previousValue . add ( currentValue ) ; let fee = new BN ( 0 ) ; const feeParticles = feeParticleGroup . getParticlesOfType ( RadixTransferrableTokensParticle , RadixSpin . DOWN ) . map ( p => p . amount . value ) ; if ( feeParticles . length > 0 ) { fee = feeParticles . reduce ( sumReduce ) ; } transaction . fee = RadixTokenDefinition . fromSubunitsToDecimal ( fee ) ; this . transactions . set ( transactionUpdate . aid , transaction ) ; if ( ! this . balanceSubject . isStopped && ! this . balanceSubject . closed ) { this . balanceSubject . next ( this . balance ) ; } else { logger . error ( `☢️ balanceSubject closed/stopped?` ) ; } if ( ! this . tokenUnitsBalanceSubject . isStopped && ! this . tokenUnitsBalanceSubject . closed ) { this . tokenUnitsBalanceSubject . next ( this . tokenUnitsBalance ) ; } else { logger . error ( `☢️ tokenUnitsBalanceSubject closed/stopped?` ) ; } if ( ! this . transactionSubject . isStopped && ! this . transactionSubject . closed ) { this . transactionSubject . next ( transactionUpdate ) ; } else { logger . error ( `☢️ transactionSubject closed/stopped?` ) ; } } private async processDeleteAtom ( atomUpdate ) { const atom = atomUpdate . atom ; if ( ! this . transactions . has ( atom . getAidString ( ) ) ) { return ; } const id = atom . getAidString ( ) ; const transaction = this . transactions . get ( id ) ; const transactionUpdate = { action : 'DELETE' , aid : id , transaction , } ; for ( const tokenId in transaction . balance ) { if ( ! ( tokenId in this . balance ) || ! this . balance [ tokenId ] ) { this . balance [ tokenId ] = new BN ( 0 ) ; } this . balance [ tokenId ] . isub ( transaction . balance [ tokenId ] ) ; transaction . tokenUnitsBalance [ tokenId ] = RadixTokenDefinition . fromSubunitsToDecimal ( transaction . balance [ tokenId ] ) ; if ( ! ( tokenId in this . tokenUnitsBalance ) || ! this . balance [ tokenId ] ) { this . tokenUnitsBalance [ tokenId ] = new Decimal ( 0 ) ; } this . tokenUnitsBalance [ tokenId ] = this . tokenUnitsBalance [ tokenId ] . sub ( transaction . tokenUnitsBalance [ tokenId ] ) ; } const consumables = atom . getSpunParticlesOfType ( RadixTransferrableTokensParticle ) ; for ( const consumable of consumables ) { const particle = consumable . particle as RadixConsumable ; const nonce = particle . nonce ; const hid = particle . getHidString ( ) ; if ( consumable . spin === RadixSpin . UP ) { if ( ! this . unspentConsumables . delete ( hid ) ) { logger . error ( `unspentConsumables did NOT contain particle with nonce ${ nonce }, but we expected it to, this bad...` ) ; } } else { if ( ! this . spentConsumables . delete ( hid ) ) { logger . error ( `spentConsumables did NOT contain particle with nonce ${ nonce }, but we expected it to, this bad...` ) ; } if ( this . unspentConsumables . has ( hid ) ) { logger . error ( `unspentConsumables contains particle with nonce ${ nonce }, this is a critical discrepancy error...` ) ; } this . unspentConsumables . set ( hid , particle ) ; } } this . transactions . delete ( transactionUpdate . aid ) ; if ( ! this . balanceSubject . isStopped && ! this . balanceSubject . closed ) { this . balanceSubject . next ( this . balance ) ; } else { logger . error ( `☢️ balanceSubject closed/stopped?` ) ; } if ( ! this . tokenUnitsBalanceSubject . isStopped && ! this . tokenUnitsBalanceSubject . closed ) { this . tokenUnitsBalanceSubject . next ( this . tokenUnitsBalance ) ; } else { logger . error ( `☢️ tokenUnitsBalanceSubject closed/stopped?` ) ; } if ( ! this . transactionSubject . isStopped && ! this . transactionSubject . closed ) { this . transactionSubject . next ( transactionUpdate ) ; } else { logger . error ( `☢️ transactionSubject closed/stopped?` ) ; } } public getAllTransactions ( ) { return Observable . create ( ( observer ) => { for ( const transaction of this . transactions . values ( ) ) { const transactionUpdate = { action : 'STORE' , aid : transaction . aid , transaction , } ; if ( ! observer . closed ) { observer . next ( transactionUpdate ) ; } else { logger . error ( `☢️ observer is closed` ) ; } } this . subs . add ( this . transactionSubject . subscribe ( observer ) ) ; } ) ; } public getUnspentConsumables ( ) { return this . unspentConsumables . values ( ) ; } public getTokenUnitsBalanceUpdates ( ) { return this . tokenUnitsBalanceSubject . pipe ( share ( ) ) ; } public snapshotOfNativeTokenBalance ( ) { return this . balance [ radixUniverse . nativeToken . toString ( ) ] ; } }