feat: routing/http/server supports /routing/v1/peers

Author: hacdias

routing/http/client/client.go

@@ -192,7 +192,7 @@ func (c *client) GetProviders(ctx context.Context, key cid.Cid) (provs iter.Resu
 		it = iter.ToResultIter(sliceIt)
 	case mediaTypeNDJSON:
 		skipBodyClose = true
-		it = ndjson.NewProvidersResponseIter(resp.Body)
+		it = ndjson.NewRecordsIter(resp.Body)
 	default:
 		logger.Errorw("unknown media type", "MediaType", mediaType, "ContentType", respContentType)
 		return nil, errors.New("unknown content type")

routing/http/client/client_test.go

@@ -35,6 +35,11 @@ func (m *mockContentRouter) GetProviders(ctx context.Context, key cid.Cid, limit
 	return args.Get(0).(iter.ResultIter[types.Record]), args.Error(1)
 }
 
+func (m *mockContentRouter) GetPeers(ctx context.Context, pid peer.ID, limit int) (iter.ResultIter[types.Record], error) {
+	args := m.Called(ctx, pid, limit)
+	return args.Get(0).(iter.ResultIter[types.Record]), args.Error(1)
+}
+
 func (m *mockContentRouter) GetIPNSRecord(ctx context.Context, name ipns.Name) (*ipns.Record, error) {
 	args := m.Called(ctx, name)
 	return args.Get(0).(*ipns.Record), args.Error(1)

routing/http/server/server.go

@@ -19,6 +19,7 @@ import (
 	"github.com/ipfs/boxo/routing/http/types/iter"
 	jsontypes "github.com/ipfs/boxo/routing/http/types/json"
 	"github.com/ipfs/go-cid"
+	"github.com/libp2p/go-libp2p/core/peer"
 
 	logging "github.com/ipfs/go-log/v2"
 )
@@ -37,6 +38,7 @@ var logger = logging.Logger("service/server/delegatedrouting")
 
 const (
 	GetProvidersPath  = "/routing/v1/providers/{cid}"
+	GetPeersPath      = "/routing/v1/peers/{peer-id}"
 	GetIPNSRecordPath = "/routing/v1/ipns/{cid}"
 )
 
@@ -50,6 +52,10 @@ type ContentRouter interface {
 	// Limit indicates the maximum amount of results to return; 0 means unbounded.
 	GetProviders(ctx context.Context, key cid.Cid, limit int) (iter.ResultIter[types.Record], error)
 
+	// GetPeers searches for peers who have the provided [peer.ID].
+	// Limit indicates the maximum amount of results to return; 0 means unbounded.
+	GetPeers(ctx context.Context, pid peer.ID, limit int) (iter.ResultIter[types.Record], error)
+
 	// GetIPNSRecord searches for an [ipns.Record] for the given [ipns.Name].
 	GetIPNSRecord(ctx context.Context, name ipns.Name) (*ipns.Record, error)
 
@@ -96,8 +102,10 @@ func Handler(svc ContentRouter, opts ...Option) http.Handler {
 
 	r := mux.NewRouter()
 	r.HandleFunc(GetProvidersPath, server.getProviders).Methods(http.MethodGet)
+	r.HandleFunc(GetPeersPath, server.getPeers).Methods(http.MethodGet)
 	r.HandleFunc(GetIPNSRecordPath, server.getIPNSRecord).Methods(http.MethodGet)
 	r.HandleFunc(GetIPNSRecordPath, server.putIPNSRecord).Methods(http.MethodPut)
+
 	return r
 }
 
@@ -108,6 +116,43 @@ type server struct {
 	streamingRecordsLimit int
 }
 
+func (s *server) detectResponseType(r *http.Request) (string, error) {
+	var (
+		supportsNDJSON bool
+		supportsJSON   bool
+
+		acceptHeaders = r.Header.Values("Accept")
+	)
+
+	if len(acceptHeaders) == 0 {
+		return mediaTypeJSON, nil
+	}
+
+	for _, acceptHeader := range acceptHeaders {
+		for _, accept := range strings.Split(acceptHeader, ",") {
+			mediaType, _, err := mime.ParseMediaType(accept)
+			if err != nil {
+				return "", fmt.Errorf("unable to parse Accept header: %w", err)
+			}
+
+			switch mediaType {
+			case mediaTypeJSON, mediaTypeWildcard:
+				supportsJSON = true
+			case mediaTypeNDJSON:
+				supportsNDJSON = true
+			}
+		}
+	}
+
+	if supportsNDJSON && !s.disableNDJSON {
+		return mediaTypeNDJSON, nil
+	} else if supportsJSON {
+		return mediaTypeJSON, nil
+	} else {
+		return "", errors.New("no supported content types")
+	}
+}
+
 func (s *server) getProviders(w http.ResponseWriter, httpReq *http.Request) {
 	vars := mux.Vars(httpReq)
 	cidStr := vars["cid"]
@@ -117,43 +162,23 @@ func (s *server) getProviders(w http.ResponseWriter, httpReq *http.Request) {
 		return
 	}
 
-	var handlerFunc func(w http.ResponseWriter, provIter iter.ResultIter[types.Record])
+	mediaType, err := s.detectResponseType(httpReq)
+	if err != nil {
+		writeErr(w, "GetProviders", http.StatusBadRequest, err)
+		return
+	}
 
-	var supportsNDJSON bool
-	var supportsJSON bool
-	var recordsLimit int
-	acceptHeaders := httpReq.Header.Values("Accept")
-	if len(acceptHeaders) == 0 {
+	var (
+		handlerFunc  func(w http.ResponseWriter, provIter iter.ResultIter[types.Record])
+		recordsLimit int
+	)
+
+	if mediaType == mediaTypeNDJSON {
+		handlerFunc = s.getProvidersNDJSON
+		recordsLimit = s.streamingRecordsLimit
+	} else {
 		handlerFunc = s.getProvidersJSON
 		recordsLimit = s.recordsLimit
-	} else {
-		for _, acceptHeader := range acceptHeaders {
-			for _, accept := range strings.Split(acceptHeader, ",") {
-				mediaType, _, err := mime.ParseMediaType(accept)
-				if err != nil {
-					writeErr(w, "GetProviders", http.StatusBadRequest, fmt.Errorf("unable to parse Accept header: %w", err))
-					return
-				}
-
-				switch mediaType {
-				case mediaTypeJSON, mediaTypeWildcard:
-					supportsJSON = true
-				case mediaTypeNDJSON:
-					supportsNDJSON = true
-				}
-			}
-		}
-
-		if supportsNDJSON && !s.disableNDJSON {
-			handlerFunc = s.getProvidersNDJSON
-			recordsLimit = s.streamingRecordsLimit
-		} else if supportsJSON {
-			handlerFunc = s.getProvidersJSON
-			recordsLimit = s.recordsLimit
-		} else {
-			writeErr(w, "GetProviders", http.StatusBadRequest, errors.New("no supported content types"))
-			return
-		}
 	}
 
 	provIter, err := s.svc.GetProviders(httpReq.Context(), cid, recordsLimit)
@@ -168,58 +193,82 @@ func (s *server) getProviders(w http.ResponseWriter, httpReq *http.Request) {
 func (s *server) getProvidersJSON(w http.ResponseWriter, provIter iter.ResultIter[types.Record]) {
 	defer provIter.Close()
 
-	var (
-		providers []types.Record
-		i         int
-	)
-
-	for provIter.Next() {
-		res := provIter.Val()
-		if res.Err != nil {
-			writeErr(w, "GetProviders", http.StatusInternalServerError, fmt.Errorf("delegate error on result %d: %w", i, res.Err))
-			return
-		}
-		providers = append(providers, res.Val)
-		i++
+	providers, err := iter.ReadAllResults(provIter)
+	if err != nil {
+		writeErr(w, "GetProviders", http.StatusInternalServerError, fmt.Errorf("delegate error: %w", err))
+		return
 	}
-	response := jsontypes.ProvidersResponse{Providers: providers}
-	writeJSONResult(w, "GetProviders", response)
+
+	writeJSONResult(w, "GetProviders", jsontypes.ProvidersResponse{
+		Providers: providers,
+	})
 }
 
 func (s *server) getProvidersNDJSON(w http.ResponseWriter, provIter iter.ResultIter[types.Record]) {
-	defer provIter.Close()
+	writeResultsIterNDJSON(w, provIter)
+}
 
-	w.Header().Set("Content-Type", mediaTypeNDJSON)
-	w.WriteHeader(http.StatusOK)
-	for provIter.Next() {
-		res := provIter.Val()
-		if res.Err != nil {
-			logger.Errorw("GetProviders ndjson iterator error", "Error", res.Err)
-			return
-		}
-		// don't use an encoder because we can't easily differentiate writer errors from encoding errors
-		b, err := drjson.MarshalJSONBytes(res.Val)
-		if err != nil {
-			logger.Errorw("GetProviders ndjson marshal error", "Error", err)
-			return
-		}
+func (s *server) getPeers(w http.ResponseWriter, r *http.Request) {
+	pidStr := mux.Vars(r)["peer-id"]
 
-		_, err = w.Write(b)
-		if err != nil {
-			logger.Warn("GetProviders ndjson write error", "Error", err)
-			return
-		}
+	// pidStr must be in CIDv1 format. Therefore, use [cid.Decode]. We can't use
+	// [peer.Decode] because that would allow other formats to pass through.
+	cid, err := cid.Decode(pidStr)
+	if err != nil {
+		writeErr(w, "GetPeers", http.StatusBadRequest, fmt.Errorf("unable to parse peer ID: %w", err))
+		return
+	}
 
-		_, err = w.Write([]byte{'\n'})
-		if err != nil {
-			logger.Warn("GetProviders ndjson write error", "Error", err)
-			return
-		}
+	pid, err := peer.FromCid(cid)
+	if err != nil {
+		writeErr(w, "GetPeers", http.StatusBadRequest, fmt.Errorf("unable to parse peer ID: %w", err))
+		return
+	}
 
-		if f, ok := w.(http.Flusher); ok {
-			f.Flush()
-		}
+	mediaType, err := s.detectResponseType(r)
+	if err != nil {
+		writeErr(w, "GetPeers", http.StatusBadRequest, err)
+		return
+	}
+
+	var (
+		handlerFunc  func(w http.ResponseWriter, provIter iter.ResultIter[types.Record])
+		recordsLimit int
+	)
+
+	if mediaType == mediaTypeNDJSON {
+		handlerFunc = s.getPeersNDJSON
+		recordsLimit = s.streamingRecordsLimit
+	} else {
+		handlerFunc = s.getPeersJSON
+		recordsLimit = s.recordsLimit
+	}
+
+	provIter, err := s.svc.GetPeers(r.Context(), pid, recordsLimit)
+	if err != nil {
+		writeErr(w, "GetPeers", http.StatusInternalServerError, fmt.Errorf("delegate error: %w", err))
+		return
+	}
+
+	handlerFunc(w, provIter)
+}
+
+func (s *server) getPeersJSON(w http.ResponseWriter, peersIter iter.ResultIter[types.Record]) {
+	defer peersIter.Close()
+
+	peers, err := iter.ReadAllResults(peersIter)
+	if err != nil {
+		writeErr(w, "GetPeers", http.StatusInternalServerError, fmt.Errorf("delegate error: %w", err))
+		return
 	}
+
+	writeJSONResult(w, "GetPeers", jsontypes.PeersResponse{
+		Peers: peers,
+	})
+}
+
+func (s *server) getPeersNDJSON(w http.ResponseWriter, peersIter iter.ResultIter[types.Record]) {
+	writeResultsIterNDJSON(w, peersIter)
 }
 
 func (s *server) getIPNSRecord(w http.ResponseWriter, r *http.Request) {
@@ -347,3 +396,40 @@ func writeErr(w http.ResponseWriter, method string, statusCode int, cause error)
 func logErr(method, msg string, err error) {
 	logger.Infow(msg, "Method", method, "Error", err)
 }
+
+func writeResultsIterNDJSON(w http.ResponseWriter, resultIter iter.ResultIter[types.Record]) {
+	defer resultIter.Close()
+
+	w.Header().Set("Content-Type", mediaTypeNDJSON)
+	w.WriteHeader(http.StatusOK)
+
+	for resultIter.Next() {
+		res := resultIter.Val()
+		if res.Err != nil {
+			logger.Errorw("ndjson iterator error", "Error", res.Err)
+			return
+		}
+		// don't use an encoder because we can't easily differentiate writer errors from encoding errors
+		b, err := drjson.MarshalJSONBytes(res.Val)
+		if err != nil {
+			logger.Errorw("ndjson marshal error", "Error", err)
+			return
+		}
+
+		_, err = w.Write(b)
+		if err != nil {
+			logger.Warn("ndjson write error", "Error", err)
+			return
+		}
+
+		_, err = w.Write([]byte{'\n'})
+		if err != nil {
+			logger.Warn("ndjson write error", "Error", err)
+			return
+		}
+
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+	}
+}