[PSL-1188] modify cascade registration endpoint with multi-volume ticket changes

Author: j-rafique

common/storage/ticketstore/activation_attempts.go

@@ -5,13 +5,13 @@ import (
 )
 
 type ActivationAttemptsQueries interface {
-	UpsertActivationAttempt(attempt types.ActivationAttempt) error
-	GetActivationAttemptByID(id string) (*types.ActivationAttempt, error)
+	UpsertActivationAttempt(attempt types.ActivationAttempt) (int64, error)
+	GetActivationAttemptByID(id int) (*types.ActivationAttempt, error)
 	GetActivationAttemptsByFileID(fileID string) ([]*types.ActivationAttempt, error)
 }
 
 // UpsertActivationAttempt upsert a new activation attempt into the activation_attempts table
-func (s *TicketStore) UpsertActivationAttempt(attempt types.ActivationAttempt) error {
+func (s *TicketStore) UpsertActivationAttempt(attempt types.ActivationAttempt) (int64, error) {
 	const upsertQuery = `
         INSERT INTO activation_attempts (
             id, file_id, activation_attempt_at, is_successful, error_message
@@ -21,26 +21,28 @@ func (s *TicketStore) UpsertActivationAttempt(attempt types.ActivationAttempt) e
             file_id = excluded.file_id,
             activation_attempt_at = excluded.activation_attempt_at,
             is_successful = excluded.is_successful,
-            error_message = excluded.error_message;`
+            error_message = excluded.error_message
+        returning id;`
 
-	_, err := s.db.Exec(upsertQuery,
+	var id int64
+	err := s.db.QueryRow(upsertQuery,
 		attempt.ID, attempt.FileID, attempt.ActivationAttemptAt,
-		attempt.IsSuccessful, attempt.ErrorMessage)
+		attempt.IsSuccessful, attempt.ErrorMessage).Scan(&id)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
-	return nil
+	return id, nil
 }
 
 // GetActivationAttemptByID retrieves an activation attempt by its ID from the activation_attempts table
-func (s *TicketStore) GetActivationAttemptByID(id string) (*types.ActivationAttempt, error) {
+func (s *TicketStore) GetActivationAttemptByID(id int) (*types.ActivationAttempt, error) {
 	const selectQuery = `
         SELECT id, file_id, activation_attempt_at, is_successful, error_message
         FROM activation_attempts
         WHERE id = ?;`
 
-	row := s.db.QueryRow(selectQuery, id)
+	row := s.db.QueryRow(selectQuery, int64(id))
 
 	var attempt types.ActivationAttempt
 	err := row.Scan(

common/storage/ticketstore/files.go

@@ -14,7 +14,7 @@ type FilesQueries interface {
 func (s *TicketStore) UpsertFile(file types.File) error {
 	const upsertQuery = `
         INSERT INTO files (
-            file_id, upload_timestamp, path, index, base_file_id, task_id, 
+            file_id, upload_timestamp, path, file_index, base_file_id, task_id, 
             reg_txid, activation_txid, req_burn_txn_amount, burn_txn_id, 
             req_amount, is_concluded, cascade_metadata_ticket_id, uuid_key, 
             hash_of_original_big_file, name_of_original_big_file_with_ext, 
@@ -23,25 +23,25 @@ func (s *TicketStore) UpsertFile(file types.File) error {
         ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         ON CONFLICT(file_id) 
         DO UPDATE SET
-            upload_timestamp = excluded.upload_timestamp,
-            path = excluded.path,
-            file_index = excluded.index,
-            base_file_id = excluded.base_file_id,
-            task_id = excluded.task_id,
-            reg_txid = excluded.reg_txid,
-            activation_txid = excluded.activation_txid,
-            req_burn_txn_amount = excluded.req_burn_txn_amount,
-            burn_txn_id = excluded.burn_txn_id,
-            req_amount = excluded.req_amount,
-            is_concluded = excluded.is_concluded,
-            cascade_metadata_ticket_id = excluded.cascade_metadata_ticket_id,
-            uuid_key = excluded.uuid_key,
-            hash_of_original_big_file = excluded.hash_of_original_big_file,
-            name_of_original_big_file_with_ext = excluded.name_of_original_big_file_with_ext,
-            size_of_original_big_file = excluded.size_of_original_big_file,
-            data_type_of_original_big_file = excluded.data_type_of_original_big_file,
-            start_block = excluded.start_block,
-            done_block = excluded.done_block;`
+            upload_timestamp = COALESCE(excluded.upload_timestamp, files.upload_timestamp),
+            path = COALESCE(excluded.path, files.path),
+            file_index = COALESCE(excluded.file_index, files.file_index),
+            base_file_id = COALESCE(excluded.base_file_id, files.base_file_id),
+            task_id = COALESCE(excluded.task_id, files.task_id),
+            reg_txid = COALESCE(excluded.reg_txid, files.reg_txid),
+            activation_txid = COALESCE(excluded.activation_txid, files.activation_txid),
+            req_burn_txn_amount = COALESCE(excluded.req_burn_txn_amount, files.req_burn_txn_amount),
+            burn_txn_id = COALESCE(excluded.burn_txn_id, files.burn_txn_id),
+            req_amount = COALESCE(excluded.req_amount, files.req_amount),
+            is_concluded = COALESCE(excluded.is_concluded, files.is_concluded),
+            cascade_metadata_ticket_id = COALESCE(excluded.cascade_metadata_ticket_id, files.cascade_metadata_ticket_id),
+            uuid_key = COALESCE(excluded.uuid_key, files.uuid_key),
+            hash_of_original_big_file = COALESCE(excluded.hash_of_original_big_file, files.hash_of_original_big_file),
+            name_of_original_big_file_with_ext = COALESCE(excluded.name_of_original_big_file_with_ext, files.name_of_original_big_file_with_ext),
+            size_of_original_big_file = COALESCE(excluded.size_of_original_big_file, files.size_of_original_big_file),
+            data_type_of_original_big_file = COALESCE(excluded.data_type_of_original_big_file, files.data_type_of_original_big_file),
+            start_block = COALESCE(excluded.start_block, files.start_block),
+            done_block = COALESCE(excluded.done_block, files.done_block);`
 
 	_, err := s.db.Exec(upsertQuery,
 		file.FileID, file.UploadTimestamp, file.Path, file.FileIndex, file.BaseFileID, file.TaskID,

common/storage/ticketstore/registration_attempts.go

@@ -5,34 +5,36 @@ import (
 )
 
 type RegistrationAttemptsQueries interface {
-	UpsertRegistrationAttempt(attempt types.RegistrationAttempt) error
+	UpsertRegistrationAttempt(attempt types.RegistrationAttempt) (int64, error)
 	GetRegistrationAttemptByID(id int) (*types.RegistrationAttempt, error)
 	GetRegistrationAttemptsByFileID(fileID string) ([]*types.RegistrationAttempt, error)
 }
 
 // UpsertRegistrationAttempt upsert a new registration attempt into the registration_attempts table
-func (s *TicketStore) UpsertRegistrationAttempt(attempt types.RegistrationAttempt) error {
+func (s *TicketStore) UpsertRegistrationAttempt(attempt types.RegistrationAttempt) (int64, error) {
 	const upsertQuery = `
         INSERT INTO registration_attempts (
             id, file_id, reg_started_at, processor_sns, finished_at, is_successful, error_message
         ) VALUES (?, ?, ?, ?, ?, ?, ?)
         ON CONFLICT(id) 
         DO UPDATE SET
-            file_id = excluded.file_id,
-            reg_started_at = excluded.reg_started_at,
-            processor_sns = excluded.processor_sns,
-            finished_at = excluded.finished_at,
-            is_successful = excluded.is_successful,
-            error_message = excluded.error_message;`
+            file_id = COALESCE(excluded.file_id, registration_attempts.file_id),
+            reg_started_at = COALESCE(excluded.reg_started_at, registration_attempts.reg_started_at),
+            processor_sns = COALESCE(excluded.processor_sns, registration_attempts.processor_sns),
+            finished_at = COALESCE(excluded.finished_at, registration_attempts.finished_at),
+            is_successful = COALESCE(excluded.is_successful, registration_attempts.is_successful),
+            error_message = COALESCE(excluded.error_message, registration_attempts.error_message)
+        returning id;`
 
-	_, err := s.db.Exec(upsertQuery,
+	var id int64
+	err := s.db.QueryRow(upsertQuery,
 		attempt.ID, attempt.FileID, attempt.RegStartedAt, attempt.ProcessorSNS,
-		attempt.FinishedAt, attempt.IsSuccessful, attempt.ErrorMessage)
+		attempt.FinishedAt, attempt.IsSuccessful, attempt.ErrorMessage).Scan(&id)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
-	return nil
+	return id, nil
 }
 
 // GetRegistrationAttemptByID retrieves a registration attempt by its ID from the registration_attempts table
@@ -43,7 +45,7 @@ func (s *TicketStore) GetRegistrationAttemptByID(id int) (*types.RegistrationAtt
         FROM registration_attempts
         WHERE id = ?;`
 
-	row := s.db.QueryRow(selectQuery, id)
+	row := s.db.QueryRow(selectQuery, int64(id))
 
 	var attempt types.RegistrationAttempt
 	err := row.Scan(

walletnode/api/services/cascade.go

@@ -3,6 +3,7 @@ package services
 import (
 	"context"
 	"fmt"
+	"github.com/pastelnetwork/gonode/common/types"
 	"os"
 	"path/filepath"
 	"strings"
@@ -26,7 +27,8 @@ import (
 )
 
 const (
-	maxFileSize = 350 * 1024 * 1024 // 350MB in bytes
+	maxFileSize                 = 350 * 1024 * 1024 // 350MB in bytes
+	maxFileRegistrationAttempts = 3
 )
 
 // CascadeAPIHandler - CascadeAPIHandler service
@@ -117,23 +119,82 @@ func (service *CascadeAPIHandler) StartProcessing(ctx context.Context, p *cascad
 		return nil, cascade.MakeUnAuthorized(errors.New("user not authorized: invalid PastelID or Key"))
 	}
 
-	taskID, err := service.register.AddTask(p)
+	baseFile, err := service.register.GetFile(p.FileID)
 	if err != nil {
-		log.WithError(err).Error("unable to add task")
 		return nil, cascade.MakeInternalServerError(err)
 	}
 
-	res = &cascade.StartProcessingResult{
-		TaskID: taskID,
-	}
-
-	fileName, err := service.register.ImageHandler.FileDb.Get(p.FileID)
+	relatedFiles, err := service.register.GetFilesByBaseFileID(baseFile.FileID)
 	if err != nil {
-		return nil, cascade.MakeBadRequest(errors.New("file not found, please re-upload and try again"))
+		return nil, cascade.MakeInternalServerError(err)
 	}
 
-	log.WithField("task_id", taskID).WithField("file_id", p.FileID).WithField("file_name", string(fileName)).
-		Info("task has been added")
+	switch {
+	case len(relatedFiles) == 1:
+		switch {
+		case baseFile.IsConcluded:
+			return nil, cascade.MakeInternalServerError(errors.New("ticket has already been registered & activated"))
+
+		case baseFile.RegTxid == "":
+			baseFileRegistrationAttempts, err := service.register.GetRegistrationAttemptsByFileID(baseFile.FileID)
+			if err != nil {
+				return nil, cascade.MakeInternalServerError(err)
+			}
+
+			switch {
+			case len(baseFileRegistrationAttempts) > maxFileRegistrationAttempts:
+				return nil, cascade.MakeInternalServerError(errors.New("ticket registration attempts have been exceeded"))
+			default:
+				regAttemptID, err := service.register.UpsertRegistrationAttempts(types.RegistrationAttempt{
+					FileID:       p.FileID,
+					RegStartedAt: time.Now().UTC(),
+				})
+				if err != nil {
+					log.WithContext(ctx).WithField("file_id", p.FileID).WithError(err).Error("error inserting registration attempt")
+					return nil, err
+				}
+
+				taskID, err := service.register.AddTask(p, regAttemptID)
+				if err != nil {
+					log.WithError(err).Error("unable to add task")
+					return nil, cascade.MakeInternalServerError(err)
+				}
+
+				res = &cascade.StartProcessingResult{
+					TaskID: taskID,
+				}
+
+				fileName, err := service.register.ImageHandler.FileDb.Get(p.FileID)
+				if err != nil {
+					return nil, cascade.MakeBadRequest(errors.New("file not found, please re-upload and try again"))
+				}
+
+				log.WithField("task_id", taskID).WithField("file_id", p.FileID).WithField("file_name", string(fileName)).
+					Info("task has been added")
+
+				file, err := service.register.GetFile(p.FileID)
+				if err != nil {
+					log.WithContext(ctx).WithError(err).Error("error retrieving file")
+					return nil, cascade.MakeBadRequest(errors.New("file not found, please re-upload and try again"))
+				}
+
+				file.TaskID = taskID
+				file.BurnTxnID = p.BurnTxid
+				err = service.register.UpsertFile(*file)
+				if err != nil {
+					log.WithField("task_id", taskID).WithField("file_id", p.FileID).
+						WithField("burn_txid", p.BurnTxid).
+						WithField("file_name", string(fileName)).
+						Errorf("Error in file upsert: %v", err.Error())
+					return nil, cascade.MakeInternalServerError(errors.New("Error in file upsert"))
+				}
+			}
+		default:
+			// Activation code
+		}
+	case len(relatedFiles) > 1:
+
+	}
 
 	return res, nil
 }

walletnode/services/cascaderegister/service.go

@@ -91,8 +91,9 @@ func (service *CascadeRegistrationService) ValidateUser(ctx context.Context, id
 }
 
 // AddTask create ticket request and start a new task with the given payload
-func (service *CascadeRegistrationService) AddTask(p *cascade.StartProcessingPayload) (string, error) {
+func (service *CascadeRegistrationService) AddTask(p *cascade.StartProcessingPayload, regAttemptID int64) (string, error) {
 	request := FromStartProcessingPayload(p)
+	request.RegAttemptID = regAttemptID
 
 	// get image filename from storage based on image_id
 	filename, err := service.ImageHandler.FileDb.Get(p.FileID)
@@ -143,16 +144,16 @@ func (service *CascadeRegistrationService) StoreFileMetadata(ctx context.Context
 		return errors.Errorf("cannot get block count: %w", err)
 	}
 
-	basefileID := uuid.NewString()
+	fileUUID := uuid.NewString()
 	err = service.ticketDB.UpsertFile(types.File{
-		FileID:                       basefileID,
+		FileID:                       m.TaskID,
 		UploadTimestamp:              time.Now().UTC(),
 		FileIndex:                    "00",
-		BaseFileID:                   basefileID,
+		BaseFileID:                   m.TaskID,
 		TaskID:                       m.TaskID,
 		ReqBurnTxnAmount:             m.ReqPreBurnAmount,
 		ReqAmount:                    m.TotalEstimatedFee,
-		UUIDKey:                      basefileID,
+		UUIDKey:                      fileUUID,
 		HashOfOriginalBigFile:        utils.GetHashStringFromBytes(m.UploadAssetReq.Bytes),
 		NameOfOriginalBigFileWithExt: *m.UploadAssetReq.Filename,
 		SizeOfOriginalBigFile:        utils.GetFileSizeInMB(m.UploadAssetReq.Bytes),
@@ -197,6 +198,46 @@ func (service *CascadeRegistrationService) GetRegistrationAttemptsByFileID(fileI
 	return registrationAttempts, nil
 }
 
+func (service *CascadeRegistrationService) GetRegistrationAttemptsByID(attemptID int) (*types.RegistrationAttempt, error) {
+	registrationAttempt, err := service.ticketDB.GetRegistrationAttemptByID(attemptID)
+	if err != nil {
+		return nil, err
+	}
+	return registrationAttempt, nil
+}
+
+func (service *CascadeRegistrationService) GetActivationAttemptByID(attemptID int) (*types.ActivationAttempt, error) {
+	actAttempt, err := service.ticketDB.GetActivationAttemptByID(attemptID)
+	if err != nil {
+		return nil, err
+	}
+	return actAttempt, nil
+}
+
+func (service *CascadeRegistrationService) UpsertRegistrationAttempts(regAttempt types.RegistrationAttempt) (int64, error) {
+	id, err := service.ticketDB.UpsertRegistrationAttempt(regAttempt)
+	if err != nil {
+		return 0, err
+	}
+	return id, nil
+}
+
+func (service *CascadeRegistrationService) UpsertActivationAttempts(actAttempt types.ActivationAttempt) (int64, error) {
+	id, err := service.ticketDB.UpsertActivationAttempt(actAttempt)
+	if err != nil {
+		return 0, err
+	}
+	return id, nil
+}
+
+func (service *CascadeRegistrationService) UpsertFile(file types.File) error {
+	err := service.ticketDB.UpsertFile(file)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 // NewService returns a new Service instance
 func NewService(config *Config, pastelClient pastel.Client, nodeClient node.ClientInterface,
 	fileStorage storage.FileStorageInterface, db storage.KeyValue,