import isEmpty from "lodash/isEmpty" ; import findIndex from "lodash/findIndex" ; import range from "lodash/range" ; import { copy as cloneDeep } from "fastest-json-copy" ; import { AttachState } from "@fluidframework/container-definitions" ; import { ISequencedDocumentMessage , MessageType , FileMode , TreeEntry , ITreeEntry , } from "@fluidframework/protocol-definitions" ; import { IChannelAttributes , IFluidDataStoreRuntime , IChannelStorageService , IChannelFactory , } from "@fluidframework/datastore-definitions" ; import { bufferToString , assert } from "@fluidframework/common-utils" ; import { ISummaryTreeWithStats } from "@fluidframework/runtime-definitions" ; import { IFluidSerializer , SharedObject } from "@fluidframework/shared-object-base" ; import { SummaryTreeBuilder } from "@fluidframework/runtime-utils" ; import { ChangeSet , Utils as ChangeSetUtils , rebaseToRemoteChanges , } from "@fluid-experimental/property-changeset" ; import { PropertyFactory , BaseProperty , NodeProperty } from "@fluid-experimental/property-properties" ; import { v4 as uuidv4 } from "uuid" ; import axios from "axios" ; import { PropertyTreeFactory } from "./propertyTreeFactory" ; export type SerializedChangeSet = any ; export type Metadata = any ; type FetchUnrebasedChangeFn = ( guid : string ) => IRemotePropertyTreeMessage ; type FetchRebasedChangesFn = ( startGuid : string , endGuid ? : string ) => IPropertyTreeMessage [ ] ; export const enum OpKind { ChangeSet = 0 } export interface IPropertyTreeMessage { op : OpKind ; changeSet : SerializedChangeSet ; metadata : Metadata ; guid : string ; referenceGuid : string ; remoteHeadGuid : string ; localBranchStart : string | undefined ; rebaseMetaInformation ? : Map < any , any > ; useMH ? : boolean ; } export interface IRemotePropertyTreeMessage extends IPropertyTreeMessage { sequenceNumber : number ; } interface ISnapshot { branchGuid : string ; summaryMinimumSequenceNumber : number ; useMH : boolean ; numChunks : number ; } interface ISnapshotSummary { remoteTipView ? : SerializedChangeSet ; remoteChanges ? : IPropertyTreeMessage [ ] ; unrebasedRemoteChanges ? : Record < string , IRemotePropertyTreeMessage > ; } export interface SharedPropertyTreeOptions { paths ? : string [ ] ; clientFiltering ? : boolean ; useMH ? : boolean ; } /**
 * Silly DDS example that models a six sided die.
 *
 * Unlike the typical 'Dice Roller' example where clients clobber each other's last roll in a
 * SharedMap, this 'SharedDie' DDS works by advancing an internal PRNG each time it sees a 'roll'
 * operation.
 *
 * Because all clients are using the same PRNG starting in the same state, they arrive at
 * consensus by simply applying the same number of rolls.  (A fun addition would be logging
 * who received which roll, which would need to change as clients learn how races are resolved
 * in the total order)
 */ export class SharedPropertyTree extends SharedObject { tipView : SerializedChangeSet = { } ; remoteTipView : SerializedChangeSet = { } ; localChanges : IPropertyTreeMessage [ ] = [ ] ; remoteChanges : IPropertyTreeMessage [ ] = [ ] ; unrebasedRemoteChanges : Record < string , IRemotePropertyTreeMessage > = { } ; transmissionsHaveBeenStopped = false ; enqueuedMessages : IPropertyTreeMessage [ ] = [ ] ; notificationDelayScope : number = 0 ; _root : any = PropertyFactory . create ( "NodeProperty" ) ; options : SharedPropertyTreeOptions ; skipSequenceNumber : number = - 1 ; headCommitGuid : string = "" ; useMH : boolean = false ; public constructor ( id , runtime , attributes , options ) { super ( id , runtime , attributes ) ; this . options = options ; this . _root . _tree = this ; this . useMH = options . useMH ?? false ; } param runtime param id returns public static create ( runtime , id ? , queryString ? ) { return runtime . createChannel ( id , PropertyTreeFactory . Type ) as SharedPropertyTree ; } returns public static getFactory ( ) { return new PropertyTreeFactory ( ) ; } /**
     * in case of partial checkout we want to send the paths we are interested in once we are connected
     */ protected onConnect ( ) { this . scopeFutureDeltasToPaths ( this . options . paths ) ; } private scopeFutureDeltasToPaths ( paths ? ) { const socket = ( this . runtime . deltaManager as any ) . deltaManager . connectionManager . connection . socket ; socket . emit ( "partial_checkout" , { paths } ) ; } public _reportDirtinessToView ( ) { if ( this . listenerCount ( "localModification" ) > 0 ) { const changes = this . _root . _serialize ( true , false , BaseProperty . MODIFIED_STATE_FLAGS . DIRTY ) ; const _changeSet = new ChangeSet ( changes ) ; if ( ! isEmpty ( _changeSet . getSerializedChangeSet ( ) ) ) { this . emit ( "localModification" , _changeSet ) ; } } this . _root . cleanDirty ( BaseProperty . MODIFIED_STATE_FLAGS . DIRTY ) ; } public get changeSet ( ) : SerializedChangeSet { return this . tipView ; } public get activeCommit ( ) : IPropertyTreeMessage { if ( this . localChanges . length > 0 ) { return this . localChanges [ this . localChanges . length - 1 ] ; } else { return this . remoteChanges [ this . remoteChanges . length - 1 ] ; } } public get root ( ) : NodeProperty { return this . _root as NodeProperty ; } public commit ( metadata ? , submitEmptyChange ? ) { const changes = this . _root . _serialize ( true , false , BaseProperty . MODIFIED_STATE_FLAGS . PENDING_CHANGE ) ; let doSubmit = ! ! submitEmptyChange ; if ( submitEmptyChange === undefined ) { doSubmit = metadata !== undefined ; } if ( doSubmit || ! isEmpty ( changes ) ) { this . applyChangeSet ( changes , metadata || { } ) ; this . root . cleanDirty ( ) ; } } private applyChangeSet ( changeSet , metadata ) { const _changeSet = new ChangeSet ( changeSet ) ; _changeSet . _toReversibleChangeSet ( this . tipView ) ; const remoteHeadGuid = this . remoteChanges . length > 0 ? this . remoteChanges [ this . remoteChanges . length - 1 ] . guid : this . headCommitGuid ; const change = { op : OpKind . ChangeSet , changeSet , metadata , guid : uuidv4 ( ) , remoteHeadGuid , referenceGuid : this . localChanges . length > 0 ? this . localChanges [ this . localChanges . length - 1 ] . guid : remoteHeadGuid , localBranchStart : this . localChanges . length > 0 ? this . localChanges [ 0 ] . guid : undefined , useMH : this . useMH } ; this . _applyLocalChangeSet ( change ) ; if ( this . transmissionsHaveBeenStopped ) { this . enqueuedMessages . push ( cloneDeep ( change ) ) ; } else { this . submitLocalMessage ( cloneDeep ( change ) ) ; } } stopTransmission ( stop ) { this . transmissionsHaveBeenStopped = stop ; if ( stop === false ) { for ( const message of this . enqueuedMessages ) { this . submitLocalMessage ( message ) ; } this . enqueuedMessages = [ ] ; } } /**
     * Delays notifications until popNotificationDelayScope has been called the same number of times as
     * pushNotificationDelayScope.
     */ public pushNotificationDelayScope ( ) { this . notificationDelayScope ++ ; if ( this . notificationDelayScope === 0 ) { this . _root . _reportDirtinessToView ( ) ; } } /**
     * Re-enables notifications when popNotificationDelayScope has been called the same number of times as
     * pushNotificationDelayScope.
     */ public popNotificationDelayScope ( ) { if ( this . notificationDelayScope === 0 ) { console . error ( "Unbalanced push/pop calls." ) ; } this . notificationDelayScope -- ; this . _root . _reportDirtinessToView ( ) ; } param message param local param localOpMetadata protected processCore ( message , local , localOpMetadata ) { if ( message . type === MessageType . Operation && message . sequenceNumber > this . skipSequenceNumber ) { const content = { ... message . contents , sequenceNumber : message . sequenceNumber } ; switch ( content . op ) { case OpKind . ChangeSet : this . _applyRemoteChangeSet ( cloneDeep ( content ) ) ; break ; default : break ; } } } public static prune ( minimumSequenceNumber , remoteChanges , unrebasedRemoteChanges ) { const remoteChangeMap = new Map < string , number > ( ) ; remoteChanges . forEach ( ( change , index ) => { remoteChangeMap . set ( change . guid , index ) ; } ) ; const visitedUnrebasedRemoteChanges = new Set < string > ( ) ; const visitedRemoteChanges = new Set < string > ( ) ; for ( const id of Object . keys ( unrebasedRemoteChanges ) ) { const unrebasedChange = unrebasedRemoteChanges [ id ] ; if ( unrebasedChange . sequenceNumber >= minimumSequenceNumber && ! visitedUnrebasedRemoteChanges . has ( unrebasedChange . guid ) ) { visitedUnrebasedRemoteChanges . add ( unrebasedChange . guid ) ; let visitor = unrebasedChange ; while ( visitor . remoteHeadGuid !== visitor . referenceGuid || visitedUnrebasedRemoteChanges . has ( visitor . referenceGuid ) ) { const guid = visitor . referenceGuid ; if ( guid === "" ) { break ; } visitor = unrebasedRemoteChanges [ visitor . referenceGuid ] ; if ( ! visitor ) { throw new Error ( `no visitor found for guid "${ guid }"` ) ; } visitedUnrebasedRemoteChanges . add ( visitor . guid ) ; } if ( visitor . remoteHeadGuid === visitor . referenceGuid && remoteChangeMap . has ( visitor . referenceGuid ) ) { visitedRemoteChanges . add ( visitor . referenceGuid ) ; } if ( visitor . remoteHeadGuid === "" ) { visitedRemoteChanges . add ( remoteChanges [ 0 ] . guid ) ; } } } let pruned = 0 ; const prunedUnrebasedRemoteChanges = { } ; for ( const key of Object . keys ( unrebasedRemoteChanges ) ) { if ( visitedUnrebasedRemoteChanges . has ( key ) ) { prunedUnrebasedRemoteChanges [ key ] = unrebasedRemoteChanges [ key ] ; } else { pruned ++ ; } } const minIndex = Math . min ( ... [ ... visitedRemoteChanges ] . map ( ( key ) => remoteChangeMap . get ( key ) as number ) ) ; const prunedRemoteChanges = remoteChanges . slice ( minIndex ) ; pruned += minIndex ; return { remoteChanges : prunedRemoteChanges , unrebasedRemoteChanges : prunedUnrebasedRemoteChanges , prunedCount : pruned } ; } public pruneHistory ( ) { const msn = this . runtime . deltaManager . minimumSequenceNumber ; const { remoteChanges , unrebasedRemoteChanges } = SharedPropertyTree . prune ( msn , this . remoteChanges , this . unrebasedRemoteChanges ) ; this . remoteChanges = remoteChanges ; this . unrebasedRemoteChanges = unrebasedRemoteChanges ; } public summarizeCore ( serializer ) { this . pruneHistory ( ) ; const snapshot = { branchGuid : this . handle . absolutePath . split ( "/" ) . pop ( ) as string , summaryMinimumSequenceNumber : this . runtime . deltaManager . minimumSequenceNumber , useMH : this . useMH , numChunks : 0 } ; const builder = new SummaryTreeBuilder ( ) ; if ( ! this . useMH ) { const summary = { remoteTipView : this . remoteTipView , remoteChanges : this . remoteChanges , unrebasedRemoteChanges : this . unrebasedRemoteChanges } ; const chunkSize = 5000 * 1024 ; let serializedSummary = JSON . stringify ( summary ) ; serializedSummary = serializedSummary . replace ( /[\u007F-\uFFFF]/g , ( c ) => `\\u${ ( "0000" + c . charCodeAt ( 0 ) . toString ( 16 ) ) . substr ( - 4 ) }` ) ; for ( let pos = 0 , i = 0 ; pos < serializedSummary . length ; pos += chunkSize , i ++ ) { builder . addBlob ( `summaryChunk_${ i }` , serializedSummary . substr ( pos , chunkSize ) ) ; snapshot . numChunks ++ ; } } builder . addBlob ( "properties" , serializer !== undefined ? serializer . stringify ( snapshot , this . handle ) : JSON . stringify ( snapshot ) ) ; return builder . getSummaryTree ( ) ; } protected async loadCore ( storage ) { const runtime = this . runtime ; const handleTableChunk = await storage . readBlob ( "properties" ) ; const utf8 = bufferToString ( handleTableChunk , "utf8" ) ; const snapshot = this . serializer . parse ( utf8 ) ; this . useMH = snapshot . useMH ; try { if ( ! snapshot . useMH ) { const chunks = await Promise . all ( range ( snapshot . numChunks ) . map ( async ( i ) => { return bufferToString ( await storage . readBlob ( `summaryChunk_${ i }` ) , "utf8" ) ; } ) ) ; const serializedSummary = chunks . reduce ( ( a , b ) => a + b , "" ) ; const snapshotSummary = JSON . parse ( serializedSummary ) ; if ( snapshotSummary . remoteChanges === undefined || snapshotSummary . remoteTipView === undefined || snapshotSummary . unrebasedRemoteChanges === undefined ) { throw new Error ( "Invalid Snapshot." ) ; } this . remoteTipView = snapshotSummary . remoteTipView ; this . remoteChanges = snapshotSummary . remoteChanges ; this . unrebasedRemoteChanges = snapshotSummary . unrebasedRemoteChanges ; const isPartialCheckoutActive = ! ! ( this . options . clientFiltering && this . options . paths ) ; if ( isPartialCheckoutActive && this . options . paths ) { this . remoteTipView = ChangeSetUtils . getFilteredChangeSetByPaths ( this . remoteTipView , this . options . paths ) ; } this . tipView = cloneDeep ( this . remoteTipView ) ; this . skipSequenceNumber = 0 ; } else { const { branchGuid , summaryMinimumSequenceNumber } = snapshot ; const branchResponse = await axios . get ( `http://localhost:3000/branch/${ branchGuid }` ) ; this . headCommitGuid = branchResponse . data . headCommitGuid ; const { commit : { meta : commitMetadata } , } = ( await axios . get ( `http://localhost:3000/branch/${ branchGuid }/commit/${ this . headCommitGuid }` ) ) . data ; let { changeSet : materializedView } = ( await axios . get ( `http://localhost:3000/branch/${ branchGuid }/commit/${ this . headCommitGuid }/materializedView` ) ) . data ; const isPartialCheckoutActive = this . options . clientFiltering && this . options . paths ; if ( isPartialCheckoutActive && this . options . paths ) { materializedView = ChangeSetUtils . getFilteredChangeSetByPaths ( materializedView , this . options . paths ) ; } this . tipView = materializedView ; this . remoteTipView = cloneDeep ( this . tipView ) ; this . remoteChanges = [ ] ; let missingDeltas = [ ] ; const firstDelta = Math . min ( commitMetadata . minimumSequenceNumber , summaryMinimumSequenceNumber ) ; const lastDelta = commitMetadata . sequenceNumber ; const dm = ( this . runtime . deltaManager as any ) . deltaManager ; await dm . getDeltas ( "DocumentOpen" , firstDelta , lastDelta , ( messages ) => { missingDeltas = messages . filter ( ( x ) => x . type === "op" ) ; } ) ; for ( let i = 0 ; i < missingDeltas . length ; i ++ ) { if ( missingDeltas [ i ] . sequenceNumber < commitMetadata . sequenceNumber ) { const remoteChange = JSON . parse ( missingDeltas [ i ] . contents ) . contents . contents . content . contents ; const { changeSet } = ( await axios . get ( `http://localhost:3000/branch/${ branchGuid }/commit/${ remoteChange . guid }/changeSet` ) ) . data ; remoteChange . changeSet = changeSet ; if ( remoteChange ) { if ( isPartialCheckoutActive && this . options . paths ) { remoteChange . changeSet = ChangeSetUtils . getFilteredChangeSetByPaths ( remoteChange . changeSet , this . options . paths ) ; } this . remoteChanges . push ( remoteChange ) ; } } else { this . processCore ( missingDeltas [ i ] , false , { } ) ; } } this . skipSequenceNumber = lastDelta ?? - 1 ; } } catch ( e ) { this . tipView = { } ; this . remoteTipView = { } ; this . remoteChanges = [ ] ; } finally { this . _root . deserialize ( this . tipView , undefined , false , false ) ; const _changeSet = new ChangeSet ( this . tipView ) ; if ( ! isEmpty ( _changeSet . getSerializedChangeSet ( ) ) ) { this . emit ( "localModification" , _changeSet ) ; } this . root . cleanDirty ( ) ; } } protected registerCore ( ) { } protected onDisconnect ( ) { } private _applyLocalChangeSet ( change ) { const changeSetWrapper = new ChangeSet ( this . tipView ) ; changeSetWrapper . applyChangeSet ( change . changeSet ) ; if ( this . runtime . attachState === AttachState . Detached ) { const remoteChangeSetWrapper = new ChangeSet ( this . remoteTipView ) ; remoteChangeSetWrapper . applyChangeSet ( change . changeSet ) ; } else { this . localChanges . push ( change ) ; } } private _applyRemoteChangeSet ( change ) { this . unrebasedRemoteChanges [ change . guid ] = cloneDeep ( change ) ; if ( this . remoteChanges . length !== 0 ) { rebaseToRemoteChanges ( change , this . getUnrebasedChange . bind ( this ) , this . getRebasedChanges . bind ( this ) ) ; } this . remoteChanges . push ( change ) ; const remoteChangeSetWrapper = new ChangeSet ( this . remoteTipView ) ; remoteChangeSetWrapper . applyChangeSet ( change . changeSet ) ; const pendingChanges = this . _root . _serialize ( true , false , BaseProperty . MODIFIED_STATE_FLAGS . PENDING_CHANGE ) ; new ChangeSet ( pendingChanges ) . _toReversibleChangeSet ( this . tipView ) ; const changesToTip = { } ; const changesNeeded = this . rebaseLocalChanges ( change , pendingChanges , changesToTip ) ; if ( changesNeeded ) { this . pushNotificationDelayScope ( ) ; this . _root . applyChangeSet ( changesToTip ) ; this . _root . cleanDirty ( BaseProperty . MODIFIED_STATE_FLAGS . PENDING_CHANGE ) ; this . _root . applyChangeSet ( pendingChanges ) ; this . popNotificationDelayScope ( ) ; } } getUnrebasedChange ( guid ) { return this . unrebasedRemoteChanges [ guid ] ; } getRebasedChanges ( startGuid , endGuid ? ) { const startIndex = findIndex ( this . remoteChanges , ( c ) => c . guid === startGuid ) ; if ( endGuid !== undefined ) { const endIndex = findIndex ( this . remoteChanges , ( c ) => c . guid === endGuid ) ; return this . remoteChanges . slice ( startIndex + 1 , endIndex + 1 ) ; } return this . remoteChanges . slice ( startIndex + 1 ) ; } private rebaseLocalChanges ( change , pendingChanges , newTipDelta ) { let rebaseBaseChangeSet = cloneDeep ( change . changeSet ) ; const accumulatedChanges = { } ; const conflicts = [ ] as any [ ] ; if ( this . localChanges . length > 0 && this . localChanges [ 0 ] . guid === change . guid ) { this . localChanges . shift ( ) ; return false ; } for ( let i = 0 ; i < this . localChanges . length ; i ++ ) { console . assert ( this . localChanges [ i ] . guid !== change . guid ) ; const rebaseMetaInformation = new Map ( ) ; const copiedChangeSet = new ChangeSet ( cloneDeep ( this . localChanges [ i ] . changeSet ) ) ; new ChangeSet ( rebaseBaseChangeSet ) . _rebaseChangeSet ( this . localChanges [ i ] . changeSet , conflicts , { applyAfterMetaInformation : rebaseMetaInformation } ) ; copiedChangeSet . toInverseChangeSet ( ) ; copiedChangeSet . applyChangeSet ( rebaseBaseChangeSet ) ; copiedChangeSet . applyChangeSet ( this . localChanges [ i ] . changeSet , { applyAfterMetaInformation : rebaseMetaInformation } ) ; rebaseBaseChangeSet = copiedChangeSet . getSerializedChangeSet ( ) ; new ChangeSet ( accumulatedChanges ) . applyChangeSet ( this . localChanges [ i ] . changeSet ) ; } const pendingChangesRebaseMetaInformation = new Map ( ) ; const deltaToTipCS = new ChangeSet ( newTipDelta ) ; deltaToTipCS . applyChangeSet ( pendingChanges ) ; deltaToTipCS . toInverseChangeSet ( ) ; new ChangeSet ( rebaseBaseChangeSet ) . _rebaseChangeSet ( pendingChanges , conflicts , { applyAfterMetaInformation : pendingChangesRebaseMetaInformation } ) ; deltaToTipCS . applyChangeSet ( rebaseBaseChangeSet ) ; deltaToTipCS . applyChangeSet ( pendingChanges , { applyAfterMetaInformation : pendingChangesRebaseMetaInformation } ) ; if ( ! this . tipView ) { this . tipView = cloneDeep ( this . remoteTipView ) ; const changeSet = new ChangeSet ( this . tipView ) ; changeSet . applyChangeSet ( accumulatedChanges ) ; } else { new ChangeSet ( this . tipView ) . applyChangeSet ( newTipDelta ) ; } return true ; } protected applyStashedOp ( ) { throw new Error ( "not implemented" ) ; } }