ObsidianDragon/src/services/note_consolidation.h

// DragonX Wallet - ImGui Edition
// Copyright 2024-2026 The Hush Developers
// Released under the GPLv3

#pragma once

// Consolidating notes WITHOUT moving funds between addresses: each address's notes are merged back into
// that same address (z_mergetoaddress [addr] addr), so funds kept apart under different labels stay apart
// while the wallet sheds notes (fewer notes = smaller wallet, faster witness rebuilds and scans).
// Shielded only: a transparent address can't receive a non-coinbase output on DragonX (private chain).
// Pure apart from the merge call it is handed, so the test target can drive it.

#include "rpc/rpc_failure.h"

#include <nlohmann/json.hpp>

#include <chrono>
#include <cstdint>
#include <functional>
#include <set>
#include <string>
#include <vector>

namespace dragonx {
namespace services {

// One shielded address's spendable notes (confirmed, spendable), from z_listunspent.
struct NoteGroup {
    std::string address;
    int notes = 0;
    double amount = 0.0;
};

// Spendable shielded notes grouped by address, most notes first. Transparent UTXOs and unconfirmed,
// watch-only or locked notes are left out (z_mergetoaddress only spends confirmed, unlocked notes it has
// keys for).
std::vector<NoteGroup> groupSpendableNotes(const nlohmann::json& zListUnspent);

// A z_listunspent entry the wallet has locked, e.g. for an operation in flight ("locked": true).
bool noteIsLocked(const nlohmann::json& zListUnspentEntry);

// Worth consolidating in place: at least two notes, and more than one batch's fee to work with.
bool worthConsolidating(const NoteGroup& group, double fee);

// Merge calls needed to bring `notes` down when each call takes at most `limit` of them. A lone leftover
// note isn't a call (the daemon refuses to merge one note into its own address).
int estimateBatches(int notes, int limit);

// What one z_mergetoaddress call reported.
struct MergeCallResult {
    std::string opid;
    int merging = 0;     // notes this call is merging
    int remaining = 0;   // notes of the address it left for a later call
    std::string error;   // non-empty: the call failed (nothing was submitted)
};
using MergeCall = std::function<MergeCallResult(const std::string& address)>;

// The outcome for one address.
struct AddressRun {
    std::string address;
    std::vector<std::string> opids;   // one per submitted batch
    int notesMerged = 0;
    int remaining = 0;                // notes still left after the last batch
    std::string error;                // the call that stopped it, if any (earlier batches still went out)
};

// Submit merges of `address` into itself until fewer than two of its notes are left unmerged. The daemon
// locks a batch's notes before the call returns, so each call takes the next batch. At most maxBatches.
AddressRun consolidateInPlace(const std::string& address, const MergeCall& call, int maxBatches = 20);

// ── Merge-call stepper ──
// One call's result into the address's run (r.error set: run.error = r.error, false). True while another call
// for this address is worth making: at least 2 unlocked notes left and fewer than maxBatches submitted.
bool recordMergeResult(AddressRun& run, const MergeCallResult& r, int maxBatches = 20);
// z_mergetoaddress's reply. Never throws. A non-empty string "opid" = accepted; mergingNotes/remainingNotes only
// when integral (clamped to [0, INT_MAX]), else 0; no opid: error "no operation id".
void readMergeCallResult(const nlohmann::json& res, MergeCallResult& out);

// ── Async op status ──
enum class OpState : std::uint8_t { Unknown, Queued, Executing, Success, Failed, Lost };
inline bool opStateTerminal(OpState s) { return s == OpState::Success || s == OpState::Failed || s == OpState::Lost; }
struct OpReport {
    std::string id, status, error, txid;   // error = error.message, txid = result.txid
};
OpState opStateFromStatus(const std::string& status);   // queued/executing/success; failed|cancelled|canceled: Failed
// Type-checked, never throws. Skips non-objects and missing/non-string/empty ids; keeps only `wanted` (all if empty).
std::vector<OpReport> parseOpReports(const nlohmann::json& statusArray, const std::set<std::string>& wanted = {});

// ── Merge-call failures ──
enum class MergeFailure : std::uint8_t {
    NotSent,      // turned away before z_mergetoaddress ran: the same batch again, later
    Syncing,      // THROW_IF_SYNCING: hold the run while the node catches up
    NeedsUnlock,  // a locked keystore ("could not find spending key")
    NothingLeft,  // no funds / already merged / too small for the fee
    Uncertain,    // it may have queued the merge (a reply timeout, an answer without an opid): never resent
    Transient,    // the HTTP layer dropped it, or the async queue is shutting down: retry with backoff
    Fatal,        // anything else: this address is done
};
MergeFailure classifyMergeFailure(rpc::RpcFailureKind kind, const std::string& message);
// The failure kind of a z_mergetoaddress call that threw. Sent, then cut off before any HTTP status
// (connection reset, an empty or partial reply): the daemon may have queued the merge, exactly as after a
// reply timeout, so it reads as one. Any other failure keeps its kind.
rpc::RpcFailureKind mergeCallFailureKind(const std::exception& e);

// ── Paced per-address consolidation run (owned by App, pumped from App::update(); main thread) ──
// The "Keep in each address" mode used to post one task that called z_mergetoaddress for every batch of every
// address back to back. Each call is a whole-wallet scan under cs_main, and the daemon runs the ops one at a
// time anyway. Here: one run RPC at a time, the next call only after the previous batch left the queue and a
// gap as long as the last call took, failures classified (nothing that may have queued is ever resent), and a
// breaker after three dead addresses or three failed ops in a row.
inline constexpr double kRunMinGapSec = 2.0;
inline constexpr double kRunMaxGapSec = 120.0;
inline constexpr double kRunPollIntervalSec = 3.0;
inline constexpr long   kRunCallTimeoutSec = 120;
inline constexpr double kRunWatchdogSec = 1800.0;   // a call or poll outstanding this long is presumed lost
inline constexpr double kRunOpMaxSec = 3600.0;      // an op not terminal after this much pollable time is presumed lost
inline constexpr double kRunOpHardMaxSec = 4 * kRunOpMaxSec;   // ...and after this long whatever happened
inline constexpr double kRunNotSentRetrySec = 5.0;
inline constexpr double kRunSyncHoldSec = 30.0;
inline constexpr double kRunUnlockHoldSec = 30.0;
inline constexpr int    kRunMaxTransientRetries = 5;
inline constexpr int    kRunMaxUnlockRefusals = 2;
inline constexpr int    kRunMaxFailStreak = 3;
inline constexpr int    kRunMaxPollMisses = 2;

struct RunConfig {
    double fee = 0.0001;
    int limit = 50;
    int maxBatches = 20;
};
struct RunEnv {
    bool nodeUsable = false;    // connected, not warming up or rescanning
    bool busy = false;          // the busy gate holds RPCs that need cs_main
    bool callsAllowed = true;   // not syncing / settling
    bool locked = false;
    bool userSendBusy = false;  // a send / shield / auto-shield / chat op of ours is in progress
    bool scanBusy = false;      // another GUI wallet scan is in the daemon
};
enum class RunAction : std::uint8_t { None, Call, Poll, Finish };
enum class RunWait : std::uint8_t { None, InFlight, Node, Busy, Syncing, Locked, UserSend, Scan, Batch };
enum class RunEnd : std::uint8_t { None, Completed, Stopped, Halted, Aborted };
enum class AddrOutcome : std::uint8_t { Pending, Running, Done, Uncertain, Failed };   // ordered: worse wins
enum class RunVerdict : std::uint8_t { None, Clean, NothingSubmitted, SomeFailed, Unconfirmed, Stopped, Halted, Aborted };
struct RunStep {
    RunAction action = RunAction::None;
    int addrIdx = -1;
};
struct RunAddress {
    NoteGroup group;
    AddressRun run;
    AddrOutcome outcome = AddrOutcome::Pending;
    bool done = false;
    int opsDone = 0;
    int transient = 0;
    int unlockRefusals = 0;
    std::string failure;
};
struct RunBatch {
    int addrIdx = 0;
    std::string opid;
    int merging = 0;
    OpState state = OpState::Unknown;
    std::string error, txid;
    int misses = 0;
    double acceptedAt = 0.0;
    double pollableSec = 0.0;   // time since accepted in which a status poll could have gone out
};
struct RunProgress {
    int addresses = 0, addressesDone = 0, batchesSubmitted = 0, batchesEstimated = 0, batchesDone = 0, notesMerged = 0;
    bool stopping = false;
    RunWait wait = RunWait::None;
};
struct RunSummary {
    RunEnd end = RunEnd::None;
    int addresses = 0, addressesFailed = 0, addressesUnconfirmed = 0, batchesSubmitted = 0, batchesSucceeded = 0,
        notesMerged = 0;
    std::string firstError;
};
RunVerdict runVerdict(const RunSummary& s);
const char* runWaitKey(RunWait wait);   // i18n key of the wait line, nullptr when there is nothing to say

class ConsolidationRun {
public:
    bool start(const std::vector<NoteGroup>& groups, RunConfig cfg, double now);
    void requestStop();
    bool abort();
    RunStep next(const RunEnv& env, double now);
    void markCallPosted(int addrIdx, double now);
    void markPollPosted(double now);
    // Returns an orphan opid (an accepted merge whose result is stale) for the caller to track, else "".
    std::string onCallResult(std::uint64_t gen, int addrIdx, const MergeCallResult& r, rpc::RpcFailureKind kind,
                             double callSec, double now);
    void onPollResult(std::uint64_t gen, bool ok, const std::vector<OpReport>& reports, double now);
    bool consumeLockResyncRequest();
    std::vector<std::string> outstandingOpids() const;
    void dismiss();
    bool active() const { return phase_ == Phase::Running; }
    bool hasRun() const { return phase_ != Phase::Idle; }
    bool stopping() const { return stopping_; }
    std::uint64_t gen() const { return gen_; }
    RunWait lastWait() const { return lastWait_; }
    RunEnd end() const { return end_; }
    RunProgress progress() const;
    RunSummary summary() const;
    const std::vector<RunAddress>& addresses() const { return addrs_; }
    const std::vector<RunBatch>& batches() const { return batches_; }
    const RunConfig& config() const { return cfg_; }

private:
    enum class Phase : std::uint8_t { Idle, Running, Finished };
    static void worsen(RunAddress& a, AddrOutcome o) { if (o > a.outcome) a.outcome = o; }
    void endAddress(RunAddress& a, AddrOutcome o, const std::string& msg);
    void applyOpState(RunBatch& b, OpState st, const OpReport* rep);
    void halt(const std::string& msg) { halted_ = true; if (haltError_.empty()) haltError_ = msg; }

    Phase phase_ = Phase::Idle;
    RunEnd end_ = RunEnd::None;
    RunConfig cfg_;
    std::vector<RunAddress> addrs_;
    std::vector<RunBatch> batches_;
    std::uint64_t gen_ = 0;
    std::size_t cur_ = 0;
    int callAddr_ = -1;
    bool callInFlight_ = false, pollInFlight_ = false, stopping_ = false, halted_ = false, lockResync_ = false;
    double callPostedAt_ = 0, pollPostedAt_ = 0, nextCallAt_ = 0, nextPollAt_ = 0, syncHoldUntil_ = 0,
           unlockHoldUntil_ = 0;
    int deadStreak_ = 0, opFailStreak_ = 0;
    std::string haltError_;
    RunWait lastWait_ = RunWait::None;
    double lastPumpAt_ = 0.0;
};

// ── Scope scan (the Merge/Consolidate dialog's listunspent + z_listunspent) ─────────────────────────
// Both calls are whole-wallet scans under the daemon's cs_main. The dialog used to post the pair again on
// the frame after any failure, every frame for as long as it stayed open (and forever on Lite, which has
// no node). This decides when a scan may be posted; the dialog owns one ScopeScan.

inline double steadySeconds()   // ImGui::GetTime() stands still while the window is minimized
{
    return std::chrono::duration<double>(std::chrono::steady_clock::now().time_since_epoch()).count();
}

inline constexpr long   kScopeCallTimeoutSec = 120;   // each call; a large wallet can take minutes
inline constexpr double kScopeInFlightMaxSec = 1800;  // a scan outstanding this long is presumed lost
inline constexpr int    kScopeMaxAutoRetries = 3;

enum class ScopePhase { Idle, Loading, Loaded, Failed };

struct ScopeScan {
    ScopePhase phase = ScopePhase::Idle;
    std::uint32_t gen = 0;          // bumped per dialog session; a result from another session is dropped
    bool inFlight = false;          // a scan is posted and its result not yet in (survives close/reopen)
    std::uint32_t inFlightGen = 0;  // the session that posted it: only its result clears inFlight
    double startedAt = 0.0;
    int failures = 0;
    bool autoRetry = false;         // the last failure was "node not ready yet": retry by itself
    double retryAt = 0.0;
    std::string error;
};

struct ScopeInputs {
    bool mergeMode = false;
    bool busyWithOp = false;   // a merge was submitted or is in flight
    bool nodeReady = false;    // App::walletScanReady()
    double now = 0.0;
};

enum class ScopeWant { Nothing, Post, WaitNode, WaitBackoff };

ScopeWant scopeScanWant(const ScopeScan& scan, const ScopeInputs& in);
// Mark a scan posted: Loading, in flight for this session. Returns the generation to finish it with.
std::uint32_t scopeScanBegin(ScopeScan& scan, double now);
// Record a scan's result. True when it belongs to the current session and succeeded: apply its data.
bool scopeScanFinish(ScopeScan& scan, std::uint32_t gen, const std::string& error, rpc::RpcFailureKind kind,
                     double now);
// The dialog opened: start a new session, unless a scan of the last one is still loading (adopt it).
void scopeScanOpen(ScopeScan& scan, double now);
// The data went stale (reconnect): the next want is a fresh scan. An in-flight scan stays guarded.
void scopeScanInvalidate(ScopeScan& scan);
// The user pressed Retry after a failure.
void scopeScanRetry(ScopeScan& scan);
double scopeRetryDelaySec(int failures);   // 10, 30, then 60 s
// Retry by itself only when the node wasn't ready (refused, warming up, held back as busy, aborted):
// not after a reply timeout or a real error, each of which would cost another pair of full scans.
bool scopeFailureRetries(rpc::RpcFailureKind kind, const std::string& message);

}  // namespace services
}  // namespace dragonx