/*
 * Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. 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.wso2.carbon.stream.processor.core.internal;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.carbon.stream.processor.core.ha.HACoordinationRecordTableHandler;
import org.wso2.carbon.stream.processor.core.ha.HACoordinationSinkHandler;
import org.wso2.carbon.stream.processor.core.ha.HACoordinationSourceHandler;
import org.wso2.carbon.stream.processor.core.ha.HAManager;
import org.wso2.carbon.stream.processor.core.ha.RetryRecordTableConnection;
import org.wso2.carbon.stream.processor.core.ha.exception.HAModeException;
import org.wso2.carbon.stream.processor.core.ha.util.CompressionUtil;
import org.wso2.carbon.stream.processor.core.internal.exception.SiddhiAppAlreadyExistException;
import org.wso2.carbon.stream.processor.core.internal.exception.SiddhiAppConfigurationException;
import org.wso2.carbon.stream.processor.core.internal.exception.SiddhiAppDeploymentException;
import org.wso2.carbon.stream.processor.core.internal.util.SiddhiAppFilesystemInvoker;
import org.wso2.carbon.stream.processor.core.internal.util.SiddhiAppProcessorConstants;
import org.wso2.carbon.stream.processor.core.model.HAStateSyncObject;
import org.wso2.siddhi.core.SiddhiAppRuntime;
import org.wso2.siddhi.core.SiddhiManager;
import org.wso2.siddhi.core.exception.ConnectionUnavailableException;
import org.wso2.siddhi.core.stream.input.InputHandler;
import org.wso2.siddhi.core.stream.input.source.Source;
import org.wso2.siddhi.core.stream.output.sink.Sink;
import org.wso2.siddhi.core.table.Table;
import org.wso2.siddhi.core.util.transport.BackoffRetryCounter;
import org.wso2.siddhi.query.api.SiddhiApp;
import org.wso2.siddhi.query.api.annotation.Element;
import org.wso2.siddhi.query.api.util.AnnotationHelper;
import org.wso2.siddhi.query.compiler.SiddhiCompiler;

import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


/**
 * Class which manage Siddhi Apps.
 */
public class StreamProcessorService {

    private Map<String, SiddhiAppData> siddhiAppMap = new ConcurrentHashMap<>();
    private BackoffRetryCounter backoffRetryCounter = new BackoffRetryCounter();
    private static final Logger log = LoggerFactory.getLogger(StreamProcessorService.class);

    public void deploySiddhiApp(String siddhiAppContent, String siddhiAppName) throws SiddhiAppConfigurationException,
            SiddhiAppAlreadyExistException, ConnectionUnavailableException {

        SiddhiAppData siddhiAppData = new SiddhiAppData(siddhiAppContent);

        if (siddhiAppMap.containsKey(siddhiAppName)) {
            throw new SiddhiAppAlreadyExistException("There is a Siddhi App with name " + siddhiAppName +
                    " is already exist");
        }

        SiddhiManager siddhiManager = StreamProcessorDataHolder.getSiddhiManager();
        SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(siddhiAppContent);

        if (siddhiAppRuntime != null) {
            Set<String> streamNames = siddhiAppRuntime.getStreamDefinitionMap().keySet();
            Map<String, InputHandler> inputHandlerMap =
                    new ConcurrentHashMap<String, InputHandler>(streamNames.size());
            for (String streamName : streamNames) {
                inputHandlerMap.put(streamName, siddhiAppRuntime.getInputHandler(streamName));
            }

            HAManager haManager = StreamProcessorDataHolder.getHAManager();

            if (haManager != null) {
                if (haManager.isActiveNode()) {
                    //Active Node
                    if (StreamProcessorDataHolder.isPersistenceEnabled()) {
                        log.info("Periodic Persistence of Active Node Enabled. Restoring From Last Saved Snapshot " +
                                "for " + siddhiAppName);
                        String revision = siddhiAppRuntime.restoreLastRevision();
                        if (revision != null) {
                            log.info("Siddhi App " + siddhiAppName + " restored to revision " + revision);
                        }
                    } else {
                        log.info("Periodic Persistence is Disabled. It is recommended to enable this feature when " +
                                "using 2 Node Minimum HA");
                    }

                    log.info("Setting SinksHandlers of " + siddhiAppName + " to Active");
                    Collection<List<Sink>> sinkCollection = siddhiAppRuntime.getSinks();
                    for (List<Sink> sinks : sinkCollection) {
                        for (Sink sink : sinks) {
                            ((HACoordinationSinkHandler) sink.getHandler()).setAsActive();
                        }
                    }

                    log.info("Setting SourceHandlers of " + siddhiAppName + " to Active");
                    Collection<List<Source>> sourceCollection = siddhiAppRuntime.getSources();
                    for (List<Source> sources : sourceCollection) {
                        for (Source source : sources) {
                            ((HACoordinationSourceHandler) source.getMapper().getHandler()).setAsActive();
                        }
                    }
                    log.info("Setting RecordTableHandlers of " + siddhiAppName + " to Active");
                    Collection<Table> tables = siddhiAppRuntime.getTables();
                    for (Table table : tables) {
                        HACoordinationRecordTableHandler recordTableHandler = (HACoordinationRecordTableHandler)
                                table.getHandler();
                        try {
                            recordTableHandler.setAsActive();
                        } catch (ConnectionUnavailableException e) {
                            backoffRetryCounter.reset();
                            log.error("HA Deployment: Error in connecting to table " + recordTableHandler.getTableId()
                                    + " while changing from passive state to active, will retry in "
                                    + backoffRetryCounter.getTimeInterval(), e);
                            ScheduledExecutorService scheduledExecutorService = Executors.
                                    newSingleThreadScheduledExecutor();
                            backoffRetryCounter.increment();
                            scheduledExecutorService.schedule(new RetryRecordTableConnection(backoffRetryCounter,
                                            table.getHandler(), scheduledExecutorService),
                                    backoffRetryCounter.getTimeIntervalMillis(), TimeUnit.MILLISECONDS);
                        }
                    }

                } else {
                    //Passive Node
                    if(haManager.isLiveStateSyncEnabled()) {
                        //Live State Sync Enabled of Passive Node
                        log.info("Live State Sync is Enabled for Passive Node. Restoring Active Node current state " +
                                "for " + siddhiAppName);
                        HAStateSyncObject haStateSyncObject = StreamProcessorDataHolder.getHAManager().
                                getActiveNodeSiddhiAppSnapshot(siddhiAppName);
                        if (haStateSyncObject.hasState()) {
                            byte[] snapshot = haStateSyncObject.getSnapshotMap().get(siddhiAppName);

                            if (log.isDebugEnabled()) {
                                log.debug("Snapshot for " + siddhiAppName + " found from Active Node Live State " +
                                        "Sync before deploying app");
                            }
                            try {
                                byte[] decompressGZIP = CompressionUtil.decompressGZIP(snapshot);
                                siddhiAppRuntime.restore(decompressGZIP);
                            } catch (IOException e) {
                                log.error("Error Decompressing Bytes " + e.getMessage(), e);
                            }
                        } else {

                            int gracePeriod = StreamProcessorDataHolder.getDeploymentConfig().getRetryAppSyncPeriod();

                            siddhiAppData.setActive(false);
                            siddhiAppData.setSiddhiAppRuntime(siddhiAppRuntime);
                            siddhiAppData.setInputHandlerMap(inputHandlerMap);
                            siddhiAppMap.put(siddhiAppName, siddhiAppData);

                            Timer timer = retrySiddhiAppLiveStateSync(gracePeriod, siddhiAppName, siddhiAppData,
                                    siddhiAppRuntime);
                            log.info("Snapshot for " + siddhiAppName + " not found. Make sure Active and Passive" +
                                    " node have deployed the same Siddhi Applications");
                            log.info("Scheduled active node state sync for " + siddhiAppName + " in " +
                                    gracePeriod/1000 + " seconds.");
                            haManager.addRetrySiddhiAppSyncTimer(timer);

                            return; // Siddhi application set to inactive state
                        }
                    } else {
                        //Live State Sync Disabled of Passive Node
                        if (StreamProcessorDataHolder.isPersistenceEnabled()) {
                            log.info("Live State Sync is Disabled for Passive Node. Restoring Active nodes last " +
                                    "persisted state for " + siddhiAppName);
                            String revision = siddhiAppRuntime.restoreLastRevision();
                            if (revision != null) {
                                log.info("Siddhi App " + siddhiAppName + " restored to revision " + revision);
                            } else {
                                int gracePeriod = StreamProcessorDataHolder.getDeploymentConfig().
                                        getRetryAppSyncPeriod();

                                siddhiAppData.setActive(false);
                                siddhiAppData.setSiddhiAppRuntime(siddhiAppRuntime);
                                siddhiAppData.setInputHandlerMap(inputHandlerMap);
                                siddhiAppMap.put(siddhiAppName, siddhiAppData);

                                Timer timer = retrySiddhiAppPersistenceStateSync(gracePeriod, siddhiAppName,
                                        siddhiAppData, siddhiAppRuntime);
                                log.info("Snapshot for " + siddhiAppName + " not found. Make sure Active and Passive" +
                                        " node have deployed the same Siddhi Applications");
                                log.info("Scheduled active node persistence sync for " + siddhiAppName + " in " +
                                        gracePeriod/1000 + " seconds.");
                                haManager.addRetrySiddhiAppSyncTimer(timer);
                                return; //Defer the app deployment until state is synced from Active Node
                            }
                        } else {
                            throw new HAModeException("Passive Node Periodic Persistence is Disabled and Live State " +
                                    "Sync Enabled. Please enable Periodic Persistence.");
                        }
                    }
                }
            } else {
                if (StreamProcessorDataHolder.isPersistenceEnabled()) {
                    log.info("Periodic State persistence enabled. Restoring last persisted state of " + siddhiAppName);
                    String revision = siddhiAppRuntime.restoreLastRevision();
                    if (revision != null) {
                        log.info("Siddhi App " + siddhiAppName + " restored to revision " + revision);
                    }
                }
            }

            siddhiAppRuntime.start();
            log.info("Siddhi App " + siddhiAppName + " deployed successfully");
            siddhiAppData.setActive(true);
            siddhiAppData.setSiddhiAppRuntime(siddhiAppRuntime);
            siddhiAppData.setInputHandlerMap(inputHandlerMap);
            siddhiAppMap.put(siddhiAppName, siddhiAppData);
        }
    }

    public void undeploySiddhiApp(String siddhiAppName) {

        if (siddhiAppMap.containsKey(siddhiAppName)) {
            SiddhiAppData siddhiAppData = siddhiAppMap.get(siddhiAppName);
            if (siddhiAppData != null) {
                if (siddhiAppData.isActive()) {
                    siddhiAppData.getSiddhiAppRuntime().shutdown();
                }
            }
            siddhiAppMap.remove(siddhiAppName);
            log.info("Siddhi App File " + siddhiAppName + " undeployed successfully.");
        }
    }

    public boolean delete(String siddhiAppName) throws SiddhiAppConfigurationException,
            SiddhiAppDeploymentException {

        if (siddhiAppMap.containsKey(siddhiAppName)) {
            SiddhiAppFilesystemInvoker.delete(siddhiAppName);
            return true;
        }
        return false;
    }

    public String validateAndSave(String siddhiApp, boolean isUpdate) throws SiddhiAppConfigurationException,
            SiddhiAppDeploymentException {

        String siddhiAppName = "";
        try {
            siddhiAppName = getSiddhiAppName(siddhiApp);
            if (isUpdate || !siddhiAppMap.containsKey(siddhiAppName)) {
                SiddhiManager siddhiManager = StreamProcessorDataHolder.getSiddhiManager();
                SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(siddhiApp);
                if (siddhiAppRuntime != null) {
                    SiddhiAppFilesystemInvoker.save(siddhiApp, siddhiAppName);
                    return siddhiAppName;
                }
            }
        } catch (SiddhiAppDeploymentException e) {
            log.error("Exception occurred when saving Siddhi App : " + siddhiAppName, e);
            throw e;
        } catch (Throwable e) {
            log.error("Exception occurred when validating Siddhi App " + siddhiAppName, e);
            throw new SiddhiAppConfigurationException(e);
        }
        return null;
    }

    public String getSiddhiAppName(String siddhiApp) throws SiddhiAppConfigurationException {
        try {
            SiddhiApp parsedSiddhiApp = SiddhiCompiler.parse(siddhiApp);
            Element nameAnnotation = AnnotationHelper.
                    getAnnotationElement(SiddhiAppProcessorConstants.ANNOTATION_NAME_NAME,
                            null, parsedSiddhiApp.getAnnotations());

            if (nameAnnotation == null || nameAnnotation.getValue().isEmpty()) {
                throw new SiddhiAppConfigurationException("Siddhi App name must " +
                        "be provided as @App:name('name').");
            }

            return nameAnnotation.getValue();

        } catch (Throwable e) {
            throw new SiddhiAppConfigurationException("Exception occurred when retrieving Siddhi App Name ", e);
        }
    }

    public boolean isExists(String siddhiApp) throws SiddhiAppConfigurationException {
        return siddhiAppMap.containsKey(getSiddhiAppName(siddhiApp));
    }

    public void addSiddhiAppFile(String siddhiAppName, SiddhiAppData siddhiAppData) {
        siddhiAppMap.put(siddhiAppName, siddhiAppData);
    }

    public Map<String, SiddhiAppData> getSiddhiAppMap() {
        return siddhiAppMap;
    }

    /**
     * When passive node gets live state sync from active node, schedule a retry for state if active node doesn't
     * send a snapshot at startup.
     *
     * @param gracePeriod time after which passive node retry to get state
     * @param siddhiAppName name of Siddhi application whose state will be requested
     * @param siddhiAppData data of relevant siddhi application
     * @param siddhiAppRuntime runtime of relevant siddhi application
     * @return reference to the timer that is started
     */
    private Timer retrySiddhiAppLiveStateSync(int gracePeriod, String siddhiAppName, SiddhiAppData siddhiAppData,
                                              SiddhiAppRuntime siddhiAppRuntime) {
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {

                HAStateSyncObject haStateSyncObject = StreamProcessorDataHolder.getHAManager().
                        getActiveNodeSiddhiAppSnapshot(siddhiAppName);
                if (haStateSyncObject.hasState()) {
                    byte[] snapshot = haStateSyncObject.getSnapshotMap().get(siddhiAppName);
                    try {
                        byte[] decompressGZIP = CompressionUtil.decompressGZIP(snapshot);
                        siddhiAppRuntime.restore(decompressGZIP);
                        siddhiAppRuntime.start();
                        siddhiAppData.setActive(true);
                        siddhiAppData.setSiddhiAppRuntime(siddhiAppRuntime);
                        siddhiAppMap.put(siddhiAppName, siddhiAppData);
                        log.info("Siddhi App " + siddhiAppName + " deployed successfully after active node sync in "
                                + gracePeriod/1000 + " seconds");
                    } catch (IOException e) {
                        log.error("Error Decompressing Bytes. " + siddhiAppName + " not deployed", e);
                    }
                } else {
                    log.error("Snapshot for " + siddhiAppName + " not found after " + gracePeriod/1000 + " seconds." +
                            " Make sure Active and Passive node have deployed the same Siddhi Applications");
                }
            }
        }, gracePeriod);

        return timer;
    }

    /**
     * When passive node gets state sync from active node persistence store, schedule a retry for state if
     * active node hasn't already persisted a state.
     *
     * @param gracePeriod time after which passive node retry to get state
     * @param siddhiAppName name of Siddhi application whose state will be requested
     * @param siddhiAppData data of relevant siddhi application
     * @param siddhiAppRuntime runtime of relevant siddhi application
     * @return reference to the timer that is started
     */
    private Timer retrySiddhiAppPersistenceStateSync(int gracePeriod, String siddhiAppName,
                                                     SiddhiAppData siddhiAppData, SiddhiAppRuntime siddhiAppRuntime) {

        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                String revision = siddhiAppRuntime.restoreLastRevision();
                if (revision != null) {
                    log.info("Siddhi App " + siddhiAppName + " restored to revision " + revision);
                    siddhiAppRuntime.start();
                    siddhiAppData.setActive(true);
                    siddhiAppData.setSiddhiAppRuntime(siddhiAppRuntime);
                    siddhiAppMap.put(siddhiAppName, siddhiAppData);
                    log.info("Siddhi App " + siddhiAppName + " deployed successfully after active node sync in "
                            + gracePeriod/1000 + " seconds");
                } else {
                    log.error("Snapshot for " + siddhiAppName + " not found after " + gracePeriod/1000 + " seconds." +
                            " Make sure Active and Passive node have deployed the same Siddhi Applications");
                }
            }
        }, gracePeriod);

        return timer;
    }
}
