#pragma once #include "chat/chat_protocol.h" #include "data/wallet_state.h" #include "refresh_scheduler.h" #include "rpc/rpc_worker.h" #include "rpc/rpc_failure.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; // A call allowed to outlast the default timeout (a whole-wallet sweep on a large wallet). // Gateways without a separate timeout just call(). virtual nlohmann::json callLong(const std::string& method, const nlohmann::json& params, long timeoutSec) { (void)timeoutSec; return call(method, params); } }; 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; std::optional txCount; // getwalletinfo's txcount (the wallet's transactions) }; struct WarmupPollResult { bool ready = false; ConnectionInfoResult info; std::string errorMessage; // Why the poll failed: a -28 warmup reply means the node is there; a refused/reset connection // means it's gone (the caller counts those toward its lost-connection check). rpc::RpcFailureKind errorKind = rpc::RpcFailureKind::None; }; struct ConnectionInitResult { ConnectionInfoResult info; WalletEncryptionResult encryption; }; struct CoreRefreshResult { bool balanceOk = false; std::optional shieldedBalance; // display (minconf=0, incl. pending change) std::optional transparentBalance; std::optional totalBalance; std::optional spendableShieldedBalance; // confirmed (minconf=1) std::optional spendableTransparentBalance; std::optional spendableTotalBalance; bool blockchainOk = false; std::optional blocks; std::optional headers; std::optional bestBlockHash; std::optional verificationProgress; std::optional longestChain; std::optional notarized; double balanceScanMs = 0.0; // wall-clock spent in z_gettotalbalance this refresh (0 if balance skipped) // How those calls failed (None when they succeeded or weren't made), so a caller can tell a busy // node (ReplyTimeout / QueueFull) from a gone one (NotListening) without reading what(). rpc::RpcFailureKind blockchainErrKind = rpc::RpcFailureKind::None; rpc::RpcFailureKind balanceErrKind = rpc::RpcFailureKind::None; // The minconf=0 call reached the daemon (it may still have failed). False when the balance was // skipped (behind, busy chain read) or held back by the busy gate (Deferred) before it was sent. bool balanceAttempted = false; bool spendableQueried = false; // the minconf=1 call was made std::optional totalBalanceZat; // the minconf=0 "total" in zatoshis, exact // getmempoolinfo (CoreRefreshPlan::fetchMempool): a change signal that takes no cs_main. bool mempoolFetched = false; bool mempoolOk = false; std::int64_t mempoolTxs = 0; std::int64_t mempoolBytes = 0; }; // What one Core refresh fetches. The minconf=1 (spendable) z_gettotalbalance is a second whole-wallet // scan under cs_main; it only tells something the minconf=0 call doesn't while funds are unconfirmed // or pending, so it runs only when needSpendable or the exact total moved since prevTotalZat. struct CoreRefreshPlan { bool includeBalance = true; bool needSpendable = true; std::optional prevTotalZat; bool fetchMempool = false; // add getmempoolinfo after a successful getblockchaininfo bool evenIfBehind = false; // the session's first balance read: also while the node is behind }; 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; }; // Minimal per-note view of a z_listunspent entry — just what a downstream consumer needs to derive // spendable-note budgets without re-scanning. Kept UI/feature-agnostic (no chat specifics here). struct UnspentNoteLite { double amount = 0.0; // note value, DRGX bool locked = false; // tied up by an in-flight send int confirmations = 0; // TRUE depth (rawconfirmations when present, else confirmations) std::string address; // the z-address holding it std::string txid; // the transaction that created it bool change = false; // change from the wallet's own spend (not a receive) }; struct AddressRefreshResult { std::vector shieldedAddresses; std::vector transparentAddresses; // False if either address-enumeration RPC (z_listaddresses / getaddressesbyaccount) threw, so the // lists may be falsely short. Consumers that treat an empty list as authoritative (e.g. the // empty-wallet warning) must not trust a 0 count unless this is true. bool addressListOk = true; // Wall-clock spent in the address scan (dominated by z_listunspent — O(mapWallet), holds the // daemon's cs_main). Lets the caller throttle how often it polls (addressRefreshDue()). double scanMs = 0.0; // The wallet's unspent notes from this same z_listunspent scan, so a consumer (e.g. the chat // note-budget) can be fed for free instead of running its own duplicate z_listunspent. std::vector unspentNotes; bool balancesOk = true; // false: a balance call failed, the per-address balances may be stale bool replyTimedOut = false; // a call hit the reply timeout (the daemon kept running it) // The lists only: no z_listunspent / listunspent, or a balance call failed (the pass then degrades to // list-only). Applying it carries the previous balances over. bool listOnly = false; bool nodeBusy = false; // a call failed because the node was busy (timeout, full queue, held back) }; 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::size_t position = 0; }; 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; // receive scans this cycle; 0 = no limit // The pass's floor (services::ShieldedHistoryPass): an address scanned below it is scanned again. // -1: only never-scanned addresses (and priority ones) are. int shieldedScanFloorHeight = -1; // Scanned first, whatever the cursor, when never scanned or scanned below their floor: new // addresses, the chat reply address, addresses holding notes the history doesn't show yet. std::unordered_map priorityScanFloors; bool skipShieldedScans = false; // a Light cycle (or a probe-less recent poll): no receive scans bool skipSentSweep = false; int sentScanFromHeight = -1; // the sweep's watermark; -1 = sweep the whole wallet std::size_t transparentRowHint = 0; // listtransactions rows last time: sizes the first page // Same tip and nothing changed since the rows on screen were fetched: reuse their transparent rows // instead of paging listtransactions again. bool reuseTransparentRows = false; std::unordered_set enrichBacklogTxids; // discovered earlier, not enriched yet // Run a windowed sent sweep even though the receive pass isn't complete (the caller rate-limits // this), so untracked sends (from the Console, or another wallet on the same seed) don't wait for a // whole pass. The whole-wallet sweep stays tied to a completed pass. bool allowMidPassSentSweep = false; }; struct TransactionRefreshResult { std::vector transactions; std::vector hushChatMetadata; int blockHeight = -1; TransactionViewCache newViewTxEntries; // Tracked sends whose view result is final although it shows no outgoing output: the op only // paid the wallet itself (shield, merge/consolidate, auto-shield, chat note split, a transfer // between own addresses) and is now mined or conflicted. Applying the result stops tracking them. std::unordered_set resolvedSendTxids; std::size_t nextShieldedScanStartIndex = 0; std::size_t shieldedAddressesScanned = 0; std::size_t shieldedAddressCount = 0; std::unordered_map shieldedScanHeights; bool shieldedScanComplete = true; // Wall-clock spent in the whole cycle: listtransactions, the receive scans and the sent sweep each // hold cs_main. Lets the caller throttle wallet-wide scans by their measured cost (txRefreshDue()). double scanMs = 0.0; std::size_t shieldedScansSucceeded = 0; // shieldedAddressesScanned counts attempts double receiveScanOkMs = 0.0; // time in the receive scans that succeeded std::size_t transparentRowCount = 0; // rows the listtransactions chain returned bool transparentChainOk = false; bool transparentRowsReused = false; bool sentSweepRan = false; bool sentSweepOk = false; bool sentSweepFull = false; int sentSweepHeight = -1; // the tip a successful sweep covered double sentSweepMs = 0.0; // time in the sweep (part of scanMs) rpc::RpcFailureKind sentSweepErrKind = rpc::RpcFailureKind::None; // Txids found but not enriched (over the per-cycle cap, or the view failed transiently): the caller // keeps them and feeds them back as enrichBacklogTxids. std::unordered_set unenrichedTxids; bool rpcError = false; // a wallet-wide call failed bool replyTimedOut = false; // ...with a reply timeout (the daemon kept running it) }; struct OperationStatusPollResult { std::vector doneOpids; std::vector staleOpids; std::vector successTxids; std::unordered_map successTxidsByOpid; std::vector failureMessages; std::unordered_map failureByOpid; // opid -> error message 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); // The node knows the chain tip: longestchain stays 0 after a start until its first UpdateTip. static bool connectionInfoTipKnown(const ConnectionInfoResult& info); // The tip is known and the node is more than 2 blocks behind it (the sync tolerance). static bool connectionInfoShowsBehind(const ConnectionInfoResult& 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, const nlohmann::json& spendableBalance, bool balanceOk, const nlohmann::json& blockInfo, bool blockOk); // includeBalance=false skips z_gettotalbalance (which takes the wallet lock + cs_main) and only // fetches getblockchaininfo — used while syncing, where the balance is incomplete anyway and the // wallet should minimise lock contention with block connection. static CoreRefreshResult collectCoreRefreshResult(RefreshRpcGateway& rpc, const CoreRefreshPlan& plan); static CoreRefreshResult collectCoreRefreshResult(RefreshRpcGateway& rpc, bool includeBalance = true) { CoreRefreshPlan plan; plan.includeBalance = includeBalance; return collectCoreRefreshResult(rpc, plan); } // An amount as the daemon formats it ("[-]digits[.up to 8 digits]") in zatoshis; nullopt for anything // else or anything that would overflow. Exact, so "did the total move" never trips on rounding. static std::optional parseMoneyZat(const std::string& text); // Whether the next balance poll needs the minconf=1 (spendable) call whatever the total does: never // fetched, unconfirmed or pending funds, or a forced refresh after the user's own send. static bool spendableBalanceNeeded(const WalletState& state, bool pendingSendWork, bool forced); // The total the next poll compares against (CoreRefreshPlan::prevTotalZat). None after a poll whose // minconf=1 call was made but brought back no total (timeout, held back, error): that left spendable // degraded to display, so the next poll must count the total as moved and measure it again. static std::optional nextTotalBaseline(const CoreRefreshResult& result, std::optional previous); 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); // Parse a CoinGecko /market_chart response ({"prices":[[ms,price],...]}) into a timestamped // (unix seconds, USD price) series, oldest->newest. Malformed rows are skipped; on any error // (or a missing "prices" array) an empty vector is returned. Pure/no-I/O. static std::vector> parseCoinGeckoMarketChart(const std::string& response); 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); // listOnly: enumerate the addresses only (cheap), without the wallet-wide balance scans; for while the // scans are held (the node catching up, the connect gate). Both lists are enumerated first, and the // balances are scanned only when both enumerations succeeded (a failed pass is discarded whole). static AddressRefreshResult collectAddressRefreshResult( RefreshRpcGateway& rpc, const AddressRefreshSnapshot& snapshot = {}, bool listOnly = false); 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 = {}, std::unordered_map>* chatMemoOutputs = nullptr); static TransactionViewCacheEntry parseViewTransactionCacheEntry(const nlohmann::json& viewTransaction); static void appendViewTransactionOutputs(std::vector& transactions, const std::string& txid, const TransactionViewCacheEntry& entry); // Build "send" rows from a z_listsentbyaddress entry's sends.transparentSends[] — the transparent // (z->t deshield) outgoing outputs that z_viewtransaction does not surface. Deduped by // txid+address+amount against rows already present (e.g. listtransactions-derived t->t sends). static void appendSentTransparentOutputs(std::vector& transactions, const std::string& txid, const nlohmann::json& sentEntry); 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); // Params for z_getoperationstatus/result filtered to these opids: ONE argument, an array of ids, so // [[id, ...]]. nullopt when no non-empty id remains: an empty filter means every op the node has // retained. Built explicitly, because a one-element braced list of a json ({ids}) collapses to the // inner array, which the daemon rejects ("JSON value is not an array as expected"). static std::optional buildOperationFilterParams(const std::vector& opids); struct DaemonRescanScan { bool foundRescan = false; bool finished = false; float rescanPct = 0.0f; std::string lastStatus; bool foundWitness = false; float witnessPct = -1.0f; int witnessRemaining = -1; int witnessReadTotal = -1; bool witnessRebuilt = false; std::vector witnessTxids; int witnessTotalTxs = -1; }; // Pure line-by-line scan of newly-appended dragonxd stdout for rescan + Sapling // witness-rebuild progress. No I/O, no state — unit-tested against fixture log snippets. static DaemonRescanScan parseDaemonRescanOutput(const std::string& output); 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); } void deferFor(Timer timer, float seconds) { scheduler_.deferFor(timer, seconds); } void retrySoon(Timer timer, float seconds) { scheduler_.retrySoon(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