Merge pull request #454 from fairDataSociety/staging

Staging

Author: asabya

cmd/dfs/cmd/server.go

@@ -206,8 +206,8 @@ can consume it.`,
 
 		ctx, cancel := context.WithCancel(cmd.Context())
 		defer cancel()
-		// datadir will be removed in some future version. it is kept for migration purpose only
-		hdlr, err := api.New(ctx, beeApi, cookieDomain, postageBlockId, corsOrigins, ensConfig, subscriptionConfig, logger)
+
+		hdlr, err := api.New(ctx, beeApi, cookieDomain, postageBlockId, corsOrigins, ensConfig, nil, logger)
 		if err != nil {
 			logger.Error(err.Error())
 			return err

pkg/blockstore/bee/client.go

@@ -46,14 +46,16 @@ const (
 	healthUrl              = "/health"
 	chunkUploadDownloadUrl = "/chunks"
 	bytesUploadDownloadUrl = "/bytes"
+	bzzUrl                 = "/bzz"
 	tagsUrl                = "/tags"
 	pinsUrl                = "/pins/"
 	_                      = pinsUrl
 	swarmPinHeader         = "Swarm-Pin"
 	swarmEncryptHeader     = "Swarm-Encrypt"
 	swarmPostageBatchId    = "Swarm-Postage-Batch-Id"
 	//swarmDeferredUploadHeader = "Swarm-Deferred-Upload"
-	swarmTagHeader = "Swarm-Tag"
+	swarmTagHeader    = "Swarm-Tag"
+	contentTypeHeader = "Content-Type"
 )
 
 // Client is a bee http client that satisfies blockstore.Client
@@ -462,6 +464,110 @@ func (s *Client) DownloadBlob(address []byte) ([]byte, int, error) {
 	return respData, response.StatusCode, nil
 }
 
+// UploadBzz uploads a file through bzz api
+func (s *Client) UploadBzz(data []byte, fileName string) (address []byte, err error) {
+	to := time.Now()
+
+	fullUrl := s.url + bzzUrl + "?name=" + fileName
+	req, err := http.NewRequest(http.MethodPost, fullUrl, bytes.NewBuffer(data))
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set(swarmPostageBatchId, s.postageBlockId)
+	req.Header.Set(contentTypeHeader, "application/json")
+
+	response, err := s.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer response.Body.Close()
+
+	req.Close = true
+
+	respData, err := io.ReadAll(response.Body)
+	if err != nil {
+		return nil, errors.New("error downloading bzz")
+	}
+
+	if response.StatusCode != http.StatusOK && response.StatusCode != http.StatusCreated {
+		var beeErr *beeError
+		err = json.Unmarshal(respData, &beeErr)
+		if err != nil {
+			return nil, errors.New(string(respData))
+		}
+		return nil, errors.New(beeErr.Message)
+	}
+
+	var resp bytesPostResponse
+	err = json.Unmarshal(respData, &resp)
+
+	fields := logrus.Fields{
+		"reference": resp.Reference.String(),
+		"size":      len(respData),
+		"duration":  time.Since(to).String(),
+	}
+	s.logger.WithFields(fields).Log(logrus.DebugLevel, "upload bzz: ")
+
+	// add the data and ref if it is not in cache
+	if !s.inBlockCache(s.downloadBlockCache, resp.Reference.String()) {
+		s.addToBlockCache(s.downloadBlockCache, resp.Reference.String(), data)
+	}
+	return resp.Reference.Bytes(), nil
+}
+
+// DownloadBzz downloads bzz data from the Swarm network.
+func (s *Client) DownloadBzz(address []byte) ([]byte, int, error) {
+	to := time.Now()
+
+	// return the data if this address is already in cache
+	addrString := swarm.NewAddress(address).String()
+	if s.inBlockCache(s.downloadBlockCache, addrString) {
+		return s.getFromBlockCache(s.downloadBlockCache, addrString), 200, nil
+	}
+
+	fullUrl := s.url + bzzUrl + "/" + addrString
+	req, err := http.NewRequest(http.MethodGet, fullUrl, http.NoBody)
+	if err != nil {
+		return nil, http.StatusNotFound, err
+	}
+
+	response, err := s.client.Do(req)
+	if err != nil {
+		return nil, http.StatusNotFound, err
+	}
+	defer response.Body.Close()
+
+	req.Close = true
+
+	respData, err := io.ReadAll(response.Body)
+	if err != nil {
+		return nil, response.StatusCode, errors.New("error downloading bzz")
+	}
+
+	if response.StatusCode != http.StatusOK {
+		var beeErr *beeError
+		err = json.Unmarshal(respData, &beeErr)
+		if err != nil {
+			return nil, response.StatusCode, errors.New(string(respData))
+		}
+		return nil, response.StatusCode, errors.New(beeErr.Message)
+	}
+
+	fields := logrus.Fields{
+		"reference": addrString,
+		"size":      len(respData),
+		"duration":  time.Since(to).String(),
+	}
+	s.logger.WithFields(fields).Log(logrus.DebugLevel, "download bzz: ")
+
+	// add the data and ref if it is not in cache
+	if !s.inBlockCache(s.downloadBlockCache, addrString) {
+		s.addToBlockCache(s.downloadBlockCache, addrString, respData)
+	}
+	return respData, response.StatusCode, nil
+}
+
 // DeleteReference unpins a reference so that it will be garbage collected by the Swarm network.
 func (s *Client) DeleteReference(address []byte) error {
 	// TODO uncomment after unpinning is fixed

pkg/blockstore/bee/mock/client.go

@@ -132,6 +132,16 @@ func (m *BeeClient) DownloadBlob(address []byte) ([]byte, int, error) {
 	return nil, http.StatusInternalServerError, fmt.Errorf("error downloading data")
 }
 
+// UploadBzz downloads  data to bzz api from the Swarm network.
+func (m *BeeClient) UploadBzz(_ []byte, _ string) ([]byte, error) {
+	return nil, nil
+}
+
+// DownloadBzz downloads bzz data from the Swarm network.
+func (m *BeeClient) DownloadBzz(_ []byte) ([]byte, int, error) {
+	return nil, 0, nil
+}
+
 // DeleteReference unpins chunk in swarm
 func (m *BeeClient) DeleteReference(address []byte) error {
 	m.storerMu.Lock()

pkg/blockstore/client.go

@@ -28,8 +28,10 @@ type Client interface {
 	UploadSOC(owner string, id string, signature string, data []byte) (address []byte, err error)
 	UploadChunk(ch swarm.Chunk) (address []byte, err error)
 	UploadBlob(data []byte, tag uint32, encrypt bool) (address []byte, err error)
+	UploadBzz(data []byte, fileName string) (address []byte, err error)
 	DownloadChunk(ctx context.Context, address []byte) (data []byte, err error)
 	DownloadBlob(address []byte) (data []byte, respCode int, err error)
+	DownloadBzz(address []byte) (data []byte, respCode int, err error)
 	DeleteReference(address []byte) error
 	CreateTag(address []byte) (uint32, error)
 	GetTag(tag uint32) (int64, int64, int64, error)

pkg/contracts/config.go

@@ -25,11 +25,7 @@ func TestnetConfig() (*ENSConfig, *SubscriptionConfig) {
 		PublicResolverAddress: "0xbfeCC6c32B224F7D0026ac86506Fe40A9607BD14",
 		ProviderDomain:        "fds",
 	}
-
-	s := &SubscriptionConfig{
-		DataHubAddress: "0x1949beB6CC2db0241Dd625dcaC09891DF5c3756b",
-	}
-	return e, s
+	return e, nil
 }
 
 // PlayConfig defines the configuration for fdp-play

pkg/contracts/datahub/Datahub.go

@@ -551,6 +551,7 @@ func (a *API) RequestSubscription(sessionId string, subHash [32]byte) error {
 	if err != nil {
 		return err
 	}
+
 	return ui.GetPod().RequestSubscription(subHash, nameHash)
 }
 
@@ -636,7 +637,7 @@ type SubscriptionInfo struct {
 }
 
 // GetSubscriptions
-func (a *API) GetSubscriptions(sessionId string) ([]*SubscriptionInfo, error) {
+func (a *API) GetSubscriptions(sessionId string) ([]SubscriptionInfo, error) {
 	// get the loggedin user information
 	ui := a.users.GetLoggedInUserInfo(sessionId)
 	if ui == nil {
@@ -647,26 +648,34 @@ func (a *API) GetSubscriptions(sessionId string) ([]*SubscriptionInfo, error) {
 		return nil, errNilSubManager
 	}
 
-	subscriptions, err := ui.GetPod().GetSubscriptions()
+	nameHash, err := a.users.GetNameHash(ui.GetUserName())
 	if err != nil {
 		return nil, err
 	}
 
-	subs := []*SubscriptionInfo{}
-	for _, item := range subscriptions {
+	subscriptions, err := ui.GetPod().GetSubscriptions(nameHash)
+	if err != nil {
+		return nil, err
+	}
+
+	subs := make([]SubscriptionInfo, len(subscriptions))
+	for i, item := range subscriptions {
 		info, err := ui.GetPod().GetSubscribablePodInfo(item.SubHash)
 		if err != nil {
 			return subs, err
 		}
-		sub := &SubscriptionInfo{
+		var infoLocation = make([]byte, 32)
+		copy(infoLocation, item.UnlockKeyLocation[:])
+		sub := SubscriptionInfo{
 			SubHash:      item.SubHash,
 			PodName:      info.PodName,
 			PodAddress:   info.PodAddress,
-			InfoLocation: item.UnlockKeyLocation[:],
+			InfoLocation: infoLocation,
 			ValidTill:    item.ValidTill.Int64(),
 			Category:     info.Category,
 		}
-		subs = append(subs, sub)
+
+		subs[i] = sub
 	}
 
 	return subs, nil
@@ -678,31 +687,37 @@ func (a *API) GetSubscribablePodInfo(sessionId string, subHash [32]byte) (*rpc.S
 	if ui == nil {
 		return nil, ErrUserNotLoggedIn
 	}
-
+	if a.sm == nil {
+		return nil, errNilSubManager
+	}
 	return a.sm.GetSubscribablePodInfo(subHash)
 }
 
 // OpenSubscribedPod
-func (a *API) OpenSubscribedPod(sessionId string, subHash [32]byte) (*pod.Info, error) {
+func (a *API) OpenSubscribedPod(sessionId string, subHash [32]byte, infoLocation string) (*pod.Info, error) {
+	// get the loggedin user information
+	ui := a.users.GetLoggedInUserInfo(sessionId)
+	if ui == nil {
+		return nil, ErrUserNotLoggedIn
+	}
 
+	if a.sm == nil {
+		return nil, errNilSubManager
+	}
 	sub, err := a.sm.GetSub(subHash)
 	if err != nil {
 		return nil, err
 	}
 
+	subHashString := utils.Encode(subHash[:])
+
 	_, ownerPublicKey, err := a.users.GetUserInfoFromENS(sub.FdpSellerNameHash)
 	if err != nil {
 		return nil, err
 	}
 
-	// get the loggedin user information
-	ui := a.users.GetLoggedInUserInfo(sessionId)
-	if ui == nil {
-		return nil, ErrUserNotLoggedIn
-	}
-
 	// open the pod
-	pi, err := ui.GetPod().OpenSubscribedPod(subHash, ownerPublicKey)
+	pi, err := ui.GetPod().OpenSubscribedPodFromReference(infoLocation, ownerPublicKey)
 	if err != nil {
 		return nil, err
 	}
@@ -711,7 +726,7 @@ func (a *API) OpenSubscribedPod(sessionId string, subHash [32]byte) (*pod.Info,
 		return nil, err
 	}
 	// Add podName in the login user session
-	ui.AddPodName(pi.GetPodName(), pi)
+	ui.AddPodName("0x"+subHashString, pi)
 	return pi, nil
 }
 
@@ -722,7 +737,9 @@ func (a *API) GetSubscribablePods(sessionId string) ([]datahub.DataHubSub, error
 	if ui == nil {
 		return nil, ErrUserNotLoggedIn
 	}
-
+	if a.sm == nil {
+		return nil, errNilSubManager
+	}
 	return ui.GetPod().GetMarketplace()
 }
 
@@ -733,6 +750,8 @@ func (a *API) GetSubsRequests(sessionId string) ([]datahub.DataHubSubRequest, er
 	if ui == nil {
 		return nil, ErrUserNotLoggedIn
 	}
-
+	if a.sm == nil {
+		return nil, errNilSubManager
+	}
 	return ui.GetPod().GetSubRequests()
 }

pkg/dfs/pod_api.go

@@ -85,7 +85,7 @@ func (a *API) ConnectPortableAccountWithWallet(userName, passPhrase, addressHex,
 }
 
 // LoginWithWallet is a controller function which calls the users login function.
-func (a *API) LoginWithWallet(addressHex, signature, sessionId string) (*user.Info, error) {
+func (a *API) LoginWithWallet(addressHex, signature, sessionId string) (*user.Info, string, error) {
 	return a.users.LoginWithWallet(addressHex, signature, a.client, a.tm, a.sm, sessionId)
 }
 

pkg/dfs/user_api.go

@@ -75,7 +75,7 @@ func (d *Directory) ListDir(dirNameWithPath, podPassword string) ([]Entry, []str
 			lsTask := newLsTask(d, dirTopic, dirPath, podPassword, listEntries, mtx, wg)
 			_, err := d.syncManager.Go(lsTask)
 			if err != nil {
-				return nil, nil, fmt.Errorf("list dir : %v", err)
+				wg.Done()
 			}
 		} else if strings.HasPrefix(fileOrDirName, "_F_") {
 			fileName := strings.TrimPrefix(fileOrDirName, "_F_")

pkg/dir/ls.go

@@ -85,7 +85,7 @@ func (d *Directory) SyncDirectoryAsync(ctx context.Context, dirNameWithPath, pod
 			syncTask := newSyncTask(d, filePath, podPassword, wg)
 			_, err = d.syncManager.Go(syncTask)
 			if err != nil { // skipcq: TCV-001
-				return err
+				wg.Done()
 			}
 		} else if strings.HasPrefix(fileOrDirName, "_D_") {
 			dirName := strings.TrimPrefix(fileOrDirName, "_D_")