package com.alibaba.otter.canal.parse.inbound.mysql.rds;

import java.io.File;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.springframework.util.Assert;

import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.exception.PositionNotFoundException;
import com.alibaba.otter.canal.parse.exception.ServerIdNotMatchException;
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
import com.alibaba.otter.canal.parse.inbound.mysql.LocalBinLogConnection;
import com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.protocol.position.LogPosition;

/**
 * 基于rds binlog备份文件的复制
 *
 * @author agapple 2017年10月15日 下午1:27:36
 * @since 1.0.25
 */
public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements CanalEventParser, LocalBinLogConnection.FileParserListener {

    private String              url;                // openapi地址
    private String              accesskey;          // 云账号的ak
    private String              secretkey;          // 云账号sk
    private String              instanceId;         // rds实例id
    private Long                startTime;
    private Long                endTime;
    private BinlogDownloadQueue binlogDownloadQueue;
    private ParseFinishListener finishListener;
    private int                 batchFileSize;

    public RdsLocalBinlogEventParser(){
    }

    public void start() throws CanalParseException {
        try {
            Assert.notNull(accesskey);
            Assert.notNull(secretkey);
            Assert.notNull(instanceId);
            Assert.notNull(url);
            Assert.notNull(directory);

            if (endTime == null) {
                endTime = System.currentTimeMillis();
            }

            EntryPosition entryPosition = findStartPosition(null);
            if (entryPosition == null) {
                throw new PositionNotFoundException("position not found!");
            }
            Long startTimeInMill = entryPosition.getTimestamp();
            if (startTimeInMill == null || startTimeInMill <= 0) {
                throw new PositionNotFoundException("position timestamp is empty!");
            }

            startTime = startTimeInMill;
            List<BinlogFile> binlogFiles = RdsBinlogOpenApi.listBinlogFiles(url,
                accesskey,
                secretkey,
                instanceId,
                new Date(startTime),
                new Date(endTime));
            if (binlogFiles.isEmpty()) {
                throw new CanalParseException("start timestamp : " + startTimeInMill + " binlog files is empty");
            }

            binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchFileSize, directory);
            binlogDownloadQueue.silenceDownload();
            needWait = true;
            // try to download one file,use to test server id
            binlogDownloadQueue.tryOne();
        } catch (Throwable e) {
            logger.error("download binlog failed", e);
            throw new CanalParseException(e);
        }
        setParserExceptionHandler(this::handleMysqlParserException);
        super.start();
    }

    private void handleMysqlParserException(Throwable throwable) {
        if (throwable instanceof ServerIdNotMatchException) {
            logger.error("server id not match, try download another rds binlog!");
            binlogDownloadQueue.notifyNotMatch();
            try {
                binlogDownloadQueue.cleanDir();
                binlogDownloadQueue.tryOne();
                binlogDownloadQueue.prepare();
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }

            try {
                binlogDownloadQueue.execute(() -> {
                    RdsLocalBinlogEventParser.super.stop();
                    RdsLocalBinlogEventParser.super.start();
                });
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

        }
    }

    @Override
    protected ErosaConnection buildErosaConnection() {
        ErosaConnection connection = super.buildErosaConnection();
        if (connection instanceof LocalBinLogConnection) {
            LocalBinLogConnection localBinLogConnection = (LocalBinLogConnection) connection;
            localBinLogConnection.setNeedWait(true);
            localBinLogConnection.setServerId(serverId);
            localBinLogConnection.setParserListener(this);
            localBinLogConnection.setRdsOssMode(true);
        }
        return connection;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        if (StringUtils.isNotEmpty(url)) {
            this.url = url;
        }
    }

    public void setAccesskey(String accesskey) {
        this.accesskey = accesskey;
    }

    public void setSecretkey(String secretkey) {
        this.secretkey = secretkey;
    }

    public void setInstanceId(String instanceId) {
        this.instanceId = instanceId;
    }

    public void setStartTime(Long startTime) {
        this.startTime = startTime;
    }

    public void setEndTime(Long endTime) {
        this.endTime = endTime;
    }

    @Override
    public void onFinish(String fileName) {
        try {
            binlogDownloadQueue.downOne();
            File needDeleteFile = new File(directory + File.separator + fileName);
            if (needDeleteFile.exists()) {
                needDeleteFile.delete();
            }
            // 处理下logManager位点问题
            LogPosition logPosition = logPositionManager.getLatestIndexBy(destination);
            Long timestamp = 0L;
            if (logPosition != null && logPosition.getPostion() != null) {
                timestamp = logPosition.getPostion().getTimestamp();
                EntryPosition position = logPosition.getPostion();
                LogPosition newLogPosition = new LogPosition();
                String journalName = position.getJournalName();
                int sepIdx = journalName.indexOf(".");
                String fileIndex = journalName.substring(sepIdx + 1);
                int index = NumberUtils.toInt(fileIndex) + 1;
                String newJournalName = journalName.substring(0, sepIdx) + "."
                                        + StringUtils.leftPad(String.valueOf(index), fileIndex.length(), "0");
                newLogPosition.setPostion(new EntryPosition(newJournalName,
                    4L,
                    position.getTimestamp(),
                    position.getServerId()));
                newLogPosition.setIdentity(logPosition.getIdentity());
                logPositionManager.persistLogPosition(destination, newLogPosition);
            }

            if (binlogDownloadQueue.isLastFile(fileName)) {
                logger.warn("last file : " + fileName + " , timestamp : " + timestamp
                            + " , all file parse complete, switch to mysql parser!");
                finishListener.onFinish();
                return;
            } else {
                logger.warn("parse local binlog file : " + fileName + " , timestamp : " + timestamp
                            + " , try the next binlog !");
            }
            binlogDownloadQueue.prepare();
        } catch (Exception e) {
            logger.error("prepare download binlog file failed!", e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public void stop() {
        this.binlogDownloadQueue.release();
        super.stop();
    }

    public void setFinishListener(ParseFinishListener finishListener) {
        this.finishListener = finishListener;
    }

    public interface ParseFinishListener {

        void onFinish();
    }

    public void setBatchFileSize(int batchFileSize) {
        this.batchFileSize = batchFileSize;
    }
}
