/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * 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.alibaba.nacos.core.distributed.raft;

import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.utils.ExceptionUtil;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.alibaba.nacos.common.utils.LoggerUtils;
import com.alibaba.nacos.consistency.RequestProcessor;
import com.alibaba.nacos.consistency.ProtoMessageUtil;
import com.alibaba.nacos.consistency.cp.RequestProcessor4CP;
import com.alibaba.nacos.consistency.entity.ReadRequest;
import com.alibaba.nacos.consistency.entity.Response;
import com.alibaba.nacos.consistency.entity.WriteRequest;
import com.alibaba.nacos.consistency.exception.ConsistencyException;
import com.alibaba.nacos.consistency.snapshot.LocalFileMeta;
import com.alibaba.nacos.consistency.snapshot.Reader;
import com.alibaba.nacos.consistency.snapshot.SnapshotOperation;
import com.alibaba.nacos.consistency.snapshot.Writer;
import com.alibaba.nacos.core.distributed.raft.utils.JRaftUtils;
import com.alibaba.nacos.core.utils.Loggers;
import com.alipay.sofa.jraft.Closure;
import com.alipay.sofa.jraft.Iterator;
import com.alipay.sofa.jraft.Node;
import com.alipay.sofa.jraft.RouteTable;
import com.alipay.sofa.jraft.Status;
import com.alipay.sofa.jraft.conf.Configuration;
import com.alipay.sofa.jraft.core.StateMachineAdapter;
import com.alipay.sofa.jraft.entity.LeaderChangeContext;
import com.alipay.sofa.jraft.entity.LocalFileMetaOutter;
import com.alipay.sofa.jraft.error.RaftError;
import com.alipay.sofa.jraft.error.RaftException;
import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader;
import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter;
import com.google.protobuf.Message;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;

/**
 * JRaft StateMachine implemented.
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 */
class NacosStateMachine extends StateMachineAdapter {
    
    protected final JRaftServer server;
    
    protected final RequestProcessor processor;
    
    private final AtomicBoolean isLeader = new AtomicBoolean(false);
    
    private final String groupId;
    
    private Collection<JSnapshotOperation> operations;
    
    private Node node;
    
    private volatile long term = -1;
    
    private volatile String leaderIp = "unknown";
    
    NacosStateMachine(JRaftServer server, RequestProcessor4CP processor) {
        this.server = server;
        this.processor = processor;
        this.groupId = processor.group();
        adapterToJraftSnapshot(processor.loadSnapshotOperate());
    }
    
    @Override
    public void onApply(Iterator iter) {
        int index = 0;
        int applied = 0;
        Message message;
        NacosClosure closure = null;
        try {
            while (iter.hasNext()) {
                Status status = Status.OK();
                try {
                    if (iter.done() != null) {
                        closure = (NacosClosure) iter.done();
                        message = closure.getMessage();
                    } else {
                        final ByteBuffer data = iter.getData();
                        message = ProtoMessageUtil.parse(data.array());
                        if (message instanceof ReadRequest) {
                            //'iter.done() == null' means current node is follower, ignore read operation
                            applied++;
                            index++;
                            iter.next();
                            continue;
                        }
                    }
                    
                    LoggerUtils.printIfDebugEnabled(Loggers.RAFT, "receive log : {}", message);
                    
                    if (message instanceof WriteRequest) {
                        Response response = processor.onApply((WriteRequest) message);
                        postProcessor(response, closure);
                    }
                    
                    if (message instanceof ReadRequest) {
                        Response response = processor.onRequest((ReadRequest) message);
                        postProcessor(response, closure);
                    }
                } catch (Throwable e) {
                    index++;
                    status.setError(RaftError.UNKNOWN, e.toString());
                    Optional.ofNullable(closure).ifPresent(closure1 -> closure1.setThrowable(e));
                    throw e;
                } finally {
                    Optional.ofNullable(closure).ifPresent(closure1 -> closure1.run(status));
                }
                
                applied++;
                index++;
                iter.next();
            }
        } catch (Throwable t) {
            Loggers.RAFT.error("processor : {}, stateMachine meet critical error: {}.", processor, t);
            iter.setErrorAndRollback(index - applied,
                    new Status(RaftError.ESTATEMACHINE, "StateMachine meet critical error: %s.",
                            ExceptionUtil.getStackTrace(t)));
        }
    }
    
    public void setNode(Node node) {
        this.node = node;
    }
    
    @Override
    public void onSnapshotSave(SnapshotWriter writer, Closure done) {
        for (JSnapshotOperation operation : operations) {
            try {
                operation.onSnapshotSave(writer, done);
            } catch (Throwable t) {
                Loggers.RAFT.error("There was an error saving the snapshot , error : {}, operation : {}", t,
                        operation.info());
                throw t;
            }
        }
    }
    
    @Override
    public boolean onSnapshotLoad(SnapshotReader reader) {
        for (JSnapshotOperation operation : operations) {
            try {
                if (!operation.onSnapshotLoad(reader)) {
                    Loggers.RAFT.error("Snapshot load failed on : {}", operation.info());
                    return false;
                }
            } catch (Throwable t) {
                Loggers.RAFT.error("Snapshot load failed on : {}, has error : {}", operation.info(), t);
                return false;
            }
        }
        return true;
    }
    
    @Override
    public void onLeaderStart(final long term) {
        super.onLeaderStart(term);
        this.term = term;
        this.isLeader.set(true);
        this.leaderIp = node.getNodeId().getPeerId().getEndpoint().toString();
        NotifyCenter.publishEvent(
                RaftEvent.builder().groupId(groupId).leader(leaderIp).term(term).raftClusterInfo(allPeers()).build());
    }
    
    @Override
    public void onLeaderStop(final Status status) {
        super.onLeaderStop(status);
        this.isLeader.set(false);
    }
    
    @Override
    public void onStartFollowing(LeaderChangeContext ctx) {
        this.term = ctx.getTerm();
        this.leaderIp = ctx.getLeaderId().getEndpoint().toString();
        NotifyCenter.publishEvent(
                RaftEvent.builder().groupId(groupId).leader(leaderIp).term(ctx.getTerm()).raftClusterInfo(allPeers())
                        .build());
    }
    
    @Override
    public void onConfigurationCommitted(Configuration conf) {
        NotifyCenter.publishEvent(
                RaftEvent.builder().groupId(groupId).raftClusterInfo(JRaftUtils.toStrings(conf.getPeers())).build());
    }
    
    @Override
    public void onError(RaftException e) {
        super.onError(e);
        processor.onError(e);
        NotifyCenter.publishEvent(
                RaftEvent.builder().groupId(groupId).leader(leaderIp).term(term).raftClusterInfo(allPeers())
                        .errMsg(e.toString())
                        .build());
    }
    
    public boolean isLeader() {
        return isLeader.get();
    }
    
    private List<String> allPeers() {
        if (node == null) {
            return Collections.emptyList();
        }
        
        if (node.isLeader()) {
            return JRaftUtils.toStrings(node.listPeers());
        }
        
        return JRaftUtils.toStrings(RouteTable.getInstance().getConfiguration(node.getGroupId()).getPeers());
    }
    
    private void postProcessor(Response data, NacosClosure closure) {
        if (Objects.nonNull(closure)) {
            closure.setResponse(data);
        }
    }
    
    public long getTerm() {
        return term;
    }
    
    private void adapterToJraftSnapshot(Collection<SnapshotOperation> userOperates) {
        List<JSnapshotOperation> tmp = new ArrayList<>();
        
        for (SnapshotOperation item : userOperates) {
            
            if (item == null) {
                Loggers.RAFT.error("Existing SnapshotOperation for null");
                continue;
            }
            
            tmp.add(new JSnapshotOperation() {
                
                @Override
                public void onSnapshotSave(SnapshotWriter writer, Closure done) {
                    final Writer wCtx = new Writer(writer.getPath());
                    
                    // Do a layer of proxy operation to shield different Raft
                    // components from implementing snapshots
                    
                    final BiConsumer<Boolean, Throwable> callFinally = (result, t) -> {
                        Boolean[] results = new Boolean[wCtx.listFiles().size()];
                        int[] index = new int[] {0};
                        wCtx.listFiles().forEach((file, meta) -> {
                            try {
                                results[index[0]++] = writer.addFile(file, buildMetadata(meta));
                            } catch (Exception e) {
                                throw new ConsistencyException(e);
                            }
                        });
                        final Status status = result
                                && Arrays.stream(results).allMatch(Boolean.TRUE::equals) ? Status.OK()
                                : new Status(RaftError.EIO, "Fail to compress snapshot at %s, error is %s",
                                        writer.getPath(), t == null ? "" : t.getMessage());
                        done.run(status);
                    };
                    item.onSnapshotSave(wCtx, callFinally);
                }
                
                @Override
                public boolean onSnapshotLoad(SnapshotReader reader) {
                    final Map<String, LocalFileMeta> metaMap = new HashMap<>(reader.listFiles().size());
                    for (String fileName : reader.listFiles()) {
                        final LocalFileMetaOutter.LocalFileMeta meta = (LocalFileMetaOutter.LocalFileMeta) reader
                                .getFileMeta(fileName);
                        
                        byte[] bytes = meta.getUserMeta().toByteArray();
                        
                        final LocalFileMeta fileMeta;
                        if (bytes == null || bytes.length == 0) {
                            fileMeta = new LocalFileMeta();
                        } else {
                            fileMeta = JacksonUtils.toObj(bytes, LocalFileMeta.class);
                        }
                        
                        metaMap.put(fileName, fileMeta);
                    }
                    final Reader rCtx = new Reader(reader.getPath(), metaMap);
                    return item.onSnapshotLoad(rCtx);
                }
                
                @Override
                public String info() {
                    return item.toString();
                }
            });
        }
        
        this.operations = Collections.unmodifiableList(tmp);
    }
    
}
