import { ChangeDetectorRef , Injectable , NgZone , OnDestroy } from '@angular/core' ; import { ActivatedRoute } from '@angular/router' ; import { merge , Observable , Subject , Subscription } from 'rxjs' ; import { auditTime , filter , map } from 'rxjs/operators' ; import { ICollaborator , LocalOperation , MetaDataType , MuteCoreFactory , MuteCoreTypes , Position , StateStrategy , StateTypes , TextDelete , TextInsert , } from '@coast-team/mute-core' ; import { SessionParameters } from '@coast-team/mute-core/dist/types/src/MuteCore' ; import { KeyState } from '@coast-team/mute-crypto' ; import { WebGroupState } from 'netflux' ; import { environment } from '@environments/environment' ; import { CryptoService } from '../core/crypto' ; import { Doc } from '../core/Doc' ; import { EProperties } from '../core/settings/EProperties.enum' ; import { SettingsService } from '../core/settings' ; import { BotStorageService } from '../core/storage/bot' ; import { UiService } from '../core/ui' ; import { DocResolverService } from './doc-resolver.service' ; import { LogsService } from './logs' ; import { NetworkService } from './network' ; import { RichCollaboratorsService } from './rich-collaborators' ; const SAVE_DOC_INTERVAL = 2000 ; const SYNC_DOC_INTERVAL = 10000 ; @ Injectable ( ) export class DocService implements OnDestroy { private subs : Subscription [ ] ; private muteCore : MuteCoreTypes ; private saveDocInterval : number | undefined ; private syncDocContentInterval : number | undefined ; private docContentChanged : boolean ; private initSubject$ : Subject < string > ; public doc : Doc ; constructor ( private zone , private route , private collabs , private settings , private network , private botStorage , public ui , private cd , private logs , private crypto , private docResolver ) { this . subs = [ ] ; this . docContentChanged = false ; this . saveDocInterval = undefined ; this . initSubject$ = new Subject ( ) ; this . subs [ this . subs . length ] = this . settings . onChange . pipe ( filter ( ( props ) => props . includes ( EProperties . profile ) ) ) . subscribe ( ( ) => window . location . reload ( ) ) ; this . zone . runOutsideAngular ( ( ) => { this . newSub = this . route . data . subscribe ( ( { doc } ) => { this . doc = doc ; this . newSub = this . collabs . onUpdate . subscribe ( ( collab ) => { if ( collab . login === this . botStorage . login ) { if ( this . doc . remotes . length === 0 ) { this . doc . addRemote ( this . botStorage . id ) ; } this . doc . remotes [ 0 ] . synchronized = new Date ( ) ; } } ) ; this . newSub = this . network . onStateChange . subscribe ( ( state ) => { if ( state === WebGroupState . JOINED && this . doc . remotes . length !== 0 ) { this . network . inviteBot ( this . botStorage . wsURL ) ; } } ) ; this . newSub = this . doc . onMetadataChanges . subscribe ( ( ) => cd . detectChanges ( ) ) ; } ) ; } ) ; this . logs . setStreamLogsPulsar ( this . network . pulsarService ) ; } async joinSession ( ) { const docContent = await this . readDocContent ( ) ; this . route . paramMap . subscribe ( ( params ) => { this . doc . pulsar = this . doc . pulsar || params . get ( 'pulsar' ) === 'true' ? true : false ; } ) ; const muteCoreOptions = { strategy : environment . crdtStrategy , profile : { displayName : this . settings . profile . displayName , login : this . settings . profile . login , email : this . settings . profile . email , avatar : this . settings . profile . avatar , deviceID : this . settings . profile . deviceID , } , docContent , metaTitle : { title : this . doc . title , titleModified : this . doc . titleModified . getTime ( ) , } , metaFixData : { docCreated : this . doc . created . getTime ( ) , cryptoKey : this . doc . cryptoKey , } , metaLogs : { share : this . doc . shareLogs , vector : this . doc . shareLogsVector , } , metaPulsar : { activatePulsar : this . doc . pulsar , } , } ; this . muteCore = MuteCoreFactory . createMuteCore ( muteCoreOptions ) ; this . initLogs ( ) ; this . muteCore . localTextOperations$ = this . doc . localContentChanges . pipe ( map ( ( ops ) => { return ops . map ( ( { index , text , length } ) => { this . docContentChanged = true ; return length ? new TextDelete ( index , length , this . muteCore . myMuteCoreId ) : new TextInsert ( index , text , this . muteCore . myMuteCoreId ) ; } ) ; } ) ) ; this . newSub = this . muteCore . remoteTextOperations$ . subscribe ( ( { collaborator , operations } ) => { this . docContentChanged = true ; this . doc . remoteContentChanges . next ( operations . map ( ( op ) => { if ( op instanceof TextInsert ) { return { collaborator , index : op . index , text : op . content } ; } else if ( op instanceof TextDelete ) { return { collaborator , index : op . index , length : op . length } ; } } ) ) ; } ) ; this . muteCore . messageIn$ = this . network . messageOut ; this . network . setMessageIn ( this . muteCore . messageOut$ ) ; this . collabs . subscribeToUpdateSource ( this . muteCore . remoteCollabUpdate$ ) ; this . collabs . subscribeToJoinSource ( this . muteCore . collabJoin$ ) ; this . collabs . subscribeToLeaveSource ( this . muteCore . collabLeave$ ) ; this . muteCore . memberJoin$ = this . network . onMemberJoin ; this . muteCore . memberLeave$ = this . network . onMemberLeave ; this . muteCore . localCollabUpdate$ = this . settings . onChange . pipe ( filter ( ( props ) => props . includes ( EProperties . profile ) || props . includes ( EProperties . profileDisplayName ) ) , map ( ( props ) => { if ( props . includes ( EProperties . profile ) ) { return { id : this . network . myId , displayName : this . settings . profile . displayName , login : this . settings . profile . login , email : this . settings . profile . email , avatar : this . settings . profile . avatar , } ; } else { return { id : this . network . myId , displayName : this . settings . profile . displayName } ; } } ) ) ; this . muteCore . localMetadataUpdate$ = merge ( this . doc . onMetadataChanges . pipe ( filter ( ( { isLocal , changedProperties } ) => isLocal && changedProperties . includes ( Doc . TITLE ) ) , map ( ( ) => { return { type : MetaDataType . Title , data : { title : this . doc . title , titleModified : this . doc . titleModified . getTime ( ) } } ; } ) ) , this . doc . onMetadataChanges . pipe ( filter ( ( { isLocal , changedProperties } ) => isLocal && changedProperties . includes ( Doc . SHARE_LOGS ) ) , map ( ( ) => { return { type : MetaDataType . Logs , data : { share : this . doc . shareLogs , vector : this . doc . shareLogsVector } } ; } ) ) , this . doc . onMetadataChanges . pipe ( filter ( ( { isLocal , changedProperties } ) => isLocal && changedProperties . includes ( Doc . PULSAR ) ) , map ( ( ) => { return { type : MetaDataType . Pulsar , data : { activatePulsar : this . doc . pulsar } } ; } ) ) ) ; this . doc . setRemoteMetadataUpdateSource ( this . muteCore . remoteMetadataUpdate$ ) ; this . newSub = this . muteCore . digestUpdate$ . subscribe ( ( digest ) => { this . ui . updateDocDigest ( digest ) ; this . cd . detectChanges ( ) ; } ) ; this . startSaveDocInterval ( ) ; this . newSub = merge ( this . network . onCryptoStateChange . pipe ( filter ( ( state ) => state === KeyState . READY ) ) , this . network . onStateChange . pipe ( filter ( ( state ) => state === WebGroupState . JOINED ) ) ) . pipe ( auditTime ( 1000 ) ) . subscribe ( ( v ) => this . restartSyncInterval ( ) ) ; if ( environment . cryptography . coniksClient ) { this . collabs . onJoin . subscribe ( ( { id , login } ) => this . crypto . verifyLoginPKConiks ( id , login ) . catch ( ( err ) => { log . info ( 'Failed to retreive Public Key of ' + login ) ; } ) ) ; } else if ( environment . cryptography . keyserver ) { this . collabs . onJoin . subscribe ( ( { id , login , deviceID } ) => { return this . crypto . verifyLoginPK ( id , login , deviceID ) . catch ( ( err ) => { log . info ( 'Failed to retreive Public Key of ' + login ) ; } ) ; } ) ; } this . network . join ( this . doc . signalingKey ) ; } ngOnDestroy ( ) { this . subs . forEach ( ( s ) => s . unsubscribe ( ) ) ; window . clearInterval ( this . saveDocInterval ) ; window . clearInterval ( this . syncDocContentInterval ) ; this . doc . dispose ( ) ; } getDocContent ( ) { return this . muteCore . state ; } indexFromId ( pos ) { return this . muteCore . indexFromId ( pos ) ; } positionFromIndex ( index ) { return this . muteCore . positionFromIndex ( index ) ; } private initLogs ( ) { const siteId = this . muteCore . myMuteCoreId ; this . subs . push ( this . doc . onMetadataChanges . pipe ( filter ( ( { changedProperties } ) => changedProperties . includes ( Doc . SHARE_LOGS ) ) ) . subscribe ( ( ) => { this . logs . setShareLogs ( this . doc . shareLogs , this . muteCore . state . vector ) ; } ) ) ; if ( this . docResolver . isCreate ) { this . logs . log ( { type : 'creation' , timestamp : Date . now ( ) , siteId , documentId : this . doc . signalingKey } ) ; } else { this . logs . log ( { type : 'opening' , timestamp : Date . now ( ) , siteId , documentId : this . doc . signalingKey } ) ; } this . subs . push ( this . network . onStateChange . pipe ( filter ( ( state ) => state === WebGroupState . JOINED ) ) . subscribe ( ( ) => { const obj = { type : 'connection' , timestamp : Date . now ( ) , siteId , networkId : this . network . myId , neighbours : { downstream : this . network . wg . neighbors , upstream : this . network . wg . neighbors , } , } ; this . logs . log ( obj ) ; } ) ) ; this . subs . push ( this . network . onLeave . subscribe ( ( ) => { this . logs . log ( { type : 'disconnection' , timestamp : Date . now ( ) , siteId , networkId : this . network . myId , neighbours : { downstream : this . network . wg . neighbors , upstream : this . network . wg . neighbors , } , } ) ; } ) ) ; this . subs . push ( this . network . onMemberJoin . subscribe ( ( peer ) => { this . logs . log ( { type : 'peerConnection' , timestamp : Date . now ( ) , siteId , remoteNetworkId : peer } ) ; } ) ) ; this . subs . push ( this . network . onMemberLeave . subscribe ( ( peer ) => { this . logs . log ( { type : 'peerDisconnection' , timestamp : Date . now ( ) , siteId , remoteNetworkId : peer } ) ; } ) ) ; this . subs . push ( this . muteCore . collabJoin$ . subscribe ( ( c ) => { this . logs . log ( { type : 'collaboratorJoin' , timestamp : Date . now ( ) , siteId , remoteSiteId : c . muteCoreId , remoteNetworkId : c . id } ) ; } ) ) ; this . subs . push ( this . muteCore . collabLeave$ . subscribe ( ( c ) => { this . logs . log ( { type : 'collaboratorLeave' , timestamp : Date . now ( ) , siteId , remoteSiteId : c . muteCoreId , remoteNetworkId : c . id } ) ; } ) ) ; this . subs . push ( ( this . muteCore . localOperationForLog$ as Observable < LocalOperation < any > > ) . subscribe ( { next : ( operation ) => { this . logs . log ( { ... operation , timestamp : Date . now ( ) , collaborators : this . network . members , neighbours : { downstream : this . network . wg . neighbors , upstream : this . network . wg . neighbors , } , } ) ; } , } ) ) ; } private restartSyncInterval ( ) { window . clearInterval ( this . syncDocContentInterval ) ; this . sync ( ) ; this . syncDocContentInterval = window . setInterval ( ( ) => this . sync ( ) , SYNC_DOC_INTERVAL ) ; } private sync ( ) { if ( this . network . cryptoState === KeyState . READY ) { this . muteCore . synchronize ( ) ; } } private async readDocContent ( ) { try { const state = await this . doc . fetchContent ( ) ; if ( ! ( state instanceof Blob ) ) { const str = StateStrategy . getStr ( state ) ; if ( str !== undefined ) { this . initSubject$ . next ( str ) ; return state ; } } } catch { return StateStrategy . emptyState ( environment . crdtStrategy ) ; } } private startSaveDocInterval ( ) { this . saveDocInterval = window . setInterval ( ( ) => { if ( this . docContentChanged ) { this . doc . saveContent ( this . muteCore . state ) . catch ( ( err ) => log . warn ( 'Failed save document content to database: ' , err ) ) ; } else { this . docContentChanged = false ; } } , SAVE_DOC_INTERVAL ) ; } private set newSub ( s ) { this . subs [ this . subs . length ] = s ; } public get initSubject ( ) : Observable < string > { return this . initSubject$ . asObservable ( ) ; } }