package com.dingzhuo.compute.engine.actor.indexcalc;

import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import akka.actor.UntypedAbstractActor;
import akka.cluster.sharding.ClusterSharding;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import com.dingzhuo.compute.engine.function.FunctionEngine;
import com.dingzhuo.compute.engine.message.calculation.LoadCalcIndexMessage;
import com.dingzhuo.compute.engine.message.calculation.UnloadCalcIndexMessage;
import com.dingzhuo.compute.engine.utils.ActorUtil;
import com.dingzhuo.energy.data.model.domain.IndexStorage;
import com.dingzhuo.energy.data.model.service.IIndexStorageService;
import com.greenpineyu.fel.parser.FelNode;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import scala.concurrent.duration.Duration;

/**
 * @author fanxinfu
 */
@Component("loadIndexActor")
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class LoadIndexActor extends UntypedAbstractActor {

  LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
  private final IIndexStorageService indexStorageService;
  public static final String ACTOR_NAME = "loadIndexActor";
  private Map<String, IndexStorage> loadedCalcIndex = new HashMap<>();
  private ActorSelection calculateActor;

  public LoadIndexActor(IIndexStorageService indexStorageService) {
    this.indexStorageService = indexStorageService;
  }

  @Override
  public void preStart() {
    calculateActor = getContext()
        .actorSelection(ActorUtil.getActorAddress(CalculationIndexActor.ACTOR_NAME));
    this.context().system().scheduler()
        .scheduleAtFixedRate(Duration.Zero(), Duration.create(5, TimeUnit.MINUTES), this.self(),
            Message.REFRESH, this.context().system().dispatcher(), null);
  }

  @Override
  public void onReceive(Object message) {
    if (message instanceof Message) {
      if (message == Message.REFRESH) {
        refreshIndex();
      } else {
        this.unhandled(message);
      }
    }
  }

  private void refreshIndex() {
    List<IndexStorage> indexStorages = indexStorageService.getAllCalcIndexStorage();
    List<IndexStorage> filterIndexStorageList = new ArrayList<>();
    indexStorages.forEach(indexStorage -> {
      try {
        if (StringUtils.isNotBlank(indexStorage.getCalcText())) {
          FelNode node = FunctionEngine.getInstance().parse(indexStorage.getCalcText());
          if (node != null && !node.getChildren().isEmpty()) {
            filterIndexStorageList.add(indexStorage);
          } else {
            log.error("ErrorIndex:" + indexStorage.getId() + ";" + indexStorage.getCalcText());
          }
        }
      } catch (Exception ex) {
        log.error("ErrorIndex:" + indexStorage.getId() + ";" + indexStorage.getCalcText());
      }
    });

    if (filterIndexStorageList.isEmpty()) {
      return;
    }

    Map<String, IndexStorage> newCalcIndex = filterIndexStorageList.stream()
        .collect(Collectors.toMap(IndexStorage::getId, indexStorage -> indexStorage));
    Set<String> needInstall = new HashSet<>();
    Set<String> needUninstall = new HashSet<>();

    loadedCalcIndex.forEach((id, indexStorage) -> {
      if (!newCalcIndex.containsKey(id)) {
        needUninstall.add(id);
      } else {
        Date nowUpdate = newCalcIndex.get(id).getUpdateTime();
        Date lastUpdate = indexStorage.getUpdateTime();
        if (lastUpdate != null && nowUpdate != null && lastUpdate.after(nowUpdate)) {
          needUninstall.add(id);
          needInstall.add(id);
        }
      }
    });

    newCalcIndex.forEach((id, indexStorage) -> {
      if (!loadedCalcIndex.containsKey(id)) {
        needInstall.add(id);
      }
    });

    needUninstall.forEach(id -> {
      IndexStorage indexStorage = loadedCalcIndex.get(id);
      loadedCalcIndex.remove(id);
      calculateActor
          .tell(new UnloadCalcIndexMessage(ActorUtil.buildActorId(indexStorage)), getSelf());
    });

    needInstall.forEach(id -> {
      IndexStorage indexStorage = newCalcIndex.get(id);
      loadedCalcIndex.put(id, indexStorage);
      calculateActor.tell(new LoadCalcIndexMessage(indexStorage), getSelf());
    });
  }

  public enum Message {
    /**
     * 检测指标
     */
    REFRESH
  }
}