import BTree from 'sorted-btree' ; import { IsoBuffer } from '@fluidframework/common-utils' ; import { ITelemetryLogger } from '@fluidframework/common-definitions' ; import { assert , assertNotUndefined , compareArrays , fail } from './Common' ; import { Edit , EditWithoutId } from './generic' ; import { EditId } from './Identifiers' ; import { compareFiniteNumbers } from './SnapshotUtilities' ; public sealed export interface OrderedEditSet < TChange > { /**
     * The length of this `OrderedEditSet`.
     */ readonly length : number ; /**
     * The edit IDs of all edits in the log.
     */ readonly editIds : readonly EditId [ ] ; returns getIndexOfId ( editId ) : number ; returns getIdAtIndex ( index ) : EditId ; returns tryGetIndexOfId ( editId ) : number | undefined ; returns getEditAtIndex ( index ) : Promise < Edit < TChange > > ; returns getEditInSessionAtIndex ( index ) : Edit < TChange > ; returns tryGetEdit ( editId ) : Promise < Edit < TChange > | undefined > ; param useHandles returns internal getEditLogSummary ( useHandles ? ) : EditLogSummary < TChange > ; } internal export interface EditLogSummary < TChange > { /**
     * A of list of serialized chunks and their corresponding keys.
     * Start revision is the index of the first edit in the chunk in relation to the edit log.
     */ readonly editChunks : readonly { readonly startRevision : number ; readonly chunk : EditChunkOrHandle < TChange > ; } [ ] ; /**
     * A list of edits IDs for all sequenced edits.
     */ readonly editIds : readonly EditId [ ] ; } internal export interface EditHandle { readonly get : ( ) => Promise < ArrayBufferLike > ; } /**
 * Server-provided metadata for edits that have been sequenced.
 */ export interface EditSequencingInfo { /**
     * The server-assigned sequence number of the op.
     */ readonly sequenceNumber : number ; /**
     * Last known sequenced edit at the time this op was issued.
     */ readonly referenceSequenceNumber : number ; } /**
 * Server-provided metadata for edits that have been sequenced.
 */ export interface MessageSequencingInfo extends EditSequencingInfo { /**
     * Last sequenced edit that all clients are guaranteed to be aware of.
     * If not specified, then some clients have not seen any edits yet.
     */ readonly minimumSequenceNumber ? : number ; } /**
 * Metadata for a sequenced edit.
 */ export interface SequencedOrderedEditId { readonly isLocal : false ; readonly index : number ; /**
     * Information about the edit's relationship to other sequenced edits.
     * Undefined iff the edit was loaded from a summary.
     */ readonly sequenceInfo ? : EditSequencingInfo ; } /**
 * Metadata for a local edit.
 */ export interface LocalOrderedEditId { readonly isLocal : true ; readonly localSequence : number ; } interface EditChunk < TChange > { handle ? : EditHandle ; edits ? : EditWithoutId < TChange > [ ] ; } internal export type EditChunkOrHandle < TChange > = EditHandle | readonly EditWithoutId < TChange > [ ] ; /**
 * Metadata for an edit.
 */ export type OrderedEditId = SequencedOrderedEditId | LocalOrderedEditId ; /**
 * Returns an object that separates an Edit into two fields, id and editWithoutId.
 */ export function separateEditAndId < TChange > ( edit ) { const editWithoutId = { ... edit , id : undefined } ; delete editWithoutId . id ; return { id : edit . id , editWithoutId } ; } function joinEditAndId < TChange > ( id , edit ) { return { id , ... edit } ; } /**
 * The number of blobs to be loaded in memory at any time.
 * TODO:#49901: Change cache size once the virtualized history summary format is being written.
 * 		 This is so the summarizer doesn't have to reload every edit to generate summaries.
 * */ const loadedChunkCacheSize = Number . POSITIVE_INFINITY ; param edit param isLocal export type EditAddedHandler < TChange > = ( edit : Edit < TChange > , isLocal : boolean , wasLocal : boolean ) => void ; sealed export class EditLog < TChange > implements OrderedEditSet < TChange > { private localEditSequence = 0 ; private _minSequenceNumber = 0 ; private readonly sequencedEditIds : EditId [ ] ; private readonly editChunks : BTree < number , EditChunk < TChange > > ; private readonly localEdits : Edit < TChange > [ ] = [ ] ; private readonly loadedChunkCache : number [ ] = [ ] ; private readonly indexOfFirstEditInSession : number ; private readonly maximumEvictableIndex : number ; private readonly allEditIds : Map < EditId , OrderedEditId > = new Map ( ) ; private readonly editAddedHandlers : EditAddedHandler < TChange > [ ] = [ ] ; private readonly logger ? : ITelemetryLogger ; /**
     * The number of edits associated with each blob.
     */ public readonly editsPerChunk : number ; returns public get earliestAvailableEditIndex ( ) : number { return this . maximumEvictableIndex + 1 ; } returns public get minSequenceNumber ( ) : number { return this . _minSequenceNumber ; } param summary param logger public constructor ( summary = { editIds : [ ] , editChunks : [ ] } , logger ? , editsPerChunk = 100 ) { const { editChunks , editIds } = summary ; this . logger = logger ; this . editsPerChunk = editsPerChunk ; this . editChunks = new BTree < number , EditChunk < TChange > > ( undefined , compareFiniteNumbers ) ; editChunks . forEach ( ( editChunkOrHandle ) => { const { startRevision , chunk } = editChunkOrHandle ; if ( Array . isArray ( chunk ) ) { this . editChunks . set ( startRevision , { edits : chunk } ) ; } else { this . editChunks . set ( startRevision , { handle : chunk as EditHandle } ) ; } } ) ; this . sequencedEditIds = editIds . slice ( ) ; this . indexOfFirstEditInSession = this . numberOfSequencedEdits ; this . maximumEvictableIndex = this . indexOfFirstEditInSession - 1 ; this . sequencedEditIds . forEach ( ( id , index ) => { const encounteredEditId = this . allEditIds . get ( id ) ; assert ( encounteredEditId === undefined , 'Duplicate acked edit.' ) ; this . allEditIds . set ( id , { isLocal : false , index } ) ; } ) ; } /**
     * Registers a handler for when an edit is added to this `EditLog`.
     */ public registerEditAddedHandler ( handler ) { this . editAddedHandlers . push ( handler ) ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.length}
     */ public get length ( ) : number { return this . numberOfSequencedEdits + this . numberOfLocalEdits ; } /**
     * The number of sequenced (acked) edits in the log.
     */ public get numberOfSequencedEdits ( ) : number { return this . sequencedEditIds . length ; } /**
     * The number of local (unacked) edits in the log.
     */ public get numberOfLocalEdits ( ) : number { return this . localEdits . length ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.editIds}
     */ public get editIds ( ) : EditId [ ] { return this . sequencedEditIds . concat ( this . localEdits . map ( ( { id } ) => id ) ) ; } returns public isLocalEdit ( editId ) { const entry = this . allEditIds . get ( editId ) ; return entry !== undefined && entry . isLocal ; } returns public isSequencedRevision ( revision ) { return revision <= this . sequencedEditIds . length ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.tryGetIndexOfId}
     */ public tryGetIndexOfId ( editId ) { const orderedEdit = this . allEditIds . get ( editId ) ; if ( orderedEdit === undefined ) { return undefined ; } if ( orderedEdit . isLocal ) { const firstLocal = assertNotUndefined ( this . allEditIds . get ( this . localEdits [ 0 ] . id ) ) ; assert ( firstLocal . isLocal ) ; return this . numberOfSequencedEdits + orderedEdit . localSequence - firstLocal . localSequence ; } return orderedEdit . index ; } returns public getOrderedEditId ( editId ) { return assertNotUndefined ( this . allEditIds . get ( editId ) , 'All edits should exist in this map' ) ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.getIndexOfId}
     */ public getIndexOfId ( editId ) { return this . tryGetIndexOfId ( editId ) ?? fail ( 'edit not found' ) ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.idOf}
     */ public getIdAtIndex ( index ) { if ( this . numberOfSequencedEdits <= index ) { return this . localEdits [ index - this . numberOfSequencedEdits ] . id ; } return this . sequencedEditIds [ index ] ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.getAtIndex}
     */ public async getEditAtIndex ( index ) { if ( index < this . numberOfSequencedEdits ) { const [ startRevision , editChunk ] = assertNotUndefined ( this . editChunks . nextLowerPair ( index + 1 ) ) ; const { handle , edits } = editChunk ; if ( edits === undefined ) { assert ( handle !== undefined , 'An edit chunk should include at least a handle or edits.' ) ; const edits = JSON . parse ( IsoBuffer . from ( await handle . get ( ) ) . toString ( ) ) . edits as EditWithoutId < TChange > [ ] ; const nextKey = this . editChunks . nextHigherKey ( index ) ; const expectedEditLength = ( nextKey === undefined ? this . numberOfSequencedEdits : nextKey ) - startRevision ; assert ( edits . length === expectedEditLength , 'The chunk does not contain the correct number of edits.' ) ; editChunk . edits = edits ; this . addKeyToCache ( startRevision ) ; return joinEditAndId ( this . getIdAtIndex ( index ) , edits [ index - startRevision ] ) ; } return joinEditAndId ( this . getIdAtIndex ( index ) , edits [ index - startRevision ] ) ; } return this . localEdits [ index - this . numberOfSequencedEdits ] ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.getAtIndexSynchronous}
     */ public getEditInSessionAtIndex ( index ) { assert ( index > this . maximumEvictableIndex , 'Edit to retrieve must have been added to the log during the current session.' ) ; if ( index < this . numberOfSequencedEdits ) { const [ startRevision , editChunk ] = assertNotUndefined ( this . editChunks . nextLowerPair ( index + 1 ) ) ; const { edits } = editChunk ; return joinEditAndId ( this . getIdAtIndex ( index ) , assertNotUndefined ( edits , 'Edits should not have been evicted.' ) [ index - startRevision ] ) ; } assert ( index - this . numberOfSequencedEdits < this . localEdits . length , 'Edit to retrieve must be in the log.' ) ; return this . localEdits [ index - this . numberOfSequencedEdits ] ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.tryGetEdit}
     */ public async tryGetEdit ( editId ) { try { const index = this . getIndexOfId ( editId ) ; return await this . getEditAtIndex ( index ) ; } catch { return undefined ; } } returns public * getEditChunksReadyForUpload ( ) { const maxStartRevision = this . editChunks . maxKey ( ) ; if ( maxStartRevision === undefined ) { return ; } for ( const [ startRevision , chunk ] of this . editChunks . entries ( undefined , [ ] ) ) { if ( chunk . handle === undefined ) { const edits = assertNotUndefined ( chunk . edits ) ; if ( maxStartRevision !== startRevision || edits . length >= this . editsPerChunk ) { yield [ startRevision , edits ] ; } } } } /**
     * Assigns provided handles to edit chunks based on chunk index specified.
     */ public processEditChunkHandle ( chunkHandle , startRevision ) { const chunk = this . editChunks . get ( startRevision ) ; if ( chunk !== undefined ) { assertNotUndefined ( chunk . edits , 'A chunk handle op should not be received before the edit ops it corresponds to.' ) ; chunk . handle = chunkHandle ; this . addKeyToCache ( startRevision ) ; } else { this . logger ?. sendErrorEvent ( { eventName : 'UnexpectedHistoryChunk' } ) ; } } /**
     * Sequences all local edits.
     */ public sequenceLocalEdits ( ) { this . localEdits . slice ( ) . forEach ( ( edit ) => this . addSequencedEditInternal ( edit ) ) ; } /**
     * Adds a sequenced (non-local) edit to the edit log.
     * If the id of the supplied edit matches a local edit already present in the log, the local edit will be replaced.
     *
     */ public addSequencedEdit ( edit , message ) { this . addSequencedEditInternal ( edit , message , message . minimumSequenceNumber ) ; } /**
     * Adds a sequenced (non-local) edit to the edit log.
     * If the id of the supplied edit matches a local edit already present in the log, the local edit will be replaced.
     */ private addSequencedEditInternal ( edit , info ? , minSequenceNumber = 0 ) { const { id , editWithoutId } = separateEditAndId ( edit ) ; assert ( minSequenceNumber >= this . minSequenceNumber , 'Sequenced edits should carry a monotonically increasing min number' ) ; this . _minSequenceNumber = minSequenceNumber ; const encounteredEditId = this . allEditIds . get ( id ) ; if ( encounteredEditId !== undefined ) { assert ( encounteredEditId . isLocal , 'Duplicate acked edit.' ) ; const oldLocalEditId = assertNotUndefined ( this . localEdits . shift ( ) , 'Local edit should exist' ) . id ; assert ( oldLocalEditId === id , 'Causal ordering should be upheld' ) ; } const startRevision = this . numberOfSequencedEdits ; const edits = [ editWithoutId ] ; const lastPair = this . editChunks . nextLowerPair ( undefined ) ; if ( lastPair === undefined ) { this . editChunks . set ( startRevision , { edits } ) ; } else { const { edits : lastEditChunk , handle } = lastPair [ 1 ] ; if ( handle === undefined && lastEditChunk !== undefined && lastEditChunk . length < this . editsPerChunk ) { lastEditChunk . push ( editWithoutId ) ; } else { assert ( handle !== undefined || lastEditChunk !== undefined , 'An edit chunk must have either a handle or a list of edits.' ) ; this . editChunks . set ( startRevision , { edits } ) ; } } this . sequencedEditIds . push ( id ) ; const sequencedEditId = { index : this . numberOfSequencedEdits - 1 , isLocal : false , sequenceInfo : info } ; this . allEditIds . set ( id , sequencedEditId ) ; this . emitAdd ( edit , false , encounteredEditId !== undefined ) ; } returns public getLastEditChunk ( ) { return this . editChunks . nextLowerPair ( undefined ) ; } /**
     * Adds a non-sequenced (local) edit to the edit log.
     * Duplicate edits are ignored.
     */ public addLocalEdit ( edit ) { this . localEdits . push ( edit ) ; const localEditId = { localSequence : this . localEditSequence ++ , isLocal : true } ; this . allEditIds . set ( edit . id , localEditId ) ; this . emitAdd ( edit , true , false ) ; } private emitAdd ( editAdded , isLocal , wasLocal ) { for ( const handler of this . editAddedHandlers ) { handler ( editAdded , isLocal , wasLocal ) ; } } returns public equals < TOtherChangeTypes > ( other ) { return compareArrays ( this . editIds , other . editIds ) ; } /**
     * {@inheritDoc @intentional/shared-tree#OrderedEditSet.getEditLogSummary}
     */ public getEditLogSummary ( useHandles = false ) { if ( useHandles ) { return { editChunks : this . editChunks . toArray ( ) . map ( ( [ startRevision , { handle , edits } ] ) => { return { startRevision , chunk : handle ?? edits ?? fail ( 'An edit chunk must have either a handle or a list of edits.' ) } ; } ) , editIds : this . sequencedEditIds } ; } return { editChunks : this . editChunks . toArray ( ) . map ( ( [ startRevision , { edits } ] ) => { return { startRevision , chunk : assertNotUndefined ( edits ) } ; } ) , editIds : this . sequencedEditIds } ; } private addKeyToCache ( newKey ) { if ( newKey <= this . maximumEvictableIndex ) { if ( newKey in this . loadedChunkCache ) { this . loadedChunkCache . splice ( this . loadedChunkCache . indexOf ( newKey ) , 1 ) ; } this . loadedChunkCache . push ( newKey ) ; if ( this . loadedChunkCache . length > loadedChunkCacheSize ) { const indexToEvict = assertNotUndefined ( this . loadedChunkCache . shift ( ) ) ; const chunkToEvict = assertNotUndefined ( this . editChunks . get ( indexToEvict ) , 'Chunk start revision added to cache should exist in the edit log.' ) ; chunkToEvict . edits = undefined ; } } } }