HTTP API support for pinning contents (#1658)

* storage;api/http: HTTP API support for pinning contents

* Fix errors and lint issues

* Fix review comments by Janos

* Removed sending message after Pin and UNpin API

* changed argument from IsRaw to raw

* Change FileInfo to PinInfo

api/client/client_test.go

@@ -32,11 +32,12 @@ import (
 	"github.com/ethersphere/swarm/storage"
 	"github.com/ethersphere/swarm/storage/feed"
 	"github.com/ethersphere/swarm/storage/feed/lookup"
+	"github.com/ethersphere/swarm/storage/pin"
 	"github.com/ethersphere/swarm/testutil"
 )
 
-func serverFunc(api *api.API) swarmhttp.TestServer {
-	return swarmhttp.NewServer(api, nil, "")
+func serverFunc(api *api.API, pinAPI *pin.API) swarmhttp.TestServer {
+	return swarmhttp.NewServer(api, pinAPI, "")
 }
 
 // TestClientUploadDownloadRaw test uploading and downloading raw data to swarm

api/http/server.go

@@ -62,6 +62,12 @@ var (
 	getFileFail     = metrics.NewRegisteredCounter("api.http.get.file.fail", nil)
 	getListCount    = metrics.NewRegisteredCounter("api.http.get.list.count", nil)
 	getListFail     = metrics.NewRegisteredCounter("api.http.get.list.fail", nil)
+	getPinCount     = metrics.NewRegisteredCounter("api.http.get.pin.count", nil)
+	getPinFail      = metrics.NewRegisteredCounter("api.http.get.pin.fail", nil)
+	postPinCount    = metrics.NewRegisteredCounter("api.http.post.pin.count", nil)
+	postPinFail     = metrics.NewRegisteredCounter("api.http.post.pin.fail", nil)
+	deletePinCount  = metrics.NewRegisteredCounter("api.http.delete.pin.count", nil)
+	deletePinFail   = metrics.NewRegisteredCounter("api.http.delete.pin.fail", nil)
 )
 
 const (
@@ -165,7 +171,20 @@ func NewServer(api *api.API, pinAPI *pin.API, corsString string) *Server {
 			defaultMiddlewares...,
 		),
 	})
-
+	mux.Handle("/bzz-pin:/", methodHandler{
+		"GET": Adapt(
+			http.HandlerFunc(server.HandleGetPins),
+			defaultMiddlewares...,
+		),
+		"POST": Adapt(
+			http.HandlerFunc(server.HandlePin),
+			defaultPostMiddlewares...,
+		),
+		"DELETE": Adapt(
+			http.HandlerFunc(server.HandleUnpin),
+			defaultMiddlewares...,
+		),
+	})
 	mux.Handle("/", methodHandler{
 		"GET": Adapt(
 			http.HandlerFunc(server.HandleRootPaths),
@@ -900,6 +919,81 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *http.Request) {
 	http.ServeContent(w, r, fileName, time.Now(), newBufferedReadSeeker(reader, getFileBufferSize))
 }
 
+// HandlePin takes a root hash as argument and pins a given file or collection in the local Swarm DB
+func (s *Server) HandlePin(w http.ResponseWriter, r *http.Request) {
+	postPinCount.Inc(1)
+	ruid := GetRUID(r.Context())
+	uri := GetURI(r.Context())
+	fileAddr := uri.Address()
+	log.Debug("handle.post.pin", "ruid", ruid, "uri", r.RequestURI)
+
+	if fileAddr == nil {
+		postPinFail.Inc(1)
+		respondError(w, r, "missig hash to pin ", http.StatusBadRequest)
+		return
+	}
+
+	isRaw := false
+	isRawString := r.URL.Query().Get("raw")
+	if strings.ToLower(isRawString) == "true" {
+		isRaw = true
+	}
+
+	err := s.pinAPI.PinFiles(fileAddr, isRaw, "")
+	if err != nil {
+		postPinFail.Inc(1)
+		respondError(w, r, fmt.Sprintf("error pinning file %s: %s", fileAddr.Hex(), err), http.StatusInternalServerError)
+		return
+	}
+
+	log.Debug("pinned content", "ruid", ruid, "key", fileAddr.Hex())
+	w.Header().Set("Content-Type", "text/plain")
+	w.WriteHeader(http.StatusOK)
+}
+
+// HandleUnpin takes a root hash as argument and unpins the file or collection from the local Swarm DB
+func (s *Server) HandleUnpin(w http.ResponseWriter, r *http.Request) {
+	deletePinCount.Inc(1)
+	ruid := GetRUID(r.Context())
+	uri := GetURI(r.Context())
+	fileAddr := uri.Address()
+	log.Debug("handle.delete.pin", "ruid", ruid, "uri", r.RequestURI)
+
+	if fileAddr == nil {
+		deletePinFail.Inc(1)
+		respondError(w, r, "missig hash to unpin ", http.StatusBadRequest)
+		return
+	}
+
+	err := s.pinAPI.UnpinFiles(fileAddr, "")
+	if err != nil {
+		deletePinFail.Inc(1)
+		respondError(w, r, fmt.Sprintf("error pinning file %s: %s", fileAddr.Hex(), err), http.StatusInternalServerError)
+		return
+	}
+
+	log.Debug("unpinned content", "ruid", ruid, "key", fileAddr.Hex())
+	w.Header().Set("Content-Type", "text/plain")
+	w.WriteHeader(http.StatusOK)
+}
+
+// HandleGetPins return information about all the hashes pinned at this moment
+func (s *Server) HandleGetPins(w http.ResponseWriter, r *http.Request) {
+	getPinCount.Inc(1)
+	ruid := GetRUID(r.Context())
+	log.Debug("handle.get.pin", "ruid", ruid, "uri", r.RequestURI)
+
+	pinnedFiles, err := s.pinAPI.ListPins()
+	if err != nil {
+		getPinFail.Inc(1)
+		respondError(w, r, fmt.Sprintf("error getting pinned files: %s", err), http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(&pinnedFiles)
+}
+
 // calculateNumberOfChunks calculates the number of chunks in an arbitrary content length
 func calculateNumberOfChunks(contentLength int64, isEncrypted bool) int64 {
 	if contentLength < 4096 {

api/http/server_test.go

@@ -20,6 +20,7 @@ import (
 	"archive/tar"
 	"bytes"
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"flag"
@@ -45,6 +46,7 @@ import (
 	"github.com/ethersphere/swarm/storage"
 	"github.com/ethersphere/swarm/storage/feed"
 	"github.com/ethersphere/swarm/storage/feed/lookup"
+	"github.com/ethersphere/swarm/storage/pin"
 	"github.com/ethersphere/swarm/testutil"
 )
 
@@ -54,8 +56,8 @@ func init() {
 	log.Root().SetHandler(log.CallerFileHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(true)))))
 }
 
-func serverFunc(api *api.API) TestServer {
-	return NewServer(api, nil, "")
+func serverFunc(api *api.API, pinAPI *pin.API) TestServer {
+	return NewServer(api, pinAPI, "")
 }
 
 func newTestSigner() (*feed.GenericSigner, error) {
@@ -66,6 +68,65 @@ func newTestSigner() (*feed.GenericSigner, error) {
 	return feed.NewGenericSigner(privKey), nil
 }
 
+// TestPinUnpinAPI function tests the pinning and unpinning through HTTP API.
+// It does the following
+//    1) upload a file
+//    2) pin file using HTTP API
+//    3) list all the files using HTTP API and check if the pinned file is present
+//    4) unpin the pinned file
+//    5) list pinned files and check if the unpinned files is not there anymore
+func TestPinUnpinAPI(t *testing.T) {
+	// Initialize Swarm test server
+	srv := NewTestSwarmServer(t, serverFunc, nil, nil)
+	defer srv.Close()
+
+	// upload a file
+	data := testutil.RandomBytes(1, 10000)
+	rootHash := uploadFile(t, srv, data)
+
+	// pin it
+	pinFile(t, srv, rootHash)
+
+	// get the list of files pinned
+	pinnedInfo := listPinnedFiles(t, srv)
+	listInfos := make([]pin.PinInfo, 0)
+	err := json.Unmarshal(pinnedInfo, &listInfos)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Check if the pinned file is present in the list pin command
+	fileInfo := listInfos[0]
+	if hex.EncodeToString(fileInfo.Address) != string(rootHash) {
+		t.Fatalf("roothash not in list of pinned files")
+	}
+	if !fileInfo.IsRaw {
+		t.Fatalf("pinned file is not raw")
+	}
+	if fileInfo.PinCounter != 1 {
+		t.Fatalf("pin counter is not 1")
+	}
+	if fileInfo.FileSize != uint64(len(data)) {
+		t.Fatalf("data size mismatch, expected %x, got %x", len(data), fileInfo.FileSize)
+	}
+
+	// unpin it
+	unpinFile(t, srv, rootHash)
+
+	// get the list of files pinned again
+	unpinnedInfo := listPinnedFiles(t, srv)
+	listInfosUnpin := make([]pin.PinInfo, 0)
+	err = json.Unmarshal(unpinnedInfo, &listInfosUnpin)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Check if the pinned file is not present in the list pin command
+	if len(listInfosUnpin) != 0 {
+		t.Fatalf("roothash is in list of pinned files")
+	}
+}
+
 // Test the transparent resolving of feed updates with bzz:// scheme
 //
 // First upload data to bzz:, and store the Swarm hash to the resulting manifest in a feed update.
@@ -1406,3 +1467,76 @@ func (t *testResolveValidator) Owner(node [32]byte) (addr common.Address, err er
 func (t *testResolveValidator) HeaderByNumber(context.Context, *big.Int) (header *types.Header, err error) {
 	return
 }
+
+func uploadFile(t *testing.T, srv *TestSwarmServer, data []byte) []byte {
+	t.Helper()
+	resp, err := http.Post(fmt.Sprintf("%s/bzz-raw:/", srv.URL), "text/plain", bytes.NewReader([]byte(data)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("err %s", resp.Status)
+	}
+	rootHash, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return rootHash
+}
+
+func pinFile(t *testing.T, srv *TestSwarmServer, rootHash []byte) []byte {
+	t.Helper()
+	pinResp, err := http.Post(fmt.Sprintf("%s/bzz-pin:/%s?raw=true", srv.URL, string(rootHash)), "text/plain", bytes.NewReader([]byte("")))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pinResp.Body.Close()
+	if pinResp.StatusCode != http.StatusOK {
+		t.Fatalf("err %s", pinResp.Status)
+	}
+	pinMessage, err := ioutil.ReadAll(pinResp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return pinMessage
+}
+
+func listPinnedFiles(t *testing.T, srv *TestSwarmServer) []byte {
+	t.Helper()
+	getPinURL := fmt.Sprintf("%s/bzz-pin:/", srv.URL)
+	listResp, err := http.Get(getPinURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer listResp.Body.Close()
+	if listResp.StatusCode != http.StatusOK {
+		t.Fatalf("err %s", listResp.Status)
+	}
+	pinnedInfo, err := ioutil.ReadAll(listResp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return pinnedInfo
+}
+
+func unpinFile(t *testing.T, srv *TestSwarmServer, rootHash []byte) []byte {
+	t.Helper()
+	req, err := http.NewRequest("DELETE", fmt.Sprintf("%s/bzz-pin:/%s", srv.URL, string(rootHash)), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	unpinResp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer unpinResp.Body.Close()
+	if unpinResp.StatusCode != http.StatusOK {
+		t.Fatalf("err %s", unpinResp.Status)
+	}
+	unpinMessage, err := ioutil.ReadAll(unpinResp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return unpinMessage
+}

api/http/test_server.go

@@ -21,27 +21,35 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/ethersphere/swarm/api"
 	"github.com/ethersphere/swarm/chunk"
+	"github.com/ethersphere/swarm/state"
 	"github.com/ethersphere/swarm/storage"
 	"github.com/ethersphere/swarm/storage/feed"
 	"github.com/ethersphere/swarm/storage/localstore"
+	"github.com/ethersphere/swarm/storage/pin"
 )
 
 type TestServer interface {
 	ServeHTTP(http.ResponseWriter, *http.Request)
 }
 
-func NewTestSwarmServer(t *testing.T, serverFunc func(*api.API) TestServer, resolver api.Resolver,
+func NewTestSwarmServer(t *testing.T, serverFunc func(*api.API, *pin.API) TestServer, resolver api.Resolver,
 	o *localstore.Options) *TestSwarmServer {
 
 	swarmDir, err := ioutil.TempDir("", "swarm-storage-test")
 	if err != nil {
 		t.Fatal(err)
 	}
 
+	stateStore, err := state.NewDBStore(filepath.Join(swarmDir, "state-store.db"))
+	if err != nil {
+		t.Fatalf("could not create state store. Error: %s", err.Error())
+	}
+
 	localStore, err := localstore.New(swarmDir, make([]byte, 32), o)
 	if err != nil {
 		os.RemoveAll(swarmDir)
@@ -63,7 +71,8 @@ func NewTestSwarmServer(t *testing.T, serverFunc func(*api.API) TestServer, reso
 	}
 
 	swarmApi := api.NewAPI(fileStore, resolver, feeds.Handler, nil, tags)
-	apiServer := httptest.NewServer(serverFunc(swarmApi))
+	pinAPI := pin.NewAPI(localStore, stateStore, nil, tags, swarmApi)
+	apiServer := httptest.NewServer(serverFunc(swarmApi, pinAPI))
 
 	tss := &TestSwarmServer{
 		Server:    apiServer,

api/uri.go

@@ -86,7 +86,7 @@ func Parse(rawuri string) (*URI, error) {
 
 	// check the scheme is valid
 	switch uri.Scheme {
-	case "bzz", "bzz-raw", "bzz-immutable", "bzz-list", "bzz-hash", "bzz-feed":
+	case "bzz", "bzz-raw", "bzz-immutable", "bzz-list", "bzz-hash", "bzz-feed", "bzz-pin":
 	default:
 		return nil, fmt.Errorf("unknown scheme %q", u.Scheme)
 	}
@@ -128,6 +128,10 @@ func (u *URI) Hash() bool {
 	return u.Scheme == "bzz-hash"
 }
 
+func (u *URI) Pin() bool {
+	return u.Scheme == "bzz-pin"
+}
+
 func (u *URI) String() string {
 	return u.Scheme + ":/" + u.Addr + "/" + u.Path
 }

cmd/swarm/feeds_test.go

@@ -31,11 +31,12 @@ import (
 	swarmhttp "github.com/ethersphere/swarm/api/http"
 	"github.com/ethersphere/swarm/storage/feed"
 	"github.com/ethersphere/swarm/storage/feed/lookup"
+	"github.com/ethersphere/swarm/storage/pin"
 	"github.com/ethersphere/swarm/testutil"
 )
 
 func TestCLIFeedUpdate(t *testing.T) {
-	srv := swarmhttp.NewTestSwarmServer(t, func(api *api.API) swarmhttp.TestServer {
+	srv := swarmhttp.NewTestSwarmServer(t, func(api *api.API, pinAPI *pin.API) swarmhttp.TestServer {
 		return swarmhttp.NewServer(api, nil, "")
 	}, nil, nil)
 	log.Info("starting a test swarm server")