rm-circle-team-router/tools/server/rmc-member-watch.py

#!/usr/bin/env python3
"""RM Circle: tell Marty the moment a watched position upgrades, and who caught the money.

Built for one question he asked on 24 September 2026: when #220 buys Fastigium, does Orlando's
#30 actually get paid? The chain said it should -- #220 sits at Apex with four uplines between
them and #30, and a level-6 upgrade skips exactly those four -- but "should" is not "did", so
this reports what the CONTRACT DID, never what the walk predicted. If somebody else catches it,
that is what the message says.

  MemberUpgraded(uint48 id, uint8 newLevel, ...)          topics[1]=id,  data[0]=new level
  UplineRewarded(uint48 to, uint48 from, uint8, uint256)  topics[1]=to, topics[2]=from,
                                                          data[0]=level, data[1]=amount wei

Two rules this thing lives by:

  1. It NEVER advances its block pointer on a failed scan. A watchdog that quietly skips the
     blocks it could not read is worse than no watchdog: the event passes, nothing is said, and
     everyone assumes silence means nothing happened.
  2. If it cannot reach the chain several runs running, it SAYS SO. Going blind is itself news.

  WATCH=220 python3 rmc-member-watch.py             # normal (cron)
  DRY=1 WATCH=220 python3 rmc-member-watch.py       # print only, no send, no state write
  BACK=50000 DRY=1 WATCH=139 python3 ...            # rehearse on history instead of the pointer
"""
import json
import os
import pathlib
import sys
import time
import urllib.request

CONTRACT = "0x33bdaeefd6d17d80ae53816c916dfb26c4fb2daf"
# (url, the widest block span that endpoint accepts in one eth_getLogs). All four were measured
# from core itself, not assumed: polygon-rpc.com, ankr, blastapi and blockpi all refuse it outright.
# The order matters -- publicnode and tenderly take the whole window in one request, so a normal
# run costs two calls; the 50-block pair are chunked fallbacks for when those rate-limit. They do:
# the RM Circle app already polls publicnode from this same IP, so 429s here are routine, not alarming.
RPCS = [("https://polygon-bor-rpc.publicnode.com", 2000),
        ("https://polygon.gateway.tenderly.co", 2000),
        ("https://polygon.drpc.org", 50),
        ("https://1rpc.io/matic", 50)]
ROUNDS = 2               # a whole pass over every endpoint, twice, before calling a run failed
ROUND_WAIT = 20          # seconds between those passes, to let a rate limit lapse
T_UPGRADED = "0xc0b79a9e133d4dcbb1a606a57591d98dce93c7d5c86197a5caae22c4a1480049"
T_UPLINE = "0x6cdacf6757bdea1fcf01918794f79ca1e71b088fbec8fde7b027a963cfab0ae9"
LEVELS = ['Scintilla', 'Ascensus', 'Fabrica', 'Culmen', 'Apex', 'Fastigium', 'Vertex', 'Corona']
NAMED = {30: 'Orlando'}

WATCH = int(os.environ.get("WATCH", "220"))
DRY = os.environ.get("DRY") == "1"
BACK = int(os.environ.get("BACK", "0"))
STATE = pathlib.Path(os.environ.get("STATE_FILE", "/root/.rmc-member-watch-%d.json" % WATCH))
TOKEN_FILE = pathlib.Path(os.environ.get("TG_TOKEN_FILE", "/root/.mbhermes-telegram-token"))
CHAT_ID = os.environ.get("TG_CHAT", "1289244227")
WARN = "\u26a0\ufe0f"
BLIND_AFTER = 4          # consecutive failed runs before we admit we have gone blind


def rpc(url, method, params, timeout=45):
    body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}).encode()
    # publicnode 403s the default Python-urllib agent, so always send one of our own
    req = urllib.request.Request(url, data=body, headers={
        "Content-Type": "application/json", "User-Agent": "rmc-member-watch/1.0"})
    out = json.load(urllib.request.urlopen(req, timeout=timeout))
    if "error" in out:
        raise RuntimeError(out["error"].get("message", "rpc error"))
    return out["result"]


def head():
    last = None
    for attempt in range(ROUNDS):
        for url, _ in RPCS:
            try:
                return int(rpc(url, "eth_blockNumber", []), 16), url
            except Exception as exc:
                last = exc
        if attempt + 1 < ROUNDS:
            time.sleep(ROUND_WAIT)
    raise last


def get_logs(lo, hi):
    """Every log for the watched member between two blocks, from whichever endpoint answers.
    Chunked to each endpoint's own range limit, so a 50-block-max fallback still works."""
    pad = "0x" + format(WATCH, "064x")
    filters = [
        {"topics": [T_UPGRADED, pad]},              # the watched member upgrading
        {"topics": [T_UPLINE, None, pad]},          # what that upgrade paid, and to whom
    ]
    last = None
    for attempt in range(ROUNDS):
        for url, span in RPCS:
            try:
                out = []
                for f in filters:
                    start = lo
                    while start <= hi:
                        end = min(start + span - 1, hi)
                        q = dict(f)
                        q.update({"address": CONTRACT, "fromBlock": hex(start), "toBlock": hex(end)})
                        out += rpc(url, "eth_getLogs", [q])
                        start = end + 1
                return out, url
            except Exception as exc:
                last = exc                          # start over on the next endpoint
        if attempt + 1 < ROUNDS:
            time.sleep(ROUND_WAIT)                  # every endpoint balked; give the limits a rest
    raise last


def send(text):
    token = TOKEN_FILE.read_text().strip()
    body = json.dumps({"chat_id": CHAT_ID, "text": text, "parse_mode": "HTML",
                       "disable_web_page_preview": True}).encode()
    req = urllib.request.Request("https://api.telegram.org/bot%s/sendMessage" % token,
                                 data=body, headers={"Content-Type": "application/json"})
    urllib.request.urlopen(req, timeout=30).read()


def lvname(n):
    return LEVELS[n - 1] if 1 <= n <= len(LEVELS) else "level %d" % n


def who(mid):
    return "#%d (%s)" % (mid, NAMED[mid]) if mid in NAMED else "#%d" % mid


def load():
    try:
        return json.loads(STATE.read_text())
    except Exception:
        return {}


def compose(tx, upgrades, pays):
    """One message per transaction: what was bought, and where every POL of it went."""
    lines = []
    for lv in sorted(upgrades):
        lines.append("<b>#%d upgraded to %s</b> (level %d)." % (WATCH, lvname(lv), lv))
    if not upgrades:
        lines.append("<b>#%d generated a pass-up payment.</b>" % WATCH)
    for to, pol in pays:
        lines.append("\n%s POL went to %s." % (format(pol, ",.2f"), who(to)))
    if not pays:
        lines.append("\nNo pass-up payment was logged against it, so it fell through to the "
                     "admin address: nobody upline qualified.")
    lines.append("\nhttps://polygonscan.com/tx/" + tx)
    return "\n".join(lines)


def stumble(st, exc, where):
    """One place for every way a run can fail to read the chain.

    Whatever broke -- the block number or the log scan -- the pointer stays put and the failure
    is COUNTED. An earlier version only counted failures from the log scan, so a total outage
    (where even the block number is unreachable) raised straight past the counter: the one kind
    of outage worth shouting about was the one that would have stayed silent.
    """
    fails = int(st.get("fails", 0)) + 1
    print("%s FAILED (%d in a row): %s" % (where, fails, exc), file=sys.stderr)
    if fails >= BLIND_AFTER and not st.get("blindAlerted") and not DRY:
        try:
            send(WARN + " <b>The #%d watchdog cannot reach the chain.</b>\n\n"
                 "%d runs in a row have failed, so nothing past block %s has been read and an "
                 "upgrade could pass unnoticed. Last error: %s"
                 % (WATCH, fails, st.get("lastBlock", "?"), exc))
            st["blindAlerted"] = True
        except Exception as e2:
            # Telegram is unreachable too; log it and let the next run try again
            print("could not send the blind alert: %s" % e2, file=sys.stderr)
    st["fails"] = fails
    if not DRY:
        STATE.write_text(json.dumps(st))
    sys.exit(1)


def main():
    st = load()
    try:
        bn, _ = head()
    except Exception as exc:
        stumble(st, exc, "block number")

    if BACK:
        lo, hi = bn - BACK, bn
    elif "lastBlock" in st:
        lo, hi = st["lastBlock"] + 1, bn
    else:
        # first run: a short look back covers the gap between writing this and the first tick
        lo, hi = bn - 1000, bn
    if lo > hi:
        print("nothing new (head %d)" % bn, flush=True)
        return

    try:
        logs, src = get_logs(lo, hi)
    except Exception as exc:
        stumble(st, exc, "scan %d-%d" % (lo, hi))

    # group by transaction so the upgrade and the payment it caused arrive as one message
    txs = {}
    for l in logs:
        tx = l["transactionHash"]
        d = txs.setdefault(tx, {"block": int(l["blockNumber"], 16), "upgrades": [], "pays": []})
        data = l["data"]
        word = lambda i: int(data[2 + i * 64: 2 + (i + 1) * 64], 16)
        if l["topics"][0] == T_UPGRADED:
            d["upgrades"].append(word(0))
        else:
            d["pays"].append((int(l["topics"][1], 16), word(1) / 1e18))

    seen = set(st.get("seen", []))
    order = sorted(txs.items(), key=lambda kv: kv[1]["block"])
    fresh = [(t, d) for t, d in order if t not in seen]
    print("scanned %d-%d via %s: %d log(s), %d tx, %d new"
          % (lo, hi, src, len(logs), len(txs), len(fresh)), flush=True)

    for tx, d in fresh:
        msg = compose(tx, d["upgrades"], d["pays"])
        if DRY:
            print("--- would send ---\n" + msg, flush=True)
        else:
            send(msg)
            print("telegram: sent for " + tx, flush=True)
        seen.add(tx)

    if not DRY:
        st.update({"lastBlock": hi, "seen": sorted(seen)[-200:], "fails": 0,
                   "blindAlerted": False, "at": int(time.time())})
        STATE.write_text(json.dumps(st))


if __name__ == "__main__":
    try:
        main()
    except SystemExit:
        raise
    except Exception as exc:
        print("member_watch failed: %s" % exc, file=sys.stderr)
        sys.exit(1)