func (s *StandAloneStorage) Write(ctx *kvrpcpb.Context, batch []storage.Modify) error {
    if len(batch) == 0 {
        return nil
    }

    var err error
    for _, m := range batch {
        key, val, cf := m.Key(), m.Value(), m.Cf()
        switch data := m.Data.(type) {
        case storage.Put:
            err = engine_util.PutCF(s.engine.Kv, cf, key, val)
        case storage.Delete:
            err = engine_util.DeleteCF(s.engine.Kv, cf, key)
        default:
            return fmt.Errorf("unsupported modify type: %T", data)
        }

        if err != nil {
            return fmt.Errorf("failed to execute modification on key %s: %w", key, err)
        }
    }
    return nil
}

	GetCF(cf string, key []byte) ([]byte, error)
	IterCF(cf string) engine_util.DBIterator
	Close()

type StandAloneStorageReader struct {
	KvTxn *badger.Txn
}

func NewStandAloneStorageReader(txn *badger.Txn) *StandAloneStorageReader {
	if txn == nil {
		return nil
	}
	return &StandAloneStorageReader{
		KvTxn: txn,
	}
}
func (s *StandAloneStorage) Reader(ctx *kvrpcpb.Context) (storage.StorageReader, error) {
	txn := s.engine.Kv.NewTransaction(false)
	return NewStandAloneStorageReader(txn), nil
}

func (s *StandAloneStorageReader) GetCF(cf string, key []byte) ([]byte, error) {
	val, err := engine_util.GetCFFromTxn(s.KvTxn, cf, key)
	if err == badger.ErrKeyNotFound {
		return nil, nil
	}
	return val, err
}

func (s *StandAloneStorageReader) IterCF(cf string) engine_util.DBIterator {
	return engine_util.NewCFIterator(cf, s.KvTxn)
}

func (s *StandAloneStorageReader) Close() {
	s.KvTxn.Discard()
}

func (server *Server) RawScan(_ context.Context, req *kvrpcpb.RawScanRequest) (*kvrpcpb.RawScanResponse, error) {
    reader, err := server.storage.Reader(req.Context)
    if err != nil {
        return nil, err
    }

    Kvs := make([]*kvrpcpb.KvPair, 0, req.Limit)

    iter := reader.IterCF(req.Cf)
    defer iter.Close()

    for iter.Seek(req.StartKey); iter.Valid(); iter.Next() {
        if req.Limit == 0 {
            break
        }

        item := iter.Item()
        key := item.Key()
        val, err := item.Value() 
        if err != nil {
            return nil, err
        }

        Kvs = append(Kvs, &kvrpcpb.KvPair{
            Key:   key,
            Value: val,
        })
        req.Limit--
    }

    resp := &kvrpcpb.RawScanResponse{
        Kvs: Kvs,
    }
    return resp, nil
}

// RaftLog manage the log entries, its struct look like:
//
//  snapshot/first.....applied....committed....stabled.....last
//  --------|------------------------------------------------|
//                            log entries
//
// for simplify the RaftLog implement should manage all log entries
// that not truncated
type RaftLog struct {
	// storage contains all stable entries since the last snapshot.
	// 存储包含自上次快照以来的所有稳定条目。
	storage Storage

	// committed is the highest log position that is known to be in
	// stable storage on a quorum of nodes.
	// 已知已提交的最高的日志条目的索引     committedIndex
	committed uint64

	// applied is the highest log position that the application has
	// been instructed to apply to its state machine.
	// Invariant: applied <= committed
	// 已经被应用到状态机的最高的日志条目的索引 appliedIndex
	applied uint64

	// log entries with index <= stabled are persisted to storage.
	// It is used to record the logs that are not persisted by storage yet.
	// Everytime handling `Ready`, the unstabled logs will be included.
	// stabled 保存的是已经持久化到 storage 的 index
	stabled uint64

	// all entries that have not yet compact.
	// 尚未压缩的所有条目
	entries []pb.Entry

	// the incoming unstable snapshot, if any.
	// 2C
	pendingSnapshot *pb.Snapshot

	// Your Data Here (2A).
	dummyIndex uint64
}

func newLog(storage Storage) *RaftLog {
	// Your Code Here (2A).
	return nil
}

func newLog(storage Storage) *RaftLog {
	firstIndex, _ := storage.FirstIndex()
	lastIndex, _ := storage.LastIndex()
	entries, _ := storage.Entries(firstIndex, lastIndex+1)
	hardState, _, _ := storage.InitialState()
	
	rl := &RaftLog{
		storage:    storage,
		committed:  hardState.Commit,
		applied:    firstIndex - 1,
		stabled:    lastIndex,
		entries:    entries,
		pendingSnapshot: nil,
		dummyIndex: firstIndex,
	}

	return rl
}

const (
	// 'MessageType_MsgHup' is a local message used for election. If an election timeout happened,
	// the node should pass 'MessageType_MsgHup' to its Step method and start a new election.
	MessageType_MsgHup MessageType = 0
	// 'MessageType_MsgBeat' is a local message that signals the leader to send a heartbeat
	// of the 'MessageType_MsgHeartbeat' type to its followers.
	MessageType_MsgBeat MessageType = 1
	// 'MessageType_MsgPropose' is a local message that proposes to append data to the leader's log entries.
	MessageType_MsgPropose MessageType = 2
	// 'MessageType_MsgAppend' contains log entries to replicate.
	MessageType_MsgAppend MessageType = 3
	// 'MessageType_MsgAppendResponse' is response to log replication request('MessageType_MsgAppend').
	MessageType_MsgAppendResponse MessageType = 4
	// 'MessageType_MsgRequestVote' requests votes for election.
	MessageType_MsgRequestVote MessageType = 5
	// 'MessageType_MsgRequestVoteResponse' contains responses from voting request.
	MessageType_MsgRequestVoteResponse MessageType = 6
	// 'MessageType_MsgSnapshot' requests to install a snapshot message.
	MessageType_MsgSnapshot MessageType = 7
	// 'MessageType_MsgHeartbeat' sends heartbeat from leader to its followers.
	MessageType_MsgHeartbeat MessageType = 8
	// 'MessageType_MsgHeartbeatResponse' is a response to 'MessageType_MsgHeartbeat'.
	MessageType_MsgHeartbeatResponse MessageType = 9
	// 'MessageType_MsgTransferLeader' requests the leader to transfer its leadership.
	MessageType_MsgTransferLeader MessageType = 11
	// 'MessageType_MsgTimeoutNow' send from the leader to the leadership transfer target, to let
	// the transfer target timeout immediately and start a new election.
	MessageType_MsgTimeoutNow MessageType = 12
)