import { RadixAtomStore , RadixAtom , RadixAtomNodeStatusUpdate , RadixAID , RadixAddress , RadixAtomObservation , logger , } from '../..' ; import { Observable , Subject } from 'rxjs' ; import Datastore from 'nedb' ; import { RadixSerializer } from '../atommodel' ; import { filter , share } from 'rxjs/operators' ; export interface RadixAtomStoreEntry { _id : string ; atom : any ; status : RadixAtomNodeStatusUpdate ; addresses : string [ ] ; createdAt ? : number ; updatedAt ? : number ; } export class RadixNEDBAtomStore implements RadixAtomStore { private db : Datastore ; private atomObservationSubject = new Subject < RadixAtomObservation > ( ) ; param options . filename param options . inMemoryOnly constructor ( options ) { if ( ! ( options . filename || options . inMemoryOnly ) ) { throw new Error ( 'Must set either the filename or inMemoryOnly options' ) ; } ( options as any ) . autoload = true ; ( options as any ) . timestampData = true ; this . db = new Datastore ( options ) ; } /**
     * Creates a non-perisisted atom store
     */ public static createInMemoryStore ( ) { return new this ( { inMemoryOnly : true } ) ; } param filename public static createPersistedStore ( filename ) { return new this ( { filename } ) ; } public insert ( atom , status ) { const dbEntry = { _id : atom . getAidString ( ) , atom : atom . toJSON ( ) , status , addresses : atom . getAddresses ( ) . map ( address => address . toString ( ) ) , } ; return this . dbInsert ( dbEntry ) . then ( ( newDoc ) => { if ( ! this . atomObservationSubject . closed && ! this . atomObservationSubject . isStopped ) { this . atomObservationSubject . next ( { atom , status , timestamp : newDoc . updatedAt , } ) ; } else { logger . error ( `☢️ this.atomObservationSubject closed or stopped` ) ; } return true ; } ) . catch ( error => { return false ; } ) ; } public updateStatus ( aid , status ) { return this . dbFindOne ( aid . toString ( ) ) . then ( atomEntry => { if ( atomEntry . status . status === status . status ) { throw new Error ( 'No change' ) ; } this . dbUpdate ( aid . toString ( ) , status ) . then ( ( ) => { if ( ! this . atomObservationSubject . closed && ! this . atomObservationSubject . isStopped ) { this . atomObservationSubject . next ( { atom : RadixSerializer . fromJSON ( atomEntry . atom ) , status , timestamp : atomEntry . updatedAt , } ) ; } else { logger . error ( `☢️ this.atomObservationSubject closed or stopped` ) ; } } ) ; return true ; } ) . catch ( error => { return false ; } ) ; } public getAtom ( aid ) { return this . dbFindOne ( aid . toString ( ) ) . then ( atomEntry => { return { atom : RadixSerializer . fromJSON ( atomEntry . atom ) , status : atomEntry . status , timestamp : atomEntry . updatedAt , } ; } ) ; } public getAtomObservations ( address ? ) { return this . atomObservationSubject . pipe ( filter ( atomObservation => { if ( address ) { for ( const atomAddress of atomObservation . atom . getAddresses ( ) ) { if ( address . equals ( atomAddress ) ) { return true ; } } return false ; } else { return true ; } } ) , share ( ) ) ; } public getAtomStatusUpdates ( aid ) { return this . atomObservationSubject . pipe ( filter ( ( atomObservation ) => { return atomObservation . atom . getAid ( ) . equals ( aid ) ; } ) , share ( ) ) . map ( atomObservation => { return atomObservation . status ; } ) ; } public getStoredAtomObservations ( address ? ) { const query = { } ; if ( address ) { ( query as any ) . addresses = address . toString ( ) ; } return Observable . create ( observer => { this . dbFind ( query ) . then ( async ( entires ) => { for ( const entry of entires ) { const deserializedAtom = await this . asyncDeserialize ( entry . atom ) ; if ( ! observer . closed ) { observer . next ( { atom : deserializedAtom , status : entry . status , timestamp : entry . updatedAt , } ) ; } else { logger . error ( `☢️ observer closed or stopped` ) ; } } observer . complete ( ) ; } ) ; } ) ; } public reset ( ) { return new Promise < boolean > ( ( resolve , reject ) => { this . db . remove ( { } , { multi : true } , ( error , numRemoved ) => { if ( error ) { reject ( error ) ; } else { resolve ( true ) ; } } ) ; } ) ; } private asyncDeserialize ( atom ) { return new Promise < RadixAtom > ( ( resolve , reject ) => { setTimeout ( ( ) => { resolve ( RadixSerializer . fromJSON ( atom ) ) ; } , 0 ) ; } ) ; } private dbFindOne = ( _id ) : Promise < RadixAtomStoreEntry > => { return new Promise ( ( resolve , reject ) => { this . db . findOne < RadixAtomStoreEntry > ( { _id } , ( error , doc ) => { if ( error ) { reject ( error ) ; } else { resolve ( doc ) ; } } ) ; } ) ; } ; private dbNotExists = ( _id ) : Promise < boolean > => { return new Promise ( ( resolve , reject ) => { this . db . findOne ( { _id } , ( error , doc ) => { if ( error ) { reject ( error ) ; } else if ( ! doc ) { resolve ( true ) ; } reject ( 'Atom already in db' ) ; } ) ; } ) ; } ; private dbFind = ( opt ) : Promise < RadixAtomStoreEntry [ ] > => { return new Promise ( ( resolve , reject ) => { this . db . find < RadixAtomStoreEntry > ( opt , ( error , doc ) => { if ( error ) { reject ( error ) ; } else { resolve ( doc ) ; } } ) ; } ) ; } ; private dbInsert = ( entry ) : Promise < RadixAtomStoreEntry > => { return new Promise ( ( resolve , reject ) => { this . db . insert ( entry , ( error , doc ) => { if ( error ) { reject ( error ) ; } else { resolve ( doc ) ; } } ) ; } ) ; } ; private dbUpdate = ( _id , status ) : Promise < boolean > => { return new Promise ( ( resolve , reject ) => { const update = { $set : { status } } ; this . db . update ( { _id } , update , { } , ( error , numReplaced ) => { if ( error || numReplaced === 0 ) { reject ( false ) ; } else { resolve ( true ) ; } } ) ; } ) ; } ; private dbRemove = ( opt ) : Promise < number > => { return new Promise ( ( resolve , reject ) => { this . db . remove ( opt , ( error , doc ) => { if ( error ) { reject ( error ) ; } else { resolve ( doc ) ; } } ) ; } ) ; } ; }