Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions api/endpoints/awards.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,5 +10,4 @@ def awards(
receiver: str = Query(pattern=r"^[a-z0-9.-]{2,25}$"),
memo: str = Query(max_length=200),
) -> dict:
t = award_stats.totals(receiver, memo)
return {"count": t["count"], "initiators": t["initiators"], "total_viz": None}
return award_stats.totals(receiver, memo)
85 changes: 75 additions & 10 deletions api/helpers/award_stats.py
Original file line number Diff line number Diff line change
@@ -1,24 +1,30 @@
"""Materialize a permanent, queryable slice of `award` ops (receiver+memo
"""Materialize a permanent, queryable slice of award activity (receiver+memo
totals) for the write-side platform's "who awarded this post" endpoint.

Called from the parser thread via op_stream (sibling to notifications.dispatch).
Best-effort: failures are logged, never block parsing.
Two writers, one collection (`kind` tells them apart):
- `award` (real op) — from the live poller via op_stream, so counts and
initiators show up the instant the op lands in a head block.
- `receive_award` (virtual op, carries the actual SHARES payout) — from the
archival parser, which is the only path that sees virtual ops (~36s behind
the head, irreversible blocks only).

No `receive_award` (virtual op) capture: the live poller (parser.live_stream)
never sees virtual ops, only the archival parser does, and that parser doesn't
call emit_block_ops. So there is no live path to the actual VIZ payout amount
today; `total_viz` stays null at the endpoint layer.
Best-effort: failures are logged, never block parsing.
"""
from __future__ import annotations

import logging
import os
import time
from typing import Any

from helpers.db_client import get_db
from helpers.viz import convertShares, vesting_rate

logger = logging.getLogger(__name__)

RATE_TTL = float(os.getenv("AWARD_RATE_TTL", "600"))
_rate_cache: tuple[float, float] = (0.0, 0.0) # (fetched_at, rate)


def _coll():
return get_db()[os.getenv("COLLECTION_AWARDS", "award_ops")]
Expand All @@ -31,13 +37,15 @@ def ensure_indexes() -> None:


def dispatch(op: dict[str, Any]) -> None:
"""Live path: record a real `award` op (count + initiator, no payout)."""
try:
op_type = op["op"][0]
if op_type != "award":
return
body = op["op"][1] if len(op["op"]) > 1 else {}
doc = {
"op_id": op["_id"],
"kind": "award",
"initiator": body.get("initiator"),
"receiver": body.get("receiver"),
"memo": body.get("memo"),
Expand All @@ -53,10 +61,67 @@ def dispatch(op: dict[str, Any]) -> None:
logger.exception("award_stats.dispatch failed for op %s", op.get("_id"))


def capture_payouts(block: list[dict[str, Any]], blocknum: int) -> None:
"""Archival path: record each `receive_award` virtual op's SHARES payout.

`block` is get_ops_in_block's list (already normalized by save_block)."""
for i, tx in enumerate(block):
op = tx.get("op") or []
if not op or op[0] != "receive_award":
continue
body = op[1] if len(op) > 1 else {}
shares = body.get("shares")
# Distinct id namespace from the live path's float ids: the archival
# op index counts virtual ops too, so positions don't line up.
op_id = f"ra:{blocknum}:{i}"
try:
_coll().update_one(
{"op_id": op_id},
{"$setOnInsert": {
"op_id": op_id,
"kind": "receive_award",
"initiator": body.get("initiator"),
"receiver": body.get("receiver"),
"memo": body.get("memo"),
"shares": convertShares(shares) if isinstance(shares, str) else shares,
"timestamp": tx.get("timestamp"),
}},
upsert=True,
)
except Exception:
logger.exception("award_stats.capture_payouts failed for %s", op_id)


def _cached_rate() -> float | None:
global _rate_cache
fetched_at, rate = _rate_cache
if time.monotonic() - fetched_at < RATE_TTL and fetched_at:
return rate
try:
rate = vesting_rate()
except Exception:
logger.exception("award_stats: vesting_rate failed")
return rate or None
_rate_cache = (time.monotonic(), rate)
return rate


def totals(receiver: str, memo: str) -> dict:
query = {"receiver": receiver, "memo": memo}
coll = _coll()
# Pre-`kind` rows are live `award` rows; keep counting them.
awards = {"receiver": receiver, "memo": memo, "kind": {"$ne": "receive_award"}}
payouts = {"receiver": receiver, "memo": memo, "kind": "receive_award"}
agg = list(coll.aggregate([
{"$match": payouts},
{"$group": {"_id": None, "shares": {"$sum": "$shares"}}},
]))
total_shares = float(agg[0]["shares"]) if agg else 0.0
rate = _cached_rate()
return {
"count": coll.count_documents(query),
"initiators": coll.distinct("initiator", query),
"count": coll.count_documents(awards),
"initiators": coll.distinct("initiator", awards),
"total_shares": total_shares,
# ponytail: converted at today's vesting rate, not the rate at award
# time; store the rate per payout row if historical precision matters.
"total_viz": round(total_shares * rate, 3) if rate is not None else None,
}
12 changes: 2 additions & 10 deletions api/helpers/richlist_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from typing import Any

from helpers.db_client import get_db
from helpers.viz import convertShares, get_client
from helpers.viz import convertShares, get_client, vesting_rate

logger = logging.getLogger(__name__)

Expand All @@ -24,14 +24,6 @@
_LOOKUP_LIMIT = 1000 # graphene caps lookup_accounts / get_accounts at 1000


def _vesting_rate() -> float:
"""VIZ per SHARE at the current vesting rate (fund / total shares)."""
dgp = get_client().rpc.get_dynamic_global_properties()
fund = convertShares(dgp["total_vesting_fund"])
shares = convertShares(dgp["total_vesting_shares"])
return fund / shares if shares else 0.0


def _all_account_names(client: Any) -> list[str]:
"""Paginate lookup_accounts over the whole account set (sorted, inclusive
of the lower bound — so each page after the first repeats one name)."""
Expand Down Expand Up @@ -80,7 +72,7 @@ def _row(acc: dict[str, Any], rate: float) -> dict[str, Any]:
def refresh_once() -> int:
"""Rebuild the snapshot. Returns the number of accounts scanned."""
client = get_client()
rate = _vesting_rate()
rate = vesting_rate()
names = _all_account_names(client)

rows: list[dict[str, Any]] = []
Expand Down
8 changes: 8 additions & 0 deletions api/helpers/viz.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,14 @@ def convertShares(shares: str) -> float:
return float(shares.split(" ", 1)[0])


def vesting_rate() -> float:
"""VIZ per SHARE at the current vesting rate (fund / total shares)."""
dgp = get_dgp()
fund = convertShares(dgp["total_vesting_fund"])
shares = convertShares(dgp["total_vesting_shares"])
return fund / shares if shares else 0.0


# Backwards-compat alias for the old recursive function name.
def change_node(selectFirst: bool = False) -> None: # noqa: N802 - legacy name
init_node(select_first=selectFirst)
2 changes: 2 additions & 0 deletions api/parser/parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from time import sleep
from typing import NoReturn

from helpers import award_stats
from helpers.mongo import get_last_blocknum, save_block
from helpers.viz import get_client, get_last_block_in_chain, get_ops_in_block

Expand Down Expand Up @@ -78,6 +79,7 @@ def start_parsing() -> NoReturn:
# crash-loop (froze the tip at 81,288,426 for ~16h).
last_db_block = _
_update_key_index(block)
award_stats.capture_payouts(block, _)
if last_db_block % 100 == 0:
print(f"Saved block {_} (ch: {last_chain_block})")
else:
Expand Down
50 changes: 48 additions & 2 deletions api/tests/test_awards.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,10 @@ def test_dispatch_dedupes_same_op_id():
assert t["count"] == 1


def test_awards_endpoint(client):
def test_awards_endpoint(client, _viz):
award_stats.ensure_indexes()
_rate(_viz)
award_stats.capture_payouts(_archival_block(), 50)
award_stats.dispatch(_op("award", {"initiator": "alice", "receiver": "bob", "memo": "post-1"}, op_id=1.0001))
award_stats.dispatch(_op("award", {"initiator": "carol", "receiver": "bob", "memo": "post-1"}, op_id=2.0001))

Expand All @@ -61,4 +63,48 @@ def test_awards_endpoint(client):
body = resp.json()
assert body["count"] == 2
assert set(body["initiators"]) == {"alice", "carol"}
assert body["total_viz"] is None
assert body["total_shares"] == 2.5
assert body["total_viz"] == 5.0


def _archival_block(ts=dt.datetime(2026, 7, 5, tzinfo=dt.UTC)):
"""get_ops_in_block shape after save_block: real op first, then virtuals."""
return [
{"timestamp": ts, "op": ["award", {"initiator": "alice", "receiver": "bob", "memo": "post-1", "energy": 100}]},
{"timestamp": ts, "op": ["receive_award", {"initiator": "alice", "receiver": "bob", "memo": "post-1", "shares": "2.500000 SHARES"}]},
{"timestamp": ts, "op": ["receive_award", {"initiator": "carol", "receiver": "bob", "memo": "post-2", "shares": "9.000000 SHARES"}]},
]


def _rate(viz, fund="200.000 VIZ", shares="100.000000 SHARES"):
viz.rpc.get_dynamic_global_properties.return_value = {
"last_irreversible_block_num": 100,
"total_vesting_fund": fund,
"total_vesting_shares": shares,
}
award_stats._rate_cache = (0.0, 0.0)


def test_capture_payouts_sums_shares_and_converts_to_viz(_viz):
award_stats.ensure_indexes()
_rate(_viz) # 2 VIZ per SHARE
award_stats.capture_payouts(_archival_block(), 50)
award_stats.capture_payouts(_archival_block(), 50) # idempotent re-run
award_stats.dispatch(_op("award", {"initiator": "alice", "receiver": "bob", "memo": "post-1"}, op_id=50.0001))

t = award_stats.totals("bob", "post-1")
assert t["count"] == 1 # payout rows never inflate the award count
assert t["initiators"] == ["alice"]
assert t["total_shares"] == 2.5
assert t["total_viz"] == 5.0


def test_totals_viz_is_null_when_rate_unavailable(_viz):
award_stats.ensure_indexes()
_viz.rpc.get_dynamic_global_properties.side_effect = RuntimeError("node down")
award_stats._rate_cache = (0.0, 0.0)
award_stats.capture_payouts(_archival_block(), 50)

t = award_stats.totals("bob", "post-1")
assert t["total_shares"] == 2.5
assert t["total_viz"] is None
10 changes: 3 additions & 7 deletions config/deploy.web.yml
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,8 @@
#
# kamal deploy -c config/deploy.web.yml
#
# Serves explorer.viz.cx (canonical) + network.viz.cx, which the app itself
# 308s to explorer.viz.cx — host-scoped redirects() in web/next.config.ts. Both
# hosts are listed under proxy.hosts so kamal-proxy issues/renews certs for both.
# network.viz.cx is a grace-window alias (canonical 2026-08 → 2026-09); drop it
# from here + DNS once traffic is gone.
# Serves explorer.viz.cx. (network.viz.cx was the canonical host 2026-08 →
# 2026-09; its 308 alias was dropped 2026-09-18 after the grace window.)
#
# The viz.cx apex + www.viz.cx belong to the UGC platform (repo viz-cx/viz.cx,
# Kamal app viz-cx-platform) since the 2026-09 cutover. kamal-proxy refuses
Expand All @@ -31,8 +28,7 @@ ssh:

proxy:
hosts:
- explorer.viz.cx # canonical
- network.viz.cx # alias — the app 308s it to explorer.viz.cx
- explorer.viz.cx
app_port: 3000
ssl: true
healthcheck:
Expand Down
14 changes: 0 additions & 14 deletions web/next.config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,20 +30,6 @@ const nextConfig: NextConfig = {
async headers() {
return [{ source: "/:path*", headers: SECURITY_HEADERS }];
},
// network.viz.cx was the canonical host 2026-08 → 2026-09 and stays as an
// alias for a grace window — permanently redirect it to explorer.viz.cx
// (path + query preserved). Host-scoped so the proxy healthcheck (container
// address as Host) is never redirected. permanent:true => 308.
async redirects() {
return [
{
source: "/:path*",
has: [{ type: "host", value: "network.viz.cx" }],
destination: "https://explorer.viz.cx/:path*",
permanent: true,
},
];
},
};

// Wrap with Sentry. Source-map upload only happens when SENTRY_AUTH_TOKEN is
Expand Down
Loading