import { Subject , Observable , Observer , Subscription } from 'rxjs' ; import { TSMap } from 'typescript-map' ; import RadixMessageUpdate from './RadixMessageUpdate' ; import { RadixAccountSystem , RadixChat , RadixMessage , RadixSerializer , RadixAtomStatusIsInsert , RadixAtomObservation } from '../..' ; import { RadixAddress , RadixAtomUpdate , RadixAtom } from '../atommodel' ; import { RadixDecryptedData , RadixDecryptionState } from '../account/RadixDecryptionAccountSystem' ; import { logger } from '../common/RadixLogger' ; export default class RadixMessagingAccountSystem implements RadixAccountSystem { public name = 'RADIX-MESSAGING' ; public messageSubject : Subject < RadixMessageUpdate > = new Subject ( ) ; public chats : TSMap < string , RadixChat > = new TSMap ( ) ; public messages : TSMap < string , RadixMessage > = new TSMap ( ) ; private subs = new Subscription ( ) ; constructor ( readonly address ) { } public async processAtomUpdate ( atomUpdate ) { if ( ! ( 'decryptedData' in atomUpdate . processedData ) || atomUpdate . processedData . decryptedData . application !== 'message' || atomUpdate . processedData . decryptedData . decryptionState === RadixDecryptionState . CANNOT_DECRYPT ) { return ; } if ( RadixAtomStatusIsInsert [ atomUpdate . status . status ] ) { this . processStoreAtom ( atomUpdate ) ; } else { this . processDeleteAtom ( atomUpdate ) ; } } public unsubscribeSubscribers ( ) { this . subs . unsubscribe ( ) ; } public startNewChat ( to ) { const chatId = to . getAddress ( ) ; if ( this . chats . has ( chatId ) ) { return ; } const chatDescription = { address : to . getAddress ( ) , chat_id : chatId , title : chatId , last_message_timestamp : Date . now ( ) , messages : new TSMap ( ) , } ; this . chats . set ( chatId , chatDescription ) ; } private processStoreAtom ( atomUpdate ) { const atom = atomUpdate . atom ; const aid = atom . getAidString ( ) ; const signatures = atom . signatures ; const decryptedData = atomUpdate . processedData . decryptedData ; if ( this . messages . has ( aid ) ) { return ; } const from = decryptedData . from ; const to = decryptedData . to . find ( a => ! a . equals ( from ) ) ; if ( ! to ) { throw new Error ( 'A message needs to have at least one other recipient' ) ; } let address = null ; let isMine = false ; if ( from . equals ( this . address ) ) { address = to ; isMine = true ; } else { address = from ; } if ( address === null ) { throw new Error ( 'Error processing a radix-message atom: neither of addresses is owned by this account' ) ; } const chatId = address . toString ( ) ; const message = { aid , chat_id : chatId , to , from , content : decryptedData . data , timestamp : 0 , is_mine : isMine , encryptionState : decryptedData . decryptionState , } ; if ( ! this . chats . has ( chatId ) ) { const newChatDescription = { address : address . getAddress ( ) , chat_id : chatId , title : chatId , last_message_timestamp : message . timestamp , messages : new TSMap ( ) , } ; this . chats . set ( chatId , newChatDescription ) ; } const chatDescription = this . chats . get ( chatId ) ; if ( message . timestamp > chatDescription . last_message_timestamp ) { chatDescription . last_message_timestamp = message . timestamp ; } chatDescription . messages . set ( aid , message ) ; this . chats . delete ( chatId ) ; this . chats . set ( chatId , chatDescription ) ; this . messages . set ( aid , message ) ; const messageUpdate = { action : 'STORE' , aid , message , } ; if ( ! this . messageSubject . closed && ! this . messageSubject . isStopped ) { this . messageSubject . next ( messageUpdate ) ; } else { logger . error ( `☢️ messageSubject closed or stopped` ) ; } } private processDeleteAtom ( atomUpdate ) { const atom = atomUpdate . atom ; const aid = atom . getAidString ( ) ; if ( ! this . messages . has ( aid ) ) { return ; } const message = this . messages . get ( aid ) ; this . chats . get ( message . chat_id ) . messages . delete ( aid ) ; this . messages . delete ( aid ) ; const messageUpdate = { action : 'DELETE' , aid , message , } ; if ( ! this . messageSubject . closed && ! this . messageSubject . isStopped ) { this . messageSubject . next ( messageUpdate ) ; } else { logger . error ( `☢️ messageSubject closed or stopped` ) ; } } public getAllMessages ( ) { return Observable . create ( ( observer ) => { for ( const message of this . messages . values ( ) ) { const messageUpdate = { action : 'STORE' , aid : message . aid , message , } ; if ( ! observer . closed ) { observer . next ( messageUpdate ) ; } else { logger . error ( `☢️ observer closed` ) ; } } this . subs . add ( this . messageSubject . subscribe ( observer ) ) ; } ) ; } }