ipfshttp.go

// Package ipfshttp implements an IPFS Cluster IPFSConnector component. It
// uses the IPFS HTTP API to communicate to IPFS.
package ipfshttp

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/ipfs-cluster/ipfs-cluster/api"
	"github.com/ipfs-cluster/ipfs-cluster/observations"
	"github.com/tv42/httpunix"

	files "github.com/ipfs/boxo/files"
	gopath "github.com/ipfs/boxo/path"
	ipfspinner "github.com/ipfs/boxo/pinning/pinner"
	cid "github.com/ipfs/go-cid"
	logging "github.com/ipfs/go-log/v2"
	rpc "github.com/libp2p/go-libp2p-gorpc"
	peer "github.com/libp2p/go-libp2p/core/peer"
	manet "github.com/multiformats/go-multiaddr/net"
	"github.com/multiformats/go-multicodec"
	multihash "github.com/multiformats/go-multihash"

	"go.opencensus.io/plugin/ochttp"
	"go.opencensus.io/plugin/ochttp/propagation/tracecontext"
	"go.opencensus.io/stats"
	"go.opencensus.io/trace"
)

// DNSTimeout is used when resolving DNS multiaddresses in this module
var DNSTimeout = 5 * time.Second

var logger = logging.Logger("ipfshttp")

// Connector implements the IPFSConnector interface
// and provides a component which  is used to perform
// on-demand requests against the configured IPFS daemom
// (such as a pin request).
type Connector struct {
	// struct alignment! These fields must be up-front.
	updateMetricCount uint64
	ipfsPinCount      int64

	ctx    context.Context
	cancel func()
	ready  chan struct{}

	config      *Config
	nodeAddr    string
	nodeNetwork string

	rpcClient *rpc.Client
	rpcReady  chan struct{}

	client *http.Client // client to ipfs daemon

	failedRequests atomic.Uint64 // count failed requests.
	reqRateLimitCh chan struct{}

	shutdownLock sync.Mutex
	shutdown     bool
	wg           sync.WaitGroup
}

type ipfsError struct {
	path    string
	code    int
	Message string
}

func (ie ipfsError) Error() string {
	return fmt.Sprintf(
		"IPFS error (%s). Code: %d. Message: %s",
		ie.path,
		ie.code,
		ie.Message,
	)
}

type ipfsUnpinnedError ipfsError

func (unpinned ipfsUnpinnedError) Is(target error) bool {
	ierr, ok := target.(ipfsError)
	if !ok {
		return false
	}
	return strings.HasSuffix(ierr.Message, "not pinned")
}

func (unpinned ipfsUnpinnedError) Error() string {
	return ipfsError(unpinned).Error()
}

type ipfsIDResp struct {
	ID        string
	Addresses []string
}

type ipfsResolveResp struct {
	Path string
}

type ipfsRepoGCResp struct {
	Key   cid.Cid
	Error string
}

type ipfsPinsResp struct {
	Pins     []string
	Progress int
}

type ipfsSwarmPeersResp struct {
	Peers []ipfsPeer
}

type ipfsBlockPutResp struct {
	Key  api.Cid
	Size int
}

type ipfsPeer struct {
	Peer string
}

// NewConnector creates the component and leaves it ready to be started
func NewConnector(cfg *Config) (*Connector, error) {
	err := cfg.Validate()
	if err != nil {
		return nil, err
	}

	nodeNetwork, nodeAddr, err := manet.DialArgs(cfg.NodeAddr)
	if err != nil {
		return nil, err
	}

	c := &http.Client{} // timeouts are handled by context timeouts

	if nodeNetwork == "unix" {
		unixTransport := &httpunix.Transport{
			DialTimeout: time.Second,
		}
		unixTransport.RegisterLocation("ipfs", nodeAddr)
		t := &http.Transport{}
		t.RegisterProtocol(httpunix.Scheme, unixTransport)
		c.Transport = t
	}

	if cfg.Tracing {
		c.Transport = &ochttp.Transport{
			Base:           http.DefaultTransport,
			Propagation:    &tracecontext.HTTPFormat{},
			StartOptions:   trace.StartOptions{SpanKind: trace.SpanKindClient},
			FormatSpanName: func(req *http.Request) string { return req.Host + ":" + req.URL.Path + ":" + req.Method },
			NewClientTrace: ochttp.NewSpanAnnotatingClientTrace,
		}
	}

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

	ipfs := &Connector{
		ctx:            ctx,
		cancel:         cancel,
		ready:          make(chan struct{}),
		config:         cfg,
		nodeAddr:       nodeAddr,
		nodeNetwork:    nodeNetwork,
		rpcReady:       make(chan struct{}, 1),
		reqRateLimitCh: make(chan struct{}),
		client:         c,
	}

	initializeMetrics(ctx)

	go ipfs.rateLimiter()

	go ipfs.run()
	return ipfs, nil
}

func initializeMetrics(ctx context.Context) {
	// initialize metrics
	stats.Record(ctx, observations.PinsIpfsPins.M(0))
	stats.Record(ctx, observations.PinsPinAdd.M(0))
	stats.Record(ctx, observations.PinsPinAddError.M(0))
	stats.Record(ctx, observations.BlocksPut.M(0))
	stats.Record(ctx, observations.BlocksAddedSize.M(0))
	stats.Record(ctx, observations.BlocksAdded.M(0))
	stats.Record(ctx, observations.BlocksAddedError.M(0))
}

// rateLimiter issues ticks in the reqRateLimitCh that allow requests to
// proceed. See doPostCtx.
func (ipfs *Connector) rateLimiter() {
	isRateLimiting := false

	// TODO: The rate-limiter is configured to start rate-limiting after
	// 10 failed requests at a rate of 1 req/s. This should probably be
	// configurable.
	for {
		failed := ipfs.failedRequests.Load()
		switch {
		case failed == 0:
			if isRateLimiting {
				// This does not print always,
				// only when there were several requests
				// waiting to read.
				logger.Warning("Lifting up rate limit")
			}
			isRateLimiting = false
		case failed > 0 && failed <= 10:
			isRateLimiting = false
		case failed > 10:
			if !isRateLimiting {
				logger.Warning("Rate-limiting requests to 1req/s")
			}
			isRateLimiting = true
			time.Sleep(time.Second)
		}

		// Send tick
		select {
		case <-ipfs.ctx.Done():
			close(ipfs.reqRateLimitCh)
			return
		case ipfs.reqRateLimitCh <- struct{}{}:
			// note that the channel is unbuffered,
			// therefore we will sit here until a method
			// wants to read from us, and they don't if
			// failed == 0.
		}
	}
}

// connects all ipfs daemons when
// we receive the rpcReady signal.
func (ipfs *Connector) run() {
	<-ipfs.rpcReady

	// wait for IPFS to be available
	i := 0
	for {
		select {
		case <-ipfs.ctx.Done():
			return
		default:
		}
		i++
		_, err := ipfs.ID(ipfs.ctx)
		if err == nil {
			close(ipfs.ready)
			break
		}
		if i%10 == 0 {
			logger.Warningf("ipfs does not seem to be available after %d retries", i)
		}

		// Requests will be rate-limited when going faster.
		time.Sleep(time.Second)
	}

	// Do not shutdown while launching threads
	// -- prevents race conditions with ipfs.wg.
	ipfs.shutdownLock.Lock()
	defer ipfs.shutdownLock.Unlock()

	if ipfs.config.ConnectSwarmsDelay == 0 {
		return
	}

	// This runs ipfs swarm connect to the daemons of other cluster members
	ipfs.wg.Add(1)
	go func() {
		defer ipfs.wg.Done()

		// It does not hurt to wait a little bit. i.e. think cluster
		// peers which are started at the same time as the ipfs
		// daemon...
		tmr := time.NewTimer(ipfs.config.ConnectSwarmsDelay)
		defer tmr.Stop()
		select {
		case <-tmr.C:
			// do not hang this goroutine if this call hangs
			// otherwise we hang during shutdown
			go ipfs.ConnectSwarms(ipfs.ctx)
		case <-ipfs.ctx.Done():
			return
		}
	}()
}

// SetClient makes the component ready to perform RPC
// requests.
func (ipfs *Connector) SetClient(c *rpc.Client) {
	ipfs.rpcClient = c
	ipfs.rpcReady <- struct{}{}
}

// Shutdown stops any listeners and stops the component from taking
// any requests.
func (ipfs *Connector) Shutdown(ctx context.Context) error {
	_, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/Shutdown")
	defer span.End()

	ipfs.shutdownLock.Lock()
	defer ipfs.shutdownLock.Unlock()

	if ipfs.shutdown {
		logger.Debug("already shutdown")
		return nil
	}

	logger.Info("stopping IPFS Connector")

	ipfs.cancel()
	close(ipfs.rpcReady)

	ipfs.wg.Wait()
	ipfs.shutdown = true

	return nil
}

// Ready returns a channel which gets notified when a testing request to the
// IPFS daemon first succeeds.
func (ipfs *Connector) Ready(ctx context.Context) <-chan struct{} {
	return ipfs.ready
}

// ID performs an ID request against the configured
// IPFS daemon. It returns the fetched information.
// If the request fails, or the parsing fails, it
// returns an error.
func (ipfs *Connector) ID(ctx context.Context) (api.IPFSID, error) {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/ID")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()

	body, err := ipfs.postCtx(ctx, "id", "", nil)
	if err != nil {
		return api.IPFSID{}, err
	}

	var res ipfsIDResp
	err = json.Unmarshal(body, &res)
	if err != nil {
		return api.IPFSID{}, err
	}

	pID, err := peer.Decode(res.ID)
	if err != nil {
		return api.IPFSID{}, err
	}

	id := api.IPFSID{
		ID: pID,
	}

	mAddrs := make([]api.Multiaddr, len(res.Addresses))
	for i, strAddr := range res.Addresses {
		mAddr, err := api.NewMultiaddr(strAddr)
		if err != nil {
			logger.Warningf("cannot parse IPFS multiaddress: %s (%w)... ignoring", strAddr, err)
			continue
		}
		mAddrs[i] = mAddr
	}
	id.Addresses = mAddrs
	return id, nil
}

func pinArgs(maxDepth api.PinDepth) string {
	q := url.Values{}
	switch {
	case maxDepth < 0:
		q.Set("recursive", "true")
	case maxDepth == 0:
		q.Set("recursive", "false")
	default:
		q.Set("recursive", "true")
		q.Set("max-depth", strconv.Itoa(int(maxDepth)))
	}
	return q.Encode()
}

// Pin performs a pin request against the configured IPFS
// daemon.
func (ipfs *Connector) Pin(ctx context.Context, pin api.Pin) error {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/Pin")
	defer span.End()

	hash := pin.Cid
	maxDepth := pin.MaxDepth

	pinStatus, err := ipfs.PinLsCid(ctx, pin)
	if err != nil {
		return err
	}

	if pinStatus.IsPinned(maxDepth) {
		logger.Debug("IPFS object is already pinned: ", hash)
		return nil
	}

	// Call at the beginning of pinning to update pinqueue
	ipfs.updateInformerMetric(ctx)
	// Call at the end of pinning to update freespace
	defer ipfs.updateInformerMetric(ctx)

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

	// If the pin has origins, tell ipfs to connect to a maximum of 10.
	bound := len(pin.Origins)
	if bound > 10 {
		bound = 10
	}
	for _, orig := range pin.Origins[0:bound] {
		// do it in the background, ignoring errors.
		go func(o string) {
			logger.Debugf("swarm-connect to origin before pinning: %s", o)
			_, err := ipfs.postCtx(
				ctx,
				fmt.Sprintf("swarm/connect?arg=%s", o),
				"",
				nil,
			)
			if err != nil {
				logger.Debug(err)
				return
			}
			logger.Debugf("swarm-connect success to origin: %s", o)
		}(url.QueryEscape(orig.String()))
	}

	// If we have a pin-update, and the old object
	// is pinned recursively, then do pin/update.
	// Otherwise do a normal pin.
	if from := pin.PinUpdate; from.Defined() {
		fromPin := api.PinWithOpts(from, pin.PinOptions)
		pinStatus, _ := ipfs.PinLsCid(ctx, fromPin)
		if pinStatus.IsPinned(-1) { // pinned recursively.
			// As a side note, if PinUpdate == pin.Cid, we are
			// somehow pinning an already pinned thing and we'd
			// better use update for that
			return ipfs.pinUpdate(ctx, from, pin.Cid)
		}
	}

	// Pin request and timeout if there is no progress
	outPins := make(chan int)
	go func() {
		var lastProgress int
		lastProgressTime := time.Now()

		ticker := time.NewTicker(ipfs.config.PinTimeout)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				if time.Since(lastProgressTime) > ipfs.config.PinTimeout {
					// timeout request
					cancelRequest()
					return
				}
			case p := <-outPins:
				// ipfs will send status messages every second
				// or so but we need make sure there was
				// progress by looking at number of nodes
				// fetched.
				if p > lastProgress {
					lastProgress = p
					lastProgressTime = time.Now()
				}
			case <-ctx.Done():
				return
			}
		}
	}()

	stats.Record(ipfs.ctx, observations.PinsPinAdd.M(1))
	err = ipfs.pinProgress(ctx, hash, maxDepth, outPins)
	if err != nil {
		stats.Record(ipfs.ctx, observations.PinsPinAddError.M(1))
		return err
	}
	totalPins := atomic.AddInt64(&ipfs.ipfsPinCount, 1)
	stats.Record(ipfs.ctx, observations.PinsIpfsPins.M(totalPins))

	logger.Info("IPFS Pin request succeeded: ", hash)
	return nil
}

// pinProgress pins an item and sends fetched node's progress on a
// channel. Blocks until done or error. pinProgress will always close the out
// channel.  pinProgress will not block on sending to the channel if it is full.
func (ipfs *Connector) pinProgress(ctx context.Context, hash api.Cid, maxDepth api.PinDepth, out chan<- int) error {
	defer close(out)

	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/pinsProgress")
	defer span.End()

	pinArgs := pinArgs(maxDepth)
	path := fmt.Sprintf("pin/add?arg=%s&%s&progress=true", hash, pinArgs)

	body, err := ipfs.postCtxStreamResponse(ctx, path, "", nil)
	if err != nil {
		return err
	}
	defer body.Close()

	dec := json.NewDecoder(body)
	for {
		var pins ipfsPinsResp
		if err := dec.Decode(&pins); err != nil {
			// If we canceled the request we should tell the user
			// (in case dec.Decode() exited cleanly with an EOF).
			select {
			case <-ctx.Done():
				return ctx.Err()
			default:
				if err == io.EOF {
					return nil // clean exit. Pinned!
				}
				return err // error decoding
			}
		}

		select {
		case out <- pins.Progress:
		default:
		}
	}
}

func (ipfs *Connector) pinUpdate(ctx context.Context, from, to api.Cid) error {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/pinUpdate")
	defer span.End()

	path := fmt.Sprintf("pin/update?arg=%s&arg=%s&unpin=false", from, to)
	_, err := ipfs.postCtx(ctx, path, "", nil)
	if err != nil {
		return err
	}
	totalPins := atomic.AddInt64(&ipfs.ipfsPinCount, 1)
	stats.Record(ipfs.ctx, observations.PinsIpfsPins.M(totalPins))
	logger.Infof("IPFS Pin Update request succeeded. %s -> %s (unpin=false)", from, to)
	return nil
}

// Unpin performs an unpin request against the configured IPFS
// daemon.
func (ipfs *Connector) Unpin(ctx context.Context, hash api.Cid) error {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/Unpin")
	defer span.End()

	if ipfs.config.UnpinDisable {
		return errors.New("ipfs unpinning is disallowed by configuration on this peer")
	}

	// Unpinning doesn't free space and doesn't matter for pinqueue so not
	// really necessary to publish metrics.
	//defer ipfs.updateInformerMetric(ctx)

	path := fmt.Sprintf("pin/rm?arg=%s", hash)

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.UnpinTimeout)
	defer cancel()

	// We will call unpin in any case, if the CID is not pinned,
	// then we ignore the error (although this is a bit flaky).
	_, err := ipfs.postCtx(ctx, path, "", nil)
	if err != nil {
		ipfsErr, ok := err.(ipfsError)
		if !ok || ipfsErr.Message != ipfspinner.ErrNotPinned.Error() {
			return err
		}
		logger.Debug("IPFS object is already unpinned: ", hash)
		return nil
	}

	totalPins := atomic.AddInt64(&ipfs.ipfsPinCount, -1)
	stats.Record(ipfs.ctx, observations.PinsIpfsPins.M(totalPins))

	logger.Info("IPFS Unpin request succeeded:", hash)
	return nil
}

// PinLs performs a "pin ls --type typeFilter" request against the configured
// IPFS daemon and sends the results on the given channel. Returns when done.
func (ipfs *Connector) PinLs(ctx context.Context, typeFilters []string, out chan<- api.IPFSPinInfo) error {
	defer close(out)
	bodies := make([]io.ReadCloser, len(typeFilters))

	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/PinLs")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()

	var err error
	var totalPinCount int64
	defer func() {
		if err != nil {
			atomic.StoreInt64(&ipfs.ipfsPinCount, totalPinCount)
			stats.Record(ipfs.ctx, observations.PinsIpfsPins.M(totalPinCount))
		}
	}()

nextFilter:
	for i, typeFilter := range typeFilters {
		// Post and read streaming response
		path := "pin/ls?stream=true&type=" + typeFilter
		bodies[i], err = ipfs.postCtxStreamResponse(ctx, path, "", nil)
		if err != nil {
			return err
		}
		defer bodies[i].Close()

		dec := json.NewDecoder(bodies[i])

		for {
			select {
			case <-ctx.Done():
				err = fmt.Errorf("aborting pin/ls operation: %w", ctx.Err())
				logger.Error(err)
				return err
			default:
			}

			var ipfsPin api.IPFSPinInfo
			err = dec.Decode(&ipfsPin)
			if err == io.EOF {
				break nextFilter
			}
			if err != nil {
				err = fmt.Errorf("error decoding ipfs pin: %w", err)
				return err
			}

			select {
			case <-ctx.Done():
				err = fmt.Errorf("aborting pin/ls operation: %w", ctx.Err())
				logger.Error(err)
				return err
			case out <- ipfsPin:
				totalPinCount++
			}
		}
	}

	return nil
}

// PinLsCid performs a "pin ls <hash>" request. It will use "type=recursive" or
// "type=direct" (or other) depending on the given pin's MaxDepth setting.
// It returns an api.IPFSPinStatus for that hash.
func (ipfs *Connector) PinLsCid(ctx context.Context, pin api.Pin) (api.IPFSPinStatus, error) {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/PinLsCid")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()

	if !pin.Defined() {
		return api.IPFSPinStatusBug, errors.New("calling PinLsCid without a defined CID")
	}

	pinType := pin.MaxDepth.ToPinMode().String()
	lsPath := fmt.Sprintf("pin/ls?stream=true&arg=%s&type=%s", pin.Cid, pinType)
	body, err := ipfs.postCtxStreamResponse(ctx, lsPath, "", nil)
	if err != nil {
		if errors.Is(ipfsUnpinnedError{}, err) {
			return api.IPFSPinStatusUnpinned, nil
		}
		return api.IPFSPinStatusError, err
	}
	defer body.Close()

	var res api.IPFSPinInfo
	dec := json.NewDecoder(body)

	err = dec.Decode(&res)
	if err != nil {
		logger.Error("error parsing pin/ls?arg=cid response")
		return api.IPFSPinStatusError, err
	}

	return res.Type, nil
}

// ConnectSwarms requests the ipfs addresses of other peers and
// triggers ipfs swarm connect requests
func (ipfs *Connector) ConnectSwarms(ctx context.Context) error {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/ConnectSwarms")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()

	in := make(chan struct{})
	close(in)
	out := make(chan api.ID)
	go func() {
		err := ipfs.rpcClient.Stream(
			ctx,
			"",
			"Cluster",
			"Peers",
			in,
			out,
		)
		if err != nil {
			logger.Error(err)
		}
	}()

	for id := range out {
		ipfsID := id.IPFS
		if id.Error != "" || ipfsID.Error != "" {
			continue
		}
		for _, addr := range ipfsID.Addresses {
			// This is a best effort attempt
			// We ignore errors which happens
			// when passing in a bunch of addresses
			_, err := ipfs.postCtx(
				ctx,
				fmt.Sprintf("swarm/connect?arg=%s", url.QueryEscape(addr.String())),
				"",
				nil,
			)
			if err != nil {
				logger.Debug(err)
				continue
			}
			logger.Debugf("ipfs successfully connected to %s", addr)
		}
	}
	return nil
}

// ConfigKey fetches the IPFS daemon configuration and retrieves the value for
// a given configuration key. For example, "Datastore/StorageMax" will return
// the value for StorageMax in the Datastore configuration object.
func (ipfs *Connector) ConfigKey(keypath string) (interface{}, error) {
	ctx, cancel := context.WithTimeout(ipfs.ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()
	res, err := ipfs.postCtx(ctx, "config/show", "", nil)
	if err != nil {
		return nil, err
	}

	var cfg map[string]interface{}
	err = json.Unmarshal(res, &cfg)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	path := strings.SplitN(keypath, "/", 2)
	if len(path) == 0 {
		return nil, errors.New("cannot lookup without a path")
	}

	return getConfigValue(path, cfg)
}

func getConfigValue(path []string, cfg map[string]interface{}) (interface{}, error) {
	value, ok := cfg[path[0]]
	if !ok {
		return nil, errors.New("key not found in configuration")
	}

	if len(path) == 1 {
		return value, nil
	}

	switch v := value.(type) {
	case map[string]interface{}:
		return getConfigValue(path[1:], v)
	default:
		return nil, errors.New("invalid path")
	}
}

// RepoStat returns the DiskUsage and StorageMax repo/stat values from the
// ipfs daemon, in bytes, wrapped as an IPFSRepoStat object.
func (ipfs *Connector) RepoStat(ctx context.Context) (api.IPFSRepoStat, error) {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/RepoStat")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()
	res, err := ipfs.postCtx(ctx, "repo/stat?size-only=true", "", nil)
	if err != nil {
		return api.IPFSRepoStat{}, err
	}

	var stats api.IPFSRepoStat
	err = json.Unmarshal(res, &stats)
	if err != nil {
		logger.Error(err)
		return api.IPFSRepoStat{}, err
	}
	return stats, nil
}

// RepoGC performs a garbage collection sweep on the cluster peer's IPFS repo.
func (ipfs *Connector) RepoGC(ctx context.Context) (api.RepoGC, error) {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/RepoGC")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.RepoGCTimeout)
	defer cancel()

	body, err := ipfs.postCtxStreamResponse(ctx, "repo/gc?stream-errors=true", "", nil)
	if err != nil {
		return api.RepoGC{}, err
	}

	defer body.Close()

	// Freespace metric might have gone down, so update it at the end.
	defer ipfs.updateInformerMetric(ctx)

	dec := json.NewDecoder(body)
	repoGC := api.RepoGC{
		Keys: []api.IPFSRepoGC{},
	}
	for {
		resp := ipfsRepoGCResp{}

		if err := dec.Decode(&resp); err != nil {
			// If we canceled the request we should tell the user
			// (in case dec.Decode() exited cleanly with an EOF).
			select {
			case <-ctx.Done():
				return repoGC, ctx.Err()
			default:
				if err == io.EOF {
					return repoGC, nil // clean exit
				}
				logger.Error(err)
				return repoGC, err // error decoding
			}
		}

		repoGC.Keys = append(repoGC.Keys, api.IPFSRepoGC{Key: api.NewCid(resp.Key), Error: resp.Error})
	}
}

// Resolve accepts ipfs or ipns path and resolves it into a cid
func (ipfs *Connector) Resolve(ctx context.Context, path string) (api.Cid, error) {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/Resolve")
	defer span.End()

	validPath, err := gopath.NewPath(path)
	if err != nil {
		validPath, err = gopath.NewPath("/ipfs/" + path)
		if err != nil {
			logger.Error("could not parse path: " + err.Error())
			return api.CidUndef, err
		}
	}
	immPath, err := gopath.NewImmutablePath(validPath)
	if err == nil && len(immPath.Segments()) == 2 { // no need to resolve
		return api.NewCid(immPath.RootCid()), nil
	}

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()
	res, err := ipfs.postCtx(ctx, "resolve?arg="+url.QueryEscape(validPath.String()), "", nil)
	if err != nil {
		return api.CidUndef, err
	}

	var resp ipfsResolveResp
	err = json.Unmarshal(res, &resp)
	if err != nil {
		logger.Error("could not unmarshal response: " + err.Error())
		return api.CidUndef, err
	}

	respPath, err := gopath.NewPath(resp.Path)
	if err != nil {
		logger.Error("invalid path in response: " + err.Error())
		return api.CidUndef, err
	}

	respImmPath, err := gopath.NewImmutablePath(respPath)
	if err != nil {
		logger.Error("resolved path is mutable: " + err.Error())
		return api.CidUndef, err
	}
	return api.NewCid(respImmPath.RootCid()), nil
}

// SwarmPeers returns the peers currently connected to this ipfs daemon.
func (ipfs *Connector) SwarmPeers(ctx context.Context) ([]peer.ID, error) {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/SwarmPeers")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()

	res, err := ipfs.postCtx(ctx, "swarm/peers", "", nil)
	if err != nil {
		return nil, err
	}
	var peersRaw ipfsSwarmPeersResp
	err = json.Unmarshal(res, &peersRaw)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	swarm := make([]peer.ID, len(peersRaw.Peers))
	for i, p := range peersRaw.Peers {
		pID, err := peer.Decode(p.Peer)
		if err != nil {
			logger.Error(err)
			return swarm, err
		}
		swarm[i] = pID
	}
	return swarm, nil
}

// chanDirectory implements the files.Directory interface
type chanDirectory struct {
	iterator files.DirIterator
}

// Close is a no-op and it is not used.
func (cd *chanDirectory) Close() error {
	return nil
}

// not implemented, I think not needed for multipart.
func (cd *chanDirectory) Size() (int64, error) {
	return 0, nil
}

func (cd *chanDirectory) Entries() files.DirIterator {
	return cd.iterator
}

// Mode: return mode for directory, but unused.
func (cd *chanDirectory) Mode() os.FileMode {
	return os.ModeDir
}

// ModeTime: not implemented
func (cd *chanDirectory) ModTime() (mtime time.Time) {
	return time.UnixMilli(0)
}

// chanIterator implements the files.DirIterator interface.
type chanIterator struct {
	ctx    context.Context
	blocks <-chan api.NodeWithMeta

	current api.NodeWithMeta
	peeked  api.NodeWithMeta
	done    bool
	err     error

	seenMu sync.Mutex
	seen   map[string]int
}

func (ci *chanIterator) Name() string {
	if !ci.current.Cid.Defined() {
		return ""
	}
	return ci.current.Cid.String()
}

// return NewBytesFile.
// This function might and is actually called multiple times for the same node
// by the multifile Reader to send the multipart.
func (ci *chanIterator) Node() files.Node {
	if !ci.current.Cid.Defined() {
		return nil
	}
	logger.Debugf("it.Node(): %s", ci.current.Cid)
	return files.NewBytesFile(ci.current.Data)
}

// Seen returns whether we have seen a multihash. It keeps count so it will
// return true as many times as we have seen it.
func (ci *chanIterator) Seen(c api.Cid) bool {
	ci.seenMu.Lock()
	n, ok := ci.seen[string(c.Cid.Hash())]
	logger.Debugf("Seen(): %s, %d, %t", c, n, ok)
	if ok {
		if n == 1 {
			delete(ci.seen, string(c.Cid.Hash()))
		} else {
			ci.seen[string(c.Cid.Hash())] = n - 1
		}
	}
	ci.seenMu.Unlock()
	return ok
}

func (ci *chanIterator) Done() bool {
	return ci.done
}

// Peek reads one block from the channel but saves it so that Next also
// returns it.
func (ci *chanIterator) Peek() (api.NodeWithMeta, bool) {
	if ci.done {
		return api.NodeWithMeta{}, false
	}

	select {
	case <-ci.ctx.Done():
		return api.NodeWithMeta{}, false
	case next, ok := <-ci.blocks:
		if !ok {
			return api.NodeWithMeta{}, false
		}
		ci.peeked = next
		return next, true
	}
}

func (ci *chanIterator) Next() bool {
	if ci.done {
		return false
	}

	seeBlock := func(b api.NodeWithMeta) {
		ci.seenMu.Lock()
		ci.seen[string(b.Cid.Hash())]++
		ci.seenMu.Unlock()
		stats.Record(ci.ctx, observations.BlocksAdded.M(1))
		stats.Record(ci.ctx, observations.BlocksAddedSize.M(int64(len(b.Data))))

	}

	if ci.peeked.Cid.Defined() {
		ci.current = ci.peeked
		ci.peeked = api.NodeWithMeta{}
		seeBlock(ci.current)
		return true
	}
	select {
	case <-ci.ctx.Done():
		ci.done = true
		ci.err = ci.ctx.Err()
		return false
	case next, ok := <-ci.blocks:
		if !ok {
			ci.done = true
			return false
		}
		// Record that we have seen this block. This has to be done
		// here, not in Node() as Node() is called multiple times per
		// block received.
		logger.Debugf("it.Next() %s", next.Cid)
		ci.current = next
		seeBlock(ci.current)
		return true
	}
}

func (ci *chanIterator) Err() error {
	return ci.err
}

func blockPutQuery(prefix cid.Prefix) (url.Values, error) {
	q := make(url.Values, 3)

	codec := multicodec.Code(prefix.Codec).String()
	if codec == "" {
		return q, fmt.Errorf("cannot find name for the blocks' CID codec: %x", prefix.Codec)
	}

	mhType, ok := multihash.Codes[prefix.MhType]
	if !ok {
		return q, fmt.Errorf("cannot find name for the blocks' Multihash type: %x", prefix.MhType)
	}

	// From go-ipfs 0.13.0 format is deprecated and we use cid-codec
	q.Set("cid-codec", codec)
	q.Set("mhtype", mhType)
	q.Set("mhlen", strconv.Itoa(prefix.MhLength))
	q.Set("pin", "false")
	q.Set("allow-big-block", "true")
	return q, nil
}

// BlockStream performs a multipart request to block/put with the blocks
// received on the channel.
func (ipfs *Connector) BlockStream(ctx context.Context, blocks <-chan api.NodeWithMeta) error {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/BlockStream")
	defer span.End()

	logger.Debug("streaming blocks to IPFS")

	// Update at the end of block-streaming to have an updated freespace
	// metric.
	defer ipfs.updateInformerMetric(ctx)

	it := &chanIterator{
		ctx:    ctx,
		blocks: blocks,
		seen:   make(map[string]int),
	}
	dir := &chanDirectory{
		iterator: it,
	}

	// We need to pick into the first block to know which Cid prefix we
	// are writing blocks with, so that ipfs calculates the expected
	// multihash (we select the function used). This means that all blocks
	// in a stream should use the same.
	peek, ok := it.Peek()
	if !ok {
		return errors.New("BlockStream: no blocks to peek in blocks channel")
	}

	q, err := blockPutQuery(peek.Cid.Prefix())
	if err != nil {
		return err
	}
	url := "block/put?" + q.Encode()

	// Now we stream the blocks to ipfs. In case of error, we return
	// directly, but leave a goroutine draining the channel until it is
	// closed, which should be soon after returning.
	stats.Record(ctx, observations.BlocksPut.M(1))
	multiFileR := files.NewMultiFileReader(dir, true, false)
	contentType := "multipart/form-data; boundary=" + multiFileR.Boundary()
	body, err := ipfs.postCtxStreamResponse(ctx, url, contentType, multiFileR)
	if err != nil {
		return err
	}
	defer body.Close()

	dec := json.NewDecoder(body)
	for {
		var res ipfsBlockPutResp
		err = dec.Decode(&res)
		if err == io.EOF {
			return nil
		}
		if err != nil {
			logger.Error(err)
			break
		}
		logger.Debugf("response block: %s", res.Key)
		if !it.Seen(res.Key) {
			logger.Warningf("blockPut response CID (%s) does not match the multihash of any blocks sent", res.Key)
		}
	}

	// keep draining blocks channel until closed.
	go func() {
		for range blocks {
		}
	}()

	if err != nil {
		stats.Record(ipfs.ctx, observations.BlocksAddedError.M(1))
	}
	return err
}

// BlockGet retrieves an ipfs block with the given cid
func (ipfs *Connector) BlockGet(ctx context.Context, c api.Cid) ([]byte, error) {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/BlockGet")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, ipfs.config.IPFSRequestTimeout)
	defer cancel()
	url := "block/get?arg=" + c.String()
	return ipfs.postCtx(ctx, url, "", nil)
}

// // FetchRefs asks IPFS to download blocks recursively to the given depth.
// // It discards the response, but waits until it completes.
// func (ipfs *Connector) FetchRefs(ctx context.Context, c api.Cid, maxDepth int) error {
// 	ctx, cancel := context.WithTimeout(ipfs.ctx, ipfs.config.PinTimeout)
// 	defer cancel()

// 	q := url.Values{}
// 	q.Set("recursive", "true")
// 	q.Set("unique", "false") // same memory on IPFS side
// 	q.Set("max-depth", fmt.Sprintf("%d", maxDepth))
// 	q.Set("arg", c.String())

// 	url := fmt.Sprintf("refs?%s", q.Encode())
// 	err := ipfs.postDiscardBodyCtx(ctx, url)
// 	if err != nil {
// 		return err
// 	}
// 	logger.Debugf("refs for %s successfully fetched", c)
// 	return nil
// }

// Returns true every updateMetricsMod-th time that we
// call this function.
func (ipfs *Connector) shouldUpdateMetric() bool {
	if ipfs.config.InformerTriggerInterval <= 0 {
		return false
	}
	curCount := atomic.AddUint64(&ipfs.updateMetricCount, 1)
	if curCount >= uint64(ipfs.config.InformerTriggerInterval) {
		atomic.StoreUint64(&ipfs.updateMetricCount, 0)
		return true
	}
	return false
}

// Trigger a broadcast of the local informer metrics.
func (ipfs *Connector) updateInformerMetric(ctx context.Context) error {
	ctx, span := trace.StartSpan(ctx, "ipfsconn/ipfshttp/updateInformerMetric")
	defer span.End()

	if !ipfs.shouldUpdateMetric() {
		return nil
	}

	err := ipfs.rpcClient.GoContext(
		ctx,
		"",
		"Cluster",
		"SendInformersMetrics",
		struct{}{},
		&struct{}{},
		nil,
	)
	if err != nil {
		logger.Error(err)
	}
	return err
}

// daemon API.
func (ipfs *Connector) apiURL() string {
	switch ipfs.nodeNetwork {
	case "unix":
		return "http+unix://ipfs/api/v0"
	default:
		return fmt.Sprintf("http://%s/api/v0", ipfs.nodeAddr)
	}
}

func (ipfs *Connector) doPostCtx(ctx context.Context, client *http.Client, apiURL, path string, contentType string, postBody io.Reader) (*http.Response, error) {
	logger.Debugf("posting /%s", path)
	urlstr := fmt.Sprintf("%s/%s", apiURL, path)

	req, err := http.NewRequest("POST", urlstr, postBody)
	if err != nil {
		logger.Error("error creating POST request:", err)
		return nil, err
	}

	req.Header.Set("Content-Type", contentType)
	req = req.WithContext(ctx)

	// Rate limiter. If we have a number of failed requests,
	// then wait for a tick.
	if failed := ipfs.failedRequests.Load(); failed > 0 {
		select {
		case <-ipfs.reqRateLimitCh:
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-ipfs.ctx.Done():
			return nil, ipfs.ctx.Err()
		}

	}

	res, err := ipfs.client.Do(req)
	if err != nil {
		// request error: ipfs was unreachable, record it.
		ipfs.failedRequests.Add(1)
		logger.Error("error posting to IPFS:", err)
	} else {
		ipfs.failedRequests.Store(0)
	}

	return res, err
}

// checkResponse tries to parse an error message on non StatusOK responses
// from ipfs.
func checkResponse(path string, res *http.Response) ([]byte, error) {
	if res.StatusCode == http.StatusOK {
		return nil, nil
	}

	body, err := io.ReadAll(res.Body)
	res.Body.Close()
	if err == nil {
		var ipfsErr ipfsError
		if err := json.Unmarshal(body, &ipfsErr); err == nil {
			ipfsErr.code = res.StatusCode
			ipfsErr.path = path
			return body, ipfsErr
		}
	}

	// No error response with useful message from ipfs
	return nil, fmt.Errorf(
		"IPFS request failed (is it running?) (%s). Code %d: %s",
		path,
		res.StatusCode,
		string(body))
}

// postCtxStreamResponse makes a POST request against the ipfs daemon, and
// returns the body reader after checking the request for errros.
func (ipfs *Connector) postCtxStreamResponse(ctx context.Context, path string, contentType string, postBody io.Reader) (io.ReadCloser, error) {
	res, err := ipfs.doPostCtx(ctx, ipfs.client, ipfs.apiURL(), path, contentType, postBody)
	if err != nil {
		return nil, err
	}

	_, err = checkResponse(path, res)
	if err != nil {
		return nil, err
	}
	return res.Body, nil
}

// postCtx makes a POST request against
// the ipfs daemon, reads the full body of the response and
// returns it after checking for errors.
func (ipfs *Connector) postCtx(ctx context.Context, path string, contentType string, postBody io.Reader) ([]byte, error) {
	rdr, err := ipfs.postCtxStreamResponse(ctx, path, contentType, postBody)
	if err != nil {
		return nil, err
	}
	defer rdr.Close()

	body, err := io.ReadAll(rdr)
	if err != nil {
		logger.Errorf("error reading response body: %s", err)
		return nil, err
	}
	return body, nil
}