Clean the Code

src/main/java/com/distributed/search/cluster/LeaderElection.java

@@ -6,7 +6,8 @@ import java.util.Collections;
 import java.util.List;
 
 /**
- * Handles the Leader Election logic using Zookeeper's ephemeral sequential nodes.
+ * Implements the Fault-Tolerant Leader Election using Zookeeper.
+ * Uses Ephemeral Sequential nodes to avoid 'Herd Effect'.
  */
 public class LeaderElection implements Watcher {
     private static final String ELECTION_NAMESPACE = "/election";

src/main/java/com/distributed/search/cluster/OnElectionAction.java

@@ -47,7 +47,6 @@ public class OnElectionAction implements OnElectionCallback {
     @Override
     public void onWorker() {
         try {
-            // نغير الترتيب: نشغل السيرفر أولاً
             boolean started = startGrpcServer();
 
             if (started) {

src/main/java/com/distributed/search/grpc/SearchClient.java

 package com.distributed.search.grpc;
 import java.net.InetSocketAddress;
-
 import com.distributed.search.model.*;
 import io.grpc.ManagedChannel;
-// انتبه: هذا الـ Import المظلل هو الذي يحل مشكلة الـ NameResolver في الـ Fat JAR
 import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
 import java.util.*;
 
+/**
+ * Acts as the Coordinator Client that manages connections to Workers
+ * and aggregates search results.
+ */
+
 public class SearchClient {
     private final Map<String, SearchServiceGrpc.SearchServiceBlockingStub> stubs = new HashMap<>();
     private final List<ManagedChannel> channels = new ArrayList<>();
-
+    /**
+     * Updates the active worker list and establishes gRPC channels.
+     */
     public void updateWorkers(List<String> workerAddresses) {
-        // إغلاق القنوات القديمة
         for (ManagedChannel channel : channels) {
             channel.shutdownNow();
         }
@@ -21,13 +25,10 @@ public class SearchClient {
 
         for (String address : workerAddresses) {
             try {
-                // تقسيم العنوان يدويًا
                 String[] parts = address.split(":");
                 String host = parts[0];
                 int port = Integer.parseInt(parts[1]);
 
-                // استخدام NettyChannelBuilder.forAddress حصراً
-                // هذا يتجاوز نظام الـ NameResolver تماماً ويمنع خطأ الـ 'unix'
                 ManagedChannel channel = NettyChannelBuilder.forAddress(new InetSocketAddress(host, port))
                         .usePlaintext()
                         .build();
@@ -38,18 +39,20 @@ public class SearchClient {
 
             } catch (Exception e) {
                 System.err.println("Failed to connect to " + address + ": " + e.getMessage());
-                e.printStackTrace(); // مهم: سيظهر مكان رمي الاستثناء بالضبط
+                e.printStackTrace();
             }
         }
     }
-
+    /**
+     * Orchestrates the 2-Phase Distributed Search.
+     */
     public void performSearch(List<String> terms, List<String> allFiles) {
         if (stubs.isEmpty()) {
             System.out.println("No workers available.");
             return;
         }
 
-        // --- Phase 1: Global Stats ---
+        // Phase 1: Aggregate Global Counts
         Map<String, Integer> globalTermCounts = new HashMap<>();
         int filesPerWorker = (int) Math.ceil((double) allFiles.size() / stubs.size());
         int currentFileIndex = 0;

src/main/java/com/distributed/search/grpc/SearchServiceImpl.java

@@ -9,6 +9,10 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+/**
+ * Implements gRPC Search Service on the Worker nodes.
+ */
+
 public class SearchServiceImpl extends SearchServiceGrpc.SearchServiceImplBase {
 
     private final String sharedDirectoryPath;
@@ -16,7 +20,9 @@ public class SearchServiceImpl extends SearchServiceGrpc.SearchServiceImplBase {
     public SearchServiceImpl(String sharedDirectoryPath) {
         this.sharedDirectoryPath = sharedDirectoryPath;
     }
-
+    /**
+     * Local Document Frequency Statistics.
+     */
     @Override
     public void getDocumentStats(StatRequest request, StreamObserver<StatResponse> responseObserver) {
         List<String> terms = request.getTermsList();
@@ -32,7 +38,6 @@ public class SearchServiceImpl extends SearchServiceGrpc.SearchServiceImplBase {
         for (String term : terms) {
             int docsWithTerm = 0;
             for (String docName : assignedFiles) {
-                // التعديل هنا: نمرر الـ Path مباشرة كما اقترح الحل الجديد
                 Path filePath = Paths.get(sharedDirectoryPath, docName);
                 List<String> words = TFIDFCalculator.getWordsFromDocument(filePath);
 
@@ -50,7 +55,9 @@ public class SearchServiceImpl extends SearchServiceGrpc.SearchServiceImplBase {
         responseObserver.onNext(response);
         responseObserver.onCompleted();
     }
-
+    /**
+     * Final Scoring using Global IDF received from Coordinator.
+     */
     @Override
     public void getFinalScores(CalculationRequest request, StreamObserver<SearchResponse> responseObserver) {
         List<String> terms = request.getTermsList();
@@ -67,7 +74,6 @@ public class SearchServiceImpl extends SearchServiceGrpc.SearchServiceImplBase {
         for (String docName : assignedFiles) {
             double docScore = 0.0;
 
-            // التعديل هنا أيضاً: نمرر الـ Path
             Path filePath = Paths.get(sharedDirectoryPath, docName);
             List<String> words = TFIDFCalculator.getWordsFromDocument(filePath);
 

src/main/java/com/distributed/search/logic/TFIDFCalculator.java

@@ -8,28 +8,26 @@ import java.io.IOException;
 import java.util.*;
 import java.util.stream.Collectors;
 
+/**
+ * Handles the mathematical part of the TF-IDF algorithm.
+ * Optimized for performance and cross-language support.
+ */
+
 public class TFIDFCalculator {
 
     /**
-     * تقرأ محتوى الملف وتحوّله إلى قائمة كلمات بشكل آمن
-     * - بدون regex خاطئ
-     * - يدعم Unicode (عربي / إنجليزي)
-     * - لا يرمي Exceptions تكسر gRPC
+     * Reads a file and extracts words (tokens).
+     * Uses ISO_8859_1 encoding to support various legacy text formats.
      */
     public static List<String> getWordsFromDocument(Path filePath) {
         try {
-            // قراءة الملف كنص UTF-8
             String content = Files.readString(filePath, StandardCharsets.ISO_8859_1);
 
-            // أحرف صغيرة
             content = content.toLowerCase(Locale.ROOT);
 
-            // استبدال أي شيء ليس حرفًا أو رقمًا بمسافة
-            // \p{L} = أي حرف (أي لغة)
-            // \p{N} = أي رقم
+            // Regex: keep only letters and numbers (Unicode supported)
             String cleaned = content.replaceAll("[^\\p{L}\\p{N}]+", " ");
 
-            // تقسيم على المسافات
             String[] tokens = cleaned.trim().split("\\s+");
 
             return Arrays.stream(tokens)
@@ -50,7 +48,7 @@ public class TFIDFCalculator {
     }
 
     /**
-     * Term Frequency
+     * Calculates Term Frequency (TF): (Count of term in doc) / (Total words in doc)
      */
     public static double calculateTermFrequency(List<String> words, String term) {
         if (words == null || words.isEmpty()) return 0.0;
@@ -63,7 +61,7 @@ public class TFIDFCalculator {
     }
 
     /**
-     * Inverse Document Frequency
+     * Calculates Inverse Document Frequency (IDF): log10(Total Docs / Docs containing term)
      */
     public static double calculateIdf(int totalDocuments, int documentsWithTerm) {
         if (documentsWithTerm <= 0 || totalDocuments <= 0) return 0.0;