xmr-remote-nodes/internal/repo/monero.go

package repo

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"net"
	"slices"
	"strings"
	"time"

	"github.com/ditatompel/xmr-nodes/internal/database"

	"github.com/jmoiron/sqlx/types"
)

type MoneroRepository interface {
	Node(id int) (MoneroNode, error)
	Add(protocol string, host string, port uint) error
	Nodes(q MoneroQueryParams) (MoneroNodes, error)
	GiveJob(acceptTor int) (MoneroNode, error)
	ProcessJob(report ProbeReport, proberId int64) error
	NetFee() []NetFee
	Countries() ([]MoneroCountries, error)
	Logs(q MoneroLogQueryParams) (MoneroNodeFetchLogs, error)
}

type MoneroRepo struct {
	db *database.DB
}

func NewMoneroRepo(db *database.DB) MoneroRepository {
	return &MoneroRepo{db}
}

type MoneroNode struct {
	Id              uint           `json:"id,omitempty" db:"id"`
	Hostname        string         `json:"hostname" db:"hostname"`
	Ip              string         `json:"ip" db:"ip_addr"`
	Port            uint           `json:"port" db:"port"`
	Protocol        string         `json:"protocol" db:"protocol"`
	IsTor           bool           `json:"is_tor" db:"is_tor"`
	IsAvailable     bool           `json:"is_available" db:"is_available"`
	NetType         string         `json:"nettype" db:"nettype"`
	Height          uint           `json:"height" db:"height"`
	AdjustedTime    uint           `json:"adjusted_time" db:"adjusted_time"`
	DatabaseSize    uint           `json:"database_size" db:"database_size"`
	Difficulty      uint           `json:"difficulty" db:"difficulty"`
	Version         string         `json:"version" db:"version"`
	Status          string         `json:"status,omitempty"`
	Uptime          float64        `json:"uptime" db:"uptime"`
	EstimateFee     uint           `json:"estimate_fee" db:"estimate_fee"`
	Asn             uint           `json:"asn" db:"asn"`
	AsnName         string         `json:"asn_name" db:"asn_name"`
	CountryCode     string         `json:"cc" db:"country"`
	CountryName     string         `json:"country_name" db:"country_name"`
	City            string         `json:"city" db:"city"`
	Lat             float64        `json:"latitude" db:"lat"`
	Lon             float64        `json:"longitude" db:"lon"`
	DateEntered     uint           `json:"date_entered,omitempty" db:"date_entered"`
	LastChecked     uint           `json:"last_checked" db:"last_checked"`
	FailedCount     uint           `json:"failed_count,omitempty" db:"failed_count"`
	LastCheckStatus types.JSONText `json:"last_check_statuses" db:"last_check_status"`
	CorsCapable     bool           `json:"cors" db:"cors_capable"`
}

func (repo *MoneroRepo) Node(id int) (MoneroNode, error) {
	node := MoneroNode{}
	err := repo.db.Get(&node, `SELECT * FROM tbl_node WHERE id = ?`, id)
	if err != nil && err != sql.ErrNoRows {
		fmt.Println("WARN:", err)
		return node, errors.New("Can't get node information")
	}
	if err == sql.ErrNoRows {
		return node, errors.New("Node not found")
	}
	return node, err
}

type MoneroNodes struct {
	TotalRows   int           `json:"total_rows"`
	RowsPerPage int           `json:"rows_per_page"`
	CurrentPage int           `json:"current_page"`
	NextPage    int           `json:"next_page"`
	Items       []*MoneroNode `json:"items"`
}

type MoneroQueryParams struct {
	Host          string
	NetType       string
	Protocol      string
	CC            string // 2 letter country code
	Status        int
	Cors          int
	RowsPerPage   int
	Page          int
	SortBy        string
	SortDirection string
}

func (repo *MoneroRepo) Nodes(q MoneroQueryParams) (MoneroNodes, error) {
	queryParams := []interface{}{}
	whereQueries := []string{}
	where := ""

	if q.Host != "" {
		whereQueries = append(whereQueries, "(hostname LIKE ? OR ip_addr LIKE ?)")
		queryParams = append(queryParams, "%"+q.Host+"%")
		queryParams = append(queryParams, "%"+q.Host+"%")
	}
	if q.NetType != "any" {
		whereQueries = append(whereQueries, "nettype = ?")
		queryParams = append(queryParams, q.NetType)
	}
	if q.Protocol != "any" {
		if q.Protocol == "tor" {
			whereQueries = append(whereQueries, "is_tor = ?")
			queryParams = append(queryParams, 1)
		} else {
			whereQueries = append(whereQueries, "(protocol = ? AND is_tor = ?)")
			queryParams = append(queryParams, q.Protocol)
			queryParams = append(queryParams, 0)
		}
	}
	if q.CC != "any" {
		whereQueries = append(whereQueries, "country = ?")
		if q.CC == "UNKNOWN" {
			queryParams = append(queryParams, "")
		} else {
			queryParams = append(queryParams, q.CC)
		}
	}
	if q.Status != -1 {
		whereQueries = append(whereQueries, "is_available = ?")
		queryParams = append(queryParams, q.Status)
	}
	if q.Cors != -1 {
		whereQueries = append(whereQueries, "cors_capable = ?")
		queryParams = append(queryParams, 1)
	}

	if len(whereQueries) > 0 {
		where = "WHERE " + strings.Join(whereQueries, " AND ")
	}

	nodes := MoneroNodes{}

	queryTotalRows := fmt.Sprintf("SELECT COUNT(id) AS total_rows FROM tbl_node %s", where)

	err := repo.db.QueryRow(queryTotalRows, queryParams...).Scan(&nodes.TotalRows)
	if err != nil {
		return nodes, err
	}
	queryParams = append(queryParams, q.RowsPerPage, (q.Page-1)*q.RowsPerPage)

	allowedSort := []string{"last_checked", "uptime"}
	sortBy := "last_checked"
	if slices.Contains(allowedSort, q.SortBy) {
		sortBy = q.SortBy
	}
	sortDirection := "DESC"
	if q.SortDirection == "asc" {
		sortDirection = "ASC"
	}

	query := fmt.Sprintf("SELECT id, protocol, hostname, port, is_tor, is_available, nettype, height, adjusted_time, database_size, difficulty, version, uptime, estimate_fee, ip_addr, asn, asn_name, country, country_name, city, lat, lon, date_entered, last_checked, last_check_status, cors_capable FROM tbl_node %s ORDER BY %s %s LIMIT ? OFFSET ?", where, sortBy, sortDirection)

	row, err := repo.db.Query(query, queryParams...)
	if err != nil {
		return nodes, err
	}
	defer row.Close()

	nodes.RowsPerPage = q.RowsPerPage
	nodes.CurrentPage = q.Page
	nodes.NextPage = q.Page + 1

	for row.Next() {
		node := MoneroNode{}
		err = row.Scan(&node.Id, &node.Protocol, &node.Hostname, &node.Port, &node.IsTor, &node.IsAvailable, &node.NetType, &node.Height, &node.AdjustedTime, &node.DatabaseSize, &node.Difficulty, &node.Version, &node.Uptime, &node.EstimateFee, &node.Ip, &node.Asn, &node.AsnName, &node.CountryCode, &node.CountryName, &node.City, &node.Lat, &node.Lon, &node.DateEntered, &node.LastChecked, &node.LastCheckStatus, &node.CorsCapable)
		if err != nil {
			return nodes, err
		}
		nodes.Items = append(nodes.Items, &node)
	}

	return nodes, nil
}

type MoneroLogQueryParams struct {
	NodeId       int    // 0 fpr all, >0 for specific node
	WorkerId     int    // 0 for all, >0 for specific worker
	Status       int    // -1 for all, 0 for failed, 1 for success
	FailedReason string // empty for all, if not empty, will be used as search from failed_reaso

	RowsPerPage   int
	Page          int
	SortBy        string
	SortDirection string
}

type ProbeLog struct {
	Id           int     `db:"id" json:"id,omitempty"`
	NodeId       int     `db:"node_id" json:"node_id"`
	ProberId     int     `db:"prober_id" json:"prober_id"`
	Status       int     `db:"is_available" json:"status"`
	Height       int     `db:"height" json:"height"`
	AdjustedTime int     `db:"adjusted_time" json:"adjusted_time"`
	DatabaseSize int     `db:"database_size" json:"database_size"`
	Difficulty   int     `db:"difficulty" json:"difficulty"`
	EstimateFee  int     `db:"estimate_fee" json:"estimate_fee"`
	DateChecked  int     `db:"date_checked" json:"date_checked"`
	FailedReason string  `db:"failed_reason" json:"failed_reason"`
	FetchRuntime float64 `db:"fetch_runtime" json:"fetch_runtime"`
}

type MoneroNodeFetchLogs struct {
	TotalRows   int         `json:"total_rows"`
	RowsPerPage int         `json:"rows_per_page"`
	Items       []*ProbeLog `json:"items"`
}

func (repo *MoneroRepo) Logs(q MoneroLogQueryParams) (MoneroNodeFetchLogs, error) {
	queryParams := []interface{}{}
	whereQueries := []string{}
	where := ""

	if q.NodeId != 0 {
		whereQueries = append(whereQueries, "node_id = ?")
		queryParams = append(queryParams, q.NodeId)
	}
	if q.Status != -1 {
		whereQueries = append(whereQueries, "is_available = ?")
		queryParams = append(queryParams, q.Status)
	}
	if q.FailedReason != "" {
		whereQueries = append(whereQueries, "failed_reason LIKE ?")
		queryParams = append(queryParams, "%"+q.FailedReason+"%")
	}

	if len(whereQueries) > 0 {
		where = "WHERE " + strings.Join(whereQueries, " AND ")
	}

	fetchLogs := MoneroNodeFetchLogs{}

	queryTotalRows := fmt.Sprintf("SELECT COUNT(id) FROM tbl_probe_log %s", where)
	err := repo.db.QueryRow(queryTotalRows, queryParams...).Scan(&fetchLogs.TotalRows)
	if err != nil {
		return fetchLogs, err
	}
	queryParams = append(queryParams, q.RowsPerPage, (q.Page-1)*q.RowsPerPage)

	allowedSort := []string{"date_checked", "fetch_runtime"}
	sortBy := "id"
	if slices.Contains(allowedSort, q.SortBy) {
		sortBy = q.SortBy
	}
	sortDirection := "DESC"
	if q.SortDirection == "asc" {
		sortDirection = "ASC"
	}

	query := fmt.Sprintf("SELECT id, node_id, prober_id, is_available, height, adjusted_time, database_size, difficulty, estimate_fee, date_checked, failed_reason, fetch_runtime FROM tbl_probe_log %s ORDER BY %s %s LIMIT ? OFFSET ?", where, sortBy, sortDirection)

	row, err := repo.db.Query(query, queryParams...)
	if err != nil {
		return fetchLogs, err
	}
	defer row.Close()

	fetchLogs.RowsPerPage = q.RowsPerPage

	for row.Next() {
		probeLog := ProbeLog{}
		err = row.Scan(&probeLog.Id, &probeLog.NodeId, &probeLog.ProberId, &probeLog.Status, &probeLog.Height, &probeLog.AdjustedTime, &probeLog.DatabaseSize, &probeLog.Difficulty, &probeLog.EstimateFee, &probeLog.DateChecked, &probeLog.FailedReason, &probeLog.FetchRuntime)
		if err != nil {
			return fetchLogs, err
		}

		fetchLogs.Items = append(fetchLogs.Items, &probeLog)
	}

	return fetchLogs, nil
}

func (repo *MoneroRepo) Add(protocol string, hostname string, port uint) error {
	if protocol != "http" && protocol != "https" {
		return errors.New("Invalid protocol, must one of or HTTP/HTTPS")
	}

	if port > 65535 || port < 1 {
		return errors.New("Invalid port number")
	}

	is_tor := false
	if strings.HasSuffix(hostname, ".onion") {
		is_tor = true
	}
	ip := ""

	if !is_tor {
		hostIps, err := net.LookupIP(hostname)
		if err != nil {
			return err
		}

		hostIp := hostIps[0].To4()
		if hostIp == nil {
			return errors.New("Host IP is not IPv4")
		}
		if hostIp.IsPrivate() {
			return errors.New("IP address is private")
		}
		if hostIp.IsLoopback() {
			return errors.New("IP address is loopback address")
		}

		ip = hostIp.String()
	}

	check := `SELECT id FROM tbl_node WHERE protocol = ? AND hostname = ? AND port = ? LIMIT 1`
	row, err := repo.db.Query(check, protocol, hostname, port)
	if err != nil {
		return err
	}
	defer row.Close()

	if row.Next() {
		return errors.New("Node already monitored")
	}
	statusDb, _ := json.Marshal([5]int{2, 2, 2, 2, 2})

	query := `INSERT INTO tbl_node (protocol, hostname, port, is_tor, nettype, ip_addr, lat, lon, date_entered, last_checked, last_check_status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
	_, err = repo.db.Exec(query, protocol, hostname, port, is_tor, "", ip, 0, 0, time.Now().Unix(), 0, string(statusDb))
	if err != nil {
		return err
	}

	return nil
}

func (repo *MoneroRepo) Delete(id uint) error {
	if _, err := repo.db.Exec(`DELETE FROM tbl_node WHERE id = ?`, id); err != nil {
		return err
	}
	if _, err := repo.db.Exec(`DELETE FROM tbl_probe_log WHERE node_id = ?`, id); err != nil {
		return err
	}

	return nil
}

func (repo *MoneroRepo) GiveJob(acceptTor int) (MoneroNode, error) {
	queryParams := []interface{}{}
	whereQueries := []string{}
	where := ""

	if acceptTor != 1 {
		whereQueries = append(whereQueries, "is_tor = ?")
		queryParams = append(queryParams, 0)
	}

	if len(whereQueries) > 0 {
		where = "WHERE " + strings.Join(whereQueries, " AND ")
	}

	node := MoneroNode{}

	query := fmt.Sprintf(`SELECT id, hostname, port, protocol, is_tor, last_check_status FROM tbl_node %s ORDER BY last_checked ASC LIMIT 1`, where)
	err := repo.db.QueryRow(query, queryParams...).Scan(&node.Id, &node.Hostname, &node.Port, &node.Protocol, &node.IsTor, &node.LastCheckStatus)
	if err != nil {
		return node, err
	}

	update := `UPDATE tbl_node SET last_checked = ? WHERE id = ?`
	_, err = repo.db.Exec(update, time.Now().Unix(), node.Id)
	if err != nil {
		return node, err
	}

	return node, nil
}

type ProbeReport struct {
	TookTime float64    `json:"took_time"`
	Message  string     `json:"message"`
	NodeInfo MoneroNode `json:"node_info"`
}

func (repo *MoneroRepo) ProcessJob(report ProbeReport, proberId int64) error {
	if report.NodeInfo.Id == 0 {
		return errors.New("Invalid node")
	}

	qInsertLog := `INSERT INTO tbl_probe_log (node_id, prober_id, is_available, height, adjusted_time, database_size, difficulty, estimate_fee, date_checked, failed_reason, fetch_runtime) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`

	_, err := repo.db.Exec(qInsertLog, report.NodeInfo.Id, proberId, report.NodeInfo.IsAvailable, report.NodeInfo.Height, report.NodeInfo.AdjustedTime, report.NodeInfo.DatabaseSize, report.NodeInfo.Difficulty, report.NodeInfo.EstimateFee, time.Now().Unix(), report.Message, report.TookTime)
	if err != nil {
		return err
	}

	now := time.Now()
	limitTs := now.AddDate(0, -1, 0).Unix()

	nodeStats := struct {
		OnlineCount  uint `db:"online"`
		OfflineCount uint `db:"offline"`
		TotalFetched uint `db:"total_fetched"`
	}{}

	qstats := `SELECT
		SUM(if(is_available='1',1,0)) AS online,
		SUM(if(is_available='0',1,0)) AS offline,
		SUM(if(id='0',0,1)) AS total_fetched
	FROM tbl_probe_log WHERE node_id = ? AND date_checked > ?`
	repo.db.Get(&nodeStats, qstats, report.NodeInfo.Id, limitTs)

	avgUptime := (float64(nodeStats.OnlineCount) / float64(nodeStats.TotalFetched)) * 100
	report.NodeInfo.Uptime = math.Ceil(avgUptime*100) / 100

	var statuses [5]int
	errUnmarshal := report.NodeInfo.LastCheckStatus.Unmarshal(&statuses)
	if errUnmarshal != nil {
		fmt.Println("Warning", errUnmarshal.Error())
		statuses = [5]int{2, 2, 2, 2, 2}
	}

	nodeAvailable := 0

	if report.NodeInfo.IsAvailable {
		nodeAvailable = 1
	}
	newStatuses := statuses[1:]
	newStatuses = append(newStatuses, nodeAvailable)
	statuesValueToDb, errMarshalStatus := json.Marshal(newStatuses)
	if errMarshalStatus != nil {
		fmt.Println("WARN", errMarshalStatus.Error())
	}

	// recheck IP
	if report.NodeInfo.Ip != "" {
		if ipInfo, errGeoIp := GetGeoIpInfo(report.NodeInfo.Ip); errGeoIp != nil {
			fmt.Println("WARN:", errGeoIp.Error())
		} else {
			report.NodeInfo.Asn = ipInfo.Asn
			report.NodeInfo.AsnName = ipInfo.AsnOrg
			report.NodeInfo.CountryCode = ipInfo.CountryCode
			report.NodeInfo.CountryName = ipInfo.CountryName
			report.NodeInfo.City = ipInfo.City
			report.NodeInfo.Lon = ipInfo.Longitude
			report.NodeInfo.Lat = ipInfo.Latitude
		}
	}

	if report.NodeInfo.IsAvailable {
		update := `UPDATE tbl_node SET
			is_available = ?, nettype = ?, height = ?, adjusted_time = ?,
			database_size = ?, difficulty = ?, version = ?, uptime = ?,
			estimate_fee = ?, ip_addr = ?, asn = ?, asn_name = ?, country = ?,
			country_name = ?, city = ?, last_checked = ?, last_check_status = ?,
			cors_capable = ?
		WHERE id = ?`
		_, err = repo.db.Exec(update,
			nodeAvailable, report.NodeInfo.NetType, report.NodeInfo.Height, report.NodeInfo.AdjustedTime, report.NodeInfo.DatabaseSize, report.NodeInfo.Difficulty, report.NodeInfo.Version, report.NodeInfo.Uptime, report.NodeInfo.EstimateFee, report.NodeInfo.Ip, report.NodeInfo.Asn, report.NodeInfo.AsnName, report.NodeInfo.CountryCode, report.NodeInfo.CountryName, report.NodeInfo.City, now.Unix(), string(statuesValueToDb), report.NodeInfo.CorsCapable, report.NodeInfo.Id)
	} else {
		update := `UPDATE tbl_node SET is_available = ?,  uptime = ?, last_checked = ?, last_check_status = ? WHERE id = ?`
		_, err = repo.db.Exec(update, nodeAvailable, report.NodeInfo.Uptime, now.Unix(), string(statuesValueToDb), report.NodeInfo.Id)
	}

	if avgUptime <= 0 && nodeStats.TotalFetched > 300 {
		fmt.Println("Deleting Monero node (0% uptime from > 300 records)")
		repo.Delete(report.NodeInfo.Id)
	}

	repo.db.Exec(`UPDATE tbl_prober SET last_submit_ts = ? WHERE id = ?`, now.Unix(), proberId)

	return err
}

type NetFee struct {
	Nettype     string `json:"nettype" db:"nettype"`
	EstimateFee uint   `json:"estimate_fee" db:"estimate_fee"`
	NodeCount   int    `json:"node_count" db:"node_count"`
}

func (repo *MoneroRepo) NetFee() []NetFee {
	netTypes := [3]string{"mainnet", "stagenet", "testnet"}
	netFees := []NetFee{}

	for _, net := range netTypes {
		fees := NetFee{}
		err := repo.db.Get(&fees, `SELECT COUNT(id) AS node_count, nettype, estimate_fee FROM tbl_node WHERE nettype = ? GROUP BY estimate_fee ORDER BY node_count DESC LIMIT 1`, net)
		if err != nil {
			fmt.Println("WARN:", err.Error())
			continue
		}
		netFees = append(netFees, fees)
	}

	return netFees
}

type MoneroCountries struct {
	TotalNodes int    `json:"total_nodes" db:"total_nodes"`
	Cc         string `json:"cc" db:"country"`
	Name       string `json:"name" db:"country_name"`
}

func (repo *MoneroRepo) Countries() ([]MoneroCountries, error) {
	countries := []MoneroCountries{}

	err := repo.db.Select(&countries, `SELECT COUNT(id) AS total_nodes, country, country_name FROM tbl_node GROUP BY country ORDER BY country ASC`)
	return countries, err
}