Implement the core federation protocol

federation/federation_test.go

@@ -0,0 +1,41 @@
+package federation_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFederationStandalone(t *testing.T) {
+	provider, err := libp2p.New()
+	require.NoError(t, err)
+
+	subject := newTestFederationMember(t)
+	require.NoError(t, subject.Start(context.Background()))
+
+	madeUpAdCid := requireRandomCid(t)
+	require.NoError(t, subject.ingester.MarkAdProcessed(provider.ID(), madeUpAdCid))
+	providerAddrInfo := peer.AddrInfo{
+		ID:    provider.ID(),
+		Addrs: provider.Addrs(),
+	}
+	require.NoError(t, subject.registry.Update(context.TODO(), providerAddrInfo, providerAddrInfo, madeUpAdCid, nil, 0))
+
+	head := subject.requireHeadEventually(t, 20*time.Second, time.Second)
+	snapshot := subject.requireSnapshot(t, head.Head)
+
+	require.True(t, snapshot.Epoch > 0)
+	require.True(t, snapshot.VectorClock > 0)
+	require.Len(t, snapshot.Providers.Keys, 1)
+	require.Len(t, snapshot.Providers.Values, 1)
+	status, found := snapshot.Providers.Get(provider.ID().String())
+	require.True(t, found)
+	require.Equal(t, madeUpAdCid.String(), status.LastAdvertisement.String())
+	require.Nil(t, status.HeightHint)
+
+	require.NoError(t, subject.Shutdown(context.TODO()))
+}

federation/handlers.go

@@ -0,0 +1,77 @@
+package federation
+
+import (
+	"errors"
+	"net/http"
+	"path"
+
+	"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"
+	"github.com/ipld/go-ipld-prime/multicodec"
+	"github.com/ipld/go-ipld-prime/schema"
+)
+
+func (f *Federation) handleV1FedHead(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		w.Header().Set("Allow", http.MethodGet)
+		http.Error(w, "", http.StatusMethodNotAllowed)
+		return
+	}
+
+	switch headNode, err := f.getHeadNode(r.Context()); {
+	case err != nil:
+		logger.Errorw("Failed to load head snapshot link", "err", err)
+		http.Error(w, "", http.StatusInternalServerError)
+	case headNode == nil:
+		w.WriteHeader(http.StatusNoContent)
+	default:
+		// TODO: Support alternative content types based on request Accept header.
+		w.Header().Add("Content-Type", "application/json")
+		if err := dagjson.Encode(headNode, w); err != nil {
+			logger.Errorw("Failed to encode head", "err", err)
+			http.Error(w, "", http.StatusInternalServerError)
+		}
+	}
+}
+
+func (f *Federation) handleV1FedSubtree(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		w.Header().Set("Allow", http.MethodGet)
+		http.Error(w, "", http.StatusMethodNotAllowed)
+		return
+	}
+	cidStr := path.Base(r.URL.Path)
+	c, err := cid.Decode(cidStr)
+	if err != nil {
+		http.Error(w, "invalid cid", http.StatusBadRequest)
+		return
+	}
+	logger := logger.With("link", cidStr)
+
+	// Fail fast if codec ask for is not known.
+	encoder, err := multicodec.LookupEncoder(c.Prefix().Codec)
+	if err != nil {
+		logger.Errorw("Failed to find codec for link", "err", err)
+		http.Error(w, "codec not supported", http.StatusBadRequest)
+		return
+	}
+
+	ctx := ipld.LinkContext{Ctx: r.Context()}
+	lnk := cidlink.Link{Cid: c}
+	node, err := f.linkSystem.Load(ctx, lnk, Prototypes.Snapshot)
+	if err != nil {
+		if errors.Is(err, datastore.ErrNotFound) || errors.Is(err, ipld.ErrNotExists{}) {
+			http.Error(w, "", http.StatusNotFound)
+			return
+		}
+		logger.Errorw("Failed to load link", "err", err)
+		http.Error(w, "", http.StatusInternalServerError)
+		return
+	}
+	if err := encoder(node.(schema.TypedNode).Representation(), w); err != nil {
+		logger.Errorw("Failed to encode node", "err", err)
+	}
+}

federation/options.go

@@ -0,0 +1,140 @@
+package federation
+
+import (
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/sync"
+	"github.com/ipld/go-ipld-prime"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/ipld/go-ipld-prime/storage/dsadapter"
+	"github.com/ipni/storetheindex/internal/ingest"
+	"github.com/ipni/storetheindex/internal/registry"
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+type (
+	Option func(*options) error
+
+	options struct {
+		datastore                datastore.Datastore
+		host                     host.Host
+		httpListenAddr           string
+		linkSystem               *ipld.LinkSystem
+		members                  []peer.AddrInfo
+		registry                 *registry.Registry
+		ingester                 *ingest.Ingester
+		reconciliationHttpClient *http.Client
+		reconciliationInterval   time.Duration
+		snapshotInterval         time.Duration
+	}
+)
+
+func newOptions(o ...Option) (*options, error) {
+	opt := options{
+		httpListenAddr:           "0.0.0.0:3004",
+		reconciliationHttpClient: http.DefaultClient,
+		reconciliationInterval:   1 * time.Hour,
+		snapshotInterval:         30 * time.Minute,
+	}
+	for _, apply := range o {
+		if err := apply(&opt); err != nil {
+			return nil, err
+		}
+	}
+	// Check required options
+	if opt.ingester == nil {
+		return nil, errors.New("ingester must be specified")
+	}
+	if opt.registry == nil {
+		return nil, errors.New("registry must be specified")
+	}
+
+	// Set defaults
+	if opt.datastore == nil {
+		opt.datastore = sync.MutexWrap(datastore.NewMapDatastore())
+	}
+	if opt.host == nil {
+		var err error
+		opt.host, err = libp2p.New()
+		if err != nil {
+			return nil, err
+		}
+	}
+	if opt.linkSystem == nil {
+		ls := cidlink.DefaultLinkSystem()
+		storage := dsadapter.Adapter{
+			Wrapped: opt.datastore,
+		}
+		ls.SetReadStorage(&storage)
+		ls.SetWriteStorage(&storage)
+		opt.linkSystem = &ls
+	}
+
+	return &opt, nil
+}
+
+func WithDatastore(v datastore.Datastore) Option {
+	return func(o *options) error {
+		o.datastore = v
+		return nil
+	}
+}
+
+func WithHost(v host.Host) Option {
+	return func(o *options) error {
+		o.host = v
+		return nil
+	}
+}
+func WithHttpListenAddr(v string) Option {
+	return func(o *options) error {
+		o.httpListenAddr = v
+		return nil
+	}
+}
+
+func WithMembers(v ...peer.AddrInfo) Option {
+	return func(o *options) error {
+		o.members = append(o.members, v...)
+		return nil
+	}
+}
+func WithLinkSystem(v *ipld.LinkSystem) Option {
+	return func(o *options) error {
+		o.linkSystem = v
+		return nil
+	}
+}
+
+func WithIngester(v *ingest.Ingester) Option {
+	return func(o *options) error {
+		o.ingester = v
+		return nil
+	}
+}
+
+func WithRegistry(v *registry.Registry) Option {
+	return func(o *options) error {
+		o.registry = v
+		return nil
+	}
+}
+
+func WithReconciliationInterval(v time.Duration) Option {
+	return func(o *options) error {
+		o.reconciliationInterval = v
+		return nil
+	}
+}
+
+func WithSnapshotInterval(v time.Duration) Option {
+	return func(o *options) error {
+		o.snapshotInterval = v
+		return nil
+	}
+}