Merge pull request #16 from filecoin-project/feat/repo-daemon

Wire up repo

cli/chain.go

@@ -18,7 +18,10 @@ var chainHeadCmd = &cli.Command{
 	Name:  "head",
 	Usage: "Print chain head",
 	Action: func(cctx *cli.Context) error {
-		api := getApi(cctx)
+		api, err := getAPI(cctx)
+		if err != nil {
+			return err
+		}
 		ctx := reqContext(cctx)
 
 		head, err := api.ChainHead(ctx)

cli/cmd.go

@@ -6,20 +6,36 @@ import (
 	"os/signal"
 	"syscall"
 
-	"github.com/filecoin-project/go-lotus/api"
+	manet "github.com/multiformats/go-multiaddr-net"
 	"gopkg.in/urfave/cli.v2"
+
+	"github.com/filecoin-project/go-lotus/api"
+	"github.com/filecoin-project/go-lotus/api/client"
+	"github.com/filecoin-project/go-lotus/node/repo"
 )
 
 const (
 	metadataContext = "context"
-	metadataAPI     = "api"
 )
 
 // ApiConnector returns API instance
 type ApiConnector func() api.API
 
-func getApi(ctx *cli.Context) api.API {
+func getAPI(ctx *cli.Context) (api.API, error) {
-	return ctx.App.Metadata[metadataAPI].(ApiConnector)()
+	r, err := repo.NewFS(ctx.String("repo"))
+	if err != nil {
+		return nil, err
+	}
+
+	ma, err := r.APIEndpoint()
+	if err != nil {
+		return nil, err
+	}
+	_, addr, err := manet.DialArgs(ma)
+	if err != nil {
+		return nil, err
+	}
+	return client.NewRPC("http://" + addr + "/rpc/v0"), nil
 }
 
 // reqContext returns context for cli execution. Calling it for the first time

cli/net.go

@@ -26,7 +26,10 @@ var netPeers = &cli.Command{
 	Name:  "peers",
 	Usage: "Print peers",
 	Action: func(cctx *cli.Context) error {
-		api := getApi(cctx)
+		api, err := getAPI(cctx)
+		if err != nil {
+			return err
+		}
 		ctx := reqContext(cctx)
 		peers, err := api.NetPeers(ctx)
 		if err != nil {
@@ -45,7 +48,10 @@ var netListen = &cli.Command{
 	Name:  "listen",
 	Usage: "List listen addresses",
 	Action: func(cctx *cli.Context) error {
-		api := getApi(cctx)
+		api, err := getAPI(cctx)
+		if err != nil {
+			return err
+		}
 		ctx := reqContext(cctx)
 
 		addrs, err := api.NetAddrsListen(ctx)
@@ -64,7 +70,10 @@ var netConnect = &cli.Command{
 	Name:  "connect",
 	Usage: "Connect to a peer",
 	Action: func(cctx *cli.Context) error {
-		api := getApi(cctx)
+		api, err := getAPI(cctx)
+		if err != nil {
+			return err
+		}
 		ctx := reqContext(cctx)
 
 		pis, err := parseAddresses(ctx, cctx.Args().Slice())

cmd/lotus/main.go

@@ -6,8 +6,6 @@ import (
 
 	"gopkg.in/urfave/cli.v2"
 
-	"github.com/filecoin-project/go-lotus/api"
-	"github.com/filecoin-project/go-lotus/api/client"
 	"github.com/filecoin-project/go-lotus/build"
 	lcli "github.com/filecoin-project/go-lotus/cli"
 	"github.com/filecoin-project/go-lotus/daemon"
@@ -22,11 +20,13 @@ func main() {
 		Name:    "lotus",
 		Usage:   "Filecoin decentralized storage network client",
 		Version: build.Version,
-		Metadata: map[string]interface{}{
+		Flags: []cli.Flag{
-			"api": lcli.ApiConnector(func() api.API {
+			&cli.StringFlag{
-				// TODO: get this from repo
+				Name:    "repo",
-				return client.NewRPC("http://127.0.0.1:1234/rpc/v0")
+				EnvVars: []string{"LOTUS_PATH"},
-			}),
+				Hidden:  true,
+				Value:   "~/.lotus", // TODO: Consider XDG_DATA_HOME
+			},
 		},
 
 		Commands: append(local, lcli.Commands...),

daemon/cmd.go

@@ -5,10 +5,11 @@ package daemon
 import (
 	"context"
 
-	"github.com/filecoin-project/go-lotus/node"
+	"github.com/multiformats/go-multiaddr"
-	"github.com/filecoin-project/go-lotus/node/config"
-
 	"gopkg.in/urfave/cli.v2"
+
+	"github.com/filecoin-project/go-lotus/node"
+	"github.com/filecoin-project/go-lotus/node/repo"
 )
 
 // Cmd is the `go-lotus daemon` command
@@ -18,22 +19,37 @@ var Cmd = &cli.Command{
 	Flags: []cli.Flag{
 		&cli.StringFlag{
 			Name:  "api",
-			Value: ":1234",
+			Value: "1234",
 		},
 	},
 	Action: func(cctx *cli.Context) error {
 		ctx := context.Background()
-
+		r, err := repo.NewFS(cctx.String("repo"))
-		cfg, err := config.FromFile("./config.toml")
 		if err != nil {
 			return err
 		}
 
-		api, err := node.New(ctx, node.Online(), node.Config(cfg))
+		if err := r.Init(); err != nil && err != repo.ErrRepoExists {
+			return err
+		}
+
+		api, err := node.New(ctx,
+			node.Online(),
+			node.Repo(r),
+
+			node.Override(node.SetApiEndpointKey, func(lr repo.LockedRepo) error {
+				apima, err := multiaddr.NewMultiaddr("/ip4/127.0.0.1/tcp/" + cctx.String("api"))
+				if err != nil {
+					return err
+				}
+				return lr.SetAPIEndpoint(apima)
+			}),
+		)
 		if err != nil {
 			return err
 		}
 
-		return serveRPC(api, cctx.String("api"))
+		// TODO: properly parse api endpoint (or make it a URL)
+		return serveRPC(api, "127.0.0.1:"+cctx.String("api"))
 	},
 }

go.mod

@@ -39,8 +39,10 @@ require (
 	github.com/libp2p/go-libp2p-yamux v0.2.1
 	github.com/libp2p/go-maddr-filter v0.0.4
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1
+	github.com/mitchellh/go-homedir v1.1.0
 	github.com/multiformats/go-multiaddr v0.0.4
 	github.com/multiformats/go-multiaddr-dns v0.0.2
+	github.com/multiformats/go-multiaddr-net v0.0.1
 	github.com/multiformats/go-multihash v0.0.5
 	github.com/pkg/errors v0.8.1
 	github.com/polydawn/refmt v0.0.0-20190408063855-01bf1e26dd14

go.sum

@@ -350,6 +350,8 @@ github.com/minio/sha256-simd v0.0.0-20190131020904-2d45a736cd16/go.mod h1:2FMWW+
 github.com/minio/sha256-simd v0.0.0-20190328051042-05b4dd3047e5/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U=
 github.com/minio/sha256-simd v0.1.0 h1:U41/2erhAKcmSI14xh/ZTUdBPOzDOIfS93ibzUSl8KM=
 github.com/minio/sha256-simd v0.1.0/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U=
+github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
+github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mr-tron/base58 v1.1.0/go.mod h1:xcD2VGqlgYjBdcBLw+TuYLr8afG+Hj8g2eTVqeSzSU8=
 github.com/mr-tron/base58 v1.1.1/go.mod h1:xcD2VGqlgYjBdcBLw+TuYLr8afG+Hj8g2eTVqeSzSU8=
 github.com/mr-tron/base58 v1.1.2 h1:ZEw4I2EgPKDJ2iEw0cNmLB3ROrEmkOtXIkaG7wZg+78=

node/builder.go

@@ -27,6 +27,7 @@ import (
 	"github.com/filecoin-project/go-lotus/node/modules/helpers"
 	"github.com/filecoin-project/go-lotus/node/modules/lp2p"
 	"github.com/filecoin-project/go-lotus/node/modules/testing"
+	"github.com/filecoin-project/go-lotus/node/repo"
 )
 
 // special is a type used to give keys to modules which
@@ -65,6 +66,8 @@ const (
 	HandleIncomingBlocksKey
 	HandleIncomingMessagesKey
 
+	SetApiEndpointKey
+
 	_nInvokes // keep this last
 )
 
@@ -193,6 +196,33 @@ func Config(cfg *config.Root) Option {
 	)
 }
 
+func Repo(r repo.Repo) Option {
+	lr, err := r.Lock()
+	if err != nil {
+		return Error(err)
+	}
+	cfg, err := lr.Config()
+	if err != nil {
+		return Error(err)
+	}
+	pk, err := lr.Libp2pIdentity()
+	if err != nil {
+		return Error(err)
+	}
+
+	return Options(
+		Config(cfg),
+		Override(new(repo.LockedRepo), modules.LockedRepo(lr)), // module handles closing
+
+		Override(new(datastore.Batching), modules.Datastore),
+		Override(new(blockstore.Blockstore), modules.Blockstore),
+
+		Override(new(ci.PrivKey), pk),
+		Override(new(ci.PubKey), ci.PrivKey.GetPublic),
+		Override(new(peer.ID), peer.IDFromPublicKey),
+	)
+}
+
 // New builds and starts new Filecoin node
 func New(ctx context.Context, opts ...Option) (api.API, error) {
 	resAPI := &API{}

node/modules/core.go

@@ -5,6 +5,7 @@ import (
 
 	"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"
 	logging "github.com/ipfs/go-log"
@@ -16,6 +17,7 @@ import (
 
 	"github.com/filecoin-project/go-lotus/chain"
 	"github.com/filecoin-project/go-lotus/node/modules/helpers"
+	"github.com/filecoin-project/go-lotus/node/repo"
 )
 
 var log = logging.Logger("modules")
@@ -43,3 +45,29 @@ func Bitswap(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, rt routin
 func SetGenesis(cs *chain.ChainStore, g Genesis) error {
 	return cs.SetGenesis(g)
 }
+
+func LockedRepo(lr repo.LockedRepo) func(lc fx.Lifecycle) repo.LockedRepo {
+	return func(lc fx.Lifecycle) repo.LockedRepo {
+		lc.Append(fx.Hook{
+			OnStop: func(_ context.Context) error {
+				return lr.Close()
+			},
+		})
+
+		return lr
+	}
+}
+
+func Datastore(r repo.LockedRepo) (datastore.Batching, error) {
+	return r.Datastore("/metadata")
+}
+
+func Blockstore(r repo.LockedRepo) (blockstore.Blockstore, error) {
+	blocks, err := r.Datastore("/blocks")
+	if err != nil {
+		return nil, err
+	}
+
+	bs := blockstore.NewBlockstore(blocks)
+	return blockstore.NewIdStore(bs), nil
+}

node/node_test.go

@@ -2,10 +2,11 @@ package node_test
 
 import (
 	"context"
-	"github.com/filecoin-project/go-lotus/node"
 	"net/http/httptest"
 	"testing"
 
+	"github.com/filecoin-project/go-lotus/node"
+
 	"github.com/filecoin-project/go-lotus/api"
 	"github.com/filecoin-project/go-lotus/api/client"
 	"github.com/filecoin-project/go-lotus/api/test"

node/opts_test.go

@@ -2,6 +2,7 @@ package node_test
 
 import (
 	"errors"
+
 	"github.com/filecoin-project/go-lotus/node"
 
 	"github.com/filecoin-project/go-lotus/node/modules/lp2p"

node/repo/fsrepo.go

@@ -6,12 +6,17 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
 	badger "github.com/ipfs/go-ds-badger"
 	fslock "github.com/ipfs/go-fs-lock"
+	logging "github.com/ipfs/go-log"
 	"github.com/libp2p/go-libp2p-core/crypto"
+	"github.com/mitchellh/go-homedir"
 	"github.com/multiformats/go-multiaddr"
+	"github.com/pkg/errors"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-lotus/node/config"
@@ -25,6 +30,10 @@ const (
 	fsLock      = "repo.lock"
 )
 
+var log = logging.Logger("repo")
+
+var ErrRepoExists = errors.New("repo exists")
+
 // FsRepo is struct for repo, use NewFS to create
 type FsRepo struct {
 	path string
@@ -34,11 +43,28 @@ var _ Repo = &FsRepo{}
 
 // NewFS creates a repo instance based on a path on file system
 func NewFS(path string) (*FsRepo, error) {
+	path, err := homedir.Expand(path)
+	if err != nil {
+		return nil, err
+	}
+
 	return &FsRepo{
 		path: path,
 	}, nil
 }
 
+func (fsr *FsRepo) Init() error {
+	if _, err := os.Stat(fsr.path); err == nil {
+		return ErrRepoExists
+	} else if !os.IsNotExist(err) {
+		return err
+	}
+
+	log.Infof("Initializing repo at '%s'", fsr.path)
+
+	return os.Mkdir(fsr.path, 0755) // nolint
+}
+
 // APIEndpoint returns endpoint of API in this repo
 func (fsr *FsRepo) APIEndpoint() (multiaddr.Multiaddr, error) {
 	p := filepath.Join(fsr.path, fsAPI)
@@ -88,6 +114,10 @@ func (fsr *FsRepo) Lock() (LockedRepo, error) {
 type fsLockedRepo struct {
 	path   string
 	closer io.Closer
+
+	ds     datastore.Batching
+	dsErr  error
+	dsOnce sync.Once
 }
 
 func (fsr *fsLockedRepo) Close() error {
@@ -114,8 +144,14 @@ func (fsr *fsLockedRepo) stillValid() error {
 	return nil
 }
 
-func (fsr *fsLockedRepo) Datastore() (datastore.Datastore, error) {
+func (fsr *fsLockedRepo) Datastore(ns string) (datastore.Batching, error) {
-	return badger.NewDatastore(fsr.join(fsDatastore), nil)
+	fsr.dsOnce.Do(func() {
+		fsr.ds, fsr.dsErr = badger.NewDatastore(fsr.join(fsDatastore), nil)
+	})
+	if fsr.dsErr != nil {
+		return nil, fsr.dsErr
+	}
+	return namespace.Wrap(fsr.ds, datastore.NewKey(ns)), nil
 }
 
 func (fsr *fsLockedRepo) Config() (*config.Root, error) {
@@ -142,6 +178,13 @@ func (fsr *fsLockedRepo) Libp2pIdentity() (crypto.PrivKey, error) {
 		if err != nil {
 			return nil, xerrors.Errorf("could not write private key: %w", err)
 		}
+	} else if err != nil {
+		return nil, err
+	}
+
+	stat, err = os.Stat(kpath)
+	if err != nil {
+		return nil, err
 	}
 
 	if stat.Mode()&0066 != 0 {
@@ -171,7 +214,7 @@ func (fsr *fsLockedRepo) SetAPIEndpoint(ma multiaddr.Multiaddr) error {
 	if err := fsr.stillValid(); err != nil {
 		return err
 	}
-	return ioutil.WriteFile(fsr.join(fsAPI), []byte(ma.String()), 0666)
+	return ioutil.WriteFile(fsr.join(fsAPI), []byte(ma.String()), 0644)
 }
 
 func (fsr *fsLockedRepo) Wallet() (interface{}, error) {

node/repo/interface.go

@@ -10,7 +10,7 @@ import (
 )
 
 var (
-	ErrNoAPIEndpoint     = xerrors.New("no API Endpoint set")
+	ErrNoAPIEndpoint     = xerrors.New("API not running (no endpoint)")
 	ErrRepoAlreadyLocked = xerrors.New("repo is already locked")
 	ErrClosedRepo        = xerrors.New("repo is no longer open")
 )
@@ -28,7 +28,7 @@ type LockedRepo interface {
 	Close() error
 
 	// Returns datastore defined in this repo.
-	Datastore() (datastore.Datastore, error)
+	Datastore(namespace string) (datastore.Batching, error)
 
 	// Returns config in this repo
 	Config() (*config.Root, error)

node/repo/memrepo.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 
 	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
 	dssync "github.com/ipfs/go-datastore/sync"
 	"github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/multiformats/go-multiaddr"
@@ -135,11 +136,12 @@ func (lmem *lockedMemRepo) Close() error {
 
 }
 
-func (lmem *lockedMemRepo) Datastore() (datastore.Datastore, error) {
+func (lmem *lockedMemRepo) Datastore(ns string) (datastore.Batching, error) {
 	if err := lmem.checkToken(); err != nil {
 		return nil, err
 	}
-	return lmem.mem.datastore, nil
+
+	return namespace.Wrap(lmem.mem.datastore, datastore.NewKey(ns)), nil
 }
 
 func (lmem *lockedMemRepo) Config() (*config.Root, error) {