118 lines
2.8 KiB
Go
118 lines
2.8 KiB
Go
package dht
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"chorus/pkg/config"
|
|
libp2p "github.com/libp2p/go-libp2p"
|
|
"github.com/libp2p/go-libp2p/core/host"
|
|
"github.com/libp2p/go-libp2p/core/peer"
|
|
"github.com/libp2p/go-libp2p/p2p/security/noise"
|
|
"github.com/libp2p/go-libp2p/p2p/transport/tcp"
|
|
"github.com/multiformats/go-multiaddr"
|
|
)
|
|
|
|
// RealDHT wraps a libp2p-based DHT to satisfy the generic DHT interface.
|
|
type RealDHT struct {
|
|
cancel context.CancelFunc
|
|
host host.Host
|
|
dht *LibP2PDHT
|
|
}
|
|
|
|
// NewRealDHT creates a new real DHT implementation backed by libp2p.
|
|
func NewRealDHT(cfg *config.HybridConfig) (DHT, error) {
|
|
if cfg == nil {
|
|
cfg = &config.HybridConfig{}
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
|
|
listenAddr, err := multiaddr.NewMultiaddr("/ip4/0.0.0.0/tcp/0")
|
|
if err != nil {
|
|
cancel()
|
|
return nil, fmt.Errorf("failed to create listen address: %w", err)
|
|
}
|
|
|
|
host, err := libp2p.New(
|
|
libp2p.ListenAddrs(listenAddr),
|
|
libp2p.Security(noise.ID, noise.New),
|
|
libp2p.Transport(tcp.NewTCPTransport),
|
|
libp2p.DefaultMuxers,
|
|
libp2p.EnableRelay(),
|
|
)
|
|
if err != nil {
|
|
cancel()
|
|
return nil, fmt.Errorf("failed to create libp2p host: %w", err)
|
|
}
|
|
|
|
opts := []Option{
|
|
WithProtocolPrefix("/CHORUS"),
|
|
}
|
|
|
|
if nodes := cfg.GetDHTBootstrapNodes(); len(nodes) > 0 {
|
|
opts = append(opts, WithBootstrapPeersFromStrings(nodes))
|
|
}
|
|
|
|
libp2pDHT, err := NewLibP2PDHT(ctx, host, opts...)
|
|
if err != nil {
|
|
host.Close()
|
|
cancel()
|
|
return nil, fmt.Errorf("failed to initialize libp2p DHT: %w", err)
|
|
}
|
|
|
|
if err := libp2pDHT.Bootstrap(); err != nil {
|
|
libp2pDHT.Close()
|
|
host.Close()
|
|
cancel()
|
|
return nil, fmt.Errorf("failed to bootstrap DHT: %w", err)
|
|
}
|
|
|
|
return &RealDHT{
|
|
cancel: cancel,
|
|
host: host,
|
|
dht: libp2pDHT,
|
|
}, nil
|
|
}
|
|
|
|
// PutValue stores a value in the DHT.
|
|
func (r *RealDHT) PutValue(ctx context.Context, key string, value []byte) error {
|
|
return r.dht.PutValue(ctx, key, value)
|
|
}
|
|
|
|
// GetValue retrieves a value from the DHT.
|
|
func (r *RealDHT) GetValue(ctx context.Context, key string) ([]byte, error) {
|
|
return r.dht.GetValue(ctx, key)
|
|
}
|
|
|
|
// Provide announces that this node can provide the given key.
|
|
func (r *RealDHT) Provide(ctx context.Context, key string) error {
|
|
return r.dht.Provide(ctx, key)
|
|
}
|
|
|
|
// FindProviders locates peers that can provide the specified key.
|
|
func (r *RealDHT) FindProviders(ctx context.Context, key string, limit int) ([]peer.AddrInfo, error) {
|
|
return r.dht.FindProviders(ctx, key, limit)
|
|
}
|
|
|
|
// GetStats exposes runtime metrics for the real DHT.
|
|
func (r *RealDHT) GetStats() DHTStats {
|
|
return r.dht.GetStats()
|
|
}
|
|
|
|
// Close releases resources associated with the DHT.
|
|
func (r *RealDHT) Close() error {
|
|
r.cancel()
|
|
|
|
var errs []error
|
|
if err := r.dht.Close(); err != nil {
|
|
errs = append(errs, err)
|
|
}
|
|
if err := r.host.Close(); err != nil {
|
|
errs = append(errs, err)
|
|
}
|
|
|
|
return errors.Join(errs...)
|
|
}
|