package id.iptek.utms.agent;

import id.iptek.utms.agent.dao.TerminalDao;
import id.iptek.utms.agent.db.DBConn;
import id.iptek.utms.agent.model.TerminalIdObj;
import id.iptek.utms.agent.queue.ConsumerMode;
import id.iptek.utms.agent.queue.DelayMessageQueue;
import id.iptek.utms.agent.queue.DeviceInitQueueMessageHandler;
import id.iptek.utms.agent.queue.DiagnosticInfoQueueMessageHandler;
import id.iptek.utms.agent.queue.HeartBeatQueueMessageHandler;
import id.iptek.utms.agent.scheduler.job.DownloadTaskSchedulerJob;
import id.iptek.utms.agent.util.Singleton;
import id.iptek.utms.agent.worker.Worker;
import id.iptek.utms.agent.worker.WorkerFactory;
import io.minio.MinioClient;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.PropertyConfigurator;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import static org.quartz.JobBuilder.newJob;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.SchedulerFactory;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author jakar
 */
public class Main {

    private Logger logger = LoggerFactory.getLogger(Main.class);
    private List<Worker> workers;
    private List<DelayMessageQueue> queueHandlers;
    private ExecutorService executorService = Executors.newCachedThreadPool();
    private final CountDownLatch shutdownLatch = new CountDownLatch(1);
    private final AtomicBoolean stopping = new AtomicBoolean(false);
    private static final long QUEUE_SHUTDOWN_TIMEOUT_SECONDS = 30L;

    public void start(String[] args) {
        String configFile = getConfigFile(args);
        File config = new File(configFile);
        File configParent = config.getAbsoluteFile().getParentFile();
        String baseDir = configParent != null && configParent.getParentFile() != null ? configParent.getParentFile().getAbsolutePath() : ".";
        if (baseDir == null) {
            baseDir = ".";
        }
        String configDir = (new StringBuilder()).append(baseDir).append(File.separator).append("conf").toString();
        AppConfig appConfig = AppConfig.getInstance();
        //init config
        try {
            initConfig(appConfig, baseDir, configDir, configFile);
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
        //init logging
        try {
            initLogging(appConfig);
        } catch (Exception ex) {
            logger.debug("Error initializing Logging: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
        //init db
        try {
            initDB(appConfig);
        } catch (Exception ex) {
            logger.error("Error initializing DB: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
        //init mqtt client
        try {
            initMqttClient(appConfig);
        } catch (Exception ex) {
            logger.error("Error init Mqtt Client: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
        //init minio client
        try {
            initMinioClient(appConfig);
        } catch (Exception ex) {
            logger.error("Error init Minio Client: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
        //load data async
        try {
            loadDataAsync(appConfig);
        } catch (Exception ex) {
            logger.error("Error load data async: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
        //build message queue handlers
        try {
            buildQueueHandlers(appConfig);
        } catch (Exception ex) {
            logger.error("Error build & start queue handlers: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
        //build & start workers
        try {
            buildAndStartWorkers(appConfig);
        } catch (Exception ex) {
            logger.error("Error build & start workers: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
        //start schedulers
        try {
            startSchedulers(appConfig);
        } catch (Exception ex) {
            logger.error("Error starting schedulers: {}", ex.getMessage(), ex);
            throw new RuntimeException(ex);
        }
    }

    private String getConfigFile(String[] args) {
        if (args == null || args.length == 0) {
            throw new IllegalArgumentException("Usage: java -jar utms-agent.jar <config-file>");
        }
        if (args.length > 1 && Main.class.getName().equals(args[0])) {
            return args[1];
        }
        return args[0];
    }

    private void initConfig(AppConfig appConfig, String baseDir, String configDir, String configFile) throws Exception {
        File _configFile = new File(configFile);
        /*if(!_configFile.isAbsolute()) {
            _configFile = new File(configDir, configFile);
        }*/
        appConfig.init(_configFile);
        appConfig.set("Config.File", _configFile.getAbsolutePath());
        appConfig.set("Base.Dir", baseDir);
        appConfig.set("Config.Dir", configDir);
    }

    private void initLogging(AppConfig appConfig) throws Exception {
        String configFile = appConfig.get("log4j.file");
        String configDir = appConfig.get("Config.Dir");
        configFile = configFile.replace("${Config.Dir}", configDir);
        BasicConfigurator.resetConfiguration();
        PropertyConfigurator.configure(configFile);
        logger.info("Config file: {}", appConfig.get("Config.File"));
        logger.info("Base dir: {}", appConfig.get("Base.Dir"));
        logger.info("Config dir: {}", configDir);
        logger.info("Logging config file: {}", configFile);
        logger.info("Logging initialized.");
    }

    private void initDB(AppConfig appConfig) throws Exception {
        logger.info("Init DB ...");
        String datasource = appConfig.get("db.datasource");
        String driver = appConfig.get("db.driver");
        String url = appConfig.get("db.url");
        String user = appConfig.get("db.user");
        String password = appConfig.get("db.password");

        if (datasource != null && !"".equals(datasource.trim())) {
            logger.debug("Using datasource.");
            DBConn.getInstance().init(datasource);
        } else {
            logger.debug("Using traditional url, username, and password.");
            DBConn.getInstance().init(driver, url, user, password);
        }
        logger.info("DB initialized.");
    }

    private void initMqttClient(AppConfig appConfig) throws Exception {
        logger.info("Init Mqtt client ...");
        String clientId = appConfig.get("mqtt.clientid_prefix") + RandomStringUtils.randomAlphanumeric(10);
        String brokerUrl = appConfig.get("mqtt.broker.url");
        String username = appConfig.get("mqtt.user");
        String password = appConfig.get("mqtt.password");
        int keepAlive = appConfig.getAsInt("mqtt.keepalive", 10);
        int maxInFlight = appConfig.getAsInt("mqtt.maxinflight", 32);
        boolean cleanSession = appConfig.getAsBoolean("mqtt.cleansession", true);
        boolean autoReconnect = appConfig.getAsBoolean("mqtt.autoreconnect", true);
        
        MemoryPersistence persistence = new MemoryPersistence();
        MqttClient mqttClient = new MqttClient(brokerUrl, clientId, persistence);
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(cleanSession);
        connOpts.setUserName(username);
        connOpts.setPassword(password.toCharArray());
        connOpts.setAutomaticReconnect(autoReconnect);
        connOpts.setKeepAliveInterval(keepAlive);
        connOpts.setMaxInflight(maxInFlight);
        logger.info("Try to connect ...");
        mqttClient.connect(connOpts);
        
        // put into Singleton
        Singleton.getInstance("MQTT_CLIENT").setObject(mqttClient);
        logger.info("Mqtt client initialized.");
    }

    private void initMinioClient(AppConfig appConfig) throws Exception {
        MinioClient minioClient
                = MinioClient.builder()
                        .endpoint(appConfig.get("fileserver.endpoint"))
                        .credentials(appConfig.get("fileserver.access_key"), appConfig.get("fileserver.secret_key"))
                        .build();
        // put into Singleton
        Singleton.getInstance("MINIO_CLIENT").setObject(minioClient);
    }
    
    private void buildAndStartWorkers(AppConfig appConfig) throws Exception {
        logger.info("Build & Start workers ...");
        workers = WorkerFactory.getInstance().buildWorkers(appConfig);
        for (Worker worker : workers) {
            logger.info("Start worker .. '{}'", worker.getLoggingId());
            worker.setExecutorService(executorService);
            worker.start();
            logger.info("Worker started");
            executorService.execute(worker);
        }
        logger.info("Workers build & started.");
    }

    private void buildQueueHandlers(AppConfig appConfig) throws Exception {
        queueHandlers = new ArrayList<>();

        logger.info("Build & Start queue handlers ...");
        // heartbeat queue handler
        logger.info("Start Heartbeat queue handler .. ");
        DelayMessageQueue hbQueue = new DelayMessageQueue(executorService, new HeartBeatQueueMessageHandler(), 10);
        hbQueue.setMode(ConsumerMode.BATCH);
        hbQueue.setTotalThread(3);
        hbQueue.run();
        // put to singleton
        Singleton.getInstance(HeartBeatQueueMessageHandler.NAME).setObject(hbQueue);
        queueHandlers.add(hbQueue);
        logger.info("Heartbeat queue handler started");

        // diagnostic queue handler
        logger.info("Start Diagnostic queue handler .. ");
        DelayMessageQueue diagQueue = new DelayMessageQueue(executorService, new DiagnosticInfoQueueMessageHandler(), 3);
        diagQueue.setMode(ConsumerMode.BATCH);
        diagQueue.setTotalThread(2);
        diagQueue.run();
        // put to singleton
        Singleton.getInstance(DiagnosticInfoQueueMessageHandler.NAME).setObject(diagQueue);
        queueHandlers.add(diagQueue);
        logger.info("Diagnostic queue handler started");

        // init queue handler
        logger.info("Start Init queue handler .. ");
        DelayMessageQueue initQueue = new DelayMessageQueue(executorService, new DeviceInitQueueMessageHandler(), 3);
        initQueue.setMode(ConsumerMode.BATCH);
        initQueue.setTotalThread(2);
        initQueue.run();
        // put to singleton
        Singleton.getInstance(DeviceInitQueueMessageHandler.NAME).setObject(initQueue);
        queueHandlers.add(initQueue);
        logger.info("Init queue handler started");

        logger.info("Queue handlers build & started.");
    }
    
    private void loadDataAsync(AppConfig appConfig) throws Exception {
        logger.info("Load data async ...");
        // load terminal Id Generic Cache
        loadTerminalIdCache();
        logger.info("End of load data async.");
    }

    private void loadTerminalIdCache() throws Exception {
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    logger.debug("Load Terminal Id data ...");
                    TerminalDao terminalDao = new TerminalDao();
                    GenericCache cache = GenericCache.getCache("terminalId");
                    
                    List<TerminalIdObj> terminalList = terminalDao.getTerminalIds();
                    terminalList.forEach((t) -> {
                        cache.put(t.getSn().toLowerCase(), t.getTerminalId());
                    });
                    logger.debug("Terminal Id data loaded: {} data.", terminalList.size());
                } catch(Exception ex) {
                    logger.error("Error loading Terminal Id cache: {}", ex.getMessage(), ex);
                }
            }
        });
    }
    
    private void startSchedulers(AppConfig appConfig) throws Exception {
        logger.info("Start schedulers ...");
        // download task scheduler
        JobDetail job = newJob(DownloadTaskSchedulerJob.class)
                .withIdentity("downloadTaskScheduler", "task")
                .build();
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity("myTrigger", "task")
                .startNow()
                .withSchedule(SimpleScheduleBuilder.simpleSchedule()
                        .withIntervalInMinutes(appConfig.getAsInt("scheduler.periode.minutes", 5))
                        .repeatForever())
                .build();

        SchedulerFactory sf = new StdSchedulerFactory();
        Scheduler sched = sf.getScheduler();
        sched.scheduleJob(job, trigger);
        sched.start();
        
        Singleton.getInstance("SCHEDULER").setObject(sched);
        logger.info("Schedulers started.");
    }

    private void stopQueueHandlers() throws Exception {
        logger.info("Stop queue handlers ...");
        if (queueHandlers == null) {
            logger.info("Queue handlers have not been started.");
            return;
        }
        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(QUEUE_SHUTDOWN_TIMEOUT_SECONDS);
        for (DelayMessageQueue queue : queueHandlers) {
            try {
                logger.info("Stop queue handler .. '{}'", queue.getClass());
                long remainingNanos = deadline - System.nanoTime();
                if (remainingNanos <= 0L) {
                    logger.warn("Queue graceful shutdown timeout reached, forcing queue handler '{}'", queue.getClass());
                    queue.stop(true);
                } else {
                    boolean graceful = queue.stop(false, remainingNanos, TimeUnit.NANOSECONDS);
                    if (!graceful) {
                        logger.warn("Force queue handler .. '{}'", queue.getClass());
                        queue.stop(true);
                    }
                }
                logger.info("Queue handler stopped");
            } catch (Exception ex) {
                logger.error("Error stopping queue handler: {}", ex.getMessage(), ex);
            }
        }
        logger.info("Queue handlers stopped.");
    }

    private void stopWorkers() throws Exception {
        logger.info("Stop workers ...");
        if (workers == null) {
            logger.info("Workers have not been started.");
            return;
        }
        for (Worker worker : workers) {
            try {
                logger.info("Stop worker .. '{}'", worker.getLoggingId());
                worker.stop();
                logger.info("Worker stopped");
            } catch (Exception ex) {
                logger.error("Error stopping worker '{}': {}", worker.getLoggingId(), ex.getMessage(), ex);
            }
        }
        logger.info("Workers stopped.");
    }

    public void stop() {
        if (stopping.compareAndSet(false, true)) {
            try {
                stopScheduler();
                stopQueueHandlers();
                stopWorkers();
                stopMqttClient();
                executorService.shutdownNow();
            } catch (Exception ex) {
                logger.error("Error stopping: {}", ex.getMessage(), ex);
            } finally {
                shutdownLatch.countDown();
            }
        }
    }

    private void stopMqttClient() throws Exception {
        Object object = Singleton.getInstance("MQTT_CLIENT").getObject();
        if (object instanceof MqttClient) {
            MqttClient mqttClient = (MqttClient) object;
            if (mqttClient.isConnected()) {
                logger.info("Disconnect MQTT client ...");
                mqttClient.disconnectForcibly(0L, 250L, false);
                logger.info("MQTT client disconnected.");
            }
            mqttClient.close(true);
        }
    }

    private void stopScheduler() throws Exception {
        Object scheduler = Singleton.getInstance("SCHEDULER").getObject();
        if (scheduler instanceof Scheduler) {
            logger.info("Stop scheduler ...");
            ((Scheduler) scheduler).shutdown(false);
            logger.info("Scheduler stopped.");
        }
    }

    private void awaitShutdown() throws InterruptedException {
        shutdownLatch.await();
    }

    public static void main(String[] args) {
        final Main main = new Main();
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                main.stop();
            }
        }, "utms-agent-shutdown"));

        try {
            main.start(args);
            main.awaitShutdown();
        } catch (Exception ex) {
            main.stop();
            System.exit(1);
        }
    }
}
