import { Deferred , bufferToString , assert } from "@fluidframework/common-utils" ; import { ChildLogger } from "@fluidframework/telemetry-utils" ; import { ISequencedDocumentMessage , MessageType , } from "@fluidframework/protocol-definitions" ; import { IChannelAttributes , IFluidDataStoreRuntime , IChannelStorageService , } from "@fluidframework/datastore-definitions" ; import { Client , createAnnotateRangeOp , createGroupOp , createInsertOp , createRemoveRangeOp , ICombiningOp , IJSONSegment , IMergeTreeAnnotateMsg , IMergeTreeDeltaOp , IMergeTreeGroupMsg , IMergeTreeOp , IMergeTreeRemoveMsg , IRelativePosition , ISegment , ISegmentAction , LocalReference , matchProperties , MergeTreeDeltaType , PropertySet , RangeStackMap , ReferencePosition , ReferenceType , SegmentGroup , } from "@fluidframework/merge-tree" ; import { ObjectStoragePartition , SummaryTreeBuilder } from "@fluidframework/runtime-utils" ; import { IFluidSerializer , makeHandlesSerializable , parseHandles , SharedObject , ISharedObjectEvents , SummarySerializer , } from "@fluidframework/shared-object-base" ; import { IEventThisPlaceHolder } from "@fluidframework/common-definitions" ; import { ISummaryTreeWithStats } from "@fluidframework/runtime-definitions" ; import { IntervalCollection , SequenceInterval , SequenceIntervalCollectionValueType , } from "./intervalCollection" ; import { MapKernel } from "./mapKernel" ; import { IValueChanged } from "./mapKernelInterfaces" ; import { SequenceDeltaEvent , SequenceMaintenanceEvent } from "./sequenceDeltaEvent" ; import { ISharedIntervalCollection } from "./sharedIntervalCollection" ; const snapshotFileName = "header" ; const contentPath = "content" ; /**
 * Events emitted in response to changes to the sequence data.
 *
 * ### "sequenceDelta"
 *
 * The sequenceDelta event is emitted when segments are inserted, annotated, or removed.
 *
 * #### Listener signature
 *
 * ```typescript
 * (event: SequenceDeltaEvent, target: IEventThisPlaceHolder) => void
 * ```
 * - `event` - Various information on the segments that were modified.
 *
 * - `target` - The sequence itself.
 *
 * ### "maintenance"
 *
 * The maintenance event is emitted when segments are modified during merge-tree maintenance.
 *
 * #### Listener signature
 *
 * ```typescript
 * (event: SequenceMaintenanceEvent, target: IEventThisPlaceHolder) => void
 * ```
 * - `event` - Various information on the segments that were modified.
 *
 * - `target` - The sequence itself.
 */ export interface ISharedSegmentSequenceEvents extends ISharedObjectEvents { ( event , listener ) ; ( event , listener ) ; } export abstract class SharedSegmentSequence < T extends ISegment > extends SharedObject < ISharedSegmentSequenceEvents > implements ISharedIntervalCollection < SequenceInterval > { get loaded ( ) : Promise < void > { return this . loadedDeferred . promise ; } private static createOpsFromDelta ( event ) { const ops = [ ] ; for ( const r of event . ranges ) { switch ( event . deltaOperation ) { case MergeTreeDeltaType . ANNOTATE : { const lastAnnotate = ops [ ops . length - 1 ] as IMergeTreeAnnotateMsg ; const props = { } ; for ( const key of Object . keys ( r . propertyDeltas ) ) { props [ key ] = r . segment . properties [ key ] === undefined ? null : r . segment . properties [ key ] ; } if ( lastAnnotate && lastAnnotate . pos2 === r . position && matchProperties ( lastAnnotate . props , props ) ) { lastAnnotate . pos2 += r . segment . cachedLength ; } else { ops . push ( createAnnotateRangeOp ( r . position , r . position + r . segment . cachedLength , props , undefined ) ) ; } break ; } case MergeTreeDeltaType . INSERT : ops . push ( createInsertOp ( r . position , r . segment . clone ( ) . toJSONObject ( ) ) ) ; break ; case MergeTreeDeltaType . REMOVE : { const lastRem = ops [ ops . length - 1 ] as IMergeTreeRemoveMsg ; if ( lastRem ?. pos1 === r . position ) { lastRem . pos2 += r . segment . cachedLength ; } else { ops . push ( createRemoveRangeOp ( r . position , r . position + r . segment . cachedLength ) ) ; } break ; } default : } } return ops ; } protected client : Client ; protected loadedDeferred = new Deferred < void > ( ) ; private readonly loadedDeferredOutgoingOps : [ IMergeTreeOp , SegmentGroup | SegmentGroup [ ] ] [ ] = [ ] ; private deferIncomingOps = true ; private readonly loadedDeferredIncomingOps : ISequencedDocumentMessage [ ] = [ ] ; private messagesSinceMSNChange : ISequencedDocumentMessage [ ] = [ ] ; private readonly intervalMapKernel : MapKernel ; constructor ( private readonly dataStoreRuntime , public id , attributes , public readonly segmentFromSpec ) { super ( id , dataStoreRuntime , attributes ) ; this . loadedDeferred . promise . catch ( ( error ) => { this . logger . sendErrorEvent ( { eventName : "SequenceLoadFailed" } , error ) ; } ) ; this . client = new Client ( segmentFromSpec , ChildLogger . create ( this . logger , "SharedSegmentSequence.MergeTreeClient" ) , dataStoreRuntime . options ) ; super . on ( "newListener" , ( event ) => { switch ( event ) { case "sequenceDelta" : if ( ! this . client . mergeTreeDeltaCallback ) { this . client . mergeTreeDeltaCallback = ( opArgs , deltaArgs ) => { this . emit ( "sequenceDelta" , new SequenceDeltaEvent ( opArgs , deltaArgs , this . client ) , this ) ; } ; } break ; case "maintenance" : if ( ! this . client . mergeTreeMaintenanceCallback ) { this . client . mergeTreeMaintenanceCallback = ( args , opArgs ) => { this . emit ( "maintenance" , new SequenceMaintenanceEvent ( opArgs , args , this . client ) , this ) ; } ; } break ; default : } } ) ; super . on ( "removeListener" , ( event ) => { switch ( event ) { case "sequenceDelta" : if ( super . listenerCount ( event ) === 0 ) { this . client . mergeTreeDeltaCallback = undefined ; } break ; case "maintenance" : if ( super . listenerCount ( event ) === 0 ) { this . client . mergeTreeMaintenanceCallback = undefined ; } break ; default : break ; } } ) ; this . intervalMapKernel = new MapKernel ( this . serializer , this . handle , ( op , localOpMetadata ) => this . submitLocalMessage ( op , localOpMetadata ) , ( ) => this . isAttached ( ) , [ new SequenceIntervalCollectionValueType ( ) ] ) ; } param start param end public removeRange ( start , end ) { const removeOp = this . client . removeRangeLocal ( start , end ) ; if ( removeOp ) { this . submitSequenceMessage ( removeOp ) ; } return removeOp ; } public groupOperation ( groupOp ) { this . client . localTransaction ( groupOp ) ; this . submitSequenceMessage ( groupOp ) ; } public getContainingSegment ( pos ) { return this . client . getContainingSegment < T > ( pos ) ; } /**
     * Returns the length of the current sequence for the client
     */ public getLength ( ) { return this . client . getLength ( ) ; } param segment public getPosition ( segment ) { return this . client . getPosition ( segment ) ; } param start param end param props param combiningOp public annotateRange ( start , end , props , combiningOp ? ) { const annotateOp = this . client . annotateRangeLocal ( start , end , props , combiningOp ) ; if ( annotateOp ) { this . submitSequenceMessage ( annotateOp ) ; } } public getPropertiesAtPosition ( pos ) { return this . client . getPropertiesAtPosition ( pos ) ; } public getRangeExtentsOfPosition ( pos ) { return this . client . getRangeExtentsOfPosition ( pos ) ; } public createPositionReference ( segment , offset , refType ) { const lref = new LocalReference ( this . client , segment , offset , refType ) ; if ( refType !== ReferenceType . Transient ) { this . addLocalReference ( lref ) ; } return lref ; } public localRefToPos ( localRef ) { if ( localRef . segment ) { return localRef . offset + this . getPosition ( localRef . segment ) ; } else { return - 1 ; } } param remoteClientPosition param remoteClientRefSeq param remoteClientId public resolveRemoteClientPosition ( remoteClientPosition , remoteClientRefSeq , remoteClientId ) { return this . client . resolveRemoteClientPosition ( remoteClientPosition , remoteClientRefSeq , remoteClientId ) ; } public submitSequenceMessage ( message ) { if ( ! this . isAttached ( ) ) { return ; } const translated = makeHandlesSerializable ( message , this . serializer , this . handle ) ; const metadata = this . client . peekPendingSegmentGroups ( message . type === MergeTreeDeltaType . GROUP ? message . ops . length : 1 ) ; if ( ! this . loadedDeferred . isCompleted ) { this . loadedDeferredOutgoingOps . push ( [ translated , metadata ] ) ; } else { this . submitLocalMessage ( translated , metadata ) ; } } public addLocalReference ( lref ) { return this . client . addLocalReference ( lref ) ; } public removeLocalReference ( lref ) { return this . client . removeLocalReference ( lref ) ; } param relativePos public posFromRelativePos ( relativePos ) { return this . client . posFromRelativePos ( relativePos ) ; } param handler param start param end param accum param splitRange public walkSegments < TClientData > ( handler , start ? , end ? , accum ? , splitRange = false ) { return this . client . walkSegments < TClientData > ( handler , start , end , accum , splitRange ) ; } public getStackContext ( startPos , rangeLabels ) { return this . client . getStackContext ( startPos , rangeLabels ) ; } public getCurrentSeq ( ) { return this . client . getCurrentSeq ( ) ; } public insertAtReferencePosition ( pos , segment ) { const insertOp = this . client . insertAtReferencePositionLocal ( pos , segment ) ; if ( insertOp ) { this . submitSequenceMessage ( insertOp ) ; } } public async waitIntervalCollection ( label ) { return this . intervalMapKernel . wait < IntervalCollection < SequenceInterval > > ( this . getIntervalCollectionPath ( label ) ) ; } public getIntervalCollection ( label ) { const labelPath = this . getIntervalCollectionPath ( label ) ; if ( ! this . intervalMapKernel . has ( labelPath ) ) { this . intervalMapKernel . createValueType ( labelPath , SequenceIntervalCollectionValueType . Name , undefined ) ; } const sharedCollection = this . intervalMapKernel . get < IntervalCollection < SequenceInterval > > ( labelPath ) ; return sharedCollection ; } protected summarizeCore ( serializer ) { const builder = new SummaryTreeBuilder ( ) ; if ( this . intervalMapKernel . size > 0 ) { builder . addBlob ( snapshotFileName , this . intervalMapKernel . serialize ( serializer ) ) ; } builder . addWithStats ( contentPath , this . summarizeMergeTree ( serializer ) ) ; return builder . getSummaryTree ( ) ; } /**
     * Runs serializer over the GC data for this SharedMatrix.
     * All the IFluidHandle's represent routes to other objects.
     */ protected processGCDataCore ( serializer ) { if ( this . intervalMapKernel . size > 0 ) { this . intervalMapKernel . serialize ( serializer ) ; } this . client . serializeGCData ( this . handle , serializer ) ; } param start param end param segment protected replaceRange ( start , end , segment ) { const insertIndex = Math . max ( start , end ) ; const insert = this . client . insertSegmentLocal ( insertIndex , segment ) ; if ( insert ) { if ( start < end ) { const remove = this . client . removeRangeLocal ( start , end ) ; this . submitSequenceMessage ( createGroupOp ( insert , remove ) ) ; } else { this . submitSequenceMessage ( insert ) ; } } } protected onConnect ( ) { this . client . startOrUpdateCollaboration ( this . runtime . clientId ) ; } protected onDisconnect ( ) { } protected reSubmitCore ( content , localOpMetadata ) { if ( ! this . intervalMapKernel . trySubmitMessage ( content , localOpMetadata ) ) { this . submitSequenceMessage ( this . client . regeneratePendingOp ( content as IMergeTreeOp , localOpMetadata as SegmentGroup | SegmentGroup [ ] ) ) ; } } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.loadCore}
     */ protected async loadCore ( storage ) { if ( await storage . contains ( snapshotFileName ) ) { const blob = await storage . readBlob ( snapshotFileName ) ; const header = bufferToString ( blob , "utf8" ) ; this . intervalMapKernel . populate ( header ) ; } try { const { catchupOpsP } = await this . client . load ( this . runtime , new ObjectStoragePartition ( storage , contentPath ) , this . serializer ) ; const loadCatchUpOps = catchupOpsP . then ( ( msgs ) => { msgs . forEach ( ( m ) => { const collabWindow = this . client . getCollabWindow ( ) ; if ( m . minimumSequenceNumber < collabWindow . minSeq || m . referenceSequenceNumber < collabWindow . minSeq || m . sequenceNumber <= collabWindow . minSeq || m . sequenceNumber <= collabWindow . currentSeq ) { throw new Error ( `Invalid catchup operations in snapshot: ${ JSON . stringify ( { op : { seq : m . sequenceNumber , minSeq : m . minimumSequenceNumber , refSeq : m . referenceSequenceNumber } , collabWindow : { seq : collabWindow . currentSeq , minSeq : collabWindow . minSeq } } ) }` ) ; } this . processMergeTreeMsg ( m ) ; } ) ; this . loadFinished ( ) ; } ) . catch ( ( error ) => { this . loadFinished ( error ) ; } ) ; if ( this . dataStoreRuntime . options ?. sequenceInitializeFromHeaderOnly !== true ) { await loadCatchUpOps ; } } catch ( error ) { this . loadFinished ( error ) ; } } protected processCore ( message , local , localOpMetadata ) { if ( this . deferIncomingOps ) { assert ( ! local , 0x072 ) ; this . loadedDeferredIncomingOps . push ( message ) ; } else { assert ( message . type === MessageType . Operation , 0x073 ) ; const handled = this . intervalMapKernel . tryProcessMessage ( message . contents , local , message , localOpMetadata ) ; if ( ! handled ) { this . processMergeTreeMsg ( message ) ; } } } protected registerCore ( ) { for ( const value of this . intervalMapKernel . values ( ) ) { if ( SharedObject . is ( value ) ) { value . bindToContext ( ) ; } } this . client . startOrUpdateCollaboration ( this . runtime . clientId ) ; } protected didAttach ( ) { if ( this . isAttached ( ) ) { this . client . startOrUpdateCollaboration ( this . runtime . clientId ?? "attached" ) ; } } protected initializeLocalCore ( ) { super . initializeLocalCore ( ) ; this . loadFinished ( ) ; } private summarizeMergeTree ( serializer ) { assert ( this . loadedDeferred . isCompleted , 0x074 ) ; const minSeq = this . runtime . deltaManager . minimumSequenceNumber ; this . processMinSequenceNumberChanged ( minSeq ) ; this . messagesSinceMSNChange . forEach ( ( m ) => m . minimumSequenceNumber = minSeq ) ; return this . client . summarize ( this . runtime , this . handle , serializer , this . messagesSinceMSNChange ) ; } private processMergeTreeMsg ( rawMessage ) { const message = parseHandles ( rawMessage , this . serializer ) ; const ops = [ ] ; function transfromOps ( event ) { ops . push ( ... SharedSegmentSequence . createOpsFromDelta ( event ) ) ; } const needsTransformation = message . referenceSequenceNumber !== message . sequenceNumber - 1 ; let stashMessage = message ; if ( this . runtime . options ?. newMergeTreeSnapshotFormat !== true ) { if ( needsTransformation ) { this . on ( "sequenceDelta" , transfromOps ) ; } } this . client . applyMsg ( message ) ; if ( this . runtime . options ?. newMergeTreeSnapshotFormat !== true ) { if ( needsTransformation ) { this . removeListener ( "sequenceDelta" , transfromOps ) ; stashMessage = { ... message , referenceSequenceNumber : stashMessage . sequenceNumber - 1 , contents : ops . length !== 1 ? createGroupOp ( ... ops ) : ops [ 0 ] } ; } this . messagesSinceMSNChange . push ( stashMessage ) ; if ( this . messagesSinceMSNChange . length > 20 && this . messagesSinceMSNChange [ 20 ] . sequenceNumber < message . minimumSequenceNumber ) { this . processMinSequenceNumberChanged ( message . minimumSequenceNumber ) ; } } } private getIntervalCollectionPath ( label ) { return `intervalCollections/${ label }` ; } private processMinSequenceNumberChanged ( minSeq ) { let index = 0 ; for ( ; index < this . messagesSinceMSNChange . length ; index ++ ) { if ( this . messagesSinceMSNChange [ index ] . sequenceNumber > minSeq ) { break ; } } if ( index !== 0 ) { this . messagesSinceMSNChange = this . messagesSinceMSNChange . slice ( index ) ; } } private loadFinished ( error ? ) { if ( ! this . loadedDeferred . isCompleted ) { this . initializeIntervalCollections ( ) ; if ( error ) { this . loadedDeferred . reject ( error ) ; throw error ; } else { this . deferIncomingOps = false ; while ( this . loadedDeferredIncomingOps . length > 0 ) { this . processCore ( this . loadedDeferredIncomingOps . shift ( ) , false , undefined ) ; } this . loadedDeferred . resolve ( ) ; while ( this . loadedDeferredOutgoingOps . length > 0 ) { const opData = this . loadedDeferredOutgoingOps . shift ( ) ; this . reSubmitCore ( opData [ 0 ] , opData [ 1 ] ) ; } } } } private initializeIntervalCollections ( ) { this . intervalMapKernel . eventEmitter . on ( "valueChanged" , ( ev ) => { const intervalCollection = this . intervalMapKernel . get < IntervalCollection < SequenceInterval > > ( ev . key ) ; if ( ! intervalCollection . attached ) { intervalCollection . attachGraph ( this . client , ev . key ) ; } } ) ; for ( const key of this . intervalMapKernel . keys ( ) ) { const intervalCollection = this . intervalMapKernel . get < IntervalCollection < SequenceInterval > > ( key ) ; intervalCollection . attachGraph ( this . client , key ) ; } } protected applyStashedOp ( ) { throw new Error ( "not implemented" ) ; } }