rpc.go

package dht

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/rpc"
	"time"

	"github.com/google/uuid"
)

var (
	errMismatchRequestID = errors.New("mismatch request id")
)

type nodeRPC struct {
	NodeId     string
	IpOrDomain string
	Port       uint32
	LastSeen   time.Time
}

// PingNode
type PingNode struct {
	Sender    *nodeRPC
	RequestID string
}

// FindNodeArgs
type FindNodeArgs struct {
	Sender       *nodeRPC
	RequestID    string
	TargetNodeID string
}

// FindNodeResult
type FindeNodeResult struct {
	Node         *nodeRPC
	RequestID    string
	ClosestNodes []*nodeRPC
}

// FindValueArgs
type FindValueArgs struct {
	Sender    *nodeRPC
	RequestID string
	Key       string
}

// FindValueResult
type FindValueResult struct {
	Node         *nodeRPC
	RequestID    string
	Value        []byte
	ClosestNodes []*nodeRPC
}

type rpcServer struct {
	ht dht
	s  *rpc.Server
}

// NewRPCServer
func NewRPCServer(kademlia dht) *rpcServer {
	rs := &rpcServer{
		ht: kademlia,
		s:  rpc.NewServer(),
	}

	rs.s.Register(rs.FindNode)
	rs.s.Register(rs.PingNode)
	rs.s.Register(rs.FindValue)

	return rs
}

// Serve
func (rs *rpcServer) Serve(ctx context.Context, serverAddr string) error {
	addr, err := net.ResolveTCPAddr("tcp", serverAddr)
	if err != nil {
		return fmt.Errorf("failed to resolve tcp addr: %w", err)
	}

	lis, err := net.ListenTCP("tcp", addr)
	if err != nil {
		return fmt.Errorf("net.Listen: %w", err)
	}
	defer lis.Close()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
			// fallthrough
		}

		if err := lis.SetDeadline(time.Now().Add(time.Millisecond * 250)); err != nil {
			time.Sleep(time.Second)
			continue
		}

		conn, err := lis.Accept()
		if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
			// listener timeout continue loop
			continue
		} else if err != nil {
			return fmt.Errorf("lis.Accept: %w", err)
		}

		go rs.s.ServeConn(conn)
	}
}

// PingNode
func (rs *rpcServer) PingNode(args *PingNode, result *PingNode) error {

	n, err := senderToNode(args.Sender)
	if err != nil {
		return fmt.Errorf("")
	}

	if err := rs.ht.addNode(&n); err != nil {
		return fmt.Errorf("")
	}

	result = newPingResponse(rs.ht.getSelf(), args.RequestID)

	return nil
}

func pingNodeRPC(sender *node, target string) error {
	client, err := rpc.Dial("tcp", target)
	if err != nil {
		return fmt.Errorf("failed to dial target: %w", err)
	}
	defer client.Close()

	args := &PingNode{
		Sender:    nodeToSender(sender),
		RequestID: uuid.New().String(),
	}

	var result PingNode
	if err := client.Call("", args, &result); err != nil {
		return fmt.Errorf("")
	}

	if result.RequestID != args.RequestID {
		return errMismatchRequestID
	}

	return nil
}

// FindNode
func (rs *rpcServer) FindNode(args *FindNodeArgs, result *FindeNodeResult) error {

	targetID, err := nodeIDFromStr(args.TargetNodeID)
	if err != nil {
		return fmt.Errorf("")
	}

	closestNodes, err := rs.ht.findNode(targetID)
	if err != nil {
		return fmt.Errorf("")
	}

	result = newFindNodeResponse(closestNodes, rs.ht.getSelf(), args.RequestID)

	return nil
}

func findNodeRPC(n *node, targetID nodeID, target string) ([]*node, error) {
	var (
		discoveredContacts = make([]*node, 0)
	)

	client, err := rpc.Dial("tcp", target)
	if err != nil {
		return nil, fmt.Errorf("failed to dial target: %w", err)
	}
	defer client.Close()

	args := &FindNodeArgs{
		Sender:       nodeToSender(n),
		RequestID:    uuid.New().String(),
		TargetNodeID: targetID.toString(),
	}

	var result FindeNodeResult
	err = client.Call("", args, &result)
	if err != nil {
		return nil, fmt.Errorf("failed to execute find node: %w", err)
	}

	if result.RequestID != args.RequestID {
		return nil, errMismatchRequestID
	}

	for _, cn := range result.ClosestNodes {
		nodeID, err := nodeIDFromStr(cn.NodeId)
		if err != nil {
			return nil, fmt.Errorf("node id from string: %w", err)
		}
		discoveredContacts = append(discoveredContacts, &node{
			id:       nodeID,
			ipOrHost: cn.IpOrDomain,
			port:     cn.Port,
			lastSeen: cn.LastSeen,
		})
	}

	return discoveredContacts, nil
}

// FindValue
func (rs *rpcServer) FindValue(args *FindNodeArgs, result *FindValueResult) error {
	return nil
}

func findValueRPC(n *node, key, target string) ([]byte, []*node, error) {
	client, err := rpc.Dial("tcp", target)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to dial target: %w", err)
	}
	defer client.Close()

	args := &FindValueArgs{
		Sender:    nodeToSender(n),
		Key:       key,
		RequestID: uuid.New().String(),
	}

	var result FindValueResult
	if err := client.Call("", args, &result); err != nil {
		return nil, nil, fmt.Errorf("execute rpc call: %w", err)
	}

	if args.RequestID != result.RequestID {
		return nil, nil, errMismatchRequestID
	}

	if result.Value != nil {
		return result.Value, nil, nil
	}

	closestNodes := make([]*node, 0, len(result.ClosestNodes))
	for _, cn := range result.ClosestNodes {
		n, err := senderToNode(cn)
		if err != nil {
			return nil, nil, fmt.Errorf("sender to node: %w", err)
		}
		closestNodes = append(closestNodes, &n)
	}

	return nil, closestNodes, nil
}

func newPingResponse(n *node, requestID string) *PingNode {
	return &PingNode{
		Sender:    nodeToSender(n),
		RequestID: requestID,
	}
}

func newFindNodeResponse(ns []*node, n *node, requestID string) *FindeNodeResult {
	closestNodes := make([]*nodeRPC, 0, len(ns))
	for _, n := range ns {
		closestNodes = append(closestNodes, nodeToSender(n))
	}
	return &FindeNodeResult{
		Node:         nodeToSender(n),
		RequestID:    requestID,
		ClosestNodes: closestNodes,
	}
}

func nodeToSender(n *node) *nodeRPC {
	return &nodeRPC{
		NodeId:     n.id.toString(),
		IpOrDomain: n.ipOrHost,
		Port:       n.port,
		LastSeen:   n.lastSeen,
	}
}

func senderToNode(n *nodeRPC) (node, error) {
	nodeID, err := nodeIDFromStr(n.NodeId)
	if err != nil {
		return node{}, fmt.Errorf("node is from string: %w", err)
	}
	return node{
		id:       nodeID,
		ipOrHost: n.IpOrDomain,
		port:     n.Port,
		lastSeen: n.LastSeen,
	}, nil
}