/**
 * Reproducer for a bug where PIT restore fatally crashes while replaying oplog entries generated
 * by the blockReplicaSetWrites / setUserWriteBlockMode critical section machinery.
 *
 * 1. Starts a single node replica set, take a checkpoint, then run the blockReplicaSetWrites
 * enable/disable sequence (borrowed from jstests/core/administrative/block_replica_set_writes.js)
 * interleaved with user writes, and shut down.
 * 2. Starts a restore node that replays the oplog entries generated above via the ORP.
 * 3. Expect the restore node to exit cleanly; today it instead fasserts during oplog application.
 */
import SLS from "src/mongo/db/modules/atlas/jstests/disagg_storage/libs/slstest.js";
import {
    disableReplicaSetWriteBlock,
    enableReplicaSetWriteBlock,
} from "jstests/libs/block_replica_set_writes_utils.js";
import {MagicRestoreTest} from "src/mongo/db/modules/enterprise/jstests/hot_backups/libs/magic_restore_test.js";

// Create a page server and log servers
const sls = new SLS.SLSBackupRestoreTest();

const CELL1 = Object.freeze({cell: "cell1", zone: "zone1"});
const MDB_LOG_ID = 1;
const MDB_RESTORE_LOG_ID = 2;
const ENCRYPTION_KEY_FILE_PATH = SLS.createKeyFile();

// blockReplicaSetWrites is still behind a feature flag.
const featureFlagParams = {featureFlagBlockReplicaSetWrites: true};

const primaryPort = allocatePort();
let replSetConfig = {
    _id: "sourceCluster",
    version: 1,
    term: 1,
    members: [{_id: 0, host: `localhost:${primaryPort}`, priority: 1}],
};

let disaggConfig = {
    logID: NumberLong(MDB_LOG_ID),
    myZoneName: CELL1.zone,
    zones: [
        {name: CELL1.zone, cell: CELL1.cell, cellMetadataServers: [sls.getNodeByName("cms").uri]},
    ],
    encryptionKeyFilePath: ENCRYPTION_KEY_FILE_PATH,
    replSetConfig: replSetConfig,
};

jsTest.log.info("Starting up primary mongod");
const rst = SLS.startSingleNodeReplSetTest(disaggConfig, {port: primaryPort}, featureFlagParams);
let primary = rst.getPrimary();
jsTest.log.info("Waiting for primary node to become primary");
assert.soon(() => {
    const res = primary.adminCommand({"hello": 1});
    return res.me && res.me == res.primary && res.isWritablePrimary;
});
jsTest.log.info("Started primary node");

const primDB = primary.getDB("test");
const primAdminDB = primary.getDB("admin");
const collName = "testColl";

const numIterations = 20;
const numDocs = 5;

const initialOpTime = assert.commandWorked(
    primDB.runCommand({insert: collName, documents: [{_id: -1}]}),
).opTime;
const originalCheckpoint = sls.waitForCheckpoint(MDB_LOG_ID, initialOpTime);

jsTest.log.info("Running repeated drop/create/block/unblock/write cycles.");
// Repeatedly drop+recreate testColl, toggle the write block on/off, then insert and update docs.
// Each cycle's drop+create is a DDL barrier, so each cycle's enable/disable/insert/update oplog
// entries land together in a fresh oplog batch/thread assignment. Across many cycles, this gives
// many chances for the critical-section document's worker thread and testColl's worker thread to
// race against each other during replay.
let lastOpTime;
for (let i = 0; i < numIterations; i++) {
    assert.commandWorked(primDB.runCommand({drop: collName}));
    assert.commandWorked(primDB.createCollection(collName));

    enableReplicaSetWriteBlock(primAdminDB, false /* allowDeletions */, "InsufficientDiskSpace");
    disableReplicaSetWriteBlock(primAdminDB, "InsufficientDiskSpace");

    const docs = [];
    for (let d = 0; d < numDocs; d++) {
        docs.push({_id: d, x: d});
    }
    assert.commandWorked(primDB.runCommand({insert: collName, documents: docs}));

    const updates = [];
    for (let d = 0; d < numDocs; d++) {
        updates.push({q: {_id: d}, u: {$set: {x: d + 100}}});
    }
    lastOpTime = assert.commandWorked(
        primDB.runCommand({update: collName, updates: updates}),
    ).opTime;
}
const pointInTime = lastOpTime;

// We insert another document after the point-in-time we're restoring to; this should not
// appear after restore.
assert.commandWorked(primDB.runCommand({insert: collName, documents: [{_id: 999}]}));

jsTest.log.info("Attempting to restore to: " + tojson(pointInTime));

sls.restoreLog(primDB, MDB_LOG_ID, MDB_RESTORE_LOG_ID, originalCheckpoint, pointInTime);

const magicRestoreTest = new MagicRestoreTest({rst: rst, pipeDir: MongoRunner.dataDir});
MongoRunner.stopMongod(primary);

jsTest.log.info("Starting a second replica set on restored log.");
const restoreConfiguration = {
    "replicaSetConfig": magicRestoreTest.getExpectedConfig(),
    "maxCheckpointTs": SLS.lsnToTimestamp(originalCheckpoint.oplogStableTimestampLsn),
    "nodeType": "replicaSet",
    "pointInTimeTimestamp": pointInTime.ts,
};

disaggConfig = {
    logID: NumberLong(MDB_RESTORE_LOG_ID),
    myZoneName: CELL1.zone,
    zones: [
        {name: CELL1.zone, cell: CELL1.cell, cellMetadataServers: [sls.getNodeByName("cms").uri]},
    ],
    objectReadProxyURI: sls.getNodeByName("orp").uri,
    encryptionKeyFilePath: ENCRYPTION_KEY_FILE_PATH,
    replSetConfig: replSetConfig,
};

// This is expected to hang/fail today: the restoring mongod fasserts while replaying the
// testColl insert oplog entry because it (incorrectly) enforces the replica set write block
// against its own oplog replay.
magicRestoreTest.writeObjsAndRunMagicRestore(
    restoreConfiguration,
    [] /* entriesAfterBackup */,
    SLS.createMongodOptions(disaggConfig, {port: primaryPort}, featureFlagParams),
    /*isClassicMagicRestore=*/ false,
);

// Finally, we assert that all the data is there, including operations that were
// done after the initial checkpoint (i.e. operations that had to be replayed).
jsTest.log.info("Starting up primary mongod post restore");
const newRST = SLS.startSingleNodeReplSetTest(disaggConfig, {port: primaryPort}, featureFlagParams);
primary = newRST.getPrimary();
const docs = primary.getDB("test")[collName].find().toArray();
const expectedDocs = [];
for (let d = 0; d < numDocs; d++) {
    expectedDocs.push({_id: d, x: d + 100});
}
assert.sameMembers(docs, expectedDocs);

SLS.stopSingleNodeReplSetTest(newRST);
sls.cleanup();