入账监听实现指南

#  ┌── getblockbylimitnext ── 一次 100 个块 ────────────┐
#  │                                                    ▼
#  │   本地解析：TransferContract / TriggerSmartContract
#  │                        │
#  │                        ▼  Redis SET 过滤（零请求）
#  │                   命中监听地址？
#  │                    否 ──► 丢弃（99.99% 走这条）
#  │                    是 ──► gettransactioninfobyblocknum
#  │                             │  确认 receipt + Transfer 日志
#  │                             ▼
#  │                    落库 status=detected（唯一索引幂等）
#  │                             │
#  │                             ▼  等 20 个块固化
#  │                    solidity 二次核验 ──► status=confirmed ──► 入账
#  └── 水位 +100 落库，继续下一批 ─────────────────────┘

# Correct: watermark-driven catch-up. Never "fetch the latest block".
BATCH = 100          # hard cap of getblockbylimitnext
cursor = db_load_cursor()                 # last processed block, from DB

while True:
    head = rpc("wallet", "getnowblock")["block_header"]["raw_data"]["number"]
    if cursor >= head:
        time.sleep(1.5)                   # caught up — idle, costs almost nothing
        continue

    start = cursor + 1
    end   = min(start + BATCH, head + 1)  # endNum is EXCLUSIVE
    blocks = rpc("wallet", "getblockbylimitnext",
                 {"startNum": start, "endNum": end}).get("block", [])

    if len(blocks) != end - start:        # never trust a short read
        log.warning("short read: want %d got %d", end - start, len(blocks))
        time.sleep(2); continue            # retry, do NOT advance the cursor

    process(blocks)
    cursor = end - 1
    db_save_cursor(cursor)                # advance only after the batch is durable

# Native TRX — everything comes straight from the block, zero extra requests
for tx in block.get("transactions", []):
    if tx.get("ret", [{}])[0].get("contractRet") != "SUCCESS":
        continue                               # failed on-chain — ignore
    c = tx["raw_data"]["contract"][0]
    if c["type"] != "TransferContract":
        continue
    v  = c["parameter"]["value"]
    to = to_base58(v["to_address"])           # 41dc6b… -> T…
    if not redis.sismember("watch:addrs", to):
        continue                               # not ours — 99.99% land here
    hit(kind="TRX", txid=tx["txID"], to=to,
        sender=to_base58(v["owner_address"]),
        amount=int(v["amount"]), decimals=6)     # amount is in SUN

USDT      = "41a614f803b6fd780986a42c78ec9c7f77e6ded13c"  # WITH 41 prefix (block)
USDT_LOG  = USDT[2:]                                        # WITHOUT it (logs!)
SELECTOR  = "a9059cbb"                                     # transfer(address,uint256)
TRANSFER  = "ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"

# --- step 1: candidates from the block, no extra request ---------------
c = tx["raw_data"]["contract"][0]
v = c["parameter"]["value"]
if c["type"] == "TriggerSmartContract" \
   and v.get("contract_address") == USDT \
   and v.get("data", "")[:8] == SELECTOR \
   and tx.get("ret", [{}])[0].get("contractRet") == "SUCCESS":
    data = v["data"]
    to   = to_base58("41" + data[8:72][-40:])   # arg is 20 bytes, add 41
    if redis.sismember("watch:addrs", to):
        candidates.append((tx["txID"], to, int(data[72:136], 16)))

# --- step 2: only for blocks with candidates, pull receipts ------------
infos = rpc("wallet", "gettransactioninfobyblocknum", {"num": block_num})
by_id = {i["id"]: i for i in infos}

for txid, to, calldata_amt in candidates:
    info = by_id.get(txid)
    if not info or info.get("receipt", {}).get("result") != "SUCCESS":
        continue                                  # fake deposit — drop it
    for lg in info.get("log", []):
        if lg.get("address") != USDT_LOG:         # must be the REAL contract
            continue
        tp = lg.get("topics", [])
        if len(tp) != 3 or tp[0] != TRANSFER:     # must be a standard Transfer
            continue
        if to_base58("41" + tp[2][-40:]) != to:
            continue
        amount = int(lg["data"], 16)             # trust the LOG, not calldata
        hit(kind="USDT", txid=txid, to=to, amount=amount, decimals=6)
        break

# Final check before crediting. Re-ask the node via the SOLIDITY endpoint:
# solidified data cannot be rolled back. Runs once per deposit, ~20 blocks later.
def verify_final(dep):
    if dep["kind"] == "TRX":
        tx = rpc("walletsolidity", "gettransactionbyid", {"value": dep["txid"]})
        if not tx.get("txID"):
            return False, "not solidified yet, or reorged out"
        if tx["ret"][0]["contractRet"] != "SUCCESS":
            return False, "execution failed"
        v = tx["raw_data"]["contract"][0]["parameter"]["value"]
        if to_base58(v["to_address"]) != dep["to"] or int(v["amount"]) != dep["amount"]:
            return False, "recipient or amount mismatch"
        return True, ""

    info = rpc("walletsolidity", "gettransactioninfobyid", {"value": dep["txid"]})
    if not info.get("id"):
        return False, "not solidified yet, or reorged out"
    if info.get("blockNumber") != dep["block"]:
        return False, "block height mismatch — possible reorg"
    if info.get("receipt", {}).get("result") != "SUCCESS":
        return False, "execution failed — fake deposit"
    total = sum(int(l["data"], 16) for l in info.get("log", [])
                if l.get("address") == USDT_LOG
                and len(l.get("topics", [])) == 3
                and l["topics"][0] == TRANSFER
                and to_base58("41" + l["topics"][2][-40:]) == dep["to"])
    if total != dep["amount"]:
        return False, "amount mismatch"
    return True, ""

# redis.conf — deposit monitor

maxmemory 4gb
maxmemory-policy noeviction     # NEVER allkeys-lru / allkeys-random

appendonly yes                  # the pending-notify queue must survive a restart
appendfsync everysec

bind 127.0.0.1                  # never expose Redis to the internet
requirepass <strong-password>
rename-command FLUSHALL ""       # flushing the watchlist = missing deposits

# --- state-ish: no TTL, rebuildable from the database -----------------
v1:watch:addrs          SET     # every watched address; the hot path filter
v1:notify:queue         LIST    # deposits pending push to your business system

# --- cache: TTL is fine, losing it costs nothing ----------------------
v1:seen:{txid}:{idx}    STRING  TTL 3h   # dedupe shortcut before hitting the DB
v1:deposit:{address}    ZSET    TTL 3h   # score=block, recent deposits per address
v1:block:{num}          STRING  TTL 3h   # raw block cache, for replay/debugging

# memory: 100k watched addresses + 3h of hits ~= tens of MB.
# caching raw blocks too: 3600 blocks * a few hundred KB ~= 1-2 GB.

-- scan watermark: the single source of truth for "how far have I got"
CREATE TABLE scan_cursor (
    chain       TEXT PRIMARY KEY,          -- 'tron'
    block_num   BIGINT NOT NULL,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE deposits (
    id          BIGSERIAL PRIMARY KEY,
    txid        TEXT   NOT NULL,
    log_index   INT    NOT NULL DEFAULT 0,   -- one tx can hold many transfers
    block_num   BIGINT NOT NULL,
    block_time  TIMESTAMPTZ NOT NULL,
    currency    TEXT   NOT NULL,            -- 'TRX' | 'USDT'
    contract    TEXT   NOT NULL DEFAULT '',  -- empty for native TRX
    from_addr   TEXT   NOT NULL,
    to_addr     TEXT   NOT NULL,
    amount      NUMERIC(38,0) NOT NULL,     -- smallest unit, integer only
    decimals    INT    NOT NULL,
    status      TEXT   NOT NULL,            -- detected|confirmed|credited|rejected|orphaned
    reject_note TEXT   NOT NULL DEFAULT '',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- THE idempotency guarantee. Not your application code.
CREATE UNIQUE INDEX uq_deposits ON deposits (txid, log_index);
CREATE INDEX ix_deposits_pending ON deposits (status, block_num);
CREATE INDEX ix_deposits_addr ON deposits (to_addr, created_at DESC);

#!/usr/bin/env python3
# TRON deposit monitor — native TRX + TRC20(USDT)
# docs: https://www.trxapi.io/docs/deposit-monitor.html
import hashlib, json, logging, time, urllib.request

API_KEY  = "YOUR_API_KEY"
BASE     = "https://api.trxapi.io/api/v1/openapi/node"
USDT     = "41a614f803b6fd780986a42c78ec9c7f77e6ded13c"
USDT_LOG = USDT[2:]
SELECTOR = "a9059cbb"
TRANSFER = "ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
BATCH    = 100      # hard cap of getblockbylimitnext
CONFIRMS = 20       # solidity lags ~18 blocks; 20 gives margin

log = logging.getLogger("deposit")

def rpc(kind, method, body=None, retries=5):
    url = f"{BASE}/{kind}/{method}?apikey={API_KEY}"
    for i in range(retries):
        try:
            req = urllib.request.Request(
                url, method="POST",
                data=json.dumps(body or {}).encode(),
                headers={"Content-Type": "application/json"})
            with urllib.request.urlopen(req, timeout=30) as r:
                return json.load(r)
        except Exception as e:
            wait = 2 ** i                       # 1s, 2s, 4s... on 429/timeout
            log.warning("rpc %s failed (%s), retry in %ds", method, e, wait)
            time.sleep(wait)
    raise RuntimeError(f"rpc {method} failed after {retries} tries")

# ---- hex(41-prefixed) -> Base58Check (T...) --------------------------
B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
def to_base58(hex41):
    raw = bytes.fromhex(hex41)
    chk = hashlib.sha256(hashlib.sha256(raw).digest()).digest()[:4]
    n   = int.from_bytes(raw + chk, "big")
    out = ""
    while n:
        n, r = divmod(n, 58)
        out  = B58[r] + out
    return out

# ---- plug in your own storage ---------------------------------------
def is_watched(addr):        ...   # redis.sismember("v1:watch:addrs", addr)
def save_deposits(rows):     ...   # INSERT ... ON CONFLICT (txid,log_index) DO NOTHING
def load_cursor():           ...   # SELECT block_num FROM scan_cursor
def save_cursor(n):          ...   # UPDATE scan_cursor SET block_num=$1

def scan_batch(start, end):
    """Fetch [start, end) in ONE request and extract everything we care about."""
    blocks = rpc("wallet", "getblockbylimitnext",
                 {"startNum": start, "endNum": end}).get("block", [])
    if len(blocks) != end - start:
        raise RuntimeError(f"short read: want {end-start}, got {len(blocks)}")

    hits = []
    for blk in blocks:
        raw = blk["block_header"]["raw_data"]
        num, ts = raw.get("number", 0), raw["timestamp"]
        cands = []
        for tx in blk.get("transactions", []):
            if tx.get("ret", [{}])[0].get("contractRet") != "SUCCESS":
                continue
            c = tx["raw_data"]["contract"][0]
            v = c["parameter"]["value"]

            if c["type"] == "TransferContract":          # native TRX
                to = to_base58(v["to_address"])
                if is_watched(to):
                    hits.append(dict(kind="TRX", currency="TRX", contract="",
                                     txid=tx["txID"], log_index=0, block=num,
                                     ts=ts, to=to,
                                     sender=to_base58(v["owner_address"]),
                                     amount=int(v["amount"]), decimals=6))

            elif c["type"] == "TriggerSmartContract":   # TRC20 candidate
                data = v.get("data", "")
                if v.get("contract_address") != USDT: continue
                if data[:8] != SELECTOR or len(data) < 136: continue
                to = to_base58("41" + data[8:72][-40:])
                if is_watched(to):
                    cands.append((tx["txID"], to,
                                  to_base58(v["owner_address"])))

        if cands:                                    # receipts ONLY when needed
            hits += confirm_trc20(num, ts, cands)
    return hits

def confirm_trc20(num, ts, cands):
    infos = rpc("wallet", "gettransactioninfobyblocknum", {"num": num})
    by_id = {i["id"]: i for i in infos}
    out   = []
    for txid, to, sender in cands:
        info = by_id.get(txid)
        if not info or info.get("receipt", {}).get("result") != "SUCCESS":
            log.info("drop fake deposit %s (execution failed)", txid)
            continue
        for idx, lg in enumerate(info.get("log", [])):
            if lg.get("address") != USDT_LOG:      # the real contract only
                continue
            tp = lg.get("topics", [])
            if len(tp) != 3 or tp[0] != TRANSFER:  # a standard Transfer only
                continue
            if to_base58("41" + tp[2][-40:]) != to:
                continue
            out.append(dict(kind="USDT", currency="USDT", contract=USDT,
                            txid=txid, log_index=idx, block=num, ts=ts,
                            to=to, sender=sender,
                            amount=int(lg["data"], 16), decimals=6))
    return out

def run():
    cursor = load_cursor()
    while True:
        head = rpc("wallet", "getnowblock")["block_header"]["raw_data"]["number"]
        if cursor >= head:
            time.sleep(1.5)
            continue
        start = cursor + 1
        end   = min(start + BATCH, head + 1)
        try:
            hits = scan_batch(start, end)
        except RuntimeError as e:
            log.warning("batch %d-%d failed: %s", start, end, e)
            time.sleep(2)
            continue                              # cursor NOT advanced
        # save_deposits() and save_cursor() belong in ONE transaction
        save_deposits(hits)
        save_cursor(end - 1)
        cursor = end - 1
        if hits:
            log.info("blocks %d-%d: %d deposits", start, end - 1, len(hits))

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run()

// TRON deposit monitor — core loop (Node.js 18+, no dependencies)
const API = 'https://api.trxapi.io/api/v1/openapi/node';
const KEY = 'YOUR_API_KEY';
const USDT = '41a614f803b6fd780986a42c78ec9c7f77e6ded13c';
const USDT_LOG = USDT.slice(2);
const TRANSFER = 'ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef';

async function rpc(kind, method, body = {}) {
  const r = await fetch(`${API}/${kind}/${method}?apikey=${KEY}`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(body),
  });
  if (r.status === 429) throw new Error('rate limited');   // back off
  return r.json();
}

async function scanBatch(start, end) {          // end is EXCLUSIVE
  const { block = [] } = await rpc('wallet', 'getblockbylimitnext',
                                     { startNum: start, endNum: end });
  if (block.length !== end - start)
    throw new Error(`short read: want ${end - start}, got ${block.length}`);

  const hits = [];
  for (const blk of block) {
    const num = blk.block_header.raw_data.number ?? 0;
    const cands = [];
    for (const tx of blk.transactions ?? []) {
      if (tx.ret?.[0]?.contractRet !== 'SUCCESS') continue;
      const c = tx.raw_data.contract[0], v = c.parameter.value;

      if (c.type === 'TransferContract') {
        const to = toBase58(v.to_address);
        if (await isWatched(to))
          hits.push({ kind: 'TRX', txid: tx.txID, logIndex: 0, block: num,
                      to, sender: toBase58(v.owner_address),
                      amount: BigInt(v.amount) });
      } else if (c.type === 'TriggerSmartContract' &&
                 v.contract_address === USDT &&
                 (v.data ?? '').startsWith('a9059cbb')) {
        const to = toBase58('41' + v.data.slice(8, 72).slice(-40));
        if (await isWatched(to)) cands.push({ txid: tx.txID, to });
      }
    }
    if (cands.length) hits.push(...await confirmTrc20(num, cands));
  }
  return hits;
}