/*
 * Copyright 2019 WeBank
 *
 * Licensed 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 com.webank.wedatasphere.linkis.udf.service.impl;

import com.google.common.collect.Iterables;
import com.webank.wedatasphere.linkis.udf.dao.UDFTreeDao;
import com.webank.wedatasphere.linkis.udf.entity.UDFInfo;
import com.webank.wedatasphere.linkis.udf.entity.UDFTree;
import com.webank.wedatasphere.linkis.udf.excepiton.UDFException;
import com.webank.wedatasphere.linkis.udf.service.UDFService;
import com.webank.wedatasphere.linkis.udf.service.UDFTreeService;
import org.apache.commons.collections.map.HashedMap;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * Created by johnnwang on 8/18/18.
 */
@Service
public class UDFTreeServiceImpl implements UDFTreeService {

    private static final Logger logger = Logger.getLogger(UDFTreeServiceImpl.class);

    @Autowired
    private UDFTreeDao udfTreeDao;

    @Autowired
    private UDFService udfService;

    Map<String, List<UDFTree>> firstFloor = new HashMap<>();

    public final  static String SYS_USER = "sys";
    public final static String BDP_USER = "bdp";
    public final static String SHARE_USER = "share";

    Map<String, String> firstFloorName = new HashedMap();

    {
        firstFloorName.put(SYS_USER, "系统函数");
        firstFloorName.put(BDP_USER, "BDAP函数");
        firstFloorName.put(SHARE_USER, "共享函数");
    }


    /**
     * 每个用户的初始化，以及第一次调用系统的初始化
     * @param userName
     * @param category
     * @return
     * @throws UDFException
     */
    @Override
    public UDFTree initTree(String userName, String category) throws UDFException {
        List<UDFTree>  childrens = new ArrayList<>();
        if(firstFloor.get(category) == null || firstFloor.get(category).size() != 3){

            List<UDFTree> root = new ArrayList<>();
            //sys
            root.add(0,getFirstFloor(SYS_USER, category));
            //bdp
            root.add(1,getFirstFloor(BDP_USER, category));
            //shared
            root.add(2,getFirstFloor(SHARE_USER, category));

            firstFloor.put(category, root);

        }
        for(int i = 0 ; i < 3; i++){
            childrens.add(firstFloor.get(category).get(i));
        }
        childrens.add(getFirstFloor(userName, category));
        UDFTree udfTree = new UDFTree();
        udfTree.setChildrens(childrens);
        return udfTree;
    }

    /**
     * Get the user's first level directory(获得用户的第一层目录)
      * @param userName
     * @param category
     * @return
     * @throws UDFException
     */
    public UDFTree getFirstFloor(String userName, String category) throws UDFException{
        logger.info(userName + " to get first Floor directory");
        Map<String, Object> params = new HashedMap();
        params.put("parent", -1L);
        params.put("userName", userName);
        params.put("category", category);
        List<UDFTree> first = udfTreeDao.getTreesByParentId(params);
        if(first == null || first.size() == 0){
            String treeName = firstFloorName.getOrDefault(userName, "个人函数");
            UDFTree udfTree = new UDFTree(null, -1L, treeName, userName, "", new Date(), new Date(), category);
            udfTreeDao.addTree(udfTree);
            return  udfTree;
        }
        if(first.size() > 1) {
            throw new UDFException("user(用户)" + userName + "There are two root directory directories(存在两个根目录目录)");
        }
        return first.get(0);
    }

    @Override
    public UDFTree addTree(UDFTree udfTree, String userName) throws UDFException {
        if (userName.equals(udfTree.getUserName())) {
            try{
                logger.info(userName + " to add directory");
                udfTreeDao.addTree(udfTree);
            } catch (Throwable e){
                if(e instanceof DuplicateKeyException){
                    throw new UDFException("Duplicate file name(文件名重复)");
                } else {
                    throw new UDFException(e.getMessage());
                }
            }

        } else {
            throw new UDFException("Current user must be consistent with the user created(当前用户必须和创建用户一致)");
        }
        return udfTree;
    }

    @Override
    public UDFTree updateTree(UDFTree udfTree, String userName) throws UDFException {
        if (userName.equals(udfTree.getUserName())) {
            if(udfTree.getId() == null){
                throw new UDFException("id Can not be empty(不能为空)");
            }
            try{
                logger.info(userName + " to update directory");
                udfTreeDao.updateTree(udfTree);
            } catch (Throwable e){
                if(e instanceof DuplicateKeyException){
                    throw new UDFException("Duplicate file name(文件名重复)");
                } else {
                    throw new UDFException(e.getMessage());
                }
            }
        } else {
            throw new UDFException("Current user must be consistent with the modified user(当前用户必须和修改用户一致)");
        }
        return udfTree;
    }

    @Override
    public Boolean deleteTree(Long id, String userName) throws UDFException {

        udfTreeDao.deleteTree(id, userName);
        return true;
    }

    /**
     * Get UDFs and subdirectories under a layer of directories(获得一层目录下面的UDF和子目录)
     * @param id
     * @param userName
     * @param type
     * @param category
     * @return
     * @throws UDFException
     */
    @Override
    public UDFTree getTreeById(Long id, String userName, String type, String category) throws UDFException {
        UDFTree udfTree = null;
        if (id == null || id < 0) {
            udfTree = initTree(userName, category);
        } else {
            udfTree = udfTreeDao.getTreeById(id, category);
            if(udfTree == null){
                return udfTree;
            }
            Map<String, Object> params = new HashedMap();
            params.put("parent", udfTree.getId());
            params.put("category", category);
            //TODO Determine if the user can list(判断用户是否可以list)
            List<UDFTree> childrens = udfTreeDao.getTreesByParentId(params);
            List<UDFInfo> udfInfos = null;
            switch (type){
                case SYS_USER: udfInfos = udfService.getUDFSByTreeIdAndUser(udfTree.getId(), type, category);break;
                case BDP_USER: udfInfos =  udfService.getUDFInfoByTreeId(udfTree.getId(), userName, category); break;
               // case "self": udfInfos =  udfService.getUDFInfoByTreeId(udfTree.getId(), userName); break;
                case SHARE_USER: udfInfos= udfService.getSharedUDFInfos(udfTree.getId(),userName, category);break;
                default: udfInfos =  udfService.getUDFInfoByTreeId(udfTree.getId(), userName, category); break;
            }
            udfTree.setChildrens(childrens);
            udfTree.setUdfInfos(udfInfos);
        }
        return udfTree;
    }

    @Override
    public UDFTree getSharedTree(String category) throws UDFException{
        Map<String, Object> params = new HashedMap();
        params.put("parent", -1L);
        params.put("userName", SHARE_USER);
        params.put("category", category);
        return Iterables.getFirst(udfTreeDao.getTreesByParentId(params), null);
    }

}
