/*
 *  Copyright 1999-2019 Seata.io Group.
 *
 *  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 io.seata.server.storage.file.session;

import io.seata.common.exception.ShouldNeverHappenException;
import io.seata.common.loader.LoadLevel;
import io.seata.common.loader.Scope;
import io.seata.common.util.CollectionUtils;
import io.seata.common.util.StringUtils;
import io.seata.config.ConfigurationFactory;
import io.seata.core.constants.ConfigurationKeys;
import io.seata.core.exception.TransactionException;
import io.seata.core.model.GlobalStatus;
import io.seata.server.session.*;
import io.seata.server.storage.file.ReloadableStore;
import io.seata.server.storage.file.TransactionWriteStore;
import io.seata.server.storage.file.store.FileTransactionStoreManager;
import io.seata.server.store.AbstractTransactionStoreManager;
import io.seata.server.store.SessionStorable;
import io.seata.server.store.TransactionStoreManager;

import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static io.seata.common.DefaultValues.DEFAULT_SERVICE_SESSION_RELOAD_READ_SIZE;


/**
 * The type File based session manager.
 *
 * @author slievrly
 */
@LoadLevel(name = "file", scope = Scope.PROTOTYPE)
public class FileSessionManager extends AbstractSessionManager implements Reloadable {

    private static final int READ_SIZE = ConfigurationFactory.getInstance().getInt(
        ConfigurationKeys.SERVICE_SESSION_RELOAD_READ_SIZE, DEFAULT_SERVICE_SESSION_RELOAD_READ_SIZE);
    /**
     * The Session map.
     */
    private Map<String, GlobalSession> sessionMap = new ConcurrentHashMap<>();

    /**
     * Instantiates a new File based session manager.
     *
     * @param name                 the name
     * @param sessionStoreFilePath the session store file path
     * @throws IOException the io exception
     */
    public FileSessionManager(String name, String sessionStoreFilePath) throws IOException {
        super(name);
        if (StringUtils.isNotBlank(sessionStoreFilePath)) {
            transactionStoreManager = new FileTransactionStoreManager(
                sessionStoreFilePath + File.separator + name, this);
        } else {
            transactionStoreManager = new AbstractTransactionStoreManager() {
                @Override
                public boolean writeSession(LogOperation logOperation, SessionStorable session) {
                    return true;
                }
            };
        }
    }

    @Override
    public void reload() {
        restoreSessions();
    }

    @Override
    public void addGlobalSession(GlobalSession session) throws TransactionException {
        CollectionUtils.computeIfAbsent(sessionMap, session.getXid(), k -> {
            try {
                super.addGlobalSession(session);
            } catch (TransactionException e) {
                LOGGER.error("addGlobalSession fail, msg: {}", e.getMessage());
            }
            return session;
        });
    }

    @Override
    public GlobalSession findGlobalSession(String xid)  {
        return sessionMap.get(xid);
    }

    @Override
    public GlobalSession findGlobalSession(String xid, boolean withBranchSessions) {
        // withBranchSessions without process in memory
        return sessionMap.get(xid);
    }

    @Override
    public void removeGlobalSession(GlobalSession session) throws TransactionException {
        if (sessionMap.remove(session.getXid()) != null) {
            super.removeGlobalSession(session);
        }
    }

    @Override
    public Collection<GlobalSession> allSessions() {
        return sessionMap.values();
    }

    @Override
    public List<GlobalSession> findGlobalSessions(SessionCondition condition) {
        List<GlobalSession> found = new ArrayList<>();

        List<GlobalStatus> globalStatuses = null;
        if (null != condition.getStatuses() && condition.getStatuses().length > 0) {
            globalStatuses = Arrays.asList(condition.getStatuses());
        }
        for (GlobalSession globalSession : sessionMap.values()) {
            if (null != condition.getOverTimeAliveMills() && condition.getOverTimeAliveMills() > 0) {
                if (System.currentTimeMillis() - globalSession.getBeginTime() <= condition.getOverTimeAliveMills()) {
                    continue;
                }
            }

            if (!StringUtils.isEmpty(condition.getXid())) {
                if (Objects.equals(condition.getXid(), globalSession.getXid())) {
                    // Only one will be found, just add and return
                    found.add(globalSession);
                    return found;
                } else {
                    continue;
                }
            }

            if (null != condition.getTransactionId() && condition.getTransactionId() > 0) {
                if (Objects.equals(condition.getTransactionId(), globalSession.getTransactionId())) {
                    // Only one will be found, just add and return
                    found.add(globalSession);
                    return found;
                } else {
                    continue;
                }
            }

            if (null != globalStatuses) {
                if (!globalStatuses.contains(globalSession.getStatus())) {
                    continue;
                }
            }

            // All test pass, add to resp
            found.add(globalSession);
        }
        return found;
    }

    @Override
    public <T> T lockAndExecute(GlobalSession globalSession, GlobalSession.LockCallable<T> lockCallable)
        throws TransactionException {
        globalSession.lock();
        try {
            return lockCallable.call();
        } finally {
            globalSession.unlock();
        }
    }

    private void restoreSessions() {
        final Set<String> removedGlobalBuffer = new HashSet<>();
        final Map<String, Map<Long, BranchSession>> unhandledBranchBuffer = new HashMap<>();

        restoreSessions(true, removedGlobalBuffer, unhandledBranchBuffer);
        restoreSessions(false, removedGlobalBuffer, unhandledBranchBuffer);

        if (!unhandledBranchBuffer.isEmpty()) {
            unhandledBranchBuffer.values().forEach(unhandledBranchSessions -> {
                unhandledBranchSessions.values().forEach(branchSession -> {
                    String xid = branchSession.getXid();
                    if (removedGlobalBuffer.contains(xid)) {
                        return;
                    }

                    long bid = branchSession.getBranchId();
                    GlobalSession found = sessionMap.get(xid);
                    if (found == null) {
                        // Ignore
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info("GlobalSession Does Not Exists For BranchSession [" + bid + "/" + xid + "]");
                        }
                    } else {
                        BranchSession existingBranch = found.getBranch(branchSession.getBranchId());
                        if (existingBranch == null) {
                            found.add(branchSession);
                        } else {
                            existingBranch.setStatus(branchSession.getStatus());
                        }
                    }
                });
            });
        }
    }

    private boolean checkSessionStatus(GlobalSession globalSession) {
        GlobalStatus globalStatus = globalSession.getStatus();
        switch (globalStatus) {
            case UnKnown:
            case Committed:
            case CommitFailed:
            case Rollbacked:
            case RollbackFailed:
            case TimeoutRollbacked:
            case TimeoutRollbackFailed:
            case RollbackRetryTimeout:
            case Finished:
                return false;
            default:
                return true;
        }
    }

    private void restoreSessions(boolean isHistory, Set<String> removedGlobalBuffer, Map<String,
        Map<Long, BranchSession>> unhandledBranchBuffer) {
        if (!(transactionStoreManager instanceof ReloadableStore)) {
            return;
        }
        while (((ReloadableStore)transactionStoreManager).hasRemaining(isHistory)) {
            List<TransactionWriteStore> stores = ((ReloadableStore)transactionStoreManager).readWriteStore(READ_SIZE,
                isHistory);
            restore(stores, removedGlobalBuffer, unhandledBranchBuffer);
        }
    }

    private void restore(List<TransactionWriteStore> stores, Set<String> removedGlobalBuffer,
        Map<String, Map<Long, BranchSession>> unhandledBranchBuffer) {
        for (TransactionWriteStore store : stores) {
            TransactionStoreManager.LogOperation logOperation = store.getOperate();
            SessionStorable sessionStorable = store.getSessionRequest();
            switch (logOperation) {
                case GLOBAL_ADD:
                case GLOBAL_UPDATE: {
                    GlobalSession globalSession = (GlobalSession)sessionStorable;
                    if (globalSession.getTransactionId() == 0) {
                        LOGGER.error(
                            "Restore globalSession from file failed, the transactionId is zero , xid:" + globalSession
                                .getXid());
                        break;
                    }
                    if (removedGlobalBuffer.contains(globalSession.getXid())) {
                        break;
                    }
                    GlobalSession foundGlobalSession = sessionMap.get(globalSession.getXid());
                    if (foundGlobalSession == null) {
                        if (this.checkSessionStatus(globalSession)) {
                            sessionMap.put(globalSession.getXid(), globalSession);
                        } else {
                            removedGlobalBuffer.add(globalSession.getXid());
                            unhandledBranchBuffer.remove(globalSession.getXid());
                        }
                    } else {
                        if (this.checkSessionStatus(globalSession)) {
                            foundGlobalSession.setStatus(globalSession.getStatus());
                        } else {
                            sessionMap.remove(globalSession.getXid());
                            removedGlobalBuffer.add(globalSession.getXid());
                            unhandledBranchBuffer.remove(globalSession.getXid());
                        }
                    }
                    break;
                }
                case GLOBAL_REMOVE: {
                    GlobalSession globalSession = (GlobalSession)sessionStorable;
                    if (globalSession.getTransactionId() == 0) {
                        LOGGER.error(
                            "Restore globalSession from file failed, the transactionId is zero , xid:" + globalSession
                                .getXid());
                        break;
                    }
                    if (removedGlobalBuffer.contains(globalSession.getXid())) {
                        break;
                    }
                    if (sessionMap.remove(globalSession.getXid()) == null) {
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info("GlobalSession To Be Removed Does Not Exists [" + globalSession.getXid() + "]");
                        }
                    }
                    removedGlobalBuffer.add(globalSession.getXid());
                    unhandledBranchBuffer.remove(globalSession.getXid());
                    break;
                }
                case BRANCH_ADD:
                case BRANCH_UPDATE: {
                    BranchSession branchSession = (BranchSession)sessionStorable;
                    if (branchSession.getTransactionId() == 0) {
                        LOGGER.error(
                            "Restore branchSession from file failed, the transactionId is zero , xid:" + branchSession
                                .getXid());
                        break;
                    }
                    if (removedGlobalBuffer.contains(branchSession.getXid())) {
                        break;
                    }
                    GlobalSession foundGlobalSession = sessionMap.get(branchSession.getXid());
                    if (foundGlobalSession == null) {
                        unhandledBranchBuffer.computeIfAbsent(branchSession.getXid(), key -> new HashMap<>())
                            .put(branchSession.getBranchId(), branchSession);
                    } else {
                        BranchSession existingBranch = foundGlobalSession.getBranch(branchSession.getBranchId());
                        if (existingBranch == null) {
                            foundGlobalSession.add(branchSession);
                        } else {
                            existingBranch.setStatus(branchSession.getStatus());
                        }
                    }
                    break;
                }
                case BRANCH_REMOVE: {
                    BranchSession branchSession = (BranchSession)sessionStorable;
                    String xid = branchSession.getXid();
                    if (removedGlobalBuffer.contains(xid)) {
                        break;
                    }
                    long bid = branchSession.getBranchId();
                    if (branchSession.getTransactionId() == 0) {
                        LOGGER.error(
                            "Restore branchSession from file failed, the transactionId is zero , xid:" + branchSession
                                .getXid());
                        break;
                    }
                    GlobalSession found = sessionMap.get(xid);
                    if (found == null) {
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info(
                                "GlobalSession To Be Updated (Remove Branch) Does Not Exists [" + bid + "/" + xid
                                    + "]");
                        }
                    } else {
                        BranchSession theBranch = found.getBranch(bid);
                        if (theBranch == null) {
                            if (LOGGER.isInfoEnabled()) {
                                LOGGER.info("BranchSession To Be Updated Does Not Exists [" + bid + "/" + xid + "]");
                            }
                        } else {
                            found.remove(theBranch);
                        }
                    }
                    break;
                }
                default:
                    throw new ShouldNeverHappenException("Unknown Operation: " + logOperation);
            }
        }

    }

    @Override
    public void destroy() {
        transactionStoreManager.shutdown();
    }

}