package meerkat.bulletinboard;

import com.google.common.util.concurrent.*;
import com.google.protobuf.Int64Value;
import com.google.protobuf.Timestamp;
import meerkat.comm.CommunicationException;
import meerkat.comm.MessageInputStream;
import meerkat.comm.MessageInputStream.MessageInputStreamFactory;
import meerkat.comm.MessageOutputStream;
import meerkat.crypto.concrete.SHA256Digest;
import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.protobuf.Crypto.Signature;
import meerkat.protobuf.Voting.*;
import meerkat.util.BulletinBoardUtils;

import javax.ws.rs.NotFoundException;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by Arbel Deutsch Peled on 15-Mar-16.
 * This client wraps a BulletinBoardServer in an asynchronous client.
 * It is meant to be used as a local cache handler and for testing purposes.
 * This means the access to the server is direct (via method calls) instead of through a TCP connection.
 * The client implements both synchronous and asynchronous method calls, but calls to the server itself are performed synchronously.
 */
public class LocalBulletinBoardClient implements DeletableSubscriptionBulletinBoardClient {

    private final DeletableBulletinBoardServer server;
    private final ListeningScheduledExecutorService executorService;
    private final BulletinBoardDigest digest;
    private final long subsrciptionDelay;

    /**
     * Initializes an instance of the client
     * @param server an initialized Bulletin Board Server instance which will perform the actual processing of the requests
     * @param threadNum is the number of concurrent threads to allocate for the client
     */
    public LocalBulletinBoardClient(DeletableBulletinBoardServer server, int threadNum, int subscriptionDelay) {
        this.server = server;
        this.executorService = MoreExecutors.listeningDecorator(Executors.newScheduledThreadPool(threadNum));
        this.digest = new GenericBulletinBoardDigest(new SHA256Digest());
        this.subsrciptionDelay = subscriptionDelay;
    }

    private class MessagePoster implements Callable<Boolean> {

        private final BulletinBoardMessage msg;

        public MessagePoster(BulletinBoardMessage msg) {
            this.msg = msg;
        }


        @Override
        public Boolean call() throws CommunicationException {
            return server.postMessage(msg).getValue();
        }

    }

    @Override
    public MessageID postMessage(BulletinBoardMessage msg, FutureCallback<Boolean> callback) {

        Futures.addCallback(executorService.submit(new MessagePoster(msg)), callback);

        digest.update(msg.getMsg());
        return digest.digestAsMessageID();

    }

    private class CompleteBatchPoster implements Callable<Boolean> {

        private final BulletinBoardMessage msg;
        private final int chunkSize;

        public CompleteBatchPoster(BulletinBoardMessage msg, int chunkSize) {
            this.msg = msg;
            this.chunkSize = chunkSize;
        }


        @Override
        public Boolean call() throws CommunicationException {

            BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder()
                    .addAllTag(msg.getMsg().getTagList())
                    .build();

            Int64Value batchId = server.beginBatch(beginBatchMessage);

            BatchMessage.Builder builder = BatchMessage.newBuilder()
                    .setBatchId(batchId.getValue());

            List<BatchChunk> batchChunkList = BulletinBoardUtils.breakToBatch(msg, chunkSize);

            int i=0;
            for (BatchChunk chunk : batchChunkList){

                server.postBatchMessage(builder.setSerialNum(i).setData(chunk).build());

                i++;

            }

            CloseBatchMessage closeBatchMessage = BulletinBoardUtils.generateCloseBatchMessage(batchId, batchChunkList.size(), msg);

            return server.closeBatch(closeBatchMessage).getValue();
        }

    }

    @Override
    public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize, FutureCallback<Boolean> callback) {

        Futures.addCallback(executorService.submit(new CompleteBatchPoster(msg, chunkSize)), callback);

        digest.reset();
        digest.update(msg);
        return digest.digestAsMessageID();

    }

    private class BatchBeginner implements Callable<SingleServerBatchIdentifier> {

        private final BeginBatchMessage msg;

        public BatchBeginner(BeginBatchMessage msg) {
            this.msg = msg;
        }


        @Override
        public SingleServerBatchIdentifier call() throws Exception {
            return new SingleServerBatchIdentifier(server.beginBatch(msg));
        }

    }

    @Override
    public void beginBatch(Iterable<String> tags, FutureCallback<BatchIdentifier> callback) {

        BeginBatchMessage beginBatchMessage = BeginBatchMessage.newBuilder()
                .addAllTag(tags)
                .build();

        Futures.addCallback(executorService.submit(new BatchBeginner(beginBatchMessage)), callback);
    }

    private class BatchDataPoster implements Callable<Boolean> {

        private final SingleServerBatchIdentifier batchId;
        private final List<BatchChunk> batchChunkList;
        private final int startPosition;

        public BatchDataPoster(SingleServerBatchIdentifier batchId, List<BatchChunk> batchChunkList, int startPosition) {
            this.batchId = batchId;
            this.batchChunkList = batchChunkList;
            this.startPosition = startPosition;
        }


        @Override
        public Boolean call() throws Exception {

            BatchMessage.Builder msgBuilder = BatchMessage.newBuilder()
                    .setBatchId(batchId.getBatchId().getValue());

            int i = startPosition;
            for (BatchChunk data : batchChunkList){

                msgBuilder.setSerialNum(i)
                        .setData(data);

                if (!server.postBatchMessage(msgBuilder.build()).getValue())
                    return false;

                i++;

            }

            batchId.setLength(i);

            return true;

        }

    }

    @Override
    public void postBatchData(BatchIdentifier batchId, List<BatchChunk> batchChunkList, int startPosition, FutureCallback<Boolean> callback)
        throws IllegalArgumentException{

        // Cast identifier to usable form

        if (!(batchId instanceof SingleServerBatchIdentifier)){
            throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
        }

        SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchId;

        // Add worker

        Futures.addCallback(executorService.submit(new BatchDataPoster(identifier, batchChunkList, startPosition)), callback);

    }

    @Override
    public void postBatchData(BatchIdentifier batchId, List<BatchChunk> batchChunkList, FutureCallback<Boolean> callback) throws IllegalArgumentException{
        postBatchData(batchId, batchChunkList, 0, callback);
    }


    private class BatchCloser implements Callable<Boolean> {

        private final CloseBatchMessage msg;

        public BatchCloser(CloseBatchMessage msg) {
            this.msg = msg;
        }


        @Override
        public Boolean call() throws Exception {
            return server.closeBatch(msg).getValue();
        }

    }

    @Override
    public void closeBatch(BatchIdentifier batchId, Timestamp timestamp, Iterable<Signature> signatures, FutureCallback<Boolean> callback) {

        // Cast identifier to usable form

        if (!(batchId instanceof SingleServerBatchIdentifier)){
            throw new IllegalArgumentException("Error: batch identifier supplied was not created by this class.");
        }

        SingleServerBatchIdentifier identifier = (SingleServerBatchIdentifier) batchId;

        // Add worker

        CloseBatchMessage closeBatchMessage = CloseBatchMessage.newBuilder()
                .setBatchId(identifier.getBatchId().getValue())
                .setBatchLength(identifier.getLength())
                .setTimestamp(timestamp)
                .addAllSig(signatures)
                .build();

        Futures.addCallback(executorService.submit(new BatchCloser(closeBatchMessage)), callback);

    }

    private class RedundancyGetter implements Callable<Float> {

        private final MessageID msgId;

        public RedundancyGetter(MessageID msgId) {
            this.msgId = msgId;
        }


        @Override
        public Float call() throws Exception {

            MessageFilterList filterList = MessageFilterList.newBuilder()
                    .addFilter(MessageFilter.newBuilder()
                            .setType(FilterType.MSG_ID)
                            .setId(msgId.getID())
                            .build())
                    .build();

            ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
            MessageOutputStream<BulletinBoardMessage> outputStream = new MessageOutputStream<>(byteOutputStream);
            server.readMessages(filterList,outputStream);

            MessageInputStream<BulletinBoardMessage> inputStream =
                    MessageInputStreamFactory.createMessageInputStream(
                            new ByteArrayInputStream(byteOutputStream.toByteArray()),
                            BulletinBoardMessage.class);

            if (inputStream.isAvailable())
                return 1.0f;
            else
                return 0.0f;

        }

    }

    @Override
    public void getRedundancy(MessageID id, FutureCallback<Float> callback) {
        Futures.addCallback(executorService.submit(new RedundancyGetter(id)), callback);
    }

    private class MessageReader implements Callable<List<BulletinBoardMessage>> {

        private final MessageFilterList filterList;

        public MessageReader(MessageFilterList filterList) {
            this.filterList = filterList;
        }


        @Override
        public List<BulletinBoardMessage> call() throws Exception {

            ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
            MessageOutputStream<BulletinBoardMessage> outputStream = new MessageOutputStream<>(byteOutputStream);
            server.readMessages(filterList, outputStream);

            MessageInputStream<BulletinBoardMessage> inputStream =
                    MessageInputStreamFactory.createMessageInputStream(
                            new ByteArrayInputStream(byteOutputStream.toByteArray()),
                            BulletinBoardMessage.class);

            return inputStream.asList();

        }

    }

    @Override
    public void readMessages(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {
        Futures.addCallback(executorService.submit(new MessageReader(filterList)), callback);
    }

    class SubscriptionCallback implements FutureCallback<List<BulletinBoardMessage>> {

        private MessageFilterList filterList;
        private final FutureCallback<List<BulletinBoardMessage>> callback;

        public SubscriptionCallback(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {
            this.filterList = filterList;
            this.callback = callback;
        }

        @Override
        public void onSuccess(List<BulletinBoardMessage> result) {

            // Report new messages to user
            if (callback != null)
                callback.onSuccess(result);

            MessageFilterList.Builder filterBuilder = filterList.toBuilder();

            // If any new messages arrived: update the MIN_ENTRY condition
            if (result.size() > 0) {

                // Remove last filter from list (MIN_ENTRY one)
                filterBuilder.removeFilter(filterBuilder.getFilterCount() - 1);

                // Add updated MIN_ENTRY filter (entry number is successor of last received entry's number)
                filterBuilder.addFilter(MessageFilter.newBuilder()
                        .setType(FilterType.MIN_ENTRY)
                        .setEntry(result.get(result.size() - 1).getEntryNum() + 1)
                        .build());

            }

            filterList = filterBuilder.build();

            // Reschedule job
            Futures.addCallback(executorService.schedule(new MessageReader(filterList), subsrciptionDelay, TimeUnit.MILLISECONDS), this);

        }

        @Override
        public void onFailure(Throwable t) {

            // Notify caller about failure and terminate subscription
            if (callback != null)
                callback.onFailure(t);

        }
    }

    @Override
    public void subscribe(MessageFilterList filterList, long startEntry, FutureCallback<List<BulletinBoardMessage>> callback) {

        MessageFilterList subscriptionFilterList =
                filterList.toBuilder()
                    .addFilter(MessageFilter.newBuilder()
                            .setType(FilterType.MIN_ENTRY)
                            .setEntry(startEntry)
                            .build())
                    .build();

        Futures.addCallback(executorService.submit(new MessageReader(subscriptionFilterList)), new SubscriptionCallback(subscriptionFilterList, callback));

    }

    @Override
    public void subscribe(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {
        subscribe(filterList, 0, callback);
    }

    private class BatchDataReader implements Callable<List<BatchChunk>> {

        private final MessageID msgID;

        public BatchDataReader(MessageID msgID) {
            this.msgID = msgID;
        }

        @Override
        public List<BatchChunk> call() throws Exception {

            BatchQuery batchQuery = BatchQuery.newBuilder()
                    .setMsgID(msgID)
                    .setStartPosition(0)
                    .build();

            ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream();
            MessageOutputStream<BatchChunk> batchOutputStream = new MessageOutputStream<>(byteOutputStream);
            server.readBatch(batchQuery,batchOutputStream);

            MessageInputStream<BatchChunk> inputStream =
                    MessageInputStreamFactory.createMessageInputStream(
                            new ByteArrayInputStream(byteOutputStream.toByteArray()),
                            BatchChunk.class);

            return inputStream.asList();

        }
    }

    private class CompleteBatchReader implements Callable<BulletinBoardMessage> {

        private final MessageID msgID;

        public CompleteBatchReader(MessageID msgID) {
            this.msgID = msgID;
        }


        @Override
        public BulletinBoardMessage call() throws Exception {

            // Read message (mat be a stub)

            MessageFilterList filterList = MessageFilterList.newBuilder()
                    .addFilter(MessageFilter.newBuilder()
                            .setType(FilterType.MSG_ID)
                            .setId(msgID.getID())
                            .build())
                    .build();

            MessageReader messageReader = new MessageReader(filterList);
            List<BulletinBoardMessage> bulletinBoardMessages = messageReader.call();

            if (bulletinBoardMessages.size() <= 0) {
                throw new NotFoundException("Message does not exist");
            }

            BulletinBoardMessage msg = bulletinBoardMessages.get(0);

            if (msg.getMsg().getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.MSGID) {

                // Read data

                BatchDataReader batchDataReader = new BatchDataReader(msgID);
                List<BatchChunk> batchChunkList = batchDataReader.call();

                // Combine and return

                return BulletinBoardUtils.gatherBatch(msg, batchChunkList);

            } else {
                return msg;
            }

        }

    }

    private class BatchDataCombiner implements Callable<BulletinBoardMessage> {

        private final BulletinBoardMessage stub;

        public BatchDataCombiner(BulletinBoardMessage stub) {
            this.stub = stub;
        }

        @Override
        public BulletinBoardMessage call() throws Exception {

            MessageID msgID = MessageID.newBuilder().setID(stub.getMsg().getMsgId()).build();

            BatchDataReader batchDataReader = new BatchDataReader(msgID);

            List<BatchChunk> batchChunkList = batchDataReader.call();

            return BulletinBoardUtils.gatherBatch(stub, batchChunkList);

        }

    }

    @Override
    public void readMessage(MessageID msgID, FutureCallback<BulletinBoardMessage> callback) {
        Futures.addCallback(executorService.submit(new CompleteBatchReader(msgID)), callback);
    }

    @Override
    public void readBatchData(BulletinBoardMessage stub, FutureCallback<BulletinBoardMessage> callback) throws IllegalArgumentException {

        if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID){
            throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
        }

        Futures.addCallback(executorService.submit(new BatchDataCombiner(stub)),callback);

    }

    private class SyncQueryHandler implements Callable<SyncQueryResponse> {

        private final SyncQuery syncQuery;

        public SyncQueryHandler(SyncQuery syncQuery) {
            this.syncQuery = syncQuery;
        }


        @Override
        public SyncQueryResponse call() throws Exception {
            return server.querySync(syncQuery);
        }

    }

    @Override
    public void querySync(SyncQuery syncQuery, FutureCallback<SyncQueryResponse> callback) {
        Futures.addCallback(executorService.submit(new SyncQueryHandler(syncQuery)), callback);
    }

    /**
     * This method is a stub, since the implementation only considers one server, and that is given in the constructor
     * @param ignored is ignored
     */
    @Override
    public void init(BulletinBoardClientParams ignored) {}

    @Override
    public MessageID postMessage(BulletinBoardMessage msg) throws CommunicationException {

            MessagePoster poster = new MessagePoster(msg);
            poster.call();

            digest.update(msg.getMsg());
            return digest.digestAsMessageID();

    }

    @Override
    public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize) throws CommunicationException {

        CompleteBatchPoster poster = new CompleteBatchPoster(msg, chunkSize);
        Boolean result = poster.call();

        if (!result)
            throw new CommunicationException("Batch post failed");

        digest.reset();
        digest.update(msg);
        return digest.digestAsMessageID();

    }

    @Override
    public float getRedundancy(MessageID id) {

        try {

            RedundancyGetter getter = new RedundancyGetter(id);
            return getter.call();

        } catch (Exception e) {
            return -1.0f;
        }

    }

    @Override
    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException{

        try {

            MessageReader reader = new MessageReader(filterList);
            return reader.call();

        } catch (Exception e){
            throw new CommunicationException("Error reading from server");
        }

    }

    @Override
    public BulletinBoardMessage readMessage(MessageID msgID) throws CommunicationException {

        MessageFilterList filterList = MessageFilterList.newBuilder()
                .addFilter(MessageFilter.newBuilder()
                        .setType(FilterType.MSG_ID)
                        .setId(msgID.getID())
                        .build())
                .build();

        CompleteBatchReader completeBatchReader = new CompleteBatchReader(msgID);

        try {
            return completeBatchReader.call();
        } catch (Exception e) {
            throw new CommunicationException(e.getMessage() + " " + e.getMessage());
        }

    }

    @Override
    public BulletinBoardMessage readBatchData(BulletinBoardMessage stub) throws CommunicationException, IllegalArgumentException {

        if (stub.getMsg().getDataTypeCase() != UnsignedBulletinBoardMessage.DataTypeCase.MSGID){
            throw new IllegalArgumentException("Message is not a stub and does not contain the required message ID");
        }

        BatchDataCombiner combiner = new BatchDataCombiner(stub);

        try {
            return combiner.call();
        } catch (Exception e) {
            throw new CommunicationException(e.getCause() + " " + e.getMessage());
        }

    }

    @Override
    public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException {
        return server.generateSyncQuery(generateSyncQueryParams);
    }

    @Override
    public void deleteMessage(MessageID msgID, FutureCallback<Boolean> callback) {

        try {
            Boolean deleted = server.deleteMessage(msgID).getValue();
            if (callback != null)
                callback.onSuccess(deleted);
        } catch (CommunicationException e) {
            if (callback != null)
                callback.onFailure(e);
        }

    }

    @Override
    public void deleteMessage(long entryNum, FutureCallback<Boolean> callback) {

        try {
            Boolean deleted = server.deleteMessage(entryNum).getValue();
            if (callback != null)
                callback.onSuccess(deleted);
        } catch (CommunicationException e) {
            if (callback != null)
                callback.onFailure(e);
        }

    }

    @Override
    public boolean deleteMessage(MessageID msgID) throws CommunicationException {
        return server.deleteMessage(msgID).getValue();
    }

    @Override
    public boolean deleteMessage(long entryNum) throws CommunicationException {
        return server.deleteMessage(entryNum).getValue();
    }

    @Override
    public void close() {
        try {
            server.close();
        } catch (CommunicationException ignored) {}
    }

}
