/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.usergrid.services.notifications.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.usergrid.batch.JobExecution;
import org.apache.usergrid.persistence.Entity;
import org.apache.usergrid.persistence.EntityManager;
import org.apache.usergrid.persistence.EntityRef;
import org.apache.usergrid.persistence.PathQuery;
import org.apache.usergrid.persistence.Query;
import org.apache.usergrid.persistence.SimpleEntityRef;
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
import org.apache.usergrid.persistence.entities.Device;
import org.apache.usergrid.persistence.entities.Notification;
import org.apache.usergrid.persistence.entities.Notifier;
import org.apache.usergrid.persistence.entities.Receipt;
import org.apache.usergrid.persistence.entities.User;
import org.apache.usergrid.persistence.index.utils.UUIDUtils;
import org.apache.usergrid.persistence.queue.LegacyQueueManager;
import org.apache.usergrid.persistence.queue.LegacyQueueMessage;
import org.apache.usergrid.services.notifications.ApplicationQueueManager;
import org.apache.usergrid.services.notifications.ApplicationQueueMessage;
import org.apache.usergrid.services.notifications.JobScheduler;
import org.apache.usergrid.services.notifications.NotificationsService;
import org.apache.usergrid.services.notifications.ProviderAdapter;
import org.apache.usergrid.services.notifications.ProviderAdapterFactory;
import org.apache.usergrid.services.notifications.TaskManager;
import org.apache.usergrid.services.notifications.TaskTracker;

import com.codahale.metrics.Meter;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;
import rx.schedulers.Schedulers;


public class ApplicationQueueManagerImpl implements ApplicationQueueManager {

    private static final Logger logger = LoggerFactory.getLogger(ApplicationQueueManagerImpl.class);

    private final EntityManager em;
    private final LegacyQueueManager qm;
    private final JobScheduler jobScheduler;
    private final MetricsFactory metricsFactory;
    private final String queueName;
    private final Meter queueMeter;
    private final Meter sendMeter;
    private int concurrencyFactor;

    private final static String PUSH_PROCESSING_MAXTHREADS_PROP = "usergrid.push.async.processing.threads";
    private final static String PUSH_PROCESSING_QUEUESIZE_PROP = "usergrid.push.async.processing.queue.size";
    private final static String PUSH_PROCESSING_CONCURRENCY_PROP = "usergrid.push.async.processing.concurrency";

    HashMap<Object, ProviderAdapter> notifierHashMap; // only retrieve notifiers once


    //private final Scheduler scheduler;


    public ApplicationQueueManagerImpl(JobScheduler jobScheduler, EntityManager entityManager,
                                       LegacyQueueManager legacyQueueManager, MetricsFactory metricsFactory,
                                       Properties properties) {
        this.em = entityManager;
        this.qm = legacyQueueManager;
        this.jobScheduler = jobScheduler;
        this.metricsFactory = metricsFactory;
        this.queueName = getQueueNames(properties);
        this.queueMeter = metricsFactory.getMeter(ApplicationQueueManagerImpl.class, "notification.queue");
        this.sendMeter = metricsFactory.getMeter(NotificationsService.class, "queue.send");
        this.concurrencyFactor = Integer.valueOf(System.getProperty(PUSH_PROCESSING_CONCURRENCY_PROP, "50"));


        /**
        int maxAsyncThreads;
        int workerQueueSize;

        try {

            maxAsyncThreads = Integer.valueOf(System.getProperty(PUSH_PROCESSING_MAXTHREADS_PROP, "200"));
            workerQueueSize = Integer.valueOf(System.getProperty(PUSH_PROCESSING_QUEUESIZE_PROP, "2000"));

        } catch (Exception e){

            // if junk is passed into the property, just default the values
            maxAsyncThreads = 200;
            workerQueueSize = 2000;
            this.concurrencyFactor = 50;

        }


        // create our own executor which has a bounded queue w/ caller runs policy for rejected tasks
        this.scheduler = Schedulers.from(TaskExecutorFactory
            .createTaskExecutor( "push-device-io", maxAsyncThreads, workerQueueSize,
                TaskExecutorFactory.RejectionAction.CALLERRUNS ));
         **/

    }

    private boolean scheduleQueueJob(Notification notification) throws Exception {
        return jobScheduler.scheduleQueueJob(notification);
    }

    @Override
    public void queueNotification(final Notification notification, final JobExecution jobExecution) throws Exception {
        if (scheduleQueueJob(notification)) {
            em.update(notification);
            return;
        }

        long startTime = System.currentTimeMillis();

        if (notification.getCanceled() == Boolean.TRUE) {
            if (logger.isDebugEnabled()) {
                logger.debug("notification " + notification.getUuid() + " canceled");
            }
            if (jobExecution != null) {
                jobExecution.killed();
            }
            return;
        }

        final PathQuery<Device> pathQuery = notification.getPathQuery().buildPathQuery(); //devices query
        final AtomicInteger deviceCount = new AtomicInteger(); //count devices so you can make a judgement on batching
        final ConcurrentLinkedQueue<String> errorMessages = new ConcurrentLinkedQueue<>(); //build up list of issues

        // Get devices in querystring, and make sure you have access
        if (pathQuery != null) {
            final HashMap<Object, ProviderAdapter> notifierMap = getAdapterMap();
            if (logger.isTraceEnabled()) {
                logger.trace("notification {} start query", notification.getUuid());
            }

            if(logger.isTraceEnabled()) {
                logger.trace("Notification {} started processing", notification.getUuid());
            }


            // The main iterator can use graph traversal or index querying based on payload property. Default is Index.
            final Iterator<Device> iterator;
            if( notification.getUseGraph()){
                iterator = pathQuery.graphIterator(em);
            }else{
                iterator = pathQuery.iterator(em);
            }

            /**** Old code to scheduler large sets of data, but now the processing is fired off async in the background.
                Leaving this only a reference of how to do it, if needed in future.

                    //if there are more pages (defined by PAGE_SIZE) you probably want this to be async,
                    //also if this is already a job then don't reschedule

                    if (iterator instanceof ResultsIterator
                                && ((ResultsIterator) iterator).hasPages() && jobExecution == null) {

                        if(logger.isTraceEnabled()){
                            logger.trace("Scheduling notification job as it has multiple pages of devices.");
                        }
                        jobScheduler.scheduleQueueJob(notification, true);
                        em.update(notification);
                        return;
                     }
             ****/

            final UUID appId = em.getApplication().getUuid();
            final Map<String, Object> payloads = notification.getPayloads();

            final Func1<EntityRef, Optional<ApplicationQueueMessage>> sendMessageFunction = deviceRef -> {

                try {

                    //logger.info("Preparing notification queue message for device: {}", deviceRef.getUuid());

                    long now = System.currentTimeMillis();

                    String notifierId = null;
                    String notifierKey = null;

                    //find the device notifier info, match it to the payload
                    for (Map.Entry<String, Object> entry : payloads.entrySet()) {
                        ProviderAdapter adapter = notifierMap.get(entry.getKey().toLowerCase());
                        now = System.currentTimeMillis();
                        String providerId = getProviderId(deviceRef, adapter.getNotifier());
                        if (providerId != null) {
                            notifierId = providerId;
                            notifierKey = entry.getKey().toLowerCase();
                            break;
                        }
                        if (logger.isTraceEnabled()) {
                            logger.trace("Provider query for notification {} device {} took {} ms", notification.getUuid(), deviceRef.getUuid(), (System.currentTimeMillis() - now));
                        }
                    }

                    if (notifierId == null) {
                        //TODO need to leverage optional here
                        return Optional.empty();
                    }

                    ApplicationQueueMessage message = new ApplicationQueueMessage(appId, notification.getUuid(), deviceRef.getUuid(), notifierKey, notifierId);
                    if (notification.getQueued() == null) {

                        // update queued time
                        notification.setQueued(System.currentTimeMillis());

                    }

                    deviceCount.incrementAndGet();

                    return Optional.of(message);


                } catch (Exception deviceLoopException) {
                    logger.error("Failed to add device", deviceLoopException);
                    errorMessages.add("Failed to add device: " + deviceRef.getUuid() + ", error:" + deviceLoopException);

                    return Optional.empty();
                }

            };


            final Map<String, Object> filters = notification.getFilters();

            Observable processMessagesObservable = Observable.create(new IteratorObservable<EntityRef>(iterator))

                .flatMap( entityRef -> {

                    return Observable.just(entityRef).flatMap(ref->{

                        List<Entity> entities = new ArrayList<>();

                            if( ref.getType().equals(User.ENTITY_TYPE)){

                                Query devicesQuery = new Query();
                                devicesQuery.setCollection("devices");
                                devicesQuery.setResultsLevel(Query.Level.CORE_PROPERTIES);
                                devicesQuery.setLimit(50); // for now, assume a user has no more than 50 devices

                                try {

                                   entities = em.searchCollection(new SimpleEntityRef("user", ref.getUuid()), devicesQuery.getCollection(), devicesQuery).getEntities();

                                }catch (Exception e){

                                    logger.error("Unable to load devices for user: {}", ref.getUuid());
                                    return Observable.empty();
                                }


                            }else if ( ref.getType().equals(Device.ENTITY_TYPE)){

                                try{
                                    entities.add(em.get(ref));

                                }catch(Exception e){

                                    logger.error("Unable to load device: {}", ref.getUuid());
                                    return Observable.empty();

                                }

                            }
                        return Observable.from(entities);

                        })
                        .filter( device -> {

                            if(logger.isTraceEnabled()) {
                                logger.trace("Filtering device: {}", device.getUuid());
                            }

                            if(notification.getUseGraph() && filters.size() > 0 ) {

                                for (Map.Entry<String, Object> entry : filters.entrySet()) {

                                    if ((device.getDynamicProperties().get(entry.getKey()) != null &&
                                        device.getDynamicProperties().get(entry.getKey()).equals(entry.getValue())) ||

                                        (device.getProperties().get(entry.getKey()) != null &&
                                            device.getProperties().get(entry.getKey()).equals(entry.getValue()))

                                        ) {


                                        return true;
                                    }

                                }
                                if(logger.isTraceEnabled()) {
                                    logger.trace("Push notification filter did not match for notification {}, so removing from notification",
                                        device.getUuid(), notification.getUuid());
                                }
                                return false;


                            }

                            return true;

                        })
                        .map(sendMessageFunction)
                        .subscribeOn(Schedulers.io());

                }, concurrencyFactor)
                .distinct( queueMessage -> {

                    if(queueMessage.isPresent()) {
                        return queueMessage.get().getDeviceId();
                    }

                    return UUIDUtils.newTimeUUID(); // this should be distinct, default handling for the Optional.empty() case

                } )
                .doOnNext( message -> {
                    try {

                        if(message.isPresent()){

                            if(logger.isTraceEnabled()) {
                                logger.trace("Queueing notification message for device: {}", message.get().getDeviceId());
                            }
                            qm.sendMessageToLocalRegion( message.get(), Boolean.TRUE );
                            queueMeter.mark();
                        }

                    } catch (Exception e) {

                        if(message.isPresent()){
                            logger.error("Unable to queue notification for notification UUID {} and device UUID {} ",
                                message.get().getNotificationId(), message.get().getDeviceId());
                        }
                        else{
                            logger.error("Unable to queue notification as it's not present when trying to send to queue");
                        }

                    }


                })
                .doOnError(throwable -> {

                    logger.error("Error while processing devices for notification : {}, error: {}", notification.getUuid(), throwable.getMessage());
                    notification.setProcessingFinished(-1L);
                    notification.setDeviceProcessedCount(deviceCount.get());
                    logger.warn("Partial notification. Only {} devices processed for notification {}",
                        deviceCount.get(), notification.getUuid());
                    try {
                        em.update(notification);
                    }catch (Exception e){
                        logger.error("Error updating negative processing status when processing failed.");
                    }

                })
                .doOnCompleted( () -> {

                    try {
                        notification.setProcessingFinished(System.currentTimeMillis());
                        notification.setDeviceProcessedCount(deviceCount.get());
                        em.update(notification);
                        if(logger.isTraceEnabled()) {
                            logger.trace("Notification {} finished processing {} device(s)", notification.getUuid(), deviceCount.get());
                        }

                    } catch (Exception e) {
                        logger.error("Unable to set processing finished timestamp for notification");
                    }

                });

            processMessagesObservable.subscribeOn(Schedulers.io()).subscribe(); // fire the queuing into the background

        }

        // update queued time
        Map<String, Object> properties = new HashMap<>(2);
        properties.put("queued", notification.getQueued());
        properties.put("state", notification.getState());
        if (errorMessages.size() > 0) {
            if (notification.getErrorMessage() == null) {
                notification.setErrorMessage("There was a problem delivering all of your notifications. See deliveryErrors in properties");
            }
        }

        notification.addProperties(properties);
        em.update(notification);


        // if no devices, go ahead and mark the batch finished
        if (deviceCount.get() <= 0 ) {
            TaskManager taskManager = new TaskManager(em, notification);
            taskManager.finishedBatch();
        }


    }

    /**
     * only need to get notifiers once. will reset on next batch
     *
     * @return
     */
    private HashMap<Object, ProviderAdapter> getAdapterMap() {
        if (notifierHashMap == null) {
            long now = System.currentTimeMillis();
            notifierHashMap = new HashMap<>();
            Query query = new Query();
            query.setCollection("notifiers");
            query.setLimit(100);
            PathQuery<Notifier> pathQuery = new PathQuery<>(
                new SimpleEntityRef(em.getApplicationRef()),
                query
            );
            Iterator<Notifier> notifierIterator = pathQuery.iterator(em);
            int count = 0;
            while (notifierIterator.hasNext()) {
                Notifier notifier = notifierIterator.next();
                String name = notifier.getName() != null ? notifier.getName() : "";
                UUID uuid = notifier.getUuid() != null ? notifier.getUuid() : UUID.randomUUID();
                ProviderAdapter providerAdapter = ProviderAdapterFactory.getProviderAdapter(notifier, em);
                notifierHashMap.put(name.toLowerCase(), providerAdapter);
                notifierHashMap.put(uuid, providerAdapter);
                notifierHashMap.put(uuid.toString(), providerAdapter);
                if (count++ >= 100) {
                    logger.error("ApplicationQueueManager: too many notifiers...breaking out ", notifierHashMap.size());
                    break;
                }
            }
        }
        return notifierHashMap;
    }

    /**
     * send batches of notifications to provider
     *
     * @param messages
     * @throws Exception
     */
    @Override
    public Observable sendBatchToProviders(final List<LegacyQueueMessage> messages, final String queuePath) {
        if (logger.isTraceEnabled()) {
            logger.trace("sending batch of {} notifications.", messages.size());
        }

        final Map<Object, ProviderAdapter> notifierMap = getAdapterMap();
        final ApplicationQueueManagerImpl proxy = this;
        final ConcurrentHashMap<UUID, TaskManager> taskMap = new ConcurrentHashMap<UUID, TaskManager>(messages.size());
        final ConcurrentHashMap<UUID, Notification> notificationMap = new ConcurrentHashMap<UUID, Notification>(messages.size());

        final Func1<LegacyQueueMessage, ApplicationQueueMessage> func = queueMessage -> {
            boolean messageCommitted = false;
            ApplicationQueueMessage message = null;
            try {
                message = (ApplicationQueueMessage) queueMessage.getBody();
                if (logger.isTraceEnabled()) {
                    logger.trace("start sending notification for device {} for Notification: {} on thread {}", message.getDeviceId(), message.getNotificationId(), Thread.currentThread().getId());
                }

                UUID deviceUUID = message.getDeviceId();

                Notification notification = notificationMap.get(message.getNotificationId());
                if (notification == null) {
                    notification = em.get(message.getNotificationId(), Notification.class);
                    notificationMap.putIfAbsent(message.getNotificationId(), notification);
                }
                TaskManager taskManager = taskMap.get(message.getNotificationId());
                if (taskManager == null) {
                    taskManager = new TaskManager(em, notification);
                    taskMap.putIfAbsent(message.getNotificationId(), taskManager);
                    taskManager = taskMap.get(message.getNotificationId());
                }

                final Map<String, Object> payloads = notification.getPayloads();
                final Map<String, Object> translatedPayloads = translatePayloads(payloads, notifierMap);
                if (logger.isTraceEnabled()) {
                    logger.trace("sending notification for device {} for Notification: {}", deviceUUID, notification.getUuid());
                }

                try {
                    String notifierName = message.getNotifierKey().toLowerCase();
                    ProviderAdapter providerAdapter = notifierMap.get(notifierName.toLowerCase());
                    Object payload = translatedPayloads.get(notifierName);

                    TaskTracker tracker = null;

                    if(notification.getSaveReceipts()){

                        final Receipt receipt =
                            new Receipt( notification.getUuid(), message.getNotifierId(), payload, deviceUUID );
                        tracker =
                            new TaskTracker( providerAdapter.getNotifier(), taskManager, receipt, deviceUUID );

                    }
                    else {

                        tracker =
                            new TaskTracker( providerAdapter.getNotifier(), taskManager, null, deviceUUID );
                    }
                    if (!isOkToSend(notification)) {
                        tracker.failed(0, "Notification is duplicate/expired/cancelled.");
                    } else {
                        if (payload == null) {
                            if (logger.isDebugEnabled()) {
                                logger.debug("selected device {} for notification {} doesn't have a valid payload. skipping.", deviceUUID, notification.getUuid());
                            }
                            tracker.failed(0, "failed to match payload to " + message.getNotifierId() + " notifier");
                        } else {
                            long now = System.currentTimeMillis();
                            try {
                                providerAdapter.sendNotification(message.getNotifierId(), payload, notification, tracker);
                            } catch (Exception e) {
                                tracker.failed(0, e.getMessage());
                            } finally {
                                if (logger.isTraceEnabled()) {
                                    logger.trace("sending to device {} for Notification: {} duration {} ms", deviceUUID, notification.getUuid(), (System.currentTimeMillis() - now));
                                }
                            }
                        }
                    }
                    messageCommitted = true;
                } finally {
                    sendMeter.mark();
                }

            } catch (Exception e) {
                logger.error("Failure while sending", e);
                try {
                    if (!messageCommitted && queuePath != null) {
                        qm.commitMessage(queueMessage);
                    }
                } catch (Exception queueException) {
                    logger.error("Failed to commit message.", queueException);
                }
            }
            return message;
        };

        //from each queue message, process them in parallel up to 10 at a time
        Observable queueMessageObservable = Observable.from(messages).flatMap(queueMessage -> {


            return Observable.just(queueMessage).map(func).buffer(messages.size()).map(queueMessages -> {
                //for gcm this will actually send notification
                for (ProviderAdapter providerAdapter : notifierMap.values()) {
                    try {
                        providerAdapter.doneSendingNotifications();
                    } catch (Exception e) {
                        logger.error("providerAdapter.doneSendingNotifications: ", e);
                    }
                }
                //TODO: check if a notification is done and mark it
                HashMap<UUID, ApplicationQueueMessage> notifications = new HashMap<>();
                for (ApplicationQueueMessage message : queueMessages) {
                    if (notifications.get(message.getNotificationId()) == null) {
                        try {
                            TaskManager taskManager = taskMap.get(message.getNotificationId());
                            notifications.put(message.getNotificationId(), message);
                            taskManager.finishedBatch();
                        } catch (Exception e) {
                            logger.error("Failed to finish batch", e);
                        }
                    }
                }
                return notifications;
            }).doOnError(throwable -> logger.error("Failed while sending", throwable));
        }, 10);

        return queueMessageObservable;
    }

    @Override
    public void stop() {
        for (ProviderAdapter adapter : getAdapterMap().values()) {
            try {
                adapter.stop();
            } catch (Exception e) {
                logger.error("failed to stop adapter", e);
            }
        }
    }


    /**
     *  Validates that a notifier and adapter exists to send notifications to. For the example payload
     *
     *  { "payloads" : {"winphone":"mymessage","apple":"mymessage"} }
     *
     *  Notifiers with name "winphone" and "apple" must exist.
     */
    private Map<String, Object> translatePayloads(Map<String, Object> payloads,
                                                  Map<Object, ProviderAdapter> notifierMap) throws Exception {

        final Map<String, Object> translatedPayloads = new HashMap<String, Object>(payloads.size());

        for (Map.Entry<String, Object> entry : payloads.entrySet()) {

            String payloadKey = entry.getKey().toLowerCase();
            Object payloadValue = entry.getValue();

            //look for adapter from payload map
            ProviderAdapter providerAdapter = notifierMap.get(payloadKey);
            if (providerAdapter != null) {

                //translate payload to usable information
                Object translatedPayload = payloadValue != null ? providerAdapter.translatePayload(payloadValue) : null;
                if (translatedPayload != null) {
                    translatedPayloads.put(payloadKey, translatedPayload);
                }

            }
        }
        return translatedPayloads;
    }

    public static String getQueueNames(Properties properties) {

        String name = properties.getProperty(ApplicationQueueManagerImpl.DEFAULT_QUEUE_PROPERTY,
            ApplicationQueueManagerImpl.DEFAULT_QUEUE_NAME);
        return name;

    }

    private static final class IteratorObservable<T> implements rx.Observable.OnSubscribe<T> {
        private final Iterator<T> input;

        private IteratorObservable(final Iterator input) {
            this.input = input;
        }

        @Override
        public void call(final Subscriber<? super T> subscriber) {

            /**
             * You would replace this code with your file reading.  Instead of emitting from an iterator,
             * you would create a bean object that represents the entity, and then emit it
             */

            try {
                while (!subscriber.isUnsubscribed() && input.hasNext()) {

                    //send our input to the next
                    subscriber.onNext((T) input.next());

                }

                //tell the subscriber we don't have any more data
                subscriber.onCompleted();

            } catch (Throwable t) {
                logger.error("failed on subscriber", t);
                subscriber.onError(t);
            }
        }
    }

    @Override
    public void asyncCheckForInactiveDevices() throws Exception {
        Collection<ProviderAdapter> providerAdapters = getAdapterMap().values();
        for (final ProviderAdapter providerAdapter : providerAdapters) {
            try {
                if (providerAdapter != null) {
                    if (logger.isTraceEnabled()) {
                        logger.trace("checking notifier {} for inactive devices", providerAdapter.getNotifier());
                    }
                    providerAdapter.removeInactiveDevices();

                    if (logger.isTraceEnabled()) {
                        logger.trace("finished checking notifier {} for inactive devices", providerAdapter.getNotifier());
                    }
                }
            } catch (Exception e) {
                // not essential so don't fail, but log
                logger.error("checkForInactiveDevices", e);

            }
        }
    }


    private boolean isOkToSend(Notification notification) {

        if (notification.getCanceled() == Boolean.TRUE) {
            if (logger.isDebugEnabled()) {
                logger.debug("Notification {} canceled. Not sending.",
                    notification.getUuid());
            }
            return false;
        }
        if (notification.isExpired()) {
            if (logger.isDebugEnabled()) {
                logger.debug("Notification {} expired. Not sending.",
                    notification.getUuid());
            }
            return false;
        }
        return true;
    }


    private String getProviderId(EntityRef device, Notifier notifier) throws Exception {
        try {
            Object value = em.getProperty(device, notifier.getName() + NOTIFIER_ID_POSTFIX);
            if (value == null) {
                value = em.getProperty(device, notifier.getUuid() + NOTIFIER_ID_POSTFIX);
            }
            return value != null ? value.toString() : null;
        } catch (Exception e) {
            logger.error("Error getting notifier for device {}, proceeding with rest of batch", device, e);
            return null;
        }
    }


}