cashless2ecash

proc-transfer.go (7393B)

---

 // 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/db"
 	"c2ec/pkg/provider"
 	"context"
 	"errors"
 	"fmt"
 	"strings"
 	"time"
 )
 
 const REFUND_RETRY_INTERVAL_SECONDS = 1
 
 const REFUND_CHANNEL_BUFFER_SIZE = 10
 const PS_REFUND_CHANNEL = "transfer"
 
 const TRANSFER_STATUS_SUCCESS = 0
 const TRANSFER_STATUS_RETRY = 1
 const TRANSFER_STATUS_FAILED = -1
 
 const MAX_TRANSFER_BACKOFF_MS = 24 * 60 * 60 * 1000 // 1 day
 
 // Sets up and runs an attestor in the background. This must be called at startup.
 func RunTransferrer(
 	ctx context.Context,
 	errs chan error,
 ) {
 
 	// go RunListener(
 	// 	ctx,
 	// 	PS_REFUND_CHANNEL,
 	// 	transferCallback,
 	// 	make(chan *Notification, REFUND_CHANNEL_BUFFER_SIZE),
 	// 	errs,
 	// )
 
 	go func() {
 		lastlog := time.Now().Add(time.Minute * -3)
 		lastlog2 := time.Now().Add(time.Minute * -3)
 		for {
 			time.Sleep(REFUND_RETRY_INTERVAL_SECONDS * time.Second)
 			if lastlog.Before(time.Now().Add(time.Second * -30)) {
 				internal_utils.LogInfo("proc-transfer", "transferrer executing transfers")
 				lastlog = time.Now()
 			}
 			executePendingTransfers(errs, lastlog2)
 			if lastlog2.Before(time.Now().Add(time.Second * -30)) {
 				lastlog2 = time.Now()
 			}
 		}
 	}()
 }
 
 // func transferCallback(notification *Notification, errs chan error) {
 
 // 	internal_utils.LogInfo("proc-transfer", fmt.Sprintf("retrieved information on channel=%s with payload=%s", notification.Channel, notification.Payload))
 
 // 	transferRequestUidBase64 := notification.Payload
 // 	if transferRequestUidBase64 == "" {
 // 		errs <- errors.New("the transfer to refund is not specified")
 // 		return
 // 	}
 
 // 	transferRequestUid, err := base64.StdEncoding.DecodeString(transferRequestUidBase64)
 // 	if err != nil {
 // 		errs <- errors.New("malformed transfer request uid: " + err.Error())
 // 		return
 // 	}
 
 // 	transfer, err := db.DB.GetTransferById(transferRequestUid)
 // 	if err != nil {
 // 		internal_utils.LogWarn("proc-transfer", "unable to retrieve transfer with requestUid")
 // 		internal_utils.LogError("proc-transfer", err)
 // 		transferFailed(transfer, errs)
 // 		errs <- err
 // 		return
 // 	}
 
 // 	if transfer == nil {
 // 		err := errors.New("expected an existing transfer. very strange")
 // 		internal_utils.LogError("proc-transfer", err)
 // 		transferFailed(transfer, errs)
 // 		errs <- err
 // 		return
 // 	}
 
 // 	paytoTargetType, tid, err := ParsePaytoUri(transfer.CreditAccount)
 // 	internal_utils.LogInfo("proc-transfer", "parsed payto-target-type="+paytoTargetType)
 // 	if err != nil {
 // 		internal_utils.LogWarn("proc-transfer", "unable to parse payto-uri="+transfer.CreditAccount)
 // 		errs <- errors.New("malformed transfer request uid: " + err.Error())
 // 		transferFailed(transfer, errs)
 // 		return
 // 	}
 
 // 	provider, err := db.DB.GetTerminalProviderByPaytoTargetType(paytoTargetType)
 // 	if err != nil {
 // 		internal_utils.LogWarn("proc-transfer", "unable to find provider for provider-target-type="+paytoTargetType)
 // 		internal_utils.LogError("proc-transfer", err)
 // 		transferFailed(transfer, errs)
 // 		errs <- err
 // 	}
 
 // 	client := PROVIDER_CLIENTS[provider.Name]
 // 	if client == nil {
 // 		errs <- errors.New("no provider client registered for provider " + provider.Name)
 // 	}
 
 // 	err = client.Refund(tid)
 // 	if err != nil {
 // 		internal_utils.LogError("proc-transfer", err)
 // 		transferFailed(transfer, errs)
 // 		return
 // 	}
 
 // 	err = db.DB.UpdateTransfer(
 // 		transfer.RequestUid,
 // 		time.Now().Unix(),
 // 		TRANSFER_STATUS_SUCCESS, // success
 // 		transfer.Retries,
 // 	)
 // 	if err != nil {
 // 		errs <- err
 // 	}
 // }
 
 func executePendingTransfers(errs chan error, lastlog time.Time) {
 
 	transfers, err := db.DB.GetTransfersByState(TRANSFER_STATUS_RETRY)
 	if err != nil {
 		internal_utils.LogError("proc-transfer-1", err)
 		errs <- err
 		return
 	}
 
 	if lastlog.Before(time.Now().Add(time.Second * -30)) {
 		internal_utils.LogInfo("proc-transfer", fmt.Sprintf("found %d pending transfers", len(transfers)))
 	}
 	for _, t := range transfers {
 
 		shouldRetry := internal_utils.ShouldStartRetry(time.Unix(t.TransferTs, 0), int(t.Retries), MAX_TRANSFER_BACKOFF_MS)
 		if !shouldRetry {
 			if lastlog.Before(time.Now().Add(time.Second * -30)) {
 				internal_utils.LogInfo("proc-transfer", fmt.Sprintf("not retrying transfer id=%d, because backoff not yet exceeded", t.RowId))
 			}
 			continue
 		}
 
 		paytoTargetType, tid, err := internal_utils.ParsePaytoUri(t.CreditAccount)
 		internal_utils.LogInfo("proc-transfer", "parsed payto-target-type="+paytoTargetType)
 		if err != nil {
 			internal_utils.LogWarn("proc-transfer", "parsing payto-target-type failed")
 			internal_utils.LogError("proc-transfer-2", err)
 			continue
 		}
 
 		prvdr, err := db.DB.GetTerminalProviderByPaytoTargetType(paytoTargetType)
 		if err != nil {
 			internal_utils.LogWarn("proc-transfer", "finding terminal by payto target type failed")
 			internal_utils.LogError("proc-transfer-3", err)
 			continue
 		}
 
 		client := provider.PROVIDER_CLIENTS[prvdr.Name]
 		if client == nil {
 			errs <- errors.New("no provider client registered for provider " + prvdr.Name)
 			continue
 		}
 
 		internal_utils.LogInfo("proc-transfer", "refunding transaction "+tid)
 		err = client.Refund(strings.Trim(tid, " \n"))
 		if err != nil {
 			internal_utils.LogWarn("proc-transfer", "refunding using provider client failed")
 			internal_utils.LogError("proc-transfer-4", err)
 			transferFailed(t, errs)
 			continue
 		}
 
 		internal_utils.LogInfo("proc-transfer", "setting transfer to success state")
 		err = db.DB.UpdateTransfer(
 			t.RowId,
 			t.RequestUid,
 			time.Now().Unix(),
 			TRANSFER_STATUS_SUCCESS, // success
 			t.Retries,
 		)
 		if err != nil {
 			internal_utils.LogWarn("proc-transfer", "failed setting refund to success state")
 			internal_utils.LogError("proc-transfer", err)
 		}
 	}
 }
 
 func transferFailed(
 	transfer *db.Transfer,
 	errs chan error,
 ) {
 
 	err := db.DB.UpdateTransfer(
 		transfer.RowId,
 		transfer.RequestUid,
 		time.Now().Unix(),
 		TRANSFER_STATUS_RETRY, // retry transfer.
 		transfer.Retries+1,
 	)
 	if err != nil {
 		errs <- err
 	}
 
 	// if transfer.Retries > 2 {
 	// 	err := db.DB.UpdateTransfer(
 	// 		transfer.RequestUid,
 	// 		time.Now().Unix(),
 	// 		TRANSFER_STATUS_FAILED, // transfer ultimatively failed.
 	// 		transfer.Retries,
 	// 	)
 	// 	if err != nil {
 	// 		errs <- err
 	// 	}
 	// } else {
 	// 	err := db.DB.UpdateTransfer(
 	// 		transfer.RequestUid,
 	// 		time.Now().Unix(),
 	// 		TRANSFER_STATUS_RETRY, // retry transfer.
 	// 		transfer.Retries+1,
 	// 	)
 	// 	if err != nil {
 	// 		errs <- err
 	// 	}
 	//}
 }