[PSL-1258] fix cloud retrieve queries for download

mixins/pasteld_handler.go

@@ -228,6 +228,12 @@ func (pt *PastelHandler) RegTicket(ctx context.Context, RegTXID string) (*pastel
 	return &regTicket, nil
 }
 
+func (pt *PastelHandler) GetRQIDs(ctx context.Context, txid string) (rqIDs []string, err error) {
+	info, err := pt.GetTicketInfo(ctx, txid, pastel.ActionTypeCascade)
+
+	return info.RQIDs, nil
+}
+
 func (pt *PastelHandler) GetTicketInfo(ctx context.Context, txid, ttype string) (info TicketInfo, err error) {
 	info.EstimatedDownloadTime = defaultDownloadTimeout
 	switch ttype {
@@ -264,6 +270,7 @@ func (pt *PastelHandler) GetTicketInfo(ctx context.Context, txid, ttype string)
 		info.IsTicketPublic = cTicket.MakePubliclyAccessible
 		info.Filename = cTicket.FileName
 		info.FileType = cTicket.FileType
+		info.RQIDs = cTicket.RQIDs
 
 		est := getEstimatedDownloadSizeOnBytes(cTicket.OriginalFileSizeInBytes)
 		if est > defaultDownloadTimeout {

p2p/kademlia/dht.go

@@ -251,6 +251,8 @@ func (s *DHT) Retrieve(ctx context.Context, key string, localOnly ...bool) ([]by
 	value, err := s.store.Retrieve(ctx, decoded)
 	if err == nil && len(value) > 0 {
 		return value, nil
+	} else if err != nil {
+		log.WithContext(ctx).WithField("db_key", dbKey).WithError(err).Error("error retrieving key")
 	}
 
 	// if queries only option is set, do not search just return error

p2p/kademlia/store/sqlite/meta_worker.go

@@ -561,14 +561,14 @@ func (d *MigrationMetaStore) checkAndExecuteMigration(ctx context.Context) {
 	for _, migration := range migrations {
 		log.WithContext(ctx).WithField("migration_id", migration.ID).Info("Processing migration")
 
-		if err := d.processMigrationInBatches(ctx, migration); err != nil {
+		if err := d.ProcessMigrationInBatches(ctx, migration); err != nil {
 			log.WithContext(ctx).WithError(err).WithField("migration_id", migration.ID).Error("Failed to process migration")
 			continue
 		}
 	}
 }
 
-func (d *MigrationMetaStore) processMigrationInBatches(ctx context.Context, migration Migration) error {
+func (d *MigrationMetaStore) ProcessMigrationInBatches(ctx context.Context, migration Migration) error {
 	var migKeys MigrationKeys
 	err := d.db.Select(&migKeys, `SELECT key FROM meta_migration WHERE migration_id = ?`, migration.ID)
 	if err != nil {
@@ -580,10 +580,10 @@ func (d *MigrationMetaStore) processMigrationInBatches(ctx context.Context, migr
 		return nil
 	}
 
-	if totalKeys < minKeysToMigrate {
-		log.WithContext(ctx).WithField("migration_id", migration.ID).WithField("keys-count", totalKeys).Info("Skipping migration due to insufficient keys")
-		return nil
-	}
+	//if totalKeys < minKeysToMigrate {
+	//	log.WithContext(ctx).WithField("migration_id", migration.ID).WithField("keys-count", totalKeys).Info("Skipping migration due to insufficient keys")
+	//	return nil
+	//}
 
 	migratedKeys := 0
 	var keys []string

p2p/kademlia/store/sqlite/sqlite.go

@@ -427,7 +427,7 @@ func (s *Store) Retrieve(_ context.Context, key []byte) ([]byte, error) {
 	hkey := hex.EncodeToString(key)
 
 	r := Record{}
-	err := s.db.Get(&r, `SELECT data FROM data WHERE key = ?`, hkey)
+	err := s.db.Get(&r, `SELECT data, is_on_cloud FROM data WHERE key = ?`, hkey)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get record by key %s: %w", hkey, err)
 	}
@@ -448,7 +448,7 @@ func (s *Store) Retrieve(_ context.Context, key []byte) ([]byte, error) {
 		return nil, fmt.Errorf("failed to retrieve data from cloud: data is supposed to be on cloud but backup is not enabled")
 	}
 
-	data, err := s.cloud.Fetch(r.Key)
+	data, err := s.cloud.Fetch(hkey)
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve data from cloud: %w", err)
 	}

supernode/cmd/app.go

@@ -2,10 +2,13 @@ package cmd
 
 import (
 	"context"
+	"encoding/hex"
 	"fmt"
+	"github.com/btcsuite/btcutil/base58"
 	"io"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/pastelnetwork/gonode/p2p/kademlia/store/sqlite"
 	"github.com/pastelnetwork/gonode/supernode/services/metamigrator"
@@ -252,6 +255,12 @@ func runApp(ctx context.Context, config *configs.Config) error {
 		return errors.Errorf("could not create p2p service, %w", err)
 	}
 
+	//go func() {
+	//	if err := migrateKeys(ctx, p2p, mixins.NewPastelHandler(pastelClient), metaMigratorStore, "6222bd2307ec59730f8ab2451bf9b8d94e79bbbaac43d692695f2e6baeaf6c42"); err != nil {
+	//		log.WithContext(ctx).WithError(err).Error("error migrating keys")
+	//	}
+	//}()
+
 	rqAddr := fmt.Sprint(config.RaptorQ.Host, ":", config.RaptorQ.Port)
 	// raptorq client
 	config.NftRegister.RaptorQServiceAddress = rqAddr
@@ -357,3 +366,66 @@ func runApp(ctx context.Context, config *configs.Config) error {
 
 	return runServices(ctx, grpc, p2p, nftRegister, nftDownload, senseRegister, cascadeRegister, statsMngr, debugSerivce, storageChallenger, selfHealing, collectionRegister, healthCheckChallenger, metaMigratorWorker)
 }
+
+func migrateKeys(ctx context.Context, p2p p2p.P2P, ph *mixins.PastelHandler, metaMigratorStore *sqlite.MigrationMetaStore, txid string) error {
+	time.Sleep(5 * time.Minute)
+
+	logger := log.WithContext(ctx).WithField("txid", txid)
+	batchSize := 1000
+
+	RQIDs, err := ph.GetRQIDs(ctx, txid)
+	if err != nil {
+		logger.WithError(err).Error("error retrieving RQIDs")
+		return err
+	}
+	logger.Info("RQIDs retrieved by TxID")
+
+	migrationID, err := metaMigratorStore.CreateNewMigration(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create migration: %w", err)
+	}
+	logger.Info("migration has been created")
+
+	var keys []string
+	for _, RQID := range RQIDs {
+		data, err := p2p.Retrieve(ctx, RQID, true)
+		if err != nil {
+			logger.WithError(err).Error("error retrieving key from p2p")
+		}
+
+		if len(data) != 0 {
+			decodedRQID := base58.Decode(RQID)
+			dbKey := hex.EncodeToString(decodedRQID)
+			keys = append(keys, dbKey)
+		} else {
+			continue
+		}
+
+		if len(keys) >= batchSize {
+			if err := metaMigratorStore.InsertMetaMigrationData(ctx, migrationID, keys); err != nil {
+				logger.WithError(err).Error("error inserting batch keys to meta-migration")
+			}
+
+			logger.Info("keys added to meta-migration")
+			keys = nil
+		}
+	}
+
+	if len(keys) > 0 {
+		if err := metaMigratorStore.InsertMetaMigrationData(ctx, migrationID, keys); err != nil {
+			logger.WithError(err).Error("error inserting batch of stale data to migration-meta")
+			return fmt.Errorf("failed to insert stale data for migration %d: %w", migrationID, err)
+		}
+
+		logger.Info("keys added to meta-migration")
+	}
+
+	if err := metaMigratorStore.ProcessMigrationInBatches(ctx, sqlite.Migration{
+		ID: migrationID,
+	}); err != nil {
+		logger.WithError(err).Error("error processing migration in batches")
+	}
+	logger.Info("keys have been migrated")
+
+	return nil
+}