namecache: ipfs name follow

License: MIT
Signed-off-by: vyzo <vyzo@hackzen.org>

core/commands/follow.go

+package commands
+
+import (
+	"errors"
+	"io"
+	"strings"
+
+	cmds "github.com/ipfs/go-ipfs/commands"
+	e "github.com/ipfs/go-ipfs/core/commands/e"
+
+	"gx/ipfs/Qmde5VP1qUkyQXKCfmEUA7bP64V2HAptbJ7phuPp7jXWwg/go-ipfs-cmdkit"
+)
+
+type ipnsFollowResult struct {
+	OK bool
+}
+
+var IpnsFollowCmd = &cmds.Command{
+	Helptext: cmdkit.HelpText{
+		Tagline: "Follow IPNS names.",
+		ShortDescription: `
+Periodically resolve and optionally pin IPNS names in the background.
+`,
+	},
+	Subcommands: map[string]*cmds.Command{
+		"add":    ipnsFollowAddCmd,
+		"pin":    ipnsFollowPinCmd,
+		"list":   ipnsFollowListCmd,
+		"cancel": ipnsFollowCancelCmd,
+	},
+}
+
+var ipnsFollowAddCmd = &cmds.Command{
+	Helptext: cmdkit.HelpText{
+		Tagline: "Follow a name without pinning",
+		ShortDescription: `
+Follows an IPNS name by periodically resolving in the backround.
+`,
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		n, err := req.InvocContext().GetNode()
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
+		if n.Namecache == nil {
+			res.SetError(errors.New("IPNS Namecache is not available"), cmdkit.ErrClient)
+			return
+		}
+
+		n.Namecache.Follow(req.Arguments()[0], false)
+
+		res.SetOutput(&ipnsFollowResult{true})
+	},
+	Arguments: []cmdkit.Argument{
+		cmdkit.StringArg("name", true, false, "IPNS Name to follow."),
+	},
+	Type: ipnsFollowResult{},
+	Marshalers: cmds.MarshalerMap{
+		cmds.Text: marshalFollowResult,
+	},
+}
+
+var ipnsFollowPinCmd = &cmds.Command{
+	Helptext: cmdkit.HelpText{
+		Tagline: "Follows and pins a name",
+		ShortDescription: `
+Follows an IPNS name by periodically resolving and recursively
+pinning in the backround.
+`,
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		n, err := req.InvocContext().GetNode()
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
+		if n.Namecache == nil {
+			res.SetError(errors.New("IPNS Namecache is not available"), cmdkit.ErrClient)
+			return
+		}
+
+		n.Namecache.Follow(req.Arguments()[0], true)
+
+		res.SetOutput(&ipnsFollowResult{true})
+	},
+	Arguments: []cmdkit.Argument{
+		cmdkit.StringArg("name", true, false, "IPNS Name to follow."),
+	},
+	Type: ipnsFollowResult{},
+	Marshalers: cmds.MarshalerMap{
+		cmds.Text: marshalFollowResult,
+	},
+}
+
+var ipnsFollowListCmd = &cmds.Command{
+	Helptext: cmdkit.HelpText{
+		Tagline: "List names followed by the daemon",
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		n, err := req.InvocContext().GetNode()
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
+		if n.Namecache == nil {
+			res.SetError(errors.New("IPNS Namecache is not available"), cmdkit.ErrClient)
+			return
+		}
+
+		res.SetOutput(&stringList{n.Namecache.ListFollows()})
+	},
+	Type: stringList{},
+	Marshalers: cmds.MarshalerMap{
+		cmds.Text: stringListMarshaler,
+	},
+}
+
+var ipnsFollowCancelCmd = &cmds.Command{
+	Helptext: cmdkit.HelpText{
+		Tagline: "Cancels a follow",
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		n, err := req.InvocContext().GetNode()
+		if err != nil {
+			res.SetError(err, cmdkit.ErrNormal)
+			return
+		}
+
+		if n.Namecache == nil {
+			res.SetError(errors.New("IPNS Namecache is not available"), cmdkit.ErrClient)
+			return
+		}
+
+		n.Namecache.Unfollow(req.Arguments()[0])
+
+		res.SetOutput(&ipnsFollowResult{true})
+	},
+	Arguments: []cmdkit.Argument{
+		cmdkit.StringArg("name", true, false, "Name follow to cancel."),
+	},
+	Type: ipnsFollowResult{},
+	Marshalers: cmds.MarshalerMap{
+		cmds.Text: marshalFollowResult,
+	},
+}
+
+func marshalFollowResult(res cmds.Response) (io.Reader, error) {
+	v, err := unwrapOutput(res.Output())
+	if err != nil {
+		return nil, err
+	}
+
+	output, ok := v.(*ipnsFollowResult)
+	if !ok {
+		return nil, e.TypeErr(output, v)
+	}
+
+	var state string
+	if output.OK {
+		state = "ok"
+	} else {
+		state = "wtf"
+	}
+
+	return strings.NewReader(state + "\n"), nil
+}

core/commands/name/name.go

@@ -63,5 +63,6 @@ Resolve the value of a dnslink:
 		"publish": PublishCmd,
 		"resolve": IpnsCmd,
 		"pubsub":  IpnsPubsubCmd,
+		"follow":  IpnsFollowCmd,
 	},
 }

core/core.go

@@ -24,6 +24,7 @@ import (
 	rp "github.com/ipfs/go-ipfs/exchange/reprovide"
 	filestore "github.com/ipfs/go-ipfs/filestore"
 	mount "github.com/ipfs/go-ipfs/fuse/mount"
+	namecache "github.com/ipfs/go-ipfs/namecache"
 	namesys "github.com/ipfs/go-ipfs/namesys"
 	ipnsrp "github.com/ipfs/go-ipfs/namesys/republisher"
 	p2p "github.com/ipfs/go-ipfs/p2p"
@@ -133,7 +134,8 @@ type IpfsNode struct {
 	Routing      routing.IpfsRouting // the routing system. recommend ipfs-dht
 	Exchange     exchange.Interface  // the block exchange + strategy (bitswap)
 	Namesys      namesys.NameSystem  // the name system, resolves paths to hashes
-	Reprovider   *rp.Reprovider      // the value reprovider system
+	Namecache    namecache.NameCache // the name system follower cache
+	Reprovider   *rp.Reprovider // the value reprovider system
 	IpnsRepub    *ipnsrp.Republisher
 
 	AutoNAT  *autonat.AutoNATService
@@ -588,6 +590,7 @@ func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, routingOptio
 
 	// setup name system
 	n.Namesys = namesys.NewNameSystem(n.Routing, n.Repo.Datastore(), size)
+	n.Namecache = namecache.NewNameCache(ctx, n.Namesys, n.Pinning, n.DAG)
 
 	// setup ipns republishing
 	return n.setupIpnsRepublisher()

namecache/namecache.go

+// Package namecache implements background following (resolution and pinning) of names
+package namecache
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+
+	namesys "github.com/ipfs/go-ipfs/namesys"
+	pin "github.com/ipfs/go-ipfs/pin"
+
+	uio "gx/ipfs/QmUnHNqhSB1JgzVCxL1Kz3yb4bdyB4q1Z9AD5AUBVmt3fZ/go-unixfs/io"
+	resolver "gx/ipfs/QmVi2uUygezqaMTqs3Yzt5FcZFHJoYD4B7jQ2BELjj7ZuY/go-path/resolver"
+	ipld "gx/ipfs/QmcKKBwfz6FyQdHR2jsXrrF6XeSBXYL86anmWNewpFpoF5/go-ipld-format"
+	logging "gx/ipfs/QmcuXC5cxs79ro2cUuHs4HQ2bkDLJUYokwL8aivcX6HW3C/go-log"
+)
+
+const (
+	followInterval = 60 * time.Minute
+)
+
+var log = logging.Logger("namecache")
+
+// NameCache represents a following cache of names
+type NameCache interface {
+	Follow(name string, pinit bool)
+	Unfollow(name string)
+	ListFollows() []string
+}
+
+type nameCache struct {
+	nsys    namesys.NameSystem
+	pinning pin.Pinner
+	dag     ipld.DAGService
+
+	ctx     context.Context
+	follows map[string]func()
+	mx      sync.Mutex
+}
+
+func NewNameCache(ctx context.Context, nsys namesys.NameSystem, pinning pin.Pinner, dag ipld.DAGService) NameCache {
+	return &nameCache{
+		ctx:     ctx,
+		nsys:    nsys,
+		pinning: pinning,
+		dag:     dag,
+		follows: make(map[string]func()),
+	}
+}
+
+// Follow spawns a goroutine that periodically resolves a name
+// and (when pinit is true) pins it in the background
+func (nc *nameCache) Follow(name string, pinit bool) {
+	nc.mx.Lock()
+	defer nc.mx.Unlock()
+
+	if _, ok := nc.follows[name]; ok {
+		return
+	}
+
+	ctx, cancel := context.WithCancel(nc.ctx)
+	go nc.followName(ctx, name, pinit)
+	nc.follows[name] = cancel
+}
+
+// Unfollow cancels a follow
+func (nc *nameCache) Unfollow(name string) {
+	nc.mx.Lock()
+	defer nc.mx.Unlock()
+
+	cancel, ok := nc.follows[name]
+	if ok {
+		cancel()
+		delete(nc.follows, name)
+	}
+}
+
+// ListFollows returns a list of names currently being followed
+func (nc *nameCache) ListFollows() []string {
+	nc.mx.Lock()
+	defer nc.mx.Unlock()
+
+	follows := make([]string, 0)
+	for name, _ := range nc.follows {
+		follows = append(follows, name)
+	}
+
+	return follows
+}
+
+func (nc *nameCache) followName(ctx context.Context, name string, pinit bool) {
+	nc.resolveAndPin(ctx, name, pinit)
+
+	ticker := time.NewTicker(followInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			nc.resolveAndPin(ctx, name, pinit)
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (nc *nameCache) resolveAndPin(ctx context.Context, name string, pinit bool) {
+	log.Debugf("resolving %s", name)
+
+	if !strings.HasPrefix(name, "/ipns/") {
+		name = "/ipns/" + name
+	}
+
+	p, err := nc.nsys.Resolve(ctx, name)
+	if err != nil {
+		log.Debugf("error resolving %s: %s", name, err.Error())
+		return
+	}
+
+	log.Debugf("resolved %s to %s", name, p)
+
+	if !pinit {
+		return
+	}
+
+	log.Debugf("pinning %s", p)
+
+	r := &resolver.Resolver{
+		DAG:         nc.dag,
+		ResolveOnce: uio.ResolveUnixfsOnce,
+	}
+
+	n, err := r.ResolvePath(ctx, p)
+	if err != nil {
+		log.Debugf("error resolving path %s to node: %s", p, err.Error())
+		return
+	}
+
+	err = nc.pinning.Pin(ctx, n, true)
+	if err != nil {
+		log.Debugf("error pinning path %s: %s", p, err.Error())
+		return
+	}
+
+	err = nc.pinning.Flush()
+	if err != nil {
+		log.Debugf("error flushing pin: %s", err.Error())
+	}
+}