/*
 * ICESOFT COMMERCIAL SOURCE CODE LICENSE V 1.1
 *
 * The contents of this file are subject to the ICEsoft Commercial Source
 * Code License Agreement V1.1 (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.icesoft.com/license/commercial-source-v1.1.html
 *
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
 * License for the specific language governing rights and limitations under
 * the License.
 *
 * Copyright 2009-2015 ICEsoft Technologies Canada, Corp. All Rights Reserved.
 */

package com.icesoft.push;

import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.ArrayList;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.icesoft.push.messaging.MessagePayload;
import com.icesoft.push.messaging.PushMessageService;
import org.icepush.Configuration;

public class StatusManager {
    private static final Logger LOGGER = Logger.getLogger(StatusManager.class.getName());

    private final UUID uuid = UUID.randomUUID();
    {
        if (LOGGER.isLoggable(Level.FINE)) {
            LOGGER.log(Level.FINE, "Enterprise Push Server UUID '" + uuid + "'");
        }
    }
    private final Comparator uuidComparator = 
        new Comparator<UUID>() {
            public int compare(final UUID uuid1, final UUID uuid2) {
                if (uuid1.getMostSignificantBits() < uuid2.getMostSignificantBits()) {
                    return -1;
                } else if (uuid1.getMostSignificantBits() > uuid2.getMostSignificantBits()) {
                    return 1;
                } else if (uuid1.getLeastSignificantBits() < uuid2.getLeastSignificantBits()) {
                    return -1;
                } else if (uuid1.getLeastSignificantBits() > uuid2.getLeastSignificantBits()) {
                    return 1;
                } else {
                    return 0;
                }
            }
        };

    private final ConcurrentHashMap<UUID, Record>
        uuidRecordMap = new ConcurrentHashMap();
    {
        uuidRecordMap.put(uuid, new Record(uuid, null));
    }

    private final Map<String, Long> clockSkewMap = new HashMap<String, Long>();
    private final long clockSkewThreshold;
    private final ScheduledThreadPoolExecutor executor;

    private long lastScanTimestamp;
    private LocalPushGroupManager localPushGroupManager;
    private long logTimestamp;
    private PushMessageService pushMessageService;
    private StatusTask statusTask;
    private ScheduledFuture scheduledFuture;

    public StatusManager(
        final ScheduledThreadPoolExecutor executor, final LocalPushGroupManager localPushGroupManager,
        final Configuration configuration) {

        this.executor = executor;
        this.localPushGroupManager = localPushGroupManager;
        this.statusTask = new StatusTask(localPushGroupManager, this);
        this.clockSkewThreshold = configuration.getAttributeAsLong("clockSkewThreshold", 250);
    }

    public UUID getUUID() {
        return uuid;
    }

    public void handleShutdown(final UUID failureUUID) {
        if (LOGGER.isLoggable(Level.INFO)) {
            LOGGER.log(
                Level.INFO,
                "Handle shutdown for " +
                    "EPS instance '" + failureUUID + "@" + uuidRecordMap.get(failureUUID).nodeAddress + "'");
        }
        // Handle graceful or non-graceful shutdown
        //not supported on JDK 1.5
//        UUID _successorUUID = uuidRecordMap.higherKey(failureUUID);
//        if (_successorUUID == null) {
//            _successorUUID = uuidRecordMap.firstKey();
//        }
        UUID _successorUUID = getHigherKey(failureUUID);
        if (LOGGER.isLoggable(Level.INFO)) {
            String _successorNodeAddress = uuidRecordMap.get(_successorUUID).nodeAddress;
            LOGGER.log(
                Level.INFO,
                "EPS instance '" + failureUUID + "@" + uuidRecordMap.get(failureUUID).nodeAddress + "' " +
                    "to be taken over by " +
                "EPS instance " +
                    "'" + _successorUUID + "@" + (_successorNodeAddress != null ? _successorNodeAddress : "localhost") +
                    "'.");
        }
        uuidRecordMap.remove(failureUUID);
        if (_successorUUID != null) {
            if (_successorUUID.equals(this.uuid)) {
                for (final Map.Entry<String, org.icepush.PushID> _entry :
                        localPushGroupManager.getPushIDMap().entrySet()) {

                    PushID _pushID = (PushID)_entry.getValue();
                    if (failureUUID.equals(((PushID.Status)_pushID.getStatus()).getConfirmationHandlerUUID())) {
                        _pushID.resumeConfirmationTimeout();
                    }
                }
            } else {
                for (final Map.Entry<String, org.icepush.PushID> _entry :
                        localPushGroupManager.getPushIDMap().entrySet()) {

                    PushID.Status _status = (PushID.Status)_entry.getValue().getStatus();
                    if (failureUUID.equals(_status.getConfirmationHandlerUUID())) {
                        _status.setConfirmationHandlerUUID(_successorUUID);
                    }
                }
            }
        }
    }

    //more efficient would be to update the sorted collection
    //only when necessary
    private UUID getHigherKey(UUID key)  {
        ArrayList<UUID> keyList = new ArrayList(uuidRecordMap.keySet());
        Collections.sort(keyList, uuidComparator);
        int keyIndex = keyList.indexOf(key);
        keyIndex++;
        if (keyIndex >= keyList.size())  {
            return keyList.get(0);
        }
        return keyList.get(keyIndex);
    }

    public void handleStatus(
        final String remoteNodeAddress, final UUID remoteUUID, final long remoteTimestamp,
        final Map<String, PushID.Status> remotePushIDStatusMap) {

        Record _record = uuidRecordMap.get(remoteUUID);
        if (_record == null) {
            uuidRecordMap.put(remoteUUID, _record = new Record(remoteUUID, remoteNodeAddress));
        }
        _record.touch();
        long _now = System.currentTimeMillis();
        long _difference = Math.abs(_now - remoteTimestamp);
        if (_difference > clockSkewThreshold) {
            if (LOGGER.isLoggable(Level.FINE)) {
                LOGGER.log(
                    Level.FINE,
                    "Remote: '" + remoteNodeAddress + "', Difference: " + _difference + " ms.");
            }
            if (clockSkewMap.put(remoteNodeAddress, _difference) == null) {
                logClockSkews();
            } else {
                logClockSkewsConditionally();
            }
        } else if (clockSkewMap.remove(remoteNodeAddress) != null) {
            logClockSkews();
        }
        for (final String remotePushID : remotePushIDStatusMap.keySet()) {
            PushID localPushID = (PushID)localPushGroupManager.getPushIDMap().get(remotePushID);
            if (localPushID != null) {
                PushID.Status _localStatus = (PushID.Status)localPushID.getStatus();
                PushID.Status _remoteStatus = remotePushIDStatusMap.get(remotePushID);
                if (remoteUUID.equals(_remoteStatus.getConfirmationHandlerUUID())) {
                    _localStatus.setConfirmationHandlerUUID(remoteUUID);
                    _localStatus.setConfirmationTimeoutSequenceNumber(
                        _remoteStatus.getConfirmationTimeoutSequenceNumber());
                }
                long remoteSequenceNumber = _remoteStatus.getSequenceNumber();
                if (localPushID.getSequenceNumber() < remoteSequenceNumber) {
                    // The PushID.Status' connectionRecreationTimeout and backupConnectionRecreationTimeout values are
                    // only changed upon a receiving listen.icepush request, which means that the PushID.Status with the
                    // highest sequenceNumber has the more up-to-date values.
                    _localStatus.setConnectionRecreationTimeout(
                        _remoteStatus.getConnectionRecreationTimeout());
                    _localStatus.setBackupConnectionRecreationTimeout(
                            _remoteStatus.getBackupConnectionRecreationTimeout());
                    localPushID.setLastHandledByLocal(false);
                    localPushID.cancelConfirmationTimeoutIfOlder(remoteSequenceNumber);
                    localPushID.restartExpiryTimeoutIfOlder(remoteSequenceNumber);
                }
                localPushID.clearConfirmationTimeoutIfMatching(_remoteStatus.getCloudPushSequenceNumber());
            }
        }
    }

    public void shutdown() {
        if (scheduledFuture != null) {
            scheduledFuture.cancel(true);
        }
        // Graceful shutdown
        MessagePayload _messagePayload = new MessagePayload();
        _messagePayload.setUUID(uuid);
        pushMessageService.publishNow(_messagePayload, PushMessageService.SHUTDOWN_MESSAGE_TYPE);
    }

    PushMessageService getPushMessageService() {
        return pushMessageService;
    }

    void setPushMessageService(final PushMessageService pushMessageService) {
        this.pushMessageService = pushMessageService;
        scheduledFuture = executor.scheduleAtFixedRate(statusTask, 1000, 1000, TimeUnit.MILLISECONDS);
    }

    private void logClockSkews() {
        if (LOGGER.isLoggable(Level.FINE)) {
            StringBuilder _stringBuilder = new StringBuilder();
            _stringBuilder.append("Clock Skews:\r\n\r\n");
            for (final Map.Entry _mapEntry : clockSkewMap.entrySet()) {
                _stringBuilder.
                    append("Remote: '").append(_mapEntry.getKey()).append("', ").
                    append("Difference: ").append(_mapEntry.getValue()).append(" ms.\r\n");
            }
            LOGGER.log(Level.FINE, _stringBuilder.toString());
            logTimestamp = System.currentTimeMillis();
        }
    }

    private void logClockSkewsConditionally() {
        if (System.currentTimeMillis() - logTimestamp >= 60000) {
            logClockSkews();
        }
    }

    private void scan() {
        long _now = System.currentTimeMillis();
        if (_now - lastScanTimestamp > 5000) {
            lastScanTimestamp = _now;
            for (final Map.Entry<UUID, Record> _entry : uuidRecordMap.entrySet()) {
                UUID _uuid = _entry.getKey();
                if (!uuid.equals(_uuid)) {
                    Record _record = _entry.getValue();
                    if (_now - _record.timestamp > 5000) {
                        // Non-graceful shutdown detected.
                        if (LOGGER.isLoggable(Level.WARNING)) {
                            LOGGER.log(
                                Level.WARNING,
                                "EPS instance '" + _uuid + "@" + _record.nodeAddress + "' has disappeared.");
                        }
                        handleShutdown(_entry.getKey());
                    }
                }
            }
        }
    }

    public static class Record {
        private final String nodeAddress;
        private final UUID uuid;

        private long timestamp;

        private Record(final UUID uuid, final String nodeAddress) {
            this.uuid = uuid;
            this.nodeAddress = nodeAddress;
        }

        private void touch() {
            timestamp = System.currentTimeMillis();
        }
    }

    public static class StatusTask
    implements Runnable {
        private final LocalPushGroupManager localPushGroupManager;
        private final StatusManager statusManager;

        private StatusTask(final LocalPushGroupManager localPushGroupManager, final StatusManager statusManager) {
            this.localPushGroupManager = localPushGroupManager;
            this.statusManager = statusManager;
        }

        public void run() {
            try {
                PushMessageService _pushMessageService = statusManager.getPushMessageService();
                if (_pushMessageService != null) {
                    MessagePayload messagePayload = new MessagePayload();
                    messagePayload.setPushIDStatusMap(
                        localPushGroupManager.getPushIDStatusMap());
                    messagePayload.setTimestamp(System.currentTimeMillis());
                    messagePayload.setUUID(statusManager.getUUID());
                    _pushMessageService.publish(
                        messagePayload,
                        PushMessageService.STATUS_MESSAGE_TYPE);
                    statusManager.scan();
                }
            } catch (final Exception exception) {
                if (LOGGER.isLoggable(Level.WARNING)) {
                    LOGGER.log(
                        Level.WARNING,
                        "Exception caught on " + this.getClass().getName() + " Runnable.",
                        exception);
                }
            }
        }
    }
}
