package Coordinator_Worker;

import ch.qos.logback.classic.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

// Worker class representing a node that processes search requests from a coordinator
public class Worker {
    private static final Logger logger = (Logger) LoggerFactory.getLogger(Worker.class);

    private String nodeName;
    private String ipAddress;
    private String userName;
    private int port;

    // Constructor to initialize Worker with the full path of the node
    public Worker(String fullPathNode,int port) {
        this.nodeName = fullPathNode;
        this.port=port;
    }

    // Method to start the Worker and listen for search requests
    public static void startWorker(int port) {
        ServerSocket serverSocket = null;
        try {
            // Step 1: Set up server socket to listen for connections
            serverSocket = new ServerSocket(port);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        while (true) {
            try {
                logger.info("Waiting Request at port: "+port);
                // Step 2: Accept connection from coordinator
                Socket coordinatorSocket = serverSocket.accept();

                // Step 3: Receive data from coordinator
                DataObject data = Communication.receiveData(coordinatorSocket);
                logger.info("Received data from coordinator: " + data.getDataString() + ", " + data.getFileNames());

                // Step 4: Process the data using IDF algorithm
                ConcurrentLinkedQueue<Document> response = new ConcurrentLinkedQueue<>();
                response.addAll(IDF(data));

                // Step 5: Send the result back to the coordinator
                Communication.sendResult(coordinatorSocket, response);

                // Step 6: Close sockets
                coordinatorSocket.close();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                throw new RuntimeException(e);
            }
        }
    }

    // Function to calculate IDF for each term in the search query across files
    private static ConcurrentLinkedQueue<Document> IDF(DataObject data) {
        ConcurrentLinkedQueue<Document> documents = new ConcurrentLinkedQueue<>();
        String searchQuery = data.getDataString();
        String[] terms = searchQuery.split("\\s+");

        // Step 2: Use ExecutorService for parallel processing
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

        try {
            for (String fileName : data.getFileNames()) {
                File file = new File("C:\\Users\\Abdullah\\Downloads\\network-communication-master\\Cordinator_Worker\\src\\main\\resources\\files\\" + fileName);
                Runnable task = () -> processFile(file, terms, documents);
                executorService.submit(task);
            }
            executorService.awaitTermination(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            // Step 5: Shutdown the ExecutorService
            executorService.shutdown();
        }

        // Step 6: Return the final IDF map
        return documents;
    }

    // Function to process each file and calculate term frequencies
    private static void processFile(File file, String[] terms, ConcurrentLinkedQueue<Document> documents) {
        Document document = new Document(file.getName());
        double totalWords = 0;

        try {
            // Read content from the file
            String content = new String(Files.readAllBytes(file.toPath()));
            String[] words = content.split("\\s+");

            // Calculate term frequencies in the file
            for (String term : terms) {
                double frequency = 0;
                for (String word : words) {
                    if (word.equals(term)) {
                        frequency++;
                    }
                    totalWords++;
                }
                Double d = frequency / totalWords;
                document.setTf(term, d);
                documents.offer(document);
            }
        } catch (IOException e) {
            e.printStackTrace();
            // Handle file reading exception
        }
    }

    // Getter and setter methods for ipAddress, userName, and nodeName
    public void setIpAddress(String ipAddress) {
        this.ipAddress = ipAddress;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public String getNodeName() {
        return nodeName;
    }

    public static List<String> getIpAddress(List<Worker> workers) {
        List<String> allAddress = new ArrayList<>();
        for (Worker worker : workers) {
            allAddress.add(worker.ipAddress);
        }
        return allAddress;
    }

    public String getIpAddress() {
        return ipAddress;
    }

    public String getUserName() {
        return userName;
    }

    public int getPort() {
        return port;
    }
}
