package meerkat.bulletinboard;

import com.google.protobuf.BoolValue;
import com.google.protobuf.ByteString;
import com.google.protobuf.Int64Value;
import meerkat.bulletinboard.workers.singleserver.*;
import meerkat.comm.CommunicationException;
import meerkat.crypto.concrete.SHA256Digest;
import meerkat.protobuf.BulletinBoardApi;
import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.protobuf.Voting.*;
import meerkat.rest.*;
import meerkat.util.BulletinBoardUtils;

import java.util.List;

import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Response;

import static meerkat.bulletinboard.BulletinBoardConstants.*;

/**
 * Created by Arbel Deutsch Peled on 05-Dec-15.
 * Implements BulletinBoardClient interface in a simple, straightforward manner
 */
public class SimpleBulletinBoardClient implements BulletinBoardClient{

    protected List<String> meerkatDBs;

    protected Client client;

    protected BulletinBoardDigest digest;

    /**
     * Stores database locations and initializes the web Client
     * @param clientParams contains the data needed to access the DBs
     */
    @Override
    public void init(BulletinBoardClientParams clientParams) {

        this.meerkatDBs = clientParams.getBulletinBoardAddressList();

        client = ClientBuilder.newClient();
        client.register(ProtobufMessageBodyReader.class);
        client.register(ProtobufMessageBodyWriter.class);

        // Wrap the Digest into a BatchDigest
        digest = new GenericBulletinBoardDigest(new SHA256Digest());

    }

    /**
     * Post message to all DBs
     * Make only one try per DB.
     * @param msg is the message,
     * @return the message ID for later retrieval
     * @throws CommunicationException
     */
    @Override
    public MessageID postMessage(BulletinBoardMessage msg) throws CommunicationException {

        WebTarget webTarget;
        Response response = null;

        // Post message to all databases
        try {
            for (String db : meerkatDBs) {

                SingleServerPostMessageWorker worker = new SingleServerPostMessageWorker(db, msg, 0);

                worker.call();

            }
        } catch (Exception e) { // Occurs only when server replies with valid status but invalid data
            throw new CommunicationException("Error accessing database: " + e.getMessage());
        }

        // Calculate the correct message ID and return it
        digest.reset();
        digest.update(msg.getMsg());
        return MessageID.newBuilder().setID(ByteString.copyFrom(digest.digest())).build();
    }

    /**
     * Access each database and search for a given message ID
     * Return the number of databases in which the message was found
     * Only try once per DB
     * Ignore communication exceptions in specific databases
     * @param id is the requested message ID
     * @return the number of DBs in which retrieval was successful
     */
    @Override
    public float getRedundancy(MessageID id) {
        WebTarget webTarget;
        Response response;

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

        float count = 0;

        for (String db : meerkatDBs) {
            try {
                webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(READ_MESSAGES_PATH);

                response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(filterList, Constants.MEDIATYPE_PROTOBUF));

                if (response.readEntity(BulletinBoardMessageList.class).getMessageCount() > 0){
                    count++;
                }

            } catch (Exception e) {}
        }

        return count / ((float) meerkatDBs.size());
    }

    /**
     * Go through the DBs and try to retrieve messages according to the specified filter
     * If at the operation is successful for some DB: return the results and stop iterating
     * If no operation is successful: return null (NOT blank list)
     * @param filterList return only messages that match the filters (null means no filtering).
     * @return the list of Bulletin Board messages that are returned from a server
     */
    @Override
    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException{

        // Replace null filter list with blank one.
        if (filterList == null){
            filterList = MessageFilterList.getDefaultInstance();
        }

        String exceptionString = "";

        for (String db : meerkatDBs) {

            try {

                SingleServerReadMessagesWorker worker = new SingleServerReadMessagesWorker(db, filterList, 0);

                List<BulletinBoardMessage> result = worker.call();

                return result;

            } catch (Exception e) {
                //TODO: log
                exceptionString += e.getMessage() + "\n";
            }
        }

        throw new CommunicationException("Could not find message in any DB. Errors follow:\n" + exceptionString);

    }

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

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

        BeginBatchMessage beginBatchMessage = BulletinBoardUtils.generateBeginBatchMessage(msg);

        boolean posted = false;

        // Post message to all databases

        for (String db : meerkatDBs) {

            try {

                int pos = 0;

                SingleServerBeginBatchWorker beginBatchWorker = new SingleServerBeginBatchWorker(db, beginBatchMessage, 0);

                Int64Value batchId = beginBatchWorker.call();

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

                for (BatchChunk batchChunk : chunkList) {

                    SingleServerPostBatchWorker postBatchWorker =
                            new SingleServerPostBatchWorker(
                                    db,
                                    builder.setData(batchChunk).setSerialNum(pos).build(),
                                    0);

                    postBatchWorker.call();

                    pos++;

                }

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

                SingleServerCloseBatchWorker closeBatchWorker = new SingleServerCloseBatchWorker(db, closeBatchMessage, 0);

                closeBatchWorker.call();

                posted = true;

            } catch(Exception ignored) {}

        }

        if (!posted){
            throw new CommunicationException("Could not post to any server");
        }

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

    }

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

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

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

        String exceptionString = "";

        for (String db : meerkatDBs) {

            try {
                SingleServerReadMessagesWorker messagesWorker = new SingleServerReadMessagesWorker(db, filterList, 0);

                List<BulletinBoardMessage> messages = messagesWorker.call();

                if (messages == null || messages.size() < 1)
                    continue;

                BulletinBoardMessage stub = messages.get(0);

                SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(db, batchQuery, 0);

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

                return BulletinBoardUtils.gatherBatch(stub, batchChunkList);

            } catch (Exception e) {
                //TODO: log
                exceptionString += e.getMessage() + "\n";
            }
        }

        throw new CommunicationException("Could not find message in any DB. Errors follow:\n" + exceptionString);

    }

    @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");
        }

        BatchQuery batchQuery = BatchQuery.newBuilder()
                .setMsgID(MessageID.newBuilder()
                        .setID(stub.getMsg().getMsgId())
                        .build())
                .setStartPosition(0)
                .build();

        String exceptionString = "";

        for (String db : meerkatDBs) {

            try {

                SingleServerReadBatchWorker batchWorker = new SingleServerReadBatchWorker(db, batchQuery, 0);

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

                return BulletinBoardUtils.gatherBatch(stub, batchChunkList);

            } catch (Exception e) {
                //TODO: log
                exceptionString += e.getMessage() + "\n";
            }
        }

        throw new CommunicationException("Could not find message in any DB. Errors follow:\n" + exceptionString);

    }

    @Override
    public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException {

        WebTarget webTarget;
        Response response;

        for (String db : meerkatDBs) {

            try {
                webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(GENERATE_SYNC_QUERY_PATH);

                response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(generateSyncQueryParams, Constants.MEDIATYPE_PROTOBUF));

                return response.readEntity(SyncQuery.class);

            } catch (Exception e) {}

        }

        throw new CommunicationException("Could not contact any server");

    }

    public void close() {
        client.close();
    }

}
