import { Injectable, Logger } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository, In, createConnection, getRepository, DataSource } from 'typeorm';
import { GeoName } from './entities/geoname.entity';
import { CityBlockIpv4, CityBlockIpv6 } from './entities/city.entity';
import * as fs from 'fs';
import * as fastCsv from 'fast-csv';
import { AsnEntity } from './entities/asn.entity';
import { BreachedPassword } from '../txt-import/entities/breached-password.entity';
import { CreateBreachDataDto } from '../txt-import/dto/create-breach-data.dto';
import * as bcrypt from 'bcrypt';
import * as ipaddr from 'ipaddr.js';
import * as crypto from 'crypto';

@Injectable()
export class ImportCSVService {
  private readonly logger = new Logger('ImportCSVService');

  constructor(
    @InjectRepository(GeoName)
    private readonly geoNameRepository: Repository<GeoName>,

    @InjectRepository(CityBlockIpv4)
    private readonly cityBlockRepository: Repository<CityBlockIpv4>,

    @InjectRepository(CityBlockIpv6)
    private readonly cityBlockIpv6Repository: Repository<CityBlockIpv6>,

    @InjectRepository(AsnEntity)
    private readonly asnRepository: Repository<AsnEntity>,

    @InjectRepository(BreachedPassword)
    private readonly breachDataRepository: Repository<BreachedPassword>,

    private dataSource: DataSource,
  ) {
    this.dataSource = dataSource;
   if (!this.dataSource.isInitialized) {
      this.dataSource.initialize()
        .then(() => this.logger.log('DataSource initialized successfully'))
        .catch(error => this.logger.error('Error initializing DataSource:', error));
   }
  }

  // Helper function to delete the file
  private deleteFile(filePath: string): void {
    fs.unlink(filePath, err => {
      if (err) {
        this.logger.error(`Failed to delete file ${filePath}: ${err.message}`);
      } else {
        this.logger.log(`File deleted: ${filePath}`);
      }
    });
  }

  /**
   * Convert an IPv6 address in CIDR format to binary (Buffer)
   * @param {string} cidr The CIDR notation (e.g., 2001:218:6040::/42)
   * @returns {Buffer} The network address in binary format
   */
  private convertCidrToBinary(cidr: string): { network: Buffer; prefix: number; endIp: Buffer } {
    const [networkStr, prefixStr] = cidr.split('/');
    const prefix = parseInt(prefixStr, 10);

    // Parse the network address
    const parsedNetwork = ipaddr.parse(networkStr);
    const networkBuffer = Buffer.from(parsedNetwork.toByteArray());

    // Manually calculate end IP address by:
    // 1. Creating a bit mask based on prefix length
    // 2. Applying the mask to the network address

    // For IPv6, we have 16 bytes (128 bits)
    const bytes = parsedNetwork.toByteArray();
    const endIpBytes = [...bytes]; // Copy the network bytes

    // Calculate how many bits need to be flipped to 1
    const totalBits = 128; // IPv6 has 128 bits
    const remainingBits = totalBits - prefix;

    // Set host bits to 1
    if (remainingBits > 0) {
      // Calculate which byte to start with
      const startByteIndex = Math.floor(prefix / 8);

      // Handle bits in the first affected byte
      if (prefix % 8 !== 0) {
        const bitOffset = prefix % 8;
        endIpBytes[startByteIndex] |= 0xff >> bitOffset;
      }

      // Set all remaining bytes to 0xFF (all bits to 1)
      for (let i = startByteIndex + (prefix % 8 !== 0 ? 1 : 0); i < 16; i++) {
        endIpBytes[i] = 0xff;
      }
    }

    const endIpBuffer = Buffer.from(endIpBytes);

    return {
      network: networkBuffer,
      prefix,
      endIp: endIpBuffer,
    };
  }

  private calculateHash(data: any): string {
    // Sort keys to ensure consistent order for hashing
    const sortedData = this.sortObjectKeysRecursively(data);

    // Convert BigInt to string during serialization
    const jsonString = JSON.stringify(sortedData, (_key, value) => (typeof value === 'bigint' ? value.toString() : value));

    // Create a SHA-256 hash of the JSON string representation
    const hash = crypto.createHash('sha256');
    hash.update(jsonString);
    return hash.digest('hex');
  }

  /**
   * Sorts object keys recursively to ensure consistent order for hashing
   * @param obj Object to sort keys
   * @returns Object with sorted keys
   */
  private sortObjectKeysRecursively(obj: any): any {
    // If not an object or is null, return as is
    if (typeof obj !== 'object' || obj === null) {
      return obj;
    }

    // If array, sort keys of each element
    if (Array.isArray(obj)) {
      return obj.map(item => this.sortObjectKeysRecursively(item));
    }

    // Process object: sort keys and recursively sort values that are objects
    return Object.keys(obj)
      .sort()
      .reduce(
        (result, key) => {
          result[key] = this.sortObjectKeysRecursively(obj[key]);
          return result;
        },
        {} as Record<string, any>,
      );
  }

  async populateDataHashIPV6() {
    this.logger.log('Starting to populate data hash for existing CityBlockIpv6 records...');

    // Settings for batch processing
    const batchSize = 1000;
    let processedCount = 0;
    let updatedCount = 0;
    let hasMore = true;
    let lastId = 0;

    try {
      // Process in batches
      while (hasMore) {
        // Get a batch of records
        const records = await this.cityBlockIpv6Repository
          .createQueryBuilder('cb')
          .where('cb.id > :lastId', { lastId })
          .andWhere('cb.dataHash IS NULL') // Only records with null hash
          .orderBy('cb.id', 'ASC')
          .take(batchSize)
          .getMany();

        // Check if we have more records to process
        hasMore = records.length === batchSize;

        if (records.length === 0) {
          this.logger.log('No more records to process.');
          break;
        }

        this.logger.log(`Processing batch of ${records.length} records...`);

        // Update each record with calculated hash
        for (const record of records) {
          // Remember the last ID for pagination
          lastId = record.id;

          // Check for NaN values and replace them with null
          if (isNaN(record.latitude)) {
            record.latitude = null;
          }
          if (isNaN(record.longitude)) {
            record.longitude = null;
          }
          if (isNaN(record.accuracyRadius)) {
            record.accuracyRadius = null;
          }

          // Create data object for hashing (same structure as in import function)
          const dataForHash = {
            network: record.network.toString('hex'),
            prefixLen: record.prefixLen,
            endIp: record.endIp ? record.endIp.toString('hex') : null,
            latitude: record.latitude,
            longitude: record.longitude,
            accuracyRadius: record.accuracyRadius,
            geonameId: record.geonameId,
            registeredCountryGeonameId: record.registeredCountryGeonameId,
            representedCountryGeonameId: record.representedCountryGeonameId,
            isAnonymousProxy: record.isAnonymousProxy,
            postalCode: record.postalCode,
            isSatelliteProvider: record.isSatelliteProvider,
            isAnycast: record.isAnycast,
          };

          // Calculate hash using the same method as in the import function
          record.dataHash = this.calculateHash(dataForHash);
          processedCount++;
        }

        // Save the batch with updated hash values
        await this.cityBlockIpv6Repository.save(records);
        updatedCount += records.length;

        this.logger.log(`Updated ${records.length} records. Total updated: ${updatedCount}`);
      }

      this.logger.log(`Completed! Total records processed: ${processedCount}, updated: ${updatedCount}`);
    } catch (error) {
      this.logger.error('Error populating data hash:', error);
    }
  }

  async populateDataHashIPV4() {
    this.logger.log('Starting to populate data hash for existing CityBlockIpv4 records...');

    // Settings for batch processing
    const batchSize = 1000;
    let processedCount = 0;
    let updatedCount = 0;
    let hasMore = true;
    let lastId = 0;

    try {
      // Process in batches
      while (hasMore) {
        // Get a batch of records
        const records = await this.cityBlockRepository
          .createQueryBuilder('cb')
          .where('cb.id > :lastId', { lastId })
          .andWhere('cb.dataHash IS NULL') // Only records with null hash
          .orderBy('cb.id', 'ASC')
          .take(batchSize)
          .getMany();

        // Check if we have more records to process
        hasMore = records.length === batchSize;

        if (records.length === 0) {
          this.logger.log('No more records to process.');
          break;
        }

        this.logger.log(`Processing batch of ${records.length} records...`);

        // Update each record with calculated hash
        for (const record of records) {
          // Remember the last ID for pagination
          lastId = record.id;

          // Check for NaN values and replace them with null
          if (isNaN(record.latitude)) {
            record.latitude = null;
          }
          if (isNaN(record.longitude)) {
            record.longitude = null;
          }
          if (isNaN(record.accuracyRadius)) {
            record.accuracyRadius = null;
          }

          // Create data object for hashing (same structure as in import function)
          const dataForHash = {
            network: record.network,
            prefixLen: record.prefixLen,
            endIp: record.endIp,
            geonameId: record.geonameId,
            registeredCountryGeonameId: record.registeredCountryGeonameId,
            representedCountryGeonameId: record.representedCountryGeonameId,
            latitude: record.latitude,
            longitude: record.longitude,
            accuracyRadius: record.accuracyRadius,
            isAnonymousProxy: record.isAnonymousProxy,
            isSatelliteProvider: record.isSatelliteProvider,
            postalCode: record.postalCode,
            isAnycast: record.isAnycast,
          };

          // Calculate hash using the same method as in the import function
          record.dataHash = this.calculateHash(dataForHash);
          processedCount++;
        }

        // Save the batch with updated hash values
        await this.cityBlockRepository.save(records);
        updatedCount += records.length;

        this.logger.log(`Updated ${records.length} records. Total updated: ${updatedCount}`);
      }
    } catch (error) {
      this.logger.error('Error populating data hash:', error);
    }
  }

  async populateDataHashASN() {
    this.logger.log('Starting to populate data hash for existing ASN records...');

    // Settings for batch processing
    const batchSize = 1000;
    let processedCount = 0;
    let updatedCount = 0;
    let hasMore = true;
    let lastId = 0;

    try {
      // Process in batches
      while (hasMore) {
        // Get a batch of records
        const records = await this.asnRepository
          .createQueryBuilder('asn')
          .where('asn.id > :lastId', { lastId })
          .andWhere('asn.dataHash IS NULL') // Only records with null hash
          .orderBy('asn.id', 'ASC')
          .take(batchSize)
          .getMany();

        // Check if we have more records to process
        hasMore = records.length === batchSize;

        if (records.length === 0) {
          this.logger.log('No more records to process.');
          break;
        }

        this.logger.log(`Processing batch of ${records.length} records...`);

        // Update each record with calculated hash
        for (const record of records) {
          // Remember the last ID for pagination
          lastId = record.id;

          // Create data object for hashing (same structure as in the import function)
          const dataForHash = {
            startIp: record.startIp,
            endIp: record.endIp,
            asn: record.asn,
            name: record.name,
            domain: record.domain,
          };

          // Calculate hash using the same method as in the import function
          record.dataHash = this.calculateHash(dataForHash);
          processedCount++;
        }

        // Save the batch with updated hash values
        await this.asnRepository.save(records);
        updatedCount += records.length;

        this.logger.log(`Updated ${records.length} records. Total updated: ${updatedCount}`);
      }
    } catch (error) {
      this.logger.error('Error populating data hash:', error);
    }
  }

  async populateDataHashGeoName() {
    this.logger.log('Starting to populate data hash for existing GeoName records...');

    // Settings for batch processing
    const batchSize = 1000;
    let processedCount = 0;
    let updatedCount = 0;
    let hasMore = true;
    let lastId = 0;

    try {
      // Process in batches
      while (hasMore) {
        // Get a batch of records
        const records = await this.geoNameRepository
          .createQueryBuilder('geoname')
          .where('geoname.id > :lastId', { lastId })
          .andWhere('geoname.dataHash IS NULL') // Only records with null hash
          .orderBy('geoname.id', 'ASC')
          .take(batchSize)
          .getMany();

        // Check if we have more records to process
        hasMore = records.length === batchSize;

        if (records.length === 0) {
          this.logger.log('No more records to process.');
          break;
        }

        this.logger.log(`Processing batch of ${records.length} records...`);

        // Update each record with calculated hash
        for (const record of records) {
          // Remember the last ID for pagination
          lastId = record.id;

          // Create data object for hashing (same structure as in the import function)
          const dataForHash = {
            geonameId: record.geonameId,
            localeCode: record.localeCode,
            continentCode: record.continentCode,
            continentName: record.continentName,
            countryIsoCode: record.countryIsoCode,
            countryName: record.countryName,
            subdivision1IsoCode: record.subdivision1IsoCode,
            subdivision1Name: record.subdivision1Name,
            cityName: record.cityName,
            timeZone: record.timeZone,
            dataHash: record.dataHash,
          };

          // Calculate hash using the same method as in the import function
          record.dataHash = this.calculateHash(dataForHash);
          processedCount++;
        }

        // Save the batch with updated hash values
        await this.geoNameRepository.save(records);
        updatedCount += records.length;

        this.logger.log(`Updated ${records.length} records. Total updated: ${updatedCount}`);
      }
    } catch (error) {
      this.logger.error('Error populating data hash:', error);
    }
  }

  async autoCalculateIPv6EndIp(): Promise<{ message: string }> {
    this.logger.log('Calculating end IP addresses for IPv6 city blocks...');
    try {
      const blocks = await this.cityBlockIpv6Repository.find({ where: { endIp: null } });

      if (blocks.length === 0) {
        this.logger.log('No city blocks found with null end IP addresses.');
        return { message: 'No city blocks found with null end IP addresses.' };
      }

      for (const block of blocks) {
        // Convert Buffer to IPv6 string representation
        const ipv6 = ipaddr.fromByteArray([...new Uint8Array(block.network)]).toString();
        const networkCidr = `${ipv6}/${block.prefixLen}`;
        const { endIp } = this.convertCidrToBinary(networkCidr);

        block.latitude = isNaN(block.latitude) ? null : block.latitude;
        block.longitude = isNaN(block.longitude) ? null : block.longitude;

        block.endIp = endIp;
        await this.cityBlockIpv6Repository.save(block);
      }

      return {
        message: 'End IP addresses calculated and saved successfully',
      };
    } catch (error) {
      this.logger.error('Error calculating end IP addresses:', error);
      throw new Error(`Error calculating end IP addresses: ${error.message}`);
    }
  }

  private calculateEndIp(network: number, prefixLen: number): bigint {
    return BigInt(network) + BigInt(Math.pow(2, 32 - prefixLen)) - 1n;
  }

  private convertCidrToInteger(cidr: string): { network: number; prefix: number } {
    const [network, prefix] = cidr.split('/');

    if (!network || !prefix) {
      throw new Error('Invalid CIDR format');
    }
    const parts = network.split('.').map(part => parseInt(part, 10));
    if (parts.length !== 4 || parts.some(part => isNaN(part) || part < 0 || part > 255)) {
      throw new Error(`Invalid IP address: ${network}`);
    }

    const networkInt = (((parts[0] << 24) >>> 0) | (parts[1] << 16) | (parts[2] << 8) | parts[3]) >>> 0; // Ensure unsigned 32-bit integer

    const prefixLength = parseInt(prefix, 10);
    if (isNaN(prefixLength) || prefixLength < 0 || prefixLength > 32) {
      throw new Error(`Invalid prefix length: ${prefix}`);
    }

    return {
      network: networkInt,
      prefix: prefixLength,
    };
  }

  // Import CityBlockIpv6
  async importCityBlocksIpv6(filePath: string): Promise<void> {
    this.logger.log(`Importing CityBlocks IPv6 from file: ${filePath}`);
    return new Promise((resolve, reject) => {
      const batchSize = 1000;
      let rowsToProcess: Record<string, any>[] = [];
      let totalProcessed = 0;
      let totalSaved = 0;
      let totalSkipped = 0;

      // Function to process a batch of rows
      const processBatch = async () => {
        // console.log(`Processing batch of ${rowsToProcess.length} rows...`);
        this.logger.log(`Processing batch of ${rowsToProcess.length} rows...`);
        if (rowsToProcess.length === 0) return;

        try {
          const cityBlocksToCheck: { entity: CityBlockIpv6; hash: string }[] = [];

          // Create city blocks
          for (const row of rowsToProcess) {
            const network = row.network;
            const latitude = row.latitude ? Number(row.latitude) : null;
            const longitude = row.longitude ? Number(row.longitude) : null;
            const accuracyRadius = row.accuracy_radius ? Number(row.accuracy_radius) : null;
            const geonameId = Number(row.geoname_id);
            const registeredCountryGeonameId = Number(row.registered_country_geoname_id);
            const representedCountryGeonameId = Number(row.represented_country_geoname_id);
            const isAnonymousProxy = row.is_anonymous_proxy || false;
            const postalCode = row.postal_code || null;
            const isSatelliteProvider = row.is_satellite_provider || false;

            if (isNaN(latitude) || isNaN(longitude) || isNaN(accuracyRadius)) {
              continue;
            }

            let cidrData: { network: Buffer; prefix: number; endIp: Buffer };
            try {
              cidrData = this.convertCidrToBinary(network);
            } catch (error) {
              this.logger.error(`Invalid IPv6 address: ${network}`, error);
              continue; // Skip this record if the IPv6 is invalid
            }

            // Create data object for hashing
            const dataForHash = {
              network: cidrData.network.toString('hex'),
              prefixLen: cidrData.prefix,
              endIp: cidrData.endIp.toString('hex'),
              latitude,
              longitude,
              accuracyRadius,
              geonameId,
              registeredCountryGeonameId,
              representedCountryGeonameId,
              isAnonymousProxy,
              postalCode,
              isSatelliteProvider,
            };

            // Calculate hash of the data
            const hash = this.calculateHash(dataForHash);

            const cityBlock = this.cityBlockIpv6Repository.create({
              network: cidrData.network,
              prefixLen: cidrData.prefix,
              endIp: cidrData.endIp,
              latitude,
              longitude,
              accuracyRadius,
              geonameId,
              registeredCountryGeonameId,
              representedCountryGeonameId,
              isAnonymousProxy,
              postalCode,
              isSatelliteProvider,
              dataHash: hash,
            });

            cityBlocksToCheck.push({ entity: cityBlock, hash });
          }

          // Save city blocks
          if (cityBlocksToCheck.length > 0) {
            const hashes = cityBlocksToCheck.map(item => item.hash);

            // Find existing records with the same hashes
            const existingHashes = await this.cityBlockIpv6Repository
              .createQueryBuilder()
              .select('data_hash')
              .where('data_hash IN (:...hashes)', { hashes })
              .getRawMany();

            const existingHashSet = new Set(existingHashes.map(item => item.data_hash));

            // Filter out duplicates
            const uniqueCityBlocks = cityBlocksToCheck.filter(item => !existingHashSet.has(item.hash)).map(item => item.entity);

            const skippedCount = cityBlocksToCheck.length - uniqueCityBlocks.length;
            totalSkipped += skippedCount;

            // Save only unique records
            if (uniqueCityBlocks.length > 0) {
              await this.cityBlockIpv6Repository.save(uniqueCityBlocks);
              totalSaved += uniqueCityBlocks.length;
              // this.logger.log(
              //   `Saved ${uniqueCityBlocks.length} city blocks (skipped ${skippedCount} duplicates). Total saved: ${totalSaved}, Total skipped: ${totalSkipped}`,
              // );
            } else {
              this.logger.log(`No new records to save in this batch. Skipped ${skippedCount} duplicates. Total skipped: ${totalSkipped}`);
            }
          }
        } catch (error) {
          console.log(`Error processing batch: ${error.message}`);
          this.logger.error('Error processing city blocks batch:', error);
        }

        // Clear the batch
        rowsToProcess = [];
      };

      const stream = fs
        .createReadStream(filePath)
        .pipe(fastCsv.parse({ headers: true }))
        .on('data', row => {
          rowsToProcess.push(row);
          totalProcessed++;

          // Process in batches
          if (rowsToProcess.length >= batchSize) {
            stream.pause();
            processBatch()
              .then(() => {
                stream.resume();
              })
              .catch(error => {
                this.logger.error('Error in batch processing:', error);
                stream.resume();
              });
          }
        })
        .on('end', async () => {
          // Process any remaining rows
          try {
            await processBatch();
            // console.log('Import completed. Processed:', totalProcessed, 'rows, saved:', totalSaved, 'city blocks.');
            this.logger.log(`Import completed. Processed ${totalProcessed} rows, saved ${totalSaved} city blocks.`);
            // Delete the file after completion
            this.deleteFile(filePath);
            resolve();
          } catch (error) {
            this.logger.error('Error processing final batch:', error);
            // Delete the file in case of error
            this.deleteFile(filePath);
            reject(error);
          }
        })
        .on('error', error => {
          console.log(`Error importing CityBlocks IPv6: ${error.message}`);
          this.logger.error('Error importing CityBlocks IPv6:', error);
          // Delete the file in case of error
          this.deleteFile(filePath);
          reject(error);
        });
    });
  }

  private ipToBigInt(ip: string): bigint {
    const bytes = ipaddr.parse(ip).toByteArray();
    return bytes.reduce((acc, byte) => (acc << 8n) + BigInt(byte), 0n);
  }

  async importAsn(filePath: string): Promise<void> {
    return new Promise((resolve, reject) => {
      console.log(`Importing ASN data from file: ${filePath}`);
      this.logger.log(`Importing ASN data from file: ${filePath}`);
      const batchSize = 1000;
      let asnDataToCheck: { entity: AsnEntity; hash: string }[] = [];
      let totalProcessed = 0;
      let totalImported = 0;
      let totalSkipped = 0;

      const processBatch = async (batch: { entity: AsnEntity; hash: string }[]) => {
        if (batch.length === 0) return 0;

        try {
          this.logger.log(`Processing batch of ${batch.length} ASN records...`);

          // Extract hashes for duplicate checking
          const hashes = batch.map(item => item.hash);

          // Find existing records with the same hashes
          const existingHashes = await this.asnRepository
            .createQueryBuilder()
            .select('data_hash')
            .where('data_hash IN (:...hashes)', { hashes })
            .getRawMany();

          this.logger.log(`Found ${existingHashes.length} existing hashes.`);
          const existingHashSet = new Set(existingHashes.map(item => item.data_hash));

          // Filter out duplicates
          this.logger.log(`Filtering out duplicates...`);
          const uniqueAsns = batch.filter(item => !existingHashSet.has(item.hash)).map(item => item.entity);

          const skippedCount = batch.length - uniqueAsns.length;
          totalSkipped += skippedCount;

          // Save only unique records
          if (uniqueAsns.length > 0) {
            await this.asnRepository.save(uniqueAsns);
            this.logger.log(
              `Saved ${uniqueAsns.length} ASN records (skipped ${skippedCount} duplicates). Total saved: ${totalImported + uniqueAsns.length}, Total skipped: ${totalSkipped}`,
            );
            return uniqueAsns.length;
          } else {
            this.logger.log(`No new records to save in this batch. Skipped ${skippedCount} duplicates. Total skipped: ${totalSkipped}`);
            return 0;
          }
        } catch (error) {
          this.logger.error('Error processing ASN batch:', error);
          return 0;
        }
      };

      fs.createReadStream(filePath)
        .pipe(fastCsv.parse({ headers: true }))
        .on('data', row => {
          totalProcessed++;
          const startIp = row.start_ip;
          const endIp = row.end_ip;
          const asn = row.asn;
          const name = row.name;
          const domain = row.domain;

          if (!startIp || !endIp) {
            this.logger.warn(`Skipping invalid row: ${JSON.stringify(row)}`);
            return;
          }

          let startIpNum: bigint | null = null;
          let endIpNum: bigint | null = null;
          let startIpBin: Buffer | null = null;
          let endIpBin: Buffer | null = null;

          try {
            const parsedStartIp = ipaddr.parse(startIp);
            const parsedEndIp = ipaddr.parse(endIp);

            if (parsedStartIp.kind() === 'ipv4') {
              startIpNum = this.ipToBigInt(startIp);
              endIpNum = this.ipToBigInt(endIp);
            } else if (parsedStartIp.kind() === 'ipv6') {
              startIpBin = Buffer.from(parsedStartIp.toByteArray());
              endIpBin = Buffer.from(parsedEndIp.toByteArray());
            } else {
              this.logger.warn(`Unsupported IP kind for: ${startIp}`);
              return;
            }
          } catch (err) {
            this.logger.warn(`Invalid IP in row: ${JSON.stringify(row)}`, err);
            return;
          }

          // Create data object for hashing
          const dataForHash = {
            startIp,
            endIp,
            startIpNum: startIpNum ? startIpNum.toString() : null, // Convert BigInt to string for hashing
            endIpNum: endIpNum ? endIpNum.toString() : null,
            startIpBin: startIpBin ? startIpBin.toString('hex') : null,
            endIpBin: endIpBin ? endIpBin.toString('hex') : null,
            asn,
            name,
            domain,
          };

          // Calculate hash of the data
          const hash = this.calculateHash(dataForHash);

          const asnEntity = this.asnRepository.create({
            startIp: startIp,
            endIp: endIp,
            startIpNum: startIpNum,
            endIpNum: endIpNum,
            startIpBin: startIpBin,
            endIpBin: endIpBin,
            asn,
            name,
            domain,
            dataHash: hash,
          });

          asnDataToCheck.push({ entity: asnEntity, hash });

          // Process in batches to reduce memory usage
          if (asnDataToCheck.length >= batchSize) {
            const batchToProcess = [...asnDataToCheck]; // Create a copy
            asnDataToCheck = []; // Clear the array

            processBatch(batchToProcess)
              .then(count => {
                totalImported += count;
              })
              .catch(err => {
                this.logger.error('Error saving ASN data batch:', err);
              });
          }
        })
        .on('end', async () => {
          // Save any remaining records
          if (asnDataToCheck.length > 0) {
            try {
              const count = await processBatch(asnDataToCheck);
              totalImported += count;
              this.logger.log(`Final batch processed: ${count} ASN records saved.`);
            } catch (error) {
              this.logger.error('Error saving final ASN data batch:', error);
            }
          }
          this.logger.log(`Completed importing ASN data. Processed: ${totalProcessed}, Imported: ${totalImported}, Skipped: ${totalSkipped}`);
          // Delete the file after completion
          this.deleteFile(filePath);
          resolve();
        })
        .on('error', error => {
          this.logger.error('Error importing ASN data:', error);
          // Delete the file in case of error
          this.deleteFile(filePath);
          reject(error);
        });
    });
  }

  async importGeoNames(filePath: string): Promise<void> {
    return new Promise((resolve, reject) => {
      this.logger.log(`Importing GeoNames from file: ${filePath}`);
      const batchSize = 1000;
      let geoNamesToCheck: { entity: GeoName; hash: string }[] = [];
      let totalProcessed = 0;
      let totalImported = 0;
      let totalSkipped = 0;

      // Define processBatch function to handle deduplication and saving
      const processBatch = async (batch: { entity: GeoName; hash: string }[]): Promise<number> => {
        if (batch.length === 0) return 0;

        try {
          this.logger.log(`Processing batch of ${batch.length} GeoNames...`);

          // Extract hashes for duplicate checking
          const hashes = batch.map(item => item.hash);

          // Find existing records with the same hashes
          const existingHashes = await this.geoNameRepository
            .createQueryBuilder()
            .select('data_hash')
            .where('data_hash IN (:...hashes)', { hashes })
            .getRawMany();

          this.logger.log(`Found ${existingHashes.length} existing hashes.`);
          const existingHashSet = new Set(existingHashes.map(item => item.data_hash));

          // Filter out duplicates
          this.logger.log(`Filtering out duplicates...`);
          const uniqueGeoNames = batch.filter(item => !existingHashSet.has(item.hash)).map(item => item.entity);

          const skippedCount = batch.length - uniqueGeoNames.length;
          totalSkipped += skippedCount;

          // Save only unique records
          if (uniqueGeoNames.length > 0) {
            await this.geoNameRepository.save(uniqueGeoNames);
            this.logger.log(
              `Saved ${uniqueGeoNames.length} GeoNames (skipped ${skippedCount} duplicates). Total saved: ${totalImported + uniqueGeoNames.length}, Total skipped: ${totalSkipped}`,
            );
            return uniqueGeoNames.length;
          } else {
            this.logger.log(`No new records to save in this batch. Skipped ${skippedCount} duplicates. Total skipped: ${totalSkipped}`);
            return 0;
          }
        } catch (error) {
          this.logger.error('Error processing GeoNames batch:', error);
          return 0;
        }
      };

      fs.createReadStream(filePath)
        .pipe(fastCsv.parse({ headers: true }))
        .on('data', row => {
          totalProcessed++;
          const geonameId = Number(row.geoname_id);

          if (isNaN(geonameId)) {
            this.logger.warn(`Skipping invalid row: ${JSON.stringify(row)}`);
            return;
          }

          // Create data object for hashing
          const dataForHash = {
            geonameId,
            localeCode: row.locale_code || null,
            continentCode: row.continent_code || null,
            continentName: row.continent_name || null,
            countryIsoCode: row.country_iso_code || null,
            countryName: row.country_name || null,
            subdivision1IsoCode: row.subdivision_1_iso_code || null,
            subdivision1Name: row.subdivision_1_name || null,
            cityName: row.city_name || null,
            timeZone: row.time_zone || null,
          };

          // Calculate hash of the data
          const hash = this.calculateHash(dataForHash);

          const geoNameEntity = this.geoNameRepository.create({
            geonameId: geonameId,
            localeCode: row.locale_code || null,
            continentCode: row.continent_code || null,
            continentName: row.continent_name || null,
            countryIsoCode: row.country_iso_code || null,
            countryName: row.country_name || null,
            subdivision1IsoCode: row.subdivision_1_iso_code || null,
            subdivision1Name: row.subdivision_1_name || null,
            cityName: row.city_name || null,
            timeZone: row.time_zone || null,
            dataHash: hash,
          });

          geoNamesToCheck.push({ entity: geoNameEntity, hash });

          // Process in batches to reduce memory usage
          if (geoNamesToCheck.length >= batchSize) {
            const batchToProcess = [...geoNamesToCheck]; // Create a copy
            geoNamesToCheck = []; // Clear the array immediately for next batch

            processBatch(batchToProcess)
              .then(count => {
                totalImported += count;
              })
              .catch(err => {
                this.logger.error('Error saving GeoNames batch:', err);
              });
          }
        })
        .on('end', async () => {
          // Save any remaining records
          if (geoNamesToCheck.length > 0) {
            try {
              const count = await processBatch(geoNamesToCheck);
              totalImported += count;
              this.logger.log(`Final batch processed: ${count} GeoNames saved.`);
            } catch (error) {
              this.logger.error('Error saving final GeoNames batch:', error);
            }
          }
          this.logger.log(`Completed importing GeoNames. Processed: ${totalProcessed}, Imported: ${totalImported}, Skipped: ${totalSkipped}`);
          // Delete the file after completion
          this.deleteFile(filePath);
          resolve();
        })
        .on('error', error => {
          this.logger.error('Error importing GeoNames:', error);
          // Delete the file in case of error
          this.deleteFile(filePath);
          reject(error);
        });
    });
  }

  async importCityBlocksIpv4(filePath: string): Promise<void> {
    this.logger.log(`Importing CityBlocksIpv4 from file: ${filePath}`);
    // console.log(`Importing CityBlocksIpv4 from file: ${filePath}`);

    return new Promise((resolve, reject) => {
      const batchSize = 1000;
      let rowsToProcess: Record<string, any>[] = [];
      let totalProcessed = 0;
      let totalSaved = 0;
      let totalSkipped = 0;

      // Function to process a batch of rows
      const processBatch = async () => {
        // console.log(`Processing batch of ${rowsToProcess.length} rows...`);
        // this.logger.log(`Processing batch of ${rowsToProcess.length} rows...`);
        if (rowsToProcess.length === 0) return;

        try {
          // Create city blocks
          const cityBlocksToCheck: { entity: CityBlockIpv4; hash: string }[] = [];

          for (const row of rowsToProcess) {
            const network = row.network;
            const geonameId = Number(row.geoname_id);
            const registeredCountryGeonameId = Number(row.registered_country_geoname_id);
            const representedCountryGeonameId = Number(row.represented_country_geoname_id);
            const latitude = Number(row.latitude);
            const longitude = Number(row.longitude);
            const accuracyRadius = Number(row.accuracy_radius);
            const isAnonymousProxy = row.is_anonymous_proxy || false;
            const isSatelliteProvider = row.is_satellite_provider || false;
            const postalCode = row.postal_code || null;
            const isAnycast = row.is_anycast || false;

            if (
              isNaN(geonameId) ||
              isNaN(registeredCountryGeonameId) ||
              isNaN(representedCountryGeonameId) ||
              isNaN(latitude) ||
              isNaN(longitude) ||
              isNaN(accuracyRadius)
            ) {
              continue;
            }

            let networkBuffer: number;
            let prefixLength: number;
            try {
              const result = this.convertCidrToInteger(network);
              networkBuffer = result.network;
              prefixLength = result.prefix;
            } catch (error) {
              this.logger.error(`Invalid IPv4 address: ${network}`, error);
              continue;
            }

            const endIp = this.calculateEndIp(networkBuffer, prefixLength);

            // Create data object for hashing
            const dataForHash = {
              network: networkBuffer,
              prefixLen: prefixLength,
              endIp: endIp,
              geonameId,
              registeredCountryGeonameId,
              representedCountryGeonameId,
              latitude,
              longitude,
              accuracyRadius,
              isAnonymousProxy,
              isSatelliteProvider,
              postalCode,
              isAnycast,
            };

            // Calculate hash of the data
            const hash = this.calculateHash(dataForHash);

            const cityBlock = this.cityBlockRepository.create({
              network: networkBuffer,
              prefixLen: prefixLength,
              endIp: endIp,
              geonameId,
              registeredCountryGeonameId,
              representedCountryGeonameId,
              latitude,
              longitude,
              accuracyRadius,
              isAnonymousProxy,
              isSatelliteProvider,
              postalCode,
              isAnycast,
              dataHash: hash,
            });

            cityBlocksToCheck.push({ entity: cityBlock, hash });
          }

          // Save city blocks
          if (cityBlocksToCheck.length > 0) {
            this.logger.log(`Processing ${cityBlocksToCheck.length} city blocks...`);
            const hashes = cityBlocksToCheck.map(item => item.hash);

            // Find existing records with the same hashes
            const existingHashes = await this.cityBlockRepository
              .createQueryBuilder()
              .select('data_hash')
              .where('data_hash IN (:...hashes)', { hashes })
              .getRawMany();

            // this.logger.log(`Found ${existingHashes.length} existing hashes.`);
            // console.log(`Found ${existingHashes.length} existing hashes.`);
            const existingHashSet = new Set(existingHashes.map(item => item.data_hash));

            // Filter out duplicates
            // this.logger.log(`Filtering out duplicates...`);
            // console.log(`Filtering out duplicates...`);
            const uniqueCityBlocks = cityBlocksToCheck.filter(item => !existingHashSet.has(item.hash)).map(item => item.entity);

            const skippedCount = cityBlocksToCheck.length - uniqueCityBlocks.length;
            totalSkipped += skippedCount;

            // Save only unique records
            // this.logger.log(`Saving unique city blocks...`);
            // console.log(`Saving unique city blocks...`);
            if (uniqueCityBlocks.length > 0) {
              await this.cityBlockRepository.save(uniqueCityBlocks);
              totalSaved += uniqueCityBlocks.length;
              // console.log(
              //   `Saved ${uniqueCityBlocks.length} city blocks (skipped ${skippedCount} duplicates). Total saved: ${totalSaved}, Total skipped: ${totalSkipped}`,
              // );
              this.logger.log(
                `Saved ${uniqueCityBlocks.length} city blocks (skipped ${skippedCount} duplicates). Total saved: ${totalSaved}, Total skipped: ${totalSkipped}`,
              );
            } else {
              this.logger.log(`No new records to save in this batch. Skipped ${skippedCount} duplicates. Total skipped: ${totalSkipped}`);
            }
          }
        } catch (error) {
          console.log('Error processing city blocks batch:', error);
          this.logger.error('Error processing city blocks batch:', error);
        }

        // Clear the batch
        rowsToProcess = [];
      };

      const stream = fs
        .createReadStream(filePath)
        .pipe(fastCsv.parse({ headers: true }))
        .on('data', row => {
          rowsToProcess.push(row);
          totalProcessed++;

          // Process in batches
          if (rowsToProcess.length >= batchSize) {
            stream.pause();
            processBatch()
              .then(() => {
                stream.resume();
              })
              .catch(error => {
                this.logger.error('Error in batch processing:', error);
                stream.resume();
              });
          }
        })
        .on('end', async () => {
          // Process any remaining rows
          try {
            await processBatch();
            this.logger.log(
              `Import completed. Processed ${totalProcessed} rows, saved ${totalSaved} city blocks, skipped ${totalSkipped} duplicates.`,
            );
            // Delete the file after completion
            this.deleteFile(filePath);
            resolve();
          } catch (error) {
            this.logger.error('Error processing final batch:', error);
            // Delete the file in case of error
            this.deleteFile(filePath);
            reject(error);
          }
        })
        .on('error', error => {
          this.logger.error('Error importing CityBlocks:', error);
          // Delete the file in case of error
          this.deleteFile(filePath);
          reject(error);
        });
    });
  }

  // Helper method to process and save a batch
  private async processBatchAndClear<T>(entities: T[], repository: Repository<T>): Promise<number> {
    if (entities.length === 0) return 0;

    const entitiesCopy = [...entities];
    entities.length = 0; // Clear the original array

    try {
      const result = await repository.save(entitiesCopy);
      return result.length;
    } catch (error) {
      this.logger.error('Error saving batch:', error);
      throw error;
    }
  }

  async importCSVPassword(
    filePath: string,
    batchSize = 500,
    deleteAfterImport = false,
    passwordColumnName: string = 'password',
    sourceColumnName: string = 'source',
  ): Promise<{ totalProcessed: number; totalSaved: number }> {
    if (!fs.existsSync(filePath)) {
      this.logger.error(`File not found: ${filePath}`);
      throw new Error(`File not found: ${filePath}`);
    }

    let totalProcessed = 0;
    let totalSaved = 0;
    let rowsToProcess: CreateBreachDataDto[] = [];
    const processedPasswords = new Set<string>();
    let actualPasswordColumn: string | null = null;
    let actualSourceColumn: string | null = null;
    let headerValidated = false;

    return new Promise<{ totalProcessed: number; totalSaved: number }>((resolve, reject) => {
      try {
        const readStream = fs.createReadStream(filePath, { highWaterMark: 512 * 1024, encoding: 'utf8' });

        const processBatch = async () => {
          if (rowsToProcess.length === 0) return 0;
          const batchToProcess = [...rowsToProcess];
          rowsToProcess = [];

          try {
            const savedCount = await this.processBatch(batchToProcess);
            this.logger.log(`Processed batch of ${batchToProcess.length} records, saved ${savedCount} records`);
            return savedCount;
          } catch (err) {
            this.logger.error(`Error processing batch: ${err.message}`);
            return 0;
          }
        };

        readStream
          .pipe(fastCsv.parse({ headers: true }))
          .on('data', async row => {
            try {
              // Validate header columns on first row
              if (!headerValidated) {
                const headers = Object.keys(row);

                // Check if provided password column exists
                if (headers.includes(passwordColumnName)) {
                  actualPasswordColumn = passwordColumnName;
                } else if (headers.includes('password')) {
                  // Fall back to default 'password' column
                  actualPasswordColumn = 'password';
                  this.logger.warn(`Provided password column '${passwordColumnName}' not found, using default 'password' column`);
                } else {
                  // No password column found, throw error
                  throw new Error(
                    `No password column found. Expected '${passwordColumnName}' or 'password' but available columns are: ${headers.join(', ')}`,
                  );
                }

                // Check if provided source column exists
                if (headers.includes(sourceColumnName)) {
                  actualSourceColumn = sourceColumnName;
                } else {
                  // Use default value if source column doesn't exist
                  actualSourceColumn = null;
                  this.logger.warn(`Source column '${sourceColumnName}' not found, using default value 'Custom'`);
                }

                headerValidated = true;
                this.logger.log(`Using columns: password='${actualPasswordColumn}', source=${actualSourceColumn || 'Default "Custom"'}`);
              }

              // Extract password from the row using validated column
              const password = row[actualPasswordColumn]?.trim();

              if (!password) {
                this.logger.warn(`Skipping row with empty password: ${JSON.stringify(row)}`);
                return; // Skip this row but continue processing
              }

              // Skip duplicate passwords
              if (processedPasswords.has(password)) {
                return;
              }

              processedPasswords.add(password);
              const hashedPassword = await bcrypt.hash(password, 8);

              // Get source value if column exists, otherwise use 'Custom'
              const source = actualSourceColumn && row[actualSourceColumn]?.trim() ? row[actualSourceColumn].trim() : 'Custom';

              rowsToProcess.push({ password: hashedPassword, source });

              totalProcessed++;
              if (rowsToProcess.length >= batchSize) {
                readStream.pause();
                totalSaved += await processBatch();
                readStream.resume();
              }
            } catch (err) {
              this.logger.error(`Error processing row: ${err.message}`);
              readStream.destroy(err); // Stop processing on error
            }
          })
          .on('end', async () => {
            if (rowsToProcess.length > 0) {
              totalSaved += await processBatch();
            }
            processedPasswords.clear();
            this.logger.log(`Import completed. Total processed: ${totalProcessed}, Total saved: ${totalSaved}`);

            if (deleteAfterImport) {
              await this.deleteFile(filePath);
            }

            resolve({ totalProcessed, totalSaved });
          })
          .on('error', err => {
            this.logger.error(`Error reading CSV file: ${err.message}`);
            reject(err);
          });
      } catch (err) {
        this.logger.error(`Unexpected error: ${err.message}`);
        reject(err);
      }
    });
  }

  private async processBatch(batch: CreateBreachDataDto[]): Promise<number> {
    if (batch.length === 0) return 0;

    try {
      const passwords = batch.map(item => item.password);
      const existingPasswords = new Set<string>();

      const existingRecords = await this.breachDataRepository.find({
        select: ['password'],
        where: { password: In(passwords) },
      });

      existingRecords.forEach(record => existingPasswords.add(record.password));
      const newRecords = batch.filter(item => !existingPasswords.has(item.password));

      if (newRecords.length === 0) return 0;

      // Using insert instead of save for better performance
      await this.breachDataRepository.createQueryBuilder().insert().into(BreachedPassword).values(newRecords).execute();

      this.logger.log(`Saved ${newRecords.length} new records`);
      return newRecords.length;
    } catch (err) {
      this.logger.error(`Error saving batch: ${err.message}`);
      return 0;
    }
  }
}