#!/bin/bash
# start-cluster.sh - 本地三节点集群

DATA_BASE="/tmp/dqlite-cluster"
mkdir -p ${DATA_BASE}/node{1,2,3}

# 启动 Node 1
echo "Starting Node 1..."
./dqlite-node 1 "127.0.0.1:9001" "${DATA_BASE}/node1" &
PID1=$!

sleep 2

# 启动 Node 2
echo "Starting Node 2..."
./dqlite-node 2 "127.0.0.1:9002" "${DATA_BASE}/node2" &
PID2=$!

sleep 2

# 启动 Node 3
echo "Starting Node 3..."
./dqlite-node 3 "127.0.0.1:9003" "${DATA_BASE}/node3" &
PID3=$!

sleep 2

echo "Cluster started:"
echo "  Node 1: PID=$PID1, address=127.0.0.1:9001"
echo "  Node 2: PID=$PID2, address=127.0.0.1:9002"
echo "  Node 3: PID=$PID3, address=127.0.0.1:9003"

# 将节点添加到集群（由第一个节点执行）
# 具体方式取决于 go-dqlite 或 C API

#!/bin/bash
# stop-cluster.sh
kill $PID1 $PID2 $PID3
wait
echo "Cluster stopped"

package main

import (
    "context"
    "database/sql"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    dqlite "github.com/canonical/go-dqlite/v2"
    "github.com/canonical/go-dqlite/v2/driver"
)

type ClusterConfig struct {
    Nodes []NodeConfig
}

type NodeConfig struct {
    ID      uint64
    Address string
    DataDir string
}

func startClusterNode(cfg NodeConfig, allNodes []driver.NodeInfo) (*dqlite.Node, *sql.DB, error) {
    // 确保数据目录
    if err := os.MkdirAll(cfg.DataDir, 0755); err != nil {
        return nil, nil, fmt.Errorf("mkdir: %w", err)
    }

    // 创建节点
    node, err := dqlite.New(cfg.ID, cfg.Address, cfg.DataDir, nil)
    if err != nil {
        return nil, nil, fmt.Errorf("new node: %w", err)
    }

    // 创建驱动（需要知道所有节点地址）
    store := driver.NewInmemNodeStore()
    store.Set(context.Background(), allNodes)

    drv, err := driver.New(store)
    if err != nil {
        node.Close()
        return nil, nil, fmt.Errorf("new driver: %w", err)
    }

    db := sql.OpenDB(drv)
    db.SetMaxOpenConns(5)

    return node, db, nil
}

func main() {
    // 集群配置
    nodes := []NodeConfig{
        {ID: 1, Address: "127.0.0.1:9001", DataDir: "/tmp/dqlite-cluster/node1"},
        {ID: 2, Address: "127.0.0.1:9002", DataDir: "/tmp/dqlite-cluster/node2"},
        {ID: 3, Address: "127.0.0.1:9003", DataDir: "/tmp/dqlite-cluster/node3"},
    }

    allNodeInfos := []driver.NodeInfo{
        {ID: 1, Address: "127.0.0.1:9001"},
        {ID: 2, Address: "127.0.0.1:9002"},
        {ID: 3, Address: "127.0.0.1:9003"},
    }

    // 启动所有节点
    var (
        nodeInstances []*dqlite.Node
        databases     []*sql.DB
    )

    for _, cfg := range nodes {
        node, db, err := startClusterNode(cfg, allNodeInfos)
        if err != nil {
            log.Fatalf("Failed to start node %d: %v", cfg.ID, err)
        }
        nodeInstances = append(nodeInstances, node)
        databases = append(databases, db)
        fmt.Printf("Node %d started on %s\n", cfg.ID, cfg.Address)
    }

    defer func() {
        for i, node := range nodeInstances {
            databases[i].Close()
            node.Close()
        }
    }()

    // 等待集群稳定
    time.Sleep(3 * time.Second)

    // 使用第一个节点的数据库进行操作
    db := databases[0]
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    if err := db.PingContext(ctx); err != nil {
        log.Fatalf("Ping failed: %v", err)
    }

    // 创建测试表
    _, err := db.Exec(`CREATE TABLE IF NOT EXISTS cluster_test (
        id INTEGER PRIMARY KEY,
        node_id INTEGER,
        created_at TEXT DEFAULT (datetime('now'))
    )`)
    if err != nil {
        log.Fatalf("Create table failed: %v", err)
    }

    fmt.Println("Cluster is ready!")

    // 等待信号
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    <-sigCh

    fmt.Println("\nShutting down cluster...")
}

# /etc/systemd/system/dqlite.service
[Unit]
Description=dqlite Node
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=dqlite
Group=dqlite
ExecStart=/usr/local/bin/dqlite-node --id %i --address :9001 --data-dir /var/lib/dqlite
Restart=always
RestartSec=5
LimitNOFILE=65536

# 安全加固
NoNewPrivileges=true
ProtectSystem=strict
ReadWritePaths=/var/lib/dqlite

[Install]
WantedBy=multi-user.target

// 通过 SQL 执行 Leader 转移（如果 go-dqlite 支持）
// 注意：具体的 API 可能随版本变化

// 方式 1：停止当前 Leader，集群自动选举新 Leader
func gracefulLeaderTransfer(db *sql.DB) error {
    // 1. 停止接受新写入
    _, err := db.Exec("PRAGMA dqlite_stop_writes")
    if err != nil {
        return err
    }

    // 2. 等待所有写入完成
    time.Sleep(2 * time.Second)

    // 3. 新的 Leader 会在选举超时后自动产生
    return nil
}

// 方式 2：使用 dqlite 客户端的 Transfer 命令
// （需要连接到当前 Leader）

// 客户端需要处理 Leader 切换期间的错误
func executeWithFailover(ctx context.Context, dbs []*sql.DB, query string, args ...interface{}) error {
    var lastErr error

    for _, db := range dbs {
        _, err := db.ExecContext(ctx, query, args...)
        if err == nil {
            return nil
        }

        // 如果是 Leader 相关错误，尝试下一个节点
        if isLeaderError(err) {
            lastErr = err
            continue
        }

        // 其他错误直接返回
        return err
    }

    return fmt.Errorf("all nodes failed: %w", lastErr)
}

func isLeaderError(err error) bool {
    if err == nil {
        return false
    }
    msg := err.Error()
    return strings.Contains(msg, "server is not the leader") ||
           strings.Contains(msg, "no leader") ||
           strings.Contains(msg, "connection refused")
}

// 集群健康检查
type ClusterHealth struct {
    NodeID    uint64
    Address   string
    IsLeader  bool
    IsHealthy bool
    Lag       time.Duration
}

func checkClusterHealth(nodeInfos []driver.NodeInfo) []ClusterHealth {
    var results []ClusterHealth

    for _, info := range nodeInfos {
        health := ClusterHealth{
            NodeID:  info.ID,
            Address: info.Address,
        }

        // 尝试连接
        store := driver.NewInmemNodeStore()
        store.Set(context.Background(), []driver.NodeInfo{info})

        drv, err := driver.New(store)
        if err != nil {
            health.IsHealthy = false
            results = append(results, health)
            continue
        }

        db := sql.OpenDB(drv)
        ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)

        start := time.Now()
        err = db.PingContext(ctx)
        health.Lag = time.Since(start)

        health.IsHealthy = (err == nil)

        db.Close()
        drv.Close()
        cancel()

        results = append(results, health)
    }

    return results
}

// 添加新节点到集群
func addNodeToCluster(ctx context.Context, leaderDB *sql.DB,
    newNodeID uint64, newNodeAddr string) error {

    // 注意：具体的 SQL 或 API 调用取决于 go-dqlite 版本
    // 以下为概念性代码

    // 使用 dqlite 客户端的 Add 命令
    // 在实际使用中，需要通过 go-dqlite 的 Client API

    log.Printf("Adding node %d at %s to cluster...", newNodeID, newNodeAddr)

    // 方式 1：通过 go-dqlite Client API
    // client.Add(ctx, dqlite.NodeInfo{ID: newNodeID, Address: newNodeAddr})

    // 方式 2：通过自定义 SQL 命令
    // _, err := leaderDB.Exec("ALTER CLUSTER ADD NODE ?", newNodeID)

    return nil
}

// 从集群移除节点
func removeNodeFromCluster(ctx context.Context, leaderDB *sql.DB,
    nodeID uint64) error {

    log.Printf("Removing node %d from cluster...", nodeID)

    // 类似添加节点，通过客户端 API 执行

    return nil
}

#!/bin/bash
# cluster-manage.sh - 集群成员管理脚本

set -euo pipefail

ACTION="$1"
NODE_ID="$2"
NODE_ADDR="${3:-}"

case "$ACTION" in
    add)
        if [ -z "$NODE_ADDR" ]; then
            echo "Usage: $0 add <node-id> <node-address>"
            exit 1
        fi
        echo "Adding node $NODE_ID ($NODE_ADDR) to cluster..."
        # 通过 go-dqlite CLI 或 API 添加
        # go-dqlite cluster add $NODE_ID $NODE_ADDR
        ;;
    remove)
        echo "Removing node $NODE_ID from cluster..."
        # go-dqlite cluster remove $NODE_ID
        ;;
    list)
        echo "Listing cluster members..."
        # go-dqlite cluster list
        ;;
    *)
        echo "Usage: $0 {add|remove|list} <node-id> [node-address]"
        exit 1
        ;;
esac

// 获取集群信息
func getClusterInfo(db *sql.DB) error {
    // 查询集群信息（具体 SQL 取决于版本）
    rows, err := db.Query("SELECT * FROM dqlite_cluster")
    if err != nil {
        return err
    }
    defer rows.Close()

    fmt.Println("=== Cluster Info ===")
    // 遍历结果...
    return nil
}

// 查看 Raft 状态
type RaftStatus struct {
    Term         uint64
    CommitIndex  uint64
    LastLogIndex uint64
    Role         string
}