import java.util.*;

/**
 * CWE-407 unit tests for Apache Pulsar.
 *
 * Three defects measured:
 *   pulsar-0001: ModularLoadManagerImpl.reapDeadBrokerPreallocations()
 *                aliveBrokers is List<String> from listLocks(); O(B^2) lookup
 *   pulsar-0002: PersistentTopic.removeOrphanReplicationCursors() and
 *                checkReplicationStatus(); configuredClusters is List<String>
 *                and .contains() is called inside a loop over cursors/replicators
 *   pulsar-0003: NamespacesBase.internalGetTopicHashPositionsAsync();
 *                allTopicsInThisBundle is List<String> from getOwnedTopicListForNamespaceBundle();
 *                .contains() called inside a for loop over topics query argument
 */
public class PulsarTest {

    // -----------------------------------------------------------------------
    // pulsar-0001: reapDeadBrokerPreallocations O(B^2)
    //
    // Pattern:
    //   for (String broker : loadData.getBrokerData().keySet()) {
    //       if (!aliveBrokers.contains(broker)) { ... }  // aliveBrokers is List
    //   }
    // -----------------------------------------------------------------------

    static long defectReapDeadBrokers(List<String> knownBrokers, List<String> aliveBrokers) {
        // Simulates: for each known broker, scan aliveBrokers list for membership
        long ops = 0;
        List<String> dead = new ArrayList<>();
        for (String broker : knownBrokers) {
            ops++;
            if (!aliveBrokers.contains(broker)) {  // O(N) linear scan per broker
                dead.add(broker);
            }
        }
        return ops;
    }

    static long fixedReapDeadBrokers(List<String> knownBrokers, List<String> aliveBrokers) {
        Set<String> aliveSet = new HashSet<>(aliveBrokers);
        long ops = 0;
        List<String> dead = new ArrayList<>();
        for (String broker : knownBrokers) {
            ops++;
            if (!aliveSet.contains(broker)) {   // O(1) hash lookup
                dead.add(broker);
            }
        }
        return ops;
    }

    static void testPulsar0001() {
        int B = 500; // number of known brokers in loadData
        List<String> knownBrokers = new ArrayList<>();
        List<String> aliveBrokers = new ArrayList<>();
        for (int i = 0; i < B; i++) {
            knownBrokers.add("broker-" + i);
        }
        // Half are alive (worst-case scan length for the rest)
        for (int i = 0; i < B / 2; i++) {
            aliveBrokers.add("broker-" + i);
        }

        int REPS = 200;

        // Warm up
        for (int i = 0; i < 5; i++) {
            defectReapDeadBrokers(knownBrokers, aliveBrokers);
            fixedReapDeadBrokers(knownBrokers, aliveBrokers);
        }

        long t0 = System.nanoTime();
        for (int r = 0; r < REPS; r++) {
            defectReapDeadBrokers(knownBrokers, aliveBrokers);
        }
        long defectNs = System.nanoTime() - t0;

        long t1 = System.nanoTime();
        for (int r = 0; r < REPS; r++) {
            fixedReapDeadBrokers(knownBrokers, aliveBrokers);
        }
        long fixedNs = System.nanoTime() - t1;

        double ratio = (double) defectNs / fixedNs;
        System.out.printf("pulsar-0001 reapDeadBrokerPreallocations  B=%d  defect=%.1fms  fixed=%.1fms  ratio=%.1fx%n",
                B,
                defectNs / 1e6 / REPS,
                fixedNs  / 1e6 / REPS,
                ratio);

        if (ratio < 3.0) {
            throw new RuntimeException("pulsar-0001 FAIL: expected ratio >= 3.0, got " + ratio);
        }
        System.out.println("pulsar-0001 PASS");
    }

    // -----------------------------------------------------------------------
    // pulsar-0002: PersistentTopic replication cluster list scan O(C*R)
    //
    // Pattern (removeOrphanReplicationCursors):
    //   List<String> replicationClusters = topicPolicies.getReplicationClusters().get();
    //   for (ManagedCursor cursor : ledger.getCursors()) {
    //       if (!replicationClusters.contains(remoteCluster)) { ... }  // O(C)
    //   }
    //
    // Pattern (checkReplicationStatus):
    //   List<String> configuredClusters = topicPolicies.getReplicationClusters().get();
    //   replicators.forEach((cluster, replicator) -> {
    //       if (!configuredClusters.contains(cluster)) { ... }  // O(C)
    //   });
    // -----------------------------------------------------------------------

    static long defectReplicationCursorScan(List<String> replicationClusters, List<String> cursors) {
        long removes = 0;
        for (String cursor : cursors) {
            // simulates: if (!replicationClusters.contains(remoteCluster))
            if (!replicationClusters.contains(cursor)) {
                removes++;
            }
        }
        return removes;
    }

    static long fixedReplicationCursorScan(List<String> replicationClusters, List<String> cursors) {
        Set<String> clusterSet = new HashSet<>(replicationClusters);
        long removes = 0;
        for (String cursor : cursors) {
            if (!clusterSet.contains(cursor)) {
                removes++;
            }
        }
        return removes;
    }

    static void testPulsar0002() {
        int C = 500; // configured replication clusters (list size)
        int R = 500; // active replicator cursors (loop iterations)
        // Simulate geo-replicated topic: R cursors, C configured clusters
        List<String> replicationClusters = new ArrayList<>();
        List<String> cursorClusters = new ArrayList<>();
        for (int i = 0; i < C; i++) {
            replicationClusters.add("cluster-" + i);
        }
        // cursors: all are orphans — each triggers full linear scan before miss
        for (int i = 0; i < R; i++) {
            cursorClusters.add("orphan-cluster-" + i);   // not in list: O(C) scan
        }

        int REPS = 500;

        // Warm up
        for (int i = 0; i < 5; i++) {
            defectReplicationCursorScan(replicationClusters, cursorClusters);
            fixedReplicationCursorScan(replicationClusters, cursorClusters);
        }

        long t0 = System.nanoTime();
        for (int r = 0; r < REPS; r++) {
            defectReplicationCursorScan(replicationClusters, cursorClusters);
        }
        long defectNs = System.nanoTime() - t0;

        long t1 = System.nanoTime();
        for (int r = 0; r < REPS; r++) {
            fixedReplicationCursorScan(replicationClusters, cursorClusters);
        }
        long fixedNs = System.nanoTime() - t1;

        double ratio = (double) defectNs / fixedNs;
        System.out.printf("pulsar-0002 replicationCluster list scan   C=%d R=%d  defect=%.2fms  fixed=%.2fms  ratio=%.1fx%n",
                C, R,
                defectNs / 1e6 / REPS,
                fixedNs  / 1e6 / REPS,
                ratio);

        if (ratio < 3.0) {
            throw new RuntimeException("pulsar-0002 FAIL: expected ratio >= 3.0, got " + ratio);
        }
        System.out.println("pulsar-0002 PASS");
    }

    // -----------------------------------------------------------------------
    // pulsar-0003: NamespacesBase getTopicHashPositions O(T*B)
    //
    // Pattern:
    //   List<String> allTopicsInThisBundle = getOwnedTopicListForNamespaceBundle(nsBundle);
    //   for (String topic : topics) {
    //       if (allTopicsInThisBundle.contains(topicName.toString())) { ... }  // O(B)
    //   }
    // -----------------------------------------------------------------------

    static long defectTopicHashPositions(List<String> allTopicsInBundle, List<String> queryTopics) {
        long hits = 0;
        for (String topic : queryTopics) {
            // simulates: if (allTopicsInThisBundle.contains(topicName.toString()))
            if (allTopicsInBundle.contains(topic)) {
                hits++;
            }
        }
        return hits;
    }

    static long fixedTopicHashPositions(List<String> allTopicsInBundle, List<String> queryTopics) {
        Set<String> topicSet = new HashSet<>(allTopicsInBundle);
        long hits = 0;
        for (String topic : queryTopics) {
            if (topicSet.contains(topic)) {
                hits++;
            }
        }
        return hits;
    }

    static void testPulsar0003() {
        int B = 1000; // topics in bundle
        int T = 500;  // topics in query list
        List<String> allTopicsInBundle = new ArrayList<>();
        List<String> queryTopics = new ArrayList<>();
        for (int i = 0; i < B; i++) {
            allTopicsInBundle.add("persistent://tenant/ns/topic-" + i);
        }
        // Query includes topics that land near the end of the bundle list (worst-case scan)
        for (int i = 0; i < T; i++) {
            queryTopics.add("persistent://tenant/ns/topic-" + (B - T + i));
        }

        int REPS = 200;

        // Warm up
        for (int i = 0; i < 5; i++) {
            defectTopicHashPositions(allTopicsInBundle, queryTopics);
            fixedTopicHashPositions(allTopicsInBundle, queryTopics);
        }

        long t0 = System.nanoTime();
        for (int r = 0; r < REPS; r++) {
            defectTopicHashPositions(allTopicsInBundle, queryTopics);
        }
        long defectNs = System.nanoTime() - t0;

        long t1 = System.nanoTime();
        for (int r = 0; r < REPS; r++) {
            fixedTopicHashPositions(allTopicsInBundle, queryTopics);
        }
        long fixedNs = System.nanoTime() - t1;

        double ratio = (double) defectNs / fixedNs;
        System.out.printf("pulsar-0003 topicHashPositions list scan   B=%d T=%d  defect=%.1fms  fixed=%.1fms  ratio=%.1fx%n",
                B, T,
                defectNs / 1e6 / REPS,
                fixedNs  / 1e6 / REPS,
                ratio);

        if (ratio < 5.0) {
            throw new RuntimeException("pulsar-0003 FAIL: expected ratio >= 5.0, got " + ratio);
        }
        System.out.println("pulsar-0003 PASS");
    }

    public static void main(String[] args) {
        testPulsar0001();
        testPulsar0002();
        testPulsar0003();
        System.out.println("All Pulsar CWE-407 tests PASS");
    }
}