走进 P2P 网络：基于 libp2p 打造去中心化集群

package p2p

import (
	"context"
	"fmt"

	"github.com/libp2p/go-libp2p"
	"github.com/libp2p/go-libp2p/core/crypto"
	"github.com/libp2p/go-libp2p/core/host"
	"github.com/libp2p/go-libp2p/core/pnet"
	"github.com/libp2p/go-libp2p/p2p/transport/tcp"
)

// CreateHost 创建一个新的 libp2p host 节点
func CreateHost(ctx context.Context, port int, priv crypto.PrivKey, psk pnet.PSK) (host.Host, error) {
	// 1. 配置基础的 libp2p 选项
	opts := []libp2p.Option{
		// 监听指定的 TCP 端口
		libp2p.ListenAddrStrings(
			fmt.Sprintf("/ip4/0.0.0.0/tcp/%d", port),
		),
		// 配置节点唯一的身份私钥
		libp2p.Identity(priv),
		// 启用 NAT 端口映射 (UPnP/NAT-PMP)
		libp2p.NATPortMap(),
		// 显式指定使用 TCP 传输层（私有网络 PSK 架构下对 QUIC 支持较弱，故锁定 TCP）
		libp2p.Transport(tcp.NewTCPTransport),
		// 启用默认的安全加密握手协议 (Noise/TLS)
		libp2p.DefaultSecurity,
		// 启用默认的多路复用器 (Yamux)
		libp2p.DefaultMuxers,
	}

	// 2. 如果配置了私有网络 Swarm Key，则加入私有网络保护
	if psk != nil {
		opts = append(opts, libp2p.PrivateNetwork(psk))
	}

	// 3. 构造并启动 Host 实例
	h, err := libp2p.New(opts...)
	if err != nil {
		return nil, err
	}

	return h, nil
}

package p2p

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	"github.com/libp2p/go-libp2p/core/host"
	"github.com/libp2p/go-libp2p/core/peer"
	"github.com/libp2p/go-libp2p/p2p/discovery/mdns"
)

const DiscoveryServiceTag = "cosmos-star"

// discoveryNotifee 实现了 mdns.Notifee 接口，用于接收发现事件
type discoveryNotifee struct {
	h host.Host
}

// HandlePeerFound 当局域网内发现新 Peer 时会被触发
func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
	// 忽略自身节点
	if pi.ID == n.h.ID() {
		return
	}

	fmt.Printf("[mDNS] 发现对等节点: %s\n", pi.ID.String())

	// 异步发起连接，加入微小的随机延迟避免本地多实例并发连接冲突
	go func() {
		time.Sleep(time.Duration(rand.Intn(900)+100) * time.Millisecond)

		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()

		err := n.h.Connect(ctx, pi)
		if err != nil {
			fmt.Printf("[mDNS] 无法连接到节点 %s: %s\n", pi.ID.String(), err)
			return
		}

		fmt.Printf("[mDNS] 成功建立连接: %s\n", pi.ID.String())
	}()
}

// SetupDiscovery 启动 mDNS 组播服务
func SetupDiscovery(h host.Host) error {
	s := mdns.NewMdnsService(h, DiscoveryServiceTag, &discoveryNotifee{h: h})
	return s.Start()
}

package p2p

import (
	"context"
	"fmt"
	"sync"
	"time"

	dht "github.com/libp2p/go-libp2p-kad-dht"
	"github.com/libp2p/go-libp2p/core/host"
	"github.com/libp2p/go-libp2p/core/peer"
	"github.com/libp2p/go-libp2p/core/protocol"
	"github.com/multiformats/go-multiaddr"
)

const DHTProtocolID = protocol.ID("/cosmos-star/kad/1.0.0")

// SetupDHT 创建并初始化引导 DHT 路由表
func SetupDHT(ctx context.Context, h host.Host, isCore bool, bootstrapPeers []string) (*dht.IpfsDHT, error) {
	var opts []dht.Option
	// 设置专属于我们集群的协议前缀，隔离公共 IPFS 网络
	opts = append(opts, dht.ProtocolPrefix(DHTProtocolID))

	// 核心骨干节点作为 Server 模式运行（响应他人查询），普通节点作为 Client 运行
	if isCore {
		opts = append(opts, dht.Mode(dht.ModeServer))
	} else {
		opts = append(opts, dht.Mode(dht.ModeClient))
	}

	// 1. 创建 DHT 实例
	kademliaDHT, err := dht.New(ctx, h, opts...)
	if err != nil {
		return nil, fmt.Errorf("failed to create DHT: %w", err)
	}

	// 2. 启动 DHT 引导自愈协程
	if err = kademliaDHT.Bootstrap(ctx); err != nil {
		return nil, fmt.Errorf("failed to bootstrap DHT: %w", err)
	}

	// 3. 异步连接到预设的骨干引导节点
	if len(bootstrapPeers) > 0 {
		go connectToBootstrapPeers(ctx, h, bootstrapPeers)
	}

	return kademliaDHT, nil
}

package p2p

import (
	"context"
	"fmt"
	"sync"

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

type PubSubManager struct {
	ps     *pubsub.PubSub
	topics map[string]*pubsub.Topic
	subs   map[string]*pubsub.Subscription
	mu     sync.RWMutex
}

// SetupPubSub 初始化 GossipSub 广播协议
func SetupPubSub(ctx context.Context, h host.Host) (*PubSubManager, error) {
	// 创建 GossipSub 实例，默认会自动启用消息签名校验
	ps, err := pubsub.NewGossipSub(ctx, h)
	if err != nil {
		return nil, err
	}

	return &PubSubManager{
		ps:     ps,
		topics: make(map[string]*pubsub.Topic),
		subs:   make(map[string]*pubsub.Subscription),
	}, nil
}

// JoinAndSubscribe 加入并订阅主题
func (m *PubSubManager) JoinAndSubscribe(topicName string) (*pubsub.Subscription, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// 1. 加入网状网络主题
	topic, ok := m.topics[topicName]
	if !ok {
		var err error
		topic, err = m.ps.Join(topicName)
		if err != nil {
			return nil, fmt.Errorf("failed to join topic: %w", err)
		}
		m.topics[topicName] = topic
	}

	// 2. 订阅消息主题
	sub, err := topic.Subscribe()
	if err != nil {
		return nil, fmt.Errorf("failed to subscribe: %w", err)
	}
	m.subs[topicName] = sub

	return sub, nil
}