//
// Basic $changeStream tests for operations that perform queryable encryption.
//
// @tags: [
// change_stream_does_not_expect_txns,
// assumes_unsharded_collection,
// requires_fcv_71
// ]
//
import {EncryptedClient, isEnterpriseShell} from "jstests/fle2/libs/encrypted_client_util.js";
import {canonicalizeEventForTesting, ChangeStreamTest} from "jstests/libs/query/change_stream_util.js";

if (!isEnterpriseShell()) {
    jsTestLog("Skipping test as it requires the enterprise module");
    quit();
}

const dbName = "qetestdb";
const collName = "qetestcoll";
const initialConn = db.getMongo();
const testDb = db.getSiblingDB(dbName);
const placeholderBinData0 = BinData(0, "WMdGo/tcDkE4UL6bgGYTN6oKFitgLXvhyhB9sbKxprk=");
const placeholderBinData6 = BinData(6, "WMdGo/tcDkE4UL6bgGYTN6oKFitgLXvhyhB9sbKxprk=");
const placeholderOID = ObjectId();

function replaceRandomDataWithPlaceholders(event) {
    for (let field in event) {
        if (!Object.prototype.hasOwnProperty.call(event, field)) {
            continue;
        }
        if (event[field] instanceof BinData) {
            if (event[field].subtype() === 6) {
                event[field] = placeholderBinData6;
            } else if (event[field].subtype() === 0) {
                event[field] = placeholderBinData0;
            }
        } else if (event[field] instanceof ObjectId) {
            event[field] = placeholderOID;
        } else if (typeof event[field] === "object") {
            replaceRandomDataWithPlaceholders(event[field]);
        }
    }
}
const eventModifier = function (event, expected) {
    if (event.hasOwnProperty("fullDocument") || event.hasOwnProperty("documentKey")) {
        replaceRandomDataWithPlaceholders(event);
    }
    return canonicalizeEventForTesting(event, expected);
};

testDb.dropDatabase();

let encryptedClient = new EncryptedClient(initialConn, dbName);
assert.commandWorked(
    encryptedClient.createEncryptionCollection(collName, {
        encryptedFields: {
            "fields": [
                {
                    "path": "first",
                    "bsonType": "string",
                    /* contention: 0 is required for the cleanup tests to work */
                    "queries": {"queryType": "equality", "contention": 0},
                },
            ],
        },
    }),
);

const cst = new ChangeStreamTest(testDb, {eventModifier});
const ecoll = encryptedClient.getDB()[collName];
const [escName, ecocName] = (() => {
    let names = encryptedClient.getStateCollectionNamespaces(collName);
    return [names.esc, names.ecoc];
})();

const escInsertChange = {
    documentKey: {_id: placeholderBinData0},
    fullDocument: {_id: placeholderBinData0},
    ns: {db: dbName, coll: escName},
    operationType: "insert",
};
const ecocInsertChange = {
    documentKey: {_id: placeholderOID},
    fullDocument: {_id: placeholderOID, fieldName: "first", value: placeholderBinData0},
    ns: {db: dbName, coll: ecocName},
    operationType: "insert",
};
function expectedEDCInsertChange(id, last, implicitShardKey = undefined) {
    let expected = {
        documentKey: {_id: id},
        fullDocument: {
            _id: id,
            first: placeholderBinData6,
            last: last,
            "__safeContent__": [placeholderBinData0],
        },
        ns: {db: dbName, coll: collName},
        operationType: "insert",
    };
    if (encryptedClient.useImplicitSharding && implicitShardKey) {
        expected.documentKey = Object.assign(expected.documentKey, implicitShardKey);
    }
    return expected;
}

let expectedChange = undefined;
const testValues = [
    ["frodo", "baggins"],
    ["merry", "brandybuck"],
    ["pippin", "took"],
    ["sam", "gamgee"],
    ["rosie", "gamgee"],
    ["paladin", "took"],
];

let cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
let cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});

// Test that if there are no changes, we return an empty batch.
assert.eq(0, cursor.firstBatch.length, "Cursor had changes: " + tojson(cursor));
assert.eq(0, cursordb.firstBatch.length, "Cursor had changes: " + tojson(cursordb));

jsTestLog("Testing single insert");
{
    assert.commandWorked(ecoll.einsert({_id: 0, first: "frodo", last: "baggins"}));
    expectedChange = expectedEDCInsertChange(0, "baggins", {last: "baggins"});

    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [expectedChange]});
    cst.assertNextChangesEqualUnordered({
        cursor: cursordb,
        expectedChanges: [expectedChange, escInsertChange, ecocInsertChange],
    });
    cst.assertNoChange(cursor);
    cst.assertNoChange(cursordb);
}

jsTestLog("Testing second insert");
{
    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});

    assert.commandWorked(ecoll.einsert({_id: 1, first: "merry", last: "brandybuck"}));
    expectedChange = expectedEDCInsertChange(1, "brandybuck", {last: "brandybuck"});

    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [expectedChange]});
    cst.assertNextChangesEqualUnordered({
        cursor: cursordb,
        expectedChanges: [expectedChange, escInsertChange, ecocInsertChange],
    });
    cst.assertNoChange(cursor);
    cst.assertNoChange(cursordb);
}

jsTestLog("Testing replacement update");
{
    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});

    assert.commandWorked(ecoll.ereplaceOne({last: "baggins"}, {first: "pippin", last: "took", location: "shire"}));
    expectedChange = expectedEDCInsertChange(0, "took", {last: "baggins"});
    expectedChange.operationType = "replace";
    expectedChange.fullDocument.location = "shire";

    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [expectedChange]});
    cst.assertNextChangesEqualUnordered({
        cursor: cursordb,
        expectedChanges: [expectedChange, escInsertChange, ecocInsertChange],
    });
}

jsTestLog("Testing upsert");
{
    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});

    assert.commandWorked(ecoll.ereplaceOne({last: "gamgee"}, {_id: 2, first: "sam", last: "gamgee"}, {upsert: true}));

    expectedChange = expectedEDCInsertChange(2, "gamgee", {last: "gamgee"});
    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [expectedChange]});
    cst.assertNextChangesEqualUnordered({
        cursor: cursordb,
        expectedChanges: [expectedChange, escInsertChange, ecocInsertChange],
    });
    cst.assertNoChange(cursor);
    cst.assertNoChange(cursordb);
}

jsTestLog("Testing modification update");
{
    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});
    assert.commandWorked(ecoll.eupdateOne({last: "gamgee"}, {$set: {first: "rosie"}}));
    expectedChange = {
        documentKey: {_id: 2},
        ns: {db: dbName, coll: collName},
        operationType: "update",
        updateDescription: {
            removedFields: [],
            updatedFields: {first: placeholderBinData6, "__safeContent__.1": placeholderBinData0},
            truncatedArrays: [],
        },
    };
    let safeContentPullChange = {
        documentKey: {_id: 2},
        ns: {db: dbName, coll: collName},
        operationType: "update",
        updateDescription: {
            removedFields: [],
            updatedFields: {"__safeContent__": [placeholderBinData0]},
            truncatedArrays: [],
        },
    };
    if (encryptedClient.useImplicitSharding) {
        expectedChange.documentKey.last = "gamgee";
        safeContentPullChange.documentKey.last = "gamgee";
    }

    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [expectedChange, safeContentPullChange]});
    cst.assertNextChangesEqualUnordered({
        cursor: cursordb,
        expectedChanges: [expectedChange, escInsertChange, ecocInsertChange, safeContentPullChange],
    });
    cst.assertNoChange(cursor);
    cst.assertNoChange(cursordb);
}

jsTestLog("Testing findAndModify");
{
    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});
    assert.commandWorked(
        ecoll.erunCommand({
            findAndModify: ecoll.getName(),
            query: {last: "took"},
            update: {$set: {first: "paladin"}, $unset: {location: ""}},
        }),
    );
    expectedChange = {
        documentKey: {_id: 0},
        ns: {db: dbName, coll: collName},
        operationType: "update",
        updateDescription: {
            removedFields: ["location"],
            updatedFields: {first: placeholderBinData6, "__safeContent__.1": placeholderBinData0},
            truncatedArrays: [],
        },
    };
    let safeContentPullChange = {
        documentKey: {_id: 0},
        ns: {db: dbName, coll: collName},
        operationType: "update",
        updateDescription: {
            removedFields: [],
            updatedFields: {"__safeContent__": [placeholderBinData0]},
            truncatedArrays: [],
        },
    };
    if (encryptedClient.useImplicitSharding) {
        expectedChange.documentKey.last = "took";
        safeContentPullChange.documentKey.last = "took";
    }

    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [expectedChange, safeContentPullChange]});
    cst.assertNextChangesEqualUnordered({
        cursor: cursordb,
        expectedChanges: [expectedChange, escInsertChange, ecocInsertChange, safeContentPullChange],
    });
    cst.assertNoChange(cursor);
    cst.assertNoChange(cursordb);
}

jsTestLog("Testing delete");
{
    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});
    assert.commandWorked(ecoll.edeleteOne({last: "gamgee"}));
    expectedChange = {
        documentKey: {_id: 2},
        ns: {db: dbName, coll: collName},
        operationType: "delete",
    };
    if (encryptedClient.useImplicitSharding) {
        expectedChange.documentKey.last = "gamgee";
    }
    cst.assertNextChangesEqual({cursor: cursor, expectedChanges: [expectedChange]});
    cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [expectedChange]});
    cst.assertNoChange(cursor);
    cst.assertNoChange(cursordb);
}

const ecocRenameChange = {
    operationType: "rename",
    ns: {db: dbName, coll: ecocName},
    to: {db: dbName, coll: ecocName + ".compact"},
};
const escDeleteChange = {
    operationType: "delete",
    ns: {db: dbName, coll: escName},
    documentKey: {_id: placeholderBinData0},
};
const ecocCompactDropChange = {
    operationType: "drop",
    ns: {db: dbName, coll: ecocName + ".compact"},
};

jsTestLog("Testing compact");
{
    // all non-anchors will be deleted by compact
    const deleteCount = testDb[escName].countDocuments({value: {$exists: false}});
    const numUniqueValues = testValues.length;

    encryptedClient.assertEncryptedCollectionCounts(collName, 2, numUniqueValues, numUniqueValues);

    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});

    encryptedClient.runEncryptionOperation(() => {
        assert.commandWorked(ecoll.compact());
    });
    encryptedClient.assertEncryptedCollectionCounts(collName, 2, numUniqueValues, 0);
    const anchorCount = testDb[escName].countDocuments({value: {$exists: true}});
    const nonAnchorCount = testDb[escName].countDocuments({value: {$exists: false}});
    assert.eq(anchorCount, numUniqueValues);
    assert.eq(nonAnchorCount, 0);

    cst.assertNoChange(cursor);

    escInsertChange.fullDocument.value = placeholderBinData0;
    // temp ecoc rename
    cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [ecocRenameChange]});
    // normal anchor inserts
    for (let i = 0; i < numUniqueValues; i++) {
        cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [escInsertChange]});
    }
    // non-anchor deletes
    for (let i = 0; i < deleteCount; i++) {
        cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [escDeleteChange]});
    }
    // temp ecoc drop
    cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [ecocCompactDropChange]});
    cst.assertNoChange(cursordb);
}

jsTestLog("Testing cleanup");
{
    // insert new documents for each test value, so the ECOC & ESC have documents to clean up
    for (let val of testValues) {
        assert.commandWorked(ecoll.einsert({first: val[0], last: val[1]}));
    }
    // ESC doesn't have null anchors yet, so the total delete count == ESC count before cleanup
    const deleteCount = testDb[escName].countDocuments({});
    const nonAnchorCount = testDb[escName].countDocuments({value: {$exists: false}});
    const anchorCount = deleteCount - nonAnchorCount;
    const numUniqueValues = testValues.length;

    encryptedClient.assertEncryptedCollectionCounts(
        collName,
        2 + numUniqueValues,
        numUniqueValues * 2,
        numUniqueValues,
    );
    assert.eq(anchorCount, numUniqueValues);
    assert.eq(nonAnchorCount, numUniqueValues);

    cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: testDb[collName]});
    cursordb = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: 1});

    encryptedClient.runEncryptionOperation(() => {
        assert.commandWorked(ecoll.cleanup());
    });
    encryptedClient.assertEncryptedCollectionCounts(collName, 2 + numUniqueValues, numUniqueValues, 0);
    encryptedClient.assertESCNonAnchorCount(collName, 0);

    cst.assertNoChange(cursor);

    // temp ecoc rename
    cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [ecocRenameChange]});
    // null anchor inserts
    escInsertChange.fullDocument.value = placeholderBinData0;
    for (let i = 0; i < anchorCount; i++) {
        cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [escInsertChange]});
    }
    // non-anchors and regular anchors are deleted from ESC
    for (let i = 0; i < deleteCount; i++) {
        cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [escDeleteChange]});
    }
    // temp ecoc.compact drop
    cst.assertNextChangesEqual({cursor: cursordb, expectedChanges: [ecocCompactDropChange]});
    cst.assertNoChange(cursordb);
}

cst.cleanUp();