package unit;

import java.util.*;

/**
 * scylladb-0001: storage_proxy::intersection O(|l1|×|l2|) linear scan in vnode range loop
 *
 * Models service/storage_proxy.cc:
 *   intersection(): std::remove_copy_if with std::find on l2 — O(|l1| × |l2|)
 *   vnode range-merge loop: calls intersection twice per vnode — O(V × RF²)
 *
 * Compile: javac -d . ScylladbTest.java
 * Run:     java unit.ScylladbTest
 */
public class ScylladbTest {

    static class HostId {
        final long id;
        HostId(long id) { this.id = id; }
        @Override public boolean equals(Object o) {
            return o instanceof HostId && ((HostId)o).id == id;
        }
        @Override public int hashCode() { return Long.hashCode(id); }
        @Override public String toString() { return "H" + id; }
    }

    // ---- DEFECTIVE: O(|l1| × |l2|) ----
    static List<HostId> intersection_defective(
            List<HostId> l1, List<HostId> l2, int[] comparisonCount) {
        List<HostId> result = new ArrayList<>();
        for (HostId a : l1) {
            // std::find on l2 — O(|l2|)
            boolean found = false;
            for (HostId b : l2) {
                comparisonCount[0]++;
                if (a.equals(b)) { found = true; break; }
            }
            if (found) result.add(a);
        }
        return result;
    }

    // Simulate vnode range-merge loop — calls intersection twice per vnode
    static int vnodeRangeMerge_defective(
            List<List<HostId>> vnodeLiveEndpoints,    // one replica set per vnode
            List<List<HostId>> vnodePreferredEndpoints,
            int[] comparisonCount) {
        int mergedRanges = 0;
        List<HostId> mergedLive = vnodeLiveEndpoints.get(0);
        List<HostId> mergedPreferred = vnodePreferredEndpoints.get(0);

        for (int i = 1; i < vnodeLiveEndpoints.size(); i++) {
            List<HostId> nextLive = vnodeLiveEndpoints.get(i);
            List<HostId> nextPreferred = vnodePreferredEndpoints.get(i);

            // Two intersection calls per vnode
            List<HostId> merged = intersection_defective(mergedLive, nextLive, comparisonCount);
            List<HostId> mergedPref = intersection_defective(mergedPreferred, nextPreferred, comparisonCount);

            if (merged.size() >= 1) {  // enough endpoints to satisfy CL
                mergedLive = merged;
                mergedPreferred = mergedPref;
                mergedRanges++;
            } else {
                break;
            }
        }
        return mergedRanges;
    }

    // ---- FIXED: O(|l1| + |l2|) using HashSet ----
    static List<HostId> intersection_fixed(
            List<HostId> l1, List<HostId> l2, int[] comparisonCount) {
        Set<HostId> s2 = new HashSet<>(l2);
        comparisonCount[0] += l2.size(); // cost of building the set
        List<HostId> result = new ArrayList<>();
        for (HostId a : l1) {
            comparisonCount[0]++;
            if (s2.contains(a)) result.add(a);
        }
        return result;
    }

    static int vnodeRangeMerge_fixed(
            List<List<HostId>> vnodeLiveEndpoints,
            List<List<HostId>> vnodePreferredEndpoints,
            int[] comparisonCount) {
        int mergedRanges = 0;
        List<HostId> mergedLive = vnodeLiveEndpoints.get(0);
        List<HostId> mergedPreferred = vnodePreferredEndpoints.get(0);

        for (int i = 1; i < vnodeLiveEndpoints.size(); i++) {
            List<HostId> nextLive = vnodeLiveEndpoints.get(i);
            List<HostId> nextPreferred = vnodePreferredEndpoints.get(i);

            List<HostId> merged = intersection_fixed(mergedLive, nextLive, comparisonCount);
            List<HostId> mergedPref = intersection_fixed(mergedPreferred, nextPreferred, comparisonCount);

            if (merged.size() >= 1) {
                mergedLive = merged;
                mergedPreferred = mergedPref;
                mergedRanges++;
            } else {
                break;
            }
        }
        return mergedRanges;
    }

    // Build vnode sets: V vnodes, RF replicas each, with rolling overlap
    static List<List<HostId>> buildVnodeEndpoints(int V, int RF, int totalNodes) {
        List<List<HostId>> result = new ArrayList<>();
        List<HostId> nodes = new ArrayList<>();
        for (int i = 0; i < totalNodes; i++) nodes.add(new HostId(i));

        for (int v = 0; v < V; v++) {
            List<HostId> replicas = new ArrayList<>();
            for (int r = 0; r < RF; r++) {
                replicas.add(nodes.get((v + r) % totalNodes));
            }
            result.add(replicas);
        }
        return result;
    }

    public static void main(String[] args) {
        int pass = 0, fail = 0;

        // -- Test 1: correctness — simple intersection
        {
            List<HostId> l1 = Arrays.asList(new HostId(1), new HostId(2), new HostId(3));
            List<HostId> l2 = Arrays.asList(new HostId(2), new HostId(3), new HostId(4));
            int[] c1 = {0}, c2 = {0};
            List<HostId> r1 = intersection_defective(l1, l2, c1);
            List<HostId> r2 = intersection_fixed(l1, l2, c2);

            if (r1.equals(r2)) {
                System.out.printf("PASS test1: intersection correctness — result=%s%n", r1);
                pass++;
            } else {
                System.out.printf("FAIL test1: defective=%s fixed=%s%n", r1, r2);
                fail++;
            }
        }

        // -- Test 2: complexity with vnode range-merge loop
        {
            int V = 256;  // vnodes (typical ScyllaDB vnode count)
            int RF = 5;   // replication factor
            int N = 10;   // nodes in cluster

            List<List<HostId>> liveEps = buildVnodeEndpoints(V, RF, N);
            List<List<HostId>> prefEps = buildVnodeEndpoints(V, RF, N);

            int[] cmpDef = {0}, cmpFix = {0};
            int mergedDef = vnodeRangeMerge_defective(liveEps, prefEps, cmpDef);
            int mergedFix = vnodeRangeMerge_fixed(liveEps, prefEps, cmpFix);

            System.out.printf("test2: defective comparisons=%d  fixed comparisons=%d  (V=%d RF=%d N=%d)%n",
                cmpDef[0], cmpFix[0], V, RF, N);
            System.out.printf("test2: merged ranges — defective=%d  fixed=%d%n", mergedDef, mergedFix);

            if (mergedDef == mergedFix && cmpFix[0] < cmpDef[0]) {
                System.out.println("PASS test2: fixed is more efficient and produces same result");
                pass++;
            } else {
                System.out.printf("FAIL test2: match=%b  efficient=%b%n",
                    mergedDef == mergedFix, cmpFix[0] < cmpDef[0]);
                fail++;
            }

            // Show expected vs actual complexity
            // Defective: per vnode, 2 intersections each O(RF²) = 2 * V * RF²
            // Fixed: per vnode, 2 intersections each O(RF) = 2 * V * RF
            int expDef = 2 * V * RF * RF;
            int expFix = 2 * V * RF * 2; // build + scan
            System.out.printf("  Expected defective ~O(2×V×RF²)=%d, actual=%d%n", expDef, cmpDef[0]);
            System.out.printf("  Expected fixed     ~O(2×V×RF)=%d,  actual=%d%n", expFix, cmpFix[0]);
        }

        // -- Test 3: empty intersection
        {
            List<HostId> l1 = Arrays.asList(new HostId(1), new HostId(2));
            List<HostId> l2 = Arrays.asList(new HostId(3), new HostId(4));
            int[] c1 = {0}, c2 = {0};
            List<HostId> r1 = intersection_defective(l1, l2, c1);
            List<HostId> r2 = intersection_fixed(l1, l2, c2);
            if (r1.isEmpty() && r2.isEmpty()) {
                System.out.println("PASS test3: empty intersection");
                pass++;
            } else {
                System.out.printf("FAIL test3: def=%s fix=%s%n", r1, r2);
                fail++;
            }
        }

        // -- Test 4: full intersection (all elements common)
        {
            List<HostId> hosts = Arrays.asList(new HostId(1), new HostId(2), new HostId(3));
            int[] c1 = {0}, c2 = {0};
            List<HostId> r1 = intersection_defective(hosts, hosts, c1);
            List<HostId> r2 = intersection_fixed(hosts, hosts, c2);
            if (r1.equals(r2) && r1.equals(hosts)) {
                System.out.println("PASS test4: full intersection");
                pass++;
            } else {
                System.out.printf("FAIL test4: def=%s fix=%s%n", r1, r2);
                fail++;
            }
        }

        System.out.printf("%nResults: %d passed, %d failed%n", pass, fail);
        if (fail > 0) System.exit(1);
    }
}