http-transport legs

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/libp2p/go-libp2p-core v0.9.0
 	github.com/libp2p/go-libp2p-pubsub v0.5.4
 	github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1
+	github.com/multiformats/go-multiaddr v0.4.0
 	github.com/multiformats/go-multicodec v0.3.0
 	github.com/whyrusleeping/cbor-gen v0.0.0-20210713220151-be142a5ae1a8
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1

http/publish.go

@@ -0,0 +1,85 @@
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"path"
+	"sync"
+
+	"github.com/filecoin-project/go-legs"
+	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-datastore"
+	"github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/codec/dagjson"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	basicnode "github.com/ipld/go-ipld-prime/node/basic"
+)
+
+type httpPublisher struct {
+	rl   sync.RWMutex
+	root cid.Cid
+	lsys ipld.LinkSystem
+}
+
+var _ legs.LegPublisher = (*httpPublisher)(nil)
+var _ http.Handler = (*httpPublisher)(nil)
+
+// NewPublisher creates a new http publisher
+func NewPublisher(ctx context.Context,
+	ds datastore.Batching,
+	lsys ipld.LinkSystem) (legs.LegPublisher, error) {
+	hp := &httpPublisher{}
+	hp.lsys = lsys
+	return hp, nil
+}
+
+func (h *httpPublisher) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	ask := path.Base(r.URL.Path)
+	if ask == "head" {
+		// serve the
+		h.rl.RLock()
+		defer h.rl.RUnlock()
+		out, err := json.Marshal(h.root.String())
+		if err != nil {
+			w.WriteHeader(500)
+			//todo: log
+		} else {
+			_, _ = w.Write(out)
+		}
+		return
+	}
+	// interpret `ask` as a CID to serve.
+	c, err := cid.Parse(ask)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte("invalid request: not a cid"))
+		return
+	}
+	item, err := h.lsys.Load(ipld.LinkContext{}, cidlink.Link{Cid: c}, basicnode.Prototype.Any)
+	if err != nil {
+		if errors.Is(err, ipld.ErrNotExists{}) {
+			w.WriteHeader(http.StatusNotFound)
+			w.Write([]byte("cid not found"))
+			return
+		}
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte("unable to load data for cid"))
+		// todo: log
+		return
+	}
+	// marshal to json and serve.
+	_ = dagjson.Encode(item, w)
+}
+
+func (h *httpPublisher) UpdateRoot(ctx context.Context, c cid.Cid) error {
+	h.rl.Lock()
+	defer h.rl.Unlock()
+	h.root = c
+	return nil
+}
+
+func (h *httpPublisher) Close() error {
+	return nil
+}

http/subscribe.go

@@ -0,0 +1,305 @@
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"path"
+	"sync"
+	"time"
+
+	"github.com/filecoin-project/go-legs"
+	"github.com/ipfs/go-cid"
+	logging "github.com/ipfs/go-log/v2"
+	"github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/datamodel"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	basicnode "github.com/ipld/go-ipld-prime/node/basic"
+	"github.com/ipld/go-ipld-prime/traversal"
+	"github.com/ipld/go-ipld-prime/traversal/selector"
+	peer "github.com/libp2p/go-libp2p-core/peer"
+	"github.com/multiformats/go-multiaddr"
+	manet "github.com/multiformats/go-multiaddr/net"
+)
+
+var defaultPollTime = time.Hour
+
+var log = logging.Logger("go-legs")
+
+// NewHTTPSubscriber creates a legs subcriber that provides subscriptions
+// from publishers identified by
+func NewHTTPSubscriber(ctx context.Context, host *http.Client, publisher multiaddr.Multiaddr, lsys *ipld.LinkSystem, topic string, selector ipld.Node) (legs.LegSubscriber, error) {
+	url, err := toURL(publisher)
+	if err != nil {
+		return nil, err
+	}
+	hs := httpSubscriber{
+		host,
+		cid.Undef,
+		url,
+
+		lsys,
+		selector,
+		sync.Mutex{},
+		make(chan req, 1),
+		make([]chan cid.Cid, 1),
+	}
+	go hs.background()
+	return &hs, nil
+}
+
+// toURL takes a multiaddr of the form:
+// /dnsaddr/thing.com/http/path/to/root
+// /ip/192.168.0.1/http
+// TODO: doesn't support including a signing key definition yet
+func toURL(ma multiaddr.Multiaddr) (string, error) {
+	// host should be either the dns name or the IP
+	_, host, err := manet.DialArgs(ma)
+	if err != nil {
+		return "", err
+	}
+	_, http := multiaddr.SplitFunc(ma, func(c multiaddr.Component) bool {
+		return c.Protocol().Code == multiaddr.P_HTTP
+	})
+	if len(http.Bytes()) == 0 {
+		return "", fmt.Errorf("publisher must be HTTP protocol for this subscriber, was: %s", ma)
+	}
+	_, path := multiaddr.SplitFirst(http)
+	return fmt.Sprintf("https://%s/%s", host, path), nil
+}
+
+type httpSubscriber struct {
+	*http.Client
+	head cid.Cid
+	root string
+
+	lsys            *ipld.LinkSystem
+	defaultSelector ipld.Node
+	submtx          sync.Mutex
+	// reqs is inbound requests for syncs from `Sync` calls
+	reqs chan req
+	subs []chan cid.Cid
+}
+
+var _ (legs.LegSubscriber) = (*httpSubscriber)(nil)
+
+type req struct {
+	cid.Cid
+	Selector ipld.Node
+	ctx      context.Context
+	resp     chan cid.Cid
+}
+
+func (h *httpSubscriber) OnChange() (chan cid.Cid, context.CancelFunc) {
+	ch := make(chan cid.Cid)
+	h.submtx.Lock()
+	defer h.submtx.Unlock()
+	h.subs = append(h.subs, ch)
+	cncl := func() {
+		h.submtx.Lock()
+		defer h.submtx.Unlock()
+		for i, ca := range h.subs {
+			if ca == ch {
+				h.subs[i] = h.subs[len(h.subs)-1]
+				h.subs[len(h.subs)-1] = nil
+				h.subs = h.subs[:len(h.subs)-1]
+				close(ch)
+				break
+			}
+		}
+	}
+	return ch, cncl
+}
+
+// Not supported, since gossip-sub is not supported by this handler.
+// `Sync` must be called explicitly to trigger a fetch instead.
+func (h *httpSubscriber) SetPolicyHandler(p legs.PolicyHandler) error {
+	return nil
+}
+
+func (h *httpSubscriber) SetLatestSync(c cid.Cid) error {
+	h.submtx.Lock()
+	defer h.submtx.Unlock()
+	h.head = c
+	return nil
+}
+
+func (h *httpSubscriber) Sync(ctx context.Context, p peer.ID, c cid.Cid, selector ipld.Node) (chan cid.Cid, context.CancelFunc, error) {
+	respChan := make(chan cid.Cid, 1)
+	cctx, cncl := context.WithCancel(ctx)
+	h.submtx.Lock()
+	defer h.submtx.Unlock()
+	// todo: error if reqs is full
+	h.reqs <- req{
+		Cid:      c,
+		Selector: selector,
+		ctx:      cctx,
+		resp:     respChan,
+	}
+	return respChan, cncl, nil
+}
+
+func (h *httpSubscriber) Close() error {
+	// cancel out subscribers.
+	h.Client.CloseIdleConnections()
+	h.submtx.Lock()
+	defer h.submtx.Unlock()
+	for _, ca := range h.subs {
+		close(ca)
+	}
+	h.subs = make([]chan cid.Cid, 0)
+	return nil
+}
+
+// background event loop for scheduling
+// a. time-scheduled fetches to the provider
+// b. interrupted fetches in response to synchronous 'Sync' calls.
+func (h *httpSubscriber) background() {
+	var nextCid cid.Cid
+	var workResp chan cid.Cid
+	var ctx context.Context
+	var sel ipld.Node
+	var xsel selector.Selector
+	var err error
+	for {
+		select {
+		case r := <-h.reqs:
+			nextCid = r.Cid
+			workResp = r.resp
+			sel = r.Selector
+			ctx = r.ctx
+		case <-time.After(defaultPollTime):
+			nextCid = cid.Undef
+			workResp = nil
+			ctx = context.Background()
+			sel = nil
+		}
+		if nextCid == cid.Undef {
+			nextCid, err = h.fetchHead(ctx)
+		}
+		if sel == nil {
+			sel = h.defaultSelector
+		}
+		if err != nil {
+			log.Warnf("failed to fetch new head: %s", err)
+			err = nil
+			goto next
+		}
+		sel = legs.ExploreRecursiveWithStopNode(selector.RecursionLimitNone(),
+			sel,
+			cidlink.Link{Cid: h.head})
+		if err := h.fetchBlock(ctx, nextCid); err != nil {
+			//log
+			goto next
+		}
+		xsel, err = selector.CompileSelector(sel)
+		if err != nil {
+			//log
+			goto next
+		}
+
+		err = h.walkFetch(ctx, nextCid, xsel)
+		if err != nil {
+			//log
+			goto next
+		}
+
+	next:
+		if workResp != nil {
+			workResp <- nextCid
+			close(workResp)
+			workResp = nil
+		}
+	}
+}
+
+func (h *httpSubscriber) walkFetch(ctx context.Context, root cid.Cid, sel selector.Selector) error {
+	getMissingLs := cidlink.DefaultLinkSystem()
+	getMissingLs.TrustedStorage = true
+	getMissingLs.StorageReadOpener = func(lc ipld.LinkContext, l ipld.Link) (io.Reader, error) {
+		r, err := h.lsys.StorageReadOpener(lc, l)
+		if err == nil {
+			return r, nil
+		}
+		// get.
+		c := l.(cidlink.Link).Cid
+		if err := h.fetchBlock(ctx, c); err != nil {
+			return nil, err
+		}
+		return h.lsys.StorageReadOpener(lc, l)
+	}
+
+	progress := traversal.Progress{
+		Cfg: &traversal.Config{
+			Ctx:                            ctx,
+			LinkSystem:                     getMissingLs,
+			LinkTargetNodePrototypeChooser: basicnode.Chooser,
+		},
+		Path: datamodel.NewPath([]datamodel.PathSegment{}),
+		LastBlock: struct {
+			Path datamodel.Path
+			Link datamodel.Link
+		}{
+			Path: ipld.Path{},
+			Link: nil,
+		},
+	}
+	// get the direct node.
+	rootNode, err := getMissingLs.Load(ipld.LinkContext{}, cidlink.Link{Cid: root}, basicnode.Prototype.Any)
+	if err != nil {
+		return err
+	}
+	return progress.WalkAdv(rootNode, sel, func(p traversal.Progress, n datamodel.Node, vr traversal.VisitReason) error {
+		return nil
+	})
+}
+
+func (h *httpSubscriber) fetch(ctx context.Context, url string, cb func(io.Reader) error) error {
+	req, err := http.NewRequestWithContext(ctx, "GET", path.Join(h.root, url), nil)
+	if err != nil {
+		return err
+	}
+	resp, err := h.Client.Do(req)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("non success http code at %s/%s: %d", h.root, url, resp.StatusCode)
+	}
+
+	defer resp.Body.Close()
+	return cb(resp.Body)
+}
+
+func (h *httpSubscriber) fetchHead(ctx context.Context) (cid.Cid, error) {
+	var cidStr string
+	if err := h.fetch(ctx, "head", func(msg io.Reader) error {
+		return json.NewDecoder(msg).Decode(&cidStr)
+	}); err != nil {
+		return cid.Undef, err
+	}
+
+	return cid.Decode(cidStr)
+}
+
+// fetch an item into the datastore at c if not locally avilable.
+func (h *httpSubscriber) fetchBlock(ctx context.Context, c cid.Cid) error {
+	n, err := h.lsys.Load(ipld.LinkContext{}, cidlink.Link{Cid: c}, basicnode.Prototype.Any)
+	// node is already present.
+	if n != nil && err == nil {
+		return nil
+	}
+
+	return h.fetch(ctx, c.String(), func(data io.Reader) error {
+		writer, committer, err := h.lsys.StorageWriteOpener(ipld.LinkContext{})
+		if err != nil {
+			return err
+		}
+		if _, err := io.Copy(writer, data); err != nil {
+			return err
+		}
+		return committer(cidlink.Link{Cid: c})
+	})
+}