server.go

package main

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"log/slog"
	"net"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	ocprom "contrib.go.opencensus.io/exporter/prometheus"
	"github.com/CAFxX/httpcompression"
	sddaemon "github.com/coreos/go-systemd/v22/daemon"
	"github.com/felixge/httpsnoop"
	"github.com/ipfs/boxo/routing/http/server"
	logging "github.com/ipfs/go-log/v2"
	"github.com/libp2p/go-libp2p"
	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/routing"
	"github.com/libp2p/go-libp2p/gologshim"
	"github.com/libp2p/go-libp2p/p2p/net/connmgr"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/rs/cors"
	"go.opencensus.io/stats/view"
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

var logger = logging.Logger(name)

// DefaultRoutingTimeout bounds how long a /routing/v1 request may spend in the
// routers. It must stay below the timeout clients put on the whole request,
// otherwise a client gives up before someguy finishes and flushes, and the
// records someguy did resolve are lost. The reference client, Helia's
// delegated routing v1 client, aborts at 30s, and its clock starts before
// ours, so the gap here also has to cover the request's network latency.
const DefaultRoutingTimeout = 25 * time.Second

func init() {
	// Set go-log's slog handler as the application-wide default.
	// This ensures all slog-based logging uses go-log's formatting.
	slog.SetDefault(slog.New(logging.SlogHandler()))

	// Wire go-log's slog bridge to go-libp2p's gologshim.
	// This provides go-libp2p loggers with the "logger" attribute
	// for per-subsystem level control.
	gologshim.SetDefaultHandler(logging.SlogHandler())

	// setup opencensus -> prometheus forwarding for delegated routing metrics
	promRegistry, ok := prometheus.DefaultRegisterer.(*prometheus.Registry)
	if !ok {
		logger.Error("delegated routing metrics: error casting DefaultRegisterer")
		return
	}
	pe, err := ocprom.NewExporter(ocprom.Options{
		Namespace: "someguy",
		Registry:  promRegistry,
		OnError: func(err error) {
			logger.Errorf("ocprom error: %w", err)
		},
	})
	if err != nil {
		logger.Errorf("delegated routing metrics: error creating exporter: %w", err)
		return
	}
	view.RegisterExporter(pe)
	view.SetReportingPeriod(2 * time.Second)
}

// newCompressionAdapter builds the response compression middleware.
//
// Do not raise MinSize. The middleware defers the compress-or-not decision
// until it has buffered MinSize bytes, and until it decides, Flush is a no-op.
// NDJSON records are routinely smaller than the default 200 bytes, so a record
// that is ready to send sits in that buffer until a later record fills it or
// the handler returns. That turns /routing/v1 streaming into a single batch
// delivered at the end, which a client waiting on early results cannot use.
// Compressing from the first byte keeps the flush after every record working.
func newCompressionAdapter() (func(http.Handler) http.Handler, error) {
	return httpcompression.DefaultAdapter(httpcompression.MinSize(0))
}

func withRequestLogger(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		m := httpsnoop.CaptureMetrics(next, w, r)
		logger.Debugw(r.Method, "url", r.URL, "host", r.Host, "code", m.Code, "duration", m.Duration, "written", m.Written, "accept", r.Header.Get("Accept"), "ua", r.UserAgent(), "referer", r.Referer())
	})
}

const (
	// DefaultRecordsLimit caps results for `Accept: application/json`
	// requests. Matches the SHOULD-cap from HTTP Routing v1 section 4.1.5.
	DefaultRecordsLimit = 100
	// DefaultStreamingRecordsLimit caps results for `Accept:
	// application/x-ndjson` requests. Sits above the JSON cap so streaming
	// returns "more results" per HTTP Routing v1 section 4.1.5. Set
	// SOMEGUY_STREAMING_RECORDS_LIMIT=0 to disable the cap.
	DefaultStreamingRecordsLimit = 1000
)

type config struct {
	listenAddress               string
	dhtType                     string
	cachedAddrBook              bool
	cachedAddrBookActiveProbing bool
	cachedAddrBookRecentTTL     time.Duration
	cachedAddrBookMaxFindPeers  int
	routingTimeout              time.Duration
	dnsAddrResolution           DNSAddrResolution
	recordsLimit                int
	streamingRecordsLimit       int

	contentEndpoints       []string
	peerEndpoints          []string
	ipnsEndpoints          []string
	blockProviderEndpoints []string
	blockProviderPeerIDs   []string

	libp2pListenAddress []string
	connMgrLow          int
	connMgrHi           int
	connMgrGrace        time.Duration
	maxMemory           uint64
	maxFD               int

	tracingAuth      string
	samplingFraction float64

	autoConf autoConfConfig
}

func start(ctx context.Context, cfg *config) error {
	h, err := newHost(cfg)
	if err != nil {
		return err
	}

	autoConf, err := startAutoConf(ctx, cfg)
	if err != nil {
		logger.Error(err.Error())
	}

	bootstrapAddrInfos := getBootstrapPeerAddrInfos(cfg, autoConf)

	// Expand delegated routing endpoints and categorize by path
	if err = expandDelegatedRoutingEndpoints(cfg, autoConf); err != nil {
		return err
	}

	// Print delegated routing endpoints
	if len(cfg.contentEndpoints) > 0 {
		fmt.Printf("Delegated routing endpoints for /routing/v1/providers: %v\n", cfg.contentEndpoints)
	}
	if len(cfg.peerEndpoints) > 0 {
		fmt.Printf("Delegated routing endpoints for /routing/v1/peers: %v\n", cfg.peerEndpoints)
	}
	if len(cfg.ipnsEndpoints) > 0 {
		fmt.Printf("Delegated routing endpoints for /routing/v1/ipns: %v\n", cfg.ipnsEndpoints)
	}

	fmt.Printf("Someguy libp2p host listening on %v\n", h.Addrs())
	var dhtRouting routing.Routing
	switch cfg.dhtType {
	case "accelerated":
		wrappedDHT, err := newBundledDHT(h, bootstrapAddrInfos)
		if err != nil {
			return err
		}
		dhtRouting = wrappedDHT
	case "standard":
		standardDHT, err := dht.New(h, dht.Mode(dht.ModeClient), dht.BootstrapPeers(bootstrapAddrInfos...))
		if err != nil {
			return err
		}
		dhtRouting = standardDHT
	case "disabled":
	default:
		return fmt.Errorf("invalid dht type %s, must be one of [accelerated, standard, disabled]", cfg.dhtType)
	}

	var cachedAddrBook *cachedAddrBook

	if cfg.cachedAddrBook && dhtRouting != nil {
		fmt.Printf("Using cached address book to speed up provider discovery (active probing enabled: %t)\n", cfg.cachedAddrBookActiveProbing)
		opts := []AddrBookOption{}

		if cfg.cachedAddrBookRecentTTL > 0 {
			opts = append(opts, WithRecentlyConnectedTTL(cfg.cachedAddrBookRecentTTL))
		}

		if cfg.cachedAddrBookMaxFindPeers > 0 {
			opts = append(opts, WithMaxConcurrentFindPeers(cfg.cachedAddrBookMaxFindPeers))
		}

		opts = append(opts, WithActiveProbing(cfg.cachedAddrBookActiveProbing))

		// Let the cache fall back to the host peerstore, which the DHT
		// populates with provider addresses during FindProviders.
		opts = append(opts, WithHostPeerstore(h.Peerstore()))

		cachedAddrBook, err = newCachedAddrBook(opts...)
		if err != nil {
			return err
		}
		go cachedAddrBook.background(ctx, h)
	}

	var blockProviderRouters []router
	if len(cfg.blockProviderEndpoints) > 0 {
		if len(cfg.blockProviderPeerIDs) != len(cfg.blockProviderEndpoints) {
			return fmt.Errorf("number of block provider peer IDs must match number of endpoints")
		}
		for i, endpoint := range cfg.blockProviderEndpoints {
			p, err := peer.Decode(cfg.blockProviderPeerIDs[i])
			if err != nil {
				return fmt.Errorf("invalid peer ID %s: %w", cfg.blockProviderPeerIDs[i], err)
			}

			r, err := newHTTPBlockRouter(endpoint, p, nil)
			if err != nil {
				return err
			}
			blockProviderRouters = append(blockProviderRouters, composableRouter{providers: r})
		}
	}

	// Create deduplicated HTTP routers - one client per unique base URL
	providerHTTPRouters, peerHTTPRouters, ipnsHTTPRouters, err := createDelegatedHTTPRouters(cfg)
	if err != nil {
		return err
	}

	// Combine HTTP routers with DHT and additional routers
	var dnsAddr *dnsAddrResolver
	if cfg.dnsAddrResolution != DNSAddrResolutionNever {
		dnsAddr, err = newDNSAddrResolver(nil)
		if err != nil {
			return err
		}
		fmt.Printf("Resolving /dnsaddr provider addresses: %s\n", cfg.dnsAddrResolution)
	}

	crRouters := combineRouters(h, dhtRouting, cachedAddrBook, providerHTTPRouters, blockProviderRouters, dnsAddr, cfg.dnsAddrResolution)
	prRouters := combineRouters(h, dhtRouting, cachedAddrBook, peerHTTPRouters, nil, dnsAddr, cfg.dnsAddrResolution)
	ipnsRouters := combineRouters(h, dhtRouting, cachedAddrBook, ipnsHTTPRouters, nil, dnsAddr, cfg.dnsAddrResolution)

	// Create DHT router for GetClosestPeers endpoint
	var dhtRouters router
	if cachedAddrBook != nil && dhtRouting != nil {
		cachedRouter := NewCachedRouter(libp2pRouter{host: h, routing: dhtRouting}, cachedAddrBook)
		dhtRouters = sanitizeRouter{cachedRouter}
	} else if dhtRouting != nil {
		dhtRouters = sanitizeRouter{libp2pRouter{host: h, routing: dhtRouting}}
	}
	if dhtRouters != nil {
		// Peerstore addresses can carry /dnsaddr too (learned via identify),
		// so the closest-peers endpoint resolves like the other endpoints.
		dhtRouters = withDNSAddrResolution(dhtRouters, dnsAddr, cfg.dnsAddrResolution)
	}

	_, port, err := net.SplitHostPort(cfg.listenAddress)
	if err != nil {
		return err
	}

	tp, err := setupTracing(ctx, cfg.samplingFraction)
	if err != nil {
		return err
	}

	defer func() {
		_ = tp.Shutdown(ctx)
	}()

	handlerOpts := []server.Option{
		server.WithPrometheusRegistry(prometheus.DefaultRegisterer),
		server.WithRecordsLimit(cfg.recordsLimit),
		server.WithStreamingRecordsLimit(cfg.streamingRecordsLimit),
	}

	// A zero timeout would cancel every request before it reached a router, so
	// treat it as unset rather than passing it through.
	if cfg.routingTimeout > 0 {
		handlerOpts = append(handlerOpts, server.WithRoutingTimeout(cfg.routingTimeout))
	} else {
		handlerOpts = append(handlerOpts, server.WithRoutingTimeout(DefaultRoutingTimeout))
	}

	handler := server.Handler(&composableRouter{
		providers: crRouters,
		peers:     prRouters,
		ipns:      ipnsRouters,
		dht:       dhtRouters,
	}, handlerOpts...)

	// Record filter-addrs so the routers can see it. Must wrap the
	// /routing/v1 handler, whose request context is the one the routers get.
	handler = withAddrFilter(handler)

	// Add CORS.
	handler = cors.New(cors.Options{
		AllowedOrigins: []string{"*"},
		AllowedMethods: []string{http.MethodGet, http.MethodOptions, http.MethodPut},
		MaxAge:         600,
	}).Handler(handler)

	// Add compression.
	compress, err := newCompressionAdapter()
	if err != nil {
		return err
	}
	handler = compress(handler)

	// Add request logging.
	handler = withRequestLogger(handler)

	// Add request tracing
	handler = withTracingAndDebug(handler, cfg.tracingAuth)

	http.Handle("/", handler)

	http.Handle("/debug/metrics/prometheus", promhttp.Handler())
	http.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintf(w, "Client: %s\n", name)
		fmt.Fprintf(w, "Version: %s\n", version)
	})

	server := &http.Server{Addr: cfg.listenAddress, Handler: nil}
	quit := make(chan os.Signal, 3)
	var wg sync.WaitGroup
	wg.Add(1)

	fmt.Printf("Metrics endpoint: http://127.0.0.1:%s/debug/metrics/prometheus\n", port)
	fmt.Printf("Delegated Routing API on http://127.0.0.1:%s/routing/v1\n", port)

	go func() {
		defer wg.Done()
		err := server.ListenAndServe()
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			logger.Fatalf("Failed to start /routing/v1 server: %v", err)
			quit <- os.Interrupt
		}
	}()

	sddaemon.SdNotify(false, sddaemon.SdNotifyReady)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
	<-quit
	sddaemon.SdNotify(false, sddaemon.SdNotifyStopping)
	fmt.Printf("\nClosing /routing/v1 server...\n")

	// Attempt a graceful shutdown
	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	if err := server.Shutdown(ctx); err != nil {
		log.Fatalf("Graceful shutdown failed:%+v\n", err)
	}

	go server.Close()
	wg.Wait()

	// The DHT constructors stopped taking a context in
	// go-libp2p-kad-dht v0.42.0, so cancelling ctx no longer stops them.
	// Close blocks until their long-lived components have shut down.
	if closer, ok := dhtRouting.(io.Closer); ok {
		if err := closer.Close(); err != nil {
			logger.Errorw("closing DHT", "err", err)
		}
	}

	fmt.Println("Shutdown finished.")
	return nil
}

func newHost(cfg *config) (host.Host, error) {
	cmgr, err := connmgr.NewConnManager(cfg.connMgrLow, cfg.connMgrHi, connmgr.WithGracePeriod(cfg.connMgrGrace))
	if err != nil {
		return nil, err
	}

	rcmgr, err := makeResourceMgrs(cfg.maxMemory, cfg.maxFD, cfg.connMgrHi)
	if err != nil {
		return nil, err
	}

	opts := []libp2p.Option{
		libp2p.UserAgent("someguy/" + buildVersion()),
		libp2p.ConnectionManager(cmgr),
		libp2p.ResourceManager(rcmgr),
		libp2p.NATPortMap(),
		libp2p.DefaultTransports,
		libp2p.DefaultMuxers,
		libp2p.EnableHolePunching(),
	}

	if len(cfg.libp2pListenAddress) == 0 {
		// Note: because the transports are set above we must also set the listen addresses
		// We need to set listen addresses in order for hole punching to work
		opts = append(opts, libp2p.DefaultListenAddrs)
	} else {
		opts = append(opts, libp2p.ListenAddrStrings(cfg.libp2pListenAddress...))
	}

	h, err := libp2p.New(opts...)
	if err != nil {
		return nil, err
	}

	return h, nil
}

// combineRouters combines delegated HTTP routers with DHT and additional routers.
// It no longer creates HTTP clients (that's done in createDelegatedHTTPRouters).
func combineRouters(h host.Host, dht routing.Routing, cachedAddrBook *cachedAddrBook, delegatedRouters, additionalRouters []router, dnsAddr *dnsAddrResolver, dnsAddrMode DNSAddrResolution) router {
	var dhtRouter router

	if cachedAddrBook != nil {
		cachedRouter := NewCachedRouter(libp2pRouter{host: h, routing: dht}, cachedAddrBook)
		dhtRouter = sanitizeRouter{cachedRouter}
	} else if dht != nil {
		dhtRouter = sanitizeRouter{libp2pRouter{host: h, routing: dht}}
	}

	if len(delegatedRouters) == 0 && len(additionalRouters) == 0 {
		if dhtRouter == nil {
			return composableRouter{}
		}
		return withDNSAddrResolution(dhtRouter, dnsAddr, dnsAddrMode)
	}

	var routers []router
	routers = append(routers, delegatedRouters...)
	if dhtRouter != nil {
		routers = append(routers, dhtRouter)
	}
	routers = append(routers, additionalRouters...)

	// Resolution wraps the composed router rather than sitting beside
	// sanitizeRouter, because /dnsaddr records reach someguy from the delegated
	// HTTP routers, which sanitizeRouter does not cover.
	return withDNSAddrResolution(parallelRouter{routers: routers}, dnsAddr, dnsAddrMode)
}

func withDNSAddrResolution(r router, resolver *dnsAddrResolver, mode DNSAddrResolution) router {
	if resolver == nil || mode == DNSAddrResolutionNever {
		return r
	}
	return dnsAddrRouter{router: r, resolver: resolver, mode: mode}
}

func withTracingAndDebug(next http.Handler, authToken string) http.Handler {
	next = otelhttp.NewHandler(next, "someguy.request")

	// Remove tracing and cache skipping headers if not authorized
	return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		// Disable tracing/debug headers if auth token missing or invalid
		if authToken == "" || request.Header.Get("Authorization") != authToken {
			if request.Header.Get("Traceparent") != "" {
				request.Header.Del("Traceparent")
			}
			if request.Header.Get("Tracestate") != "" {
				request.Header.Del("Tracestate")
			}
		}

		next.ServeHTTP(writer, request)
	})
}