package meerkat.bulletinboard;

import com.google.common.util.concurrent.FutureCallback;
import com.google.protobuf.*;
import com.google.protobuf.Timestamp;

import static meerkat.bulletinboard.BulletinBoardSynchronizer.SyncStatus;

import meerkat.bulletinboard.sqlserver.BulletinBoardSQLServer;
import meerkat.bulletinboard.sqlserver.H2QueryProvider;

import meerkat.comm.CommunicationException;
import meerkat.crypto.concrete.ECDSASignature;
import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.util.BulletinBoardMessageComparator;
import meerkat.util.BulletinBoardMessageGenerator;
import org.junit.*;

import java.io.IOException;
import java.io.InputStream;
import java.security.*;
import java.security.cert.CertificateException;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.fail;

/**
 * Created by Arbel on 6/1/2016.
 */
public class BulletinBoardSynchronizerTest {

    private static final String REMOTE_SERVER_ADDRESS = "remoteDB";
    private static final String LOCAL_SERVER_ADDRESS = "localDB";
    private static int testCount;

    private static final int THREAD_NUM = 3;
    private static final int SUBSCRIPTION_INTERVAL = 1000;

    private static final int SYNC_SLEEP_INTERVAL = 500;
    private static final int SYNC_WAIT_CAP = 1000;

    private DeletableSubscriptionBulletinBoardClient localClient;
    private AsyncBulletinBoardClient remoteClient;

    private BulletinBoardSynchronizer synchronizer;

    private static BulletinBoardMessageGenerator messageGenerator;
    private static BulletinBoardMessageComparator messageComparator;

    private static String KEYFILE_EXAMPLE = "/certs/enduser-certs/user1-key-with-password-secret.p12";
    private static String KEYFILE_PASSWORD1 = "secret";
    private static String CERT1_PEM_EXAMPLE = "/certs/enduser-certs/user1.crt";

    private static BulletinBoardSignature[] signers;
    private static ByteString[] signerIDs;

    private Semaphore semaphore;
    private List<Throwable> thrown;

    @BeforeClass
    public static void build() {

        messageGenerator = new BulletinBoardMessageGenerator(new Random(0));
        messageComparator = new BulletinBoardMessageComparator();

        signers = new BulletinBoardSignature[1];
        signerIDs = new ByteString[1];

        signers[0] = new GenericBulletinBoardSignature(new ECDSASignature());
        signerIDs[0] = signers[0].getSignerID();

        InputStream keyStream = BulletinBoardSynchronizerTest.class.getResourceAsStream(KEYFILE_EXAMPLE);
        char[] password = KEYFILE_PASSWORD1.toCharArray();

        try {

            KeyStore.Builder keyStoreBuilder = signers[0].getPKCS12KeyStoreBuilder(keyStream, password);

            signers[0].loadSigningCertificate(keyStoreBuilder);

            signers[0].loadVerificationCertificates(BulletinBoardSynchronizerTest.class.getResourceAsStream(CERT1_PEM_EXAMPLE));

        } catch (IOException e) {
            System.err.println("Failed reading from signature file " + e.getMessage());
            fail("Failed reading from signature file " + e.getMessage());
        } catch (CertificateException e) {
            System.err.println("Failed reading certificate " + e.getMessage());
            fail("Failed reading certificate " + e.getMessage());
        } catch (KeyStoreException e) {
            System.err.println("Failed reading keystore " + e.getMessage());
            fail("Failed reading keystore " + e.getMessage());
        } catch (NoSuchAlgorithmException e) {
            System.err.println("Couldn't find signing algorithm " + e.getMessage());
            fail("Couldn't find signing algorithm " + e.getMessage());
        } catch (UnrecoverableKeyException e) {
            System.err.println("Couldn't find signing key " + e.getMessage());
            fail("Couldn't find signing key " + e.getMessage());
        }

        signerIDs[0] = signers[0].getSignerID();

        testCount = 0;

    }

    @Before
    public void init() throws CommunicationException {

        DeletableBulletinBoardServer remoteServer = new BulletinBoardSQLServer(new H2QueryProvider(REMOTE_SERVER_ADDRESS + testCount));
        remoteServer.init();

        remoteClient = new LocalBulletinBoardClient(
                remoteServer,
                THREAD_NUM,
                SUBSCRIPTION_INTERVAL);

        DeletableBulletinBoardServer localServer = new BulletinBoardSQLServer(new H2QueryProvider(LOCAL_SERVER_ADDRESS + testCount));
        localServer.init();

        localClient = new LocalBulletinBoardClient(
                localServer,
                THREAD_NUM,
                SUBSCRIPTION_INTERVAL);

        synchronizer = new SimpleBulletinBoardSynchronizer(SYNC_SLEEP_INTERVAL, SYNC_WAIT_CAP);
        synchronizer.init(localClient, remoteClient);

        semaphore = new Semaphore(0);
        thrown = new LinkedList<>();

        testCount++;

    }

    private class SyncStatusCallback implements FutureCallback<SyncStatus> {

        private final SyncStatus statusToWaitFor;
        private AtomicBoolean stillWaiting;

        public SyncStatusCallback(SyncStatus statusToWaitFor) {
            this.statusToWaitFor = statusToWaitFor;
            stillWaiting = new AtomicBoolean(true);
        }

        @Override
        public void onSuccess(SyncStatus result) {

            if (result == statusToWaitFor && stillWaiting.compareAndSet(true, false)){
                semaphore.release();
            }

        }

        @Override
        public void onFailure(Throwable t) {
            thrown.add(t);
            if (stillWaiting.compareAndSet(true,false)) {
                semaphore.release();
            }
        }
    }

    private class MessageCountCallback implements FutureCallback<Integer> {

        private int[] expectedCounts;
        private int currentIteration;

        public MessageCountCallback(int[] expectedCounts) {
            this.expectedCounts = expectedCounts;
            this.currentIteration = 0;
        }

        @Override
        public void onSuccess(Integer result) {

            if (currentIteration < expectedCounts.length){
                if (result != expectedCounts[currentIteration]){
                    onFailure(new AssertionError("Wrong message count. Expected " + expectedCounts[currentIteration] + " but received " + result));
                    currentIteration = expectedCounts.length;
                    return;
                }
            }

            currentIteration++;

            if (currentIteration == expectedCounts.length)
                semaphore.release();
        }

        @Override
        public void onFailure(Throwable t) {
            thrown.add(t);
            semaphore.release();
        }
    }

    @Test
    public void testSync() throws SignatureException, CommunicationException, InterruptedException {

        Timestamp timestamp = Timestamp.newBuilder()
                .setSeconds(15252162)
                .setNanos(85914)
                .build();

        BulletinBoardMessage msg = messageGenerator.generateRandomMessage(signers, timestamp, 10, 10);

        MessageID msgID = localClient.postMessage(msg);

        timestamp = Timestamp.newBuilder()
                .setSeconds(51511653)
                .setNanos(3625)
                .build();

        BulletinBoardMessage batchMessage = messageGenerator.generateRandomMessage(signers,timestamp,  100, 10);

        MessageID batchMsgID = localClient.postAsBatch(batchMessage, 10);

        BulletinBoardMessage test = localClient.readMessage(batchMsgID);

        BulletinBoardMessage stub = localClient.readMessages(MessageFilterList.newBuilder()
                .addFilter(MessageFilter.newBuilder()
                        .setType(FilterType.MSG_ID)
                        .setId(batchMsgID.getID())
                        .build())
                .build()).get(0);

        BulletinBoardMessage test2 = localClient.readBatchData(stub);

        synchronizer.subscribeToSyncStatus(new SyncStatusCallback(SyncStatus.SYNCHRONIZED));

        int[] expectedCounts = {2,0};
        synchronizer.subscribeToRemainingMessagesCount(new MessageCountCallback(expectedCounts));

        Thread syncThread = new Thread(synchronizer);
        syncThread.start();

        if (!semaphore.tryAcquire(2, 4000, TimeUnit.MILLISECONDS)) {
            thrown.add(new TimeoutException("Timeout occurred while waiting for synchronizer to sync."));
        }

        synchronizer.stop();
        syncThread.join();

        if (thrown.size() > 0) {
            for (Throwable t : thrown)
                System.err.println(t.getMessage());
            assertThat("Exception thrown by Synchronizer: " + thrown.get(0).getMessage(), false);
        }

        List<BulletinBoardMessage> msgList = remoteClient.readMessages(MessageFilterList.newBuilder()
                .addFilter(MessageFilter.newBuilder()
                        .setType(FilterType.MSG_ID)
                        .setId(msgID.getID())
                        .build())
                .build());

        assertThat("Wrong number of messages returned.", msgList.size() == 1);
        assertThat("Returned message is not equal to original one", messageComparator.compare(msgList.get(0),msg) == 0);

        BulletinBoardMessage returnedBatchMsg = remoteClient.readMessage(batchMsgID);

        assertThat("Returned batch does not equal original one.", messageComparator.compare(returnedBatchMsg, batchMessage) == 0);

    }

    @Test
    public void testServerError() throws SignatureException, CommunicationException, InterruptedException {

        Timestamp timestamp = Timestamp.newBuilder()
                .setSeconds(945736256)
                .setNanos(276788)
                .build();

        BulletinBoardMessage msg = messageGenerator.generateRandomMessage(signers, timestamp, 10, 10);

        remoteClient.close();

        synchronizer.subscribeToSyncStatus(new SyncStatusCallback(SyncStatus.SERVER_ERROR));

        localClient.postMessage(msg);

        Thread thread = new Thread(synchronizer);

        thread.start();

        if (!semaphore.tryAcquire(4000, TimeUnit.MILLISECONDS)) {
            thrown.add(new TimeoutException("Timeout occurred while waiting for synchronizer to sync."));
        }

        synchronizer.stop();
        thread.join();

    }

    @After
    public void close() {

        if (thrown.size() > 0) {
            for (Throwable t : thrown) {
                System.err.println(t.getMessage());
            }
            assertThat("Exception thrown by Synchronizer: " + thrown.get(0).getMessage(), false);
        }

        synchronizer.stop();
        localClient.close();
        remoteClient.close();

    }

}
