package meerkat.bulletinboard;

import com.google.common.util.concurrent.FutureCallback;
import com.google.protobuf.Timestamp;
import meerkat.comm.CommunicationException;
import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.protobuf.Crypto.Signature;
import meerkat.protobuf.Voting.*;

import java.util.List;

/**
 * Created by Arbel Deutsch Peled on 03-Mar-16.
 * This is a full-fledged implementation of a Bulletin Board Client
 * It provides asynchronous access to several remote servers, as well as a local cache
 * Read operations are performed on the local server
 * Batch reads are performed on the local server and, if they fail, also on the remote servers
 * Write operations are performed on the local server
 * A Synchronizer is employed in order to keep the remote server up to date
 * After any read is carried out, a subscription is made for the specific query to make sure the local DB will be updated
 * The database also employs a synchronizer which makes sure local data is sent to the remote servers
 */
public class CachedBulletinBoardClient implements SubscriptionBulletinBoardClient {

    private final AsyncBulletinBoardClient localClient;
    private final AsyncBulletinBoardClient remoteClient;
    private final AsyncBulletinBoardClient queueClient;
    private final BulletinBoardSubscriber subscriber;
    private final BulletinBoardSynchronizer synchronizer;

    private Thread syncThread;

    private final static int DEFAULT_WAIT_CAP = 3000;
    private final static int DEFAULT_SLEEP_INTERVAL = 3000;

    private class SubscriptionStoreCallback implements FutureCallback<List<BulletinBoardMessage>>  {

        private final FutureCallback<List<BulletinBoardMessage>> callback;

        public SubscriptionStoreCallback(){
            callback = null;
        }

        public SubscriptionStoreCallback(FutureCallback<List<BulletinBoardMessage>> callback){
            this.callback = callback;
        }

        @Override
        public void onSuccess(List<BulletinBoardMessage> result) {
            for (BulletinBoardMessage msg : result) {
                try {

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

                        // This is a batch message: need to upload batch data as well as the message itself
                        BulletinBoardMessage completeMessage = localClient.readBatchData(msg);

                        localClient.postMessage(completeMessage);

                    } else {

                        // This is a regular message: post it
                        localClient.postMessage(msg);

                    }

                } catch (CommunicationException ignored) {
                    // TODO: log
                }
            }
        }

        @Override
        public void onFailure(Throwable t) {
            if (callback != null) {
                callback.onFailure(t);  // This is some hard error that cannot be dealt with
            }
        }

    }

    /**
     * Creates a Cached Client
     * Assumes all parameters are initialized
     * @param localClient is a Client for the local instance
     * @param remoteClient is a Client for the remote instance(s); Should have endless retries for post operations
     * @param subscriber is a subscription service to the remote instance(s)
     * @param queueClient is a client for a local deletable server to be used as a queue for not-yet-uploaded messages
     */
    public CachedBulletinBoardClient(AsyncBulletinBoardClient localClient,
                                     AsyncBulletinBoardClient remoteClient,
                                     BulletinBoardSubscriber subscriber,
                                     DeletableSubscriptionBulletinBoardClient queueClient,
                                     int sleepInterval,
                                     int waitCap) {

        this.localClient = localClient;
        this.remoteClient = remoteClient;
        this.subscriber = subscriber;
        this.queueClient = queueClient;

        this.synchronizer = new SimpleBulletinBoardSynchronizer(sleepInterval,waitCap);
        synchronizer.init(queueClient, remoteClient);
        syncThread = new Thread(synchronizer);
        syncThread.start();
    }

    /**
     * Creates a Cached Client
     * Used default values foe the time caps
     * */
    public CachedBulletinBoardClient(AsyncBulletinBoardClient localClient,
                                     AsyncBulletinBoardClient remoteClient,
                                     BulletinBoardSubscriber subscriber,
                                     DeletableSubscriptionBulletinBoardClient queue) {

        this(localClient, remoteClient, subscriber, queue, DEFAULT_SLEEP_INTERVAL, DEFAULT_WAIT_CAP);
    }

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

        return localClient.postMessage(msg, new FutureCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean result) {
                remoteClient.postMessage(msg, callback);
            }

            @Override
            public void onFailure(Throwable t) {
                if (callback != null)
                    callback.onFailure(t);
            }
        });

    }

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

        return localClient.postAsBatch(msg, chunkSize, new FutureCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean result) {
                remoteClient.postAsBatch(msg, chunkSize, callback);
            }

            @Override
            public void onFailure(Throwable t) {
                if (callback != null)
                    callback.onFailure(t);
            }
        });

    }

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

        localClient.beginBatch(tags, new FutureCallback<BatchIdentifier>() {

            private BatchIdentifier localIdentifier;

            @Override
            public void onSuccess(BatchIdentifier result) {

                localIdentifier = result;

                remoteClient.beginBatch(tags, new FutureCallback<BatchIdentifier>() {
                    @Override
                    public void onSuccess(BatchIdentifier result) {
                        if (callback != null)
                            callback.onSuccess(new CachedClientBatchIdentifier(localIdentifier, result));
                    }

                    @Override
                    public void onFailure(Throwable t) {
                        if (callback != null)
                            callback.onFailure(t);
                    }
                });

            }

            @Override
            public void onFailure(Throwable t) {
                if (callback != null)
                    callback.onFailure(t);
            }
        });

    }

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

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

        final CachedClientBatchIdentifier identifier = (CachedClientBatchIdentifier) batchIdentifier;

        localClient.postBatchData(identifier.getLocalIdentifier(), batchChunkList, startPosition, new FutureCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean result) {
                remoteClient.postBatchData(identifier.getRemoteIdentifier(), batchChunkList, startPosition, callback);
            }

            @Override
            public void onFailure(Throwable t) {
                if (callback != null)
                    callback.onFailure(t);
            }
        });

    }

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

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

        final CachedClientBatchIdentifier identifier = (CachedClientBatchIdentifier) batchIdentifier;

        localClient.postBatchData(identifier.getLocalIdentifier(), batchChunkList, new FutureCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean result) {
                remoteClient.postBatchData(identifier.getRemoteIdentifier(), batchChunkList, callback);
            }

            @Override
            public void onFailure(Throwable t) {
                if (callback != null)
                    callback.onFailure(t);
            }
        });

    }

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

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

        final CachedClientBatchIdentifier identifier = (CachedClientBatchIdentifier) batchIdentifier;

        localClient.closeBatch(identifier.getLocalIdentifier(), timestamp, signatures, new FutureCallback<Boolean>() {
            @Override
            public void onSuccess(Boolean result) {

                remoteClient.closeBatch(identifier.getRemoteIdentifier(), timestamp, signatures, callback);

            }

            @Override
            public void onFailure(Throwable t) {
                if (callback != null)
                    callback.onFailure(t);
            }
        });

    }

    @Override
    public void getRedundancy(MessageID id, FutureCallback<Float> callback) {

        remoteClient.getRedundancy(id, callback);

    }

    @Override
    public void readMessages(MessageFilterList filterList, final FutureCallback<List<BulletinBoardMessage>> callback) {

        localClient.readMessages(filterList, callback);

        subscriber.subscribe(filterList, new SubscriptionStoreCallback(callback));

    }

    @Override
    public void readMessage(final MessageID msgID, final FutureCallback<BulletinBoardMessage> callback) {

        localClient.readMessage(msgID, new FutureCallback<BulletinBoardMessage>() {

            @Override
            public void onSuccess(BulletinBoardMessage result) {
                if (callback != null)
                    callback.onSuccess(result);     // Read from local client was successful
            }

            @Override
            public void onFailure(Throwable t) {

                // Read from local unsuccessful: try to read from remote

                remoteClient.readMessage(msgID, new FutureCallback<BulletinBoardMessage>() {

                    @Override
                    public void onSuccess(BulletinBoardMessage result) {

                        // Read from remote was successful: store in local and return result

                        localClient.postMessage(result, null);

                        if (callback != null)
                            callback.onSuccess(result);

                    }

                    @Override
                    public void onFailure(Throwable t) {

                        // Read from remote was unsuccessful: report error
                        if (callback != null)
                            callback.onFailure(t);

                    }

                });

            }

        });

    }

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

        localClient.readBatchData(stub, new FutureCallback<BulletinBoardMessage>() {

            @Override
            public void onSuccess(BulletinBoardMessage result) {
                if (callback != null)
                    callback.onSuccess(result);     // Read from local client was successful
            }

            @Override
            public void onFailure(Throwable t) {

                // Read from local unsuccessful: try to read from remote

                remoteClient.readBatchData(stub, new FutureCallback<BulletinBoardMessage>() {

                    @Override
                    public void onSuccess(BulletinBoardMessage result) {

                        // Read from remote was successful: store in local and return result

                        localClient.postMessage(result, null);

                        if (callback != null)
                            callback.onSuccess(result);

                    }

                    @Override
                    public void onFailure(Throwable t) {

                        // Read from remote was unsuccessful: report error
                        if (callback != null)
                            callback.onFailure(t);

                    }

                });

            }

        });

    }

    @Override
    public void querySync(SyncQuery syncQuery, FutureCallback<SyncQueryResponse> callback) {

        localClient.querySync(syncQuery, callback);

    }

    @Override
    /**
     * This is a stub method
     * All resources are assumed to be initialized
     */
    public void init(BulletinBoardClientParams clientParams) {}

    @Override
    public MessageID postMessage(BulletinBoardMessage msg) throws CommunicationException {
        return localClient.postMessage(msg);
    }

    @Override
    public MessageID postAsBatch(BulletinBoardMessage msg, int chunkSize) throws CommunicationException {
        MessageID result = localClient.postAsBatch(msg, chunkSize);
        remoteClient.postAsBatch(msg, chunkSize);
        return  result;
    }

    @Override
    public float getRedundancy(MessageID id) throws CommunicationException {
        return remoteClient.getRedundancy(id);
    }

    @Override
    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException {
        subscriber.subscribe(filterList, new SubscriptionStoreCallback());
        return localClient.readMessages(filterList);
    }

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

        BulletinBoardMessage result = null;
        try {
            result = localClient.readMessage(msgID);
        } catch (CommunicationException e) {
            //TODO: log
        }

        if (result == null){
            result = remoteClient.readMessage(msgID);

            if (result != null){
                localClient.postMessage(result);
            }

        }

        return result;

    }

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

        BulletinBoardMessage result = null;
        try {
            result = localClient.readBatchData(stub);
        } catch (CommunicationException e) {
            //TODO: log
        }

        if (result == null){
            result = remoteClient.readBatchData(stub);

            if (result != null){
                localClient.postMessage(result);
            }

        }

        return result;

    }

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

    @Override
    public void close() {
        localClient.close();
        remoteClient.close();
        synchronizer.stop();
        try {
            syncThread.join();
        } catch (InterruptedException e) {
            //TODO: log interruption
        }
    }

    @Override
    public void subscribe(MessageFilterList filterList, FutureCallback<List<BulletinBoardMessage>> callback) {
        subscriber.subscribe(filterList, new SubscriptionStoreCallback(callback));
    }

    @Override
    public void subscribe(MessageFilterList filterList, long startEntry, FutureCallback<List<BulletinBoardMessage>> callback) {
        subscriber.subscribe(filterList, startEntry, new SubscriptionStoreCallback(callback));
    }

}