shell.go

package shell

// package shell implements a remote API interface for a running ipfs daemon
import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	gohttp "net/http"
	"os"
	"path"
	"strings"
	"time"

	cid "github.com/ipfs/go-cid"
	files "github.com/ipfs/go-ipfs-cmdkit/files"
	p2pmetrics "github.com/libp2p/go-libp2p-metrics"
	homedir "github.com/mitchellh/go-homedir"
	ma "github.com/multiformats/go-multiaddr"
	manet "github.com/multiformats/go-multiaddr-net"
	tar "github.com/whyrusleeping/tar-utils"

	"github.com/uppercaveman/go-ipfs-api/model/ls"
)

const (
	DefaultPathName = ".ipfs"
	DefaultPathRoot = "~/" + DefaultPathName
	DefaultApiFile  = "api"
	EnvDir          = "IPFS_PATH"
)

type Shell struct {
	url     string
	httpcli *gohttp.Client
}

func NewLocalShell() *Shell {
	baseDir := os.Getenv(EnvDir)
	if baseDir == "" {
		baseDir = DefaultPathRoot
	}

	baseDir, err := homedir.Expand(baseDir)
	if err != nil {
		return nil
	}

	apiFile := path.Join(baseDir, DefaultApiFile)

	if _, err := os.Stat(apiFile); err != nil {
		return nil
	}

	api, err := ioutil.ReadFile(apiFile)
	if err != nil {
		return nil
	}

	return NewShell(strings.TrimSpace(string(api)))
}

func NewShell(url string) *Shell {
	c := &gohttp.Client{
		Transport: &gohttp.Transport{
			Proxy:             gohttp.ProxyFromEnvironment,
			DisableKeepAlives: true,
		},
	}

	return NewShellWithClient(url, c)
}

func NewShellWithClient(url string, c *gohttp.Client) *Shell {
	if a, err := ma.NewMultiaddr(url); err == nil {
		_, host, err := manet.DialArgs(a)
		if err == nil {
			url = host
		}
	}

	return &Shell{
		url:     url,
		httpcli: c,
	}
}

func (s *Shell) SetTimeout(d time.Duration) {
	s.httpcli.Timeout = d
}

func (s *Shell) Request(command string, args ...string) *RequestBuilder {
	return &RequestBuilder{
		command: command,
		args:    args,
		shell:   s,
	}
}

type IdOutput struct {
	ID              string
	PublicKey       string
	Addresses       []string
	AgentVersion    string
	ProtocolVersion string
}

// ID gets information about a given peer.  Arguments:
//
// peer: peer.ID of the node to look up.  If no peer is specified,
//   return information about the local peer.
func (s *Shell) ID(peer ...string) (*IdOutput, error) {
	if len(peer) > 1 {
		return nil, fmt.Errorf("Too many peer arguments")
	}

	var out IdOutput
	if err := s.Request("id", peer...).Exec(context.Background(), &out); err != nil {
		return nil, err
	}
	return &out, nil
}

// Cat the content at the given path. Callers need to drain and close the returned reader after usage.
func (s *Shell) Cat(path string) (io.ReadCloser, error) {
	resp, err := s.Request("cat", path).Send(context.Background())
	if err != nil {
		return nil, err
	}
	if resp.Error != nil {
		return nil, resp.Error
	}

	return resp.Output, nil
}

type object struct {
	Hash string
}

// Add a file to ipfs from the given reader, returns the hash of the added file
func (s *Shell) Add(r io.Reader) (string, error) {
	return s.AddWithOpts(r, true, false)
}

// AddNoPin a file to ipfs from the given reader, returns the hash of the added file without pinning the file
func (s *Shell) AddNoPin(r io.Reader) (string, error) {
	return s.AddWithOpts(r, false, false)
}

func (s *Shell) AddWithOpts(r io.Reader, pin bool, rawLeaves bool) (string, error) {
	var rc io.ReadCloser
	if rclose, ok := r.(io.ReadCloser); ok {
		rc = rclose
	} else {
		rc = ioutil.NopCloser(r)
	}

	// handler expects an array of files
	fr := files.NewReaderFile("", "", rc, nil)
	slf := files.NewSliceFile("", "", []files.File{fr})
	fileReader := files.NewMultiFileReader(slf, true)

	var out object
	return out.Hash, s.Request("add").
		Option("progress", false).
		Option("pin", pin).
		Option("raw-leaves", rawLeaves).
		Body(fileReader).
		Exec(context.Background(), &out)
}

func (s *Shell) AddLink(target string) (string, error) {
	link := files.NewLinkFile("", "", target, nil)
	slf := files.NewSliceFile("", "", []files.File{link})
	reader := files.NewMultiFileReader(slf, true)

	var out object
	return out.Hash, s.Request("add").Body(reader).Exec(context.Background(), &out)
}

// AddDir adds a directory recursively with all of the files under it
func (s *Shell) AddDir(dir string) (string, error) {
	stat, err := os.Lstat(dir)
	if err != nil {
		return "", err
	}

	sf, err := files.NewSerialFile(path.Base(dir), dir, false, stat)
	if err != nil {
		return "", err
	}
	slf := files.NewSliceFile("", dir, []files.File{sf})
	reader := files.NewMultiFileReader(slf, true)

	resp, err := s.Request("add").
		Option("recursive", true).
		Body(reader).
		Send(context.Background())

	if err != nil {
		return "", nil
	}

	defer resp.Close()

	if resp.Error != nil {
		return "", resp.Error
	}

	dec := json.NewDecoder(resp.Output)
	var final string
	for {
		var out object
		err = dec.Decode(&out)
		if err != nil {
			if err == io.EOF {
				break
			}
			return "", err
		}
		final = out.Hash
	}

	if final == "" {
		return "", errors.New("no results received")
	}

	return final, nil
}

const (
	TRaw = iota
	TDirectory
	TFile
	TMetadata
	TSymlink
)

// List entries at the given path
func (s *Shell) List(path string) ([]*LsLink, error) {
	var out struct{ Objects []LsObject }
	err := s.Request("ls", path).Exec(context.Background(), &out)
	if err != nil {
		return nil, err
	}
	if len(out.Objects) != 1 {
		return nil, errors.New("bad response from server")
	}
	return out.Objects[0].Links, nil
}

type LsLink struct {
	Hash string
	Name string
	Size uint64
	Type int
}

type LsObject struct {
	Links []*LsLink
	LsLink
}

// Pin the given path
func (s *Shell) Pin(path string) error {
	return s.Request("pin/add", path).
		Option("recursive", true).
		Exec(context.Background(), nil)
}

// Unpin the given path
func (s *Shell) Unpin(path string) error {
	return s.Request("pin/rm", path).
		Option("recursive", true).
		Exec(context.Background(), nil)
}

const (
	DirectPin    = "direct"
	RecursivePin = "recursive"
	IndirectPin  = "indirect"
)

type PinInfo struct {
	Type string
}

// Pins returns a map of the pin hashes to their info (currently just the
// pin type, one of DirectPin, RecursivePin, or IndirectPin. A map is returned
// instead of a slice because it is easier to do existence lookup by map key
// than unordered array searching. The map is likely to be more useful to a
// client than a flat list.
func (s *Shell) Pins() (map[string]PinInfo, error) {
	var raw struct{ Keys map[string]PinInfo }
	return raw.Keys, s.Request("pin/ls").Exec(context.Background(), &raw)
}

type PeerInfo struct {
	Addrs []string
	ID    string
}

func (s *Shell) FindPeer(peer string) (*PeerInfo, error) {
	var peers struct{ Responses []PeerInfo }
	err := s.Request("dht/findpeer", peer).Exec(context.Background(), &peers)
	if err != nil {
		return nil, err
	}
	if len(peers.Responses) == 0 {
		return nil, errors.New("peer not found")
	}
	return &peers.Responses[0], nil
}

func (s *Shell) Refs(hash string, recursive bool) (<-chan string, error) {
	resp, err := s.Request("refs", hash).
		Option("recursive", recursive).
		Send(context.Background())
	if err != nil {
		return nil, err
	}

	defer resp.Close()
	if resp.Error != nil {
		return nil, resp.Error
	}

	out := make(chan string)
	go func() {
		var ref struct {
			Ref string
		}
		defer close(out)
		dec := json.NewDecoder(resp.Output)
		for {
			err := dec.Decode(&ref)
			if err != nil {
				return
			}
			if len(ref.Ref) > 0 {
				out <- ref.Ref
			}
		}
	}()

	return out, nil
}

func (s *Shell) Patch(root, action string, args ...string) (string, error) {
	info := new(object)
	err := s.Request("object/patch/"+action, root).
		Arguments(args...).
		Exec(context.Background(), info)
	if err != nil {
		return "", err
	}
	return info.Hash, nil
}

func (s *Shell) PatchData(root string, set bool, data interface{}) (string, error) {
	var read io.Reader
	switch d := data.(type) {
	case io.Reader:
		read = d
	case []byte:
		read = bytes.NewReader(d)
	case string:
		read = strings.NewReader(d)
	default:
		return "", fmt.Errorf("unrecognized type: %#v", data)
	}

	cmd := "append-data"
	if set {
		cmd = "set-data"
	}

	fr := files.NewReaderFile("", "", ioutil.NopCloser(read), nil)
	slf := files.NewSliceFile("", "", []files.File{fr})
	fileReader := files.NewMultiFileReader(slf, true)

	info := new(object)
	err := s.Request("object/patch/"+cmd, root).
		Body(fileReader).
		Exec(context.Background(), info)
	if err != nil {
		return "", err
	}
	return info.Hash, nil
}

// PatchLink : Add a link to a given object.
// arg [string]: The hash of the node to modify. Required: yes.
// arg [string]: Name of link to create. Required: yes.
// arg [string]: IPFS object to add link to. Required: yes.
// create [bool]: Create intermediary nodes. Default: “false”. Required: no.
func (s *Shell) PatchLink(root, path, childhash string, create bool) (string, error) {
	info := new(object)
	err := s.Request("object/patch/add-link", root, path, childhash).
		Option("create", create).
		Exec(context.Background(), info)
	if err != nil {
		return "", err
	}
	return info.Hash, nil
}

func (s *Shell) Get(hash, outdir string) error {
	resp, err := s.Request("get", hash).Option("create", true).Send(context.Background())
	if err != nil {
		return err
	}
	defer resp.Close()

	if resp.Error != nil {
		return resp.Error
	}

	extractor := &tar.Extractor{Path: outdir}
	return extractor.Extract(resp.Output)
}

// NewObject : Create a new object from an ipfs template.
// arg [string]: Template to use. Optional. Required: no.
func (s *Shell) NewObject(template string) (string, error) {
	var out object
	req := s.Request("object/new")
	if template != "" {
		req.Arguments(template)
	}
	return out.Hash, req.Exec(context.Background(), &out)
}

func (s *Shell) ResolvePath(path string) (string, error) {
	var out struct {
		Path string
	}
	err := s.Request("resolve", path).Exec(context.Background(), &out)
	if err != nil {
		return "", err
	}

	return strings.TrimPrefix(out.Path, "/ipfs/"), nil
}

// Version : returns ipfs version and commit sha
func (s *Shell) Version() (string, string, error) {
	ver := struct {
		Version string
		Commit  string
	}{}

	if err := s.Request("version").Exec(context.Background(), &ver); err != nil {
		return "", "", err
	}
	return ver.Version, ver.Commit, nil
}

// IsUp : return bool
func (s *Shell) IsUp() bool {
	_, _, err := s.Version()
	return err == nil
}

// BlockStat :
func (s *Shell) BlockStat(path string) (string, int, error) {
	var inf struct {
		Key  string
		Size int
	}

	err := s.Request("block/stat", path).Exec(context.Background(), &inf)
	if err != nil {
		return "", 0, err
	}
	return inf.Key, inf.Size, nil
}

func (s *Shell) BlockGet(path string) ([]byte, error) {
	resp, err := s.Request("block/get", path).Send(context.Background())
	if err != nil {
		return nil, err
	}
	defer resp.Close()

	if resp.Error != nil {
		return nil, resp.Error
	}

	return ioutil.ReadAll(resp.Output)
}

func (s *Shell) BlockPut(block []byte, format, mhtype string, mhlen int) (string, error) {
	var info struct {
		Key string
	}

	data := bytes.NewReader(block)
	rc := ioutil.NopCloser(data)
	fr := files.NewReaderFile("", "", rc, nil)
	slf := files.NewSliceFile("", "", []files.File{fr})
	fileReader := files.NewMultiFileReader(slf, true)

	err := s.Request("block/put").
		Option("mhtype", mhtype).
		Option("format", format).
		Option("mhlen", mhlen).
		Body(fileReader).
		Exec(context.Background(), &info)
	if err != nil {
		return "", err
	}
	return info.Key, nil
}

type IpfsObject struct {
	Links []ObjectLink
	Data  string
}

type ObjectLink struct {
	Name, Hash string
	Size       uint64
}

func (s *Shell) ObjectGet(path string) (*IpfsObject, error) {
	info := new(IpfsObject)
	err := s.Request("object/get", path).Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

func (s *Shell) ObjectPut(obj *IpfsObject) (string, error) {
	var data bytes.Buffer
	err := json.NewEncoder(&data).Encode(obj)
	if err != nil {
		return "", err
	}

	rc := ioutil.NopCloser(&data)

	fr := files.NewReaderFile("", "", rc, nil)
	slf := files.NewSliceFile("", "", []files.File{fr})
	fileReader := files.NewMultiFileReader(slf, true)

	info := new(object)
	err = s.Request("object/put").Body(fileReader).Exec(context.Background(), info)
	if err != nil {
		return "", err
	}
	return info.Hash, nil
}

func (s *Shell) PubSubSubscribe(topic string) (*PubSubSubscription, error) {
	// connect
	resp, err := s.Request("pubsub/sub", topic).Send(context.Background())
	if err != nil {
		return nil, err
	}
	return newPubSubSubscription(resp), nil
}

func (s *Shell) PubSubPublish(topic, data string) (err error) {
	resp, err := s.Request("pubsub/pub", topic, data).Send(context.Background())
	if err != nil {
		return err
	}
	defer resp.Close()
	return resp.Error
}

type ObjectStats struct {
	Hash           string
	BlockSize      int
	CumulativeSize int
	DataSize       int
	LinksSize      int
	NumLinks       int
}

// ObjectStat gets stats for the DAG object named by key. It returns
// the stats of the requested Object or an error.
func (s *Shell) ObjectStat(key string) (*ObjectStats, error) {
	stat := new(ObjectStats)
	err := s.Request("object/stat", key).Exec(context.Background(), stat)
	if err != nil {
		return nil, err
	}
	return stat, nil
}

// ObjectStat gets stats for the DAG object named by key. It returns
// the stats of the requested Object or an error.
func (s *Shell) StatsBW() (*p2pmetrics.Stats, error) {
	info := new(p2pmetrics.Stats)
	err := s.Request("stats/bw").Exec(context.Background(), &info)
	if err != nil {
		return nil, err
	}
	return info, err
}

type SwarmStreamInfo struct {
	Protocol string
}

type SwarmConnInfo struct {
	Addr    string
	Peer    string
	Latency string
	Muxer   string
	Streams []SwarmStreamInfo
}

// SwarmConnInfos :
type SwarmConnInfos struct {
	Peers []SwarmConnInfo
}

// SwarmPeers gets all the swarm peers
func (s *Shell) SwarmPeers(verbose, streams, latency bool) (*SwarmConnInfos, error) {
	info := new(SwarmConnInfos)

	err := s.Request("swarm/peers").
		Option("verbose", verbose).
		Option("streams", streams).
		Option("latency", latency).
		Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

type swarmConnection struct {
	Strings []string
}

// SwarmConnect : opens a swarm connection to a specific address.
func (s *Shell) SwarmConnect(addr ...string) error {
	info := new(swarmConnection)
	err := s.Request("swarm/connect").Arguments(addr...).Exec(context.Background(), info)
	if err != nil {
		return err
	}
	return nil
}

// Receipt : Receipt
type Receipt struct {
	Peer      string
	Value     float64
	Sent      uint64
	Recv      uint64
	Exchanged uint64
}

// BitswapLedger : Show the current ledger for a peer.
func (s *Shell) BitswapLedger(arg ...string) (*Receipt, error) {
	info := new(Receipt)
	err := s.Request("bitswap/ledger", arg...).Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

// IpfsLs : List directory contents for Unix filesystem objects. 分片对象列表
func (s *Shell) IpfsLs(arg string) (*ls.LsOutput, error) {
	info := new(ls.LsOutput)
	err := s.Request("ls", arg).Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

// RepoFsckOutput :
type RepoFsckOutput struct {
	Message string
}

// RepoFsck : Remove repo lockfiles.
func (s *Shell) RepoFsck() (*RepoFsckOutput, error) {
	info := new(RepoFsckOutput)
	err := s.Request("repo/fsck").Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

// RepoGCOutput :
type RepoGCOutput struct {
	Key   *cid.Cid
	Error string
}

// RepoGC : Perform a garbage collection sweep on the repo.
// quiet [bool]: Write minimal output. Default: “false”. Required: no.
// stream-errors [bool]: Stream errors. Default: “false”. Required: no.
func (s *Shell) RepoGC(quiet, streamErrors bool) (*RepoGCOutput, error) {
	info := new(RepoGCOutput)
	err := s.Request("repo/gc").
		Option("quiet", quiet).
		Option("stream-errors", streamErrors).
		Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

// StatOutput :
type StatOutput struct {
	NumObjects uint64
	RepoSize   uint64
	RepoPath   string
	Version    string
	StorageMax uint64
}

// RepoStat : Get stats for the currently used repo.
// human [bool]: Output RepoSize in MiB. Default: “false”. Required: no.
func (s *Shell) RepoStat(human bool) (*StatOutput, error) {
	info := new(StatOutput)
	err := s.Request("repo/stat").
		Option("human", human).
		Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

// RepoVerifyOutput :
type RepoVerifyOutput struct {
	Message  string
	Progress int
}

// RepoVerify : Verify all blocks in repo are not corrupted.
func (s *Shell) RepoVerify() (*RepoVerifyOutput, error) {
	info := new(RepoVerifyOutput)
	err := s.Request("repo/verify").Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}

// RepoVersionOutput :
type RepoVersionOutput struct {
	Version string
}

// RepoVersion : Show the repo version.
// quiet [bool]: Write minimal output. Required: no.
func (s *Shell) RepoVersion(quiet bool) (*RepoVersionOutput, error) {
	info := new(RepoVersionOutput)
	err := s.Request("repo/version").
		Option("quiet", quiet).
		Exec(context.Background(), info)
	if err != nil {
		return nil, err
	}
	return info, nil
}