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

import io.seata.common.exception.StoreException;
import io.seata.common.thread.NamedThreadFactory;
import io.seata.common.util.CollectionUtils;
import io.seata.server.session.BranchSession;
import io.seata.server.session.GlobalSession;
import io.seata.server.session.SessionCondition;
import io.seata.server.session.SessionManager;
import io.seata.server.storage.file.FlushDiskMode;
import io.seata.server.storage.file.ReloadableStore;
import io.seata.server.storage.file.TransactionWriteStore;
import io.seata.server.store.AbstractTransactionStoreManager;
import io.seata.server.store.SessionStorable;
import io.seata.server.store.StoreConfig;
import io.seata.server.store.TransactionStoreManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import static io.seata.core.context.RootContext.MDC_KEY_BRANCH_ID;

/**
 * The type File transaction store manager.
 *
 * @author slievrly
 */
public class FileTransactionStoreManager extends AbstractTransactionStoreManager
    implements TransactionStoreManager, ReloadableStore {
    private static final Logger LOGGER = LoggerFactory.getLogger(FileTransactionStoreManager.class);

    private static final int MAX_THREAD_WRITE = 1;

    private ExecutorService fileWriteExecutor;

    private volatile boolean stopping = false;

    private static final int MAX_SHUTDOWN_RETRY = 3;

    private static final int SHUTDOWN_CHECK_INTERVAL = 1 * 1000;

    private static final int MAX_WRITE_RETRY = 5;

    private static final String HIS_DATA_FILENAME_POSTFIX = ".1";

    private static final AtomicLong FILE_TRX_NUM = new AtomicLong(0);

    private static final AtomicLong FILE_FLUSH_NUM = new AtomicLong(0);

    private static final int MARK_SIZE = 4;

    private static final int MAX_WAIT_TIME_MILLS = 2 * 1000;

    private static final int MAX_FLUSH_TIME_MILLS = 2 * 1000;

    private static final int MAX_FLUSH_NUM = 10;

    private static final int PER_FILE_BLOCK_SIZE = 65535 * 8;

    private static final long MAX_TRX_TIMEOUT_MILLS = 30 * 60 * 1000;

    private static volatile long trxStartTimeMills = System.currentTimeMillis();

    private File currDataFile;

    private RandomAccessFile currRaf;

    private FileChannel currFileChannel;

    private long recoverCurrOffset = 0;

    private long recoverHisOffset = 0;

    private SessionManager sessionManager;

    private String currFullFileName;

    private String hisFullFileName;

    private WriteDataFileRunnable writeDataFileRunnable;

    private ReentrantLock writeSessionLock = new ReentrantLock();

    private volatile long lastModifiedTime;

    private static final int MAX_WRITE_BUFFER_SIZE = StoreConfig.getFileWriteBufferCacheSize();

    private final ByteBuffer writeBuffer = ByteBuffer.allocateDirect(MAX_WRITE_BUFFER_SIZE);

    private static final FlushDiskMode FLUSH_DISK_MODE = StoreConfig.getFlushDiskMode();

    private static final int MAX_WAIT_FOR_FLUSH_TIME_MILLS = 2 * 1000;

    private static final int MAX_WAIT_FOR_CLOSE_TIME_MILLS = 2 * 1000;

    private static final int INT_BYTE_SIZE = 4;

    /**
     * Instantiates a new File transaction store manager.
     *
     * @param fullFileName   the dir path
     * @param sessionManager the session manager
     * @throws IOException the io exception
     */
    public FileTransactionStoreManager(String fullFileName, SessionManager sessionManager) throws IOException {
        initFile(fullFileName);
        fileWriteExecutor = new ThreadPoolExecutor(MAX_THREAD_WRITE, MAX_THREAD_WRITE, Integer.MAX_VALUE,
            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),
            new NamedThreadFactory("fileTransactionStore", MAX_THREAD_WRITE, true));
        writeDataFileRunnable = new WriteDataFileRunnable();
        fileWriteExecutor.submit(writeDataFileRunnable);
        this.sessionManager = sessionManager;
    }

    private void initFile(String fullFileName) throws IOException {
        this.currFullFileName = fullFileName;
        this.hisFullFileName = fullFileName + HIS_DATA_FILENAME_POSTFIX;
        try {
            currDataFile = new File(currFullFileName);
            if (!currDataFile.exists()) {
                // create parent dir first
                if (currDataFile.getParentFile() != null && !currDataFile.getParentFile().exists()) {
                    currDataFile.getParentFile().mkdirs();
                }
                currDataFile.createNewFile();
                trxStartTimeMills = System.currentTimeMillis();
            } else {
                trxStartTimeMills = currDataFile.lastModified();
            }
            lastModifiedTime = System.currentTimeMillis();
            currRaf = new RandomAccessFile(currDataFile, "rw");
            currRaf.seek(currDataFile.length());
            currFileChannel = currRaf.getChannel();
        } catch (IOException exx) {
            LOGGER.error("init file error,{}", exx.getMessage(), exx);
            throw exx;
        }
    }

    @Override
    public boolean writeSession(LogOperation logOperation, SessionStorable session) {
        long curFileTrxNum;
        writeSessionLock.lock();
        try {
            if (!writeDataFile(new TransactionWriteStore(session, logOperation).encode())) {
                return false;
            }
            lastModifiedTime = System.currentTimeMillis();
            curFileTrxNum = FILE_TRX_NUM.incrementAndGet();
            if (curFileTrxNum % PER_FILE_BLOCK_SIZE == 0
                && (System.currentTimeMillis() - trxStartTimeMills) > MAX_TRX_TIMEOUT_MILLS) {
                return saveHistory();
            }
        } catch (Exception exx) {
            LOGGER.error("writeSession error, {}", exx.getMessage(), exx);
            return false;
        } finally {
            writeSessionLock.unlock();
        }
        flushDisk(curFileTrxNum, currFileChannel);
        return true;
    }

    private void flushDisk(long curFileNum, FileChannel currFileChannel) {

        if (FLUSH_DISK_MODE == FlushDiskMode.SYNC_MODEL) {
            SyncFlushRequest syncFlushRequest = new SyncFlushRequest(curFileNum, currFileChannel);
            writeDataFileRunnable.putRequest(syncFlushRequest);
            syncFlushRequest.waitForFlush(MAX_WAIT_FOR_FLUSH_TIME_MILLS);
        } else {
            writeDataFileRunnable.putRequest(new AsyncFlushRequest(curFileNum, currFileChannel));
        }
    }

    /**
     * get all overTimeSessionStorables
     * merge write file
     *
     * @throws IOException
     */
    private boolean saveHistory() throws IOException {
        boolean result;
        try {
            result = findTimeoutAndSave();
            CloseFileRequest request = new CloseFileRequest(currFileChannel, currRaf);
            writeDataFileRunnable.putRequest(request);
            request.waitForClose(MAX_WAIT_FOR_CLOSE_TIME_MILLS);
            Files.move(currDataFile.toPath(), new File(hisFullFileName).toPath(), StandardCopyOption.REPLACE_EXISTING);
        } catch (IOException exx) {
            LOGGER.error("save history data file error, {}", exx.getMessage(), exx);
            result = false;
        } finally {
            initFile(currFullFileName);
        }
        return result;
    }

    private boolean writeDataFrame(byte[] data) {
        if (data == null || data.length <= 0) {
            return true;
        }
        int dataLength = data.length;
        int bufferRemainingSize = writeBuffer.remaining();
        if (bufferRemainingSize <= INT_BYTE_SIZE) {
            if (!flushWriteBuffer(writeBuffer)) {
                return false;
            }
        }
        bufferRemainingSize = writeBuffer.remaining();
        if (bufferRemainingSize <= INT_BYTE_SIZE) {
            throw new IllegalStateException(
                String.format("Write buffer remaining size %d was too small", bufferRemainingSize));
        }
        writeBuffer.putInt(dataLength);
        bufferRemainingSize = writeBuffer.remaining();
        int dataPos = 0;
        while (dataPos < dataLength) {
            int dataLengthToWrite = dataLength - dataPos;
            dataLengthToWrite = Math.min(dataLengthToWrite, bufferRemainingSize);
            writeBuffer.put(data, dataPos, dataLengthToWrite);
            bufferRemainingSize = writeBuffer.remaining();
            if (bufferRemainingSize == 0) {
                if (!flushWriteBuffer(writeBuffer)) {
                    return false;
                }
                bufferRemainingSize = writeBuffer.remaining();
            }
            dataPos += dataLengthToWrite;
        }
        return true;
    }

    private boolean flushWriteBuffer(ByteBuffer writeBuffer) {
        writeBuffer.flip();
        if (!writeDataFileByBuffer(writeBuffer)) {
            return false;
        }
        writeBuffer.clear();
        return true;
    }

    private boolean findTimeoutAndSave() throws IOException {
        List<GlobalSession> globalSessionsOverMaxTimeout = sessionManager.findGlobalSessions(
            new SessionCondition(MAX_TRX_TIMEOUT_MILLS));
        if (CollectionUtils.isEmpty(globalSessionsOverMaxTimeout)) {
            return true;
        }
        for (GlobalSession globalSession : globalSessionsOverMaxTimeout) {
            TransactionWriteStore globalWriteStore = new TransactionWriteStore(globalSession, LogOperation.GLOBAL_ADD);
            byte[] data = globalWriteStore.encode();
            if (!writeDataFrame(data)) {
                return false;
            }
            List<BranchSession> branchSessIonsOverMaXTimeout = globalSession.getSortedBranches();
            if (branchSessIonsOverMaXTimeout != null) {
                for (BranchSession branchSession : branchSessIonsOverMaXTimeout) {
                    try {
                        MDC.put(MDC_KEY_BRANCH_ID, String.valueOf(branchSession.getBranchId()));
                        TransactionWriteStore branchWriteStore = new TransactionWriteStore(branchSession,
                            LogOperation.BRANCH_ADD);
                        data = branchWriteStore.encode();
                        if (!writeDataFrame(data)) {
                            return false;
                        }
                    } finally {
                        MDC.remove(MDC_KEY_BRANCH_ID);
                    }
                }
            }
        }
        if (flushWriteBuffer(writeBuffer)) {
            currFileChannel.force(false);
            return true;
        }
        return false;
    }

    @Override
    public GlobalSession readSession(String xid) {
        throw new StoreException("unsupport for read from file, xid:" + xid);
    }

    @Override
    public List<GlobalSession> readSession(SessionCondition sessionCondition) {
        throw new StoreException("unsupport for read from file");
    }

    @Override
    public void shutdown() {
        if (fileWriteExecutor != null) {
            fileWriteExecutor.shutdown();
            stopping = true;
            int retry = 0;
            while (!fileWriteExecutor.isTerminated() && retry < MAX_SHUTDOWN_RETRY) {
                ++retry;
                try {
                    Thread.sleep(SHUTDOWN_CHECK_INTERVAL);
                } catch (InterruptedException ignore) {
                }
            }
            if (retry >= MAX_SHUTDOWN_RETRY) {
                fileWriteExecutor.shutdownNow();
            }
        }
        try {
            if (currFileChannel.isOpen()) {
                currFileChannel.force(true);
            }
        } catch (IOException e) {
            LOGGER.error("fileChannel force error: {}", e.getMessage(), e);
        }
        closeFile(currRaf);
    }

    @Override
    public List<TransactionWriteStore> readWriteStore(int readSize, boolean isHistory) {
        File file = null;
        long currentOffset = 0;
        if (isHistory) {
            file = new File(hisFullFileName);
            currentOffset = recoverHisOffset;
        } else {
            file = new File(currFullFileName);
            currentOffset = recoverCurrOffset;
        }
        if (file.exists()) {
            return parseDataFile(file, readSize, currentOffset, isHistory);
        }
        return null;
    }

    @Override
    public boolean hasRemaining(boolean isHistory) {
        File file;
        RandomAccessFile raf = null;
        long currentOffset;
        if (isHistory) {
            file = new File(hisFullFileName);
            currentOffset = recoverHisOffset;
        } else {
            file = new File(currFullFileName);
            currentOffset = recoverCurrOffset;
        }
        try {
            raf = new RandomAccessFile(file, "r");
            return currentOffset < raf.length();

        } catch (IOException ignore) {
        } finally {
            closeFile(raf);
        }
        return false;
    }

    private List<TransactionWriteStore> parseDataFile(File file, int readSize, long currentOffset, boolean isHistory) {
        List<TransactionWriteStore> transactionWriteStores = new ArrayList<>(readSize);
        RandomAccessFile raf = null;
        FileChannel fileChannel = null;
        try {
            raf = new RandomAccessFile(file, "r");
            raf.seek(currentOffset);
            fileChannel = raf.getChannel();
            fileChannel.position(currentOffset);
            long size = raf.length();
            ByteBuffer buffSize = ByteBuffer.allocate(MARK_SIZE);
            while (fileChannel.position() < size) {
                try {
                    buffSize.clear();
                    int avilReadSize = fileChannel.read(buffSize);
                    if (avilReadSize != MARK_SIZE) {
                        break;
                    }
                    buffSize.flip();
                    int bodySize = buffSize.getInt();
                    byte[] byBody = new byte[bodySize];
                    ByteBuffer buffBody = ByteBuffer.wrap(byBody);
                    avilReadSize = fileChannel.read(buffBody);
                    if (avilReadSize != bodySize) {
                        break;
                    }
                    TransactionWriteStore writeStore = new TransactionWriteStore();
                    writeStore.decode(byBody);
                    transactionWriteStores.add(writeStore);
                    if (transactionWriteStores.size() == readSize) {
                        break;
                    }
                } catch (Exception ex) {
                    LOGGER.error("decode data file error:{}", ex.getMessage(), ex);
                    break;
                }
            }
            return transactionWriteStores;
        } catch (IOException exx) {
            LOGGER.error("parse data file error:{},file:{}", exx.getMessage(), file.getName(), exx);
            return null;
        } finally {
            try {
                if (fileChannel != null) {
                    if (isHistory) {
                        recoverHisOffset = fileChannel.position();
                    } else {
                        recoverCurrOffset = fileChannel.position();
                    }
                }
                closeFile(raf);
            } catch (IOException exx) {
                LOGGER.error("file close error{}", exx.getMessage(), exx);
            }
        }
    }

    private void closeFile(RandomAccessFile raf) {
        try {
            if (raf != null) {
                raf.close();
                raf = null;
            }
        } catch (IOException exx) {
            LOGGER.error("file close error,{}", exx.getMessage(), exx);
        }
    }

    private boolean writeDataFile(byte[] bs) {
        if (bs == null || bs.length >= Integer.MAX_VALUE - 3) {
            return false;
        }
        if (!writeDataFrame(bs)) {
            return false;
        }
        return flushWriteBuffer(writeBuffer);
    }

    private boolean writeDataFileByBuffer(ByteBuffer byteBuffer) {
        for (int retry = 0; retry < MAX_WRITE_RETRY; retry++) {
            try {
                while (byteBuffer.hasRemaining()) {
                    currFileChannel.write(byteBuffer);
                }
                return true;
            } catch (Exception exx) {
                LOGGER.error("write data file error:{}", exx.getMessage(), exx);
            }
        }
        LOGGER.error("write dataFile failed,retry more than :{}", MAX_WRITE_RETRY);
        return false;
    }

    interface StoreRequest {

    }

    abstract static class AbstractFlushRequest implements StoreRequest {
        private final long curFileTrxNum;

        private final FileChannel curFileChannel;

        protected AbstractFlushRequest(long curFileTrxNum, FileChannel curFileChannel) {
            this.curFileTrxNum = curFileTrxNum;
            this.curFileChannel = curFileChannel;
        }

        public long getCurFileTrxNum() {
            return curFileTrxNum;
        }

        public FileChannel getCurFileChannel() {
            return curFileChannel;
        }
    }

    class SyncFlushRequest extends AbstractFlushRequest {

        private final CountDownLatch countDownLatch = new CountDownLatch(1);

        public SyncFlushRequest(long curFileTrxNum, FileChannel curFileChannel) {
            super(curFileTrxNum, curFileChannel);
        }

        public void wakeup() {
            this.countDownLatch.countDown();
        }

        public void waitForFlush(long timeout) {
            try {
                this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                LOGGER.error("Interrupted", e);
            }
        }
    }

    class AsyncFlushRequest extends AbstractFlushRequest {

        public AsyncFlushRequest(long curFileTrxNum, FileChannel curFileChannel) {
            super(curFileTrxNum, curFileChannel);
        }

    }

    static class CloseFileRequest implements StoreRequest {
        private final CountDownLatch countDownLatch = new CountDownLatch(1);
        private FileChannel fileChannel;

        private RandomAccessFile file;

        public CloseFileRequest(FileChannel fileChannel, RandomAccessFile file) {
            this.fileChannel = fileChannel;
            this.file = file;
        }

        public FileChannel getFileChannel() {
            return fileChannel;
        }

        public RandomAccessFile getFile() {
            return file;
        }

        public void wakeup() {
            this.countDownLatch.countDown();
        }

        public void waitForClose(long timeout) {
            try {
                this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                LOGGER.error("Interrupted", e);
            }
        }
    }

    /**
     * The type Write data file runnable.
     */
    class WriteDataFileRunnable implements Runnable {

        private LinkedBlockingQueue<StoreRequest> storeRequests = new LinkedBlockingQueue<>();

        public void putRequest(final StoreRequest request) {
            storeRequests.add(request);
        }

        @Override
        public void run() {
            while (!stopping) {
                try {
                    StoreRequest storeRequest = storeRequests.poll(MAX_WAIT_TIME_MILLS, TimeUnit.MILLISECONDS);
                    handleStoreRequest(storeRequest);
                } catch (Exception exx) {
                    LOGGER.error("write file error: {}", exx.getMessage(), exx);
                }
            }
            handleRestRequest();
        }

        /**
         * handle the rest requests when stopping is true
         */
        private void handleRestRequest() {
            int remainNums = storeRequests.size();
            for (int i = 0; i < remainNums; i++) {
                handleStoreRequest(storeRequests.poll());
            }
        }

        private void handleStoreRequest(StoreRequest storeRequest) {
            if (storeRequest == null) {
                flushOnCondition(currFileChannel);
            }
            if (storeRequest instanceof SyncFlushRequest) {
                syncFlush((SyncFlushRequest)storeRequest);
            } else if (storeRequest instanceof AsyncFlushRequest) {
                async((AsyncFlushRequest)storeRequest);
            } else if (storeRequest instanceof CloseFileRequest) {
                closeAndFlush((CloseFileRequest)storeRequest);
            }
        }

        private void closeAndFlush(CloseFileRequest req) {
            long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get();
            flush(req.getFileChannel());
            FILE_FLUSH_NUM.addAndGet(diff);
            closeFile(req.getFile());
            req.wakeup();
        }

        private void async(AsyncFlushRequest req) {
            flushOnCondition(req.getCurFileChannel());
        }

        private void syncFlush(SyncFlushRequest req) {
            if (req.getCurFileTrxNum() > FILE_FLUSH_NUM.get()) {
                long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get();
                flush(req.getCurFileChannel());
                FILE_FLUSH_NUM.addAndGet(diff);
            }
            // notify
            req.wakeup();
        }

        private void flushOnCondition(FileChannel fileChannel) {
            if (FLUSH_DISK_MODE == FlushDiskMode.SYNC_MODEL) {
                return;
            }
            long diff = FILE_TRX_NUM.get() - FILE_FLUSH_NUM.get();
            if (diff == 0) {
                return;
            }
            if (diff % MAX_FLUSH_NUM == 0 || System.currentTimeMillis() - lastModifiedTime > MAX_FLUSH_TIME_MILLS) {
                flush(fileChannel);
                FILE_FLUSH_NUM.addAndGet(diff);
            }
        }

        private void flush(FileChannel fileChannel) {
            try {
                fileChannel.force(false);
            } catch (IOException exx) {
                LOGGER.error("flush error: {}", exx.getMessage(), exx);
            }
        }
    }
}