"use strict";
/*!
 * Copyright (c) Microsoft Corporation and contributors. All rights reserved.
 * Licensed under the MIT License.
 */
Object.defineProperty(exports, "__esModule", { value: true });
const node_assert_1 = require("node:assert");
const container_definitions_1 = require("@fluidframework/container-definitions");
const internal_1 = require("@fluidframework/core-interfaces/internal");
const internal_2 = require("@fluidframework/driver-definitions/internal");
const internal_3 = require("@fluidframework/telemetry-utils/internal");
const internal_4 = require("@fluidframework/test-runtime-utils/internal");
const sinon_1 = require("sinon");
const containerRuntime_js_1 = require("../containerRuntime.js");
const dataStoreRegistry_js_1 = require("../dataStoreRegistry.js");
const messageTypes_js_1 = require("../messageTypes.js");
describe("Runtime batching", () => {
    const mockClientId = "mockClientId";
    const getMockContext = (deltaManager) => {
        const mockContext = {
            attachState: container_definitions_1.AttachState.Attached,
            deltaManager,
            audience: new internal_4.MockAudience(),
            quorum: new internal_4.MockQuorumClients(),
            taggedLogger: (0, internal_3.mixinMonitoringContext)(new internal_3.MockLogger()).logger,
            clientDetails: { capabilities: { interactive: true } },
            closeFn: (_error) => { },
            updateDirtyContainerState: (_dirty) => { },
            getLoadedFromVersion: () => undefined,
            clientId: mockClientId,
            connected: true,
        };
        return mockContext;
    };
    const mockProvideEntryPoint = async () => ({
        myProp: "myValue",
    });
    let containerRuntime;
    let mockDeltaManager;
    let sandbox;
    let clock;
    /**
     * Overwrites channelCollection property to make process a no-op
     */
    function patchContainerRuntime(cr) {
        const fakeProcess = () => { };
        const patched = cr;
        return sandbox.stub(patched.channelCollection, "processMessages").callsFake(fakeProcess);
    }
    before(() => {
        clock = (0, sinon_1.useFakeTimers)();
        sandbox = (0, sinon_1.createSandbox)();
    });
    beforeEach(async () => {
        mockDeltaManager = new internal_4.MockDeltaManager();
        ({ runtime: containerRuntime } = await containerRuntime_js_1.ContainerRuntime.loadRuntime2({
            context: getMockContext(mockDeltaManager),
            registry: new dataStoreRegistry_js_1.FluidDataStoreRegistry([]),
            existing: false,
            runtimeOptions: {},
            provideEntryPoint: mockProvideEntryPoint,
        }));
    });
    afterEach(() => {
        clock.reset();
    });
    after(() => {
        sandbox.restore();
        clock.restore();
    });
    /**
     * Returns a batch of messages with the first and last message marked as batch start and end respectively.
     */
    function getMessages(count, modernRuntimeMessage = true) {
        const messageContents = {
            address: "address",
            contents: "dummy content",
        };
        const messages = [];
        for (let i = 0; i < count; i++) {
            messages.push({
                type: modernRuntimeMessage
                    ? internal_2.MessageType.Operation
                    : messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                clientId: mockClientId,
                sequenceNumber: i,
                minimumSequenceNumber: 0,
                contents: modernRuntimeMessage
                    ? {
                        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                        contents: messageContents,
                    }
                    : messageContents,
                clientSequenceNumber: i,
                timestamp: Date.now(),
            });
        }
        if (count > 1) {
            messages[0].metadata = { batch: true };
            messages[messages.length - 1].metadata = { batch: false };
        }
        return messages;
    }
    /**
     * Processes the given batch. The batch is processed by pushing each message to the inbound queue and then
     * processing the messages in the queue.
     */
    function processBatch(batch, cr) {
        // Push the messages in the inbound queue. This is done because InboundBatchAggregator listens to the "push" event
        // emitted by the inbound queue to do batch validations.
        for (const batchMessage of batch) {
            mockDeltaManager.inbound.push(batchMessage);
        }
        // Process the messages in the inbound queue.
        // Process is called on the delta manager because InboundBatchAggregator listens to the "op" event on delta manager
        // as well to do validation.
        // Process is called on the container runtime because it is the one that actually processes the messages and
        // has its own set of validations.
        let message = mockDeltaManager.inbound.pop();
        while (message !== undefined) {
            (0, node_assert_1.strict)(message !== undefined, "Message should not be undefined");
            mockDeltaManager.process(message);
            cr.process(message, false /* local */);
            message = mockDeltaManager.inbound.pop();
        }
    }
    describe("Batch validation", () => {
        let containerRuntimeStub;
        beforeEach(async () => {
            containerRuntimeStub = patchContainerRuntime(containerRuntime);
        });
        afterEach(() => {
            containerRuntimeStub.restore();
        });
        it("successfully processes messages that are not part of batch", async () => {
            const messageCount = 5;
            const batch = getMessages(messageCount);
            // Remove the batch metadata essentially making the messages not part of a batch.
            batch[0].metadata = undefined;
            batch[messageCount - 1].metadata = undefined;
            node_assert_1.strict.doesNotThrow(() => processBatch(batch, containerRuntime), "Non batch messages should be processed successfully");
        });
        it("successfully processes a batch containing ops from a single client", async () => {
            const messageCount = 5;
            const batch = getMessages(messageCount);
            node_assert_1.strict.doesNotThrow(() => processBatch(batch, containerRuntime), "Batch from a single client should be processed successfully");
        });
        it("fails processing a batch with batch end but no batch start", async () => {
            const messageCount = 5;
            const batch = getMessages(messageCount);
            // Remove the batch begin metadata.
            batch[0].metadata = undefined;
            node_assert_1.strict.throws(() => processBatch(batch, containerRuntime), (0, internal_4.validateAssertionError)("batch presence was validated above"), "Batch end without batch start should fail");
        });
        it("fails processing a batch with multiple batch starts", async () => {
            const messageCount = 5;
            const batch = getMessages(messageCount);
            batch[2].metadata = { batch: true };
            node_assert_1.strict.throws(() => processBatch(batch, containerRuntime), (0, internal_4.validateAssertionError)("there can't be active batch"), "Batch with multiple batch starts should fail");
        });
        it("fails processing a batch containing ops from multiple clients", async () => {
            const messageCount = 5;
            const batch = getMessages(messageCount);
            // Change the clientId of the second message to a different client.
            batch[1].clientId = "otherClientId";
            node_assert_1.strict.throws(() => processBatch(batch, containerRuntime), (e) => {
                (0, node_assert_1.strict)(e.errorType === internal_1.FluidErrorTypes.dataCorruptionError);
                (0, node_assert_1.strict)(e.message === "OpBatchIncomplete");
                return true;
            }, "Batch with ops from multiple clients should fail");
        });
        it("fails processing a batch containing a non-runtime op along with runtime ops", async () => {
            const messageCount = 5;
            const batch = getMessages(messageCount);
            // Change the type of the second message to a non-runtime op.
            batch[1].type = internal_2.MessageType.NoOp;
            node_assert_1.strict.throws(() => processBatch(batch, containerRuntime), (e) => {
                (0, node_assert_1.strict)(e.errorType === internal_1.FluidErrorTypes.dataProcessingError);
                (0, node_assert_1.strict)(e.message === "Received a system message during batch processing");
                return true;
            }, "Batch with non-runtime op along with runtime ops should fail");
        });
        it("fails processing a batch containing an unknown runtime op along with known ops", async () => {
            const messageCount = 5;
            const batch = getMessages(messageCount);
            // Change the type of the second message to an unknown runtime op.
            const unknownMessage = batch[1];
            const unknownMessageType = "unknown";
            // eslint-disable-next-line @typescript-eslint/no-explicit-any, @typescript-eslint/no-unsafe-member-access
            unknownMessage.contents.type = unknownMessageType;
            node_assert_1.strict.throws(() => processBatch(batch, containerRuntime), (e) => {
                (0, node_assert_1.strict)(e.errorType === internal_1.FluidErrorTypes.dataProcessingError);
                (0, node_assert_1.strict)(e.message === "Runtime message of unknown type");
                return true;
            }, "Batch with unknown runtime op along with known ops should fail");
        });
    });
    /**
     * These tests validate that container runtime handles batch begin and end correctly. It should emit
     * batch begin and end events.
     */
    describe("Batch begin and end", () => {
        let batchBeginCount = 0;
        let batchEndCount = 0;
        let containerRuntimeStub;
        beforeEach(async () => {
            containerRuntimeStub = patchContainerRuntime(containerRuntime);
            containerRuntime.on("batchBegin", () => {
                batchBeginCount++;
            });
            containerRuntime.on("batchEnd", () => {
                batchEndCount++;
            });
        });
        afterEach(() => {
            containerRuntimeStub.restore();
            batchBeginCount = 0;
            batchEndCount = 0;
        });
        function setupOpProcessingFailure() {
            containerRuntime.once("op", () => {
                throw new Error("Failed processing op in test");
            });
        }
        function validateBatchBeginAndEnd() {
            node_assert_1.strict.strictEqual(batchBeginCount, 1, "Batch begin should have been emitted once");
            node_assert_1.strict.strictEqual(batchEndCount, 1, "Batch end should have been emitted once");
        }
        it("handles batch begin and end for successfully processing modern runtime messages", async () => {
            const modernRuntimeMessage = {
                type: internal_2.MessageType.Operation,
                clientId: mockClientId,
                sequenceNumber: 1,
                minimumSequenceNumber: 0,
                contents: {
                    type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                    contents: {
                        address: "address",
                        contents: "dummy content",
                    },
                },
                clientSequenceNumber: 1,
                timestamp: Date.now(),
            };
            node_assert_1.strict.doesNotThrow(() => containerRuntime.process(modernRuntimeMessage, false /* local */), "Message processing should have succeeded");
            validateBatchBeginAndEnd();
        });
        it("handles batch begin and end for failed processing of modern runtime messages", async () => {
            setupOpProcessingFailure();
            const modernRuntimeMessage = {
                type: internal_2.MessageType.Operation,
                clientId: mockClientId,
                sequenceNumber: 1,
                minimumSequenceNumber: 0,
                contents: {
                    type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                    contents: {
                        address: "address",
                        contents: "dummy content",
                    },
                },
                clientSequenceNumber: 1,
                timestamp: Date.now(),
            };
            node_assert_1.strict.throws(() => containerRuntime.process(modernRuntimeMessage, false /* local */), (0, internal_4.validateAssertionError)("Failed processing op in test"), "Message processing should have failed");
            validateBatchBeginAndEnd();
        });
        it("handles batch begin and end for successfully processing legacy runtime messages", async () => {
            const legacyMessage = {
                type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                clientId: mockClientId,
                sequenceNumber: 1,
                minimumSequenceNumber: 0,
                contents: {
                    address: "address",
                    contents: "dummy content",
                },
                clientSequenceNumber: 1,
                timestamp: Date.now(),
            };
            node_assert_1.strict.doesNotThrow(() => containerRuntime.process(legacyMessage, false /* local */), "Non batch messages should be processed successfully");
            validateBatchBeginAndEnd();
        });
        it("handles batch begin and end for failed processing of legacy runtime messages", async () => {
            setupOpProcessingFailure();
            const legacyMessage = {
                type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                clientId: mockClientId,
                sequenceNumber: 1,
                minimumSequenceNumber: 0,
                contents: {
                    address: "address",
                    contents: "dummy content",
                },
                clientSequenceNumber: 1,
                timestamp: Date.now(),
            };
            node_assert_1.strict.throws(() => containerRuntime.process(legacyMessage, false /* local */), (0, internal_4.validateAssertionError)("Failed processing op in test"), "Message processing should have failed");
            validateBatchBeginAndEnd();
        });
        it("handles batch begin and end for successfully processing non runtime messages", async () => {
            const nonRuntimeMessage = {
                type: internal_2.MessageType.Summarize,
                clientId: mockClientId,
                sequenceNumber: 1,
                minimumSequenceNumber: 0,
                contents: {
                    handle: "test-handle",
                },
                clientSequenceNumber: 1,
                timestamp: Date.now(),
            };
            node_assert_1.strict.doesNotThrow(() => containerRuntime.process(nonRuntimeMessage, false /* local */), "Non batch messages should be processed successfully");
            validateBatchBeginAndEnd();
        });
        it("handles batch begin and end for failed processing of non runtime messages", async () => {
            setupOpProcessingFailure();
            const nonRuntimeMessage = {
                type: internal_2.MessageType.Summarize,
                clientId: mockClientId,
                sequenceNumber: 1,
                minimumSequenceNumber: 0,
                contents: {
                    handle: "test-handle",
                },
                clientSequenceNumber: 1,
                timestamp: Date.now(),
            };
            node_assert_1.strict.throws(() => containerRuntime.process(nonRuntimeMessage, false /* local */), (0, internal_4.validateAssertionError)("Failed processing op in test"), "Message processing should have failed");
            validateBatchBeginAndEnd();
        });
        it("handles batch begin and end for successfully processing empty batch", async () => {
            const emptyBatch = {
                type: internal_2.MessageType.Operation,
                clientId: mockClientId,
                sequenceNumber: 1,
                minimumSequenceNumber: 0,
                clientSequenceNumber: 1,
                contents: {
                    type: "groupedBatch",
                    contents: [],
                },
            };
            node_assert_1.strict.doesNotThrow(() => containerRuntime.process(emptyBatch, false /* local */), "Non batch messages should be processed successfully");
            validateBatchBeginAndEnd();
        });
    });
});
//# sourceMappingURL=batching.spec.js.map