func (pm *PaymentMgr) handlePayments(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			pm.cfg.HubWg.Done()
			return
		case msg := <-pm.paymentCh:
			if !pm.cfg.SoloPool {
				err := pm.payDividends(ctx, msg.CurrentHeight, msg.TreasuryActive)
				if err != nil {
					log.Errorf("unable to process payments: %v", err)
					close(msg.Done)
					continue
				}
				pm.cfg.SignalCache(DividendsPaid)
				close(msg.Done)
			}
		}
	}
}
handlePayments processes dividend payment signals.
package pool
import (
	"context"
	"errors"
	"fmt"
	"math/big"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"decred.org/dcrwallet/rpc/walletrpc"
	txrules "decred.org/dcrwallet/wallet/txrules"
	"decred.org/dcrwallet/wallet/txsizes"
	"github.com/decred/dcrd/chaincfg/chainhash"
	"github.com/decred/dcrd/chaincfg/v3"
	"github.com/decred/dcrd/dcrutil/v3"
	chainjson "github.com/decred/dcrd/rpc/jsonrpc/types/v2"
	"github.com/decred/dcrd/wire"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	errs "github.com/decred/dcrpool/errors"
)
const (
	PPS = "pps"
	PPLNS = "pplns"
	maxRoundingDiff = dcrutil.Amount(500)
	maxTxConfThreshold = uint32(3)
	paymentBufferSize = uint32(30)
)
type TxCreator interface {
	GetTxOut(context.Context, *chainhash.Hash, uint32, bool) (*chainjson.GetTxOutResult, error)
	CreateRawTransaction(context.Context, []chainjson.TransactionInput, map[dcrutil.Address]dcrutil.Amount, *int64, *int64) (*wire.MsgTx, error)
	GetBlock(ctx context.Context, blockHash *chainhash.Hash) (*wire.MsgBlock, error)
}
type TxBroadcaster interface {
	SignTransaction(context.Context, *walletrpc.SignTransactionRequest, ...grpc.CallOption) (*walletrpc.SignTransactionResponse, error)
	PublishTransaction(context.Context, *walletrpc.PublishTransactionRequest, ...grpc.CallOption) (*walletrpc.PublishTransactionResponse, error)
	GetTransaction(context.Context, *walletrpc.GetTransactionRequest, ...grpc.CallOption) (*walletrpc.GetTransactionResponse, error)
	Rescan(ctx context.Context, in *walletrpc.RescanRequest, opts ...grpc.CallOption) (walletrpc.WalletService_RescanClient, error)
}
type rescanMsg struct {
	resp *walletrpc.RescanResponse
	err  error
}
type PaymentMgrConfig struct {
	db Database
	ActiveNet *chaincfg.Params
	PoolFee float64
	LastNPeriod time.Duration
	SoloPool bool
	PaymentMethod string
	PoolFeeAddrs []dcrutil.Address
	WalletAccount uint32
	WalletPass string
	GetBlockConfirmations func(context.Context, *chainhash.Hash) (int64, error)
	FetchTxCreator func() TxCreator
	FetchTxBroadcaster func() TxBroadcaster
	CoinbaseConfTimeout time.Duration
	SignalCache func(event CacheUpdateEvent)
	HubWg *sync.WaitGroup
}
type paymentMsg struct {
	CurrentHeight  uint32
	TreasuryActive bool
	Done           chan bool
}
type PaymentMgr struct {
	failedTxConfs uint32  
	txConfHashes  map[chainhash.Hash]uint32
	processing bool
	paymentCh  chan *paymentMsg
	cfg        *PaymentMgrConfig
	mtx        sync.Mutex
}
func NewPaymentMgr(pCfg *PaymentMgrConfig) (*PaymentMgr, error) {
	pm := &PaymentMgr{
		cfg:          pCfg,
		txConfHashes: make(map[chainhash.Hash]uint32),
		paymentCh:    make(chan *paymentMsg, paymentBufferSize),
	}
	rand.Seed(time.Now().UnixNano())
	_, _, err := pm.cfg.db.loadLastPaymentInfo()
	if err != nil {
		if errors.Is(err, errs.ValueNotFound) {
			err = pm.cfg.db.persistLastPaymentInfo(0, 0)
			if err != nil {
				return nil, err
			}
		} else {
			return nil, err
		}
	}
	_, err = pm.cfg.db.loadLastPaymentCreatedOn()
	if err != nil {
		if errors.Is(err, errs.ValueNotFound) {
			err = pm.cfg.db.persistLastPaymentCreatedOn(0)
			if err != nil {
				return nil, err
			}
		} else {
			return nil, err
		}
	}
	return pm, nil
}
func (pm *PaymentMgr) isProcessing() bool {
	pm.mtx.Lock()
	defer pm.mtx.Unlock()
	return pm.processing
}
func (pm *PaymentMgr) setProcessing(status bool) {
	pm.mtx.Lock()
	pm.processing = status
	pm.mtx.Unlock()
}
func (pm *PaymentMgr) sharePercentages(shares []*Share) (map[string]*big.Rat, error) {
	totalShares := new(big.Rat)
	tally := make(map[string]*big.Rat)
	percentages := make(map[string]*big.Rat)
	for _, share := range shares {
		totalShares = totalShares.Add(totalShares, share.Weight)
		if _, ok := tally[share.Account]; ok {
			tally[share.Account] = tally[share.Account].
				Add(tally[share.Account], share.Weight)
			continue
		}
		tally[share.Account] = share.Weight
	}
	for account, shareCount := range tally {
		if tally[account].Cmp(ZeroRat) == 0 {
			return nil, errs.PoolError(errs.DivideByZero, "division by zero")
		}
		accPercent := new(big.Rat).Quo(shareCount, totalShares)
		percentages[account] = accPercent
	}
	return percentages, nil
}
func (pm *PaymentMgr) PPSSharePercentages(workCreatedOn int64) (map[string]*big.Rat, error) {
	shares, err := pm.cfg.db.ppsEligibleShares(workCreatedOn)
	if err != nil {
		return nil, err
	}
	if len(shares) == 0 {
		return make(map[string]*big.Rat), nil
	}
	percentages, err := pm.sharePercentages(shares)
	if err != nil {
		return nil, err
	}
	return percentages, nil
}
func (pm *PaymentMgr) PPLNSSharePercentages() (map[string]*big.Rat, error) {
	now := time.Now()
	min := now.Add(-pm.cfg.LastNPeriod)
	shares, err := pm.cfg.db.pplnsEligibleShares(min.UnixNano())
	if err != nil {
		return nil, err
	}
	if len(shares) == 0 {
		return make(map[string]*big.Rat), nil
	}
	percentages, err := pm.sharePercentages(shares)
	if err != nil {
		return nil, err
	}
	return percentages, nil
}
func (pm *PaymentMgr) calculatePayments(ratios map[string]*big.Rat, source *PaymentSource,
	total dcrutil.Amount, poolFee float64, height uint32, estMaturity uint32) ([]*Payment, int64, error) {
	funcName := "calculatePayments"
	if len(ratios) == 0 {
		desc := fmt.Sprintf("%s: valid share ratios required to "+
			"generate payments", funcName)
		return nil, 0, errs.PoolError(errs.ShareRatio, desc)
	}
	fee := total.MulF64(poolFee)
	amtSansFees := total - fee
	sansFees := new(big.Rat).SetInt64(int64(amtSansFees))
	paymentTotal := dcrutil.Amount(0)
	dustAmts := make([]dcrutil.Amount, 0)
	payments := make([]*Payment, 0)
	for account, ratio := range ratios {
		amtRat := new(big.Rat).Mul(sansFees, ratio)
		amtI, accuracy := new(big.Float).SetRat(amtRat).Int64()
		amt := dcrutil.Amount(amtI)
		if accuracy > 0 {
			amt -= dcrutil.Amount(1)
		}
		paymentTotal += amt
		if txrules.IsDustAmount(amt, txsizes.P2PKHOutputSize,
			txrules.DefaultRelayFeePerKb) {
			dustAmts = append(dustAmts, amt)
		} else {
			payments = append(payments, NewPayment(account, source, amt, height,
				estMaturity))
		}
	}
	if amtSansFees < paymentTotal {
		diff := paymentTotal - amtSansFees
		desc := fmt.Sprintf("%s: total payments (%s) is greater than "+
			"the remaining coinbase amount after fees (%s). Difference is %s",
			funcName, paymentTotal, amtSansFees, diff)
		return nil, 0, errs.PoolError(errs.PaymentSource, desc)
	}
	var dustTotal dcrutil.Amount
	for _, amt := range dustAmts {
		dustTotal += amt
	}
	feePayment := NewPayment(PoolFeesK, source, fee+dustTotal, height, estMaturity)
	payments = append(payments, feePayment)
	return payments, feePayment.CreatedOn, nil
}
func (pm *PaymentMgr) payPerShare(source *PaymentSource, amt dcrutil.Amount, height uint32, workCreatedOn int64) error {
	percentages, err := pm.PPSSharePercentages(workCreatedOn)
	if err != nil {
		return err
	}
	estMaturity := height + uint32(pm.cfg.ActiveNet.CoinbaseMaturity)
	payments, lastPmtCreatedOn, err := pm.calculatePayments(percentages,
		source, amt, pm.cfg.PoolFee, height, estMaturity)
	if err != nil {
		return err
	}
	for _, payment := range payments {
		err := pm.cfg.db.PersistPayment(payment)
		if err != nil {
			return err
		}
	}
	err = pm.cfg.db.persistLastPaymentCreatedOn(lastPmtCreatedOn)
	if err != nil {
		return err
	}
	return pm.cfg.db.pruneShares(workCreatedOn)
}
func (pm *PaymentMgr) payPerLastNShares(source *PaymentSource, amt dcrutil.Amount, height uint32) error {
	percentages, err := pm.PPLNSSharePercentages()
	if err != nil {
		return err
	}
	estMaturity := height + uint32(pm.cfg.ActiveNet.CoinbaseMaturity)
	payments, lastPmtCreatedOn, err := pm.calculatePayments(percentages,
		source, amt, pm.cfg.PoolFee, height, estMaturity)
	if err != nil {
		return err
	}
	for _, payment := range payments {
		err := pm.cfg.db.PersistPayment(payment)
		if err != nil {
			return err
		}
	}
	err = pm.cfg.db.persistLastPaymentCreatedOn(lastPmtCreatedOn)
	if err != nil {
		return err
	}
	minNano := time.Now().Add(-pm.cfg.LastNPeriod).UnixNano()
	return pm.cfg.db.pruneShares(minNano)
}
func (pm *PaymentMgr) generatePayments(height uint32, source *PaymentSource, amt dcrutil.Amount, workCreatedOn int64) error {
	switch pm.cfg.PaymentMethod {
	case PPS:
		return pm.payPerShare(source, amt, height, workCreatedOn)
	case PPLNS:
		return pm.payPerLastNShares(source, amt, height)
	default:
		return fmt.Errorf("unknown payment method provided %v", pm.cfg.PaymentMethod)
	}
}
func (pm *PaymentMgr) pruneOrphanedPayments(ctx context.Context, pmts map[string][]*Payment) (map[string][]*Payment, error) {
	toDelete := make([]string, 0, len(pmts))
	for key := range pmts {
		blockHash, err := chainhash.NewHashFromStr(key)
		if err != nil {
			desc := fmt.Sprintf("unable to generate hash: %v", err)
			return nil, errs.PoolError(errs.CreateHash, desc)
		}
		confs, err := pm.cfg.GetBlockConfirmations(ctx, blockHash)
		if err != nil {
			return nil, err
		}
		if confs <= 0 {
			toDelete = append(toDelete, key)
		}
	}
	for _, k := range toDelete {
		delete(pmts, k)
	}
	return pmts, nil
}
func (pm *PaymentMgr) applyTxFees(inputs []chainjson.TransactionInput, outputs map[string]dcrutil.Amount,
	tOut dcrutil.Amount, feeAddr dcrutil.Address) (dcrutil.Amount, dcrutil.Amount, error) {
	funcName := "applyTxFees"
	if len(inputs) == 0 {
		desc := fmt.Sprint("%s: cannot create a payout transaction "+
			"without a tx input", funcName)
		return 0, 0, errs.PoolError(errs.TxIn, desc)
	}
	if len(outputs) == 0 {
		desc := fmt.Sprint("%s:cannot create a payout transaction "+
			"without a tx output", funcName)
		return 0, 0, errs.PoolError(errs.TxOut, desc)
	}
	inSizes := make([]int, len(inputs))
	for range inputs {
		inSizes = append(inSizes, txsizes.RedeemP2PKHSigScriptSize)
	}
	outSizes := make([]int, len(outputs))
	for range outputs {
		outSizes = append(outSizes, txsizes.P2PKHOutputSize)
	}
	changeScriptSize := 0
	estSize := txsizes.EstimateSerializeSizeFromScriptSizes(inSizes, outSizes,
		changeScriptSize)
	estFee := txrules.FeeForSerializeSize(txrules.DefaultRelayFeePerKb, estSize)
	sansFees := tOut - estFee
	for addr, v := range outputs {
		if addr == feeAddr.String() {
			continue
		}
		ratio := float64(int64(sansFees)) / float64(int64(v))
		outFee := estFee.MulF64(ratio)
		outputs[addr] -= outFee
	}
	return sansFees, estFee, nil
}
func (pm *PaymentMgr) confirmCoinbases(ctx context.Context, txB TxBroadcaster, txHashes map[chainhash.Hash]uint32) error {
	funcName := "confirmCoinbases"
	maxSpendableConfs := int32(pm.cfg.ActiveNet.CoinbaseMaturity) + 1
	keys := make([]chainhash.Hash, 0, len(txHashes))
	for hash := range txHashes {
		keys = append(keys, hash)
	}
	for idx := range keys {
		select {
		case <-ctx.Done():
			log.Tracef("%s: context cancelled fetching tx "+
				"confirmations", funcName)
			return errs.ContextCancelled
		default:
			req := &walletrpc.GetTransactionRequest{
				TransactionHash: keys[idx][:],
			}
			resp, err := txB.GetTransaction(ctx, req)
			if err != nil {
				state, ok := status.FromError(err)
				if ok {
					if state.Code() != codes.NotFound {
						return err
					}
					log.Errorf("%s: transaction %v not found by wallet",
						funcName, err)
					continue
				}
				return err
			}
			if resp.Confirmations >= maxSpendableConfs {
				delete(txHashes, keys[idx])
			}
		}
	}
	if len(txHashes) != 0 {
		log.Tracef("%d coinbase(s) are not confirmed:", len(txHashes))
		for coinbase := range txHashes {
			log.Trace(coinbase)
		}
		return errs.TxConf
	}
	return nil
}
func fetchRescanResponse(ctx context.Context, rescanSource func() (*walletrpc.RescanResponse, error)) (*walletrpc.RescanResponse, error) {
	funcName := "fetchRescanResponse"
	respCh := make(chan *rescanMsg)
	go func(ch chan *rescanMsg) {
		resp, err := rescanSource()
		ch <- &rescanMsg{
			resp: resp,
			err:  err,
		}
	}(respCh)
	select {
	case <-ctx.Done():
		log.Tracef("%s: context cancelled fetching rescan response", funcName)
		return nil, errs.ContextCancelled
	case msg := <-respCh:
		close(respCh)
		if msg.err != nil {
			desc := fmt.Sprintf("%s: unable fetch wallet rescan response, %s",
				funcName, msg.err)
			return nil, errs.PoolError(errs.Rescan, desc)
		}
		return msg.resp, nil
	}
}
func (pm *PaymentMgr) monitorRescan(ctx context.Context, rescanSource walletrpc.WalletService_RescanClient, height int32) error {
	funcName := "monitorRescan"
	for {
		resp, err := fetchRescanResponse(ctx, rescanSource.Recv)
		if err != nil {
			if errors.Is(err, errs.ContextCancelled) {
				desc := fmt.Sprintf("%s: cancelled wallet rescan", funcName)
				return errs.PoolError(errs.ContextCancelled, desc)
			}
			return err
		}
		if resp.RescannedThrough >= height {
			log.Infof("wallet rescanned through height #%d", height)
			return nil
		}
	}
}
func (pm *PaymentMgr) generatePayoutTxDetails(ctx context.Context, txC TxCreator, feeAddr dcrutil.Address, payments map[string][]*Payment, treasuryActive bool) ([]chainjson.TransactionInput,
	map[chainhash.Hash]uint32, map[string]dcrutil.Amount, dcrutil.Amount, error) {
	funcName := "generatePayoutTxDetails"
	coinbaseIndex := uint32(1)
	if !treasuryActive {
		coinbaseIndex = 2
	}
	var tIn, tOut dcrutil.Amount
	inputs := make([]chainjson.TransactionInput, 0)
	inputTxHashes := make(map[chainhash.Hash]uint32)
	outputs := make(map[string]dcrutil.Amount)
	for _, pmtSet := range payments {
		coinbaseTx := pmtSet[0].Source.Coinbase
		txHash, err := chainhash.NewHashFromStr(coinbaseTx)
		if err != nil {
			desc := fmt.Sprintf("%s: unable to create tx hash: %v",
				funcName, err)
			return nil, nil, nil, 0, errs.PoolError(errs.CreateHash, desc)
		}
		txOutResult, err := txC.GetTxOut(ctx, txHash, coinbaseIndex, false)
		if err != nil {
			desc := fmt.Sprintf("%s: unable to find tx output: %v",
				funcName, err)
			return nil, nil, nil, 0, errs.PoolError(errs.TxOut, desc)
		}
		if txOutResult.Confirmations < int64(pm.cfg.ActiveNet.CoinbaseMaturity+1) {
			desc := fmt.Sprintf("%s: referenced coinbase at "+
				"index %d for tx %v is not spendable", funcName,
				coinbaseIndex, txHash.String())
			return nil, nil, nil, 0, errs.PoolError(errs.Coinbase, desc)
		}
		in := chainjson.TransactionInput{
			Amount: txOutResult.Value,
			Txid:   txHash.String(),
			Vout:   coinbaseIndex,
			Tree:   wire.TxTreeRegular,
		}
		inputs = append(inputs, in)
		inputTxHashes[*txHash] = pmtSet[0].Height
		prevOutV, err := dcrutil.NewAmount(in.Amount)
		if err != nil {
			desc := fmt.Sprintf("%s: unable create the input amount: %v",
				funcName, err)
			return nil, nil, nil, 0, errs.PoolError(errs.CreateAmount, desc)
		}
		tIn += prevOutV
		for _, pmt := range pmtSet {
			if pmt.Account == PoolFeesK {
				_, ok := outputs[feeAddr.String()]
				if !ok {
					outputs[feeAddr.String()] = pmt.Amount
					tOut += pmt.Amount
					continue
				}
				outputs[feeAddr.String()] += pmt.Amount
				tOut += pmt.Amount
				continue
			}
			acc, err := pm.cfg.db.fetchAccount(pmt.Account)
			if err != nil {
				return nil, nil, nil, 0, err
			}
			_, ok := outputs[acc.Address]
			if !ok {
				outputs[acc.Address] = pmt.Amount
				tOut += pmt.Amount
				continue
			}
			outputs[acc.Address] += pmt.Amount
			tOut += pmt.Amount
		}
	}
	if tOut > tIn {
		desc := fmt.Sprintf("%s: total output values for the "+
			"transaction (%s) is greater than the provided inputs (%s)",
			funcName, tOut, tIn)
		return nil, nil, nil, 0, errs.PoolError(errs.CreateTx, desc)
	}
	diff := tIn - tOut
	if diff > maxRoundingDiff {
		desc := fmt.Sprintf("%s: difference between total output "+
			"values and the provided inputs (%s) exceeds the maximum "+
			"allowed for rounding errors (%s)", funcName, diff, maxRoundingDiff)
		return nil, nil, nil, 0, errs.PoolError(errs.CreateTx, desc)
	}
	return inputs, inputTxHashes, outputs, tOut, nil
}
func (pm *PaymentMgr) payDividends(ctx context.Context, height uint32, treasuryActive bool) error {
	funcName := "payDividends"
	mPmts, err := pm.cfg.db.maturePendingPayments(height)
	if err != nil {
		return err
	}
	if len(mPmts) == 0 {
		return nil
	}
	if pm.isProcessing() {
		log.Info("payment processing already in progress, terminating")
		return nil
	}
	pm.setProcessing(true)
	defer pm.setProcessing(false)
	txB := pm.cfg.FetchTxBroadcaster()
	if txB == nil {
		desc := fmt.Sprintf("%s: tx broadcaster cannot be nil", funcName)
		return errs.PoolError(errs.Disconnected, desc)
	}
	pCtx, pCancel := context.WithTimeout(ctx, pm.cfg.CoinbaseConfTimeout)
	defer pCancel()
	txConfCount := atomic.LoadUint32(&pm.failedTxConfs)
	if txConfCount >= maxTxConfThreshold {
		beginHeight := uint32(0)
		pm.mtx.Lock()
		if len(pm.txConfHashes) == 0 {
			pm.mtx.Unlock()
			desc := fmt.Sprintf("%s: no tx conf hashes to rescan for",
				funcName)
			return errs.PoolError(errs.TxConf, desc)
		}
		for _, height := range pm.txConfHashes {
			if beginHeight == 0 {
				beginHeight = height
			}
			if beginHeight > height {
				beginHeight = height
			}
		}
		pm.mtx.Unlock()
		rescanBeginHeight := beginHeight - uint32(pm.cfg.ActiveNet.CoinbaseMaturity)
		log.Infof("wallet rescanning from height #%d", rescanBeginHeight)
		rescanReq := &walletrpc.RescanRequest{
			BeginHeight: int32(rescanBeginHeight),
		}
		rescanSource, err := txB.Rescan(pCtx, rescanReq)
		if err != nil {
			desc := fmt.Sprintf("%s: rescan source cannot be nil", funcName)
			return errs.PoolError(errs.Rescan, desc)
		}
		err = pm.monitorRescan(pCtx, rescanSource, int32(height))
		if err != nil {
			return err
		}
	}
	txC := pm.cfg.FetchTxCreator()
	if txC == nil {
		desc := fmt.Sprintf("%s: tx creator cannot be nil", funcName)
		return errs.PoolError(errs.Disconnected, desc)
	}
	pmts, err := pm.pruneOrphanedPayments(ctx, mPmts)
	if err != nil {
		return err
	}
	feeAddr := pm.cfg.PoolFeeAddrs[rand.Intn(len(pm.cfg.PoolFeeAddrs))]
	inputs, inputTxHashes, outputs, tOut, err :=
		pm.generatePayoutTxDetails(ctx, txC, feeAddr, pmts, treasuryActive)
	if err != nil {
		return err
	}
	_, estFee, err := pm.applyTxFees(inputs, outputs, tOut, feeAddr)
	if err != nil {
		return err
	}
	outs := make(map[dcrutil.Address]dcrutil.Amount, len(outputs))
	for sAddr, amt := range outputs {
		addr, err := dcrutil.DecodeAddress(sAddr, pm.cfg.ActiveNet)
		if err != nil {
			desc := fmt.Sprintf("%s: unable to decode payout address: %v",
				funcName, err)
			return errs.PoolError(errs.Decode, desc)
		}
		outs[addr] = amt
	}
	err = pm.confirmCoinbases(pCtx, txB, inputTxHashes)
	if err != nil {
		atomic.AddUint32(&pm.failedTxConfs, 1)
		pm.mtx.Lock()
		for k, v := range inputTxHashes {
			pm.txConfHashes[k] = v
		}
		pm.mtx.Unlock()
		if !errors.Is(err, errs.ContextCancelled) {
			return err
		}
		return nil
	}
	tx, err := txC.CreateRawTransaction(ctx, inputs, outs, nil, nil)
	if err != nil {
		desc := fmt.Sprintf("%s: unable to create transaction: %v",
			funcName, err)
		return errs.PoolError(errs.CreateTx, desc)
	}
	txBytes, err := tx.Bytes()
	if err != nil {
		return err
	}
	signTxReq := &walletrpc.SignTransactionRequest{
		SerializedTransaction: txBytes,
		Passphrase:            []byte(pm.cfg.WalletPass),
	}
	signedTxResp, err := txB.SignTransaction(ctx, signTxReq)
	if err != nil {
		desc := fmt.Sprintf("%s: unable to sign transaction: %v",
			funcName, err)
		return errs.PoolError(errs.SignTx, desc)
	}
	pubTxReq := &walletrpc.PublishTransactionRequest{
		SignedTransaction: signedTxResp.Transaction,
	}
	pubTxResp, err := txB.PublishTransaction(ctx, pubTxReq)
	if err != nil {
		desc := fmt.Sprintf("%s: unable to publish transaction: %v",
			funcName, err)
		return errs.PoolError(errs.PublishTx, desc)
	}
	txid, err := chainhash.NewHash(pubTxResp.TransactionHash)
	if err != nil {
		desc := fmt.Sprintf("unable to create transaction hash: %v", err)
		return errs.PoolError(errs.CreateHash, desc)
	}
	fees := outputs[feeAddr.String()]
	log.Infof("paid a total of %v in tx %s, including %v in pool fees. "+
		"Tx fee: %v", tOut, txid.String(), fees, estFee)
	for _, set := range pmts {
		for _, pmt := range set {
			pmt.PaidOnHeight = height
			pmt.TransactionID = txid.String()
			err := pm.cfg.db.updatePayment(pmt)
			if err != nil {
				desc := fmt.Sprintf("%s: unable to update payment: %v",
					funcName, err)
				return errs.PoolError(errs.PersistEntry, desc)
			}
			err = pm.cfg.db.ArchivePayment(pmt)
			if err != nil {
				desc := fmt.Sprintf("%s: unable to archive payment: %v",
					funcName, err)
				return errs.PoolError(errs.PersistEntry, desc)
			}
		}
	}
	err = pm.cfg.db.persistLastPaymentInfo(height, time.Now().UnixNano())
	if err != nil {
		return err
	}
	atomic.StoreUint32(&pm.failedTxConfs, 0)
	pm.mtx.Lock()
	pm.txConfHashes = make(map[chainhash.Hash]uint32)
	pm.mtx.Unlock()
	return nil
}
func (pm *PaymentMgr) processPayments(msg *paymentMsg) {
	pm.paymentCh <- msg
}