/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import { SessionPool } from "../../src/client/SessionPool";
import { TSDataType } from "../../src/utils/DataTypes";

describe("Multi-Node E2E Tests", () => {
  const IOTDB_HOST = process.env.IOTDB_HOST || "localhost";
  const IOTDB_PORT_1 = parseInt(process.env.IOTDB_PORT || "6667", 10);
  const IOTDB_PORT_2 = parseInt(process.env.IOTDB_PORT_2 || "6668", 10);
  const IOTDB_PORT_3 = parseInt(process.env.IOTDB_PORT_3 || "6669", 10);
  const IOTDB_USER = process.env.IOTDB_USER || "root";
  const IOTDB_PASSWORD = process.env.IOTDB_PASSWORD || "root";
  const IS_MULTI_NODE = process.env.MULTI_NODE === "true";

  let pool1: SessionPool;
  let pool2: SessionPool;
  let pool3: SessionPool;
  let isConnected = false;

  beforeAll(async () => {
    // Skip multi-node tests if not in multi-node environment
    if (!IS_MULTI_NODE) {
      console.log(
        "Skipping Multi-Node tests - not in multi-node environment (MULTI_NODE env var not set)",
      );
      return;
    }

    // For 3C3D setup, connect to all three DataNode ports
    // This enables true multi-node load distribution and testing
    console.log(`Connecting to IoTDB cluster:`);
    console.log(`  DataNode 1: ${IOTDB_HOST}:${IOTDB_PORT_1}`);
    console.log(`  DataNode 2: ${IOTDB_HOST}:${IOTDB_PORT_2}`);
    console.log(`  DataNode 3: ${IOTDB_HOST}:${IOTDB_PORT_3}`);

    try {
      // Create three pools, each connected to a different DataNode
      pool1 = new SessionPool(IOTDB_HOST, IOTDB_PORT_1, {
        username: IOTDB_USER,
        password: IOTDB_PASSWORD,
        maxPoolSize: 5,
        minPoolSize: 2,
      });

      pool2 = new SessionPool(IOTDB_HOST, IOTDB_PORT_2, {
        username: IOTDB_USER,
        password: IOTDB_PASSWORD,
        maxPoolSize: 5,
        minPoolSize: 2,
      });

      pool3 = new SessionPool(IOTDB_HOST, IOTDB_PORT_3, {
        username: IOTDB_USER,
        password: IOTDB_PASSWORD,
        maxPoolSize: 5,
        minPoolSize: 2,
      });

      await pool1.init();
      await pool2.init();
      await pool3.init();

      isConnected = true;
      console.log(
        `Successfully connected to IoTDB cluster on ports ${IOTDB_PORT_1}, ${IOTDB_PORT_2}, ${IOTDB_PORT_3}`,
      );
    } catch (error) {
      console.warn(
        "Could not connect to IoTDB. Multi-node E2E tests will be skipped.",
      );
      console.warn("Error:", error);
      await Promise.allSettled([
        pool1?.close(),
        pool2?.close(),
        pool3?.close(),
      ]);
    }
  }, 60000);

  // No beforeEach cleanup - tests handle "already exists" errors like other test files

  afterAll(async () => {
    if (!IS_MULTI_NODE) {
      return;
    }
    if (isConnected) {
      try {
        await pool1.executeNonQueryStatement("DROP DATABASE root.test");
      } catch (error) {
        // Ignore cleanup errors
      }

      // Close pools in parallel with timeout protection
      await Promise.allSettled([pool1.close(), pool2.close(), pool3.close()]);
    }
  }, 90000); // Increased timeout for multi-node cleanup

  test("Should initialize pools with connections to all three DataNodes", async () => {
    if (!IS_MULTI_NODE || !isConnected) {
      console.log(
        "Skipping test - not in multi-node environment or no IoTDB connection",
      );
      return;
    }

    expect(pool1.getPoolSize()).toBeGreaterThanOrEqual(2);
    expect(pool2.getPoolSize()).toBeGreaterThanOrEqual(2);
    expect(pool3.getPoolSize()).toBeGreaterThanOrEqual(2);
    console.log(
      `Pools initialized: DataNode1=${pool1.getPoolSize()}, DataNode2=${pool2.getPoolSize()}, DataNode3=${pool3.getPoolSize()} connections`,
    );
  });

  test("Should create database and timeseries on first node", async () => {
    if (!isConnected) {
      console.log("Skipping test - no IoTDB connection");
      return;
    }

    // Create database - ignore if already exists (like other tests)
    try {
      await pool1.executeNonQueryStatement("CREATE DATABASE root.test");
    } catch (error: any) {
      if (
        !error.message?.includes("already exist") &&
        !error.message?.includes("has already been created")
      ) {
        throw error;
      }
    }

    // Create timeseries - ignore if already exist
    try {
      await pool1.executeNonQueryStatement(
        "CREATE TIMESERIES root.test.device1.temperature WITH DATATYPE=FLOAT",
      );
    } catch (error: any) {
      if (!error.message.includes("already")) {
        throw error;
      }
    }

    try {
      await pool1.executeNonQueryStatement(
        "CREATE TIMESERIES root.test.device1.humidity WITH DATATYPE=FLOAT",
      );
    } catch (error: any) {
      if (!error.message.includes("already")) {
        throw error;
      }
    }
  });

  test("Should handle concurrent load distributed across all three DataNodes", async () => {
    if (!isConnected) {
      console.log("Skipping test - no IoTDB connection");
      return;
    }

    const operationsPerNode = 7;
    const promises: Promise<any>[] = [];

    // Distribute operations across all three DataNode pools
    for (let i = 0; i < operationsPerNode; i++) {
      // Insert to DataNode 1
      promises.push(
        pool1.insertTablet({
          deviceId: "root.test.device1",
          measurements: ["temperature", "humidity"],
          dataTypes: [TSDataType.FLOAT, TSDataType.FLOAT],
          timestamps: [Date.now() + i * 1000],
          values: [[20 + i * 0.1, 60 + i * 0.2]],
        }),
      );

      // Query from DataNode 2
      promises.push(
        pool2.executeQueryStatement("SELECT * FROM root.test.device1 LIMIT 10"),
      );

      // Insert to DataNode 3
      promises.push(
        pool3.insertTablet({
          deviceId: "root.test.device1",
          measurements: ["temperature", "humidity"],
          dataTypes: [TSDataType.FLOAT, TSDataType.FLOAT],
          timestamps: [Date.now() + i * 1000 + 500],
          values: [[21 + i * 0.1, 61 + i * 0.2]],
        }),
      );
    }

    const results = await Promise.all(promises);
    expect(results).toHaveLength(operationsPerNode * 3);

    // Close all SessionDataSet objects from query operations
    for (const result of results) {
      if (result && typeof result.close === "function") {
        await result.close();
      }
    }

    console.log(
      `Completed ${operationsPerNode * 3} concurrent operations across 3 DataNodes`,
    );
    console.log(
      `Pool stats: DN1=${pool1.getPoolSize()}/${pool1.getAvailableSize()}, DN2=${pool2.getPoolSize()}/${pool2.getAvailableSize()}, DN3=${pool3.getPoolSize()}/${pool3.getAvailableSize()}`,
    );
  });

  test("Should verify data replication across all DataNodes", async () => {
    if (!isConnected) {
      console.log("Skipping test - no IoTDB connection");
      return;
    }

    // Insert data through DataNode 1
    await pool1.insertTablet({
      deviceId: "root.test.device1",
      measurements: ["temperature", "humidity"],
      dataTypes: [TSDataType.FLOAT, TSDataType.FLOAT],
      timestamps: [Date.now()],
      values: [[99.9, 99.9]],
    });

    // Wait briefly for replication (optimized from 2000ms to 200ms)
    await new Promise((resolve) => setTimeout(resolve, 200));

    // Query from all three DataNodes - should see the same data
    const dataSet1 = await pool1.executeQueryStatement(
      "SELECT COUNT(*) FROM root.test.device1",
    );
    const dataSet2 = await pool2.executeQueryStatement(
      "SELECT COUNT(*) FROM root.test.device1",
    );
    const dataSet3 = await pool3.executeQueryStatement(
      "SELECT COUNT(*) FROM root.test.device1",
    );

    let count1 = 0;
    let count2 = 0;
    let count3 = 0;

    while (await dataSet1.hasNext()) {
      await dataSet1.next();
      count1++;
    }

    while (await dataSet2.hasNext()) {
      await dataSet2.next();
      count2++;
    }

    while (await dataSet3.hasNext()) {
      await dataSet3.next();
      count3++;
    }

    await dataSet1.close();
    await dataSet2.close();
    await dataSet3.close();

    expect(count1).toBeGreaterThan(0);
    expect(count2).toBeGreaterThan(0);
    expect(count3).toBeGreaterThan(0);

    console.log(
      `Data replicated across all DataNodes - verified queries from ports 6667, 6668, 6669`,
    );
  });

  test("Should handle large batch inserts across multiple DataNodes", async () => {
    if (!isConnected) {
      console.log("Skipping test - no IoTDB connection");
      return;
    }

    const batchSize = 100;
    const timestamps: number[] = [];
    const values: number[][] = [];
    const baseTime = Date.now();

    for (let i = 0; i < batchSize; i++) {
      timestamps.push(baseTime + i * 1000);
      values.push([25 + Math.random() * 5, 65 + Math.random() * 10]);
    }

    // Insert through different DataNodes
    await pool1.insertTablet({
      deviceId: "root.test.device1",
      measurements: ["temperature", "humidity"],
      dataTypes: [TSDataType.FLOAT, TSDataType.FLOAT],
      timestamps,
      values,
    });

    // Verify data from different DataNode
    const dataSet = await pool2.executeQueryStatement(
      "SELECT COUNT(*) FROM root.test.device1",
    );

    let rowCount = 0;
    while (await dataSet.hasNext()) {
      await dataSet.next();
      rowCount++;
    }

    await dataSet.close();

    expect(rowCount).toBeGreaterThan(0);
    console.log(
      `Inserted ${batchSize} records via DataNode1, queried via DataNode2`,
    );
  });

  test("Should maintain all pool healths under stress", async () => {
    if (!isConnected) {
      console.log("Skipping test - no IoTDB connection");
      return;
    }

    const initialSize1 = pool1.getPoolSize();
    const initialSize2 = pool2.getPoolSize();
    const initialSize3 = pool3.getPoolSize();

    // Execute operations across all DataNodes (reduced to 5 per pool to avoid timeout)
    const promises: Promise<any>[] = [];
    for (let i = 0; i < 5; i++) {
      promises.push(pool1.executeQueryStatement("SHOW DATABASES"));
      promises.push(pool2.executeQueryStatement("SHOW DATABASES"));
      promises.push(pool3.executeQueryStatement("SHOW DATABASES"));
    }

    const dataSets = await Promise.all(promises);

    // Close all SessionDataSets in parallel (much faster)
    await Promise.all(dataSets.map((ds) => ds.close()));

    // Pools should maintain their sizes
    expect(pool1.getPoolSize()).toBeGreaterThanOrEqual(initialSize1);
    expect(pool2.getPoolSize()).toBeGreaterThanOrEqual(initialSize2);
    expect(pool3.getPoolSize()).toBeGreaterThanOrEqual(initialSize3);
    console.log("All three pool healths maintained after stress test");
  }, 30000); // Increased timeout from default to 30s

  test("Should handle queries across all DataNodes simultaneously", async () => {
    if (!isConnected) {
      console.log("Skipping test - no IoTDB connection");
      return;
    }

    // First insert some data to ensure there's something to query
    await pool1.insertTablet({
      deviceId: "root.test.device1",
      measurements: ["temperature", "humidity"],
      dataTypes: [TSDataType.FLOAT, TSDataType.FLOAT],
      timestamps: [Date.now(), Date.now() + 1000, Date.now() + 2000],
      values: [
        [20.0, 60.0],
        [21.0, 61.0],
        [22.0, 62.0],
      ],
    });

    // Brief delay for data availability (optimized from 500ms to 100ms)
    await new Promise((resolve) => setTimeout(resolve, 100));

    // Execute queries simultaneously on all three DataNodes
    const [dataSet1, dataSet2, dataSet3] = await Promise.all([
      pool1.executeQueryStatement("SELECT * FROM root.test.device1 LIMIT 5"),
      pool2.executeQueryStatement("SELECT * FROM root.test.device1 LIMIT 5"),
      pool3.executeQueryStatement("SELECT * FROM root.test.device1 LIMIT 5"),
    ]);

    // Count rows in parallel for better performance
    const [count1, count2, count3] = await Promise.all([
      (async () => {
        let count = 0;
        while (await dataSet1.hasNext()) {
          await dataSet1.next();
          count++;
        }
        return count;
      })(),
      (async () => {
        let count = 0;
        while (await dataSet2.hasNext()) {
          await dataSet2.next();
          count++;
        }
        return count;
      })(),
      (async () => {
        let count = 0;
        while (await dataSet3.hasNext()) {
          await dataSet3.next();
          count++;
        }
        return count;
      })(),
    ]);

    // Close all in parallel
    await Promise.all([dataSet1.close(), dataSet2.close(), dataSet3.close()]);

    expect(count1).toBeGreaterThan(0);
    expect(count2).toBeGreaterThan(0);
    expect(count3).toBeGreaterThan(0);
    console.log(
      `Simultaneous queries across 3 DataNodes: DN1=${count1}, DN2=${count2}, DN3=${count3} rows`,
    );
  }, 20000); // Increased timeout to 20s

  test("Should support nodeUrls configuration for multi-node setup", async () => {
    if (!IS_MULTI_NODE) {
      console.log("Skipping test - not in multi-node environment");
      return;
    }

    // Create a pool using nodeUrls in string format (RECOMMENDED)
    const nodeUrlsPool = new SessionPool({
      nodeUrls: [
        `${IOTDB_HOST}:${IOTDB_PORT_1}`,
        `${IOTDB_HOST}:${IOTDB_PORT_2}`,
        `${IOTDB_HOST}:${IOTDB_PORT_3}`,
      ],
      username: IOTDB_USER,
      password: IOTDB_PASSWORD,
      maxPoolSize: 6,
      minPoolSize: 3,
    });

    try {
      await nodeUrlsPool.init();
      expect(nodeUrlsPool.getPoolSize()).toBeGreaterThanOrEqual(3);

      // Execute a query to verify it works
      const dataSet =
        await nodeUrlsPool.executeQueryStatement("SHOW DATABASES");
      expect(dataSet.getColumnNames()).toBeDefined();
      await dataSet.close();

      console.log("nodeUrls string format configuration working correctly");
    } finally {
      await nodeUrlsPool.close();
    }
  }, 10000); // Add explicit 10s timeout

  test("Should support nodeUrls configuration in object format", async () => {
    if (!IS_MULTI_NODE) {
      console.log("Skipping test - not in multi-node environment");
      return;
    }

    // Create a pool using nodeUrls in object format (also supported)
    const nodeUrlsPool = new SessionPool({
      nodeUrls: [
        { host: IOTDB_HOST, port: IOTDB_PORT_1 },
        { host: IOTDB_HOST, port: IOTDB_PORT_2 },
        { host: IOTDB_HOST, port: IOTDB_PORT_3 },
      ],
      username: IOTDB_USER,
      password: IOTDB_PASSWORD,
      maxPoolSize: 6,
      minPoolSize: 3,
    });

    try {
      await nodeUrlsPool.init();
      expect(nodeUrlsPool.getPoolSize()).toBeGreaterThanOrEqual(3);

      // Execute a query to verify it works
      const dataSet =
        await nodeUrlsPool.executeQueryStatement("SHOW DATABASES");
      expect(dataSet.getColumnNames()).toBeDefined();
      await dataSet.close();

      console.log("nodeUrls object format configuration working correctly");
    } finally {
      await nodeUrlsPool.close();
    }
  }, 10000); // Add explicit 10s timeout
});