Client import

node/api.go

@@ -8,6 +8,7 @@ import (
 	"github.com/filecoin-project/go-lotus/chain"
 	"github.com/filecoin-project/go-lotus/chain/address"
 	"github.com/filecoin-project/go-lotus/miner"
+	"github.com/filecoin-project/go-lotus/node/client"
 
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
@@ -16,6 +17,8 @@ import (
 )
 
 type API struct {
+	client.LocalStorage
+
 	Host   host.Host
 	Chain  *chain.ChainStore
 	PubSub *pubsub.PubSub

node/builder.go

@@ -18,6 +18,7 @@ import (
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	record "github.com/libp2p/go-libp2p-record"
 	"go.uber.org/fx"
+	ipld "github.com/ipfs/go-ipld-format"
 
 	"github.com/filecoin-project/go-lotus/api"
 	"github.com/filecoin-project/go-lotus/chain"
@@ -166,6 +167,7 @@ func Online() Option {
 		Override(new(blockstore.GCLocker), blockstore.NewGCLocker),
 		Override(new(blockstore.GCBlockstore), blockstore.NewGCBlockstore),
 		Override(new(exchange.Interface), modules.Bitswap),
+		Override(new(ipld.DAGService), modules.ClientDAG),
 
 		// Filecoin services
 		Override(new(*chain.Syncer), chain.NewSyncer),

node/client/import.go

@@ -0,0 +1,57 @@
+package client
+
+import (
+	"context"
+	"go.uber.org/fx"
+	"os"
+
+	"github.com/ipfs/go-cid"
+	chunker "github.com/ipfs/go-ipfs-chunker"
+	files "github.com/ipfs/go-ipfs-files"
+	ipld "github.com/ipfs/go-ipld-format"
+	"github.com/ipfs/go-unixfs/importer/balanced"
+	ihelper "github.com/ipfs/go-unixfs/importer/helpers"
+)
+
+type LocalStorage struct {
+	fx.In
+
+	LocalDAG ipld.DAGService
+}
+
+func (s *LocalStorage) ClientImport(ctx context.Context, path string) (cid.Cid, error)  {
+	f, err := os.Open(path)
+	if err != nil {
+		return cid.Undef, err
+	}
+	stat, err := f.Stat()
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	file, err := files.NewReaderPathFile(path, f, stat)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	bufferedDS := ipld.NewBufferedDAG(ctx, s.LocalDAG)
+
+	params := ihelper.DagBuilderParams{
+		Maxlinks:   ihelper.DefaultLinksPerBlock,
+		RawLeaves:  true,
+		CidBuilder: nil,
+		Dagserv:    bufferedDS, // flush?
+		NoCopy:     false,
+	}
+
+	db, err := params.New(chunker.DefaultSplitter(file))
+	if err != nil {
+		return cid.Undef, err
+	}
+	nd, err := balanced.Layout(db)
+	if err != nil {
+		return cid.Undef, err
+	}
+
+	return nd.Cid(), bufferedDS.Commit()
+}

node/modules/core.go

@@ -2,12 +2,19 @@ package modules
 
 import (
 	"context"
+	"github.com/ipfs/go-blockservice"
+	"github.com/ipfs/go-datastore/namespace"
+	"github.com/ipfs/go-ipfs/filestore"
+	"github.com/ipfs/go-merkledag"
+	"path/filepath"
 
 	"github.com/ipfs/go-bitswap"
 	"github.com/ipfs/go-bitswap/network"
 	"github.com/ipfs/go-datastore"
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
 	exchange "github.com/ipfs/go-ipfs-exchange-interface"
+	offline "github.com/ipfs/go-ipfs-exchange-offline"
+	ipld "github.com/ipfs/go-ipld-format"
 	logging "github.com/ipfs/go-log"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peerstore"
@@ -71,3 +78,27 @@ func Blockstore(r repo.LockedRepo) (blockstore.Blockstore, error) {
 	bs := blockstore.NewBlockstore(blocks)
 	return blockstore.NewIdStore(bs), nil
 }
+
+func ClientDAG(lc fx.Lifecycle, r repo.LockedRepo) (ipld.DAGService, error) {
+	clientds, err := r.Datastore("/client")
+	if err != nil {
+		return nil, err
+	}
+	blocks := namespace.Wrap(clientds, datastore.NewKey("blocks"))
+
+	fm := filestore.NewFileManager(clientds, filepath.Dir(r.Path()))
+
+	bs := blockstore.NewBlockstore(blocks)
+	fstore := filestore.NewFilestore(bs, fm)
+	ibs := blockstore.NewIdStore(fstore)
+	bsvc := blockservice.New(ibs, offline.Exchange(ibs))
+	dag := merkledag.NewDAGService(bsvc)
+
+	lc.Append(fx.Hook{
+		OnStop: func(_ context.Context) error {
+			return bsvc.Close()
+		},
+	})
+
+	return dag, nil
+}

node/repo/fsrepo.go

@@ -120,6 +120,10 @@ type fsLockedRepo struct {
 	dsOnce sync.Once
 }
 
+func (fsr *fsLockedRepo) Path() string {
+	return fsr.path
+}
+
 func (fsr *fsLockedRepo) Close() error {
 	err := os.Remove(fsr.join(fsAPI))
 

node/repo/interface.go

@@ -42,4 +42,7 @@ type LockedRepo interface {
 
 	// Wallet returns store of private keys for Filecoin transactions
 	Wallet() (interface{}, error)
+
+	// Path returns absolute path of the repo (or empty string if in-memory)
+	Path() string
 }

node/repo/memrepo.go

@@ -35,6 +35,10 @@ type lockedMemRepo struct {
 	token *byte
 }
 
+func (lmem *lockedMemRepo) Path() string {
+	return ""
+}
+
 var _ Repo = &MemRepo{}
 
 // MemRepoOptions contains options for memory repo