1  package rpc
  2  
  3  import (
  4  	"bytes"
  5  	"context"
  6  	"fmt"
  7  	"io"
  8  
  9  	"github.com/ipfs/boxo/path"
 10  	blocks "github.com/ipfs/go-block-format"
 11  	"github.com/ipfs/go-cid"
 12  	format "github.com/ipfs/go-ipld-format"
 13  	"github.com/ipfs/kubo/core/coreiface/options"
 14  	multicodec "github.com/multiformats/go-multicodec"
 15  )
 16  
 17  type (
 18  	httpNodeAdder        HttpApi
 19  	HttpDagServ          httpNodeAdder
 20  	pinningHttpNodeAdder httpNodeAdder
 21  )
 22  
 23  func (api *HttpDagServ) Get(ctx context.Context, c cid.Cid) (format.Node, error) {
 24  	r, err := api.core().Block().Get(ctx, path.FromCid(c))
 25  	if err != nil {
 26  		return nil, err
 27  	}
 28  
 29  	data, err := io.ReadAll(r)
 30  	if err != nil {
 31  		return nil, err
 32  	}
 33  
 34  	blk, err := blocks.NewBlockWithCid(data, c)
 35  	if err != nil {
 36  		return nil, err
 37  	}
 38  
 39  	return api.ipldDecoder.DecodeNode(ctx, blk)
 40  }
 41  
 42  func (api *HttpDagServ) GetMany(ctx context.Context, cids []cid.Cid) <-chan *format.NodeOption {
 43  	out := make(chan *format.NodeOption)
 44  
 45  	for _, c := range cids {
 46  		// TODO: Consider limiting concurrency of this somehow
 47  		go func(c cid.Cid) {
 48  			n, err := api.Get(ctx, c)
 49  
 50  			select {
 51  			case out <- &format.NodeOption{Node: n, Err: err}:
 52  			case <-ctx.Done():
 53  			}
 54  		}(c)
 55  	}
 56  	return out
 57  }
 58  
 59  func (api *httpNodeAdder) add(ctx context.Context, nd format.Node, pin bool) error {
 60  	c := nd.Cid()
 61  	prefix := c.Prefix()
 62  
 63  	// preserve 'cid-codec' when sent over HTTP
 64  	cidCodec := multicodec.Code(prefix.Codec).String()
 65  
 66  	// 'format' got replaced by 'cid-codec' in https://github.com/ipfs/interface-go-ipfs-core/pull/80
 67  	// but we still support it here for backward-compatibility with use of CIDv0
 68  	format := ""
 69  	if prefix.Version == 0 {
 70  		cidCodec = ""
 71  		format = "v0"
 72  	}
 73  
 74  	stat, err := api.core().Block().Put(ctx, bytes.NewReader(nd.RawData()),
 75  		options.Block.Hash(prefix.MhType, prefix.MhLength),
 76  		options.Block.CidCodec(cidCodec),
 77  		options.Block.Format(format),
 78  		options.Block.Pin(pin))
 79  	if err != nil {
 80  		return err
 81  	}
 82  	if !stat.Path().RootCid().Equals(c) {
 83  		return fmt.Errorf("cids didn't match - local %s, remote %s", c.String(), stat.Path().RootCid().String())
 84  	}
 85  	return nil
 86  }
 87  
 88  func (api *httpNodeAdder) addMany(ctx context.Context, nds []format.Node, pin bool) error {
 89  	for _, nd := range nds {
 90  		// TODO: optimize
 91  		if err := api.add(ctx, nd, pin); err != nil {
 92  			return err
 93  		}
 94  	}
 95  	return nil
 96  }
 97  
 98  func (api *HttpDagServ) AddMany(ctx context.Context, nds []format.Node) error {
 99  	return (*httpNodeAdder)(api).addMany(ctx, nds, false)
100  }
101  
102  func (api *HttpDagServ) Add(ctx context.Context, nd format.Node) error {
103  	return (*httpNodeAdder)(api).add(ctx, nd, false)
104  }
105  
106  func (api *pinningHttpNodeAdder) Add(ctx context.Context, nd format.Node) error {
107  	return (*httpNodeAdder)(api).add(ctx, nd, true)
108  }
109  
110  func (api *pinningHttpNodeAdder) AddMany(ctx context.Context, nds []format.Node) error {
111  	return (*httpNodeAdder)(api).addMany(ctx, nds, true)
112  }
113  
114  func (api *HttpDagServ) Pinning() format.NodeAdder {
115  	return (*pinningHttpNodeAdder)(api)
116  }
117  
118  func (api *HttpDagServ) Remove(ctx context.Context, c cid.Cid) error {
119  	return api.core().Block().Rm(ctx, path.FromCid(c)) // TODO: should we force rm?
120  }
121  
122  func (api *HttpDagServ) RemoveMany(ctx context.Context, cids []cid.Cid) error {
123  	for _, c := range cids {
124  		// TODO: optimize
125  		if err := api.Remove(ctx, c); err != nil {
126  			return err
127  		}
128  	}
129  	return nil
130  }
131  
132  func (api *httpNodeAdder) core() *HttpApi {
133  	return (*HttpApi)(api)
134  }
135  
136  func (api *HttpDagServ) core() *HttpApi {
137  	return (*HttpApi)(api)
138  }