// 创建OrbitDB实例
import { createLibp2p } from 'libp2p'
import { createHelia } from 'helia'
import { createOrbitDB } from '@orbitdb/core'

const libp2p = await createLibp2p({ /* Libp2p options */ })
const ipfs = await createHelia({ libp2p }) // Helia是存储和网络通信所必需的
const orbitdb = await createOrbitDB({ ipfs })

// 创建数据库
const mydb = await orbitdb.open('mydb')
console.log(mydb.address) // /orbitdb/zdpuAuK3BHpS7NvMBivynypqciYCuy2UW77XYBPUYRnLjnw13

// 添加数据
await mydb.add("hello world!")

// 兰伯特时钟示例，用于在CRDTs中确定操作顺序
type lamportClock struct {
    time int
    id   crypto.PublicKey
}

// 比较函数，用于确定两个操作的顺序
func compareClock(a, b lamportClock) int {
    dist := a.time - b.time

    if dist == 0 {
        dist = comparePubKey(a.id, b.id) // 返回字典序距离
    }

    return dist
}

// 在Berty/Wesh协议中使用OrbitDB的示例
// 初始化OrbitDB实例
orbitdb, err := orbitdb.NewOrbitDB(ctx, ipfs, &orbitdb.NewOrbitDBOptions{
    Directory: &dbPath,
})

// 打开群组的元数据日志
metadataStore, err := orbitdb.Open(ctx, group.MetadataLogAddr, &orbitdb.OpenOptions{})
if err != nil {
    return nil, err
}

// 打开群组的消息日志
messageStore, err := orbitdb.Open(ctx, group.MessageLogAddr, &orbitdb.OpenOptions{})
if err != nil {
    return nil, err
}

// 添加新成员到元数据日志
memberEntry := &protocol.MemberEntry{
    DevicePK: devicePK,
    MemberPK: memberPK,
}

memberEntryBytes, err := proto.Marshal(memberEntry)
if err != nil {
    return err
}

_, err = metadataStore.Add(ctx, memberEntryBytes, &orbitdb.WriteOptions{})
if err != nil {
    return err
}

// Gossipsub评分机制的简化示例
type PeerScore struct {
    score float64
    topics map[string]*TopicScore
}

// 更新节点评分
func (ps *PeerScore) Update(topic string, valid bool) {
    topicScore, ok := ps.topics[topic]
    if !ok {
        topicScore = &TopicScore{score: 0}
        ps.topics[topic] = topicScore
    }
    
    if valid {
        // 有效消息增加评分
        topicScore.score += 1.0
    } else {
        // 无效消息降低评分
        topicScore.score -= 10.0
    }
    
    // 更新总体评分
    ps.score = 0
    for _, ts := range ps.topics {
        ps.score += ts.score
    }
}

// 在Berty/Wesh协议中使用Gossipsub的简化示例
import (
    "context"
    
    pubsub "github.com/libp2p/go-libp2p-pubsub"
    "github.com/libp2p/go-libp2p-core/host"
)

// 初始化Gossipsub
func setupGossipsub(ctx context.Context, h host.Host) (*pubsub.PubSub, error) {
    // 创建Gossipsub实例
    gossipSub, err := pubsub.NewGossipSub(ctx, h, 
        pubsub.WithMessageSignature(true),
        pubsub.WithStrictSignatureVerification(true),
        pubsub.WithPeerScore(
            // 配置评分参数
            &pubsub.PeerScoreParams{
                // 评分参数配置
            },
            // 评分阈值配置
            &pubsub.PeerScoreThresholds{
                // 阈值配置
            },
        ),
    )
    
    if err != nil {
        return nil, err
    }
    
    return gossipSub, nil
}

// 加入主题并订阅消息
func joinTopic(ctx context.Context, ps *pubsub.PubSub, topicName string) (*pubsub.Topic, *pubsub.Subscription, error) {
    // 加入主题
    topic, err := ps.Join(topicName)
    if err != nil {
        return nil, nil, err
    }
    
    // 订阅主题
    sub, err := topic.Subscribe()
    if err != nil {
        return nil, nil, err
    }
    
    return topic, sub, nil
}

go get github.com/libp2p/go-libp2p-pubsub

import (
    "context"
    "github.com/libp2p/go-libp2p"
    pubsub "github.com/libp2p/go-libp2p-pubsub"
)

// 创建 libp2p 主机
ctx := context.Background()
host, err := libp2p.New()
if err != nil {
    panic(err)
}

// 创建 Gossipsub 实例
ps, err := pubsub.NewGossipSub(ctx, host)
if err != nil {
    panic(err)
}

// 创建带有高级配置的 Gossipsub 实例
ps, err := pubsub.NewGossipSub(ctx, host,
    // 启用消息签名
    pubsub.WithMessageSignature(true),
    
    // 启用严格签名验证
    pubsub.WithStrictSignatureVerification(true),
    
    // 配置对等节点评分机制
    pubsub.WithPeerScore(
        &pubsub.PeerScoreParams{
            // 评分参数配置
            AppSpecificScore: func(p peer.ID) float64 {
                // 自定义评分函数
                return 0
            },
            AppSpecificWeight: 1.0,
            IPColocationFactorWeight: -35.31,
            IPColocationFactorThreshold: 10,
            BehaviourPenaltyWeight: -15.92,
            BehaviourPenaltyThreshold: 6,
            DecayInterval: 12 * time.Second,
            DecayToZero: 0.01,
            RetainScore: 10 * time.Minute,
        },
        &pubsub.PeerScoreThresholds{
            // 评分阈值配置
            GossipThreshold: -100,
            PublishThreshold: -200,
            GraylistThreshold: -400,
            AcceptPXThreshold: 100,
            OpportunisticGraftThreshold: 5,
        },
    ),
    
    // 配置其他选项
    pubsub.WithFloodPublish(true),
    pubsub.WithSeenMessagesTTL(2*time.Minute),
)

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/libp2p/go-libp2p"
    "github.com/libp2p/go-libp2p-core/peer"
    pubsub "github.com/libp2p/go-libp2p-pubsub"
)

func main() {
    ctx := context.Background()
    
    // 创建 libp2p 主机
    host, err := libp2p.New()
    if err != nil {
        panic(err)
    }
    
    // 创建 Gossipsub 实例
    ps, err := pubsub.NewGossipSub(ctx, host)
    if err != nil {
        panic(err)
    }
    
    // 定义主题名称
    topicName := "example-topic"
    
    // 加入主题
    topic, err := ps.Join(topicName)
    if err != nil {
        panic(err)
    }
    
    // 订阅主题
    sub, err := topic.Subscribe()
    if err != nil {
        panic(err)
    }
    
    // 启动消息处理协程
    go handleMessages(sub)
    
    // 发布消息
    publishMessages(ctx, topic)
    
    // 等待程序退出
    select {}
}

// 处理接收到的消息
func handleMessages(sub *pubsub.Subscription) {
    for {
        msg, err := sub.Next(context.Background())
        if err != nil {
            fmt.Println("Error reading from subscription:", err)
            continue
        }
        
        fmt.Printf("Received message from %s: %s\n", msg.ReceivedFrom, string(msg.Data))
    }
}

// 发布消息
func publishMessages(ctx context.Context, topic *pubsub.Topic) {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()
    
    for i := 0; ; i++ {
        select {
        case <-ticker.C:
            message := fmt.Sprintf("Message #%d", i)
            err := topic.Publish(ctx, []byte(message))
            if err != nil {
                fmt.Println("Error publishing message:", err)
            } else {
                fmt.Println("Published message:", message)
            }
        case <-ctx.Done():
            return
        }
    }
}

package berty

import (
    "context"
    "crypto/rand"
    "encoding/hex"
    
    "github.com/libp2p/go-libp2p-core/crypto"
    "github.com/libp2p/go-libp2p-core/peer"
    pubsub "github.com/libp2p/go-libp2p-pubsub"
)

// Group 表示一个 Berty/Wesh 群组
type Group struct {
    ID        string
    SecretKey  crypto.PrivKey
    PublicKey crypto.PubKey
    Topic     *pubsub.Topic
    Sub       *pubsub.Subscription
    PubSub    *pubsub.PubSub
}

// NewGroup 创建一个新的群组
func NewGroup(ctx context.Context, ps *pubsub.PubSub) (*Group, error) {
    // 生成群组密钥对
    privKey, pubKey, err := crypto.GenerateEd25519Key(rand.Reader)
    if err != nil {
        return nil, err
    }
    
    // 生成群组ID
    pubKeyBytes, err := pubKey.Raw()
    if err != nil {
        return nil, err
    }
    groupID := hex.EncodeToString(pubKeyBytes)
    
    // 加入主题
    topicName := "/berty/group/" + groupID
    topic, err := ps.Join(topicName)
    if err != nil {
        return nil, err
    }
    
    // 订阅主题
    sub, err := topic.Subscribe()
    if err != nil {
        return nil, err
    }
    
    return &Group{
        ID:        groupID,
        SecretKey:  privKey,
        PublicKey: pubKey,
        Topic:     topic,
        Sub:       sub,
        PubSub:    ps,
    }, nil
}

// SendMessage 向群组发送消息
func (g *Group) SendMessage(ctx context.Context, message []byte) error {
    // 在实际应用中，这里应该对消息进行加密
    // 加密后的消息 = Encrypt(message, groupPublicKey)
    
    // 发布消息
    return g.Topic.Publish(ctx, message)
}

// ReceiveMessages 处理群组消息
func (g *Group) ReceiveMessages(ctx context.Context, handler func(peer.ID, []byte)) {
    go func() {
        for {
            msg, err := g.Sub.Next(ctx)
            if err != nil {
                continue
            }
            
            // 在实际应用中，这里应该对消息进行解密
            // 原始消息 = Decrypt(msg.Data, groupSecretKey)
            
            handler(msg.ReceivedFrom, msg.Data)
        }
    }()
}

// 启用消息签名
ps, err := pubsub.NewGossipSub(ctx, host,
    pubsub.WithMessageSignature(true),
    pubsub.WithStrictSignatureVerification(true),
)

// 发布签名消息
err = topic.Publish(ctx, []byte("Hello, world!"),
    pubsub.WithSignMessage(true), // 启用消息签名
)

// 验证接收到的消息
msg, err := sub.Next(ctx)
if err != nil {
    return err
}

// 验证消息签名
valid, err := msg.Verify()
if err != nil {
    return err
}

if !valid {
    // 消息签名验证失败
    return fmt.Errorf("invalid message signature")
}

// 配置对等节点评分机制
ps, err := pubsub.NewGossipSub(ctx, host,
    pubsub.WithPeerScore(
        &pubsub.PeerScoreParams{
            // 应用特定评分函数
            AppSpecificScore: func(p peer.ID) float64 {
                // 根据对等节点的行为返回评分
                // 例如，可以根据历史交互、信誉等因素评分
                return 0
            },
            AppSpecificWeight: 1.0,
            
            // IP 共置惩罚
            IPColocationFactorWeight: -35.31,
            IPColocationFactorThreshold: 10,
            
            // 行为惩罚
            BehaviourPenaltyWeight: -15.92,
            BehaviourPenaltyThreshold: 6,
            
            // 评分衰减参数
            DecayInterval: 12 * time.Second,
            DecayToZero: 0.01,
            RetainScore: 10 * time.Minute,
        },
        &pubsub.PeerScoreThresholds{
            // 评分阈值
            GossipThreshold: -100,
            PublishThreshold: -200,
            GraylistThreshold: -400,
            AcceptPXThreshold: 100,
            OpportunisticGraftThreshold: 5,
        },
    ),
)

// 1. 调整 mesh 参数
ps, err := pubsub.NewGossipSub(ctx, host,
    // 设置 mesh 中的目标连接数
    pubsub.WithDegree(10),
    pubsub.WithDegreeLow(5),
    pubsub.WithDegreeHigh(15),
    
    // 设置 fanout 中的目标连接数
    pubsub.WithFanoutTTL(60*time.Second),
    
    // 设置消息缓存时间
    pubsub.WithSeenMessagesTTL(2*time.Minute),
)

// 2. 使用连接池
// go-libp2p-pubsub 内部使用连接池管理连接，可以通过调整连接池参数优化性能
host, err := libp2p.New(
    libp2p.ConnectionManager(lowWater, highWater, gracePeriod),
)

// 3. 限制消息处理速率
// 使用速率限制器控制消息处理速率，防止资源耗尽
limiter := rate.NewLimiter(rate.Limit(1000), 100) // 每秒最多处理1000条消息，突发100条

go func() {
    for {
        msg, err := sub.Next(ctx)
        if err != nil {
            continue
        }
        
        // 等待速率限制器允许
        if err := limiter.Wait(ctx); err != nil {
            continue
        }
        
        // 处理消息
        handleMessage(msg)
    }
}()

// 4. 并行处理消息
// 使用工作池并行处理消息，提高吞吐量
workerCount := runtime.NumCPU()
messageCh := make(chan *pubsub.Message, 100)

// 启动工作池
for i := 0; i < workerCount; i++ {
    go func() {
        for msg := range messageCh {
            handleMessage(msg)
        }
    }()
}

// 从订阅中读取消息并发送到工作池
go func() {
    for {
        msg, err := sub.Next(ctx)
        if err != nil {
            continue
        }
        
        select {
        case messageCh <- msg:
        default:
            // 工作池已满，丢弃消息
        }
    }
}()

// 1. 启用事件追踪
ps, err := pubsub.NewGossipSub(ctx, host,
    pubsub.WithEventTracer(true),
    pubsub.WithTracer(&pubsub.JSONTracer{
        Writer: os.Stdout, // 将事件输出到标准输出
    }),
)

// 2. 监控对等节点评分
// 定期检查对等节点评分，识别异常节点
go func() {
    ticker := time.NewTicker(1 * time.Minute)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            // 获取所有对等节点的评分
            scores := ps.Score()
            
            // 输出评分最低的节点
            var lowestScorePeer peer.ID
            var lowestScore float64 = math.MaxFloat64
            
            for p, score := range scores {
                if score < lowestScore {
                    lowestScore = score
                    lowestScorePeer = p
                }
            }
            
            if lowestScorePeer != "" {
                log.Printf("Lowest score peer: %s, score: %f", lowestScorePeer, lowestScore)
            }
        case <-ctx.Done():
            return
        }
    }
}()

// 3. 监控主题统计信息
// 定期检查主题统计信息，了解网络状况
go func() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            // 获取主题统计信息
            stats := topic.Stats()
            
            log.Printf("Topic stats: %+v", stats)
        case <-ctx.Done():
            return
        }
    }
}()