#pragma once #include "data/wallet_state.h" #include "refresh_scheduler.h" #include "rpc/rpc_worker.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include namespace dragonx { namespace services { class NetworkRefreshService { public: using Timer = RefreshScheduler::Timer; using Intervals = RefreshScheduler::Intervals; class RefreshRpcGateway { public: virtual ~RefreshRpcGateway() = default; virtual nlohmann::json call(const std::string& method, const nlohmann::json& params) = 0; }; enum class Job { Core, Addresses, Transactions, Mining, Peers, Price, Encryption, ConnectionInit, Count }; struct DispatchTicket { Job job = Job::Core; std::uint64_t generation = 0; bool accepted = false; }; struct JobStats { std::uint64_t started = 0; std::uint64_t finished = 0; std::uint64_t skippedInFlight = 0; std::uint64_t skippedQueuePressure = 0; std::uint64_t staleCallbacks = 0; std::size_t lastQueueDepth = 0; }; struct EnqueueResult { DispatchTicket ticket; bool enqueued = false; std::size_t queueDepth = 0; }; struct ConnectionInfoResult { bool ok = false; std::optional daemonVersion; std::optional protocolVersion; std::optional p2pPort; std::optional longestChain; std::optional notarized; std::optional blocks; }; struct WalletEncryptionResult { bool ok = false; bool encrypted = false; std::int64_t unlockedUntil = 0; }; struct WarmupPollResult { bool ready = false; ConnectionInfoResult info; std::string errorMessage; }; struct ConnectionInitResult { ConnectionInfoResult info; WalletEncryptionResult encryption; }; struct CoreRefreshResult { bool balanceOk = false; std::optional shieldedBalance; std::optional transparentBalance; std::optional totalBalance; bool blockchainOk = false; std::optional blocks; std::optional headers; std::optional bestBlockHash; std::optional verificationProgress; std::optional longestChain; std::optional notarized; }; struct MiningRefreshResult { std::optional localHashrate; bool miningOk = false; std::optional generate; std::optional genproclimit; std::optional blocks; std::optional difficulty; std::optional networkHashrate; std::optional chain; double daemonMemoryMb = 0.0; }; struct PeerRefreshResult { std::vector peers; std::vector bannedPeers; }; struct PriceRefreshResult { MarketInfo market; }; struct PriceHttpResponse { bool transportOk = false; long httpStatus = 0; std::string body; std::string transportError; }; struct PriceHttpResult { std::optional price; std::string errorMessage; }; struct AddressRefreshResult { std::vector shieldedAddresses; std::vector transparentAddresses; }; struct AddressRefreshSnapshot { std::unordered_map shieldedSpendingKeys; }; struct TransactionViewCacheEntry { std::string from_address; std::int64_t timestamp = 0; int confirmations = 0; struct Output { std::string address; double value = 0.0; std::string memo; }; std::vector outgoing_outputs; }; using TransactionViewCache = std::unordered_map; struct TransactionRefreshSnapshot { std::vector shieldedAddresses; std::unordered_set fullyEnrichedTxids; TransactionViewCache viewTxCache; std::unordered_set sendTxids; std::unordered_set pendingOpids; std::vector previousTransactions; std::set miningAddresses; std::unordered_map shieldedScanHeights; std::size_t shieldedScanStartIndex = 0; std::size_t maxShieldedReceiveScans = 0; }; struct TransactionRefreshResult { std::vector transactions; int blockHeight = -1; TransactionViewCache newViewTxEntries; std::size_t nextShieldedScanStartIndex = 0; std::size_t shieldedAddressesScanned = 0; std::size_t shieldedAddressCount = 0; std::unordered_map shieldedScanHeights; bool shieldedScanComplete = true; }; struct OperationStatusPollResult { std::vector doneOpids; std::vector staleOpids; std::vector successTxids; std::unordered_map successTxidsByOpid; std::vector failureMessages; bool anySuccess = false; }; struct TransactionCacheUpdate { TransactionViewCache& viewTxCache; std::unordered_set& sendTxids; std::vector& confirmedTxCache; std::unordered_set& confirmedTxIds; int& confirmedCacheBlock; int& lastTxBlockHeight; }; static Intervals intervalsForPage(ui::NavPage page) { return RefreshScheduler::intervalsForPage(page); } static ConnectionInfoResult parseConnectionInfoResult(const nlohmann::json& info); static WalletEncryptionResult parseWalletEncryptionResult(const nlohmann::json& walletInfo); static WarmupPollResult collectWarmupPollResult(RefreshRpcGateway& rpc); static ConnectionInitResult collectConnectionInitResult( RefreshRpcGateway& rpc, const std::optional& prefetchedInfo = std::nullopt); static CoreRefreshResult parseCoreRefreshResult(const nlohmann::json& totalBalance, bool balanceOk, const nlohmann::json& blockInfo, bool blockOk); static CoreRefreshResult collectCoreRefreshResult(RefreshRpcGateway& rpc); static MiningRefreshResult parseMiningRefreshResult(const nlohmann::json& miningInfo, bool miningOk, const nlohmann::json& localHashrate, bool hashrateOk, double daemonMemoryMb); static MiningRefreshResult collectMiningRefreshResult(RefreshRpcGateway& rpc, double daemonMemoryMb, bool includeSlowRefresh, bool includeLocalHashrate = true); static PeerRefreshResult parsePeerRefreshResult(const nlohmann::json& peers, const nlohmann::json& bannedPeers); static PeerRefreshResult collectPeerRefreshResult(RefreshRpcGateway& rpc); static std::optional parseCoinGeckoPriceResponse(const std::string& response, std::time_t fetchedAt); static PriceHttpResult parsePriceHttpResponse(const PriceHttpResponse& response, std::time_t fetchedAt); static AddressInfo buildShieldedAddressInfo(const std::string& address, const nlohmann::json& validation, bool validationSucceeded); static AddressInfo buildTransparentAddressInfo(const std::string& address); static std::vector parseTransparentAddressList(const nlohmann::json& addressList); static void applyShieldedBalancesFromUnspent(std::vector& addresses, const nlohmann::json& unspent); static void applyTransparentBalancesFromUnspent(std::vector& addresses, const nlohmann::json& unspent); static AddressRefreshSnapshot buildAddressRefreshSnapshot(const WalletState& state); static AddressRefreshResult collectAddressRefreshResult( RefreshRpcGateway& rpc, const AddressRefreshSnapshot& snapshot = {}); static TransactionRefreshSnapshot buildTransactionRefreshSnapshot(const WalletState& state, const TransactionViewCache& viewTxCache, const std::unordered_set& sendTxids); static void appendTransparentTransactions(std::vector& transactions, std::set& knownTxids, const nlohmann::json& result, const std::set& miningAddresses = {}); static void appendShieldedReceivedTransactions(std::vector& transactions, std::set& knownTxids, const std::string& address, const nlohmann::json& received, const std::set& miningAddresses = {}); static TransactionViewCacheEntry parseViewTransactionCacheEntry(const nlohmann::json& viewTransaction); static void appendViewTransactionOutputs(std::vector& transactions, const std::string& txid, const TransactionViewCacheEntry& entry); static void sortTransactionsNewestFirst(std::vector& transactions); static TransactionRefreshResult collectTransactionRefreshResult(RefreshRpcGateway& rpc, const TransactionRefreshSnapshot& snapshot, int currentBlockHeight, int maxViewTransactionsPerCycle); static TransactionRefreshResult collectRecentTransactionRefreshResult( RefreshRpcGateway& rpc, const TransactionRefreshSnapshot& snapshot, int currentBlockHeight, int pageSize = 100); static OperationStatusPollResult parseOperationStatusPoll(const nlohmann::json& result, const std::vector& requestedOpids); static void applyConnectionInfoResult(WalletState& state, const ConnectionInfoResult& result); static void applyWalletEncryptionResult(WalletState& state, const WalletEncryptionResult& result); static void applyConnectionInitResult(WalletState& state, const ConnectionInitResult& result); static void applyCoreRefreshResult(WalletState& state, const CoreRefreshResult& result, std::time_t updatedAt); static void applyMiningRefreshResult(WalletState& state, const MiningRefreshResult& result, std::time_t updatedAt); static void applyPeerRefreshResult(WalletState& state, PeerRefreshResult&& result, std::time_t updatedAt); static void markPriceRefreshStarted(WalletState& state); static void applyPriceRefreshResult(WalletState& state, const PriceRefreshResult& result, std::chrono::steady_clock::time_point fetchedAt); static void applyPriceRefreshFailure(WalletState& state, const std::string& errorMessage); static void applyAddressRefreshResult(WalletState& state, AddressRefreshResult&& result); static void applyTransactionRefreshResult(WalletState& state, TransactionCacheUpdate cacheUpdate, TransactionRefreshResult&& result, std::time_t updatedAt); void applyPage(ui::NavPage page) { scheduler_.applyPage(page); } void setIntervals(Intervals intervals) { scheduler_.setIntervals(intervals); } const Intervals& intervals() const { return scheduler_.intervals(); } void tick(float deltaSeconds) { scheduler_.tick(deltaSeconds); } bool isDue(Timer timer) const { return scheduler_.isDue(timer); } bool consumeDue(Timer timer) { return scheduler_.consumeDue(timer); } void reset(Timer timer) { scheduler_.reset(timer); } void markDue(Timer timer) { scheduler_.markDue(timer); } void setTimer(Timer timer, float seconds) { scheduler_.setTimer(timer, seconds); } float timer(Timer timer) const { return scheduler_.timer(timer); } float interval(Timer timer) const { return scheduler_.interval(timer); } void markImmediateRefresh() { scheduler_.markImmediateRefresh(); } void markWalletMutationRefresh() { scheduler_.markWalletMutationRefresh(); } void resetTxAge() { scheduler_.resetTxAge(); } bool shouldRefreshTransactions(int lastTxBlockHeight, int currentBlockHeight, bool transactionsDirty) const; bool beginJob(Job job); bool beginJob(Job job, std::size_t queuedWork, std::size_t maxQueuedWork); void finishJob(Job job); bool jobInProgress(Job job) const; void resetJobs(); DispatchTicket beginDispatch(Job job, std::size_t queuedWork = 0, std::size_t maxQueuedWork = 0); bool completeDispatch(const DispatchTicket& ticket); void cancelDispatch(const DispatchTicket& ticket); JobStats stats(Job job) const; template EnqueueResult enqueue(Job job, Worker& worker, WorkFn&& work, std::size_t maxQueuedWork = 0) { std::size_t queueDepth = worker.pendingTaskCount(); auto ticket = beginDispatch(job, queueDepth, maxQueuedWork); if (!ticket.accepted) return {ticket, false, queueDepth}; worker.post([this, ticket, work = std::forward(work)]() mutable -> rpc::RPCWorker::MainCb { rpc::RPCWorker::MainCb mainCallback; try { mainCallback = work(); } catch (...) { mainCallback = nullptr; } return [this, ticket, mainCallback = std::move(mainCallback)]() mutable { if (!completeDispatch(ticket)) return; if (mainCallback) mainCallback(); }; }); return {ticket, true, queueDepth}; } private: std::atomic& jobFlag(Job job); const std::atomic& jobFlag(Job job) const; static std::size_t jobIndex(Job job); RefreshScheduler scheduler_; std::atomic coreInProgress_{false}; std::atomic addressesInProgress_{false}; std::atomic transactionsInProgress_{false}; std::atomic miningInProgress_{false}; std::atomic peersInProgress_{false}; std::atomic priceInProgress_{false}; std::atomic encryptionInProgress_{false}; std::atomic connectionInitInProgress_{false}; std::array, static_cast(Job::Count)> generations_{}; std::array, static_cast(Job::Count)> started_{}; std::array, static_cast(Job::Count)> finished_{}; std::array, static_cast(Job::Count)> skippedInFlight_{}; std::array, static_cast(Job::Count)> skippedQueuePressure_{}; std::array, static_cast(Job::Count)> staleCallbacks_{}; std::array, static_cast(Job::Count)> lastQueueDepth_{}; }; } // namespace services } // namespace dragonx