ipfs-cluster/api/rest/restapi.go

// Package rest implements an IPFS Cluster API component. It provides
// a REST-ish API to interact with Cluster.
//
// The implented API is based on the common.API component (refer to module
// description there). The only thing this module does is to provide route
// handling for the otherwise common API component.
package rest

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math/rand"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/ipfs-cluster/ipfs-cluster/adder/adderutils"
	types "github.com/ipfs-cluster/ipfs-cluster/api"
	"github.com/ipfs-cluster/ipfs-cluster/api/common"

	logging "github.com/ipfs/go-log/v2"
	"github.com/libp2p/go-libp2p-core/host"
	peer "github.com/libp2p/go-libp2p-core/peer"
	rpc "github.com/libp2p/go-libp2p-gorpc"

	mux "github.com/gorilla/mux"
)

func init() {
	rand.Seed(time.Now().UnixNano())
}

var (
	logger    = logging.Logger("restapi")
	apiLogger = logging.Logger("restapilog")
)

type peerAddBody struct {
	PeerID string `json:"peer_id"`
}

// API implements the REST API Component.
// It embeds a common.API.
type API struct {
	*common.API

	rpcClient *rpc.Client
	config    *Config
}

// NewAPI creates a new REST API component.
func NewAPI(ctx context.Context, cfg *Config) (*API, error) {
	return NewAPIWithHost(ctx, cfg, nil)
}

// NewAPIWithHost creates a new REST API component using the given libp2p Host.
func NewAPIWithHost(ctx context.Context, cfg *Config, h host.Host) (*API, error) {
	api := API{
		config: cfg,
	}
	capi, err := common.NewAPIWithHost(ctx, &cfg.Config, h, api.routes)
	api.API = capi
	return &api, err
}

// Routes returns endpoints supported by this API.
func (api *API) routes(c *rpc.Client) []common.Route {
	api.rpcClient = c
	return []common.Route{
		{
			Name:        "ID",
			Method:      "GET",
			Pattern:     "/id",
			HandlerFunc: api.idHandler,
		},

		{
			Name:        "Version",
			Method:      "GET",
			Pattern:     "/version",
			HandlerFunc: api.versionHandler,
		},

		{
			Name:        "Peers",
			Method:      "GET",
			Pattern:     "/peers",
			HandlerFunc: api.peerListHandler,
		},
		{
			Name:        "PeerAdd",
			Method:      "POST",
			Pattern:     "/peers",
			HandlerFunc: api.peerAddHandler,
		},
		{
			Name:        "PeerRemove",
			Method:      "DELETE",
			Pattern:     "/peers/{peer}",
			HandlerFunc: api.peerRemoveHandler,
		},
		{
			Name:        "Add",
			Method:      "POST",
			Pattern:     "/add",
			HandlerFunc: api.addHandler,
		},
		{
			Name:        "Allocations",
			Method:      "GET",
			Pattern:     "/allocations",
			HandlerFunc: api.allocationsHandler,
		},
		{
			Name:        "Allocation",
			Method:      "GET",
			Pattern:     "/allocations/{hash}",
			HandlerFunc: api.allocationHandler,
		},
		{
			Name:        "StatusAll",
			Method:      "GET",
			Pattern:     "/pins",
			HandlerFunc: api.statusAllHandler,
		},
		{
			Name:        "Recover",
			Method:      "POST",
			Pattern:     "/pins/{hash}/recover",
			HandlerFunc: api.recoverHandler,
		},
		{
			Name:        "RecoverAll",
			Method:      "POST",
			Pattern:     "/pins/recover",
			HandlerFunc: api.recoverAllHandler,
		},
		{
			Name:        "Status",
			Method:      "GET",
			Pattern:     "/pins/{hash}",
			HandlerFunc: api.statusHandler,
		},
		{
			Name:        "Pin",
			Method:      "POST",
			Pattern:     "/pins/{hash}",
			HandlerFunc: api.pinHandler,
		},
		{
			Name:        "PinPath",
			Method:      "POST",
			Pattern:     "/pins/{keyType:ipfs|ipns|ipld}/{path:.*}",
			HandlerFunc: api.pinPathHandler,
		},
		{
			Name:        "Unpin",
			Method:      "DELETE",
			Pattern:     "/pins/{hash}",
			HandlerFunc: api.unpinHandler,
		},
		{
			Name:        "UnpinPath",
			Method:      "DELETE",
			Pattern:     "/pins/{keyType:ipfs|ipns|ipld}/{path:.*}",
			HandlerFunc: api.unpinPathHandler,
		},
		{
			Name:        "RepoGC",
			Method:      "POST",
			Pattern:     "/ipfs/gc",
			HandlerFunc: api.repoGCHandler,
		},
		{
			Name:        "ConnectionGraph",
			Method:      "GET",
			Pattern:     "/health/graph",
			HandlerFunc: api.graphHandler,
		},
		{
			Name:        "Alerts",
			Method:      "GET",
			Pattern:     "/health/alerts",
			HandlerFunc: api.alertsHandler,
		},
		{
			Name:        "Metrics",
			Method:      "GET",
			Pattern:     "/monitor/metrics/{name}",
			HandlerFunc: api.metricsHandler,
		},
		{
			Name:        "MetricNames",
			Method:      "GET",
			Pattern:     "/monitor/metrics",
			HandlerFunc: api.metricNamesHandler,
		},
		{
			Name:        "GetToken",
			Method:      "POST",
			Pattern:     "/token",
			HandlerFunc: api.GenerateTokenHandler,
		},
	}
}

func (api *API) idHandler(w http.ResponseWriter, r *http.Request) {
	var id types.ID
	err := api.rpcClient.CallContext(
		r.Context(),
		"",
		"Cluster",
		"ID",
		struct{}{},
		&id,
	)

	api.SendResponse(w, common.SetStatusAutomatically, err, &id)
}

func (api *API) versionHandler(w http.ResponseWriter, r *http.Request) {
	var v types.Version
	err := api.rpcClient.CallContext(
		r.Context(),
		"",
		"Cluster",
		"Version",
		struct{}{},
		&v,
	)

	api.SendResponse(w, common.SetStatusAutomatically, err, v)
}

func (api *API) graphHandler(w http.ResponseWriter, r *http.Request) {
	var graph types.ConnectGraph
	err := api.rpcClient.CallContext(
		r.Context(),
		"",
		"Cluster",
		"ConnectGraph",
		struct{}{},
		&graph,
	)
	api.SendResponse(w, common.SetStatusAutomatically, err, graph)
}

func (api *API) metricsHandler(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	name := vars["name"]

	var metrics []types.Metric
	err := api.rpcClient.CallContext(
		r.Context(),
		"",
		"PeerMonitor",
		"LatestMetrics",
		name,
		&metrics,
	)
	api.SendResponse(w, common.SetStatusAutomatically, err, metrics)
}

func (api *API) metricNamesHandler(w http.ResponseWriter, r *http.Request) {
	var metricNames []string
	err := api.rpcClient.CallContext(
		r.Context(),
		"",
		"PeerMonitor",
		"MetricNames",
		struct{}{},
		&metricNames,
	)
	api.SendResponse(w, common.SetStatusAutomatically, err, metricNames)
}

func (api *API) alertsHandler(w http.ResponseWriter, r *http.Request) {
	var alerts []types.Alert
	err := api.rpcClient.CallContext(
		r.Context(),
		"",
		"Cluster",
		"Alerts",
		struct{}{},
		&alerts,
	)
	api.SendResponse(w, common.SetStatusAutomatically, err, alerts)
}

func (api *API) addHandler(w http.ResponseWriter, r *http.Request) {
	reader, err := r.MultipartReader()
	if err != nil {
		api.SendResponse(w, http.StatusBadRequest, err, nil)
		return
	}

	params, err := types.AddParamsFromQuery(r.URL.Query())
	if err != nil {
		api.SendResponse(w, http.StatusBadRequest, err, nil)
		return
	}

	api.SetHeaders(w)

	// any errors sent as trailer
	adderutils.AddMultipartHTTPHandler(
		r.Context(),
		api.rpcClient,
		params,
		reader,
		w,
		nil,
	)
}

func (api *API) peerListHandler(w http.ResponseWriter, r *http.Request) {
	in := make(chan struct{})
	close(in)
	out := make(chan types.ID, common.StreamChannelSize)
	errCh := make(chan error, 1)
	go func() {
		defer close(errCh)

		errCh <- api.rpcClient.Stream(
			r.Context(),
			"",
			"Cluster",
			"Peers",
			in,
			out,
		)
	}()

	iter := func() (interface{}, bool, error) {
		p, ok := <-out
		return p, ok, nil
	}
	api.StreamResponse(w, iter, errCh)
}

func (api *API) peerAddHandler(w http.ResponseWriter, r *http.Request) {
	dec := json.NewDecoder(r.Body)
	defer r.Body.Close()

	var addInfo peerAddBody
	err := dec.Decode(&addInfo)
	if err != nil {
		api.SendResponse(w, http.StatusBadRequest, errors.New("error decoding request body"), nil)
		return
	}

	pid, err := peer.Decode(addInfo.PeerID)
	if err != nil {
		api.SendResponse(w, http.StatusBadRequest, errors.New("error decoding peer_id"), nil)
		return
	}

	var id types.ID
	err = api.rpcClient.CallContext(
		r.Context(),
		"",
		"Cluster",
		"PeerAdd",
		pid,
		&id,
	)
	api.SendResponse(w, common.SetStatusAutomatically, err, &id)
}

func (api *API) peerRemoveHandler(w http.ResponseWriter, r *http.Request) {
	if p := api.ParsePidOrFail(w, r); p != "" {
		err := api.rpcClient.CallContext(
			r.Context(),
			"",
			"Cluster",
			"PeerRemove",
			p,
			&struct{}{},
		)
		api.SendResponse(w, common.SetStatusAutomatically, err, nil)
	}
}

func (api *API) pinHandler(w http.ResponseWriter, r *http.Request) {
	if pin := api.ParseCidOrFail(w, r); pin.Defined() {
		api.config.Logger.Debugf("rest api pinHandler: %s", pin.Cid)
		// span.AddAttributes(trace.StringAttribute("cid", pin.Cid))
		var pinObj types.Pin
		err := api.rpcClient.CallContext(
			r.Context(),
			"",
			"Cluster",
			"Pin",
			pin,
			&pinObj,
		)
		api.SendResponse(w, common.SetStatusAutomatically, err, pinObj)
		api.config.Logger.Debug("rest api pinHandler done")
	}
}

func (api *API) unpinHandler(w http.ResponseWriter, r *http.Request) {
	if pin := api.ParseCidOrFail(w, r); pin.Defined() {
		api.config.Logger.Debugf("rest api unpinHandler: %s", pin.Cid)
		// span.AddAttributes(trace.StringAttribute("cid", pin.Cid))
		var pinObj types.Pin
		err := api.rpcClient.CallContext(
			r.Context(),
			"",
			"Cluster",
			"Unpin",
			pin,
			&pinObj,
		)
		api.SendResponse(w, common.SetStatusAutomatically, err, pinObj)
		api.config.Logger.Debug("rest api unpinHandler done")
	}
}

func (api *API) pinPathHandler(w http.ResponseWriter, r *http.Request) {
	var pin types.Pin
	if pinpath := api.ParsePinPathOrFail(w, r); pinpath.Defined() {
		api.config.Logger.Debugf("rest api pinPathHandler: %s", pinpath.Path)
		err := api.rpcClient.CallContext(
			r.Context(),
			"",
			"Cluster",
			"PinPath",
			pinpath,
			&pin,
		)

		api.SendResponse(w, common.SetStatusAutomatically, err, pin)
		api.config.Logger.Debug("rest api pinPathHandler done")
	}
}

func (api *API) unpinPathHandler(w http.ResponseWriter, r *http.Request) {
	var pin types.Pin
	if pinpath := api.ParsePinPathOrFail(w, r); pinpath.Defined() {
		api.config.Logger.Debugf("rest api unpinPathHandler: %s", pinpath.Path)
		err := api.rpcClient.CallContext(
			r.Context(),
			"",
			"Cluster",
			"UnpinPath",
			pinpath,
			&pin,
		)
		api.SendResponse(w, common.SetStatusAutomatically, err, pin)
		api.config.Logger.Debug("rest api unpinPathHandler done")
	}
}

func (api *API) allocationsHandler(w http.ResponseWriter, r *http.Request) {
	queryValues := r.URL.Query()
	filterStr := queryValues.Get("filter")
	var filter types.PinType
	for _, f := range strings.Split(filterStr, ",") {
		filter |= types.PinTypeFromString(f)
	}

	if filter == types.BadType {
		api.SendResponse(w, http.StatusBadRequest, errors.New("invalid filter value"), nil)
		return
	}

	in := make(chan struct{})
	close(in)

	out := make(chan types.Pin, common.StreamChannelSize)
	errCh := make(chan error, 1)

	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()

	go func() {
		defer close(errCh)

		errCh <- api.rpcClient.Stream(
			r.Context(),
			"",
			"Cluster",
			"Pins",
			in,
			out,
		)
	}()

	iter := func() (interface{}, bool, error) {
		var p types.Pin
		var ok bool
	iterloop:
		for {

			select {
			case <-ctx.Done():
				break iterloop
			case p, ok = <-out:
				if !ok {
					break iterloop
				}
				// this means we keep iterating if no filter
				// matched
				if filter == types.AllType || filter&p.Type > 0 {
					break iterloop
				}
			}
		}
		return p, ok, ctx.Err()
	}

	api.StreamResponse(w, iter, errCh)
}

func (api *API) allocationHandler(w http.ResponseWriter, r *http.Request) {
	if pin := api.ParseCidOrFail(w, r); pin.Defined() {
		var pinResp types.Pin
		err := api.rpcClient.CallContext(
			r.Context(),
			"",
			"Cluster",
			"PinGet",
			pin.Cid,
			&pinResp,
		)
		api.SendResponse(w, common.SetStatusAutomatically, err, pinResp)
	}
}

func (api *API) statusAllHandler(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()

	queryValues := r.URL.Query()
	if queryValues.Get("cids") != "" {
		api.statusCidsHandler(w, r)
		return
	}

	local := queryValues.Get("local")

	filterStr := queryValues.Get("filter")
	filter := types.TrackerStatusFromString(filterStr)
	// FIXME: This is a bit lazy, as "invalidxx,pinned" would result in a
	// valid "pinned" filter.
	if filter == types.TrackerStatusUndefined && filterStr != "" {
		api.SendResponse(w, http.StatusBadRequest, errors.New("invalid filter value"), nil)
		return
	}

	var iter common.StreamIterator
	in := make(chan types.TrackerStatus, 1)
	in <- filter
	close(in)
	errCh := make(chan error, 1)

	if local == "true" {
		out := make(chan types.PinInfo, common.StreamChannelSize)
		iter = func() (interface{}, bool, error) {
			select {
			case <-ctx.Done():
				return nil, false, ctx.Err()
			case p, ok := <-out:
				return p.ToGlobal(), ok, nil
			}
		}

		go func() {
			defer close(errCh)

			errCh <- api.rpcClient.Stream(
				r.Context(),
				"",
				"Cluster",
				"StatusAllLocal",
				in,
				out,
			)
		}()

	} else {
		out := make(chan types.GlobalPinInfo, common.StreamChannelSize)
		iter = func() (interface{}, bool, error) {
			select {
			case <-ctx.Done():
				return nil, false, ctx.Err()
			case p, ok := <-out:
				return p, ok, nil
			}
		}
		go func() {
			defer close(errCh)

			errCh <- api.rpcClient.Stream(
				r.Context(),
				"",
				"Cluster",
				"StatusAll",
				in,
				out,
			)
		}()
	}

	api.StreamResponse(w, iter, errCh)
}

// request statuses for multiple CIDs in parallel.
func (api *API) statusCidsHandler(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()

	queryValues := r.URL.Query()
	filterCidsStr := strings.Split(queryValues.Get("cids"), ",")
	var cids []types.Cid

	for _, cidStr := range filterCidsStr {
		c, err := types.DecodeCid(cidStr)
		if err != nil {
			api.SendResponse(w, http.StatusBadRequest, fmt.Errorf("error decoding Cid: %w", err), nil)
			return
		}
		cids = append(cids, c)
	}

	local := queryValues.Get("local")

	gpiCh := make(chan types.GlobalPinInfo, len(cids))
	errCh := make(chan error, len(cids))
	var wg sync.WaitGroup
	wg.Add(len(cids))

	// Close channel when done
	go func() {
		wg.Wait()
		close(errCh)
		close(gpiCh)
	}()

	if local == "true" {
		for _, ci := range cids {
			go func(c types.Cid) {
				defer wg.Done()
				var pinInfo types.PinInfo
				err := api.rpcClient.CallContext(
					ctx,
					"",
					"Cluster",
					"StatusLocal",
					c,
					&pinInfo,
				)
				if err != nil {
					errCh <- err
					return
				}
				gpiCh <- pinInfo.ToGlobal()
			}(ci)
		}
	} else {
		for _, ci := range cids {
			go func(c types.Cid) {
				defer wg.Done()
				var pinInfo types.GlobalPinInfo
				err := api.rpcClient.CallContext(
					ctx,
					"",
					"Cluster",
					"Status",
					c,
					&pinInfo,
				)
				if err != nil {
					errCh <- err
					return
				}
				gpiCh <- pinInfo
			}(ci)
		}
	}

	iter := func() (interface{}, bool, error) {
		gpi, ok := <-gpiCh
		return gpi, ok, nil
	}

	api.StreamResponse(w, iter, errCh)
}

func (api *API) statusHandler(w http.ResponseWriter, r *http.Request) {
	queryValues := r.URL.Query()
	local := queryValues.Get("local")

	if pin := api.ParseCidOrFail(w, r); pin.Defined() {
		if local == "true" {
			var pinInfo types.PinInfo
			err := api.rpcClient.CallContext(
				r.Context(),
				"",
				"Cluster",
				"StatusLocal",
				pin.Cid,
				&pinInfo,
			)
			api.SendResponse(w, common.SetStatusAutomatically, err, pinInfo.ToGlobal())
		} else {
			var pinInfo types.GlobalPinInfo
			err := api.rpcClient.CallContext(
				r.Context(),
				"",
				"Cluster",
				"Status",
				pin.Cid,
				&pinInfo,
			)
			api.SendResponse(w, common.SetStatusAutomatically, err, pinInfo)
		}
	}
}

func (api *API) recoverAllHandler(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithCancel(r.Context())
	defer cancel()

	queryValues := r.URL.Query()
	local := queryValues.Get("local")

	var iter common.StreamIterator
	in := make(chan struct{})
	close(in)
	errCh := make(chan error, 1)

	if local == "true" {
		out := make(chan types.PinInfo, common.StreamChannelSize)
		iter = func() (interface{}, bool, error) {
			select {
			case <-ctx.Done():
				return nil, false, ctx.Err()
			case p, ok := <-out:
				return p.ToGlobal(), ok, nil
			}
		}

		go func() {
			defer close(errCh)

			errCh <- api.rpcClient.Stream(
				r.Context(),
				"",
				"Cluster",
				"RecoverAllLocal",
				in,
				out,
			)
		}()

	} else {
		out := make(chan types.GlobalPinInfo, common.StreamChannelSize)
		iter = func() (interface{}, bool, error) {
			select {
			case <-ctx.Done():
				return nil, false, ctx.Err()
			case p, ok := <-out:
				return p, ok, nil
			}
		}
		go func() {
			defer close(errCh)

			errCh <- api.rpcClient.Stream(
				r.Context(),
				"",
				"Cluster",
				"RecoverAll",
				in,
				out,
			)
		}()
	}

	api.StreamResponse(w, iter, errCh)
}

func (api *API) recoverHandler(w http.ResponseWriter, r *http.Request) {
	queryValues := r.URL.Query()
	local := queryValues.Get("local")

	if pin := api.ParseCidOrFail(w, r); pin.Defined() {
		if local == "true" {
			var pinInfo types.PinInfo
			err := api.rpcClient.CallContext(
				r.Context(),
				"",
				"Cluster",
				"RecoverLocal",
				pin.Cid,
				&pinInfo,
			)
			api.SendResponse(w, common.SetStatusAutomatically, err, pinInfo.ToGlobal())
		} else {
			var pinInfo types.GlobalPinInfo
			err := api.rpcClient.CallContext(
				r.Context(),
				"",
				"Cluster",
				"Recover",
				pin.Cid,
				&pinInfo,
			)
			api.SendResponse(w, common.SetStatusAutomatically, err, pinInfo)
		}
	}
}

func (api *API) repoGCHandler(w http.ResponseWriter, r *http.Request) {
	queryValues := r.URL.Query()
	local := queryValues.Get("local")

	if local == "true" {
		var localRepoGC types.RepoGC
		err := api.rpcClient.CallContext(
			r.Context(),
			"",
			"Cluster",
			"RepoGCLocal",
			struct{}{},
			&localRepoGC,
		)

		api.SendResponse(w, common.SetStatusAutomatically, err, repoGCToGlobal(localRepoGC))
		return
	}

	var repoGC types.GlobalRepoGC
	err := api.rpcClient.CallContext(
		r.Context(),
		"",
		"Cluster",
		"RepoGC",
		struct{}{},
		&repoGC,
	)
	api.SendResponse(w, common.SetStatusAutomatically, err, repoGC)
}

func repoGCToGlobal(r types.RepoGC) types.GlobalRepoGC {
	return types.GlobalRepoGC{
		PeerMap: map[string]types.RepoGC{
			peer.Encode(r.Peer): r,
		},
	}
}