import dev.modzuozhi.core.compat.PayloadWorkQueue;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class PayloadWorkQueueRegression {
    static void check(boolean ok, String why) { if (!ok) throw new AssertionError(why); }
    public static void main(String[] args) throws Exception {
        fifoAndExactlyOnce(); capacityAndCancellation(); batchCountAndTime();
        closeAndReentrantCompletion(); concurrentProducers();
        System.out.println("PAYLOAD QUEUE PASS");
    }
    static void fifoAndExactlyOnce() {
        var q = new PayloadWorkQueue<String>(() -> 0L);
        var seen = new ArrayList<Integer>();
        var futures = new ArrayList<CompletableFuture<Integer>>();
        for (int i=0;i<256;i++) { int n=i; futures.add(q.submit("a", () -> { seen.add(n); return n; })); }
        while(q.size()>0) q.drainBatch();
        check(seen.size()==256, "exactly once");
        for(int i=0;i<256;i++) check(seen.get(i)==i && futures.get(i).join()==i, "FIFO/results");
        check(q.drainBatch()==0, "no replay");
    }
    static void capacityAndCancellation() {
        var q = new PayloadWorkQueue<Integer>(() -> 0L);
        var fs = new ArrayList<CompletableFuture<Integer>>();
        for(int c=0;c<16;c++) for(int i=0;i<256;i++) fs.add(q.submit(c, () -> 1));
        check(q.size()==4096, "server capacity");
        check(q.submit(99, () -> 2).isCompletedExceptionally(), "reject server overflow");
        fs.getFirst().cancel(false);
        check(q.size()==4095, "cancel returns capacity immediately");
        check(!q.submit(0, () -> 3).isDone(), "cancel frees connection slot");
        while(q.size()>0) q.drainBatch();
        for(int i=0;i<256;i++) q.submit(7, () -> 1);
        check(q.submit(7, () -> 9).isCompletedExceptionally(), "reject connection overflow");
        check(!q.submit(8, () -> 1).isDone(), "other connection still allowed");
        q.close(new IllegalStateException("stop"));
    }
    static void batchCountAndTime() {
        var clock = new AtomicLong(); var q = new PayloadWorkQueue<String>(clock::get);
        for(int i=0;i<65;i++) q.submit("a", () -> 1);
        check(q.drainBatch()==64 && q.size()==1, "64 budget"); q.drainBatch();
        var done = new AtomicInteger();
        q.submit("a", () -> { clock.addAndGet(2_000_000); return done.incrementAndGet(); });
        q.submit("a", done::incrementAndGet);
        check(q.drainBatch()==1 && done.get()==1 && q.size()==1, "2ms boundary");
        q.drainBatch();
        var slow = q.submit("a", () -> { clock.addAndGet(100_000_000); return 42; });
        check(q.drainBatch()==1 && slow.join()==42, "slow task not interrupted");
    }
    static void closeAndReentrantCompletion() {
        var q = new PayloadWorkQueue<String>(() -> 0L);
        var count = new AtomicInteger(); var f=q.submit("a", count::incrementAndGet);
        f.thenRun(() -> q.submit("a", count::incrementAndGet)); q.drainBatch();
        check(count.get()==2, "completion may enqueue");
        var boom = new IllegalArgumentException("handler failure");
        var bad=q.submit("a", () -> { throw boom; }); var good=q.submit("a", () -> 42);
        q.drainBatch(); check(bad.isCompletedExceptionally() && good.join()==42, "isolate exceptions");
        var pending=q.submit("a", count::incrementAndGet);
        q.close(boom); check(pending.isCompletedExceptionally() && q.size()==0, "close settles pending");
        check(q.submit("a", count::incrementAndGet).isCompletedExceptionally(), "close terminal");
        q.drainBatch(); check(count.get()==2, "closed work never executes");
        var q2=new PayloadWorkQueue<String>(() -> 0L);
        var cancel=q2.submit("a", count::incrementAndGet); cancel.cancel(false); q2.drainBatch();
        check(count.get()==2, "cancelled action skipped");
    }
    static void concurrentProducers() throws Exception {
        var q = new PayloadWorkQueue<Integer>(() -> 0L);
        var executed=new AtomicInteger(); var futures=new ConcurrentLinkedQueue<CompletableFuture<Integer>>();
        var pool=Executors.newFixedThreadPool(8);
        try {
            var jobs=new ArrayList<Future<?>>();
            for(int c=0;c<8;c++) { int key=c; jobs.add(pool.submit(() -> {
                for(int i=0;i<250;i++) { var f=q.submit(key, executed::incrementAndGet); futures.add(f); if(i%2==0) f.cancel(false); }
            })); }
            for(var job:jobs) job.get(5,TimeUnit.SECONDS);
            while(q.size()>0) q.drainBatch();
            check(executed.get()==1000 && q.size()==0, "concurrent cancel/submit accounting");
            check(futures.stream().allMatch(CompletableFuture::isDone), "no pending futures");
        } finally { pool.shutdownNow(); }
    }
}