"use strict";
/*!
 * Copyright (c) Microsoft Corporation and contributors. All rights reserved.
 * Licensed under the MIT License.
 */
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
/* eslint-disable @typescript-eslint/consistent-type-assertions */
const node_assert_1 = require("node:assert");
const test_pairwise_generator_1 = require("@fluid-private/test-pairwise-generator");
const internal_1 = require("@fluidframework/container-definitions/internal");
const internal_2 = require("@fluidframework/driver-definitions/internal");
const internal_3 = require("@fluidframework/telemetry-utils/internal");
const sinon_1 = __importDefault(require("sinon"));
const messageTypes_js_1 = require("../messageTypes.js");
const index_js_1 = require("../opLifecycle/index.js");
const pendingStateManager_js_1 = require("../pendingStateManager.js");
// Make a mock op with distinguishable contents
function op(data = "") {
    return {
        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
        contents: data,
    };
}
function withBatchMetadata(messages, batchId) {
    return (0, index_js_1.addBatchMetadata)({ messages, referenceSequenceNumber: -1 }, batchId).messages;
}
const testAddressedDataStoreMessage = {
    address: "test-address",
    contents: {
        type: "op",
        content: "test-content",
    },
};
describe("Pending State Manager", () => {
    const sandbox = sinon_1.default.createSandbox();
    function getStateHandlerStub() {
        const stubs = {
            applyStashedOp: sandbox.stub(),
            clientId: sandbox.stub(),
            connected: sandbox.stub(),
            reSubmitBatch: sandbox.stub(),
            isActiveConnection: sandbox.stub(),
            isAttached: sandbox.stub(),
        };
        stubs.applyStashedOp.resolves(undefined);
        stubs.clientId.returns("clientId");
        stubs.connected.returns(true);
        stubs.isActiveConnection.returns(true);
        stubs.isAttached.returns(true);
        return stubs;
    }
    const mockLogger = new internal_3.MockLogger();
    const logger = (0, internal_3.createChildLogger)({ logger: mockLogger });
    const opGroupingManager = new index_js_1.OpGroupingManager({ groupedBatchingEnabled: true }, logger);
    function newPendingStateManager(stubs, stashedLocalState) {
        return new pendingStateManager_js_1.PendingStateManager(stubs, stashedLocalState, logger);
    }
    afterEach("Sinon sandbox restore", () => {
        sandbox.restore();
    });
    afterEach("ThrowOnErrorLogs", () => {
        // Note: If mockLogger is used within a test,
        // it may inadvertently clear errors such that they're not noticed here
        mockLogger.assertNoErrors();
        mockLogger.clear();
    });
    describe("Rollback", () => {
        let rollbackCalled;
        let rollbackContent;
        let rollbackShouldThrow;
        let batchManager;
        function getMessage(payload) {
            return {
                runtimeOp: op(payload),
            };
        }
        const rollBackCallback = (m) => {
            rollbackCalled = true;
            rollbackContent.push(m);
            if (rollbackShouldThrow) {
                throw new Error("test error");
            }
        };
        beforeEach(async () => {
            rollbackCalled = false;
            rollbackContent = [];
            rollbackShouldThrow = false;
            batchManager = new index_js_1.BatchManager({ disableGroupedBatching: false });
        });
        it("should do nothing when rolling back empty pending stack", () => {
            const checkpoint = batchManager.checkpoint();
            checkpoint.rollback(rollBackCallback);
            node_assert_1.strict.strictEqual(rollbackCalled, false);
            node_assert_1.strict.strictEqual(batchManager.empty, true);
        });
        it("should do nothing when rolling back nothing", () => {
            batchManager.push(getMessage("1"), /* reentrant */ false);
            const checkpoint = batchManager.checkpoint();
            checkpoint.rollback(rollBackCallback);
            node_assert_1.strict.strictEqual(rollbackCalled, false);
            node_assert_1.strict.strictEqual(batchManager.empty, false);
        });
        it("should succeed when rolling back entire pending stack", () => {
            const checkpoint = batchManager.checkpoint();
            batchManager.push(getMessage("11"), /* reentrant */ false);
            batchManager.push(getMessage("22"), /* reentrant */ false);
            batchManager.push(getMessage("33"), /* reentrant */ false);
            checkpoint.rollback(rollBackCallback);
            node_assert_1.strict.strictEqual(rollbackCalled, true);
            node_assert_1.strict.strictEqual(rollbackContent.length, 3);
            node_assert_1.strict.strictEqual(rollbackContent[0].runtimeOp.contents, "33");
            node_assert_1.strict.strictEqual(rollbackContent[1].runtimeOp.contents, "22");
            node_assert_1.strict.strictEqual(rollbackContent[2].runtimeOp.contents, "11");
            node_assert_1.strict.strictEqual(batchManager.empty, true);
        });
        it("should succeed when rolling back part of pending stack", () => {
            batchManager.push(getMessage("11"), /* reentrant */ false);
            const checkpoint = batchManager.checkpoint();
            batchManager.push(getMessage("22"), /* reentrant */ false);
            batchManager.push(getMessage("33"), /* reentrant */ false);
            checkpoint.rollback(rollBackCallback);
            node_assert_1.strict.strictEqual(rollbackCalled, true);
            node_assert_1.strict.strictEqual(rollbackContent.length, 2);
            node_assert_1.strict.strictEqual(rollbackContent[0].runtimeOp.contents, "33");
            node_assert_1.strict.strictEqual(rollbackContent[1].runtimeOp.contents, "22");
            node_assert_1.strict.strictEqual(batchManager.empty, false);
        });
        it("should throw and close when rollback fails", () => {
            rollbackShouldThrow = true;
            const checkpoint = batchManager.checkpoint();
            batchManager.push(getMessage("11"), /* reentrant */ false);
            node_assert_1.strict.throws(() => {
                checkpoint.rollback(rollBackCallback);
            });
            node_assert_1.strict.strictEqual(rollbackCalled, true);
        });
    });
    describe("Op processing", () => {
        let pendingStateManager;
        const clientId = "clientId";
        beforeEach(async () => {
            pendingStateManager = new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: () => {
                    throw new Error("test error");
                },
                clientId: () => "oldClientId",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, undefined /* initialLocalState */, logger);
        });
        const submitBatch = (messages, clientSequenceNumber, localOpMetadata) => {
            pendingStateManager.onFlushBatch(messages.map((message) => ({
                runtimeOp: {
                    type: message.type,
                    contents: message.contents,
                },
                referenceSequenceNumber: message.referenceSequenceNumber, // eslint-disable-line @typescript-eslint/no-non-null-assertion
                metadata: message.metadata,
                localOpMetadata,
            })), clientSequenceNumber ?? messages[0]?.clientSequenceNumber, false /* staged */);
        };
        const processFullBatch = (messages, batchStartCsn, groupedBatch, emptyBatchSequenceNumber, resubmittedBatchId) => pendingStateManager.processInboundMessages({
            type: "fullBatch",
            messages: messages,
            batchStart: {
                batchStartCsn,
                keyMessage: {
                    sequenceNumber: emptyBatchSequenceNumber,
                },
                clientId,
                batchId: resubmittedBatchId,
            },
            length: messages.length,
            groupedBatch,
        }, true /* local */);
        it("Grouped batch is processed correctly", () => {
            const messages = [
                {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 0,
                    referenceSequenceNumber: 0,
                    metadata: { batch: true },
                },
                {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 1,
                    referenceSequenceNumber: 0,
                },
                {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    metadata: { batch: false },
                    clientSequenceNumber: 2,
                    referenceSequenceNumber: 0,
                },
            ];
            submitBatch(messages);
            processFullBatch(messages, 0 /* batchStartCsn */, true /* groupedBatch */);
        });
        it("Ungrouped batch is processed correctly", () => {
            const messages = [
                {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 0,
                    referenceSequenceNumber: 0,
                    metadata: { batch: true },
                },
                {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 1,
                    referenceSequenceNumber: 0,
                },
                {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    metadata: { batch: false },
                    clientSequenceNumber: 2,
                    referenceSequenceNumber: 0,
                },
            ];
            submitBatch(messages);
            pendingStateManager.processInboundMessages({
                type: "batchStartingMessage",
                nextMessage: messages[0],
                batchStart: {
                    batchStartCsn: 0,
                    keyMessage: messages[0],
                    clientId,
                    batchId: undefined,
                },
            }, true /* local */);
            pendingStateManager.processInboundMessages({
                type: "nextBatchMessage",
                nextMessage: messages[1],
            }, true /* local */);
            pendingStateManager.processInboundMessages({
                type: "nextBatchMessage",
                nextMessage: messages[2],
                batchEnd: true,
            }, true /* local */);
        });
        it("empty batch is processed correctly", () => {
            const { placeholderMessage } = opGroupingManager.createEmptyGroupedBatch("batchId", 0);
            pendingStateManager.onFlushEmptyBatch(placeholderMessage, 1 /* clientSequenceNumber */, false /* staged */);
            // A groupedBatch is supposed to have nested messages inside its contents,
            // but an empty batch has no nested messages. When processing en empty grouped batch,
            // the psm will expect the next pending message to be an "empty" message as portrayed above.
            processFullBatch([], 1 /* batchStartCsn */, true /* groupedBatch */, 3 /* emptyBatchSequenceNumber */, "batchId" /* resubmittedBatchId */);
        });
        describe("processing out of sync messages will throw and log", () => {
            it("messageTypes do not match", () => {
                const messages = [
                    {
                        clientId,
                        type: internal_2.MessageType.Operation,
                        clientSequenceNumber: 0,
                        referenceSequenceNumber: 0,
                    },
                ];
                submitBatch(messages);
                node_assert_1.strict.throws(() => processFullBatch(messages.map((message) => ({
                    ...message,
                    type: "otherType",
                })), 0 /* batchStartCsn */, false /* groupedBatch */), (closeError) => closeError.errorType === internal_1.ContainerErrorTypes.dataProcessingError);
                mockLogger.assertMatch([
                    {
                        eventName: "unexpectedAckReceived",
                        pendingContentScrubbed: JSON.stringify({ type: "op" }),
                        incomingContentScrubbed: JSON.stringify({ type: "otherType" }),
                        contentsMatch: true,
                        pendingLength: 13,
                        incomingLength: 20,
                        mismatchStartIndex: 10,
                        pendingChar: "p",
                        incomingChar: "t",
                    },
                ], "Expected to log scrubbed messages", true /* inlineDetailsProp */);
            });
            it("only one message has undefined content", () => {
                const messages = [
                    {
                        clientId,
                        type: internal_2.MessageType.Operation,
                        clientSequenceNumber: 0,
                        referenceSequenceNumber: 0,
                        contents: {},
                    },
                ];
                submitBatch(messages);
                node_assert_1.strict.throws(() => processFullBatch(messages.map((message) => ({
                    ...message,
                    contents: undefined,
                })), 0 /* batchStartCsn */, false /* groupedBatch */), (closeError) => closeError.errorType === internal_1.ContainerErrorTypes.dataProcessingError);
                mockLogger.assertMatch([
                    {
                        eventName: "unexpectedAckReceived",
                        pendingContentScrubbed: JSON.stringify({ type: "op", contents: {} }),
                        incomingContentScrubbed: JSON.stringify({ type: "op" }),
                        contentsMatch: false,
                        pendingLength: 27,
                        incomingLength: 13,
                        mismatchStartIndex: 12,
                        pendingChar: ",",
                        incomingChar: "}",
                    },
                ], "Expected to log scrubbed messages", true /* inlineDetailsProp */);
            });
            it("stringified message content does not match", () => {
                const messages = [
                    {
                        clientId,
                        type: internal_2.MessageType.Operation,
                        clientSequenceNumber: 0,
                        referenceSequenceNumber: 0,
                        contents: {},
                    },
                ];
                submitBatch(messages);
                node_assert_1.strict.throws(() => processFullBatch(messages.map((message) => ({
                    ...message,
                    contents: { prop1: true },
                })), 0 /* batchStartCsn */, false /* groupedBatch */), (closeError) => closeError.errorType === internal_1.ContainerErrorTypes.dataProcessingError);
                mockLogger.assertMatch([
                    {
                        eventName: "unexpectedAckReceived",
                        pendingContentScrubbed: JSON.stringify({ type: "op", contents: {} }),
                        incomingContentScrubbed: JSON.stringify({
                            type: "op",
                            contents: { prop1: "boolean" },
                        }),
                        contentsMatch: false,
                        pendingLength: 27,
                        incomingLength: 39,
                        mismatchStartIndex: 25,
                        pendingChar: "}",
                        incomingChar: '"',
                    },
                ], "Expected to log scrubbed messages", true /* inlineDetailsProp */);
            });
            it("stringified message content out of order", () => {
                const message = {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 0,
                    referenceSequenceNumber: 0,
                    contents: {},
                };
                // contents and type are swapped in the stringified message relative to what we typically do/expect
                pendingStateManager.onFlushBatch([
                    {
                        runtimeOp: {
                            contents: message.contents,
                            type: message.type,
                        },
                        referenceSequenceNumber: 0,
                    },
                ], 0 /* clientSequenceNumber */, false /* staged */);
                node_assert_1.strict.throws(() => processFullBatch([message], 0 /* batchStartCsn */, false /* groupedBatch */), (closeError) => closeError.errorType === internal_1.ContainerErrorTypes.dataProcessingError);
                mockLogger.assertMatch([
                    {
                        eventName: "unexpectedAckReceived",
                        pendingContentScrubbed: JSON.stringify({ contents: {}, type: "op" }),
                        incomingContentScrubbed: JSON.stringify({ type: "op", contents: {} }),
                        contentsMatch: true,
                        pendingLength: 27,
                        incomingLength: 27,
                        mismatchStartIndex: 2,
                        pendingChar: "c",
                        incomingChar: "t",
                    },
                ], "Expected to log scrubbed messages", true /* inlineDetailsProp */);
            });
            it("stringified message content with unexpected keys", () => {
                const message = {
                    clientId,
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 0,
                    referenceSequenceNumber: 0,
                    contents: {},
                };
                // contents and type are swapped in the stringified message relative to what we typically do/expect
                pendingStateManager.onFlushBatch([
                    {
                        runtimeOp: {
                            type: message.type,
                            contents: message.contents,
                            somethingElse: 123, // Unexpected key
                        },
                        referenceSequenceNumber: 0,
                    },
                ], 0 /* clientSequenceNumber */, false /* staged */);
                node_assert_1.strict.throws(() => processFullBatch([message], 0 /* batchStartCsn */, false /* groupedBatch */), (closeError) => closeError.errorType === internal_1.ContainerErrorTypes.dataProcessingError);
                mockLogger.assertMatch([
                    {
                        eventName: "unexpectedAckReceived",
                        pendingContentScrubbed: JSON.stringify({
                            type: "op",
                            contents: {},
                            somethingElse: "number",
                        }),
                        incomingContentScrubbed: JSON.stringify({ type: "op", contents: {} }),
                        contentsMatch: true,
                        pendingLength: 47,
                        incomingLength: 27,
                        mismatchStartIndex: 26,
                        pendingChar: ",",
                        incomingChar: "}",
                    },
                ], "Expected to log scrubbed messages", true /* inlineDetailsProp */);
            });
            it("processing in sync messages will not throw", () => {
                const messages = [
                    {
                        clientId,
                        type: internal_2.MessageType.Operation,
                        clientSequenceNumber: 0,
                        referenceSequenceNumber: 0,
                        contents: { prop1: true },
                    },
                ];
                submitBatch(messages);
                processFullBatch(messages.map((message) => ({
                    ...message,
                    contents: { prop1: true },
                })), 0 /* batchStartCsn */, false /* groupedBatch */);
            });
            it("findFirstCharacterMismatched", () => {
                const nonAsciiChar = String.fromCodePoint(0x80);
                const testCases = [
                    { input: ["", ""], expected: [-1] },
                    { input: ["", "b"], expected: [0, undefined, "b"] },
                    { input: ["a", "b"], expected: [0, "a", "b"] },
                    { input: ["xyz", "xxx"], expected: [1, "y", "x"] },
                    { input: ["xyz", "xy"], expected: [2, "z", undefined] },
                    { input: ["xy", "xxx"], expected: [1, "y", "x"] },
                    { input: ["xyz", "xyz"], expected: [-1] },
                    { input: ["QQ", `Q${nonAsciiChar}`], expected: [1, "Q", "[non-ASCII]"] },
                ];
                for (const { input: [a, b], expected: [i, charA, charB], } of testCases) {
                    node_assert_1.strict.deepEqual((0, pendingStateManager_js_1.findFirstCharacterMismatched)(a, b), [i, charA, charB], `Failed input: "${a}", "${b}"`);
                    node_assert_1.strict.deepEqual((0, pendingStateManager_js_1.findFirstCharacterMismatched)(b, a), [i, charB, charA], `Failed input: "${b}", "${a}"`);
                }
            });
        });
        describe("getLocalState", () => {
            it("removes ops with seq num lower than snapshot", () => {
                const messages = Array.from({ length: 10 }, (_, i) => ({
                    clientId: "clientId",
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 0,
                    contents: { prop1: true },
                    sequenceNumber: i + 1, // starting with sequence number 1 so first assert does not filter any op
                }));
                submitBatch(messages);
                processFullBatch(messages, 0 /* batchStartCsn */, false /* groupedBatch */);
                let pendingState = pendingStateManager.getLocalState(0).pending.pendingStates;
                node_assert_1.strict.strictEqual(pendingState.length, 10);
                pendingState = pendingStateManager.getLocalState(5).pending.pendingStates;
                node_assert_1.strict.strictEqual(pendingState.length, 5);
                pendingState = pendingStateManager.getLocalState(10).pending.pendingStates;
                node_assert_1.strict.strictEqual(pendingState.length, 0);
            });
            it("throws when trying to get unprocessed ops older than snapshot", () => {
                const messages = Array.from({ length: 10 }, (_, i) => ({
                    clientId: "clientId",
                    type: internal_2.MessageType.Operation,
                    clientSequenceNumber: 0,
                    contents: { prop1: true },
                    referenceSequenceNumber: i,
                }));
                submitBatch(messages);
                node_assert_1.strict.throws(() => pendingStateManager.getLocalState(1));
                const pendingState = pendingStateManager.getLocalState(0).pending.pendingStates;
                node_assert_1.strict.strictEqual(pendingState.length, 10);
            });
        });
    });
    describe("Local state processing", () => {
        function createPendingStateManager(pendingStates) {
            return new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: async () => undefined,
                clientId: () => "CLIENT_ID",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, pendingStates ? { pendingStates } : undefined, logger);
        }
        describe("Constructor pendingStates", () => {
            it("Empty local state", () => {
                {
                    const pendingStateManager = createPendingStateManager(undefined);
                    node_assert_1.strict.deepStrictEqual(pendingStateManager.initialMessages.toArray(), []);
                }
                {
                    const pendingStateManager = createPendingStateManager([]);
                    node_assert_1.strict.deepStrictEqual(pendingStateManager.initialMessages.toArray(), []);
                }
            });
            it("New format", () => {
                const messages = [
                    {
                        type: "message",
                        content: '{"type":"component"}',
                        referenceSequenceNumber: 10,
                    },
                    {
                        type: "message",
                        content: '{"type": "component", "contents": {"prop1": "value"}}',
                        referenceSequenceNumber: 10,
                    },
                ];
                const pendingStateManager = createPendingStateManager(messages);
                node_assert_1.strict.deepStrictEqual(pendingStateManager.initialMessages.toArray(), messages);
            });
        });
    });
    describe("applyStashedOpsAt", () => {
        it("applyStashedOpsAt", async () => {
            const applyStashedOps = [];
            const messages = [
                {
                    type: "message",
                    content: '{"type":"component"}',
                    referenceSequenceNumber: 10,
                    localOpMetadata: undefined,
                    opMetadata: undefined,
                    batchInfo: { clientId: "CLIENT_ID", batchStartCsn: 1, length: 1, staged: false },
                    runtimeOp: undefined,
                },
                {
                    type: "message",
                    content: '{"type": "component", "contents": {"prop1": "value"}}',
                    referenceSequenceNumber: 11,
                    localOpMetadata: undefined,
                    opMetadata: undefined,
                    batchInfo: { clientId: "CLIENT_ID", batchStartCsn: 2, length: 1, staged: false },
                    runtimeOp: undefined,
                },
            ];
            const pendingStateManager = new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: async (content) => applyStashedOps.push(content),
                clientId: () => "clientId",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, { pendingStates: messages }, logger);
            await pendingStateManager.applyStashedOpsAt();
            node_assert_1.strict.strictEqual(applyStashedOps.length, 2);
            node_assert_1.strict.strictEqual(pendingStateManager.pendingMessagesCount, 2);
        });
        it("applyStashedOpsAt for empty batch", async () => {
            const oldPsm = new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: async () => undefined,
                clientId: () => "1",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, undefined /* initialLocalState */, logger);
            const { placeholderMessage } = opGroupingManager.createEmptyGroupedBatch("batchId", 0);
            oldPsm.onFlushEmptyBatch(placeholderMessage, 0, false /* staged */);
            const localStateWithEmptyBatch = oldPsm.getLocalState(0).pending;
            const applyStashedOps = [];
            const pendingStateManager = new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: async (content) => applyStashedOps.push(content),
                clientId: () => "clientId",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, localStateWithEmptyBatch, logger);
            await pendingStateManager.applyStashedOpsAt();
            node_assert_1.strict.strictEqual(applyStashedOps.length, 0);
            node_assert_1.strict.strictEqual(pendingStateManager.pendingMessagesCount, 1);
        });
    });
    describe("Pending messages state", () => {
        const forInitialMessages = [
            { type: "message", content: '{"type":"component"}', referenceSequenceNumber: 10 },
            {
                type: "message",
                content: '{"type": "component", "contents": {"prop1": "value"}}',
                referenceSequenceNumber: 10,
            },
        ];
        const forFlushedMessages = [
            {
                runtimeOp: { type: "component" },
                referenceSequenceNumber: 10,
            },
            {
                runtimeOp: { type: "component" },
                referenceSequenceNumber: 10,
            },
        ];
        function createPendingStateManager(pendingStates) {
            return new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: async () => undefined,
                clientId: () => "CLIENT_ID",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, pendingStates ? { pendingStates } : undefined, logger);
        }
        it("no pending or initial messages", () => {
            const pendingStateManager = createPendingStateManager(undefined);
            node_assert_1.strict.strictEqual(pendingStateManager.hasPendingMessages(), false, "There shouldn't be pending messages");
            node_assert_1.strict.strictEqual(pendingStateManager.pendingMessagesCount, 0, "Pending messages count should be 0");
        });
        it("has pending messages but no initial messages", () => {
            const pendingStateManager = createPendingStateManager(undefined);
            // let each message be its own batch
            for (const message of forFlushedMessages) {
                pendingStateManager.onFlushBatch([
                    {
                        runtimeOp: message.runtimeOp,
                        referenceSequenceNumber: message.referenceSequenceNumber,
                    },
                ], 0, false /* staged */);
            }
            node_assert_1.strict.strictEqual(pendingStateManager.hasPendingMessages(), true, "There should be pending messages");
            node_assert_1.strict.strictEqual(pendingStateManager.pendingMessagesCount, forFlushedMessages.length, "Pending messages count should be same as pending messages");
        });
        it("has initial messages but no pending messages", () => {
            const pendingStateManager = createPendingStateManager(forInitialMessages);
            node_assert_1.strict.strictEqual(pendingStateManager.hasPendingMessages(), true, "There should be initial messages");
            node_assert_1.strict.strictEqual(pendingStateManager.pendingMessagesCount, forFlushedMessages.length, "Pending messages count should be same as initial messages");
        });
        it("has both pending messages and initial messages", () => {
            const pendingStateManager = createPendingStateManager(forInitialMessages);
            // let each message be its own batch
            for (const message of forFlushedMessages) {
                pendingStateManager.onFlushBatch([
                    {
                        runtimeOp: message.runtimeOp,
                        referenceSequenceNumber: message.referenceSequenceNumber,
                    },
                ], 0, false /* staged */);
            }
            node_assert_1.strict.strictEqual(pendingStateManager.hasPendingMessages(), true, "There should be pending messages");
            node_assert_1.strict.strictEqual(pendingStateManager.pendingMessagesCount, forInitialMessages.length + forFlushedMessages.length, "Pending messages count should be same as pending + initial messages");
        });
    });
    describe("Minimum sequence number", () => {
        const forInitialMessages = [
            {
                type: "message",
                content: '{"type":"component"}',
                referenceSequenceNumber: 10,
                localOpMetadata: undefined,
                opMetadata: undefined,
                batchInfo: { clientId: "CLIENT_ID", batchStartCsn: 1, length: 1, staged: false },
                runtimeOp: undefined,
            },
            {
                type: "message",
                content: '{"type": "component", "contents": {"prop1": "value"}}',
                referenceSequenceNumber: 11,
                localOpMetadata: undefined,
                opMetadata: undefined,
                batchInfo: { clientId: "CLIENT_ID", batchStartCsn: 2, length: 1, staged: false },
                runtimeOp: undefined,
            },
            {
                type: "message",
                content: '{"type": "component", "contents": {"prop2": "value"}}',
                referenceSequenceNumber: 12,
                localOpMetadata: undefined,
                opMetadata: undefined,
                batchInfo: { clientId: "CLIENT_ID", batchStartCsn: 3, length: 1, staged: false },
                runtimeOp: undefined,
            },
            {
                type: "message",
                content: '{"type": "component", "contents": {"prop3": "value"}}',
                referenceSequenceNumber: 12,
                localOpMetadata: undefined,
                opMetadata: undefined,
                batchInfo: { clientId: "CLIENT_ID", batchStartCsn: 4, length: 1, staged: false },
                runtimeOp: undefined,
            },
        ];
        const forFlushedMessages = forInitialMessages.map((message) => ({
            runtimeOp: JSON.parse(message.content),
            referenceSequenceNumber: message.referenceSequenceNumber,
        }));
        function createPendingStateManager(pendingStates) {
            return new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: async () => undefined,
                clientId: () => "123",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, pendingStates ? { pendingStates } : undefined /* initialLocalState */, logger);
        }
        it("minimum sequence number can be retrieved from initial messages", async () => {
            const pendingStateManager = createPendingStateManager(forInitialMessages);
            await pendingStateManager.applyStashedOpsAt();
            node_assert_1.strict.strictEqual(pendingStateManager.minimumPendingMessageSequenceNumber, 10, "minimum sequence number should be the first message");
            pendingStateManager.replayPendingStates();
            node_assert_1.strict.strictEqual(pendingStateManager.minimumPendingMessageSequenceNumber, undefined, "Should have processed messages and thus have no min seq number");
        });
        it("minimum sequence number can be retrieved from pending messages", async () => {
            const pendingStateManager = createPendingStateManager();
            node_assert_1.strict.strictEqual(pendingStateManager.minimumPendingMessageSequenceNumber, undefined, "No pending messages should mean no minimum seq number");
            // Each message has a different reference sequence number so let them each be their own batch
            for (const message of forFlushedMessages) {
                pendingStateManager.onFlushBatch([
                    {
                        runtimeOp: message.runtimeOp,
                        referenceSequenceNumber: message.referenceSequenceNumber,
                    },
                ], 0, false /* staged */);
            }
            node_assert_1.strict.strictEqual(pendingStateManager.minimumPendingMessageSequenceNumber, 10, "has pending messages and thus a minimum seq number");
            pendingStateManager.replayPendingStates();
            node_assert_1.strict.strictEqual(pendingStateManager.minimumPendingMessageSequenceNumber, undefined, "Should no minimum sequence number as there are no messages");
        });
    });
    describe("hasPendingUserChanges", () => {
        function createPendingStateManager(pendingMessages = [], initialMessages = []) {
            const psm = new pendingStateManager_js_1.PendingStateManager({
                applyStashedOp: async () => undefined,
                clientId: () => "CLIENT_ID",
                connected: () => true,
                reSubmitBatch: () => { },
                isActiveConnection: () => false,
                isAttached: () => true,
            }, { pendingStates: initialMessages }, logger);
            psm.pendingMessages.push(...pendingMessages);
            return psm;
        }
        const dirtyableOp = {
            type: messageTypes_js_1.ContainerMessageType.Alias,
        };
        const nonDirtyableOp = {
            type: messageTypes_js_1.ContainerMessageType.GC,
        };
        it("returns false when there are no pending or initial messages", () => {
            const psm = createPendingStateManager();
            node_assert_1.strict.strictEqual(psm.hasPendingUserChanges(), false, "Should be false with no messages");
        });
        it("returns true if any pending message is dirtyable", () => {
            const pendingMessages = [
                {
                    runtimeOp: dirtyableOp,
                },
            ];
            const psm = createPendingStateManager(pendingMessages);
            node_assert_1.strict.strictEqual(psm.hasPendingUserChanges(), true, "Should be true with dirtyable op");
        });
        it("returns false if all pending messages are not dirtyable", () => {
            const pendingMessages = [
                {
                    runtimeOp: nonDirtyableOp,
                },
            ];
            const psm = createPendingStateManager(pendingMessages);
            node_assert_1.strict.strictEqual(psm.hasPendingUserChanges(), false, "Should be false with non-dirtyable op");
        });
        it("returns false if all pending messages are empty batches (runtimeOp undefined)", () => {
            const psm = createPendingStateManager();
            const { placeholderMessage } = opGroupingManager.createEmptyGroupedBatch("batchId", 0);
            psm.onFlushEmptyBatch(placeholderMessage, 0, false);
            node_assert_1.strict.strictEqual(psm.hasPendingUserChanges(), false, "Should be false for empty batch");
        });
        it("returns true if there are any initial messages, even if non-dirtyable", () => {
            const initialMessages = [
                {
                    runtimeOp: nonDirtyableOp,
                },
            ];
            const psm = createPendingStateManager([], initialMessages);
            node_assert_1.strict.strictEqual(psm.hasPendingUserChanges(), true, "Should be true with initial messages");
        });
    });
    describe("replayPendingStates", () => {
        const clientId = "clientId";
        for (const { firstBatchSize, secondBatchSize, secondBatchStaged, } of (0, test_pairwise_generator_1.generatePairwiseOptions)({
            firstBatchSize: [0, 1], // 0 means no ops resubmitted so it becomes an empty batch during replay
            secondBatchSize: [undefined, 1, 2], // undefined means no second batch
            secondBatchStaged: test_pairwise_generator_1.booleanCases,
        })) {
            it(`should replay all pending states as batches [batch sizes: ${firstBatchSize} / ${secondBatchSize} ${secondBatchStaged ? "(staged)" : ""}]`, () => {
                const stubs = getStateHandlerStub();
                const pendingStateManager = newPendingStateManager(stubs);
                const RefSeqInitial_10 = 10;
                const refSeqResubmit_15 = 15;
                stubs.reSubmitBatch.callsFake((batch, metadata) => {
                    // Here's where we implement [firstBatchSize === 0] case - Flush an empty batch on resubmit
                    if (firstBatchSize === 0 && stubs.reSubmitBatch.callCount === 1) {
                        (0, node_assert_1.strict)(metadata.batchId !== undefined, "PRECONDITION: Expected batchId for empty batch");
                        const { placeholderMessage } = opGroupingManager.createEmptyGroupedBatch(metadata.batchId, refSeqResubmit_15);
                        pendingStateManager.onFlushEmptyBatch(placeholderMessage, 
                        /* clientSequenceNumber: */ 1, 
                        /* staged: */ metadata.staged);
                        return;
                    }
                    // Otherwise, simulate the typical re-submission of the batch through the ContainerRuntime
                    pendingStateManager.onFlushBatch(withBatchMetadata(batch.map(({ runtimeOp, opMetadata, localOpMetadata }) => ({
                        runtimeOp,
                        referenceSequenceNumber: refSeqResubmit_15,
                        metadata: opMetadata,
                        localOpMetadata: `${localOpMetadata}_RESUBMITTED`,
                    })), metadata.batchId), 
                    /* clientSequenceNumber: */ metadata.staged ? undefined : 1, 
                    /* staged: */ metadata.staged);
                });
                // First batch - Always starts with 1 op, but will become empty on replay if firstBatchSize is 0
                pendingStateManager.onFlushBatch([
                    {
                        runtimeOp: {
                            type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                            contents: testAddressedDataStoreMessage,
                        },
                        referenceSequenceNumber: RefSeqInitial_10,
                        metadata: undefined, // Single message batch has no batch metadata
                        localOpMetadata: "FIRST_BATCH_MSG1",
                    },
                ], 
                /* clientSequenceNumber: */ 1, 
                /* staged: */ false);
                // Second batch (if applicable) - May have multiple ops or be skipped altogether
                if (secondBatchSize !== undefined) {
                    pendingStateManager.onFlushBatch(withBatchMetadata(Array.from({ length: secondBatchSize }, (_, i) => ({
                        runtimeOp: {
                            type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                            contents: testAddressedDataStoreMessage,
                        },
                        referenceSequenceNumber: RefSeqInitial_10,
                        localOpMetadata: `SECOND_BATCH_MSG${i + 1}`,
                    }))), 
                    /* clientSequenceNumber: */ secondBatchStaged ? undefined : 2, 
                    /* staged: */ secondBatchStaged);
                }
                pendingStateManager.replayPendingStates();
                const resubmittedMessages = pendingStateManager.pendingMessages.toArray();
                node_assert_1.strict.equal(resubmittedMessages.length, 1 + (secondBatchSize ?? 0), // Even if firstBatchSize is 0, we still have the empty batch placeholder
                "Incorrect number of resubmitted messages");
                // First batch expectations - Should be 1 pending message, either the empty batch placeholder or the first message
                const [firstResubmittedBatchPendingMessage] = resubmittedMessages.splice(0, 1);
                (({ batchInfo: { length, staged }, opMetadata, referenceSequenceNumber, }) => {
                    node_assert_1.strict.strictEqual(length, 1, "First batch size incorrect");
                    node_assert_1.strict.strictEqual(opMetadata?.batchId, `${clientId}_[1]`);
                    node_assert_1.strict.strictEqual(staged, false);
                    node_assert_1.strict.strictEqual(referenceSequenceNumber, refSeqResubmit_15);
                })(firstResubmittedBatchPendingMessage);
                // Second batch expectations
                if (secondBatchSize === undefined) {
                    (0, node_assert_1.strict)(resubmittedMessages.length === 0, "No second batch expected");
                    return;
                }
                const secondResubmittedBatchPendingMessages = resubmittedMessages.splice(0, secondBatchSize);
                // The first messages should have batchInfo and batchId on it
                (({ batchInfo: { length, staged }, opMetadata }) => {
                    node_assert_1.strict.strictEqual(length, secondBatchSize, "Wrong batch size (2nd)");
                    if (secondBatchStaged) {
                        (0, node_assert_1.strict)(opMetadata?.batchId?.length === 41, "Wrong clientId (2nd)");
                        (0, node_assert_1.strict)(opMetadata?.batchId?.includes("-1"), "Wrong clientId (2nd)");
                    }
                    else {
                        node_assert_1.strict.strictEqual(opMetadata?.batchId, `${clientId}_[2]`, "Wrong clientId (2nd)");
                    }
                    node_assert_1.strict.strictEqual(staged, secondBatchStaged, "Wrong staged flag (2nd)");
                })(secondResubmittedBatchPendingMessages[0]);
                // Every message should have the same reference sequence number
                (0, node_assert_1.strict)(secondResubmittedBatchPendingMessages.every((m) => m.referenceSequenceNumber === refSeqResubmit_15), "Second batch reference sequence number incorrect");
            });
        }
        it("should replay only staged batches when committingStagedBatches is true", () => {
            const stubs = getStateHandlerStub();
            const pendingStateManager = newPendingStateManager(stubs);
            const reSubmittedBatches = [];
            stubs.reSubmitBatch.callsFake((batch, metadata) => {
                reSubmittedBatches.push({ batch, metadata });
            });
            // Enqueue an unstaged one first, then staged.  The opposite order is not possible/supported
            pendingStateManager.onFlushBatch([
                {
                    runtimeOp: {
                        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                        contents: testAddressedDataStoreMessage,
                    },
                    referenceSequenceNumber: 13,
                    metadata: undefined,
                    localOpMetadata: { foo: "unstaged" },
                },
            ], 
            /* clientSequenceNumber: */ 1, 
            /* staged: */ false);
            pendingStateManager.onFlushBatch([
                {
                    runtimeOp: {
                        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                        contents: testAddressedDataStoreMessage,
                    },
                    referenceSequenceNumber: 12,
                    metadata: undefined,
                    localOpMetadata: { foo: "staged" },
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ true);
            pendingStateManager.replayPendingStates({
                committingStagedBatches: true,
                squash: false,
            });
            // We should only resubmit the staged batch, with the staged flag cleared
            node_assert_1.strict.strictEqual(reSubmittedBatches.length, 1, "Should resubmit only the staged batch when committingStagedBatches is true");
            node_assert_1.strict.deepStrictEqual(reSubmittedBatches[0].batch[0].localOpMetadata, { foo: "staged" }, "Should resubmit the staged batch");
            node_assert_1.strict.strictEqual(reSubmittedBatches[0].metadata.staged, false, "Staged flag should be cleared on resubmission");
            // The unstaged batch should remain in the pendingMessages queue
            node_assert_1.strict.strictEqual(pendingStateManager.pendingMessages.length, 1, "Unstaged batch should remain in the queue");
            node_assert_1.strict.deepStrictEqual(pendingStateManager.pendingMessages.peekFront()?.localOpMetadata, {
                foo: "unstaged",
            });
        });
        it("should throw if replayPendingStates is called twice for same clientId without committingStagedBatches", () => {
            const stubs = getStateHandlerStub();
            const pendingStateManager = newPendingStateManager(stubs);
            const reSubmittedBatches = [];
            stubs.reSubmitBatch.callsFake((batch, metadata) => {
                reSubmittedBatches.push({ batch, metadata });
            });
            pendingStateManager.onFlushBatch([
                {
                    runtimeOp: {
                        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                        contents: testAddressedDataStoreMessage,
                    },
                    referenceSequenceNumber: 15,
                    metadata: undefined,
                    localOpMetadata: { foo: "bar" },
                },
            ], 
            /* clientSequenceNumber: */ 1, 
            /* staged: */ false);
            // This will set clientIdFromLastReplay
            pendingStateManager.replayPendingStates();
            // Add another batch to allow replay again
            pendingStateManager.onFlushBatch([
                {
                    runtimeOp: {
                        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                        contents: testAddressedDataStoreMessage,
                    },
                    referenceSequenceNumber: 15,
                    metadata: undefined,
                    localOpMetadata: { foo: "bar" },
                },
            ], 
            /* clientSequenceNumber: */ 1, 
            /* staged: */ false);
            // This will throw since clientIdFromLastReplay is already set to the same clientId
            node_assert_1.strict.throws(() => pendingStateManager.replayPendingStates(), /0x173/, "Should throw if replayPendingStates is called twice for same clientId");
        });
        it("should set squash flag when replayPendingStates is called with squash: true", () => {
            const stubs = getStateHandlerStub();
            const pendingStateManager = newPendingStateManager(stubs);
            pendingStateManager.onFlushBatch([
                {
                    runtimeOp: {
                        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                        contents: testAddressedDataStoreMessage,
                    },
                    referenceSequenceNumber: 16,
                    metadata: undefined,
                    localOpMetadata: { foo: "bar" },
                },
            ], 
            /* clientSequenceNumber: */ 1, 
            /* staged: */ false);
            pendingStateManager.replayPendingStates({
                squash: true,
                committingStagedBatches: false,
            });
            (0, node_assert_1.strict)(stubs.reSubmitBatch.calledOnceWith(sinon_1.default.match.any, {
                squash: true,
                staged: sinon_1.default.match.bool,
                batchId: sinon_1.default.match.string,
            }), "Squash flag should be set to true");
        });
    });
    describe("popStagedBatches", () => {
        it("should pop all staged batch messages in LIFO order and invoke callback", () => {
            const stubs = getStateHandlerStub();
            const psm = newPendingStateManager(stubs);
            // Add staged batch 1
            psm.onFlushBatch([
                {
                    runtimeOp: op("foo1"),
                    referenceSequenceNumber: 1,
                    metadata: undefined,
                    localOpMetadata: undefined,
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ true);
            // Add staged batch 2
            psm.onFlushBatch([
                {
                    runtimeOp: op("foo2"),
                    referenceSequenceNumber: 2,
                    metadata: undefined,
                    localOpMetadata: undefined,
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ true);
            const popped = [];
            psm.popStagedBatches((msg) => {
                popped.push(msg.runtimeOp.contents);
            });
            node_assert_1.strict.deepStrictEqual(popped, ["foo2", "foo1"], "Should pop in LIFO order");
            node_assert_1.strict.strictEqual(psm.hasPendingMessages(), false, "All staged messages should be popped");
        });
        it("should not pop unstaged messages", () => {
            const stubs = getStateHandlerStub();
            const psm = newPendingStateManager(stubs);
            // Add unstaged batch
            psm.onFlushBatch([
                {
                    runtimeOp: op("foo1"),
                    referenceSequenceNumber: 1,
                    metadata: undefined,
                    localOpMetadata: undefined,
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ false);
            // Add staged batch
            psm.onFlushBatch([
                {
                    runtimeOp: op("foo2"),
                    referenceSequenceNumber: 2,
                    metadata: undefined,
                    localOpMetadata: undefined,
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ true);
            const popped = [];
            psm.popStagedBatches((msg) => {
                popped.push(msg.runtimeOp.contents);
            });
            node_assert_1.strict.deepStrictEqual(popped, ["foo2"], "Should only pop staged messages");
            node_assert_1.strict.strictEqual(psm.hasPendingMessages(), true, "Unstaged message should remain");
            node_assert_1.strict.strictEqual(psm.pendingMessages.length, 1, "Only unstaged message should remain in queue");
        });
        it("should not invoke callback for empty batch or messages without runtimeOp", () => {
            const stubs = getStateHandlerStub();
            const psm = newPendingStateManager(stubs);
            // Add staged empty batch
            const { placeholderMessage } = opGroupingManager.createEmptyGroupedBatch("batchId", 3);
            psm.onFlushEmptyBatch(placeholderMessage, 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ true);
            // Add staged message with no runtimeOp (simulate by direct mutation after adding)
            psm.onFlushBatch([
                {
                    runtimeOp: op("foo2"),
                    referenceSequenceNumber: 4,
                    metadata: undefined,
                    localOpMetadata: undefined,
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ true);
            // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
            psm.pendingMessages.peekBack().runtimeOp = undefined;
            // Add staged normal batch
            psm.onFlushBatch([
                {
                    runtimeOp: op("foo3"),
                    referenceSequenceNumber: 5,
                    metadata: undefined,
                    localOpMetadata: undefined,
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ true);
            const popped = [];
            psm.popStagedBatches((msg) => {
                popped.push(msg.runtimeOp.contents);
            });
            node_assert_1.strict.deepStrictEqual(popped, ["foo3"], "Should only invoke the callback on staged messages with typical runtimeOp");
            node_assert_1.strict.strictEqual(psm.pendingMessages.length, 0, "Non-typical staged messages should also be popped, just without invoking callback");
        });
        it("should do nothing if there are no staged messages", () => {
            const stubs = getStateHandlerStub();
            const psm = newPendingStateManager(stubs);
            // Add unstaged batch
            psm.onFlushBatch([
                {
                    runtimeOp: op("foo1"),
                    referenceSequenceNumber: 1,
                    metadata: undefined,
                    localOpMetadata: undefined,
                },
            ], 
            /* clientSequenceNumber: */ undefined, 
            /* staged: */ false);
            const popped = [];
            psm.popStagedBatches((msg) => {
                popped.push(msg.runtimeOp.type);
            });
            node_assert_1.strict.deepStrictEqual(popped, [], "Should not pop any messages");
            node_assert_1.strict.strictEqual(psm.pendingMessages.length, 1, "Unstaged message should remain in queue");
        });
    });
});
//# sourceMappingURL=pendingStateManager.spec.js.map