package org.apache.helix.messaging.handling;

/*
 * 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.
 */

import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.helix.AccessOption;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixRollbackException;
import org.apache.helix.InstanceType;
import org.apache.helix.NotificationContext;
import org.apache.helix.NotificationContext.MapKey;
import org.apache.helix.PropertyKey;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.messaging.handling.MessageHandler.ErrorCode;
import org.apache.helix.messaging.handling.MessageHandler.ErrorType;
import org.apache.helix.model.Message;
import org.apache.helix.model.Message.Attributes;
import org.apache.helix.model.Message.MessageType;
import org.apache.helix.monitoring.StateTransitionContext;
import org.apache.helix.monitoring.StateTransitionDataPoint;
import org.apache.helix.monitoring.mbeans.ParticipantMessageMonitor;
import org.apache.helix.util.HelixUtil;
import org.apache.helix.util.StatusUpdateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HelixTask implements MessageTask {
  private static Logger logger = LoggerFactory.getLogger(HelixTask.class);
  private final Message _message;
  private final MessageHandler _handler;
  private final NotificationContext _notificationContext;
  private final HelixManager _manager;
  StatusUpdateUtil _statusUpdateUtil;
  HelixTaskExecutor _executor;
  volatile boolean _isTimeout = false;
  volatile boolean _isStarted = false;
  volatile boolean _isCancelled = false;

  public HelixTask(Message message, NotificationContext notificationContext,
      MessageHandler handler, HelixTaskExecutor executor) {
    this._notificationContext = notificationContext;
    this._message = message;
    this._handler = handler;
    this._manager = notificationContext.getManager();
    _statusUpdateUtil = new StatusUpdateUtil();
    _executor = executor;
  }

  @Override
  public HelixTaskResult call() {
    HelixTaskResult taskResult = null;

    ErrorType type = null;
    ErrorCode code = null;
    Long handlerStart = null;
    Long handlerEnd = null;

    long start = System.currentTimeMillis();
    logger.info("handling task: " + getTaskId() + " begin, at: " + start);
    _statusUpdateUtil.logInfo(_message, HelixTask.class, "Message handling task begin execute",
        _manager);
    _message.setExecuteStartTimeStamp(new Date().getTime());

    // add a concurrent map to hold currentStateUpdates for sub-messages of a batch-message
    // partitionName -> csUpdate
    if (_message.getBatchMessageMode()) {
      _notificationContext.add(MapKey.CURRENT_STATE_UPDATE.toString(),
          new ConcurrentHashMap<String, CurrentStateUpdate>());
    }

    // Handle the message
    try {
      setStarted();
      handlerStart = System.currentTimeMillis();
      taskResult = _handler.handleMessage();
      handlerEnd = System.currentTimeMillis();

      // cancel timeout task
      _executor.cancelTimeoutTask(this);
    } catch (InterruptedException e) {
      taskResult = new HelixTaskResult();
      taskResult.setException(e);
      taskResult.setInterrupted(true);

      _statusUpdateUtil.logError(_message, HelixTask.class, e,
          "State transition interrupted, timeout:" + _isTimeout, _manager);
      logger.info("Message " + _message.getMsgId() + " is interrupted");
    } catch (Exception e) {
      taskResult = new HelixTaskResult();
      taskResult.setException(e);
      taskResult.setMessage(e.getMessage());

      String errorMessage =
          "Exception while executing a message. " + e + " msgId: " + _message.getMsgId()
              + " type: " + _message.getMsgType();
      logger.error(errorMessage, e);
      _statusUpdateUtil.logError(_message, HelixTask.class, e, errorMessage, _manager);
    }

    Exception exception = null;
    try {
      if (taskResult.isSuccess()) {
        _statusUpdateUtil
            .logInfo(_message, _handler.getClass(), "Message handling task completed successfully", _manager);
        logger.info("Message " + _message.getMsgId() + " completed.");
        _executor.getParticipantMonitor().reportProcessedMessage(_message, ParticipantMessageMonitor.ProcessedMessageState.COMPLETED);
      } else {
        type = ErrorType.INTERNAL;

        if (taskResult.isInterrupted()) {
          logger.info("Message " + _message.getMsgId() + " is interrupted");
          code = _isTimeout ? ErrorCode.TIMEOUT : ErrorCode.CANCEL;
          if (_isTimeout) {
            int retryCount = _message.getRetryCount();
            logger.info("Message timeout, retry count: " + retryCount + " msgId:"
                + _message.getMsgId());
            _statusUpdateUtil.logInfo(_message, _handler.getClass(),
                "Message handling task timeout, retryCount:" + retryCount, _manager);
            // Notify the handler that timeout happens, and the number of retries left
            // In case timeout happens (time out and also interrupted)
            // we should retry the execution of the message by re-schedule it in
            if (retryCount > 0) {
              _message.setRetryCount(retryCount - 1);
              HelixTask task = new HelixTask(_message, _notificationContext, _handler, _executor);
              _executor.scheduleTask(task);
              return taskResult;
            }
          }
          _executor.getParticipantMonitor().reportProcessedMessage(
              _message, ParticipantMessageMonitor.ProcessedMessageState.DISCARDED);
        } else if (taskResult.isCancelled()) {
          type = null;
          _statusUpdateUtil
              .logInfo(_message, _handler.getClass(), "Cancellation completed successfully",
                  _manager);
          _executor.getParticipantMonitor().reportProcessedMessage(
              _message, ParticipantMessageMonitor.ProcessedMessageState.DISCARDED);
        } else {// logging for errors
          code = ErrorCode.ERROR;
          String errorMsg =
              "Message execution failed. msgId: " + getTaskId() + ", errorMsg: "
                  + taskResult.getMessage();
          logger.error(errorMsg);
          _statusUpdateUtil.logError(_message, _handler.getClass(), errorMsg, _manager);
          _executor.getParticipantMonitor().reportProcessedMessage(
              _message, ParticipantMessageMonitor.ProcessedMessageState.FAILED);
        }
      }

      HelixDataAccessor accessor = _manager.getHelixDataAccessor();
      // forward relay messages attached to this message to other participants
      if (taskResult.isSuccess()) {
        try {
          forwardRelayMessages(accessor, _message, taskResult.getCompleteTime());
        } catch (Exception e) {
          // Fail to send relay message should not result in a task execution failure
          // Currently we don't log error to ZK to reduce writes as when accessor throws
          // exception, ZK might not be in good condition.
          logger.warn("Failed to send relay messages.", e);
        }
      }

      finalCleanup(taskResult);
    } catch (Exception e) {
      finalCleanup(taskResult);
      exception = e;
      type = ErrorType.FRAMEWORK;
      code = ErrorCode.ERROR;

      String errorMessage =
          "Exception after executing a message, msgId: " + _message.getMsgId() + e;
      logger.error(errorMessage, e);
      _statusUpdateUtil.logError(_message, HelixTask.class, errorMessage, _manager);
    } finally {
      long end = System.currentTimeMillis();
      long totalDuration = end - start;
      long handlerDuration =
          handlerStart != null && handlerEnd != null ? handlerEnd - handlerStart : -1;
      logger.info(
          "Message: {} (parent: {}) handling task for {}:{} completed at: {}, results: {}. FrameworkTime: {} ms; HandlerTime: {} ms.",
          _message.getMsgId(), _message.getAttribute(Attributes.PARENT_MSG_ID), _message.getResourceName(),
          _message.getPartitionName(), end, taskResult.isSuccess(), totalDuration - handlerDuration,
          handlerDuration);

      // Notify the handler about any error happened in the handling procedure, so that
      // the handler have chance to finally cleanup
      if (type == ErrorType.INTERNAL) {
        _handler.onError(taskResult.getException(), code, type);
      } else if (type == ErrorType.FRAMEWORK) {
        _handler.onError(exception, code, type);
      }
    }

    return taskResult;
  }

  private void removeMessageFromZk(HelixDataAccessor accessor, Message message) {
    if (!HelixUtil.removeMessageFromZK(accessor, message, _manager.getInstanceName())) {
      logger.warn("Failed to delete message " + message.getId() + " from zk!");
    } else {
      logger.info("Delete message " + message.getId() + " from zk!");
    }
  }

  private void forwardRelayMessages(HelixDataAccessor accessor, Message message,
      long taskCompletionTime) {
    if (message.hasRelayMessages()) {
      Map<String, Message> relayMessages = message.getRelayMessages();
      Builder keyBuilder = accessor.keyBuilder();

      // Ignore all relay messages if participant's session has changed.
      if (!_manager.getSessionId().equals(message.getTgtSessionId())) {
        logger.info(
            "Session id has been changed, ignore all relay messages attached with " + message
                .getId());
        return;
      }

      for (String instance : relayMessages.keySet()) {
        Message msg = relayMessages.get(instance);
        if (msg.getMsgSubType().equals(MessageType.RELAYED_MESSAGE.name())) {
          msg.setRelayTime(taskCompletionTime);
          if (msg.isExpired()) {
            logger.info(
                "Relay message expired, ignore " + msg.getId() + " to instance " + instance);
            continue;
          }
          PropertyKey msgKey = keyBuilder.message(instance, msg.getId());
          boolean success = accessor.getBaseDataAccessor()
              .create(msgKey.getPath(), msg.getRecord(), AccessOption.PERSISTENT);
          if (!success) {
            logger.warn("Failed to send relay message " + msg.getId() + " to " + instance);
          } else {
            logger.info("Send relay message " + msg.getId() + " to " + instance);
          }
        }
      }
    }
  }

  private HelixDataAccessor getSrcClusterDataAccessor(final Message message) {
    HelixDataAccessor helixDataAccessor = _manager.getHelixDataAccessor();
    String clusterName = message.getSrcClusterName();
    if (clusterName != null && !clusterName.equals(_manager.getClusterName())) {
      // for cross cluster message, create different HelixDataAccessor for replying message.
      /*
        TODO On frequent cross clsuter messaging request, keeping construct data accessor may cause
        performance issue. We should consider adding cache in this class or HelixManager. --JJ
       */
      helixDataAccessor = new ZKHelixDataAccessor(clusterName, helixDataAccessor.getBaseDataAccessor());
    }
    return helixDataAccessor;
  }

  private void sendReply(HelixDataAccessor replyDataAccessor, Message message,
      HelixTaskResult taskResult) {
    if (message.getCorrelationId() != null && !message.getMsgType()
        .equals(MessageType.TASK_REPLY.name())) {
      logger.info("Sending reply for message " + message.getCorrelationId());
      _statusUpdateUtil.logInfo(message, HelixTask.class, "Sending reply", _manager);

      taskResult.getTaskResultMap().put("SUCCESS", "" + taskResult.isSuccess());
      taskResult.getTaskResultMap().put("INTERRUPTED", "" + taskResult.isInterrupted());
      if (!taskResult.isSuccess()) {
        taskResult.getTaskResultMap().put("ERRORINFO", taskResult.getMessage());
      }
      Message replyMessage = Message
          .createReplyMessage(message, _manager.getInstanceName(), taskResult.getTaskResultMap());
      replyMessage.setSrcInstanceType(_manager.getInstanceType());

      Builder keyBuilder = replyDataAccessor.keyBuilder();
      if (message.getSrcInstanceType() == InstanceType.PARTICIPANT) {
        replyDataAccessor
            .setProperty(keyBuilder.message(message.getMsgSrc(), replyMessage.getMsgId()),
                replyMessage);
      } else if (message.getSrcInstanceType() == InstanceType.CONTROLLER) {
        replyDataAccessor
            .setProperty(keyBuilder.controllerMessage(replyMessage.getMsgId()), replyMessage);
      }
      _statusUpdateUtil.logInfo(message, HelixTask.class, String
          .format("1 msg replied to %s in cluster %s.", replyMessage.getTgtName(),
              message.getSrcClusterName() == null ?
                  _manager.getClusterName() :
                  message.getSrcClusterName()), _manager);
    }
  }

  private void reportMessageStat(HelixManager manager, Message message, HelixTaskResult taskResult) {
    // report stat
    if (!message.getMsgType().equals(MessageType.STATE_TRANSITION.name())) {
      return;
    }
    long now = new Date().getTime();
    long msgReadTime = message.getReadTimeStamp();
    long msgExecutionStartTime = message.getExecuteStartTimeStamp();
    if (msgReadTime != 0 && msgExecutionStartTime != 0) {
      long totalDelay = now - msgReadTime;
      long executionDelay = now - msgExecutionStartTime;
      long msgLatency = msgReadTime - message.getCreateTimeStamp();
      if (totalDelay >= 0 && executionDelay >= 0) {
        String fromState = message.getFromState();
        String toState = message.getToState();
        String transition = fromState + "--" + toState;
        transition = transition.replaceAll("\\*", "ANY");

        StateTransitionContext cxt =
            new StateTransitionContext(manager.getClusterName(), manager.getInstanceName(),
                message.getResourceName(), transition);

        StateTransitionDataPoint data =
            new StateTransitionDataPoint(totalDelay, executionDelay, msgLatency,
                taskResult.isSuccess());
        _executor.getParticipantMonitor().reportTransitionStat(cxt, data);
      }
    } else {
      logger.warn(
          "message read time and start execution time not recorded. State transition delay time is not available, message read time {}, Execute start time {}.",
          msgReadTime, msgExecutionStartTime);
    }
  }

  @Override
  public String getTaskId() {
    return _message.getId();
  }

  @Override
  public Message getMessage() {
    return _message;
  }

  @Override
  public NotificationContext getNotificationContext() {
    return _notificationContext;
  }

  @Override
  public void onTimeout() {
    _isTimeout = true;
    _handler.onTimeout();
  }

  @Override
  public synchronized boolean cancel() {
    if (!_isStarted) {
      _isCancelled = true;
      _handler.cancel();
      return true;
    }
    return false;
  }

  private synchronized void setStarted() {
    if (_isCancelled) {
      throw new HelixRollbackException("Task has already been cancelled");
    }
    _isStarted = true;
  }

  private void finalCleanup(HelixTaskResult taskResult) {
    try {
      if (_message.getAttribute(Attributes.PARENT_MSG_ID) == null) {
        removeMessageFromZk(_manager.getHelixDataAccessor(), _message);
        reportMessageStat(_manager, _message, taskResult);
        sendReply(getSrcClusterDataAccessor(_message), _message, taskResult);
        _executor.finishTask(this);
      }
    } catch (Exception e) {
      logger.error(String.format("Error to final clean up for message : %s", _message.getId()));
    }
  }
}
