#include"Boss/Mod/FeeMonitor.hpp" #include"Boss/Msg/ChannelDestruction.hpp" #include"Boss/Msg/CommandFail.hpp" #include"Boss/Msg/CommandRequest.hpp" #include"Boss/Msg/CommandResponse.hpp" #include"Boss/Msg/DbResource.hpp" #include"Boss/Msg/ManifestCommand.hpp" #include"Boss/Msg/Manifestation.hpp" #include"Boss/Msg/MonitorFeeByBalance.hpp" #include"Boss/Msg/MonitorFeeByTheory.hpp" #include"Boss/Msg/MonitorFeeSetChannel.hpp" #include"Boss/Msg/MonitorFeeBySize.hpp" #include"Boss/Msg/PeerMedianChannelFee.hpp" #include"Ev/Io.hpp" #include"Ev/coroutine.hpp" #include"Ev/now.hpp" #include"Ev/yield.hpp" #include"Json/Out.hpp" #include"S/Bus.hpp" #include"Sqlite3.hpp" #include namespace { template void bind_optional(Sqlite3::Query& q, char const* name , std::optional const& value) { if (value) q.bind(name, *value); else q.bind(name, nullptr); } template void add_optional_int( Obj& obj , char const* name , Sqlite3::Row& r , std::size_t& idx ) { auto value = r.get(idx++); auto is_null = r.get(idx++) != 0; if (is_null) obj.field(name, nullptr); else obj.field(name, value); } template void add_optional_double( Obj& obj , char const* name , Sqlite3::Row& r , std::size_t& idx ) { auto value = r.get(idx++); auto is_null = r.get(idx++) != 0; if (is_null) obj.field(name, nullptr); else obj.field(name, value); } bool parse_optional_number( Jsmn::Object const& value , std::optional& out ) { if (value.is_null()) return true; if (!value.is_number()) return false; out = (double) value; return true; } } namespace Boss { namespace Mod { void FeeMonitor::start() { bus.subscribe([this](Msg::DbResource const& m) { return on_db(m); }); bus.subscribe([this](Msg::PeerMedianChannelFee const& m) { return on_baseline(m); }); bus.subscribe([this](Msg::ChannelDestruction const& d) { auto it = peers.find(d.peer); if (it == peers.end()) return Ev::lift(); peers.erase(it); return Ev::lift(); }); bus.subscribe([this](Msg::MonitorFeeBySize const& m) { return on_size(m); }); bus.subscribe([this](Msg::MonitorFeeByBalance const& m) { return on_balance(m); }); bus.subscribe([this](Msg::MonitorFeeByTheory const& m) { return on_price(m); }); bus.subscribe([this](Msg::MonitorFeeSetChannel const& m) { return on_set(m); }); bus.subscribe([this](Msg::Manifestation const&) { return bus.raise(Msg::ManifestCommand{ "clboss-feemon-history", "nodeid [since] [before]", "Show fee modifier history for nodeid between since and before.", false }) + bus.raise(Msg::ManifestCommand{ "clboss-feemon-peers", "[since] [before]", "Show peer nodeids that have fee modifier history between since and before.", false }); }); bus.subscribe([this](Msg::CommandRequest const& req) { auto id = req.id; auto paramfail = [this, id]() { return bus.raise(Msg::CommandFail{ id, -32602, "Parameter failure", Json::Out::empty_object() }); }; if (req.command == "clboss-feemon-history") { auto nodeid_j = Jsmn::Object(); auto since_j = Jsmn::Object(); auto before_j = Jsmn::Object(); auto params = req.params; if (params.is_object()) { auto has_nodeid = params.has("nodeid"); auto has_since = params.has("since"); auto has_before = params.has("before"); if (!has_nodeid) return paramfail(); if (params.size() != std::size_t( has_nodeid + has_since + has_before )) return paramfail(); nodeid_j = params["nodeid"]; if (has_since) since_j = params["since"]; if (has_before) before_j = params["before"]; } else if (params.is_array()) { if (params.size() < 1 || params.size() > 3) return paramfail(); nodeid_j = params[0]; if (params.size() >= 2) since_j = params[1]; if (params.size() >= 3) before_j = params[2]; } else { return paramfail(); } if (!nodeid_j.is_string()) return paramfail(); auto nodeid_s = std::string(nodeid_j); if (!Ln::NodeId::valid_string(nodeid_s)) return paramfail(); nodeid_s = std::string(Ln::NodeId(nodeid_s)); auto since = std::optional(); auto before = std::optional(); if (!parse_optional_number(since_j, since)) return paramfail(); if (!parse_optional_number(before_j, before)) return paramfail(); if (since && before && *since > *before) return paramfail(); return db_transact().then([this, id, nodeid_s, since, before](Sqlite3::Tx tx) { auto q = tx.query(R"QRY( SELECT e.id, e.ts, e.peer_id, e.set_base, e.set_base IS NULL, e.set_ppm, e.set_ppm IS NULL, e.baseline_base, e.baseline_base IS NULL, e.baseline_ppm, e.baseline_ppm IS NULL, e.size_mult, e.size_mult IS NULL, e.size_total_peers, e.size_total_peers IS NULL, e.size_less_peers, e.size_less_peers IS NULL, e.balance_mult, e.balance_mult IS NULL, e.balance_our_msat, e.balance_our_msat IS NULL, e.balance_total_msat, e.balance_total_msat IS NULL, e.price_level, e.price_level IS NULL, e.price_mult, e.price_mult IS NULL, e.price_cards_left, e.price_cards_left IS NULL, e.price_center, e.price_center IS NULL, e.mult_product, e.mult_product IS NULL, e.est_base, e.est_base IS NULL, e.est_ppm, e.est_ppm IS NULL FROM feemon_change_events e JOIN feemon_peers p ON e.peer_id = p.id WHERE p.node_id = :node_id AND (:since IS NULL OR e.ts >= :since) AND (:before IS NULL OR e.ts <= :before) ORDER BY e.ts ASC; )QRY"); q.bind(":node_id", nodeid_s); bind_optional(q, ":since", since); bind_optional(q, ":before", before); auto fetch = q.execute(); auto out = Json::Out(); auto top = out.start_object(); top.field("nodeid", nodeid_s); if (since) top.field("since", *since); if (before) top.field("before", *before); auto history = top.start_array("history"); for (auto& r : fetch) { auto row = history.start_object(); auto idx = std::size_t(0); row.field("id", r.get(idx++)); row.field("ts", static_cast(r.get(idx++))); row.field("peer_id", r.get(idx++)); add_optional_int(row, "set_base", r, idx); add_optional_int(row, "set_ppm", r, idx); add_optional_int(row, "baseline_base", r, idx); add_optional_int(row, "baseline_ppm", r, idx); add_optional_double(row, "size_mult", r, idx); add_optional_int(row, "size_total_peers", r, idx); add_optional_int(row, "size_less_peers", r, idx); add_optional_double(row, "balance_mult", r, idx); add_optional_int(row, "balance_our_msat", r, idx); add_optional_int(row, "balance_total_msat", r, idx); add_optional_int(row, "price_level", r, idx); add_optional_double(row, "price_mult", r, idx); add_optional_int(row, "price_cards_left", r, idx); add_optional_int(row, "price_center", r, idx); add_optional_double(row, "mult_product", r, idx); add_optional_int(row, "est_base", r, idx); add_optional_int(row, "est_ppm", r, idx); row.end_object(); } history.end_array(); top.end_object(); tx.commit(); return bus.raise(Msg::CommandResponse{id, out}); }); } else if (req.command == "clboss-feemon-peers") { auto since_j = Jsmn::Object(); auto before_j = Jsmn::Object(); auto params = req.params; if (params.is_object()) { auto has_since = params.has("since"); auto has_before = params.has("before"); if (params.size() != std::size_t(has_since + has_before)) return paramfail(); if (has_since) since_j = params["since"]; if (has_before) before_j = params["before"]; } else if (params.is_array()) { if (params.size() > 2) return paramfail(); if (params.size() >= 1) since_j = params[0]; if (params.size() >= 2) before_j = params[1]; } else if (!params.is_null()) { return paramfail(); } auto since = std::optional(); auto before = std::optional(); if (!parse_optional_number(since_j, since)) return paramfail(); if (!parse_optional_number(before_j, before)) return paramfail(); if (since && before && *since > *before) return paramfail(); return db_transact().then([this, id, since, before](Sqlite3::Tx tx) { auto q = tx.query(R"QRY( SELECT DISTINCT p.node_id FROM feemon_change_events e JOIN feemon_peers p ON e.peer_id = p.id WHERE (:since IS NULL OR e.ts >= :since) AND (:before IS NULL OR e.ts <= :before) ORDER BY p.node_id ASC; )QRY"); bind_optional(q, ":since", since); bind_optional(q, ":before", before); auto fetch = q.execute(); auto out = Json::Out(); auto top = out.start_object(); if (since) top.field("since", *since); if (before) top.field("before", *before); auto peers = top.start_array("peers"); for (auto& r : fetch) peers.entry(r.get(0)); peers.end_array(); top.end_object(); tx.commit(); return bus.raise(Msg::CommandResponse{id, out}); }); } return Ev::lift(); }); } Ev::Io FeeMonitor::on_db(Msg::DbResource const& m) { db = m.db; co_await initialize_db(); co_return; } Ev::Io FeeMonitor::initialize_db() { auto tx = co_await db_transact(); tx.query_execute(R"QRY( CREATE TABLE IF NOT EXISTS feemon_peers ( id INTEGER PRIMARY KEY, node_id TEXT NOT NULL UNIQUE ); CREATE TABLE IF NOT EXISTS feemon_change_events ( id INTEGER PRIMARY KEY, ts REAL NOT NULL, peer_id INTEGER NOT NULL, set_base INTEGER, set_ppm INTEGER, baseline_base INTEGER, baseline_ppm INTEGER, size_mult REAL, size_total_peers INTEGER, size_less_peers INTEGER, balance_mult REAL, balance_our_msat INTEGER, balance_total_msat INTEGER, price_level INTEGER, price_mult REAL, price_cards_left INTEGER, price_center INTEGER, mult_product REAL, est_base INTEGER, est_ppm INTEGER, FOREIGN KEY(peer_id) REFERENCES feemon_peers(id) ); CREATE INDEX IF NOT EXISTS feemon_change_events_peer_ts_idx ON feemon_change_events(peer_id, ts); CREATE INDEX IF NOT EXISTS feemon_change_events_ts_peer_idx ON feemon_change_events(ts, peer_id); )QRY"); tx.commit(); co_return; } Ev::Io FeeMonitor::db_transact() { while (!db) { co_await Ev::yield(); } co_return co_await db.transact(); } std::uint64_t FeeMonitor::get_peer_id(Sqlite3::Tx& tx, Ln::NodeId const& node) { auto fetch = tx.query(R"QRY( SELECT id FROM feemon_peers WHERE node_id = :node_id ; )QRY") .bind(":node_id", std::string(node)) .execute() ; for (auto& r : fetch) return r.get(0); tx.query(R"QRY( INSERT OR IGNORE INTO feemon_peers VALUES(NULL, :node_id); )QRY") .bind(":node_id", std::string(node)) .execute() ; return get_peer_id(tx, node); } Ev::Io FeeMonitor::on_baseline(Msg::PeerMedianChannelFee const& m) { auto& info = peers[m.node]; info.baseline_base = m.base; info.baseline_ppm = m.proportional; co_return; } Ev::Io FeeMonitor::on_size(Msg::MonitorFeeBySize const& m) { auto& info = peers[m.node]; info.size_mult = m.mult; info.size_total_peers = m.total_peers; info.size_less_peers = m.less_peers; co_return; } Ev::Io FeeMonitor::on_balance(Msg::MonitorFeeByBalance const& m) { auto& info = peers[m.node]; info.balance_mult = m.mult; info.balance_our_msat = m.our_msat; info.balance_total_msat = m.total_msat; co_return; } Ev::Io FeeMonitor::on_price(Msg::MonitorFeeByTheory const& m) { auto& info = peers[m.node]; info.price_level = m.level; info.price_mult = m.mult; info.price_cards_left = m.cards_left; info.price_center = m.center; co_return; } Ev::Io FeeMonitor::on_set(Msg::MonitorFeeSetChannel const& m) { auto snapshot = peers[m.node]; auto ts = Ev::now(); auto mult_product = std::optional(); auto est_base = std::optional(); auto est_ppm = std::optional(); if ( snapshot.baseline_base && snapshot.baseline_ppm && snapshot.size_mult && snapshot.balance_mult && snapshot.price_mult ) { mult_product = (*snapshot.size_mult) * (*snapshot.balance_mult) * (*snapshot.price_mult); auto base = std::llround( double(*snapshot.baseline_base) * *mult_product ); auto ppm = std::llround( double(*snapshot.baseline_ppm) * *mult_product ); if (ppm == 0) ppm = 1; est_base = base; est_ppm = ppm; } auto tx = co_await db_transact(); auto peer_id = get_peer_id(tx, m.node); auto q = tx.query(R"QRY( INSERT INTO feemon_change_events ( ts, peer_id, set_base, set_ppm, baseline_base, baseline_ppm, size_mult, size_total_peers, size_less_peers, balance_mult, balance_our_msat, balance_total_msat, price_level, price_mult, price_cards_left, price_center, mult_product, est_base, est_ppm ) VALUES ( :ts, :peer_id, :set_base, :set_ppm, :baseline_base, :baseline_ppm, :size_mult, :size_total_peers, :size_less_peers, :balance_mult, :balance_our_msat, :balance_total_msat, :price_level, :price_mult, :price_cards_left, :price_center, :mult_product, :est_base, :est_ppm ); )QRY"); q.bind(":ts", ts) .bind(":peer_id", peer_id) .bind(":set_base", m.base) .bind(":set_ppm", m.proportional); bind_optional(q, ":baseline_base", snapshot.baseline_base); bind_optional(q, ":baseline_ppm", snapshot.baseline_ppm); bind_optional(q, ":size_mult", snapshot.size_mult); bind_optional(q, ":size_total_peers", snapshot.size_total_peers); bind_optional(q, ":size_less_peers", snapshot.size_less_peers); bind_optional(q, ":balance_mult", snapshot.balance_mult); bind_optional(q, ":balance_our_msat", snapshot.balance_our_msat); bind_optional(q, ":balance_total_msat", snapshot.balance_total_msat); bind_optional(q, ":price_level", snapshot.price_level); bind_optional(q, ":price_mult", snapshot.price_mult); bind_optional(q, ":price_cards_left", snapshot.price_cards_left); bind_optional(q, ":price_center", snapshot.price_center); bind_optional(q, ":mult_product", mult_product); bind_optional(q, ":est_base", est_base); bind_optional(q, ":est_ppm", est_ppm); q.execute(); tx.commit(); co_return; } FeeMonitor::FeeMonitor(S::Bus& bus_) : bus(bus_) { start(); } FeeMonitor::~FeeMonitor() =default; }}