cashless2ecash

proc-attestor.go (6383B)

---

 // This file is part of taler-cashless2ecash.
 // Copyright (C) 2024 Joel Häberli
 //
 // taler-cashless2ecash is free software: you can redistribute it and/or modify it
 // under the terms of the GNU Affero General Public License as published
 // by the Free Software Foundation, either version 3 of the License,
 // or (at your option) any later version.
 //
 // taler-cashless2ecash is distributed in the hope that it will be useful, but
 // WITHOUT ANY WARRANTY; without even the implied warranty of
 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 // Affero General Public License for more details.
 //
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 //
 // SPDX-License-Identifier: AGPL3.0-or-later
 
 package internal_proc
 
 import (
 	internal_utils "c2ec/internal/utils"
 	"c2ec/pkg/config"
 	"c2ec/pkg/db"
 	"c2ec/pkg/provider"
 	"context"
 	"errors"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 )
 
 const PAYMENT_NOTIFICATION_CHANNEL_BUFFER_SIZE = 10
 const PS_PAYMENT_NOTIFICATION_CHANNEL = "payment_notification"
 const MAX_BACKOFF_MS = 30 * 60000 // thirty minutes
 
 // Sets up and runs an attestor in the background. This must be called at startup.
 func RunAttestor(
 	ctx context.Context,
 	errs chan error,
 ) {
 
 	go RunListener(
 		ctx,
 		PS_PAYMENT_NOTIFICATION_CHANNEL,
 		confirmationCallback,
 		make(chan *db.Notification, PAYMENT_NOTIFICATION_CHANNEL_BUFFER_SIZE),
 		errs,
 	)
 }
 
 func confirmationCallback(notification *db.Notification, errs chan error) {
 
 	internal_utils.LogInfo("proc-attestor", fmt.Sprintf("retrieved information on channel=%s with payload=%s", notification.Channel, notification.Payload))
 
 	// The payload is formatted like: "{PROVIDER_NAME}|{WITHDRAWAL_ID}|{PROVIDER_TRANSACTION_ID}"
 	// the validation is strict. This means, that the dispatcher emits an error
 	// and returns, if a property is malformed.
 	payload := strings.Split(notification.Payload, "|")
 	if len(payload) != 3 {
 		errs <- errors.New("malformed notification payload: " + notification.Payload)
 		return
 	}
 
 	providerName := payload[0]
 	if providerName == "" {
 		errs <- errors.New("the provider of the payment is not specified")
 		return
 	}
 	withdrawalRowId, err := strconv.Atoi(payload[1])
 	if err != nil {
 		errs <- errors.New("malformed withdrawal_row_id: " + err.Error())
 		return
 	}
 	providerTransactionId := payload[2]
 
 	client := provider.PROVIDER_CLIENTS[providerName]
 	if client == nil {
 		errs <- errors.New("no provider client registered for provider " + providerName)
 	}
 
 	transaction, err := client.GetTransaction(providerTransactionId)
 	if err != nil {
 		internal_utils.LogError("proc-attestor", err)
 		prepareRetryOrAbort(withdrawalRowId, errs)
 		return
 	}
 
 	finaliseOrSetRetry(
 		transaction,
 		withdrawalRowId,
 		errs,
 	)
 }
 
 func finaliseOrSetRetry(
 	transaction provider.ProviderTransaction,
 	withdrawalRowId int,
 	errs chan error,
 ) {
 
 	if transaction == nil {
 		err := errors.New("transaction was nil. will set retry or abort")
 		internal_utils.LogError("proc-attestor", err)
 		errs <- err
 		prepareRetryOrAbort(withdrawalRowId, errs)
 		return
 	}
 
 	if w, err := db.DB.GetWithdrawalById(withdrawalRowId); err != nil {
 		internal_utils.LogError("proc-attestor", err)
 		errs <- err
 		prepareRetryOrAbort(withdrawalRowId, errs)
 		return
 	} else {
 		if w.WithdrawalStatus == internal_utils.CONFIRMED || w.WithdrawalStatus == internal_utils.ABORTED {
 			return
 		}
 		if err := transaction.Confirm(w); err != nil {
 			internal_utils.LogError("proc-attestor", err)
 			errs <- err
 			prepareRetryOrAbort(withdrawalRowId, errs)
 			return
 		}
 	}
 
 	completionProof := transaction.Bytes()
 	if len(completionProof) > 0 {
 		// only allow finalization operation, when the completion
 		// proof of the transaction could be retrieved
 		if transaction.AllowWithdrawal() {
 
 			err := db.DB.FinaliseWithdrawal(withdrawalRowId, internal_utils.CONFIRMED, completionProof)
 			if err != nil {
 				internal_utils.LogError("proc-attestor", err)
 				prepareRetryOrAbort(withdrawalRowId, errs)
 			}
 		} else {
 			// when the received transaction is not allowed, we first check if the
 			// transaction is in a final state which will not allow the withdrawal
 			// and therefore the operation can be aborted, without further retries.
 			if transaction.AbortWithdrawal() {
 				err := db.DB.FinaliseWithdrawal(withdrawalRowId, internal_utils.ABORTED, completionProof)
 				if err != nil {
 					internal_utils.LogError("proc-attestor", err)
 					prepareRetryOrAbort(withdrawalRowId, errs)
 					return
 				}
 			}
 			prepareRetryOrAbort(withdrawalRowId, errs)
 		}
 		return
 	}
 	// when the transaction proof was not present (empty proof), retry.
 	prepareRetryOrAbort(withdrawalRowId, errs)
 }
 
 // Checks wether the maximal amount of retries was already
 // reached and the withdrawal operation shall be aborted or
 // triggers the next retry by setting the last_retry_ts field
 // which will trigger the stored procedure triggering the retry
 // process. The retry counter of the retries is handled by the
 // retrier logic and shall not be set here!
 func prepareRetryOrAbort(
 	withdrawalRowId int,
 	errs chan error,
 ) {
 
 	withdrawal, err := db.DB.GetWithdrawalById(withdrawalRowId)
 	if err != nil {
 		internal_utils.LogError("proc-attestor", err)
 		errs <- err
 		return
 	}
 
 	if config.CONFIG.Server.MaxRetries < 0 {
 		prepareRetry(withdrawal, errs)
 	} else {
 
 		if withdrawal.RetryCounter >= config.CONFIG.Server.MaxRetries {
 
 			internal_utils.LogInfo("proc-attestor", fmt.Sprintf("max retries for withdrawal with id=%d was reached. withdrawal is aborted.", withdrawal.WithdrawalRowId))
 			err := db.DB.FinaliseWithdrawal(withdrawalRowId, internal_utils.ABORTED, make([]byte, 0))
 			if err != nil {
 				internal_utils.LogError("proc-attestor", err)
 			}
 		} else {
 			prepareRetry(withdrawal, errs)
 		}
 	}
 }
 
 func prepareRetry(w *db.Withdrawal, errs chan error) {
 	// refactor this section to set retry counter and last retry field in one query...
 	err := db.DB.SetRetryCounter(int(w.WithdrawalRowId), int(w.RetryCounter)+1)
 	if err != nil {
 		internal_utils.LogError("proc-attestor", err)
 		errs <- err
 		return
 	}
 	lastRetryTs := time.Now().Unix()
 	err = db.DB.SetLastRetry(int(w.WithdrawalRowId), lastRetryTs)
 	if err != nil {
 		internal_utils.LogError("proc-attestor", err)
 		errs <- err
 		return
 	}
 }