ytsync/manager/setup.go

package manager

import (
	"fmt"
	"math"
	"strconv"
	"time"

	"github.com/lbryio/lbry.go/v2/extras/errors"
	"github.com/lbryio/lbry.go/v2/extras/jsonrpc"
	"github.com/lbryio/lbry.go/v2/extras/util"
	"github.com/lbryio/ytsync/v5/timing"
	logUtils "github.com/lbryio/ytsync/v5/util"
	"github.com/lbryio/ytsync/v5/ytapi"

	"github.com/lbryio/ytsync/v5/tags_manager"
	"github.com/lbryio/ytsync/v5/thumbs"

	"github.com/shopspring/decimal"
	log "github.com/sirupsen/logrus"
)

func (s *Sync) enableAddressReuse() error {
	accountsResponse, err := s.daemon.AccountList(1, 50)
	if err != nil {
		return errors.Err(err)
	}
	accounts := make([]jsonrpc.Account, 0, len(accountsResponse.Items))
	ledger := "lbc_mainnet"
	if logUtils.IsRegTest() {
		ledger = "lbc_regtest"
	}
	for _, a := range accountsResponse.Items {
		if *a.Ledger == ledger {
			accounts = append(accounts, a)
		}
	}

	for _, a := range accounts {
		_, err = s.daemon.AccountSet(a.ID, jsonrpc.AccountSettings{
			ChangeMaxUses:    util.PtrToInt(1000),
			ReceivingMaxUses: util.PtrToInt(100),
		})
		if err != nil {
			return errors.Err(err)
		}
	}
	return nil
}
func (s *Sync) walletSetup() error {
	start := time.Now()
	defer func(start time.Time) {
		timing.TimedComponent("walletSetup").Add(time.Since(start))
	}(start)
	//prevent unnecessary concurrent execution and publishing while refilling/reallocating UTXOs
	s.walletMux.Lock()
	defer s.walletMux.Unlock()
	err := s.ensureChannelOwnership()
	if err != nil {
		return err
	}

	balanceResp, err := s.daemon.AccountBalance(nil)
	if err != nil {
		return err
	} else if balanceResp == nil {
		return errors.Err("no response")
	}
	balance, err := strconv.ParseFloat(balanceResp.Available.String(), 64)
	if err != nil {
		return errors.Err(err)
	}
	log.Debugf("Starting balance is %.4f", balance)

	videosOnYoutube, err := ytapi.CountVideosInChannel(s.YoutubeChannelID)
	if err != nil {
		return err
	}

	log.Debugf("Source channel has %d videos", videosOnYoutube)
	if videosOnYoutube == 0 {
		return nil
	}

	s.syncedVideosMux.RLock()
	publishedCount := 0
	notUpgradedCount := 0
	failedCount := 0
	for _, sv := range s.syncedVideos {
		if sv.Published {
			publishedCount++
			if sv.MetadataVersion < 2 {
				notUpgradedCount++
			}
		} else {
			failedCount++
		}
	}
	s.syncedVideosMux.RUnlock()

	log.Debugf("We already allocated credits for %d published videos and %d failed videos", publishedCount, failedCount)

	if videosOnYoutube > s.Manager.videosLimit {
		videosOnYoutube = s.Manager.videosLimit
	}
	unallocatedVideos := videosOnYoutube - (publishedCount + failedCount)
	channelFee := channelClaimAmount
	channelAlreadyClaimed := s.lbryChannelID != ""
	if channelAlreadyClaimed {
		channelFee = 0.0
	}
	requiredBalance := float64(unallocatedVideos)*(publishAmount+estimatedMaxTxFee) + channelFee
	if s.Manager.SyncFlags.UpgradeMetadata {
		requiredBalance += float64(notUpgradedCount) * 0.001
	}

	refillAmount := 0.0
	if balance < requiredBalance || balance < minimumAccountBalance {
		refillAmount = math.Max(math.Max(requiredBalance-balance, minimumAccountBalance-balance), minimumRefillAmount)
	}

	if s.Refill > 0 {
		refillAmount += float64(s.Refill)
	}

	if refillAmount > 0 {
		err := s.addCredits(refillAmount)
		if err != nil {
			return errors.Err(err)
		}
	}

	claimAddress, err := s.daemon.AddressList(nil, nil, 1, 20)
	if err != nil {
		return err
	} else if claimAddress == nil {
		return errors.Err("could not get an address")
	}
	s.claimAddress = string(claimAddress.Items[0].Address)
	if s.claimAddress == "" {
		return errors.Err("found blank claim address")
	}
	if s.shouldTransfer() {
		s.claimAddress = s.clientPublishAddress
	}

	err = s.ensureEnoughUTXOs()
	if err != nil {
		return err
	}

	return nil
}

func (s *Sync) getDefaultAccount() (string, error) {
	start := time.Now()
	defer func(start time.Time) {
		timing.TimedComponent("getDefaultAccount").Add(time.Since(start))
	}(start)
	if s.defaultAccountID == "" {
		accountsResponse, err := s.daemon.AccountList(1, 50)
		if err != nil {
			return "", errors.Err(err)
		}
		ledger := "lbc_mainnet"
		if logUtils.IsRegTest() {
			ledger = "lbc_regtest"
		}
		for _, a := range accountsResponse.Items {
			if *a.Ledger == ledger {
				if a.IsDefault {
					s.defaultAccountID = a.ID
					break
				}
			}
		}

		if s.defaultAccountID == "" {
			return "", errors.Err("No default account found")
		}
	}
	return s.defaultAccountID, nil
}

func (s *Sync) ensureEnoughUTXOs() error {
	start := time.Now()
	defer func(start time.Time) {
		timing.TimedComponent("ensureEnoughUTXOs").Add(time.Since(start))
	}(start)
	defaultAccount, err := s.getDefaultAccount()
	if err != nil {
		return err
	}

	utxolist, err := s.daemon.UTXOList(&defaultAccount, 1, 10000)
	if err != nil {
		return err
	} else if utxolist == nil {
		return errors.Err("no response")
	}

	target := 40
	slack := int(float32(0.1) * float32(target))
	count := 0
	confirmedCount := 0

	for _, utxo := range utxolist.Items {
		amount, _ := strconv.ParseFloat(utxo.Amount, 64)
		if utxo.IsMyOutput && utxo.Type == "payment" && amount > 0.001 {
			if utxo.Confirmations > 0 {
				confirmedCount++
			}
			count++
		}
	}
	log.Infof("utxo count: %d (%d confirmed)", count, confirmedCount)
	UTXOWaitThreshold := 16
	if count < target-slack {
		balance, err := s.daemon.AccountBalance(&defaultAccount)
		if err != nil {
			return err
		} else if balance == nil {
			return errors.Err("no response")
		}

		balanceAmount, err := strconv.ParseFloat(balance.Available.String(), 64)
		if err != nil {
			return errors.Err(err)
		}
		maxUTXOs := uint64(500)
		desiredUTXOCount := uint64(math.Floor((balanceAmount) / 0.1))
		if desiredUTXOCount > maxUTXOs {
			desiredUTXOCount = maxUTXOs
		}
		if desiredUTXOCount < uint64(confirmedCount) {
			return nil
		}
		availableBalance, _ := balance.Available.Float64()
		log.Infof("Splitting balance of %.3f evenly between %d UTXOs", availableBalance, desiredUTXOCount)

		broadcastFee := 0.1
		prefillTx, err := s.daemon.AccountFund(defaultAccount, defaultAccount, fmt.Sprintf("%.4f", balanceAmount-broadcastFee), desiredUTXOCount, false)
		if err != nil {
			return err
		} else if prefillTx == nil {
			return errors.Err("no response")
		}
		if confirmedCount < UTXOWaitThreshold {
			err = s.waitForNewBlock()
			if err != nil {
				return err
			}
		}
	} else if confirmedCount < UTXOWaitThreshold {
		log.Println("Waiting for previous txns to confirm")
		err := s.waitForNewBlock()
		if err != nil {
			return err
		}
	}

	return nil
}

func (s *Sync) waitForNewBlock() error {
	defer func(start time.Time) { timing.TimedComponent("waitForNewBlock").Add(time.Since(start)) }(time.Now())

	log.Printf("regtest: %t, docker: %t", logUtils.IsRegTest(), logUtils.IsUsingDocker())
	status, err := s.daemon.Status()
	if err != nil {
		return err
	}

	for status.Wallet.Blocks == 0 || status.Wallet.BlocksBehind != 0 {
		time.Sleep(5 * time.Second)
		status, err = s.daemon.Status()
		if err != nil {
			return err
		}
	}

	currentBlock := status.Wallet.Blocks
	for i := 0; status.Wallet.Blocks <= currentBlock; i++ {
		if i%3 == 0 {
			log.Printf("Waiting for new block (%d)...", currentBlock+1)
		}
		if logUtils.IsRegTest() && logUtils.IsUsingDocker() {
			err = s.GenerateRegtestBlock()
			if err != nil {
				return err
			}
		}
		time.Sleep(10 * time.Second)
		status, err = s.daemon.Status()
		if err != nil {
			return err
		}
	}
	return nil
}

func (s *Sync) GenerateRegtestBlock() error {
	lbrycrd, err := logUtils.GetLbrycrdClient(s.LbrycrdString)
	if err != nil {
		return errors.Prefix("error getting lbrycrd client: ", err)
	}

	txs, err := lbrycrd.Generate(1)
	if err != nil {
		return errors.Prefix("error generating new block: ", err)
	}

	for _, tx := range txs {
		log.Info("Generated tx: ", tx.String())
	}
	return nil
}

func (s *Sync) ensureChannelOwnership() error {
	defer func(start time.Time) { timing.TimedComponent("ensureChannelOwnership").Add(time.Since(start)) }(time.Now())

	if s.LbryChannelName == "" {
		return errors.Err("no channel name set")
	}

	channels, err := s.daemon.ChannelList(nil, 1, 50, nil)
	if err != nil {
		return err
	} else if channels == nil {
		return errors.Err("no channel response")
	}

	var channelToUse *jsonrpc.Transaction
	if len((*channels).Items) > 0 {
		if s.lbryChannelID == "" {
			return errors.Err("this channel does not have a recorded claimID in the database. To prevent failures, updates are not supported until an entry is manually added in the database")
		}
		for _, c := range (*channels).Items {
			log.Debugf("checking listed channel %s (%s)", c.ClaimID, c.Name)
			if c.ClaimID != s.lbryChannelID {
				continue
			}
			if c.Name != s.LbryChannelName {
				return errors.Err("the channel in the wallet is different than the channel in the database")
			}
			channelToUse = &c
			break
		}
		if channelToUse == nil {
			return errors.Err("this wallet has channels but not a single one is ours! Expected claim_id: %s (%s)", s.lbryChannelID, s.LbryChannelName)
		}
	} else if s.transferState == TransferStateComplete {
		return errors.Err("the channel was transferred but appears to have been abandoned!")
	} else if s.lbryChannelID != "" {
		return errors.Err("the database has a channel recorded (%s) but nothing was found in our control", s.lbryChannelID)
	}

	channelUsesOldMetadata := false
	if channelToUse != nil {
		channelUsesOldMetadata = channelToUse.Value.GetThumbnail() == nil
		if !channelUsesOldMetadata {
			return nil
		}
	}

	channelBidAmount := channelClaimAmount

	balanceResp, err := s.daemon.AccountBalance(nil)
	if err != nil {
		return err
	} else if balanceResp == nil {
		return errors.Err("no response")
	}
	balance, err := decimal.NewFromString(balanceResp.Available.String())
	if err != nil {
		return errors.Err(err)
	}

	if balance.LessThan(decimal.NewFromFloat(channelBidAmount)) {
		err = s.addCredits(channelBidAmount + 0.3)
		if err != nil {
			return err
		}
	}

	channelInfo, channelBranding, err := ytapi.ChannelInfo(s.APIConfig.YoutubeAPIKey, s.YoutubeChannelID)
	if err != nil {
		return err
	}

	thumbnail := thumbs.GetBestThumbnail(nil) //Ignore this for now ( ChannelInfo Panics)
	thumbnailURL, err := thumbs.MirrorThumbnail(thumbnail.URL, s.YoutubeChannelID, s.Manager.GetS3AWSConfig())
	if err != nil {
		return err
	}

	var bannerURL *string
	if channelBranding.Image != nil && channelBranding.Image.BannerImageUrl != "" {
		bURL, err := thumbs.MirrorThumbnail(channelBranding.Image.BannerImageUrl, "banner-"+s.YoutubeChannelID, s.Manager.GetS3AWSConfig())
		if err != nil {
			return err
		}
		bannerURL = &bURL
	}

	var languages []string = nil
	if channelInfo.DefaultLanguage != "" {
		if channelInfo.DefaultLanguage == "iw" {
			channelInfo.DefaultLanguage = "he"
		}
		languages = []string{channelInfo.DefaultLanguage}
	}
	var locations []jsonrpc.Location = nil
	if channelInfo.Country != "" {
		locations = []jsonrpc.Location{{Country: util.PtrToString(channelInfo.Country)}}
	}
	var c *jsonrpc.TransactionSummary
	claimCreateOptions := jsonrpc.ClaimCreateOptions{
		Title:        &channelInfo.Title,
		Description:  &channelInfo.Description,
		Tags:         tags_manager.GetTagsForChannel(s.YoutubeChannelID),
		Languages:    languages,
		Locations:    locations,
		ThumbnailURL: &thumbnailURL,
	}
	if channelUsesOldMetadata {
		if s.transferState <= 1 {
			c, err = s.daemon.ChannelUpdate(s.lbryChannelID, jsonrpc.ChannelUpdateOptions{
				ClearTags:      util.PtrToBool(true),
				ClearLocations: util.PtrToBool(true),
				ClearLanguages: util.PtrToBool(true),
				ChannelCreateOptions: jsonrpc.ChannelCreateOptions{
					ClaimCreateOptions: claimCreateOptions,
					CoverURL:           bannerURL,
				},
			})
		} else {
			logUtils.SendInfoToSlack("%s (%s) has a channel with old metadata but isn't in our control anymore. Ignoring", s.LbryChannelName, s.lbryChannelID)
			return nil
		}
	} else {
		c, err = s.daemon.ChannelCreate(s.LbryChannelName, channelBidAmount, jsonrpc.ChannelCreateOptions{
			ClaimCreateOptions: claimCreateOptions,
			CoverURL:           bannerURL,
		})
	}

	if err != nil {
		return err
	}

	s.lbryChannelID = c.Outputs[0].ClaimID
	return s.Manager.apiConfig.SetChannelClaimID(s.YoutubeChannelID, s.lbryChannelID)
}

func (s *Sync) addCredits(amountToAdd float64) error {
	start := time.Now()
	defer func(start time.Time) {
		timing.TimedComponent("addCredits").Add(time.Since(start))
	}(start)
	log.Printf("Adding %f credits", amountToAdd)
	lbrycrdd, err := logUtils.GetLbrycrdClient(s.LbrycrdString)
	if err != nil {
		return err
	}

	defaultAccount, err := s.getDefaultAccount()
	if err != nil {
		return err
	}
	addressResp, err := s.daemon.AddressUnused(&defaultAccount)
	if err != nil {
		return err
	} else if addressResp == nil {
		return errors.Err("no response")
	}
	address := string(*addressResp)

	_, err = lbrycrdd.SimpleSend(address, amountToAdd)
	if err != nil {
		return err
	}

	wait := 15 * time.Second
	log.Println("Waiting " + wait.String() + " for lbryum to let us know we have the new transaction")
	time.Sleep(wait)

	return nil
}