'use strict';

const fs = require('fs');
const net = require('net');
const os = require('os');
const path = require('path');
const { spawn } = require('child_process');
const { MongoClient } = require('E:\\aisfor\\node_modules\\mongodb');
const { Server: SocketServer } = require('E:\\aisfor\\node_modules\\socket.io');
const { createAdapter: createMongoSocketAdapter } = require('E:\\aisfor\\node_modules\\@socket.io\\mongo-adapter');
const MallMongoSessionStore = require('E:\\aismall\\services\\mongoSessionStore');

const promotionGateModules = [
  ['aisfor', 'E:\\aisfor\\services\\promotedActiveGate'],
  ['aistalk', 'F:\\aistalk\\services\\promotedActiveGate'],
  ['aisfor-group', 'E:\\aisfor-group\\services\\promotedActiveGate'],
  ['aismall', 'E:\\aismall\\services\\promotedActiveGate'],
  ['plai-sharing', 'F:\\plai-sharing\\services\\promotedActiveGate']
];

const execute = process.argv.includes('--execute');
const mongodPath = 'C:\\Program Files\\MongoDB\\Server\\8.2\\bin\\mongod.exe';
const replicaSetName = 'plaiSyntheticHaRehearsal';
const ports = [57117, 57118, 57119];
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
let temporaryRoot = '';
const children = new Map();
const clients = new Set();
const socketServers = new Set();
const promotionGates = new Set();

function portOpen(port) {
  return new Promise(resolve => {
    const socket = net.createConnection({ host: '127.0.0.1', port });
    const done = value => { socket.destroy(); resolve(value); };
    socket.setTimeout(500);
    socket.once('connect', () => done(true));
    socket.once('timeout', () => done(false));
    socket.once('error', () => done(false));
  });
}

async function waitForPort(port, expected, timeoutMs = 30_000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (await portOpen(port) === expected) return;
    await sleep(200);
  }
  throw new Error(`MONGO_REHEARSAL_PORT_STATE_TIMEOUT:${port}:${expected}`);
}

function startMongod(port, dbPath) {
  const child = spawn(mongodPath, [
    '--replSet', replicaSetName,
    '--port', String(port),
    '--dbpath', dbPath,
    '--bind_ip', '127.0.0.1',
    '--oplogSize', '64',
    '--quiet'
  ], { windowsHide: true, stdio: 'ignore' });
  children.set(port, child);
  return child;
}

async function connect(uri, options = {}) {
  const client = new MongoClient(uri, {
    serverSelectionTimeoutMS: 2000,
    connectTimeoutMS: 2000,
    ...options
  });
  await client.connect();
  clients.add(client);
  return client;
}

async function waitForPrimary(uri, excludedPort = 0, timeoutMs = 90_000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    let client;
    try {
      client = await connect(uri);
      const hello = await client.db('admin').command({ hello: 1 });
      const primary = String(hello.primary || '');
      const port = Number(primary.split(':').pop() || 0);
      if (hello.isWritablePrimary === true && port && port !== excludedPort) return { client, port, hello };
    } catch (_) {
      if (client) { clients.delete(client); await client.close().catch(() => {}); }
    }
    await sleep(500);
  }
  throw new Error('MONGO_REHEARSAL_PRIMARY_ELECTION_TIMEOUT');
}

async function stopPort(port) {
  const child = children.get(port);
  if (!child) return;
  if (child.exitCode === null) {
    const exited = new Promise(resolve => child.once('exit', resolve));
    child.kill();
    await Promise.race([exited, sleep(10_000)]);
  }
  await waitForPort(port, false, 15_000);
  children.delete(port);
}

async function proveSocketBroadcast(sender, receiver, phase, timeoutMs = 30_000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const nonce = `${phase}-${Date.now()}-${Math.random().toString(16).slice(2)}`;
    let handler;
    const received = new Promise(resolve => {
      handler = payload => {
        if (payload?.nonce === nonce && payload?.phase === phase) resolve(true);
      };
      receiver.on('plai-synthetic-cluster-event', handler);
    });
    sender.serverSideEmit('plai-synthetic-cluster-event', { nonce, phase });
    const ok = await Promise.race([received, sleep(1000).then(() => false)]);
    receiver.off('plai-synthetic-cluster-event', handler);
    if (ok) return true;
  }
  throw new Error(`MONGO_SOCKET_ADAPTER_BROADCAST_TIMEOUT:${phase}`);
}

async function provePromotionGates(database) {
  const results = [];
  for (const [serviceId, modulePath] of promotionGateModules) {
    const { createPromotedActiveGate } = require(modulePath);
    const gate = createPromotedActiveGate({
      serviceId,
      env: {
        PLAI_SERVICE_MODE: 'PROMOTED_ACTIVE',
        PLAI_REPLICA_ID: `synthetic-${serviceId}`,
        PLAI_PROMOTION_LEASE_TTL_MS: '120000'
      },
      onLeaseLost() {}
    });
    const result = await gate.verifyAndOpen({
      mongoose: { connection: { readyState: 1, db: database } },
      sharedSessionStoreReady: true,
      socketAdapter: { enabled: true },
      requireSocketAdapter: serviceId !== 'aisfor-group'
    });
    if (!result.ready || result.replicaSet !== replicaSetName) throw new Error(`PROMOTION_GATE_LIVE_PROOF_FAILED:${serviceId}`);
    promotionGates.add(gate);
    results.push({ serviceId, gate, fencingToken: result.fencingToken });
  }
  return results;
}

function sessionOperation(store, method, ...args) {
  return new Promise((resolve, reject) => store[method](...args, (error, value) => error ? reject(error) : resolve(value)));
}

async function safeRemoveTemporaryRoot() {
  if (!temporaryRoot || !fs.existsSync(temporaryRoot)) return;
  const resolved = path.resolve(temporaryRoot);
  const tempBase = `${path.resolve(os.tmpdir())}${path.sep}`.toLowerCase();
  if (!resolved.toLowerCase().startsWith(tempBase)
    || !path.basename(resolved).startsWith('plai-mongo-ha-')) {
    throw new Error('MONGO_REHEARSAL_TEMP_CLEANUP_TARGET_REJECTED');
  }
  let lastError;
  for (let attempt = 0; attempt < 20; attempt += 1) {
    try {
      fs.rmSync(resolved, { recursive: true, force: true });
      temporaryRoot = '';
      return;
    } catch (error) {
      lastError = error;
      if (!['EPERM', 'EBUSY', 'ENOTEMPTY'].includes(String(error.code || ''))) throw error;
      await sleep(250);
    }
  }
  throw lastError;
}

async function cleanup() {
  for (const gate of promotionGates) gate.stop();
  promotionGates.clear();
  for (const io of [...socketServers]) {
    await Promise.resolve(io.of('/').adapter?.close?.()).catch(() => {});
    io.removeAllListeners();
  }
  socketServers.clear();
  for (const client of [...clients]) await client.close().catch(() => {});
  clients.clear();
  for (const port of [...children.keys()]) await stopPort(port).catch(() => {});
  await safeRemoveTemporaryRoot();
}

(async () => {
  if (!fs.existsSync(mongodPath)) throw new Error('MONGOD_BINARY_NOT_FOUND');
  const occupied = [];
  for (const port of ports) if (await portOpen(port)) occupied.push(port);
  process.stdout.write(`${JSON.stringify({
    ok: occupied.length === 0,
    mode: execute ? 'EXECUTE_SYNTHETIC_THREE_NODE_MONGO_FAILOVER' : 'PLAN_ONLY_NO_PROCESSES',
    replicaNodes: 3,
    ports,
    portsAvailable: occupied.length === 0,
    productionMongoTargeted: false,
    productionDatabaseRead: false,
    syntheticDataOnly: true,
    temporaryDataAutoDeleted: true
  }, null, 2)}\n`);
  if (!execute || occupied.length) return;

  temporaryRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'plai-mongo-ha-'));
  for (const port of ports) {
    const dbPath = path.join(temporaryRoot, `node-${port}`);
    fs.mkdirSync(dbPath, { recursive: true });
    startMongod(port, dbPath);
  }
  for (const port of ports) await waitForPort(port, true);

  const direct = await connect(`mongodb://127.0.0.1:${ports[0]}/?directConnection=true`);
  await direct.db('admin').command({
    replSetInitiate: {
      _id: replicaSetName,
      members: ports.map((port, index) => ({ _id: index, host: `127.0.0.1:${port}`, priority: 1 }))
    }
  });
  const uri = `mongodb://${ports.map(port => `127.0.0.1:${port}`).join(',')}/plaiSynthetic?replicaSet=${replicaSetName}`;
  const first = await waitForPrimary(uri);
  const collection = first.client.db('plaiSynthetic').collection('continuityProof');
  await collection.insertOne({ sequence: 1, payload: 'synthetic-before-failover' }, { writeConcern: { w: 'majority', wtimeoutMS: 10_000 } });
  const beforeCount = await collection.countDocuments({});
  if (beforeCount !== 1) throw new Error('MONGO_REHEARSAL_INITIAL_MAJORITY_WRITE_MISSING');
  const promotionGatesBeforeFailure = await provePromotionGates(first.client.db('plaiSynthetic'));
  const sessionMongoose = {
    connection: {
      readyState: 1,
      collection: name => first.client.db('plaiSynthetic').collection(name)
    }
  };
  const sessionStore = new MallMongoSessionStore({ mongoose: sessionMongoose, collectionName: 'mallSessionsSynthetic' });
  await sessionOperation(sessionStore, 'set', 'synthetic-session', { cookie: {}, user: { id: 'synthetic-user' } });
  const sessionBeforeFailure = await sessionOperation(sessionStore, 'get', 'synthetic-session');
  if (sessionBeforeFailure?.user?.id !== 'synthetic-user') throw new Error('MONGO_SESSION_INITIAL_WRITE_MISSING');

  const socketClientA = await connect(uri);
  const socketClientB = await connect(uri);
  const socketCollectionA = socketClientA.db('plaiSynthetic').collection('socketIoAdapterEvents');
  const socketCollectionB = socketClientB.db('plaiSynthetic').collection('socketIoAdapterEvents');
  await socketCollectionA.createIndex({ createdAt: 1 }, {
    name: 'plaiSyntheticSocketEventTtl', expireAfterSeconds: 60
  });
  const ioA = new SocketServer();
  const ioB = new SocketServer();
  socketServers.add(ioA);
  socketServers.add(ioB);
  ioA.adapter(createMongoSocketAdapter(socketCollectionA, { addCreatedAtField: true }));
  ioB.adapter(createMongoSocketAdapter(socketCollectionB, { addCreatedAtField: true }));
  await sleep(1000);
  const socketBroadcastBeforeFailure = await proveSocketBroadcast(ioA, ioB, 'before-primary-failure');

  const failedPrimaryPort = first.port;
  await stopPort(failedPrimaryPort);
  const second = await waitForPrimary(uri, failedPrimaryPort);
  sessionMongoose.connection.collection = name => second.client.db('plaiSynthetic').collection(name);
  const promotedCollection = second.client.db('plaiSynthetic').collection('continuityProof');
  await promotedCollection.insertOne({ sequence: 2, payload: 'synthetic-after-failover' }, { writeConcern: { w: 'majority', wtimeoutMS: 10_000 } });
  const rows = await promotedCollection.find({}, { projection: { _id: 0, sequence: 1 } }).sort({ sequence: 1 }).toArray();
  if (rows.length !== 2 || rows[0].sequence !== 1 || rows[1].sequence !== 2) {
    throw new Error('MONGO_REHEARSAL_FAILOVER_DATA_CONTINUITY_MISSING');
  }
  const socketBroadcastAfterFailure = await proveSocketBroadcast(ioB, ioA, 'after-primary-failure');
  const sessionAfterFailure = await sessionOperation(sessionStore, 'get', 'synthetic-session');
  if (sessionAfterFailure?.user?.id !== 'synthetic-user') throw new Error('MONGO_SESSION_FAILOVER_CONTINUITY_MISSING');
  await sessionOperation(sessionStore, 'touch', 'synthetic-session', { cookie: { maxAge: 60_000 } });
  await Promise.all(promotionGatesBeforeFailure.map(row => row.gate.renewNow()));
  const promotionGatesAfterFailure = promotionGatesBeforeFailure.filter(row => row.gate.ready);
  const electedDifferentNode = second.port !== failedPrimaryPort;
  await cleanup();
  process.stdout.write(`${JSON.stringify({
    ok: true,
    status: 'SYNTHETIC_MONGO_THREE_NODE_MAJORITY_FAILOVER_PROVEN_AND_CLEANED',
    replicaNodes: 3,
    majorityWriteBeforeFailure: true,
    primaryProcessStopped: true,
    differentPrimaryElected: electedDifferentNode,
    majorityWriteAfterFailure: true,
    preFailureRecordPreserved: true,
    syntheticRecordCount: rows.length,
    socketMongoAdapterVersion: '0.4.0',
    socketBroadcastBeforeFailure,
    socketBroadcastAfterFailure,
    promotedActiveGatesBeforeFailure: promotionGatesBeforeFailure.length,
    promotedActiveGatesAfterFailure: promotionGatesAfterFailure.length,
    promotionLeaseMajorityWriteVerified: true,
    promotionLeaseRenewedAfterPrimaryElection: promotionGatesAfterFailure.length === promotionGateModules.length,
    promotionFencingTokensIssued: promotionGatesBeforeFailure.every(row => row.fencingToken >= 1),
    mongoSessionWriteBeforeFailure: true,
    mongoSessionReadAndTouchAfterFailure: true,
    allMongodProcessesStopped: true,
    allPortsClosed: true,
    temporaryDataDeleted: true,
    productionMongoChanged: false
  }, null, 2)}\n`);
})().catch(async error => {
  await cleanup().catch(() => {});
  console.error(JSON.stringify({ ok: false, error: String(error.message || error) }, null, 2));
  process.exitCode = 1;
});
