import { ITelemetryLogger } from "@fluidframework/common-definitions" ; import { performance } from "@fluidframework/common-utils" ; import { ChildLogger , loggerToMonitoringContext , MonitoringContext , } from "@fluidframework/telemetry-utils" ; import { IDocumentDeltaConnection , IDocumentDeltaStorageService , IDocumentService , IResolvedUrl , IDocumentStorageService , IDocumentServicePolicies , DriverErrorType , } from "@fluidframework/driver-definitions" ; import { DeltaStreamConnectionForbiddenError , NonRetryableError } from "@fluidframework/driver-utils" ; import { IFacetCodes } from "@fluidframework/odsp-doclib-utils" ; import { IClient , ISequencedDocumentMessage , } from "@fluidframework/protocol-definitions" ; import { IOdspResolvedUrl , TokenFetchOptions , IEntry , HostStoragePolicy , InstrumentedStorageTokenFetcher , OdspErrorType , } from "@fluidframework/odsp-driver-definitions" ; import { HostStoragePolicyInternal , ISocketStorageDiscovery } from "./contracts" ; import { IOdspCache } from "./odspCache" ; import { OdspDeltaStorageService , OdspDeltaStorageWithCache } from "./odspDeltaStorageService" ; import { OdspDocumentDeltaConnection } from "./odspDocumentDeltaConnection" ; import { OdspDocumentStorageService } from "./odspDocumentStorageManager" ; import { getWithRetryForTokenRefresh , getOdspResolvedUrl , TokenFetchOptionsEx } from "./odspUtils" ; import { fetchJoinSession } from "./vroom" ; import { isOdcOrigin } from "./odspUrlHelper" ; import { EpochTracker } from "./epochTracker" ; import { OpsCache } from "./opsCaching" ; import { RetryErrorsStorageAdapter } from "./retryErrorsStorageAdapter" ; /**
 * The DocumentService manages the Socket.IO connection and manages routing requests to connected
 * clients
 */ export class OdspDocumentService implements IDocumentService { private _policies : IDocumentServicePolicies ; param resolvedUrl param getStorageToken param getWebsocketToken param logger param socketIoClientFactory param cache param hostPolicy param epochTracker param socketReferenceKeyPrefix public static async create ( resolvedUrl , getStorageToken , getWebsocketToken , logger , socketIoClientFactory , cache , hostPolicy , epochTracker , socketReferenceKeyPrefix ? ) { return new OdspDocumentService ( getOdspResolvedUrl ( resolvedUrl ) , getStorageToken , getWebsocketToken , logger , socketIoClientFactory , cache , hostPolicy , epochTracker , socketReferenceKeyPrefix ) ; } private storageManager ? : OdspDocumentStorageService ; private readonly mc : MonitoringContext ; private readonly joinSessionKey : string ; private readonly hostPolicy : HostStoragePolicyInternal ; private _opsCache ? : OpsCache ; private currentConnection ? : OdspDocumentDeltaConnection ; param odspResolvedUrl param getStorageToken param getWebsocketToken param logger param socketIoClientFactory param cache param hostPolicy param epochTracker param socketReferenceKeyPrefix private constructor ( public readonly odspResolvedUrl , private readonly getStorageToken , private readonly getWebsocketToken , logger , private readonly socketIoClientFactory , private readonly cache , hostPolicy , private readonly epochTracker , private readonly socketReferenceKeyPrefix ? ) { this . _policies = { storageOnly : odspResolvedUrl . fileVersion !== undefined } ; this . joinSessionKey = `${ this . odspResolvedUrl . hashedDocumentId }/joinsession` ; this . mc = loggerToMonitoringContext ( ChildLogger . create ( logger , undefined , { all : { odc : isOdcOrigin ( new URL ( this . odspResolvedUrl . endpoints . snapshotStorageUrl ) . origin ) } } ) ) ; this . hostPolicy = hostPolicy ; this . hostPolicy . fetchBinarySnapshotFormat ??= this . mc . config . getBoolean ( "Fluid.Driver.Odsp.binaryFormatSnapshot" ) ; if ( this . odspResolvedUrl . summarizer ) { this . hostPolicy = { ... this . hostPolicy , summarizerClient : true } ; } } public get resolvedUrl ( ) : IResolvedUrl { return this . odspResolvedUrl ; } public get policies ( ) { return this . _policies ; } returns public async connectToStorage ( ) { if ( ! this . storageManager ) { this . storageManager = new OdspDocumentStorageService ( this . odspResolvedUrl , this . getStorageToken , this . mc . logger , true , this . cache , this . hostPolicy , this . epochTracker , async ( ) => { if ( this . currentConnection !== undefined && ! this . currentConnection . disposed ) { return this . currentConnection . flush ( ) ; } throw new Error ( "Disconnected while uploading summary (attempt to perform flush())" ) ; } ) ; } return new RetryErrorsStorageAdapter ( this . storageManager , this . mc . logger ) ; } returns public async connectToDeltaStorage ( ) { const snapshotOps = this . storageManager ?. ops ?? [ ] ; const service = new OdspDeltaStorageService ( this . odspResolvedUrl . endpoints . deltaStorageUrl , this . getStorageToken , this . epochTracker , this . mc . logger ) ; const batchSize = this . hostPolicy . opsBatchSize ?? 5000 ; const concurrency = this . hostPolicy . concurrentOpsBatches ?? 1 ; return new OdspDeltaStorageWithCache ( snapshotOps , this . mc . logger , batchSize , concurrency , async ( from , to , telemetryProps , fetchReason ) => service . get ( from , to , telemetryProps , fetchReason ) , async ( from , to ) => { const res = await this . opsCache ?. get ( from , to ) ; return res as ISequencedDocumentMessage [ ] ?? [ ] ; } , ( from , to ) => { if ( this . currentConnection !== undefined && ! this . currentConnection . disposed ) { this . currentConnection . requestOps ( from , to ) ; } } , ( ops ) => this . opsReceived ( ops ) ) ; } returns public async connectToDeltaStream ( client ) { return getWithRetryForTokenRefresh < IDocumentDeltaConnection > ( async ( options ) => { const requestWebsocketTokenFromJoinSession = this . getWebsocketToken === undefined ; const websocketTokenPromise = requestWebsocketTokenFromJoinSession ? Promise . resolve ( null ) : this . getWebsocketToken ! ( options ) ; const joinSessionPromise = this . joinSession ( requestWebsocketTokenFromJoinSession , options ) . catch ( ( e ) => { const likelyFacetCodes = e as IFacetCodes ; if ( Array . isArray ( likelyFacetCodes . facetCodes ) ) { for ( const code of likelyFacetCodes . facetCodes ) { switch ( code ) { case "sessionForbiddenOnPreservedFiles" : case "sessionForbiddenOnModerationEnabledLibrary" : case "sessionForbiddenOnRequireCheckout" : this . _policies = { ... this . _policies , storageOnly : true } ; throw new DeltaStreamConnectionForbiddenError ( code ) ; default : continue ; } } } throw e ; } ) ; const [ websocketEndpoint , websocketToken , io ] = await Promise . all ( [ joinSessionPromise , websocketTokenPromise , this . socketIoClientFactory ( ) , ] ) ; const finalWebsocketToken = websocketToken ?? ( websocketEndpoint . socketToken || null ) ; if ( finalWebsocketToken === null ) { throw new NonRetryableError ( "pushTokenIsNull" , "Websocket token is null" , OdspErrorType . fetchTokenError ) ; } try { const connection = await this . connectToDeltaStreamWithRetry ( websocketEndpoint . tenantId , websocketEndpoint . id , finalWebsocketToken , io , client , websocketEndpoint . deltaStreamSocketUrl ) ; connection . on ( "op" , ( documentId , ops ) => { this . opsReceived ( ops ) ; } ) ; connection . on ( "disconnect" , ( error ) => { if ( typeof error === "object" && error !== null && error . errorType === DriverErrorType . authorizationError ) { this . cache . sessionJoinCache . remove ( this . joinSessionKey ) ; } } ) ; this . currentConnection = connection ; return connection ; } catch ( error ) { this . cache . sessionJoinCache . remove ( this . joinSessionKey ) ; if ( typeof error === "object" && error !== null ) { error . socketDocumentId = websocketEndpoint . id ; } throw error ; } } ) ; } private async joinSession ( requestSocketToken , options ) { const executeFetch = async ( ) => fetchJoinSession ( this . odspResolvedUrl , "opStream/joinSession" , "POST" , this . mc . logger , this . getStorageToken , this . epochTracker , requestSocketToken , options , this . hostPolicy . sessionOptions ?. unauthenticatedUserDisplayName ) ; return this . cache . sessionJoinCache . addOrGet ( this . joinSessionKey , executeFetch ) ; } param tenantId param documentId param token param io param client param webSocketUrl private async connectToDeltaStreamWithRetry ( tenantId , documentId , token , io , client , webSocketUrl ) { const startTime = performance . now ( ) ; const connection = await OdspDocumentDeltaConnection . create ( tenantId , documentId , token , io , client , webSocketUrl , this . mc . logger , 60000 , this . epochTracker , this . socketReferenceKeyPrefix ) ; const duration = performance . now ( ) - startTime ; if ( duration >= 2000 ) { this . mc . logger . sendPerformanceEvent ( { eventName : "ConnectionSuccess" , duration } ) ; } return connection ; } public dispose ( error ? ) { if ( error !== undefined ) { this . epochTracker . removeEntries ( ) . catch ( ( ) => { } ) ; } else { this . _opsCache ?. flushOps ( ) ; } this . _opsCache ?. dispose ( ) ; } protected get opsCache ( ) { if ( this . _opsCache ) { return this . _opsCache ; } const seqNumber = this . storageManager ?. snapshotSequenceNumber ; const batchSize = this . hostPolicy . opsCaching ?. batchSize ?? 100 ; if ( seqNumber === undefined || batchSize < 1 ) { return ; } const opsKey = { type : "ops" } ; this . _opsCache = new OpsCache ( seqNumber , this . mc . logger , { write : async ( key , opsData ) => { return this . cache . persistedCache . put ( { ... opsKey , key } , opsData ) ; } , read : async ( key ) => this . cache . persistedCache . get ( { ... opsKey , key } ) , remove : ( ) => { this . cache . persistedCache . removeEntries ( ) . catch ( ( ) => { } ) ; } } , batchSize , this . hostPolicy . opsCaching ?. timerGranularity ?? 5000 , this . hostPolicy . opsCaching ?. totalOpsToCache ?? 5000 ) ; return this . _opsCache ; } protected opsReceived ( ops ) { if ( ops . length === 0 || this . odspResolvedUrl . summarizer ) { return ; } this . opsCache ?. addOps ( ops ) ; } }