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{}{}
		}
		supportedArchitecturesMap := make(map[string]struct{})
		for _, arch := range rewards.Architectures {
			supportedArchitecturesMap[arch] = struct{}{}
		}
		allowArchMapAll := make(map[string]struct{})
		for _, arch := range append(append([]string{}, allowArchitectures1...), allowArchitectures2...) {
			if _, isValid := supportedArchitecturesMap[arch]; !isValid {
				log.Fatal("Error: Architecture is not valid: ", arch)
			}
			if _, isDisallowed := disallowedMap[arch]; !isDisallowed {
				allowArchMapAll[arch] = struct{}{}
			}
		}

		if !requireBandwidth {
			log.Warn("No -b: the bandwidth pool is skipped and Pool 2 repeats Pool 1")
			noBWPool, bwSkipReason = true, "no -b"
		}

		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 !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 (Pool 2 repeats Pool 1)", bwFile)
				noBWPool, bwSkipReason = true, "bandwidth file not found"
			default:
				var v2 BandwidthData
				var v3 tpdstore.VisorBWDay
				if err := json.Unmarshal(data, &v3); err == nil && v3.Version == tpdstore.VisorBWVersion {

					for pk, byType := range v3.Visors {
						for typ, n := range byType {
							if typ != string(tptypes.DMSG) {
								bandwidthMap[pk] += n
							}
						}
					}
					if len(bandwidthMap) == 0 {
						log.Warnf("Bandwidth file %s parsed as v3 but contains zero visors: Pool 2 repeats Pool 1.", bwFile)
						noBWPool, bwSkipReason = true, "bandwidth file has no visors"
					} else {
						log.Infof("Loaded v3 bandwidth data for %d visors from %s (%d of %d transports left out as same-network)",
							len(bandwidthMap), bwFile, v3.SameNetworkExcluded, v3.Transports)
					}
				} else 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: Pool 2 repeats Pool 1. "+
							"Verify bw-collect succeeded and the TPD /metrics endpoint is healthy.", bwFile)
						noBWPool, bwSkipReason = true, "bandwidth file has no transports"
					} 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: Pool 2 repeats Pool 1.", bwFile)
						noBWPool, bwSkipReason = true, "bandwidth file has no visors"
					} 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 (Pool 2 repeats Pool 1)", bwFile, err)
					noBWPool, bwSkipReason = true, "bandwidth file unreadable"
				}
			}
		}

		// 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 // every eligible visor: presence, and bandwidth when there is data
		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

			_, _, addrErr := rewardconfig.ValidateRewardAddress(sky)

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

			_, archAllowed := allowArchMapAll[arch]

			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"
				}
			}

			nodesInfos1 = append(nodesInfos1, 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 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 := nodesInfos1
		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)

		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 {
				reason := bwSkipReason
				if reason == "" {
					reason = "--no-bw-pool"
				}
				fmt.Printf("reward mode: presence only, Pool 2 repeats Pool 1 (%s)\n", reason)
			} 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)
			}
		}

	},
}