"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 internal_1 = require("@fluidframework/telemetry-utils/internal");
const internal_2 = require("@fluidframework/test-runtime-utils/internal");
const compressionDefinitions_js_1 = require("../../compressionDefinitions.js");
const containerRuntime_js_1 = require("../../containerRuntime.js");
const messageTypes_js_1 = require("../../messageTypes.js");
const metadata_js_1 = require("../../metadata.js");
const index_js_1 = require("../../opLifecycle/index.js");
// Make a mock op with distinguishable contents
function op(data) {
    // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
    return {
        type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
        contents: data,
    };
}
describe("Outbox", () => {
    const maxBatchSizeInBytes = 1024;
    // state will be set to defaults in beforeEach
    const state = {};
    const mockLogger = new internal_1.MockLogger();
    const getMockDeltaManager = () => ({
        flush() {
            state.deltaManagerFlushCalls++;
        },
    });
    const getMockContext = () => ({
        deltaManager: getMockDeltaManager(),
        clientDetails: { capabilities: { interactive: true } },
        updateDirtyContainerState: (_dirty) => { },
        submitFn: (type, contents, batch, appData) => {
            state.individualOpsSubmitted.push({ type, contents, batch, appData });
            state.opsSubmitted++;
            return state.opsSubmitted;
        },
        submitBatchFn: (messages, referenceSequenceNumber) => {
            state.batchesSubmitted.push({ messages, referenceSequenceNumber });
            state.opsSubmitted += messages.length;
            return state.opsSubmitted;
        },
    });
    const getMockLegacyContext = () => ({
        deltaManager: getMockDeltaManager(),
        clientDetails: { capabilities: { interactive: true } },
        updateDirtyContainerState: (_dirty) => { },
        submitFn: (type, contents, batch, appData) => {
            state.individualOpsSubmitted.push({ type, contents, batch, appData });
            state.opsSubmitted++;
            return state.opsSubmitted;
        },
        connected: true,
    });
    const getMockCompressor = () => ({
        compressBatch: (batch) => {
            state.batchesCompressed.push(batch);
            return batch;
        },
    });
    const getMockSplitter = (enabled, chunkSizeInBytes) => ({
        chunkSizeInBytes,
        isBatchChunkingEnabled: enabled,
        splitSingletonBatchMessage: (batch) => {
            state.batchesSplit.push(batch);
            return batch;
        },
    });
    const getMockPendingStateManager = () => ({
        // Similar implementation as the real PSM - queue each message 1-by-1
        onFlushBatch: (batch, clientSequenceNumber) => {
            for (const { runtimeOp, referenceSequenceNumber, metadata: opMetadata, localOpMetadata, } of batch)
                state.pendingOpContents.push({
                    runtimeOp,
                    referenceSequenceNumber,
                    opMetadata,
                    localOpMetadata,
                    batchStartCsn: clientSequenceNumber ?? -1,
                });
        },
        onFlushEmptyBatch: ({ metadata: opMetadata, localOpMetadata, referenceSequenceNumber, }, clientSequenceNumber) => state.pendingOpContents.push({
            referenceSequenceNumber,
            opMetadata,
            localOpMetadata,
            batchStartCsn: clientSequenceNumber,
        }),
    });
    const createMessage = (type, fakeContents) => ({
        // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
        runtimeOp: { type, contents: fakeContents },
        metadata: undefined,
        localOpMetadata: {},
        referenceSequenceNumber: Number.POSITIVE_INFINITY,
    });
    const toSubmittedMessage = (message, batchMarker = undefined) => ({
        contents: message.runtimeOp === undefined
            ? message.contents
            : (0, index_js_1.serializeOp)(message.runtimeOp).content,
        metadata: batchMarker === undefined
            ? message.metadata
            : { ...message.metadata, batch: batchMarker },
        compression: undefined,
        referenceSequenceNumber: message.referenceSequenceNumber,
    });
    // Also converts to an OutboundBatchMessage
    const addBatchMetadata = (messages) => {
        if (messages.length > 1) {
            messages[0].metadata = {
                ...messages[0].metadata,
                batch: true,
            };
            messages[messages.length - 1].metadata = {
                ...messages[messages.length - 1].metadata,
                batch: false,
            };
        }
    };
    const toOutboundBatch = (messages) => {
        const outbound = (0, index_js_1.localBatchToOutboundBatch)({
            messages,
            referenceSequenceNumber: messages.length === 0 ? undefined : messages[0].referenceSequenceNumber,
            hasReentrantOps: false,
        });
        addBatchMetadata(outbound.messages);
        return outbound;
    };
    const DefaultCompressionOptions = {
        minimumBatchSizeInBytes: Number.POSITIVE_INFINITY,
        compressionAlgorithm: compressionDefinitions_js_1.CompressionAlgorithms.lz4,
    };
    let currentSeqNumbers = {};
    const getOutbox = (params) => {
        const { submitFn, submitBatchFn, deltaManager } = params.context;
        const legacySendBatchFn = (0, containerRuntime_js_1.makeLegacySendBatchFn)(submitFn, deltaManager);
        return new index_js_1.Outbox({
            shouldSend: () => state.canSendOps,
            pendingStateManager: getMockPendingStateManager(),
            submitBatchFn,
            legacySendBatchFn,
            compressor: getMockCompressor(),
            splitter: getMockSplitter(params.enableChunking ?? false, params.chunkSizeInBytes ?? Number.POSITIVE_INFINITY),
            config: {
                maxBatchSizeInBytes: params.maxBatchSize ?? maxBatchSizeInBytes,
                compressionOptions: params.compressionOptions ?? DefaultCompressionOptions,
            },
            logger: mockLogger,
            groupingManager: new index_js_1.OpGroupingManager(params.opGroupingConfig ?? {
                groupedBatchingEnabled: false,
            }, mockLogger),
            getCurrentSequenceNumbers: () => currentSeqNumbers,
            reSubmit: (message) => {
                state.opsResubmitted++;
            },
            opReentrancy: () => state.isReentrant,
            generateIdAllocationOp: params.generateIdAllocationOp ?? (() => undefined),
        });
    };
    const opGroupingManager = new index_js_1.OpGroupingManager({
        groupedBatchingEnabled: true,
    }, mockLogger);
    beforeEach(() => {
        state.deltaManagerFlushCalls = 0;
        state.canSendOps = true;
        state.batchesSubmitted = [];
        state.batchesCompressed = [];
        state.batchesSplit = [];
        state.individualOpsSubmitted = [];
        state.pendingOpContents = [];
        state.opsSubmitted = 0;
        state.opsResubmitted = 0;
        state.isReentrant = false;
        currentSeqNumbers = {};
        mockLogger.clear();
    });
    it("localBatchToOutboundBatch", () => {
        const localMessages = [
            { runtimeOp: op("hello"), referenceSequenceNumber: 4 },
            { runtimeOp: op("world"), referenceSequenceNumber: 4 },
            { runtimeOp: op("!"), referenceSequenceNumber: 4 },
        ];
        const localBatch = {
            messages: localMessages,
            referenceSequenceNumber: localMessages[0].referenceSequenceNumber,
            hasReentrantOps: false,
        };
        const outboundBatch = (0, index_js_1.localBatchToOutboundBatch)(localBatch);
        // Check that contentSizeInBytes and messages' contents are set propertly
        node_assert_1.strict.equal(outboundBatch.contentSizeInBytes, JSON.stringify(op("")).length * 3 + "helloworld!".length, "contentSizeInBytes is incorrect");
        node_assert_1.strict.equal(outboundBatch.messages.length, 3);
        node_assert_1.strict.deepEqual(localMessages.map((m) => m.runtimeOp), 
        // eslint-disable-next-line @typescript-eslint/no-unsafe-return, @typescript-eslint/no-non-null-assertion
        outboundBatch.messages.map((m) => JSON.parse(m.contents)), "Serialized contents do not match");
    });
    it("Sending batches", () => {
        const idAllocMessage = createMessage(messageTypes_js_1.ContainerMessageType.IdAllocation, "idAlloc");
        let idAllocReturned = false;
        const outbox = getOutbox({
            context: getMockContext(),
            // JIT callback: return an ID alloc op on the first call, then undefined
            generateIdAllocationOp: () => {
                if (!idAllocReturned) {
                    idAllocReturned = true;
                    return idAllocMessage;
                }
                return undefined;
            },
        });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "4"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "5"),
        ];
        // Flush 1 — JIT ID alloc op is prepended to the first non-empty batch (blobAttach is empty, so main batch)
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.flush();
        // Flush 2 — no more ID alloc ops
        outbox.submit(messages[2]);
        outbox.flush();
        // Not Flushed
        outbox.submit(messages[3]);
        node_assert_1.strict.equal(state.opsSubmitted, 4); // 2 main + idAlloc from flush 1, 1 main from flush 2
        node_assert_1.strict.equal(state.individualOpsSubmitted.length, 0);
        node_assert_1.strict.deepEqual(state.batchesSubmitted.map((x) => x.messages), [
            // Flush 1 (Main) — idAlloc prepended
            [
                toSubmittedMessage(idAllocMessage, true),
                toSubmittedMessage(messages[0]),
                toSubmittedMessage(messages[1], false),
            ],
            // Flush 2 (Main)
            [toSubmittedMessage(messages[2])],
        ], "Submitted batches are incorrect");
        node_assert_1.strict.equal(state.deltaManagerFlushCalls, 0);
        // Note the expected CSN here is fixed to the batch's starting CSN
        const expectedMessageOrderWithCsn = [
            // Flush 1 (Main with prepended idAlloc)
            [idAllocMessage, 1],
            [messages[0], 1],
            [messages[1], 1],
            // Flush 2 (Main)
            [messages[2], 4],
        ];
        node_assert_1.strict.deepEqual(state.pendingOpContents, expectedMessageOrderWithCsn.map(([message, csn]) => ({
            runtimeOp: message.runtimeOp,
            referenceSequenceNumber: message.referenceSequenceNumber,
            localOpMetadata: message.localOpMetadata,
            opMetadata: message.metadata,
            batchStartCsn: csn,
        })), "Pending messages are incorrect");
    });
    it("Flush empty (GroupedBatching enabled)", () => {
        const outbox = getOutbox({
            context: getMockContext(),
            opGroupingConfig: {
                groupedBatchingEnabled: true,
            },
        });
        currentSeqNumbers.referenceSequenceNumber = 0;
        // Typically, flushing with nothing submitted should be a no-op...
        outbox.flush();
        node_assert_1.strict.equal(state.opsSubmitted, 0);
        node_assert_1.strict.equal(state.batchesSubmitted.length, 0);
        node_assert_1.strict.equal(state.deltaManagerFlushCalls, 0);
        node_assert_1.strict.equal(state.pendingOpContents.length, 0);
        const batchId = "batchId";
        // ...But if batchId is provided, it's resubmit, and we need to send an empty batch with the batchId
        outbox.flush({ batchId, staged: false });
        node_assert_1.strict.equal(state.opsSubmitted, 1);
        node_assert_1.strict.equal(state.batchesSubmitted.length, 1);
        node_assert_1.strict.equal(state.batchesSubmitted[0].messages[0].contents, '{"type":"groupedBatch","contents":[]}');
        node_assert_1.strict.equal(state.batchesSubmitted[0].messages[0].metadata?.batchId, batchId);
        node_assert_1.strict.equal((0, metadata_js_1.asEmptyBatchLocalOpMetadata)(state.pendingOpContents[0].localOpMetadata)?.emptyBatch, true);
    });
    it("Batch ID added when applicable (ungrouped batch)", () => {
        const outbox = getOutbox({
            context: getMockContext(),
            opGroupingConfig: {
                groupedBatchingEnabled: false,
            },
        });
        // Flush 1 - resubmit multi-message batch
        outbox.submit(createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"));
        outbox.submit(createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"));
        outbox.flush({ batchId: "batchId-A", staged: false });
        // Flush 2 - resubmit single-message batch
        outbox.submit(createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "3"));
        outbox.flush({ batchId: "batchId-B", staged: false });
        // Flush 3 - resubmit blob attach batch
        outbox.submitBlobAttach(createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "4"));
        outbox.submitBlobAttach(createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "5"));
        currentSeqNumbers.referenceSequenceNumber = 0;
        outbox.flush({ batchId: "batchId-C", staged: false });
        // Flush 4 - no batch ID given
        outbox.submit(createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "6"));
        outbox.flush(); // Ignored - No batchID given (not resubmit)
        // Not Flushed (will not appear in batchesSubmitted or pendingOpContents)
        outbox.submit(createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "7"));
        node_assert_1.strict.deepEqual(state.batchesSubmitted.map((x) => x.messages.map((m) => m.metadata?.batchId)), [
            ["batchId-A", undefined], // Flush 1 - Main
            ["batchId-B"], // Flush 2 - Main
            ["batchId-C", undefined], // Flush 3 - Blob Attach
            [undefined], // Flush 4 - Main (no batch ID given)
        ], "Submitted batches have incorrect batch ID");
        node_assert_1.strict.deepEqual(state.pendingOpContents.map(({ opMetadata }) => (0, metadata_js_1.asBatchMetadata)(opMetadata)?.batchId), [
            "batchId-A",
            undefined, // second message in batch
            "batchId-B",
            "batchId-C",
            undefined, // second message in batch
            undefined, // no batchId given
        ], "Pending messages have incorrect batch ID");
    });
    it("Will send messages only when allowed, but will store them in the pending state", () => {
        const outbox = getOutbox({ context: getMockContext() });
        const messages = [
            // First batch (canSendOps = true)
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            // Second batch (canSendOps = false)
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"),
        ];
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.flush();
        outbox.submit(messages[2]);
        state.canSendOps = false;
        outbox.flush();
        // First two submitted
        node_assert_1.strict.equal(state.opsSubmitted, 2);
        node_assert_1.strict.deepEqual(state.batchesSubmitted.map((x) => x.messages), [[toSubmittedMessage(messages[0]), toSubmittedMessage(messages[1])]]);
        // All three pending
        node_assert_1.strict.deepEqual(state.pendingOpContents, messages.map((message, i) => ({
            runtimeOp: message.runtimeOp,
            referenceSequenceNumber: message.referenceSequenceNumber,
            localOpMetadata: message.localOpMetadata,
            opMetadata: message.metadata,
            batchStartCsn: i === 2 ? -1 : 1, // Third batch got no CSN as it was not submitted
        })));
    });
    it("Uses legacy path for legacy contexts", () => {
        const outbox = getOutbox({ context: getMockLegacyContext() });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "3"),
        ];
        // Flush 1
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.submit(messages[2]);
        outbox.flush();
        // Flush 2
        outbox.submit(messages[3]);
        outbox.flush();
        node_assert_1.strict.equal(state.opsSubmitted, messages.length);
        node_assert_1.strict.equal(state.batchesSubmitted.length, 0);
        node_assert_1.strict.deepEqual(state.individualOpsSubmitted.length, messages.length);
        node_assert_1.strict.equal(state.deltaManagerFlushCalls, 2);
        // Note the expected CSN here is fixed to the batch's starting CSN
        const expectedMessageOrderWithCsn = [
            // Flush 1 (Main)
            [messages[0], 1],
            [messages[1], 1],
            [messages[2], 1],
            // Flush 2 (Main)
            [messages[3], 4],
        ];
        node_assert_1.strict.deepEqual(state.pendingOpContents, expectedMessageOrderWithCsn.map(([message, csn]) => ({
            runtimeOp: message.runtimeOp,
            referenceSequenceNumber: message.referenceSequenceNumber,
            localOpMetadata: message.localOpMetadata,
            opMetadata: message.metadata,
            batchStartCsn: csn,
        })));
    });
    it("Compress if compression and grouping are enabled", () => {
        const outbox = getOutbox({
            context: getMockContext(),
            compressionOptions: {
                minimumBatchSizeInBytes: 1,
                compressionAlgorithm: compressionDefinitions_js_1.CompressionAlgorithms.lz4,
            },
            opGroupingConfig: {
                groupedBatchingEnabled: true,
            },
        });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"),
        ];
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.submit(messages[2]);
        outbox.flush();
        const groupedMessages = opGroupingManager.groupBatch(toOutboundBatch(messages));
        // Submits 1 grouped batch
        node_assert_1.strict.equal(state.opsSubmitted, 1);
        node_assert_1.strict.equal(state.batchesSubmitted.length, 1);
        node_assert_1.strict.equal(state.individualOpsSubmitted.length, 0);
        node_assert_1.strict.equal(state.deltaManagerFlushCalls, 0);
        node_assert_1.strict.deepEqual(state.batchesCompressed, [groupedMessages], "Compressed batches don't match expected");
        node_assert_1.strict.deepEqual(state.batchesSubmitted.map((x) => x.messages), [[toSubmittedMessage(groupedMessages.messages[0])]], "Submitted batches don't match expected");
        // Note the expected CSN here is fixed to the batch's starting CSN
        const expectedMessageOrderWithCsn = [
            // Flush 1 (Main)
            [messages[0], 1],
            [messages[1], 1],
            [messages[2], 1],
        ];
        node_assert_1.strict.deepEqual(state.pendingOpContents, expectedMessageOrderWithCsn.map(([message, csn]) => ({
            runtimeOp: message.runtimeOp,
            referenceSequenceNumber: message.referenceSequenceNumber,
            localOpMetadata: message.localOpMetadata,
            opMetadata: message.metadata,
            batchStartCsn: csn,
        })), "Pending messages don't match expected order/properties");
    });
    it("Does not compress if the batch is smaller than the configured limit", () => {
        const outbox = getOutbox({
            context: getMockContext(),
            maxBatchSize: 1024,
            opGroupingConfig: {
                groupedBatchingEnabled: true, // Required for compression
            },
            compressionOptions: {
                minimumBatchSizeInBytes: 512,
                compressionAlgorithm: compressionDefinitions_js_1.CompressionAlgorithms.lz4,
            },
        });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"),
        ];
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.submit(messages[2]);
        outbox.flush();
        node_assert_1.strict.equal(state.opsSubmitted, 1, "Expected 1 op to be submitted for the grouped batch");
        node_assert_1.strict.equal(state.batchesSubmitted.length, 1);
        node_assert_1.strict.equal(state.individualOpsSubmitted.length, 0);
        node_assert_1.strict.equal(state.deltaManagerFlushCalls, 0);
        node_assert_1.strict.deepEqual(state.batchesCompressed, []);
        // Note the expected CSN here is fixed to the batch's starting CSN
        const expectedMessageOrderWithCsn = [
            // Flush 1 (Main)
            [messages[0], 1],
            [messages[1], 1],
            [messages[2], 1],
        ];
        node_assert_1.strict.deepEqual(state.pendingOpContents, expectedMessageOrderWithCsn.map(([message, csn]) => ({
            runtimeOp: message.runtimeOp,
            referenceSequenceNumber: message.referenceSequenceNumber,
            localOpMetadata: message.localOpMetadata,
            opMetadata: message.metadata,
            batchStartCsn: csn,
        })));
    });
    it("Throws at flush, when compression and grouping are enabled and the compressed batch is still larger than the threshold", () => {
        const outbox = getOutbox({
            context: getMockContext(),
            maxBatchSize: 1,
            compressionOptions: {
                minimumBatchSizeInBytes: 1,
                compressionAlgorithm: compressionDefinitions_js_1.CompressionAlgorithms.lz4,
            },
            opGroupingConfig: {
                groupedBatchingEnabled: true,
            },
        });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"),
        ];
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.submit(messages[2]);
        node_assert_1.strict.throws(() => outbox.flush(), (e) => "dataProcessingCodepath" in e &&
            e.dataProcessingCodepath === "CompressionInsufficient", "Expected 'CompressionInsufficient' error");
        // The batch is compressed
        node_assert_1.strict.deepEqual(state.batchesCompressed, [
            opGroupingManager.groupBatch(toOutboundBatch(messages)),
        ]);
        // The batch is not persisted
        node_assert_1.strict.deepEqual(state.pendingOpContents, []);
    });
    it("Chunks when compression is enabled, compressed batch is larger than the threshold and chunking is enabled", () => {
        const outbox = getOutbox({
            context: getMockContext(),
            maxBatchSize: 1024,
            compressionOptions: {
                minimumBatchSizeInBytes: 1,
                compressionAlgorithm: compressionDefinitions_js_1.CompressionAlgorithms.lz4,
            },
            enableChunking: true,
            chunkSizeInBytes: 2,
            opGroupingConfig: {
                groupedBatchingEnabled: true,
            },
        });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"),
        ];
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.submit(messages[2]);
        outbox.flush();
        const groupedMessages = opGroupingManager.groupBatch(toOutboundBatch(messages));
        node_assert_1.strict.deepEqual(state.batchesCompressed, [groupedMessages]);
        node_assert_1.strict.deepEqual(state.batchesSplit, [groupedMessages]);
        node_assert_1.strict.deepEqual(state.batchesSubmitted.map((x) => x.messages), [[toSubmittedMessage(groupedMessages.messages[0])]]);
        // Note the expected CSN here is fixed to the batch's starting CSN
        const expectedMessageOrderWithCsn = [
            // Flush 1 (Main)
            [messages[0], 1],
            [messages[1], 1],
            [messages[2], 1],
        ];
        node_assert_1.strict.deepEqual(state.pendingOpContents, expectedMessageOrderWithCsn.map(([message, csn]) => ({
            runtimeOp: message.runtimeOp,
            referenceSequenceNumber: message.referenceSequenceNumber,
            localOpMetadata: message.localOpMetadata,
            opMetadata: message.metadata,
            batchStartCsn: csn,
        })));
    });
    it("Does not chunk when compression and grouping are enabled, compressed batch is smaller than the threshold and chunking is enabled", () => {
        const outbox = getOutbox({
            context: getMockContext(),
            maxBatchSize: 1024,
            compressionOptions: {
                minimumBatchSizeInBytes: 1,
                compressionAlgorithm: compressionDefinitions_js_1.CompressionAlgorithms.lz4,
            },
            enableChunking: true,
            chunkSizeInBytes: 10000,
            opGroupingConfig: {
                groupedBatchingEnabled: true,
            },
        });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "2"),
        ];
        outbox.submit(messages[0]);
        outbox.submit(messages[1]);
        outbox.submit(messages[2]);
        outbox.flush();
        const groupedMessages = opGroupingManager.groupBatch(toOutboundBatch(messages));
        node_assert_1.strict.deepEqual(state.batchesCompressed, [groupedMessages]);
        node_assert_1.strict.deepEqual(state.batchesSplit, []);
        node_assert_1.strict.deepEqual(state.batchesSubmitted.map((x) => x.messages), [[toSubmittedMessage(groupedMessages.messages[0])]]);
    });
    it("Throws when an out of order message is detected", () => {
        const outbox = getOutbox({ context: getMockContext() });
        const messages = [
            {
                ...createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
                referenceSequenceNumber: 0,
            },
            {
                ...createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
                referenceSequenceNumber: 1,
            },
        ];
        currentSeqNumbers.referenceSequenceNumber = 1;
        outbox.submit(messages[0]);
        node_assert_1.strict.throws(() => outbox.submit(messages[1]), "Since we incremented referenceSequenceNumber to 1, this should throw");
    });
    it("Log at most 3 reference sequence number mismatch events", () => {
        state.isReentrant = true; // This avoids the error being thrown - but it will still log
        const outbox = getOutbox({
            maxBatchSize: Number.POSITIVE_INFINITY,
            context: getMockContext(),
        });
        for (let i = 0; i < 10; i++) {
            currentSeqNumbers.referenceSequenceNumber = 0;
            outbox.submit({
                ...createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
                referenceSequenceNumber: 0,
            });
            currentSeqNumbers.referenceSequenceNumber = 1;
            outbox.submit({
                ...createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
                referenceSequenceNumber: 1,
            });
        }
        mockLogger.assertMatch(Array.from({ length: 3 }).fill({
            eventName: "Outbox:ReferenceSequenceNumberMismatch",
        }));
    });
    it("blobAttach ops always flush before regular ops", () => {
        const outbox = getOutbox({ context: getMockContext() });
        const messages = [
            createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "0"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "2"),
            createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "3"),
            createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "4"),
        ];
        outbox.submitBlobAttach(messages[0]);
        outbox.submit(messages[1]);
        outbox.submitBlobAttach(messages[2]);
        outbox.submit(messages[3]);
        outbox.submitBlobAttach(messages[4]);
        outbox.flush();
        node_assert_1.strict.deepEqual(state.batchesSubmitted.map((x) => x.messages), [
            [
                toSubmittedMessage(messages[0], true),
                toSubmittedMessage(messages[2]),
                toSubmittedMessage(messages[4], false),
            ],
            [toSubmittedMessage(messages[1], true), toSubmittedMessage(messages[3], false)],
        ]);
        // Note the expected CSN here is fixed to the batch's starting CSN
        const expectedMessageOrderWithCsn = [
            // Flush 1 (Blob Attach)
            [messages[0], 1],
            [messages[2], 1],
            [messages[4], 1],
            // Flush 1 (Main)
            [messages[1], 4],
            [messages[3], 4],
        ];
        node_assert_1.strict.deepEqual(state.pendingOpContents, expectedMessageOrderWithCsn.map(([message, csn]) => ({
            runtimeOp: message.runtimeOp,
            referenceSequenceNumber: message.referenceSequenceNumber,
            localOpMetadata: message.localOpMetadata,
            opMetadata: message.metadata,
            batchStartCsn: csn,
        })));
    });
    describe("flush", () => {
        function validateCounts(opsSubmitted, batchesSubmitted, opsResubmitted) {
            node_assert_1.strict.strictEqual(state.opsSubmitted, opsSubmitted, "unexpected opsSubmitted");
            node_assert_1.strict.strictEqual(state.batchesSubmitted.length, batchesSubmitted, "unexpected batchesSubmitted");
            node_assert_1.strict.strictEqual(state.opsResubmitted, opsResubmitted, "unexpected opsResubmitted");
        }
        it("should not assert when flushing while reentrant with empty batches", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            // Mark context as reentrant
            state.isReentrant = true;
            // Flush with no messages - should not throw
            node_assert_1.strict.doesNotThrow(() => {
                outbox.flush();
            }, "Should not assert when flushing empty batches while reentrant");
            validateCounts(0, 0, 0);
        });
        it("should assert when flushing while reentrant with non-empty batches", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            const messages = [createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0")];
            // Submit a message (not reentrant)
            state.isReentrant = false;
            outbox.submit(messages[0]);
            // Now mark context as reentrant and try to flush - should throw
            state.isReentrant = true;
            node_assert_1.strict.throws(() => outbox.flush(), (0, internal_2.validateAssertionError)("Flushing must not happen while incoming changes are being processed"), "Should assert when flushing non-empty batches while reentrant");
            // Verify nothing was submitted
            validateCounts(0, 0, 0);
        });
        it("batch has reentrant ops, but grouped batching is off: rebase still happens", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: false,
                },
            });
            const messages = [
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            ];
            state.isReentrant = true;
            outbox.submit(messages[0]);
            outbox.submit(messages[1]);
            state.isReentrant = false;
            outbox.flush();
            validateCounts(0, 0, 2);
        });
        it("batch has reentrant ops", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            const messages = [
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            ];
            state.isReentrant = true;
            outbox.submit(messages[0]);
            outbox.submit(messages[1]);
            state.isReentrant = false;
            outbox.flush();
            validateCounts(0, 0, 2);
        });
        it("batch has a single reentrant op", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            const messages = [createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0")];
            state.isReentrant = true;
            outbox.submit(messages[0]);
            state.isReentrant = false;
            outbox.flush();
            validateCounts(0, 0, 1);
        });
        it("reentrant blobAttach ops trigger rebase", () => {
            // blobAttach batch has disableGroupedBatching:true, but reentrant ops must still rebase
            // (previously, the disableGroupedBatching flag suppressed rebase for blobAttach)
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            const messages = [
                createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "0"),
                createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "1"),
            ];
            state.isReentrant = true;
            outbox.submitBlobAttach(messages[0]);
            outbox.submitBlobAttach(messages[1]);
            state.isReentrant = false;
            outbox.flush();
            validateCounts(0, 0, 2);
        });
        it("non-reentrant blobAttach ops do not trigger rebase", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            const messages = [
                createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "0"),
                createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "1"),
            ];
            outbox.submitBlobAttach(messages[0]);
            outbox.submitBlobAttach(messages[1]);
            outbox.flush();
            validateCounts(2, 1, 0);
        });
        it("throws when resubmitting a batch that has reentrant ops", () => {
            // Resubmitting (flush with batchId) a batch that contains reentrant ops is not supported
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            state.isReentrant = true;
            outbox.submit(createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"));
            state.isReentrant = false;
            node_assert_1.strict.throws(() => outbox.flush({ batchId: "batchId", staged: false }), (e) => e.message === "0xcf2");
        });
        it("should group the batch", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            const messages = [
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            ];
            outbox.submit(messages[0]);
            outbox.submit(messages[1]);
            outbox.flush();
            validateCounts(1, 1, 0);
        });
        it("should not group the batch", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: false,
                },
            });
            const messages = [
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "0"),
                createMessage(messageTypes_js_1.ContainerMessageType.FluidDataStoreOp, "1"),
            ];
            outbox.submit(messages[0]);
            outbox.submit(messages[1]);
            outbox.flush();
            validateCounts(2, 1, 0);
        });
        it("should not group blobAttach ops", () => {
            const outbox = getOutbox({
                context: getMockContext(),
                opGroupingConfig: {
                    groupedBatchingEnabled: true,
                },
            });
            const messages = [
                createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "0"),
                createMessage(messageTypes_js_1.ContainerMessageType.BlobAttach, "1"),
            ];
            outbox.submitBlobAttach(messages[0]);
            outbox.submitBlobAttach(messages[1]);
            outbox.flush();
            validateCounts(2, 1, 0);
        });
    });
    describe("containsUserChanges", () => {
        it("returns false when all batches are empty", () => {
            const outbox = getOutbox({ context: getMockContext() });
            node_assert_1.strict.equal(outbox.containsUserChanges(), false, "Should be false when all batches are empty");
        });
        it("returns true when mainBatch has user changes", () => {
            const outbox = getOutbox({ context: getMockContext() });
            const dirtyableMessage = {
                runtimeOp: {
                    type: messageTypes_js_1.ContainerMessageType.FluidDataStoreOp,
                    contents: {
                        address: "",
                        contents: {
                            type: "op",
                            content: { address: "test-address", contents: "test-contents" },
                        },
                    },
                },
                referenceSequenceNumber: 1,
                metadata: undefined,
                localOpMetadata: {},
            };
            outbox.submit(dirtyableMessage);
            node_assert_1.strict.equal(outbox.containsUserChanges(), true, "Should be true when mainBatch has user changes");
        });
        it("returns false when mainBatch has only non-user changes", () => {
            const outbox = getOutbox({ context: getMockContext() });
            const dirtyableMessage = {
                runtimeOp: {
                    type: messageTypes_js_1.ContainerMessageType.GC,
                },
                referenceSequenceNumber: 1,
                metadata: undefined,
                localOpMetadata: {},
            };
            outbox.submit(dirtyableMessage);
            node_assert_1.strict.equal(outbox.containsUserChanges(), false, "Should be false when mainBatch has only non-user changes");
        });
        it("returns true when blobAttachBatch has user changes", () => {
            const outbox = getOutbox({ context: getMockContext() });
            const blobAttachOp = {
                runtimeOp: {
                    type: messageTypes_js_1.ContainerMessageType.BlobAttach,
                },
                referenceSequenceNumber: 1,
                metadata: undefined,
                localOpMetadata: {},
            };
            outbox.submitBlobAttach(blobAttachOp);
            node_assert_1.strict.equal(outbox.containsUserChanges(), true, "Should be true when blobAttachBatch has user changes");
        });
        it("returns false when only JIT ID allocation ops would be generated", () => {
            // With JIT ID allocation, pending IDs are tracked by the compressor, not the outbox.
            // The outbox has no queued messages, so containsUserChanges should be false.
            const outbox = getOutbox({
                context: getMockContext(),
                generateIdAllocationOp: () => ({
                    runtimeOp: {
                        type: messageTypes_js_1.ContainerMessageType.IdAllocation,
                    },
                    referenceSequenceNumber: 1,
                    metadata: undefined,
                    localOpMetadata: {},
                }),
            });
            node_assert_1.strict.equal(outbox.containsUserChanges(), false, "Should be false when no messages are queued (JIT ID alloc is not queued)");
        });
    });
});
//# sourceMappingURL=outbox.spec.js.map