// Primary 端：拦截写入，转发 WAL 到 Replica
class ReplicatedDB {
public:
    bool Put(const std::string& key, const std::string& value) {
        // 1. 写入本地
        leveldb::Status s = db_->Put(wopts_, key, value);
        if (!s.ok()) return false;

        // 2. 发送到 Replica（简化示意）
        std::string wal_entry = SerializeWALEntry(key, value);
        SendToReplica(wal_entry);
        return true;
    }

private:
    void SendToReplica(const std::string& entry) {
        // 通过 TCP/gRPC 发送到 Replica
        // 需要处理：网络超时、重试、顺序保证
    }
};

// 1. 初始同步：复制整个数据库
void InitialSync(leveldb::DB* primary, leveldb::DB* replica) {
    const leveldb::Snapshot* snap = primary->GetSnapshot();
    leveldb::ReadOptions ropts;
    ropts.snapshot = snap;

    leveldb::WriteBatch batch;
    auto* it = primary->NewIterator(ropts);
    int count = 0;
    for (it->SeekToFirst(); it->Valid(); it->Next()) {
        batch.Put(it->key(), it->value());
        if (++count % 10000 == 0) {
            replica->Write(leveldb::WriteOptions(), &batch);
            batch = leveldb::WriteBatch();
        }
    }
    if (count % 10000 != 0) {
        replica->Write(leveldb::WriteOptions(), &batch);
    }

    delete it;
    primary->ReleaseSnapshot(snap);
}

// 2. 增量同步：记录写入日志
class WriteLogger {
public:
    void Log(const std::string& op, const std::string& key,
             const std::string& value) {
        std::lock_guard<std::mutex> lock(mu_);
        LogEntry entry{op, key, value, sequence_++};
        entries_.push_back(entry);
        // 持久化到文件或发送到消息队列
    }

    std::vector<LogEntry> GetEntriesSince(uint64_t seq) {
        std::lock_guard<std::mutex> lock(mu_);
        return {entries_.begin() + seq, entries_.end()};
    }

private:
    std::mutex mu_;
    uint64_t sequence_ = 0;
    std::vector<LogEntry> entries_;
};

// 简单但不推荐（不保证一致性）
bool DoubleWrite(leveldb::DB* db1, leveldb::DB* db2,
                 const std::string& key, const std::string& value) {
    leveldb::Status s1 = db1->Put(wopts, key, value);
    leveldb::Status s2 = db2->Put(wopts, key, value);
    return s1.ok() && s2.ok();
    // 问题：db1 成功但 db2 失败怎么办？
}

// Raft 状态机接口
class LevelDBStateMachine : public raft::StateMachine {
public:
    LevelDBStateMachine(const std::string& db_path) {
        leveldb::Options opts;
        opts.create_if_missing = true;
        leveldb::DB::Open(opts, db_path, &db_);
    }

    // 应用已提交的日志条目
    void Apply(const raft::LogEntry& entry) override {
        // 解析命令
        Command cmd = Deserialize(entry.data());

        switch (cmd.type) {
            case Command::PUT:
                db_->Put(leveldb::WriteOptions(), cmd.key, cmd.value);
                break;
            case Command::DELETE:
                db_->Delete(leveldb::WriteOptions(), cmd.key);
                break;
            case Command::BATCH:
                db_->Write(leveldb::WriteOptions(), &cmd.batch);
                break;
        }
    }

    // 创建快照
    raft::Snapshot* TakeSnapshot() override {
        const leveldb::Snapshot* snap = db_->GetSnapshot();
        // 序列化所有数据...
        return new LevelDBSnapshot(snap);
    }

    // 恢复快照
    void RestoreSnapshot(raft::Snapshot* snapshot) override {
        // 从快照恢复数据到 LevelDB...
    }

private:
    leveldb::DB* db_;
};

package main

import (
    "encoding/json"
    "io"
    "net"

    "github.com/hashicorp/raft"
    "github.com/hashicorp/raft-boltdb"
    "github.com/syndtr/goleveldb/leveldb"
)

type LevelDBFSM struct {
    db *leveldb.DB
}

func (fsm *LevelDBFSM) Apply(l *raft.Log) interface{} {
    var cmd struct {
        Op    string `json:"op"`
        Key   string `json:"key"`
        Value string `json:"value,omitempty"`
    }
    json.Unmarshal(l.Data, &cmd)

    switch cmd.Op {
    case "put":
        return fsm.db.Put([]byte(cmd.Key), []byte(cmd.Value), nil)
    case "delete":
        return fsm.db.Delete([]byte(cmd.Key), nil)
    }
    return nil
}

func (fsm *LevelDBFSM) Snapshot() (raft.FSMSnapshot, error) {
    // 创建一致性快照
    snap, _ := fsm.db.GetSnapshot()
    return &LevelDBSnapshot{snap: snap}, nil
}

func (fsm *LevelDBFSM) Restore(rc io.ReadCloser) error {
    // 从快照恢复
    // ...
    return nil
}

func main() {
    db, _ := leveldb.OpenFile("/data/raft-node-1", nil)
    fsm := &LevelDBFSM{db: db}

    config := raft.DefaultConfig()
    config.LocalID = raft.ServerID("node1")

    addr, _ := net.ResolveTCPAddr("tcp", "127.0.0.1:12000")
    transport, _ := raft.NewTCPTransport(addr.String(), addr, 3, 10*time.Second, os.Stderr)

    logStore, _ := raftboltdb.NewBoltStore("/data/raft-log-1")
    stableStore, _ := raftboltdb.NewBoltStore("/data/raft-stable-1")
    snapshotStore, _ := raft.NewFileSnapshotStore("/data/raft-snap-1", 3, os.Stderr)

    r, _ := raft.NewRaft(config, fsm, logStore, stableStore, snapshotStore, transport)
    // ...
}