import { IContext , IQueuedMessage , ILogger , IContextErrorData , IRoutingKey , } from "@fluidframework/server-services-core" ; import { TestKafka , DebugLogger } from "@fluidframework/server-test-utils" ; import { strict as assert } from "assert" ; import { DocumentContextManager } from "../../document-router/contextManager" ; class TestContext implements IContext { public offset = - 1 ; public checkpoint ( queuedMessage ) { assert ( queuedMessage . offset >= this . offset , `${ queuedMessage . offset } >= ${ this . offset }` ) ; this . offset = queuedMessage . offset ; } public error ( error , errorData ) { throw new Error ( "Method not implemented." ) ; } public readonly log : ILogger = DebugLogger . create ( "fluid-server:TestContextManager" ) ; } describe ( "document-router" , ( ) => { describe ( "DocumentContextManager" , ( ) => { let testContext ; let testContextManager ; let offset0 , offset5 , offset10 , offset12 , offset15 , offset20 , offset25 ; let routingKey ; beforeEach ( async ( ) => { testContext = new TestContext ( ) ; testContextManager = new DocumentContextManager ( testContext ) ; routingKey = { tenantId : "test-tenant-id" , documentId : "test-document-id" } ; offset0 = TestKafka . createdQueuedMessage ( 0 ) ; offset5 = TestKafka . createdQueuedMessage ( 5 ) ; offset10 = TestKafka . createdQueuedMessage ( 10 ) ; offset12 = TestKafka . createdQueuedMessage ( 12 ) ; offset15 = TestKafka . createdQueuedMessage ( 15 ) ; offset20 = TestKafka . createdQueuedMessage ( 20 ) ; offset25 = TestKafka . createdQueuedMessage ( 25 ) ; } ) ; describe ( ".createContext" , ( ) => { it ( "Should be able to create and then track a document context" , ( ) => { testContextManager . setHead ( offset0 ) ; const context = testContextManager . createContext ( routingKey , offset0 ) ; testContextManager . setTail ( offset0 ) ; testContextManager . setHead ( offset5 ) ; context . setHead ( offset5 ) ; context . checkpoint ( offset5 ) ; testContextManager . setTail ( offset5 ) ; assert . equal ( testContext . offset , 5 ) ; } ) ; it ( "Should be able to create and track multiple document contexts" , ( ) => { testContextManager . setHead ( offset0 ) ; const context0 = testContextManager . createContext ( routingKey , offset0 ) ; testContextManager . setTail ( offset0 ) ; testContextManager . setHead ( offset5 ) ; const context1 = testContextManager . createContext ( routingKey , offset5 ) ; testContextManager . setTail ( offset5 ) ; testContextManager . setHead ( offset10 ) ; const context2 = testContextManager . createContext ( routingKey , offset10 ) ; testContextManager . setTail ( offset10 ) ; assert . equal ( testContext . offset , - 1 ) ; testContextManager . setHead ( offset12 ) ; context0 . setHead ( offset12 ) ; context0 . checkpoint ( offset0 ) ; testContextManager . setTail ( offset12 ) ; assert . equal ( testContext . offset , 0 ) ; testContextManager . setHead ( offset15 ) ; context1 . setHead ( offset15 ) ; context1 . checkpoint ( offset5 ) ; testContextManager . setTail ( offset15 ) ; assert . equal ( testContext . offset , 0 ) ; context2 . checkpoint ( offset10 ) ; assert . equal ( testContext . offset , 0 ) ; context0 . checkpoint ( offset12 ) ; assert . equal ( testContext . offset , 5 ) ; testContextManager . setHead ( offset20 ) ; testContextManager . setTail ( offset20 ) ; assert . equal ( testContext . offset , 5 ) ; context1 . checkpoint ( offset15 ) ; assert . equal ( testContext . offset , 20 ) ; } ) ; it ( "Should correctly compute the checkpointed offset after contexts switch pending work state" , ( ) => { testContextManager . setHead ( offset0 ) ; const context = testContextManager . createContext ( routingKey , offset0 ) ; testContextManager . setTail ( offset0 ) ; testContextManager . setHead ( offset12 ) ; context . setHead ( offset12 ) ; context . checkpoint ( offset12 ) ; testContextManager . setTail ( offset12 ) ; testContextManager . setHead ( offset20 ) ; testContextManager . setTail ( offset20 ) ; assert . equal ( testContext . offset , 20 ) ; testContextManager . setHead ( offset25 ) ; context . setHead ( offset25 ) ; testContextManager . setTail ( offset25 ) ; assert . equal ( testContext . offset , 20 ) ; } ) ; it ( "Should ignore contexts without pending work" , ( ) => { testContextManager . setHead ( offset0 ) ; const context = testContextManager . createContext ( routingKey , offset0 ) ; context . checkpoint ( offset0 ) ; testContextManager . setTail ( offset0 ) ; testContextManager . setHead ( offset5 ) ; testContextManager . setTail ( offset5 ) ; assert . equal ( testContext . offset , 5 ) ; } ) ; it ( "Should not checkpoint until the starting offset changes" , ( ) => { testContextManager . setHead ( offset0 ) ; testContextManager . createContext ( routingKey , offset0 ) ; testContextManager . setTail ( offset0 ) ; assert . equal ( testContext . offset , - 1 ) ; testContextManager . setHead ( offset5 ) ; testContextManager . setTail ( offset5 ) ; assert . equal ( testContext . offset , - 1 ) ; } ) ; it ( "Should emit an error if a created context emits an error" , async ( ) => { testContextManager . setHead ( offset0 ) ; const context = testContextManager . createContext ( routingKey , offset0 ) ; testContextManager . setTail ( offset0 ) ; return new Promise < void > ( ( resolve , reject ) => { testContextManager . on ( "error" , ( error , errorData ) => { assert . ok ( error ) ; assert . ok ( errorData . restart ) ; resolve ( ) ; } ) ; context . error ( "Test Error" , { restart : true } ) ; } ) ; } ) ; } ) ; } ) ; } ) ;