import { bufferToString , IsoBuffer } from '@fluidframework/common-utils' ; import { IFluidHandle } from '@fluidframework/core-interfaces' ; import { ISequencedDocumentMessage } from '@fluidframework/protocol-definitions' ; import { IFluidDataStoreRuntime , IChannelStorageService , IChannelAttributes , } from '@fluidframework/datastore-definitions' ; import { AttachState } from '@fluidframework/container-definitions' ; import { createSingleBlobSummary , IFluidSerializer , ISharedObjectEvents , serializeHandles , SharedObject , } from '@fluidframework/shared-object-base' ; import { ITelemetryLogger } from '@fluidframework/common-definitions' ; import { ChildLogger , ITelemetryLoggerPropertyBags , PerformanceEvent } from '@fluidframework/telemetry-utils' ; import { ISummaryTreeWithStats } from '@fluidframework/runtime-definitions' ; import { assert , assertNotUndefined , fail } from '../Common' ; import { EditLog , OrderedEditSet } from '../EditLog' ; import { EditId } from '../Identifiers' ; import { Snapshot } from '../Snapshot' ; import { initialTree } from '../InitialTree' ; import { CachingLogViewer , EditStatusCallback , LogViewer } from '../LogViewer' ; import { convertSummaryToReadFormat , deserialize , readFormatVersion } from '../SummaryBackCompatibility' ; import { Edit , SharedTreeOpType , SharedTreeEditOp , SharedTreeHandleOp , EditWithoutId , SharedTreeOp , EditStatus , } from './PersistedTypes' ; import { serialize , SharedTreeSummarizer , SharedTreeSummary , SharedTreeSummaryBase } from './Summary' ; import { GenericTransaction } from './GenericTransaction' ; import { newEdit } from './GenericEditUtilities' ; /**
 * Filename where the snapshot is stored.
 */ const snapshotFileName = 'header' ; const initialSummary = { version : readFormatVersion , currentTree : initialTree , editHistory : { editChunks : [ ] , editIds : [ ] } } ; /**
 * An event emitted by a `SharedTree` to indicate a state change. See  ISharedTreeEvents for event argument information.
 *  public export enum SharedTreeEvent { /**
     * An edit has been committed to the log.
     * This happens when either:
     * 	1. A locally generated edit is added to the log.
     * 	2. A remotely generated edit is added to the log.
     * Note that, for locally generated edits, this event will not be emitted again when that edit is sequenced.
     * Passed the EditId of the committed edit, i.e. supports callbacks of type  EditCommittedHandler .
      EditCommitted = 'committedEdit' } /**
 * An event emitted by a `SharedTree` for diagnostic purposes.
 * See  ISharedTreeEvents for event argument information.
  export enum SharedTreeDiagnosticEvent { /**
     * A single catch up blob has been uploaded.
     */ CatchUpBlobUploaded = 'uploadedCatchUpBlob' , /**
     * A valid edit (local or remote) has been applied.
     * Passed the EditId of the applied edit.
     * Note that this may be called multiple times, due to concurrent edits causing reordering,
     * and/or due to not caching the output of every edit.
     */ AppliedEdit = 'appliedEdit' , /**
     * An invalid edit (local or remote) has been dropped.
     * Passed the EditId of the dropped edit.
     * Note that this may be called multiple times, due to concurrent edits causing reordering,
     * and/or due to not caching the output of every edit.
     */ DroppedInvalidEdit = 'droppedInvalidEdit' , /**
     * A malformed edit (local or remote) has been dropped.
     * Passed the EditId of the dropped edit.
     * Note that this may be called multiple times, due to concurrent edits causing reordering,
     * and/or due to not caching the output of every edit.
     */ DroppedMalformedEdit = 'droppedMalformedEdit' } public export interface EditCommittedEventArguments < TSharedTree > { /** The ID of the edit committed. */ editId : EditId ; /** Whether or not this is a local edit. */ local : boolean ; /** The tree the edit was committed on. Required for local edit events handled by SharedTreeUndoRedoHandler. */ tree : TSharedTree ; } /**
 * Events which may be emitted by `SharedTree`. See  SharedTreeEvent for documentation of event semantics.
  export interface ISharedTreeEvents < TSharedTree > extends ISharedObjectEvents { ( event , listener ) ; } /**
 * Expected type for a handler of the `EditCommitted` event.
 */ export type EditCommittedHandler < TSharedTree > = ( args : EditCommittedEventArguments < TSharedTree > ) => void ; const sharedTreeTelemetryProperties = { all : { isSharedTreeEvent : true } } ; public export abstract class GenericSharedTree < TChange > extends SharedObject < ISharedTreeEvents < TChange > > { /**
     * The log of completed edits for this SharedTree.
     */ private editLog : EditLog < TChange > ; /**
     * As an implementation detail, SharedTree uses a log viewer that caches snapshots at different revisions.
     * It is not exposed to avoid accidental correctness issues, but `logViewer` is exposed in order to give clients a way
     * to access the revision history.
     */ private cachingLogViewer : CachingLogViewer < TChange > ; /**
     * Viewer for trees defined by editLog. This allows access to views of the tree at different revisions (various points in time).
     */ public get logViewer ( ) : LogViewer { return this . cachingLogViewer ; } protected readonly logger : ITelemetryLogger ; public readonly transactionFactory : ( snapshot : Snapshot ) => GenericTransaction < TChange > ; /** Indicates if the client is the oldest member of the quorum. */ private currentIsOldest : boolean ; private readonly processEditResult = ( editResult , editId ) : void => { this . emit ( GenericSharedTree . eventFromEditResult ( editResult ) , editId ) ; } ; param runtime param id param expensiveValidation param summarizeHistory public constructor ( runtime , id , transactionFactory , attributes , private readonly expensiveValidation = false , protected readonly summarizeHistory = true ) { super ( id , runtime , attributes ) ; this . expensiveValidation = expensiveValidation ; this . transactionFactory = transactionFactory ; const quorum = this . runtime . getQuorum ( ) ; this . currentIsOldest = this . computeIsOldest ( ) ; quorum . on ( 'addMember' , this . updateOldest ) ; quorum . on ( 'removeMember' , this . updateOldest ) ; runtime . on ( 'connected' , this . updateOldest ) ; runtime . on ( 'disconnected' , this . updateOldest ) ; this . logger = ChildLogger . create ( runtime . logger , 'SharedTree' , sharedTreeTelemetryProperties ) ; const { editLog , cachingLogViewer } = this . createEditLogFromSummary ( initialSummary , this . processEditResult ) ; this . editLog = editLog ; this . cachingLogViewer = cachingLogViewer ; } /**
     * Re-computes currentIsOldest and emits an event if it has changed.
     * TODO:#55900: Get rid of copy-pasted OldestClientObserver code
     */ private readonly updateOldest = ( ) => { const oldest = this . computeIsOldest ( ) ; if ( this . currentIsOldest !== oldest ) { this . currentIsOldest = oldest ; if ( oldest ) { this . emit ( 'becameOldest' ) ; } else { this . emit ( 'lostOldest' ) ; } } } ; /**
     * Computes the oldest client in the quorum, true by default if the container is detached and false by default if the client isn't connected.
     * TODO:#55900: Get rid of copy-pasted OldestClientObserver code
     */ private computeIsOldest ( ) { if ( this . runtime . attachState === AttachState . Detached ) { return true ; } if ( ! this . runtime . connected ) { return false ; } assert ( this . runtime . clientId !== undefined , 'Client id should be set if connected.' ) ; const quorum = this . runtime . getQuorum ( ) ; const selfSequencedClient = quorum . getMember ( this . runtime . clientId ) ; if ( selfSequencedClient === undefined ) { return false ; } const members = quorum . getMembers ( ) ; for ( const sequencedClient of members . values ( ) ) { if ( sequencedClient . sequenceNumber < selfSequencedClient . sequenceNumber ) { return false ; } } return true ; } returns public get currentView ( ) : Snapshot { return this . logViewer . getSnapshotInSession ( Number . POSITIVE_INFINITY ) ; } returns public get edits ( ) : OrderedEditSet < TChange > { return this . editLog ; } internal public applyEdit ( ... changes ) { const edit = newEdit ( changes ) ; this . processLocalEdit ( edit ) ; return edit . id ; } private deserializeHandle ( serializedHandle ) { const deserializeHandle = this . serializer . parse ( serializedHandle ) ; assert ( typeof deserializeHandle === 'object' ) ; return deserializeHandle as IFluidHandle < ArrayBufferLike > ; } /**
     * Uploads the edit chunk and sends the chunk starting revision along with the resulting handle as an op.
     */ private async uploadEditChunk ( edits , startRevision ) { try { const editHandle = await this . runtime . uploadBlob ( IsoBuffer . from ( JSON . stringify ( { edits } ) ) ) ; this . submitLocalMessage ( { editHandle : serializeHandles ( editHandle , this . serializer , this . handle ) , startRevision , type : SharedTreeOpType . Handle } ) ; } catch ( error ) { this . logger . sendErrorEvent ( { eventName : 'EditChunkUploadFailure' } , error ) ; } } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.summarizeCore}
     */ public summarizeCore ( serializer ) { return createSingleBlobSummary ( snapshotFileName , this . saveSerializedSummary ( { serializer } ) ) ; } param options internal public saveSerializedSummary ( options ? ) { const { serializer , summarizer } = options || { } ; return serialize ( summarizer ? summarizer ( this . editLog , this . currentView ) : this . saveSummary ( ) , serializer || this . serializer , this . handle ) ; } internal public saveSummary ( ) { if ( this . editLog . numberOfLocalEdits > 0 ) { assert ( this . runtime . attachState !== AttachState . Attached , 'Summarizing should not occur with local edits except on first attach.' ) ; this . editLog . sequenceLocalEdits ( ) ; } assert ( this . editLog . numberOfLocalEdits === 0 , 'generateSummary must not be called with local edits' ) ; return this . generateSummary ( this . editLog ) ; } /**
     * Generates a SharedTree summary for the current state of the tree.
     * Will never be called when local edits are present.
     */ protected abstract generateSummary ( editLog ) ; internal public loadSummary ( summary ) { const { editLog , cachingLogViewer } = this . createEditLogFromSummary ( summary , this . processEditResult ) ; this . editLog = editLog ; this . cachingLogViewer = cachingLogViewer ; if ( this . runtime . connected ) { const noChunksReadyForUpload = this . editLog . getEditChunksReadyForUpload ( ) [ Symbol . iterator ] ( ) . next ( ) . done ; if ( noChunksReadyForUpload === undefined || ! noChunksReadyForUpload ) { if ( this . runtime . getQuorum ( ) . getMembers ( ) . size === 0 ) { const noop = { type : SharedTreeOpType . NoOp } ; this . submitLocalMessage ( noop ) ; } else if ( this . currentIsOldest ) { void this . uploadCatchUpBlobs ( ) ; } } this . on ( 'becameOldest' , ( ) => void this . uploadCatchUpBlobs ( ) ) ; } } private static eventFromEditResult ( editStatus ) { switch ( editStatus ) { case EditStatus . Applied : return SharedTreeDiagnosticEvent . AppliedEdit ; case EditStatus . Invalid : return SharedTreeDiagnosticEvent . DroppedInvalidEdit ; default : return SharedTreeDiagnosticEvent . DroppedMalformedEdit ; } } private createEditLogFromSummary ( summary , callback ) { const convertedSummary = convertSummaryToReadFormat < TChange > ( summary ) ; if ( typeof convertedSummary === 'string' ) { fail ( convertedSummary ) ; } const { editHistory , currentTree } = convertedSummary ; const currentView = Snapshot . fromTree ( currentTree ) ; const editLog = new EditLog ( editHistory , this . logger ) ; const logViewer = new CachingLogViewer ( editLog , Snapshot . fromTree ( initialTree ) , [ [ editLog . length , { snapshot : currentView } ] ] , this . expensiveValidation , callback , this . logger , this . transactionFactory , 0 ) ; return { editLog , cachingLogViewer : logViewer } ; } /**
     * Upload any full chunks that have yet to be uploaded.
     */ private async uploadCatchUpBlobs ( ) { for ( const [ startRevision , chunk ] of this . editLog . getEditChunksReadyForUpload ( ) ) { await this . uploadEditChunk ( chunk , startRevision ) ; this . emit ( SharedTreeDiagnosticEvent . CatchUpBlobUploaded ) ; } } /**
     * Compares this shared tree to another for equality.
     *
     * Equality means that the histories as captured by the EditLogs are equal.
     *
     * Equality does not include:
     *   - if an edit is open
     *   - the shared tree's id
     *   - local vs sequenced status of edits
     *   - registered event listeners
     *   - state of caches
     * */ public equals < TOtherChangeTypes > ( sharedTree ) { if ( ! this . currentView . equals ( sharedTree . currentView ) ) { return false ; } return this . editLog . equals ( sharedTree . editLog ) ; } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.loadCore}
     */ protected async loadCore ( storage ) { const summaryLoadPerformanceEvent = PerformanceEvent . start ( this . logger , { eventName : 'SummaryLoad' } ) ; try { const newBlob = await storage . readBlob ( snapshotFileName ) ; const blobData = bufferToString ( newBlob , 'utf8' ) ; const summary = deserialize ( blobData , this . serializer ) ; if ( typeof summary === 'string' ) { fail ( summary ) ; } this . loadSummary ( summary ) ; summaryLoadPerformanceEvent . end ( { historySize : this . edits . length } ) ; } catch ( error ) { summaryLoadPerformanceEvent . cancel ( undefined , error ) ; throw error ; } } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.processCore}
     */ protected processCore ( message , local ) { this . cachingLogViewer . setMinimumSequenceNumber ( message . minimumSequenceNumber ) ; const { type } = message . contents ; if ( type === SharedTreeOpType . Handle ) { const { editHandle , startRevision } = message . contents as SharedTreeHandleOp ; this . editLog . processEditChunkHandle ( this . deserializeHandle ( editHandle ) , startRevision ) ; } else if ( type === SharedTreeOpType . Edit ) { const semiSerializedEdit = message . contents . edit ; const stringEdit = JSON . stringify ( semiSerializedEdit ) ; const parsedEdit = this . serializer . parse ( stringEdit ) ; const edit = parsedEdit as Edit < TChange > ; this . processSequencedEdit ( edit , message ) ; } } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.registerCore}
     */ protected registerCore ( ) { } /**
     * {@inheritDoc @fluidframework/shared-object-base#SharedObject.onDisconnect}
     */ protected onDisconnect ( ) { } private processSequencedEdit ( edit , message ) { const { id : editId } = edit ; const wasLocalEdit = this . editLog . isLocalEdit ( editId ) ; const shouldIgnoreEdit = this . editLog . tryGetIndexOfId ( editId ) !== undefined && ! wasLocalEdit ; if ( shouldIgnoreEdit ) { return ; } this . editLog . addSequencedEdit ( edit , message ) ; if ( ! wasLocalEdit ) { const eventArguments = { editId , local : false , tree : this } ; this . emit ( SharedTreeEvent . EditCommitted , eventArguments ) ; } else { const lastPair = this . editLog . getLastEditChunk ( ) ; if ( lastPair !== undefined ) { const [ startRevision , chunk ] = lastPair ; const edits = assertNotUndefined ( chunk . edits ) ; if ( edits . length === this . editLog . editsPerChunk ) { void this . uploadEditChunk ( edits , startRevision ) ; } } } } internal public processLocalEdit ( edit ) { const editOp = { type : SharedTreeOpType . Edit , edit } ; const semiSerialized = this . serializer . encode ( editOp , this . handle ) ; this . submitLocalMessage ( semiSerialized ) ; this . editLog . addLocalEdit ( edit ) ; const eventArguments = { editId : edit . id , local : true , tree : this } ; this . emit ( SharedTreeEvent . EditCommitted , eventArguments ) ; } public getRuntime ( ) { return this . runtime ; } protected applyStashedOp ( ) { throw new Error ( 'not implemented' ) ; } }