factory.go

package server

import (
	"context"
	"fmt"
	"net/netip"
	"time"

	"github.com/bitmagnet-io/bitmagnet/internal/boilerplate/lazy"
	"github.com/bitmagnet-io/bitmagnet/internal/concurrency"
	"github.com/bitmagnet-io/bitmagnet/internal/protocol/dht"
	"github.com/bitmagnet-io/bitmagnet/internal/protocol/dht/responder"
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/fx"
	"go.uber.org/zap"
	"golang.org/x/time/rate"
)

type Params struct {
	fx.In
	Config    Config
	Responder responder.Responder
	Logger    *zap.SugaredLogger
}

type Result struct {
	fx.Out
	Server            lazy.Lazy[Server]
	LastResponses     *concurrency.AtomicValue[LastResponses] `name:"dht_server_last_responses"`
	AppHook           fx.Hook                                 `group:"app_hooks"`
	QueryDuration     prometheus.Collector                    `group:"prometheus_collectors"`
	QuerySuccessTotal prometheus.Collector                    `group:"prometheus_collectors"`
	QueryErrorTotal   prometheus.Collector                    `group:"prometheus_collectors"`
	QueryConcurrency  prometheus.Collector                    `group:"prometheus_collectors"`
}

const namespace = "bitmagnet"
const subsystem = "dht_server"

func New(p Params) Result {
	lastResponses := &concurrency.AtomicValue[LastResponses]{}
	collector := newPrometheusCollector()
	addr, err := netip.ParseAddr(p.Config.Addr)
	socket_ip_type := 4

	if err != nil {
		addr = netip.IPv4Unspecified()
	}
	if addr.Is4() {
		socket_ip_type = 4
	}
	if addr.Is6() || addr.Is4In6() {
		socket_ip_type = 6
	}
	ls := lazy.New(func() (Server, error) {
		s := queryLimiter{
			server: prometheusServerWrapper{
				prometheusCollector: collector,
				server: healthCollector{
					baseServer: &server{
						stopped:          make(chan struct{}),
						localAddr:        netip.AddrPortFrom(addr, p.Config.Port),
						socket:           NewSocket(socket_ip_type),
						queries:          make(map[string]chan dht.RecvMsg),
						queryTimeout:     p.Config.QueryTimeout,
						responder:        p.Responder,
						responderTimeout: time.Second * 5,
						idIssuer:         &variantIdIssuer{},
						logger:           p.Logger.Named(subsystem),
					},
					lastResponses: lastResponses,
				},
			},
			queryLimiter: concurrency.NewKeyedLimiter(rate.Every(time.Second), 4, 1000, time.Second*20),
		}
		if err := s.start(); err != nil {
			return nil, fmt.Errorf("could not start server: %w", err)
		}
		return s, nil
	})
	return Result{
		Server: ls,
		AppHook: fx.Hook{
			OnStop: func(context.Context) error {
				return ls.IfInitialized(func(s Server) error {
					s.stop()
					return nil
				})
			},
		},
		LastResponses:     lastResponses,
		QueryDuration:     collector.queryDuration,
		QuerySuccessTotal: collector.querySuccessTotal,
		QueryErrorTotal:   collector.queryErrorTotal,
		QueryConcurrency:  collector.queryConcurrency,
	}
}