package communication.overlays.clique;

import communication.CommLayer;
import communication.data.Data;
import proto.types.block.*;
import proto.types.comm.*;
import utils.config.yaml.ServerPublicDetails;

import java.util.*;
import java.util.stream.Collectors;

import static crypto.BlockDigSig.verfiyBlockWRTheader;
import static java.lang.String.format;

public class Clique implements CommLayer {
    private final static org.apache.log4j.Logger logger = org.apache.log4j.Logger.getLogger(Clique.class);



    private int id;
    private int n;
    private CliqueRpcs rpcs;

    public Clique(int id, int workers, int n,  ServerPublicDetails[] cluster) {
        this.id = id;
        this.n = n;
        new Data(n, workers);
        rpcs = new CliqueRpcs(id, cluster, n, workers);
    }

    @Override
    public void join() {
        rpcs.start();
    }

    @Override
    public void leave() {
       rpcs.shutdown();
    }

    @Override
    public void reconfigure() {
        rpcs.reconfigure();
    }

    @Override
    public void broadcast(int worker, Block data) {
        logger.debug(format("[%d-%d] broadcast block data [bid=%d]", id, worker, data.getId().getBid()));
        rpcs.broadcast(worker, data);
    }

    // Meant to test Byzantine activity

    @Override
    public void send(int worker, Block data, int[] recipients) {
        rpcs.send(worker, data, recipients);
    }

    private Block getBlockFromData(int channel, BlockID bid, BlockHeader proof) {
        final Block[] res = {null};
        int pid = bid.getPid();
        Data.blocks[pid][channel].computeIfPresent(bid, (k, v) -> {
            v = v.stream()
                    .filter(b -> verfiyBlockWRTheader(b, proof)).collect(Collectors.toCollection(LinkedList::new));
            res[0] = v.poll();
            return v;
        });
        return res[0];
    }

    @Override
    public Block recBlock(int channel, BlockHeader proof) throws InterruptedException {
        BlockID bid = proof.getBid();
        Block res = getBlockFromData(channel, bid, proof);
        if (res != null) return res;
        rpcs.broadcastCommReq(CommReq.newBuilder().setProof(proof).build());
        int pid = bid.getPid();
        synchronized (Data.blocks[pid][channel]) {
            while (res == null) {
                Data.blocks[pid][channel].wait();
                res = getBlockFromData(channel, bid, proof);
            }
        }
        return res;
    }

    @Override
    public boolean contains(int channel, BlockHeader proof) {
        if (proof.getEmpty()) return true;
        BlockID bid = proof.getBid();
        int pid = bid.getPid();
        Data.blocks[pid][channel].computeIfPresent(bid, (k, v) -> {
            int bef = v.size();
            v = v.stream().filter(b -> verfiyBlockWRTheader(b, proof))
                    .collect(Collectors.toCollection(LinkedList::new));
            logger.debug(format("[%d-%d] invalidate %d records [pid=%d ; bid=%d]", id, channel, bef - v.size(),
                    bid.getPid(), bid.getBid()));
            return v;
        });
        return Data.blocks[pid][channel].containsKey(bid) && Data.blocks[pid][channel].get(bid).size() > 0;
    }

}
