package id.iptek.utms.agent.worker;

import com.google.gson.Gson;
import id.iptek.utms.agent.AppConfig;
import id.iptek.utms.agent.GenericCache;
import id.iptek.utms.agent.dao.DeleteTaskDao;
import id.iptek.utms.agent.dao.DownloadTaskDao;
import id.iptek.utms.agent.model.Application;
import id.iptek.utms.agent.model.ApplicationSimple;
import id.iptek.utms.agent.model.PendingDeleteTask;
import id.iptek.utms.agent.model.PendingDownloadTask;
import id.iptek.utms.agent.model.SimplePendingDownloadTask;
import id.iptek.utms.agent.util.IOUtil;
import java.io.File;
import java.io.FileInputStream;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang.RandomStringUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Jaka
 */
public abstract class Worker implements MqttCallbackExtended, Runnable {

    protected Logger logger;
    private String brokerUrl;
    private String loggingId;
    private String topic;
    private String clientIdPrefix;
    private String user;
    private String password;
    private long sleep;
    private long keepAlive;
    private boolean cleanSession;
    private int maxInflight;
    private boolean autoReconnect;
    private long reconnectDelayMillis;
    
    /**
     * mqtt *
     */
    private MemoryPersistence persistence = new MemoryPersistence();
    protected MqttClient mqttClient;
    private volatile boolean running = false;

    private DeleteTaskDao deleteTaskDao = new DeleteTaskDao();
    private DownloadTaskDao downloadTaskDao = new DownloadTaskDao();
    private static final SimpleDateFormat dateTimeFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
    private static final SimpleDateFormat sdfYEAR = new SimpleDateFormat("yyyy");
    private static final SimpleDateFormat sdfMONTH = new SimpleDateFormat("MM");
    private static final SimpleDateFormat sdfDATE = new SimpleDateFormat("dd");

    protected ExecutorService executorService = null;
    
    public Worker() {
    }

    public void setExecutorService(ExecutorService executorService) {
        this.executorService = executorService;
    }

    /**
     * @return the brokerUrl
     */
    public String getBrokerUrl() {
        return brokerUrl;
    }

    /**
     * @return the loggingId
     */
    public String getLoggingId() {
        return loggingId;
    }

    /**
     * @return the topic
     */
    public String getTopic() {
        return topic;
    }

    /**
     * @return the clientIdPrefix
     */
    public String getClientIdPrefix() {
        return clientIdPrefix;
    }

    /**
     * @return the user
     */
    public String getUser() {
        return user;
    }

    /**
     * @return the password
     */
    public String getPassword() {
        return password;
    }

    /**
     * @return the sleep
     */
    public long getSleep() {
        return sleep;
    }

    /**
     * @return the keepAlive
     */
    public long getKeepAlive() {
        return keepAlive;
    }

    /**
     * @return the cleanSession
     */
    public boolean isCleanSession() {
        return cleanSession;
    }

    /**
     * @return the maxInflight
     */
    public int getMaxInflight() {
        return maxInflight;
    }

    /**
     * @return the autoReconnect
     */
    public boolean isAutoReconnect() {
        return autoReconnect;
    }

    void init(String consumerName, AppConfig appConfig) throws Exception {
        String _brokerUrl = appConfig.get("mqtt.broker.url");
        String _loggingId = appConfig.get("mqtt.consumer." + consumerName + ".loggingid");
        String _topic = appConfig.get("mqtt.consumer." + consumerName + ".topic");
        String _clientIdPrefix = appConfig.get("mqtt.consumer." + consumerName + ".clientid_prefix");
        String _user = appConfig.get("mqtt.consumer." + consumerName + ".user");
        String _password = appConfig.get("mqtt.consumer." + consumerName + ".password");
        long _sleep = appConfig.getAsLong("mqtt.consumer." + consumerName + ".sleep");
        long _keepAlive = appConfig.getAsLong("mqtt.consumer." + consumerName + ".keepalive", 10L);
        int _maxInflight = appConfig.getAsInt("mqtt.consumer." + consumerName + ".maxinflight", 32);
        boolean _cleanSession = appConfig.getAsBoolean("mqtt.consumer." + consumerName + ".cleansession", true);
        boolean _autoReconnect = appConfig.getAsBoolean("mqtt.consumer." + consumerName + ".autoreconnect", true);
        long _reconnectDelayMillis = appConfig.getAsLong("mqtt.consumer." + consumerName + ".reconnect.delay.millis", 5000L);

        this.brokerUrl = _brokerUrl;
        this.loggingId = _loggingId;
        this.topic = _topic;
        this.clientIdPrefix = _clientIdPrefix;
        this.user = _user;
        this.password = _password;
        this.sleep = _sleep;
        this.keepAlive = _keepAlive;
        this.maxInflight = _maxInflight;
        this.cleanSession = _cleanSession;
        this.autoReconnect = _autoReconnect;
        this.reconnectDelayMillis = _reconnectDelayMillis;

        // init logger
        //logger = LoggerFactory.getLogger(getClass() + "." + _loggingId);
        logger = LoggerFactory.getLogger(getClass());
    }

    public void start() throws Exception {
        try {
            this.running = true;

            logger.info("Starting ...");
            String clientId = getClientIdPrefix() + RandomStringUtils.randomAlphanumeric(10);
            mqttClient = new MqttClient(brokerUrl, clientId, persistence);

        } finally {
        }
    }

    public void stop() throws Exception {
        try {
            this.running = false;
            if (mqttClient != null) {
                mqttClient.disconnectForcibly(0L, 250L, false);
                mqttClient.close(true);
            }
        } finally {
            // no-op
        }
    }

    @Override
    public void connectionLost(Throwable thrwbl) {
        logger.warn("Connection Lost: " + thrwbl);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken imdt) {
        // no-op
    }

    @Override
    public void connectComplete(boolean bln, String string) {
        if (getTopic() != null) {
            try {
                logger.info("Connected, subscribe to '{}'", getTopic());
                mqttClient.subscribe(getTopic());
            } catch (Exception ex) {
                logger.error("Error subscribe topic: " + topic);
            }
        } else {
            logger.debug("No topic subscribed!");
        }
    }

    @Override
    public void run() {
        do {
            try {
                if (mqttClient != null && !mqttClient.isConnected()) {
                    MqttConnectOptions connOpts = new MqttConnectOptions();
                    connOpts.setCleanSession(isCleanSession());
                    connOpts.setUserName(getUser());
                    if (getPassword() != null) {
                        connOpts.setPassword(getPassword().toCharArray());
                    }
                    connOpts.setAutomaticReconnect(isAutoReconnect());
                    connOpts.setKeepAliveInterval((int) getKeepAlive());
                    connOpts.setMaxInflight(getMaxInflight());
                    mqttClient.setCallback(this);
                    logger.info("Started ...");

                    logger.info("Try to connect ...");
                    try {
                        mqttClient.connect(connOpts);
                    } catch (Exception ex) {
                        logger.error("Error connect MQTT broker: {}", ex.getMessage(), ex);
                        sleepBeforeReconnect();
                        continue;
                    }
                }

                // do something
                try {
                    doSomething();
                } catch (Exception ex) {
                    logger.error("Error doing something: {}", ex.getMessage(), ex);
                }

                if (getSleep() != 0L) {
                    try {
                        logger.info("Sleep for {} ms", getSleep());
                        TimeUnit.MILLISECONDS.sleep(getSleep());
                    } catch (InterruptedException ex) {
                        logger.error("Sleep got interrupted: {}", ex.getMessage(), ex);
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            } catch (Exception ex) {
                logger.error("Error run: {}", ex.getMessage(), ex);
            }
        } while (running);
    }

    private void sleepBeforeReconnect() {
        try {
            logger.info("Wait {} ms before reconnecting MQTT broker", reconnectDelayMillis);
            TimeUnit.MILLISECONDS.sleep(reconnectDelayMillis);
        } catch (InterruptedException ex) {
            logger.error("Reconnect wait interrupted: {}", ex.getMessage(), ex);
            Thread.currentThread().interrupt();
            running = false;
        }
    }

    protected void doSomething() throws Exception {
    }

    /*
    protected void publishPendingTasks(String terminalSN) {
        boolean dummy = true;
        if (dummy) {
            return;
        }

        // pending delete task
        List<PendingDeleteTask> pendingDeleteTaskList = deleteTaskDao.getPendingTaskForTerminal(terminalSN);
        pendingDeleteTaskList.forEach((t) -> {
            try {
                broadcastDeleteTasksToDevice(mqttClient, t);
            } catch (Exception ex) {
                logger.error("Error push delete task: [{}, {}, {}]", t.getTerminalSN(), t.getId(), t.getLogId());
            }
        });

        // pending download task
        List<PendingDownloadTask> pendingDownloadTaskList = downloadTaskDao.getPendingTaskForTerminal(terminalSN);
        pendingDownloadTaskList.forEach((t) -> {
            try {
                broadcastDownloadTasksToDevice(mqttClient, t);
            } catch (Exception ex) {
                logger.error("Error push download task: [{}, {}, {}]", t.getTerminalSN(), t.getId(), t.getLogId());
            }
        });
    }

    protected void publishPendingTasksForAll(int limit, int periodeMinutes, boolean checkOnline) {
        // pending delete task
        Date currentTime = downloadTaskDao.getCurrentTime();
        logger.debug("Current server time: {}", currentTime);
        logger.debug("Get all pending delete tasks ..");
        List<PendingDeleteTask> pendingDeleteTaskList = null;
        
        if(checkOnline) {
            pendingDeleteTaskList = deleteTaskDao.getPendingTaskForOnlineTerminals(periodeMinutes, limit);
        } else {
            pendingDeleteTaskList = deleteTaskDao.getPendingTaskForTerminals(periodeMinutes, limit);
        }
        logger.debug("... got {} task item.", pendingDeleteTaskList.size());
        pendingDeleteTaskList.forEach((t) -> {
            try {
                broadcastDeleteTasksToDevice(mqttClient, t);
            } catch (Exception ex) {
                logger.error("Error push delete task: [{}, {}, {}]", t.getTerminalSN(), t.getId(), t.getLogId(), ex);
            }
        });
        
        logger.debug("Get all pending download tasks ..");
        List<PendingDownloadTask> pendingDownloadTaskList = null;
        if(checkOnline) {
            pendingDownloadTaskList = downloadTaskDao.getPendingTaskForOnlineTerminals(periodeMinutes, limit);
        } else {
            pendingDownloadTaskList = downloadTaskDao.getPendingTaskForTerminals(periodeMinutes, limit);
        }
        logger.debug("... got {} task item.", pendingDownloadTaskList.size());
        pendingDownloadTaskList.forEach((t) -> {
            try {
                broadcastDownloadTasksToDevice(mqttClient, t);
            } catch (Exception ex) {
                logger.error("Error push download task: [{}, {}, {}]", t.getTerminalSN(), t.getId(), t.getLogId(), ex);
            }
        });
    }

    protected void publishBroadcastPendingTasksForAll() {
        // pending delete task
        Date currentTime = downloadTaskDao.getCurrentTime();
        logger.debug("Current server time: {}", currentTime);
        logger.debug("Get all pending download tasks ..");
        // pending download task
        logger.debug("Get next BROADCAST pending download tasks ..");
        PendingDownloadTask nextBroadcastDownloadTask = downloadTaskDao.getNextPendingTaskForBroadcast();
        if(nextBroadcastDownloadTask != null) {
            logger.debug("Pending BROADCAST download task: \"{}\"", nextBroadcastDownloadTask.getName());
            try {
                // cache last broadcasted task
                GenericCache taskCache = GenericCache.getCache("LAST_BROADCAST_DOWNLOAD_TASK");
                taskCache.put("task_id", nextBroadcastDownloadTask.getId());
                
                broadcastDownloadTasksToDevice(mqttClient, nextBroadcastDownloadTask);
            } catch(Exception ex) {
                logger.error("Error BROADCAST download task: [{}]", nextBroadcastDownloadTask.getId());
            }
        }
    }

    // send delete task to device topic via mqtt
    private void broadcastDeleteTasksToDevice(MqttClient mqttClient,
            PendingDeleteTask task) throws Exception {
        logger.debug("Broadcast tasks ...");
        try {
            int qos = 2;
            String topicName = task.getTerminalSN().toUpperCase() + "_IN";
            // generate json
            Gson gson = new Gson();
            Map messageMap = new HashMap<>();
            messageMap.put("req_id", UUID.randomUUID().toString());
            messageMap.put("req_time", dateTimeFormat.format(new Date()));
            // we'll use log's id as ack_id
            messageMap.put("ack_id", task.getLogId());
            messageMap.put("req_type", "DELETE_APP_TASK");
            // task map
            Map taskMap = new HashMap<>();
            taskMap.put("id", task.getId());
            taskMap.put("name", task.getName());
            taskMap.put("delete_time", task.getDeleteTime() != null ? dateTimeFormat.format(task.getDeleteTime()) : null);
            messageMap.put("task", taskMap);
            // app map
            ApplicationSimple app = task.getApp();
            Map appMap = new HashMap<>();
            appMap.put("id", app.getId());
            appMap.put("name", app.getAppName());
            appMap.put("package_name", app.getPackageName());
            messageMap.put("app", appMap);

            String messageData = gson.toJson(messageMap);

            logger.debug("Try publish message to: {}", topicName);
            mqttClient.publish(topicName, messageData.getBytes(), qos, false);
            logger.debug("Message published!");

            // change log's last broadcast time
            boolean updated = deleteTaskDao.saveLastBroadcastTs(task);
            logger.debug("Delete task log last broadcast time updated? {}", updated);
        } finally {
            logger.debug("Start broadcast delete tasks DONE");
        }
    }

    // send download task to device topic via mqtt
    protected void broadcastDownloadTasksToDevice(MqttClient mqttClient,
            PendingDownloadTask task) throws Exception {
        logger.debug("Broadcast tasks ...");
        try {
            int qos = 2;
            String topicName;
            if(!"SN".equals(task.getTerminalSN().toUpperCase())) {
                topicName = task.getTerminalSN().toUpperCase() + "_IN";
            } else {
                topicName = "SERVER_OUT";
            }
            // generate json
            Gson gson = new Gson();
            Map messageMap = new HashMap<>();
            messageMap.put("req_id", UUID.randomUUID().toString());
            messageMap.put("req_time", dateTimeFormat.format(new Date()));
            // we'll use log's id as ack_id
            messageMap.put("ack_id", task.getLogId());
            messageMap.put("req_type", "DOWNLOAD_APP_TASK");
            // task map
            Map taskMap = new HashMap<>();
            taskMap.put("id", task.getId());
            taskMap.put("name", task.getName());
            taskMap.put("download_time_type", getDownloadTimeTypeName(task.getDownloadTimeType()));
            if (task.getDownloadTimeType() == 2) {
                taskMap.put("download_time", dateTimeFormat.format(task.getDownloadTime()));
            }
            taskMap.put("installation_time_type", getInstallationTimeTypeName(task.getInstallationTimeType()));
            if (task.getInstallationTimeType() == 2) {
                taskMap.put("installation_time", dateTimeFormat.format(task.getInstallationTime()));
            }
            taskMap.put("installation_notification", getInstallationNotificationName(task.getInstallationNotification()));
            messageMap.put("task", taskMap);
            // app map
            Application app = task.getApp();
            Map appMap = new HashMap<>();
            appMap.put("id", app.getId());
            appMap.put("name", app.getAppName());
            appMap.put("package_name", app.getPackageName());
            appMap.put("version", app.getVersion());
            appMap.put("company", app.getCompanyName());
            appMap.put("uninstallable", app.isUninstallable());
            appMap.put("description", app.getDescription());
            if (app.getDownloadUrl() != null) {
                appMap.put("download_url", app.getDownloadUrl());
                appMap.put("md5_checksum", app.getChecksum());
            } else {
                String downloadURL = AppConfig.getInstance().get("file.download.url");
                if (downloadURL != null) {
                    downloadURL = downloadURL.replace("{$id}", app.getApkId());
                }
                appMap.put("download_url", downloadURL);
                String filePath = getFilePath(app.getFileId(), app.getFileExt(), app.getFileCreateDate());
                appMap.put("md5_checksum", md5Checksum(IOUtil.read(new FileInputStream(filePath))));
            }
            messageMap.put("app", appMap);

            String messageData = gson.toJson(messageMap);

            logger.debug("Try publish message to: {}", topicName);
            mqttClient.publish(topicName, messageData.getBytes(), qos, false);
            logger.debug("Message published!");

            // change log's last broadcast time
            if(!"BROADCAST".equals(task.getLogId())) {
                boolean updated = downloadTaskDao.saveLastBroadcastTs(task);
                logger.debug("Download task log last broadcast time updated? {}", updated);
            } else {
                logger.debug("Do not update last broadcast time because this is broadcasted task!");
            }
        } finally {
            logger.debug("Start broadcast download tasks DONE");
        }
    }
    
    // send NONE ack for invalid download task-SN to device topic via mqtt
    protected void broadcastNoneDownloadTasksToDevice(MqttClient mqttClient,
            String sn, String taskId) throws Exception {
        logger.debug("Broadcast NONE tasks ...");
        try {
            int qos = 2;
            String topicName = sn.toUpperCase() + "_IN";
            // generate json
            Gson gson = new Gson();
            Map messageMap = new HashMap<>();
            messageMap.put("req_id", UUID.randomUUID().toString());
            messageMap.put("req_type", "NONE");
            messageMap.put("req_time", dateTimeFormat.format(new Date()));
            messageMap.put("task_id", taskId);

            String messageData = gson.toJson(messageMap);

            logger.debug("Try publish message to: {}", topicName);
            mqttClient.publish(topicName, messageData.getBytes(), qos, false);
            logger.debug("Message published!");
        } finally {
            logger.debug("Start broadcast NONE download tasks DONE");
        }
    }

    protected void publishBroadcastCheckPendingTasksForAll() {
        // pending check task
        Date currentTime = downloadTaskDao.getCurrentTime();
        logger.debug("Current server time: {}", currentTime);
        logger.debug("Get all pending check tasks ..");
        List<SimplePendingDownloadTask> tasks = downloadTaskDao.getPendingTasksForBroadcast();
        if(!tasks.isEmpty()) {
            tasks.forEach((task) -> {
                try {
                    logger.debug("Pending BROADCAST check task: \"{} - {}\"", task.getId(), task.getName());
                    broadcastCheckTaskToDevice(mqttClient, task);
                } catch(Exception ex) {
                    logger.error("Error BROADCAST check task: [{} - {}]", task.getId(), task.getName());
                }
            });
        }
    }

    // broadcast check task to all device topic via mqtt
    private void broadcastCheckTaskToDevice(MqttClient mqttClient,
            SimplePendingDownloadTask task) throws Exception {
        logger.debug("Broadcast tasks ...");
        try {
            int qos = 2;
            String topicName = "SERVER_OUT";
            // generate json
            Gson gson = new Gson();
            Map messageMap = new HashMap<>();
            messageMap.put("req_id", UUID.randomUUID().toString());
            messageMap.put("req_time", dateTimeFormat.format(new Date()));
            // we'll put task id here as key so same task will be ignored by device if received
            // more than 1 time
            messageMap.put("task_id", task.getId());
            messageMap.put("req_type", "CHECK_TASK");
            messageMap.put("group_list", task.getGroupIds());

            String messageData = gson.toJson(messageMap);

            logger.debug("Try publish message to: {}", topicName);
            mqttClient.publish(topicName, messageData.getBytes(), qos, false);
            logger.debug("Message published!");
        } finally {
            logger.debug("Start broadcast check tasks DONE");
        }
    }
    
    private String getDownloadTimeTypeName(int type) {
        switch (type) {
            case 1:
                return "NEXT_CONTACT";
            case 2:
                return "DATETIME";
        }
        return null;
    }

    private String getInstallationTimeTypeName(int type) {
        switch (type) {
            case 1:
                return "IMMEDIATE_AFTER_DOWNLOAD";
            case 2:
                return "DATETIME";
        }
        return null;
    }

    private String getInstallationNotificationName(int type) {
        switch (type) {
            case 1:
                return "SILENT";
            case 2:
                return "NEED_CONFIRMATION";
        }
        return null;
    }

    private String getFilePath(String fileId, String fileExt, Date fileCreateDate) {
        String filePath = AppConfig.getInstance().get("file.dir") + File.separator
                + sdfYEAR.format(fileCreateDate) + File.separator
                + sdfMONTH.format(fileCreateDate) + File.separator
                + sdfDATE.format(fileCreateDate) + File.separator
                + fileId + "." + fileExt;
        return filePath;
    }

    private String md5Checksum(byte[] bytes) throws Exception {
        try {
            String checksum = DigestUtils.md5Hex(bytes);
            return checksum;
        } finally {
        }
    }*/
}
