methods.go

package client

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"os"
	"path"
	"path/filepath"
	"strings"
	"time"

	"github.com/ipfs-cluster/ipfs-cluster/api"

	files "github.com/ipfs/boxo/files"
	gopath "github.com/ipfs/boxo/path"
	peer "github.com/libp2p/go-libp2p/core/peer"

	"go.opencensus.io/trace"
)

// ID returns information about the cluster Peer.
func (c *defaultClient) ID(ctx context.Context) (api.ID, error) {
	ctx, span := trace.StartSpan(ctx, "client/ID")
	defer span.End()

	var id api.ID
	err := c.do(ctx, "GET", "/id", nil, nil, &id)
	return id, err
}

// Peers requests ID information for all cluster peers.
func (c *defaultClient) Peers(ctx context.Context, out chan<- api.ID) error {
	defer close(out)

	ctx, span := trace.StartSpan(ctx, "client/Peers")
	defer span.End()

	handler := func(dec *json.Decoder) error {
		var obj api.ID
		err := dec.Decode(&obj)
		if err != nil {
			return err
		}
		out <- obj
		return nil
	}

	return c.doStream(ctx, "GET", "/peers", nil, nil, handler)

}

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

// PeerAdd adds a new peer to the cluster.
func (c *defaultClient) PeerAdd(ctx context.Context, pid peer.ID) (api.ID, error) {
	ctx, span := trace.StartSpan(ctx, "client/PeerAdd")
	defer span.End()

	body := peerAddBody{pid.String()}

	var buf bytes.Buffer
	enc := json.NewEncoder(&buf)
	enc.Encode(body)

	var id api.ID
	err := c.do(ctx, "POST", "/peers", nil, &buf, &id)
	return id, err
}

// PeerRm removes a current peer from the cluster
func (c *defaultClient) PeerRm(ctx context.Context, id peer.ID) error {
	ctx, span := trace.StartSpan(ctx, "client/PeerRm")
	defer span.End()

	return c.do(ctx, "DELETE", fmt.Sprintf("/peers/%s", id), nil, nil, nil)
}

// Pin tracks a Cid with the given replication factor and a name for
// human-friendliness.
func (c *defaultClient) Pin(ctx context.Context, ci api.Cid, opts api.PinOptions) (api.Pin, error) {
	ctx, span := trace.StartSpan(ctx, "client/Pin")
	defer span.End()

	query, err := opts.ToQuery()
	if err != nil {
		return api.Pin{}, err
	}
	var pin api.Pin
	err = c.do(
		ctx,
		"POST",
		fmt.Sprintf(
			"/pins/%s?%s",
			ci.String(),
			query,
		),
		nil,
		nil,
		&pin,
	)
	return pin, err
}

// Unpin untracks a Cid from cluster.
func (c *defaultClient) Unpin(ctx context.Context, ci api.Cid) (api.Pin, error) {
	ctx, span := trace.StartSpan(ctx, "client/Unpin")
	defer span.End()
	var pin api.Pin
	err := c.do(ctx, "DELETE", fmt.Sprintf("/pins/%s", ci.String()), nil, nil, &pin)
	return pin, err
}

// PinPath allows to pin an element by the given IPFS path.
func (c *defaultClient) PinPath(ctx context.Context, path string, opts api.PinOptions) (api.Pin, error) {
	ctx, span := trace.StartSpan(ctx, "client/PinPath")
	defer span.End()

	var pin api.Pin
	ipfspath, err := gopath.NewPath(path)
	if err != nil {
		ipfspath, err = gopath.NewPath("/ipfs/" + path)
		if err != nil {
			return api.Pin{}, err
		}
	}
	query, err := opts.ToQuery()
	if err != nil {
		return api.Pin{}, err
	}
	err = c.do(
		ctx,
		"POST",
		fmt.Sprintf(
			"/pins%s?%s",
			ipfspath.String(),
			query,
		),
		nil,
		nil,
		&pin,
	)

	return pin, err
}

// UnpinPath allows to unpin an item by providing its IPFS path.
// It returns the unpinned api.Pin information of the resolved Cid.
func (c *defaultClient) UnpinPath(ctx context.Context, p string) (api.Pin, error) {
	ctx, span := trace.StartSpan(ctx, "client/UnpinPath")
	defer span.End()

	var pin api.Pin
	ipfspath, err := gopath.NewPath(p)
	if err != nil {
		ipfspath, err = gopath.NewPath("/ipfs/" + p)
		if err != nil {
			return api.Pin{}, err
		}
	}

	err = c.do(ctx, "DELETE", fmt.Sprintf("/pins%s", ipfspath.String()), nil, nil, &pin)
	return pin, err
}

// Allocations returns the consensus state listing all tracked items and
// the peers that should be pinning them.
func (c *defaultClient) Allocations(ctx context.Context, filter api.PinType, out chan<- api.Pin) error {
	defer close(out)

	ctx, span := trace.StartSpan(ctx, "client/Allocations")
	defer span.End()

	types := []api.PinType{
		api.DataType,
		api.MetaType,
		api.ClusterDAGType,
		api.ShardType,
	}

	var strFilter []string

	if filter == api.AllType {
		strFilter = []string{"all"}
	} else {
		for _, t := range types {
			if t&filter > 0 { // the filter includes this type
				strFilter = append(strFilter, t.String())
			}
		}
	}

	handler := func(dec *json.Decoder) error {
		var obj api.Pin
		err := dec.Decode(&obj)
		if err != nil {
			return err
		}
		out <- obj
		return nil
	}

	f := url.QueryEscape(strings.Join(strFilter, ","))
	return c.doStream(
		ctx,
		"GET",
		fmt.Sprintf("/allocations?filter=%s", f),
		nil,
		nil,
		handler)
}

// Allocation returns the current allocations for a given Cid.
func (c *defaultClient) Allocation(ctx context.Context, ci api.Cid) (api.Pin, error) {
	ctx, span := trace.StartSpan(ctx, "client/Allocation")
	defer span.End()

	var pin api.Pin
	err := c.do(ctx, "GET", fmt.Sprintf("/allocations/%s", ci.String()), nil, nil, &pin)
	return pin, err
}

// Status returns the current ipfs state for a given Cid. If local is true,
// the information affects only the current peer, otherwise the information
// is fetched from all cluster peers.
func (c *defaultClient) Status(ctx context.Context, ci api.Cid, local bool) (api.GlobalPinInfo, error) {
	ctx, span := trace.StartSpan(ctx, "client/Status")
	defer span.End()

	var gpi api.GlobalPinInfo
	err := c.do(
		ctx,
		"GET",
		fmt.Sprintf("/pins/%s?local=%t", ci.String(), local),
		nil,
		nil,
		&gpi,
	)
	return gpi, err
}

// StatusCids returns Status() information for the given Cids. If local is
// true, the information affects only the current peer, otherwise the
// information is fetched from all cluster peers.
func (c *defaultClient) StatusCids(ctx context.Context, cids []api.Cid, local bool, out chan<- api.GlobalPinInfo) error {
	return c.statusAllWithCids(ctx, api.TrackerStatusUndefined, cids, local, out)
}

// StatusAll gathers Status() for all tracked items. If a filter is
// provided, only entries matching the given filter statuses
// will be returned. A filter can be built by merging TrackerStatuses with
// a bitwise OR operation (st1 | st2 | ...). A "0" filter value (or
// api.TrackerStatusUndefined), means all.
func (c *defaultClient) StatusAll(ctx context.Context, filter api.TrackerStatus, local bool, out chan<- api.GlobalPinInfo) error {
	return c.statusAllWithCids(ctx, filter, nil, local, out)
}

func (c *defaultClient) statusAllWithCids(ctx context.Context, filter api.TrackerStatus, cids []api.Cid, local bool, out chan<- api.GlobalPinInfo) error {
	defer close(out)
	ctx, span := trace.StartSpan(ctx, "client/StatusAll")
	defer span.End()

	filterStr := ""
	if filter != api.TrackerStatusUndefined { // undefined filter means "all"
		filterStr = filter.String()
		if filterStr == "" {
			return errors.New("invalid filter value")
		}
	}

	cidsStr := make([]string, len(cids))
	for i, c := range cids {
		cidsStr[i] = c.String()
	}

	handler := func(dec *json.Decoder) error {
		var obj api.GlobalPinInfo
		err := dec.Decode(&obj)
		if err != nil {
			return err
		}
		out <- obj
		return nil
	}

	return c.doStream(
		ctx,
		"GET",
		fmt.Sprintf("/pins?local=%t&filter=%s&cids=%s",
			local, url.QueryEscape(filterStr), strings.Join(cidsStr, ",")),
		nil,
		nil,
		handler,
	)
}

// Recover retriggers pin or unpin ipfs operations for a Cid in error state.
// If local is true, the operation is limited to the current peer, otherwise
// it happens on every cluster peer.
func (c *defaultClient) Recover(ctx context.Context, ci api.Cid, local bool) (api.GlobalPinInfo, error) {
	ctx, span := trace.StartSpan(ctx, "client/Recover")
	defer span.End()

	var gpi api.GlobalPinInfo
	err := c.do(ctx, "POST", fmt.Sprintf("/pins/%s/recover?local=%t", ci.String(), local), nil, nil, &gpi)
	return gpi, err
}

// RecoverAll triggers Recover() operations on all tracked items. If local is
// true, the operation is limited to the current peer. Otherwise, it happens
// everywhere.
func (c *defaultClient) RecoverAll(ctx context.Context, local bool, out chan<- api.GlobalPinInfo) error {
	defer close(out)

	ctx, span := trace.StartSpan(ctx, "client/RecoverAll")
	defer span.End()

	handler := func(dec *json.Decoder) error {
		var obj api.GlobalPinInfo
		err := dec.Decode(&obj)
		if err != nil {
			return err
		}
		out <- obj
		return nil
	}

	return c.doStream(
		ctx,
		"POST",
		fmt.Sprintf("/pins/recover?local=%t", local),
		nil,
		nil,
		handler)
}

// Alerts returns information health events in the cluster (expired metrics
// etc.).
func (c *defaultClient) Alerts(ctx context.Context) ([]api.Alert, error) {
	ctx, span := trace.StartSpan(ctx, "client/Alert")
	defer span.End()

	var alerts []api.Alert
	err := c.do(ctx, "GET", "/health/alerts", nil, nil, &alerts)
	return alerts, err
}

// Version returns the ipfs-cluster peer's version.
func (c *defaultClient) Version(ctx context.Context) (api.Version, error) {
	ctx, span := trace.StartSpan(ctx, "client/Version")
	defer span.End()

	var ver api.Version
	err := c.do(ctx, "GET", "/version", nil, nil, &ver)
	return ver, err
}

// GetConnectGraph returns an ipfs-cluster connection graph.
// The serialized version, strings instead of pids, is returned
func (c *defaultClient) GetConnectGraph(ctx context.Context) (api.ConnectGraph, error) {
	ctx, span := trace.StartSpan(ctx, "client/GetConnectGraph")
	defer span.End()

	var graph api.ConnectGraph
	err := c.do(ctx, "GET", "/health/graph", nil, nil, &graph)
	return graph, err
}

// Metrics returns a map with the latest valid metrics of the given name
// for the current cluster peers.
func (c *defaultClient) Metrics(ctx context.Context, name string) ([]api.Metric, error) {
	ctx, span := trace.StartSpan(ctx, "client/Metrics")
	defer span.End()

	if name == "" {
		return nil, errors.New("bad metric name")
	}
	var metrics []api.Metric
	err := c.do(ctx, "GET", fmt.Sprintf("/monitor/metrics/%s", name), nil, nil, &metrics)
	return metrics, err
}

// MetricNames lists names of all metrics.
func (c *defaultClient) MetricNames(ctx context.Context) ([]string, error) {
	ctx, span := trace.StartSpan(ctx, "client/MetricNames")
	defer span.End()

	var metricsNames []string
	err := c.do(ctx, "GET", "/monitor/metrics", nil, nil, &metricsNames)
	return metricsNames, err
}

// RepoGC runs garbage collection on IPFS daemons of cluster peers and
// returns collected CIDs. If local is true, it would garbage collect
// only on contacted peer, otherwise on all peers' IPFS daemons.
func (c *defaultClient) RepoGC(ctx context.Context, local bool) (api.GlobalRepoGC, error) {
	ctx, span := trace.StartSpan(ctx, "client/RepoGC")
	defer span.End()

	var repoGC api.GlobalRepoGC
	err := c.do(
		ctx,
		"POST",
		fmt.Sprintf("/ipfs/gc?local=%t", local),
		nil,
		nil,
		&repoGC,
	)

	return repoGC, err
}

// WaitFor is a utility function that allows for a caller to wait until a CID
// status target is reached (as given in StatusFilterParams).
// It returns the final status for that CID and an error, if there was one.
//
// WaitFor works by calling Status() repeatedly and checking that returned
// peers have transitioned to the target TrackerStatus. It immediately returns
// an error when the an error is among the statuses (and an empty
// GlobalPinInfo).
//
// A special case exists for TrackerStatusPinned targets: in this case,
// TrackerStatusRemote statuses are ignored, so WaitFor will return when
// all Statuses are Pinned or Remote by default.
//
// The Limit parameter allows to specify finer-grained control to, for
// example, only wait until a number of peers reaches a status.
func WaitFor(ctx context.Context, c Client, fp StatusFilterParams) (api.GlobalPinInfo, error) {
	ctx, span := trace.StartSpan(ctx, "client/WaitFor")
	defer span.End()

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

	sf := newStatusFilter()

	go sf.pollStatus(ctx, c, fp)
	go sf.filter(ctx, fp)

	var status api.GlobalPinInfo

	for {
		select {
		case <-ctx.Done():
			return status, ctx.Err()
		case err := <-sf.Err:
			return status, err
		case st, ok := <-sf.Out:
			if !ok { // channel closed
				return status, nil
			}
			status = st
		}
	}
}

// StatusFilterParams contains the parameters required
// to filter a stream of status results.
type StatusFilterParams struct {
	Cid       api.Cid
	Local     bool // query status from the local peer only
	Target    api.TrackerStatus
	Limit     int // wait for N peers reaching status. 0 == all
	CheckFreq time.Duration
}

type statusFilter struct {
	In, Out chan api.GlobalPinInfo
	Done    chan struct{}
	Err     chan error
}

func newStatusFilter() *statusFilter {
	return &statusFilter{
		In:   make(chan api.GlobalPinInfo),
		Out:  make(chan api.GlobalPinInfo),
		Done: make(chan struct{}),
		Err:  make(chan error),
	}
}

func (sf *statusFilter) filter(ctx context.Context, fp StatusFilterParams) {
	defer close(sf.Done)
	defer close(sf.Out)

	for {
		select {
		case <-ctx.Done():
			sf.Err <- ctx.Err()
			return
		case gblPinInfo, more := <-sf.In:
			if !more {
				return
			}
			ok, err := statusReached(fp.Target, gblPinInfo, fp.Limit)
			if err != nil {
				sf.Err <- err
				return
			}

			sf.Out <- gblPinInfo
			if !ok {
				continue
			}
			return
		}
	}
}

func (sf *statusFilter) pollStatus(ctx context.Context, c Client, fp StatusFilterParams) {
	ticker := time.NewTicker(fp.CheckFreq)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			sf.Err <- ctx.Err()
			return
		case <-ticker.C:
			gblPinInfo, err := c.Status(ctx, fp.Cid, fp.Local)
			if err != nil {
				sf.Err <- err
				return
			}
			logger.Debugf("pollStatus: status: %#v", gblPinInfo)
			sf.In <- gblPinInfo
		case <-sf.Done:
			close(sf.In)
			return
		}
	}
}

func statusReached(target api.TrackerStatus, gblPinInfo api.GlobalPinInfo, limit int) (bool, error) {
	// Specific case: return error if there are errors
	for _, pinInfo := range gblPinInfo.PeerMap {
		switch pinInfo.Status {
		case api.TrackerStatusUndefined,
			api.TrackerStatusClusterError,
			api.TrackerStatusPinError,
			api.TrackerStatusUnpinError:
			return false, fmt.Errorf("error has occurred while attempting to reach status: %s", target.String())
		}
	}

	// Specific case: when limit it set, just count how many targets we
	// reached.
	if limit > 0 {
		total := 0
		for _, pinInfo := range gblPinInfo.PeerMap {
			if pinInfo.Status == target {
				total++
			}
		}
		return total >= limit, nil
	}

	// General case: all statuses should be the target.
	// Specific case: when looking for Pinned, ignore status remote.
	for _, pinInfo := range gblPinInfo.PeerMap {
		if pinInfo.Status == api.TrackerStatusRemote && target == api.TrackerStatusPinned {
			continue
		}
		if pinInfo.Status == target {
			continue
		}
		return false, nil
	}

	// All statuses are the target, as otherwise we would have returned
	// false.
	return true, nil
}

// logic drawn from go-ipfs-cmds/cli/parse.go: appendFile
func makeSerialFile(fpath string, params api.AddParams) (string, files.Node, error) {
	if fpath == "." {
		cwd, err := os.Getwd()
		if err != nil {
			return "", nil, err
		}
		cwd, err = filepath.EvalSymlinks(cwd)
		if err != nil {
			return "", nil, err
		}
		fpath = cwd
	}

	fpath = filepath.ToSlash(filepath.Clean(fpath))

	stat, err := os.Lstat(fpath)
	if err != nil {
		return "", nil, err
	}

	if stat.IsDir() {
		if !params.Recursive {
			return "", nil, fmt.Errorf("%s is a directory, but Recursive option is not set", fpath)
		}
	}

	sf, err := files.NewSerialFile(fpath, params.Hidden, stat)
	return path.Base(fpath), sf, err
}

// Add imports files to the cluster from the given paths. A path can
// either be a local filesystem location or an web url (http:// or https://).
// In the latter case, the destination will be downloaded with a GET request.
// The AddParams allow to control different options, like enabling the
// sharding the resulting DAG across the IPFS daemons of multiple cluster
// peers. The output channel will receive regular updates as the adding
// process progresses.
func (c *defaultClient) Add(
	ctx context.Context,
	paths []string,
	params api.AddParams,
	out chan<- api.AddedOutput,
) error {
	ctx, span := trace.StartSpan(ctx, "client/Add")
	defer span.End()

	addFiles := make([]files.DirEntry, len(paths))
	for i, p := range paths {
		u, err := url.Parse(p)
		if err != nil {
			close(out)
			return fmt.Errorf("error parsing path: %s", err)
		}
		var name string
		var addFile files.Node
		if strings.HasPrefix(u.Scheme, "http") {
			addFile = files.NewWebFile(u)
			name = path.Base(u.Path)
		} else {
			if params.NoCopy {
				close(out)
				return fmt.Errorf("nocopy option is only valid for URLs")
			}
			name, addFile, err = makeSerialFile(p, params)
			if err != nil {
				close(out)
				return err
			}
		}
		addFiles[i] = files.FileEntry(name, addFile)
	}

	sliceFile := files.NewSliceDirectory(addFiles)
	// If `form` is set to true, the multipart data will have
	// a Content-Type of 'multipart/form-data', if `form` is false,
	// the Content-Type will be 'multipart/mixed'.
	return c.AddMultiFile(ctx, files.NewMultiFileReader(sliceFile, true, false), params, out)
}

// AddMultiFile imports new files from a MultiFileReader. See Add().
func (c *defaultClient) AddMultiFile(
	ctx context.Context,
	multiFileR *files.MultiFileReader,
	params api.AddParams,
	out chan<- api.AddedOutput,
) error {
	ctx, span := trace.StartSpan(ctx, "client/AddMultiFile")
	defer span.End()

	defer close(out)

	headers := make(map[string]string)
	headers["Content-Type"] = "multipart/form-data; boundary=" + multiFileR.Boundary()

	// This method must run with StreamChannels set.
	params.StreamChannels = true
	queryStr, err := params.ToQueryString()
	if err != nil {
		return err
	}

	// our handler decodes an AddedOutput and puts it
	// in the out channel.
	handler := func(dec *json.Decoder) error {
		if out == nil {
			return nil
		}
		var obj api.AddedOutput
		err := dec.Decode(&obj)
		if err != nil {
			return err
		}
		out <- obj
		return nil
	}

	err = c.doStream(ctx,
		"POST",
		"/add?"+queryStr,
		headers,
		multiFileR,
		handler,
	)
	return err
}

func (c *defaultClient) Health(ctx context.Context) error {
	ctx, span := trace.StartSpan(ctx, "client/Health")
	defer span.End()
	err := c.do(ctx, "GET", "/health", nil, nil, nil)
	return err
}