/*
 * Copyright (C) 2016-2022 ActionTech.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */

package com.actiontech.dble.backend.mysql.xa.recovery.impl;

import com.actiontech.dble.backend.mysql.xa.CoordinatorLogEntry;
import com.actiontech.dble.backend.mysql.xa.Deserializer;
import com.actiontech.dble.backend.mysql.xa.Serializer;
import com.actiontech.dble.backend.mysql.xa.recovery.DeserializationException;
import com.actiontech.dble.backend.mysql.xa.recovery.Repository;
import com.actiontech.dble.config.model.SystemConfig;
import com.actiontech.dble.util.KVPathUtil;
import com.actiontech.dble.util.ZKUtils;
import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by huqing.yan on 2017/6/30.
 */
public class KVStoreRepository implements Repository {
    public static final Logger LOGGER = LoggerFactory.getLogger(KVStoreRepository.class);
    private String logPath;

    public KVStoreRepository() {
        init();
    }

    @Override
    public void init() {
        logPath = KVPathUtil.XALOG + SystemConfig.getInstance().getInstanceName();
    }

    @Override
    public void put(String id, CoordinatorLogEntry coordinatorLogEntry) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void remove(String id) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CoordinatorLogEntry get(String coordinatorId) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Collection<CoordinatorLogEntry> getAllCoordinatorLogEntries(boolean first) {
        String data = null;
        try {
            if (getZkConn().checkExists().forPath(logPath) != null) {
                try {
                    byte[] raw = getZkConn().getData().forPath(logPath);
                    if (raw != null) {
                        data = new String(raw, StandardCharsets.UTF_8);
                    }
                } catch (Exception e) {
                    if (!first) {
                        LOGGER.warn("KVStoreRepository.getAllCoordinatorLogEntries error", e);
                    }
                }
            }
        } catch (Exception e2) {
            LOGGER.warn("KVStoreRepository error", e2);
        }
        if (data == null) {
            return Collections.emptyList();
        }
        Map<String, CoordinatorLogEntry> coordinatorLogEntries = new HashMap<>();
        String[] logs = data.split(Serializer.LINE_SEPARATOR);
        for (String log : logs) {
            try {
                CoordinatorLogEntry coordinatorLogEntry = Deserializer.fromJson(log);
                coordinatorLogEntries.put(coordinatorLogEntry.getId(), coordinatorLogEntry);
            } catch (DeserializationException e) {
                LOGGER.warn("Unexpected EOF - logfile not closed properly last time? ", e);
            }
        }
        return coordinatorLogEntries.values();
    }

    @Override
    public boolean writeCheckpoint(Collection<CoordinatorLogEntry> checkpointContent) {
        try {
            StringBuilder sb = new StringBuilder();
            for (CoordinatorLogEntry coordinatorLogEntry : checkpointContent) {
                sb.append(Serializer.toJson(coordinatorLogEntry));
            }
            byte[] data = sb.toString().getBytes(StandardCharsets.UTF_8);
            if (getZkConn().checkExists().forPath(logPath) == null) {
                getZkConn().create().creatingParentsIfNeeded().forPath(logPath);
            }
            getZkConn().setData().forPath(logPath, data);
            return true;
        } catch (Exception e) {
            LOGGER.warn("Failed to writeDirectly checkpoint", e);
            return false;
        }
    }

    @Override
    public void close() {
    }

    private CuratorFramework getZkConn() {
        return ZKUtils.getConnection();
    }
}
