import { default as AbortController } from "abort-controller" ; import { v4 as uuid } from "uuid" ; import { ITelemetryLogger , ITelemetryProperties } from "@fluidframework/common-definitions" ; import { assert } from "@fluidframework/common-utils" ; import { ISequencedDocumentMessage } from "@fluidframework/protocol-definitions" ; import { InstrumentedStorageTokenFetcher } from "@fluidframework/odsp-driver-definitions" ; import { IDeltasFetchResult , IDocumentDeltaStorageService } from "@fluidframework/driver-definitions" ; import { requestOps , streamObserver , } from "@fluidframework/driver-utils" ; import { IDeltaStorageGetResponse , ISequencedDeltaOpMessage } from "./contracts" ; import { EpochTracker } from "./epochTracker" ; import { getWithRetryForTokenRefresh } from "./odspUtils" ; /**
 * Provides access to the underlying delta storage on the server for sharepoint driver.
 */ export class OdspDeltaStorageService { constructor ( private readonly deltaFeedUrl , private readonly getStorageToken , private readonly epochTracker , private readonly logger ) { } param from param to param telemetryProps returns public async get ( from , to , telemetryProps , fetchReason ? ) { return getWithRetryForTokenRefresh ( async ( options ) => { const baseUrl = this . buildUrl ( from , to ) ; const storageToken = await this . getStorageToken ( options , "DeltaStorage" ) ; const formBoundary = uuid ( ) ; let postBody = `--${ formBoundary }\r\n` ; postBody += `Authorization: Bearer ${ storageToken }\r\n` ; postBody += `X-HTTP-Method-Override: GET\r\n` ; if ( fetchReason !== undefined ) { postBody += `fetchReason: ${ fetchReason }\r\n` ; } postBody += `_post: 1\r\n` ; postBody += `\r\n--${ formBoundary }--` ; const headers = { "Content-Type" : `multipart/form-data;boundary=${ formBoundary }` } ; const abort = new AbortController ( ) ; const timer = setTimeout ( ( ) => abort . abort ( ) , 30000 ) ; const response = await this . epochTracker . fetchAndParseAsJSON < IDeltaStorageGetResponse > ( baseUrl , { headers , body : postBody , method : "POST" , signal : abort . signal } , "ops" , true ) ; clearTimeout ( timer ) ; const deltaStorageResponse = response . content ; let messages ; if ( deltaStorageResponse . value . length > 0 && "op" in deltaStorageResponse . value [ 0 ] ) { messages = ( deltaStorageResponse . value as ISequencedDeltaOpMessage [ ] ) . map ( ( operation ) => operation . op ) ; } else { messages = deltaStorageResponse . value as ISequencedDocumentMessage [ ] ; } this . logger . sendPerformanceEvent ( { eventName : "OpsFetch" , headers : Object . keys ( headers ) . length !== 0 ? true : undefined , length : messages . length , duration : response . duration , ... response . commonSpoHeaders , attempts : options . refresh ? 2 : 1 , from , to , ... telemetryProps } ) ; return { messages , partialResult : false } ; } ) ; } public buildUrl ( from , to ) { const filter = encodeURIComponent ( `sequenceNumber ge ${ from } and sequenceNumber le ${ to - 1 }` ) ; const queryString = `?ump=1&filter=${ filter }` ; return `${ this . deltaFeedUrl }${ queryString }` ; } } export class OdspDeltaStorageWithCache implements IDocumentDeltaStorageService { private firstCacheMiss = Number . MAX_SAFE_INTEGER ; public constructor ( private snapshotOps , private readonly logger , private readonly batchSize , private readonly concurrency , private readonly getFromStorage , private readonly getCached , private readonly requestFromSocket , private readonly opsReceived ) { } protected validateMessages ( reason , messages , from ) { if ( messages . length !== 0 ) { const start = messages [ 0 ] . sequenceNumber ; const length = messages . length ; const last = messages [ length - 1 ] . sequenceNumber ; if ( start !== from ) { this . logger . sendErrorEvent ( { eventName : "OpsFetchViolation" , reason , from , start , last , length } ) ; messages . length = 0 ; } if ( last + 1 !== from + length ) { this . logger . sendErrorEvent ( { eventName : "OpsFetchViolation" , reason , from , start , last , length } ) ; messages . length = 0 ; } } } public fetchMessages ( fromTotal , toTotal , abortSignal ? , cachedOnly ? , fetchReason ? ) { assert ( ! cachedOnly || toTotal === undefined , 0x1e3 ) ; let opsFromSnapshot = 0 ; let opsFromCache = 0 ; let opsFromStorage = 0 ; const requestCallback = async ( from , to , telemetryProps ) => { if ( this . snapshotOps !== undefined && this . snapshotOps . length !== 0 ) { const messages = this . snapshotOps . filter ( ( op ) => op . sequenceNumber >= from && op . sequenceNumber < to ) ; this . validateMessages ( "cached" , messages , from ) ; if ( messages . length > 0 && messages [ 0 ] . sequenceNumber === from ) { this . snapshotOps = this . snapshotOps . filter ( ( op ) => op . sequenceNumber >= to ) ; opsFromSnapshot = messages . length ; return { messages , partialResult : true } ; } this . snapshotOps = undefined ; } this . requestFromSocket ( from , to ) ; if ( from < this . firstCacheMiss ) { const messagesFromCache = await this . getCached ( from , to ) ; this . validateMessages ( "cached" , messagesFromCache , from ) ; if ( messagesFromCache . length !== 0 ) { opsFromCache += messagesFromCache . length ; return { messages : messagesFromCache , partialResult : true } ; } this . firstCacheMiss = Math . min ( this . firstCacheMiss , from ) ; } if ( cachedOnly ) { return { messages : [ ] , partialResult : false } ; } const ops = await this . getFromStorage ( from , to , telemetryProps , fetchReason ) ; this . validateMessages ( "storage" , ops . messages , from ) ; opsFromStorage += ops . messages . length ; this . opsReceived ( ops . messages ) ; return ops ; } ; const stream = requestOps ( async ( from , to , telemetryProps ) => { const result = await requestCallback ( from , to , telemetryProps ) ; this . validateMessages ( "catch all" , result . messages , from ) ; return result ; } , this . concurrency , fromTotal , toTotal , this . batchSize , this . logger , abortSignal , fetchReason ) ; return streamObserver ( stream , ( result ) => { if ( result . done && opsFromSnapshot + opsFromCache + opsFromStorage !== 0 ) { this . logger . sendPerformanceEvent ( { eventName : "CacheOpsRetrieved" , opsFromSnapshot , opsFromCache , opsFromStorage } ) ; } } ) ; } }