clirewards

var RootCmd = &cobra.Command{
	Use:   "rewards",
	Short: "reward-system operator toolchain (calculate & distribute rewards)",
	Long: `
This plural 'rewards' command is the reward-system OPERATOR toolchain that
runs the daily reward distribution: collect surveys/transports/bandwidth,
fetch uptimes, calculate per-visor shares, and serve the metrics UI.

Node operators who only want to set their own reward address or check their
payout history want the singular command instead: 'skywire cli reward'.

Typical manual pipeline:
  Collect surveys:  skywire-cli log
  Fetch uptimes:    skywire-cli ut > ut.txt
  Process rewards:  skywire-cli rewards --process

The 'rewards run' subcommand performs the whole hourly cycle in-process
(the systemd unit calls it); the bare 'rewards' command calculates a single
day from data already on disk.

Architectures:
` + fmt.Sprintf("%v", append(rewards.Architectures, "null", "all")) + `

`,
	Run: func(_ *cobra.Command, _ []string) {
		var err error
		if log == nil {
			log = logging.MustGetLogger("rewards")
		}
		if logLvl != "" {
			if lvl, err := logging.LevelFromString(logLvl); err == nil {
				logging.SetLevel(lvl)
			}
		}

		if processRewards {
			runRewardProcessing()
			return
		}

		sConf, err := script.Echo(string(deployment.ServicesJSON)).JQ(`.prod  | del(.stun_servers)`).Bytes()
		if err != nil {
			log.Fatal("error parsing json with jq:\n", err)
		}
		dConf, err := script.Echo(string(deployment.ServicesJSON)).JQ(`.prod`).Bytes()
		if err != nil {
			log.Fatal("error parsing json with jq:\n", err)
		}

		wDate, err = time.Parse("2006-01-02", wdate)
		if err != nil {
			log.Fatal("Error parsing date:", err)
			return
		}
		_, err = os.Stat(hwSurveyPath)
		if os.IsNotExist(err) {
			log.Fatal("the path to the surveys does not exist\n", err, "\nfetch the surveys with:\n$ skywire-cli log")
		}

		// Initialize GeoIP database for regional saturation scaling
		var geoDB *geoip2.Reader
		if saturationExponent < 1.0 {
			geoDB, err = geoip2.OpenBytes(geoip.EmbeddedDB())
			if err != nil {
				log.Warn("Failed to load embedded GeoIP database, disabling regional saturation: ", err)
				saturationExponent = 1.0
			} else {
				defer func() { _ = geoDB.Close() }()
			}
		}

		_, err = os.Stat(utfile)
		if os.IsNotExist(err) {
			log.Fatal("uptime tracker data file not found\n", err, "\nfetch the uptime tracker data with:\n$ skywire-cli ut > ut.txt")
		}

		disallowedMap := make(map[string]struct{})
		for _, arch := range disallowArchitectures {
			disallowedMap[arch] = struct{}{}
		}

		allowArchMap1 := make(map[string]struct{})
		allowArchMap2 := make(map[string]struct{})
		supportedArchitecturesMap := make(map[string]struct{})
		for _, arch := range rewards.Architectures {
			supportedArchitecturesMap[arch] = struct{}{}
		}
		for _, arch := range allowArchitectures1 {
			if _, isDisallowed := disallowedMap[arch]; !isDisallowed {
				allowArchMap1[arch] = struct{}{}
			}
		}
		for _, arch := range allowArchitectures2 {
			if _, isDisallowed := disallowedMap[arch]; !isDisallowed {
				allowArchMap2[arch] = struct{}{}
			}
		}

		if !requireBandwidth {
			for arch := range allowArchMap1 {
				if _, exists := allowArchMap2[arch]; exists {
					log.Fatal("Error: Architecture cannot be specified in both pools: " + arch)
				}
			}
		}
		for arch := range allowArchMap1 {
			if _, isValid := supportedArchitecturesMap[arch]; !isValid {
				log.Fatal("Error: Architecture is not valid: ", arch)
			}
		}
		for arch := range allowArchMap2 {
			if _, isValid := supportedArchitecturesMap[arch]; !isValid {
				log.Fatal("Error: Architecture is not valid: ", arch)
			}
		}

		allowArchMapAll := make(map[string]struct{})
		if requireBandwidth {
			for arch := range allowArchMap1 {
				allowArchMapAll[arch] = struct{}{}
			}
			for arch := range allowArchMap2 {
				allowArchMapAll[arch] = struct{}{}
			}
		}

		transportMap := make(map[string]struct{})
		if requireTransports {
			transportFile := fmt.Sprintf("%s/%s_transports.txt", transportHistPath, wdate)
			if data, err := os.ReadFile(transportFile); err == nil {
				for _, line := range strings.Split(string(data), "\n") {
					line = strings.TrimSpace(line)
					if line != "" {
						transportMap[line] = struct{}{}
					}
				}

				if len(transportMap) == 0 {
					log.Warnf("Transport file %s exists but contains zero qualifying visors. "+
						"Treating as upstream data unavailability (likely transient TPD/CXO outage at tp-collect time) "+
						"and skipping the transport requirement for this calc. "+
						"If unexpected, verify tp-collect succeeded and the TPD per-key-stats endpoint is healthy "+
						"— a genuine zero would indicate a network-wide failure.", transportFile)
					requireTransports = false
				} else {
					log.Infof("Loaded %d visors with sufficient transports from %s", len(transportMap), transportFile)
				}
			} else {
				log.Warnf("Transport file not found: %s (transport requirement will be skipped)", transportFile)
				requireTransports = false
			}
		}

		bandwidthMap := make(map[string]uint64)
		var bwTransports []TransportBW // populated only for v2
		if requireBandwidth && !noBWPool {
			bwFile := fmt.Sprintf("%s/%s_bandwidth.json", transportHistPath, wdate)
			data, readErr := os.ReadFile(bwFile)
			switch {
			case readErr != nil:
				log.Warnf("Bandwidth file not found: %s (bandwidth pool will be skipped)", bwFile)
				requireBandwidth = false
			default:
				var v2 BandwidthData
				if err := json.Unmarshal(data, &v2); err == nil && v2.Version >= BandwidthDataVersion {
					bwTransports = v2.Transports
					if len(bwTransports) == 0 {

						log.Warnf("Bandwidth file %s parsed as v2 but contains zero transports. "+
							"Treating as upstream data unavailability and skipping the bandwidth pool. "+
							"Verify bw-collect succeeded and the TPD /metrics endpoint is healthy.", bwFile)
						requireBandwidth = false
					} else {
						log.Infof("Loaded v2 bandwidth data: %d transports from %s", len(bwTransports), bwFile)
					}
				} else if err := json.Unmarshal(data, &bandwidthMap); err == nil {
					if len(bandwidthMap) == 0 {
						log.Warnf("Bandwidth file %s parsed as v1 but contains zero visors. "+
							"Treating as upstream data unavailability and skipping the bandwidth pool.", bwFile)
						requireBandwidth = false
					} else {
						log.Infof("Loaded v1 bandwidth data for %d visors from %s (legacy aggregated format — no counterparty eligibility filter)", len(bandwidthMap), bwFile)
					}
				} else {
					log.Warnf("Failed to parse bandwidth file %s: %v", bwFile, err)
					requireBandwidth = false
				}
			}
		}

		// Get public keys that met uptime requirement
		var res []string
		if pubkey == "" {

			res, _ = script.File(utfile).Match(strings.TrimRight(wdate, "\n")).Column(1).Slice()
			if len(res) == 0 {
				log.Fatal("No keys achieved minimum uptime on " + wdate + " !")
			}
		} else {

			res, _ = script.File(utfile).Match(strings.TrimRight(wdate, "\n")).Column(1).Match(pubkey).Slice()
			if len(res) == 0 {
				log.Fatal("Specified key " + pubkey + "\n did not achieve minimum uptime on " + wdate + " !")
			}
		}

		// Collect eligible nodes by parsing surveys
		var nodesInfos1 []nodeinfo // pool 1 (presence in bandwidth mode, or arch pool 1 in legacy)
		var nodesInfos2 []nodeinfo // pool 2 (unused in bandwidth mode, or arch pool 2 in legacy)
		var grrInfos []nodeinfo    // ineligible nodes (for error reporting)

		for _, pk := range res {
			surveyPath := fmt.Sprintf("%s/%s/node-info.json", hwSurveyPath, pk)
			survey, parseErr := parseSurvey(surveyPath)
			if parseErr != nil {
				log.Debug(parseErr.Error())

				grrInfos = append(grrInfos, nodeinfo{PK: pk, Reason: "survey not found"})
				continue
			}

			svcconf := checkServiceConfig(surveyPath, sConf, dConf)
			ifc, macs := extractInterfaces(survey)

			ip := survey.IPAddr
			sky := survey.SkyAddr
			arch := survey.Arch
			hv := survey.SysInfo.Node.Hypervisor
			if hv == "" {
				hv = "null"
			}
			uu := survey.UUID

			_, allowed1 := allowArchMap1[arch]
			_, allowed2 := allowArchMap2[arch]
			_, _, addrErr := rewardconfig.ValidateRewardAddress(sky)

			_, hasTransports := transportMap[pk]
			meetsTransportReq := !requireTransports || hasTransports
			visorBW := bandwidthMap[pk]

			var archAllowed bool
			if requireBandwidth {
				_, archAllowed = allowArchMapAll[arch]
			} else {
				archAllowed = allowed1 || allowed2
			}

			ni := nodeinfo{
				IPAddr:     ip,
				SkyAddr:    sky,
				PK:         pk,
				Arch:       arch,
				Interfaces: ifc,
				MacAddr:    macs[0],
				UUID:       uu,
				SvcConf:    svcconf,
				HV:         hv,
				Bandwidth:  visorBW,
			}

			baseEligible := archAllowed && strings.Count(ip, ".") == 3 && uu != "" && ifc != "" && len(macs) > 0 && macs[0] != "" && hv == "null" && addrErr == nil && meetsTransportReq

			if !baseEligible {
				ni.Reason = func() string {
					switch {
					case !archAllowed:
						return arch
					case strings.Count(ip, ".") != 3:
						return ip
					case uu == "":
						return ip
					case ifc == "":
						return ifc
					case len(macs) == 0 || macs[0] == "":
						return macs[0]
					case hv != "null":
						return hv
					case addrErr != nil:
						return "Invalid Skycoin address"
					case !meetsTransportReq:
						return "No transports"
					default:
						return "Unknown reason"
					}
				}()
				if grr {
					grrInfos = append(grrInfos, ni)
				}
				continue
			}

			if geoDB != nil {
				if geoRes, geoErr := geoip.Lookup(geoDB, ip); geoErr == nil && geoRes.CountryCode != "" {
					ni.Country = geoRes.CountryCode
				} else {
					ni.Country = "XX"
				}
			}

			if requireBandwidth {
				nodesInfos1 = append(nodesInfos1, ni)
			} else {
				if allowed1 {
					nodesInfos1 = append(nodesInfos1, ni)
				}
				if allowed2 {
					nodesInfos2 = append(nodesInfos2, ni)
				}
			}
		}

		if grr {
			for _, ni := range grrInfos {
				fmt.Printf("%s, %s, %s, %.6f, %.6f, %s, %s, %s, %s \n", ni.SkyAddr, ni.PK, ni.Reason, ni.Share, ni.Reward, ni.IPAddr, ni.Arch, ni.UUID, ni.Interfaces)
			}
			return
		}

		if requireBandwidth && bwTransports != nil {
			eligibleSet := make(map[string]struct{}, len(nodesInfos1))
			for _, ni := range nodesInfos1 {
				eligibleSet[ni.PK] = struct{}{}
			}
			var creditedEdges int
			for _, tp := range bwTransports {
				if _, ok := eligibleSet[tp.A]; ok {
					bandwidthMap[tp.A] += tp.SentA
					creditedEdges++
				}
				if _, ok := eligibleSet[tp.B]; ok {
					bandwidthMap[tp.B] += tp.SentB
					creditedEdges++
				}
			}
			for i := range nodesInfos1 {
				nodesInfos1[i].Bandwidth = bandwidthMap[nodesInfos1[i].PK]
			}
			log.Infof("v2 bandwidth filter: %d eligible sender-side credits applied across %d transports",
				creditedEdges, len(bwTransports))
		}

		daysThisMonth := time.Date(wDate.Year(), wDate.Month()+1, 0, 0, 0, 0, 0, time.UTC).Day()
		daysThisYear := int(time.Date(wDate.Year(), 12, 31, 23, 59, 59, 999999999, time.UTC).Sub(time.Date(wDate.Year(), 1, 1, 0, 0, 0, 0, time.UTC)).Hours()) / 24
		monthReward := (float64(yearlyTotal) / float64(daysThisYear)) * float64(daysThisMonth)
		dayReward := monthReward / float64(daysThisMonth)
		wdate = strings.ReplaceAll(wdate, " ", "0")

		allEligible := append(nodesInfos1, nodesInfos2...)
		allIPs := make([]string, 0, len(allEligible))
		allMACs := make([]string, 0, len(allEligible))
		allUUIDs := make([]string, 0, len(allEligible))
		for _, ni := range allEligible {
			allIPs = append(allIPs, ni.IPAddr)
			allMACs = append(allMACs, ni.MacAddr)
			allUUIDs = append(allUUIDs, ni.UUID)
		}
		ipCounts := countFrequency(allIPs)
		macCounts := countFrequency(allMACs)
		uuidCounts := countFrequency(allUUIDs)

		if requireBandwidth {

			presenceBudget := dayReward
			if noBWPool {
				presenceBudget = 2 * dayReward
			}

			computePoolShares(nodesInfos1, ipCounts, macCounts)
			totalPresenceShares := computePoolRewards(nodesInfos1, presenceBudget)

			for i := range nodesInfos1 {
				nodesInfos1[i].PresenceReward = nodesInfos1[i].Reward
			}

			ipBytes := bytesPerIP(nodesInfos1, minBWThreshold)
			var totalBWShares float64
			var bwPoolCount int
			if !noBWPool {
				for _, ni := range nodesInfos1 {
					if ni.Bandwidth >= minBWThreshold {
						totalBWShares += pool2Weight(ni.Bandwidth, ipBytes[ni.IPAddr], bwSaturationExponent)
						bwPoolCount++
					}
				}
				if totalBWShares > 0 {
					for i := range nodesInfos1 {
						if nodesInfos1[i].Bandwidth >= minBWThreshold {
							weight := pool2Weight(nodesInfos1[i].Bandwidth, ipBytes[nodesInfos1[i].IPAddr], bwSaturationExponent)
							bw := weight * dayReward / totalBWShares
							nodesInfos1[i].BandwidthReward = bw
							nodesInfos1[i].Reward += bw
						}
					}
				}
			}

			// Sum the two pools for the stats block. By construction
			// pool1Sum ≈ presenceBudget and pool2Sum ≈ dayReward (or 0
			// when noBWPool), but reporting the actual sums catches any
			// future drift in the share-aggregation math.
			var pool1Sum, pool2Sum float64
			for _, ni := range nodesInfos1 {
				pool1Sum += ni.PresenceReward
				pool2Sum += ni.BandwidthReward
			}

			if !h0 {
				fmt.Printf("date: %s\n", wdate)
				fmt.Printf("days this month: %d\n", daysThisMonth)
				fmt.Printf("days in the year: %d\n", daysThisYear)
				fmt.Printf("this month's rewards: %.6f\n", monthReward)
				fmt.Printf("reward per pool: %.6f\n", dayReward)
				if noBWPool {
					fmt.Printf("reward mode: presence only (bandwidth pool disabled, 2× presence)\n")
				} else {
					fmt.Printf("reward mode: presence + bandwidth\n")
				}
				fmt.Printf("\n--- Presence Pool (equal shares, IP/MAC dedup) ---\n")
				fmt.Printf("presence pool budget: %.6f\n", presenceBudget)
				fmt.Printf("Pool 1 Total: %.6f\n", pool1Sum)
				fmt.Printf("qualifying visors: %d\n", len(nodesInfos1))
				fmt.Printf("total presence shares: %.6f\n", totalPresenceShares)
				if totalPresenceShares > 0 {
					fmt.Printf("Skycoin Per Share (Pool 1): %.6f\n", presenceBudget/totalPresenceShares)
				}
				if noBWPool {
					fmt.Printf("\n--- Bandwidth Pool: DISABLED (budget folded into presence) ---\n")
				} else {
					fmt.Printf("\n--- Bandwidth Pool (sender-pays, per-IP weighted bytes^%.2f) ---\n", bwSaturationExponent)
					fmt.Printf("bandwidth pool budget: %.6f\n", dayReward)
					fmt.Printf("Pool 2 Total: %.6f\n", pool2Sum)
					fmt.Printf("minimum bandwidth threshold: %d bytes\n", minBWThreshold)
					fmt.Printf("qualifying visors: %d\n", bwPoolCount)
					var totalBW uint64
					for _, ni := range nodesInfos1 {
						if ni.Bandwidth >= minBWThreshold {
							totalBW += ni.Bandwidth
						}
					}
					fmt.Printf("total network bandwidth: %s\n", formatBytes(totalBW))
					fmt.Printf("contributing IPs: %d\n", len(ipBytes))
					fmt.Printf("bandwidth saturation exponent: %.2f\n", bwSaturationExponent)
					if totalBWShares > 0 {

						fmt.Printf("Skycoin Per Pool 2 Weight Unit: %.12f\n", dayReward/totalBWShares)
						if bwSaturationExponent == 1.0 {
							fmt.Printf("Skycoin Per GB (Pool 2): %.6f\n", dayReward/totalBWShares*1024*1024*1024)
						}
					}
				}
				fmt.Printf("\nUnique mac addresses: %d\n", len(macCounts))
				fmt.Printf("Unique IP Addresses: %d\n", len(ipCounts))
				fmt.Printf("Unique UUIDs: %d\n", len(uuidCounts))
				printSaturationStats(nodesInfos1, presenceBudget, totalPresenceShares)
			}

			if !h1 {
				fmt.Println("Skycoin Address, Skywire Public Key, Presence Share, Bandwidth (bytes), Total Reward SKY, IP, Architecture, UUID, Interfaces, Country, XPub")
				for _, ni := range nodesInfos1 {
					resolved := resolveRewardAddress(ni.SkyAddr)
					xpub := ""
					if strings.HasPrefix(ni.SkyAddr, "xpub") {
						xpub = ni.SkyAddr
					}
					fmt.Printf("%s, %s, %.6f, %d, %.6f, %s, %s, %s, %s, %s, %s \n", resolved, ni.PK, ni.Share, ni.Bandwidth, ni.Reward, ni.IPAddr, ni.Arch, ni.UUID, ni.Interfaces, ni.Country, xpub)
				}
			}

			sortedAddrs := sumRewardsByAddress(nodesInfos1)
			if !h0 {
				total := 0.0
				for _, a := range sortedAddrs {
					total += a.Reward
				}
				fmt.Printf("\nTotal Reward Amount (both pools): %.6f\n", total)
			}
			if !h2 {
				fmt.Println("Skycoin Address, Reward Amount")
				for _, a := range sortedAddrs {
					fmt.Printf("%s, %.6f\n", resolveRewardAddress(a.SkyAddr), a.Reward)
				}
			}

			if pubkey == "" {
				if err := writePerPoolFiles(transportHistPath, wdate, nodesInfos1, bwSaturationExponent, minBWThreshold); err != nil {
					log.Warnf("Failed to write per-pool detail files: %v", err)
				}
			}
		} else {

			computePoolShares(nodesInfos1, ipCounts, macCounts)
			computePoolShares(nodesInfos2, ipCounts, macCounts)
			totalShares1 := computePoolRewards(nodesInfos1, dayReward)
			totalShares2 := computePoolRewards(nodesInfos2, dayReward)

			if !h0 {
				fmt.Printf("date: %s\n", wdate)
				fmt.Printf("days this month: %d\n", daysThisMonth)
				fmt.Printf("days in the year: %d\n", daysThisYear)
				fmt.Printf("this month's rewards: %.6f\n", monthReward)
				fmt.Printf("reward total per pool: %.6f\n", dayReward)
				fmt.Printf("Visors meeting uptime & other requirements (Pool 1): %d\n", len(nodesInfos1))
				fmt.Printf("Visors meeting uptime & other requirements (Pool 2): %d\n", len(nodesInfos2))
				fmt.Printf("Unique mac addresses for first interface after lo: %d\n", len(macCounts))
				fmt.Printf("Unique IP Addresses: %d\n", len(ipCounts))
				fmt.Printf("Unique UUIDs: %d\n", len(uuidCounts))
				if saturationExponent < 1.0 {
					fmt.Printf("Regional saturation exponent: %.2f\n", saturationExponent)
				}
				fmt.Printf("Total valid shares (Pool 1): %.6f\n", totalShares1)
				fmt.Printf("Total valid shares (Pool 2): %.6f\n", totalShares2)
				if totalShares1 != 0 {
					fmt.Printf("Skycoin Per Share (Pool 1): %.6f\n", dayReward/totalShares1)
				} else {
					fmt.Printf("Skycoin Per Share (Pool 1): 0\n")
				}
				if totalShares2 != 0 {
					fmt.Printf("Skycoin Per Share (Pool 2): %.6f\n", dayReward/totalShares2)
				} else {
					fmt.Printf("Skycoin Per Share (Pool 2): 0\n")
				}
			}

			combinedNodes := append(nodesInfos1, nodesInfos2...)
			if !h1 {
				fmt.Println("Skycoin Address, Skywire Public Key, Reward Shares, Reward SKY Amount, IP, Architecture, UUID, Interfaces, Country, XPub")
				for _, ni := range combinedNodes {
					resolved := resolveRewardAddress(ni.SkyAddr)
					xpub := ""
					if strings.HasPrefix(ni.SkyAddr, "xpub") {
						xpub = ni.SkyAddr
					}
					fmt.Printf("%s, %s, %.6f, %.6f, %s, %s, %s, %s, %s, %s \n", resolved, ni.PK, ni.Share, ni.Reward, ni.IPAddr, ni.Arch, ni.UUID, ni.Interfaces, ni.Country, xpub)
				}
			}

			sortedAddrs := sumRewardsByAddress(combinedNodes)
			if !h0 {
				total := 0.0
				for _, a := range sortedAddrs {
					total += a.Reward
				}
				fmt.Printf("Total Reward Amount: %.6f\n", total)
			}
			if !h2 {
				fmt.Println("Skycoin Address, Reward Amount")
				for _, a := range sortedAddrs {
					fmt.Printf("%s, %.6f\n", resolveRewardAddress(a.SkyAddr), a.Reward)
				}
			}
		}

	},
}